Allow valid object references for Subscription's channel

Gopkg.toml

@@ -47,6 +47,13 @@ required = [
   name = "k8s.io/apimachinery"
   version = "kubernetes-1.12.6"
 
+# Overridden to ensure compatibility with GKE
+# GKE version as of 2019-01-24 is 1.11
+# controller-runtime 0.1.9 requires at least 1.12
+[[override]]
+  name = "k8s.io/apiextensions-apiserver"
+  version = "kubernetes-1.12.6"
+
 # Overridden to ensure compatibility with GKE
 # GKE version as of 2019-01-24 is 1.11
 # controller-runtime 0.1.9 requires at least 1.12

cmd/controller/main.go

@@ -43,6 +43,7 @@ import (
 	pkgmetrics "github.com/knative/pkg/metrics"
 	"github.com/knative/pkg/signals"
 	"go.uber.org/zap"
+	apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/rest"
 )
@@ -80,6 +81,7 @@ func main() {
 
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(opt.KubeClientSet, opt.ResyncPeriod)
 	eventingInformerFactory := informers.NewSharedInformerFactory(opt.EventingClientSet, opt.ResyncPeriod)
+	apiExtensionsInformerFactory := apiextensionsinformers.NewSharedInformerFactory(opt.ApiExtensionsClientSet, opt.ResyncPeriod)
 
 	// Eventing
 	triggerInformer := eventingInformerFactory.Eventing().V1alpha1().Triggers()
@@ -96,6 +98,9 @@ func main() {
 	serviceAccountInformer := kubeInformerFactory.Core().V1().ServiceAccounts()
 	roleBindingInformer := kubeInformerFactory.Rbac().V1().RoleBindings()
 
+	// Api Extensions
+	customResourceDefinitionInformer := apiExtensionsInformerFactory.Apiextensions().V1beta1().CustomResourceDefinitions()
+
 	// Duck
 	addressableInformer := duck.NewAddressableInformer(opt)
 
@@ -107,6 +112,7 @@ func main() {
 			opt,
 			subscriptionInformer,
 			addressableInformer,
+			customResourceDefinitionInformer,
 		),
 		namespace.NewController(
 			opt,
@@ -180,6 +186,8 @@ func main() {
 		deploymentInformer.Informer(),
 		serviceAccountInformer.Informer(),
 		roleBindingInformer.Informer(),
+		// Api Extensions
+		customResourceDefinitionInformer.Informer(),
 	); err != nil {
 		logger.Fatalf("Failed to start informers: %v", err)
 	}

config/200-controller-clusterrole.yaml

@@ -90,3 +90,13 @@ rules:
       - "apiserversources/status"
       - "apiserversources/finalizers"
     verbs: *everything
+
+  # The subscription controller needs to retrieve and watch CustomResourceDefinitions.
+  - apiGroups:
+      - "apiextensions.k8s.io"
+    resources:
+      - "customresourcedefinitions"
+    verbs:
+      - "get"
+      - "list"
+      - "watch"

config/300-subscription.yaml

@@ -62,7 +62,6 @@ spec:
                   minLength: 1
                 kind:
                   type: string
-                  pattern: "^Channel$"
                 name:
                   type: string
                   minLength: 1
@@ -106,7 +105,6 @@ spec:
                       minLength: 1
                     kind:
                       type: string
-                      pattern: "^Channel$"
                     name:
                       type: string
                       minLength: 1

config/channels/in-memory-channel/300-in-memory-channel.yaml

@@ -17,6 +17,7 @@ metadata:
  name: inmemorychannels.messaging.knative.dev
  labels:
     knative.dev/crd-install: "true"
+    messaging.knative.dev/subscribable: "true"
 spec:
   group: messaging.knative.dev
   version: v1alpha1
@@ -88,4 +89,4 @@ spec:
                         minLength: 1
                       replyURI:
                         type: string
-                        minLength: 1
+                        minLength: 1

pkg/apis/duck/v1alpha1/subscribable_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"github.com/knative/pkg/apis"
 	"github.com/knative/pkg/apis/duck"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -34,6 +35,9 @@ type Subscribable struct {
 	Subscribers []ChannelSubscriberSpec `json:"subscribers,omitempty" patchStrategy:"merge" patchMergeKey:"uid"`
 }
 
+// Subscribable is an Implementable "duck type".
+var _ duck.Implementable = (*Subscribable)(nil)
+
 // ChannelSubscriberSpec defines a single subscriber to a Channel.
 // Ref is a reference to the Subscription this ChannelSubscriberSpec was created for
 // SubscriberURI is the endpoint for the subscriber
@@ -52,6 +56,9 @@ type ChannelSubscriberSpec struct {
 	ReplyURI string `json:"replyURI,omitempty"`
 }
 
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
 // Channel is a skeleton type wrapping Subscribable in the manner we expect resource writers
 // defining compatible resources to embed it. We will typically use this type to deserialize
 // Channel ObjectReferences and access the Subscription data.  This is not a real resource.
@@ -69,6 +76,12 @@ type ChannelSpec struct {
 	Subscribable *Subscribable `json:"subscribable,omitempty"`
 }
 
+var (
+	// Verify Channel resources meet duck contracts.
+	_ duck.Populatable = (*Channel)(nil)
+	_ apis.Listable    = (*Channel)(nil)
+)
+
 // GetFullType implements duck.Implementable
 func (s *Subscribable) GetFullType() duck.Populatable {
 	return &Channel{}

pkg/apis/eventing/v1alpha1/subscribable_channelable_validation.go

@@ -29,25 +29,9 @@ func isChannelEmpty(f corev1.ObjectReference) bool {
 	return equality.Semantic.DeepEqual(f, corev1.ObjectReference{})
 }
 
-// Valid from only contains the following fields:
-// - Kind       == 'Channel'
-// - APIVersion == 'eventing.knative.dev/v1alpha1'
-// - Name       == not empty
+// Valid if it is a valid object reference.
 func isValidChannel(f corev1.ObjectReference) *apis.FieldError {
-	errs := isValidObjectReference(f)
-
-	if f.Kind != "Channel" {
-		fe := apis.ErrInvalidValue(f.Kind, "kind")
-		fe.Paths = []string{"kind"}
-		fe.Details = "only 'Channel' kind is allowed"
-		errs = errs.Also(fe)
-	}
-	if f.APIVersion != "eventing.knative.dev/v1alpha1" {
-		fe := apis.ErrInvalidValue(f.APIVersion, "apiVersion")
-		fe.Details = "only eventing.knative.dev/v1alpha1 is allowed for apiVersion"
-		errs = errs.Also(fe)
-	}
-	return errs
+	return isValidObjectReference(f)
 }
 
 func isValidObjectReference(f corev1.ObjectReference) *apis.FieldError {

pkg/apis/eventing/v1alpha1/subscribable_channelable_validation_test.go

@@ -30,40 +30,22 @@ var validationTests = []struct {
 	want *apis.FieldError
 }{
 	{
-		name: "invalid kind",
+		name: "valid object ref",
 		ref: corev1.ObjectReference{
 			Name:       "boaty-mcboatface",
 			APIVersion: "eventing.knative.dev/v1alpha1",
-			Kind:       "Strait",
-		},
-		want: &apis.FieldError{
-			Message: "invalid value: Strait",
-			Paths:   []string{"kind"},
-			Details: "only 'Channel' kind is allowed",
-		},
-	},
-	{
-		name: "invalid api version",
-		ref: corev1.ObjectReference{
-			Name:       "boaty-mcboatface",
-			APIVersion: "eventing.knative.dev/v1alpha2",
-			Kind:       "Channel",
-		},
-		want: &apis.FieldError{
-			Message: `invalid value: eventing.knative.dev/v1alpha2`,
-			Paths:   []string{"apiVersion"},
-			Details: "only eventing.knative.dev/v1alpha1 " +
-				"is allowed for apiVersion",
+			Kind:       "MyChannel",
 		},
+		want: nil,
 	},
 	{
-		name: "valid channel",
+		name: "invalid object ref",
 		ref: corev1.ObjectReference{
 			Name:       "boaty-mcboatface",
 			APIVersion: "eventing.knative.dev/v1alpha1",
-			Kind:       "Channel",
+			Kind:       "",
 		},
-		want: nil,
+		want: apis.ErrMissingField("kind"),
 	},
 }
 

pkg/apis/eventing/v1alpha1/subscription_types.go

@@ -77,9 +77,10 @@ type SubscriptionSpec struct {
 	//   - Kind
 	//   - APIVersion
 	//   - Name
-	// Kind must be "Channel" and APIVersion must be
-	// "eventing.knative.dev/v1alpha1"
-	//
+	//  The resource pointed by this ObjectReference must meet the Subscribable contract
+	//  with a pointer to the Subscribable duck type. If the resource does not meet this contract,
+	//  it will be reflected in the Subscription's status.
+
 	// This field is immutable. We have no good answer on what happens to
 	// the events that are currently in the channel being consumed from
 	// and what the semantics there should be. For now, you can always
@@ -154,14 +155,13 @@ type SubscriberSpec struct {
 // ReplyStrategy specifies the handling of the SubscriberSpec's returned replies.
 // If no SubscriberSpec is specified, the identity function is assumed.
 type ReplyStrategy struct {
-	// This object must be a Channel.
-	//
 	// You can specify only the following fields of the ObjectReference:
 	//   - Kind
 	//   - APIVersion
 	//   - Name
-	// Kind must be "Channel" and APIVersion must be
-	// "eventing.knative.dev/v1alpha1"
+	//  The resource pointed by this ObjectReference must meet the Addressable contract
+	//  with a reference to the Addressable duck type. If the resource does not meet this contract,
+	//  it will be reflected in the Subscription's status.
 	// +optional
 	Channel *corev1.ObjectReference `json:"channel,omitempty"`
 }

pkg/apis/eventing/v1alpha1/subscription_validation.go

@@ -115,17 +115,6 @@ func isValidReply(r ReplyStrategy) *apis.FieldError {
 	if fe := isValidObjectReference(*r.Channel); fe != nil {
 		return fe.ViaField("channel")
 	}
-	if r.Channel.Kind != "Channel" {
-		fe := apis.ErrInvalidValue(r.Channel.Kind, "kind")
-		fe.Paths = []string{"kind"}
-		fe.Details = "only 'Channel' kind is allowed"
-		return fe
-	}
-	if r.Channel.APIVersion != "eventing.knative.dev/v1alpha1" {
-		fe := apis.ErrInvalidValue(r.Channel.APIVersion, "apiVersion")
-		fe.Details = "only eventing.knative.dev/v1alpha1 is allowed for apiVersion"
-		return fe
-	}
 	return nil
 }