Skip to content

Commit 5c73631

Browse files
zugaozugao
authored
Merge pull request #73 from vshn/improve-validation
improve version validation on DBaaS
2 parents b15136a + e879838 commit 5c73631

File tree

16 files changed

+403
-129
lines changed

16 files changed

+403
-129
lines changed

‎generate_sample.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -299,6 +299,7 @@ func newKafkaSample() *exoscalev1.Kafka {
299299
},
300300
IPFilter: exoscalev1.IPFilter{"0.0.0.0/0"},
301301
},
302+
Version: "3.2",
302303
KafkaSettings: runtime.RawExtension{Raw: []byte(`{"connections_max_idle_ms": 60000}`)},
303304
},
304305
},

‎operator/kafkacontroller/observe_test.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -389,6 +389,7 @@ func sampleKafka(name string) exoscalev1.Kafka {
389389
Name: "foo",
390390
},
391391
}
392+
instance.Spec.ForProvider.Version = "3.2"
392393
instance.Spec.ForProvider.Size.Plan = "businesss-8"
393394
instance.Spec.ForProvider.IPFilter = []string{
394395
"0.0.0.0/0",

‎operator/kafkacontroller/webhook.go

Lines changed: 56 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,42 +3,93 @@ package kafkacontroller
33
import (
44
"context"
55
"fmt"
6+
"github.com/exoscale/egoscale/v2/oapi"
67
"strings"
78

9+
exoscalesdk "github.com/exoscale/egoscale/v2"
810
exoscalev1 "github.com/vshn/provider-exoscale/apis/exoscale/v1"
11+
"github.com/vshn/provider-exoscale/operator/pipelineutil"
912
"github.com/vshn/provider-exoscale/operator/webhook"
1013
ctrl "sigs.k8s.io/controller-runtime"
14+
"sigs.k8s.io/controller-runtime/pkg/client"
1115

1216
"github.com/go-logr/logr"
1317
"k8s.io/apimachinery/pkg/runtime"
1418
)
1519

20+
const serviceType = "kafka"
21+
1622
// SetupWebhook adds a webhook for kafka resources.
1723
func SetupWebhook(mgr ctrl.Manager) error {
1824
return ctrl.NewWebhookManagedBy(mgr).
1925
For(&exoscalev1.Kafka{}).
2026
WithValidator(&Validator{
21-
log: mgr.GetLogger().WithName("webhook").WithName(strings.ToLower(exoscalev1.KafkaKind)),
27+
log: mgr.GetLogger().WithName("webhook").WithName(strings.ToLower(exoscalev1.KafkaKind)),
28+
kube: mgr.GetClient(),
2229
}).
2330
Complete()
2431
}
2532

2633
// Validator validates kafka admission requests.
2734
type Validator struct {
28-
log logr.Logger
35+
log logr.Logger
36+
kube client.Client
2937
}
3038

3139
// ValidateCreate validates the spec of a created kafka resource.
32-
func (v *Validator) ValidateCreate(_ context.Context, obj runtime.Object) error {
40+
func (v *Validator) ValidateCreate(ctx context.Context, obj runtime.Object) error {
41+
instance := obj.(*exoscalev1.Kafka)
42+
v.log.V(1).Info("get kafka available versions")
43+
exo, err := pipelineutil.OpenExoscaleClient(ctx, v.kube, instance.GetProviderConfigName(), exoscalesdk.ClientOptWithAPIEndpoint(fmt.Sprintf("https://api-%s.exoscale.com", instance.Spec.ForProvider.Zone)))
44+
if err != nil {
45+
return fmt.Errorf("open exoscale client failed: %w", err)
46+
}
47+
return v.validateCreateWithExoClient(ctx, obj, exo.Exoscale)
48+
}
49+
50+
func (v *Validator) validateCreateWithExoClient(ctx context.Context, obj runtime.Object, exo oapi.ClientWithResponsesInterface) error {
3351
instance, ok := obj.(*exoscalev1.Kafka)
3452
if !ok {
3553
return fmt.Errorf("invalid managed resource type %T for kafka webhook", obj)
3654
}
3755
v.log.V(2).WithValues("instance", instance).Info("validate create")
3856

57+
availableVersions, err := v.getAvailableVersions(ctx, exo)
58+
if err != nil {
59+
return err
60+
}
61+
62+
err = v.validateVersion(ctx, obj, *availableVersions)
63+
if err != nil {
64+
return err
65+
}
66+
3967
return validateSpec(instance.Spec.ForProvider)
4068
}
4169

70+
func (v *Validator) getAvailableVersions(ctx context.Context, exo oapi.ClientWithResponsesInterface) (*[]string, error) {
71+
// get kafka available versions
72+
resp, err := exo.GetDbaasServiceTypeWithResponse(ctx, serviceType)
73+
if err != nil {
74+
return nil, fmt.Errorf("get DBaaS service type failed: %w", err)
75+
}
76+
77+
v.log.V(1).Info("DBaaS service type", "body", string(resp.Body))
78+
79+
serviceType := *resp.JSON200
80+
if serviceType.AvailableVersions == nil {
81+
return nil, fmt.Errorf("kafka available versions not found")
82+
}
83+
return serviceType.AvailableVersions, nil
84+
}
85+
86+
func (v *Validator) validateVersion(_ context.Context, obj runtime.Object, availableVersions []string) error {
87+
instance := obj.(*exoscalev1.Kafka)
88+
89+
v.log.V(1).Info("validate version")
90+
return webhook.ValidateVersions(instance.Spec.ForProvider.Version, availableVersions)
91+
}
92+
4293
// ValidateUpdate validates the spec of an updated kafka resource and checks that no immutable field has been modified.
4394
func (v *Validator) ValidateUpdate(_ context.Context, oldObj, newObj runtime.Object) error {
4495
newInstance, ok := newObj.(*exoscalev1.Kafka)
@@ -47,7 +98,7 @@ func (v *Validator) ValidateUpdate(_ context.Context, oldObj, newObj runtime.Obj
4798
}
4899
oldInstance, ok := oldObj.(*exoscalev1.Kafka)
49100
if !ok {
50-
return fmt.Errorf("invalid managed resource type %T for kafka webhook", newObj)
101+
return fmt.Errorf("invalid managed resource type %T for kafka webhook", oldObj)
51102
}
52103
v.log.V(2).WithValues("old", oldInstance, "new", newInstance).Info("VALIDATE update")
53104

@@ -111,13 +162,9 @@ func compareVersion(oldInst, newInst exoscalev1.Kafka) error {
111162
if oldInst.Spec.ForProvider.Version == newInst.Spec.ForProvider.Version {
112163
return nil
113164
}
114-
if newInst.Spec.ForProvider.Version == "" {
115-
// Setting version to empyt string should always be fine
116-
return nil
117-
}
118165
if oldInst.Spec.ForProvider.Version == "" {
119166
// Fall back to reported version if no version was set before
120167
oldInst.Spec.ForProvider.Version = oldInst.Status.AtProvider.Version
121168
}
122-
return webhook.ValidateVersion(oldInst.Status.AtProvider.Version, oldInst.Spec.ForProvider.Version, newInst.Spec.ForProvider.Version)
169+
return webhook.ValidateUpdateVersion(oldInst.Status.AtProvider.Version, oldInst.Spec.ForProvider.Version, newInst.Spec.ForProvider.Version)
123170
}

‎operator/kafkacontroller/webhook_test.go

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@ package kafkacontroller
22

33
import (
44
"context"
5+
"github.com/exoscale/egoscale/v2/oapi"
6+
"github.com/stretchr/testify/mock"
7+
"github.com/vshn/provider-exoscale/internal/operatortest"
58
"testing"
69

710
"github.com/go-logr/logr"
@@ -15,26 +18,34 @@ func TestWebhook_Create(t *testing.T) {
1518
log: logr.Discard(),
1619
}
1720

21+
exoMock := &operatortest.ClientWithResponsesInterface{}
22+
mockAvailableVersionsCall(exoMock)
1823
base := sampleKafka("foo")
1924

2025
t.Run("valid", func(t *testing.T) {
21-
err := v.ValidateCreate(ctx, &base)
26+
err := v.validateCreateWithExoClient(ctx, &base, exoMock)
2227
assert.NoError(t, err)
2328
})
2429
t.Run("invalid empty", func(t *testing.T) {
25-
err := v.ValidateCreate(ctx, &exoscalev1.Kafka{})
30+
err := v.validateCreateWithExoClient(ctx, &exoscalev1.Kafka{}, exoMock)
2631
assert.Error(t, err)
2732
})
2833
t.Run("invalid no ipfilter", func(t *testing.T) {
2934
inst := base
3035
inst.Spec.ForProvider.IPFilter = nil
31-
err := v.ValidateCreate(ctx, &inst)
36+
err := v.validateCreateWithExoClient(ctx, &inst, exoMock)
3237
assert.Error(t, err)
3338
})
3439
t.Run("invalid no time", func(t *testing.T) {
3540
inst := base
3641
inst.Spec.ForProvider.Maintenance.TimeOfDay = ""
37-
err := v.ValidateCreate(ctx, &inst)
42+
err := v.validateCreateWithExoClient(ctx, &inst, exoMock)
43+
assert.Error(t, err)
44+
})
45+
t.Run("invalid version", func(t *testing.T) {
46+
inst := base
47+
inst.Spec.ForProvider.Version = "non-valid-version"
48+
err := v.validateCreateWithExoClient(ctx, &inst, exoMock)
3849
assert.Error(t, err)
3950
})
4051
}
@@ -91,3 +102,12 @@ func TestWebhook_Update(t *testing.T) {
91102
})
92103

93104
}
105+
106+
func mockAvailableVersionsCall(m *operatortest.ClientWithResponsesInterface) {
107+
m.On("GetDbaasServiceTypeWithResponse", mock.Anything, mock.Anything).
108+
Return(&oapi.GetDbaasServiceTypeResponse{
109+
JSON200: &oapi.DbaasServiceType{
110+
AvailableVersions: &[]string{"3.2", "4.3"},
111+
},
112+
}, nil)
113+
}

‎operator/mysqlcontroller/setup.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,8 @@ func SetupWebhook(mgr ctrl.Manager) error {
4141
return ctrl.NewWebhookManagedBy(mgr).
4242
For(&exoscalev1.MySQL{}).
4343
WithValidator(&Validator{
44-
log: mgr.GetLogger().WithName("webhook").WithName(strings.ToLower(exoscalev1.MySQLKind)),
44+
log: mgr.GetLogger().WithName("webhook").WithName(strings.ToLower(exoscalev1.MySQLKind)),
45+
kube: mgr.GetClient(),
4546
}).
4647
Complete()
4748
}

0 commit comments

Comments
 (0)