SinkBinding v1beta1 Reconciler & E2E test

pkg/reconciler/sinkbinding/controller.go

@@ -19,7 +19,7 @@ package sinkbinding
 import (
 	"context"
 
-	sbinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1alpha1/sinkbinding"
+	sbinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1beta1/sinkbinding"
 	"knative.dev/pkg/client/injection/ducks/duck/v1/podspecable"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/namespace"
 	"knative.dev/pkg/reconciler"
@@ -31,7 +31,7 @@ import (
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
-	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
+	"knative.dev/eventing/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/apis/duck"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -73,7 +73,7 @@ func NewController(
 				return nil
 			},
 		},
-		GVR: v1alpha1.SchemeGroupVersion.WithResource("sinkbindings"),
+		GVR: v1beta1.SchemeGroupVersion.WithResource("sinkbindings"),
 		Get: func(namespace string, name string) (psbinding.Bindable, error) {
 			return sbInformer.Lister().SinkBindings(namespace).Get(name)
 		},
@@ -125,12 +125,12 @@ func WithContextFactory(ctx context.Context, handler func(types.NamespacedName))
 	r := resolver.NewURIResolver(ctx, handler)
 
 	return func(ctx context.Context, b psbinding.Bindable) (context.Context, error) {
-		sb := b.(*v1alpha1.SinkBinding)
+		sb := b.(*v1beta1.SinkBinding)
 		uri, err := r.URIFromDestinationV1(sb.Spec.Sink, sb)
 		if err != nil {
 			return nil, err
 		}
 		sb.Status.SinkURI = uri
-		return v1alpha1.WithSinkURI(ctx, sb.Status.SinkURI), nil
+		return v1beta1.WithSinkURI(ctx, sb.Status.SinkURI), nil
 	}
 }

pkg/reconciler/testing/sinkbinding.go

@@ -21,6 +21,7 @@ import (
 
 	sourcesv1alpha1 "knative.dev/eventing/pkg/apis/sources/v1alpha1"
 	sourcesv1alpha2 "knative.dev/eventing/pkg/apis/sources/v1alpha2"
+	sourcesv1beta1 "knative.dev/eventing/pkg/apis/sources/v1beta1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/tracker"
 )
@@ -31,6 +32,9 @@ type SinkBindingV1Alpha1Option func(*sourcesv1alpha1.SinkBinding)
 // SinkBindingV1Alpha2Option enables further configuration of a SinkBinding.
 type SinkBindingV1Alpha2Option func(*sourcesv1alpha2.SinkBinding)
 
