From aa5b8d6695cde2439836eefb9afdc812ce993d85 Mon Sep 17 00:00:00 2001 From: Adam Harwayne Date: Fri, 26 Jul 2019 14:02:35 -0700 Subject: [PATCH] ApiServerSource updates its receive adapter, if needed. --- .../apiserversource/apiserversource.go | 42 ++--- .../apiserversource/apiserversource_test.go | 173 +++++++++++++++++- 2 files changed, 190 insertions(+), 25 deletions(-) diff --git a/pkg/reconciler/apiserversource/apiserversource.go b/pkg/reconciler/apiserversource/apiserversource.go index 2403f3913be..b445aa94629 100644 --- a/pkg/reconciler/apiserversource/apiserversource.go +++ b/pkg/reconciler/apiserversource/apiserversource.go @@ -50,6 +50,8 @@ const ( apiserversourceReconciled = "ApiServerSourceReconciled" apiServerSourceReadinessChanged = "ApiServerSourceReadinessChanged" apiserversourceUpdateStatusFailed = "ApiServerSourceUpdateStatusFailed" + apiserversourceDeploymentCreated = "ApiServerSourceDeploymentCreated" + apiserversourceDeploymentUpdated = "ApiServerSourceDeploymentUpdated" // raImageEnvVar is the name of the environment variable that contains the receive adapter's // image. It must be defined. @@ -173,15 +175,6 @@ func (r *Reconciler) getReceiveAdapterImage() string { } func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1alpha1.ApiServerSource, sinkURI string) (*appsv1.Deployment, error) { - ra, err := r.getReceiveAdapter(ctx, src) - if err != nil && !apierrors.IsNotFound(err) { - logging.FromContext(ctx).Error("Unable to get an existing receive adapter", zap.Error(err)) - return nil, err - } - if ra != nil { - logging.FromContext(ctx).Info("Reusing existing receive adapter", zap.Any("receiveAdapter", ra)) - return ra, nil - } adapterArgs := resources.ReceiveAdapterArgs{ Image: r.getReceiveAdapterImage(), Source: src, @@ -189,24 +182,25 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1alpha1.Api SinkURI: sinkURI, } expected := resources.MakeReceiveAdapter(&adapterArgs) - if ra != nil { - if r.podSpecChanged(ra.Spec.Template.Spec, expected.Spec.Template.Spec) { - ra.Spec.Template.Spec = expected.Spec.Template.Spec - if ra, err = r.KubeClientSet.AppsV1().Deployments(src.Namespace).Update(ra); err != nil { - return ra, err - } - logging.FromContext(ctx).Info("Receive Adapter updated.", zap.Any("receiveAdapter", ra)) - } else { - logging.FromContext(ctx).Info("Reusing existing receive adapter", zap.Any("receiveAdapter", ra)) + + ra, err := r.getReceiveAdapter(ctx, src) + if apierrors.IsNotFound(err) { + ra, err = r.KubeClientSet.AppsV1().Deployments(src.Namespace).Create(expected) + r.Recorder.Eventf(src, corev1.EventTypeNormal, apiserversourceDeploymentCreated, "Deployment created, error: %v", err) + return ra, err + } else if err != nil { + return nil, fmt.Errorf("error getting receive adapter: %v", err) + } else if r.podSpecChanged(ra.Spec.Template.Spec, expected.Spec.Template.Spec) { + ra.Spec.Template.Spec = expected.Spec.Template.Spec + if ra, err = r.KubeClientSet.AppsV1().Deployments(src.Namespace).Update(ra); err != nil { + return ra, err } + r.Recorder.Eventf(src, corev1.EventTypeNormal, apiserversourceDeploymentUpdated, "Deployment updated") return ra, nil + } else { + logging.FromContext(ctx).Debug("Reusing existing receive adapter", zap.Any("receiveAdapter", ra)) } - - if ra, err = r.KubeClientSet.AppsV1().Deployments(src.Namespace).Create(expected); err != nil { - return nil, err - } - logging.FromContext(ctx).Info("Receive Adapter created.", zap.Any("receiveAdapter", expected)) - return ra, err + return ra, nil } func (r *Reconciler) reconcileEventTypes(ctx context.Context, src *v1alpha1.ApiServerSource) error { diff --git a/pkg/reconciler/apiserversource/apiserversource_test.go b/pkg/reconciler/apiserversource/apiserversource_test.go index 60e73d9cd09..ba90de22b6e 100644 --- a/pkg/reconciler/apiserversource/apiserversource_test.go +++ b/pkg/reconciler/apiserversource/apiserversource_test.go @@ -63,7 +63,6 @@ var ( const ( image = "github.com/knative/test/image" sourceName = "test-apiserver-source" - sourceUID = "1234-5678-90" testNS = "testnamespace" sinkName = "testsink" @@ -124,6 +123,7 @@ func TestReconcile(t *testing.T) { }, Key: testNS + "/" + sourceName, WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentCreated", "Deployment created, error: "), Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), }, @@ -149,6 +149,152 @@ func TestReconcile(t *testing.T) { makeReceiveAdapter(), }, }, + { + Name: "valid - deployment update due to env", + Objects: []runtime.Object{ + NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + }), + ), + NewChannel(sinkName, testNS, + WithInitChannelConditions, + WithChannelAddress(sinkDNS), + ), + makeReceiveAdapterWithDifferentEnv(), + }, + Key: testNS + "/" + sourceName, + WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentUpdated", "Deployment updated"), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + }), + // Status Update: + WithInitApiServerSourceConditions, + WithApiServerSourceDeployed, + WithApiServerSourceSink(sinkURI), + WithApiServerSourceEventTypes, + ), + }}, + WantUpdates: []clientgotesting.UpdateActionImpl{{ + Object: makeReceiveAdapter(), + }}, + }, + { + Name: "valid - deployment update due to service account", + Objects: []runtime.Object{ + NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + ServiceAccountName: "malin", + }), + ), + NewChannel(sinkName, testNS, + WithInitChannelConditions, + WithChannelAddress(sinkDNS), + ), + makeReceiveAdapterWithDifferentServiceAccount("morgan"), + }, + Key: testNS + "/" + sourceName, + WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentUpdated", "Deployment updated"), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + ServiceAccountName: "malin", + }), + // Status Update: + WithInitApiServerSourceConditions, + WithApiServerSourceDeployed, + WithApiServerSourceSink(sinkURI), + WithApiServerSourceEventTypes, + ), + }}, + WantUpdates: []clientgotesting.UpdateActionImpl{{ + Object: makeReceiveAdapterWithDifferentServiceAccount("malin"), + }}, + }, + { + Name: "valid - deployment update due to container count", + Objects: []runtime.Object{ + NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + }), + ), + NewChannel(sinkName, testNS, + WithInitChannelConditions, + WithChannelAddress(sinkDNS), + ), + makeReceiveAdapterWithDifferentContainerCount(), + }, + Key: testNS + "/" + sourceName, + WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentUpdated", "Deployment updated"), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), + Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewApiServerSource(sourceName, testNS, + WithApiServerSourceSpec(sourcesv1alpha1.ApiServerSourceSpec{ + Resources: []sourcesv1alpha1.ApiServerResource{ + { + APIVersion: "", + Kind: "Namespace", + }, + }, + Sink: &sinkRef, + }), + // Status Update: + WithInitApiServerSourceConditions, + WithApiServerSourceDeployed, + WithApiServerSourceSink(sinkURI), + WithApiServerSourceEventTypes, + ), + }}, + WantUpdates: []clientgotesting.UpdateActionImpl{{ + Object: makeReceiveAdapter(), + }}, + }, { Name: "valid with event types to delete", Objects: []runtime.Object{ @@ -171,6 +317,7 @@ func TestReconcile(t *testing.T) { }, Key: testNS + "/" + sourceName, WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentCreated", "Deployment created, error: "), Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), }, @@ -220,6 +367,7 @@ func TestReconcile(t *testing.T) { }, Key: testNS + "/" + sourceName, WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentCreated", "Deployment created, error: "), Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), }, @@ -275,6 +423,7 @@ func TestReconcile(t *testing.T) { }, Key: testNS + "/" + sourceName, WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentCreated", "Deployment created, error: "), Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), }, @@ -330,6 +479,7 @@ func TestReconcile(t *testing.T) { }, Key: testNS + "/" + sourceName, WantEvents: []string{ + Eventf(corev1.EventTypeNormal, "ApiServerSourceDeploymentCreated", "Deployment created, error: "), Eventf(corev1.EventTypeNormal, "ApiServerSourceReconciled", `ApiServerSource reconciled: "%s/%s"`, testNS, sourceName), Eventf(corev1.EventTypeNormal, "ApiServerSourceReadinessChanged", `ApiServerSource %q became ready`, sourceName), }, @@ -408,6 +558,27 @@ func makeReceiveAdapter() *appsv1.Deployment { return resources.MakeReceiveAdapter(&args) } +func makeReceiveAdapterWithDifferentEnv() *appsv1.Deployment { + ra := makeReceiveAdapter() + ra.Spec.Template.Spec.Containers[0].Env = append(ra.Spec.Template.Spec.Containers[0].Env, corev1.EnvVar{ + Name: "not-in", + Value: "the-original", + }) + return ra +} + +func makeReceiveAdapterWithDifferentServiceAccount(name string) *appsv1.Deployment { + ra := makeReceiveAdapter() + ra.Spec.Template.Spec.ServiceAccountName = name + return ra +} + +func makeReceiveAdapterWithDifferentContainerCount() *appsv1.Deployment { + ra := makeReceiveAdapter() + ra.Spec.Template.Spec.Containers = append(ra.Spec.Template.Spec.Containers, corev1.Container{}) + return ra +} + func makeEventTypeWithName(eventType, name string) *v1alpha1.EventType { et := makeEventType(eventType) et.Name = name