+// SinkBindingV1Beta1Option enables further configuration of a SinkBinding.
+type SinkBindingV1Beta1Option func(*sourcesv1beta1.SinkBinding)
+
 // NewSinkBindingV1Alpha1 creates a SinkBinding with SinkBindingOptions
 func NewSinkBindingV1Alpha1(name, namespace string, o ...SinkBindingV1Alpha1Option) *sourcesv1alpha1.SinkBinding {
 	c := &sourcesv1alpha1.SinkBinding{
@@ -61,6 +65,21 @@ func NewSinkBindingV1Alpha2(name, namespace string, o ...SinkBindingV1Alpha2Opti
 	return c
 }
 
+// NewSinkBindingV1Beta1 creates a SinkBinding with SinkBindingOptions
+func NewSinkBindingV1Beta1(name, namespace string, o ...SinkBindingV1Beta1Option) *sourcesv1beta1.SinkBinding {
+	c := &sourcesv1beta1.SinkBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+	for _, opt := range o {
+		opt(c)
+	}
+	//c.SetDefaults(context.Background()) // TODO: We should add defaults and validation.
+	return c
+}
+
 // WithSubjectV1A1 assigns the subject of the SinkBinding.
 func WithSubjectV1A1(subject tracker.Reference) SinkBindingV1Alpha1Option {
 	return func(sb *sourcesv1alpha1.SinkBinding) {
@@ -102,3 +121,24 @@ func WithCloudEventOverridesV1A2(overrides duckv1.CloudEventOverrides) SinkBindi
 		sb.Spec.CloudEventOverrides = &overrides
 	}
 }
+
+// WithSubjectV1B1 assigns the subject of the SinkBinding.
+func WithSubjectV1B1(subject tracker.Reference) SinkBindingV1Beta1Option {
+	return func(sb *sourcesv1beta1.SinkBinding) {
+		sb.Spec.Subject = subject
+	}
+}
+
+// WithSinkV1B1 assigns the sink of the SinkBinding.
+func WithSinkV1B1(sink duckv1.Destination) SinkBindingV1Beta1Option {
+	return func(sb *sourcesv1beta1.SinkBinding) {
+		sb.Spec.Sink = sink
+	}
+}
+
+// WithCloudEventOverridesV1B1 assigns the CloudEventsOverrides of the SinkBinding.
+func WithCloudEventOverridesV1B1(overrides duckv1.CloudEventOverrides) SinkBindingV1Beta1Option {
+	return func(sb *sourcesv1beta1.SinkBinding) {
+		sb.Spec.CloudEventOverrides = &overrides
+	}
+}

test/e2e/source_sinkbinding_v1beta1_test.go

@@ -0,0 +1,211 @@
+// +build e2e
+
+/*
+Copyright 2020 The Knative Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"fmt"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+
+	. "github.com/cloudevents/sdk-go/v2/test"
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	pkgTest "knative.dev/pkg/test"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/resources"
+
+	eventingtesting "knative.dev/eventing/pkg/reconciler/testing"
+)
+
+func TestSinkBindingV1Beta1Deployment(t *testing.T) {
+	const (
+		sinkBindingName = "e2e-sink-binding"
+		deploymentName  = "e2e-sink-binding-deployment"
+		// the heartbeats image is built from test_images/heartbeats
+		imageName = "heartbeats"
+
+		recordEventPodName = "e2e-sink-binding-recordevent-pod-v1beta1dep"
+	)
+
+	client := setup(t, true)
+	defer tearDown(client)
+
+	// create event logger pod and service
+	eventTracker, _ := recordevents.StartEventRecordOrFail(client, recordEventPodName)
+	defer eventTracker.Cleanup()
+
+	extensionSecret := string(uuid.NewUUID())
+
+	// create sink binding
+	sinkBinding := eventingtesting.NewSinkBindingV1Beta1(
+		sinkBindingName,
+		client.Namespace,
+		eventingtesting.WithSinkV1B1(duckv1.Destination{Ref: resources.KnativeRefForService(recordEventPodName, client.Namespace)}),
+		eventingtesting.WithSubjectV1B1(tracker.Reference{
+			APIVersion: "apps/v1",
+			Kind:       "Deployment",
+			Namespace:  client.Namespace,
+			Name:       deploymentName,
+		}),
+		eventingtesting.WithCloudEventOverridesV1B1(duckv1.CloudEventOverrides{Extensions: map[string]string{
+			"sinkbinding": extensionSecret,
+		}}),
+	)
+	client.CreateSinkBindingV1Beta1OrFail(sinkBinding)
+
+	message := fmt.Sprintf("TestSinkBindingDeployment%s", uuid.NewUUID())
+	client.CreateDeploymentOrFail(&appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: client.Namespace,
+			Name:      deploymentName,
+		},
+		Spec: appsv1.DeploymentSpec{
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"foo": "bar",
+				},
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"foo": "bar",
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{{
+						Name:            imageName,
+						Image:           pkgTest.ImagePath(imageName),
+						ImagePullPolicy: corev1.PullAlways,
+						Args:            []string{"--msg=" + message},
+						Env: []corev1.EnvVar{{
+							Name:  "POD_NAME",
+							Value: deploymentName,
+						}, {
+							Name:  "POD_NAMESPACE",
+							Value: client.Namespace,
+						}},
+					}},
+				},
+			},
+		},
+	})
+
+	// wait for all test resources to be ready
+	client.WaitForAllTestResourcesReadyOrFail()
+
+	// Look for events with expected data, and sinkbinding extension
+	eventTracker.AssertAtLeast(2, recordevents.MatchEvent(
+		recordevents.MatchHeartBeatsImageMessage(message),
+		HasSource(fmt.Sprintf("https://knative.dev/eventing/test/heartbeats/#%s/%s", client.Namespace, deploymentName)),
+		HasExtension("sinkbinding", extensionSecret),
+	))
+}
+
+func TestSinkBindingV1Beta1CronJob(t *testing.T) {
+	const (
+		sinkBindingName = "e2e-sink-binding"
+		deploymentName  = "e2e-sink-binding-cronjob"
+		// the heartbeats image is built from test_images/heartbeats
+		imageName = "heartbeats"
+
+		recordEventPodName = "e2e-sink-binding-recordevent-pod-v1beta1c"
+	)
+
+	client := setup(t, true)
+	defer tearDown(client)
+
+	// create event logger pod and service
+	eventTracker, _ := recordevents.StartEventRecordOrFail(client, recordEventPodName)
+	defer eventTracker.Cleanup()
+
+	// create sink binding
+	sinkBinding := eventingtesting.NewSinkBindingV1Beta1(
+		sinkBindingName,
+		client.Namespace,
+		eventingtesting.WithSinkV1B1(duckv1.Destination{Ref: resources.KnativeRefForService(recordEventPodName, client.Namespace)}),
+		eventingtesting.WithSubjectV1B1(tracker.Reference{
+			APIVersion: "batch/v1",
+			Kind:       "Job",
+			Namespace:  client.Namespace,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"foo": "bar",
+				},
+			},
+		}),
+	)
+	client.CreateSinkBindingV1Beta1OrFail(sinkBinding)
+
+	message := fmt.Sprintf("TestSinkBindingCronJob%s", uuid.NewUUID())
+	client.CreateCronJobOrFail(&batchv1beta1.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: client.Namespace,
+			Name:      deploymentName,
+		},
+		Spec: batchv1beta1.CronJobSpec{
+			Schedule: "* * * * *",
+			JobTemplate: batchv1beta1.JobTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"foo": "bar",
+					},
+				},
+				Spec: batchv1.JobSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							RestartPolicy: corev1.RestartPolicyNever,
+							Containers: []corev1.Container{{
+								Name:            imageName,
+								Image:           pkgTest.ImagePath(imageName),
+								ImagePullPolicy: corev1.PullAlways,
+								Args:            []string{"--msg=" + message},
+								Env: []corev1.EnvVar{{
+									Name:  "ONE_SHOT",
+									Value: "true",
+								}, {
+									Name:  "POD_NAME",
+									Value: deploymentName,
+								}, {
+									Name:  "POD_NAMESPACE",
+									Value: client.Namespace,
+								}},
+							}},
+						},
+					},
+				},
+			},
+		},
+	})
+
+	// wait for all test resources to be ready
+	client.WaitForAllTestResourcesReadyOrFail()
+
+	// verify the logger service receives the event
+	eventTracker.AssertAtLeast(2, recordevents.MatchEvent(
+		recordevents.MatchHeartBeatsImageMessage(message),
+		HasSource(fmt.Sprintf("https://knative.dev/eventing/test/heartbeats/#%s/%s", client.Namespace, deploymentName)),
+	))
+
+}

test/lib/creation.go

@@ -19,8 +19,6 @@ package lib
 import (
 	"fmt"
 
-	sourcesv1alpha2 "knative.dev/eventing/pkg/apis/sources/v1alpha2"
-
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	corev1 "k8s.io/api/core/v1"
@@ -38,6 +36,8 @@ import (
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	messagingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
 	sourcesv1alpha1 "knative.dev/eventing/pkg/apis/sources/v1alpha1"
+	sourcesv1alpha2 "knative.dev/eventing/pkg/apis/sources/v1alpha2"
+	sourcesv1beta1 "knative.dev/eventing/pkg/apis/sources/v1beta1"
 	"knative.dev/eventing/pkg/utils"
 	"knative.dev/eventing/test/lib/duck"
 	"knative.dev/eventing/test/lib/resources"
@@ -322,6 +322,17 @@ func (c *Client) CreateSinkBindingV1Alpha2OrFail(sb *sourcesv1alpha2.SinkBinding
 	c.Tracker.AddObj(sb)
 }
 
+// CreateSinkBindingV1Beta1OrFail will create a SinkBinding or fail the test if there is an error.
+func (c *Client) CreateSinkBindingV1Beta1OrFail(sb *sourcesv1beta1.SinkBinding) {
+	c.T.Logf("Creating sinkbinding %+v", sb)
+	sbInterface := c.Eventing.SourcesV1beta1().SinkBindings(c.Namespace)
+	_, err := sbInterface.Create(sb)
+	if err != nil {
+		c.T.Fatalf("Failed to create sinkbinding %q: %v", sb.Name, err)
+	}
+	c.Tracker.AddObj(sb)
+}
+
 // CreateApiServerSourceOrFail will create an ApiServerSource
 func (c *Client) CreateApiServerSourceOrFail(apiServerSource *sourcesv1alpha2.ApiServerSource) {
 	c.T.Logf("Creating apiserversource %+v", apiServerSource)