Adds Kafka Channel Provisioner Controllers

config/provisioners/kafka/clusterprovisioner.yaml

@@ -0,0 +1,22 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+apiVersion: eventing.knative.dev/v1alpha1
+kind: ClusterProvisioner
+metadata:
+  name: kafka
+spec:
+  reconciles:
+    group: eventing.knative.dev
+    kind: Channel

config/provisioners/kafka/clusterrole.yaml

@@ -0,0 +1,28 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: kafka-provisioner
+rules:
+- apiGroups: ["eventing.knative.dev"]
+  resources: ["clusterprovisioners"]
+  verbs: ["get", "watch", "list", "update", "patch"]
+- apiGroups: ["eventing.knative.dev"]
+  resources: ["channels"]
+  verbs: ["get", "watch", "list", "update", "patch"]
+- apiGroups: [""]
+  resources: ["events"]
+  verbs: ["create", "patch"]

config/provisioners/kafka/clusterrolebinding.yaml

@@ -0,0 +1,26 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRoleBinding
+metadata:
+  name: kafka-provisioner-manage
+subjects:
+  - kind: ServiceAccount
+    name: kafka-provisioner
+    namespace: knative-eventing
+roleRef:
+  kind: ClusterRole
+  name: kafka-provisioner
+  apiGroup: rbac.authorization.k8s.io

config/provisioners/kafka/controller.yaml

@@ -0,0 +1,41 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+apiVersion: apps/v1beta1
+kind: Deployment
+metadata:
+  name: kafka-provisioner
+  namespace: knative-eventing
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: kafka-provisioner
+    spec:
+      serviceAccountName: kafka-provisioner
+      containers:
+      - name: kafka-provisioner-controller
+        image: github.com/knative/eventing/pkg/provisioners/kafka
+        args: [
+              "-logtostderr",
+              "-stderrthreshold", "INFO",
+        ]
+        volumeMounts:
+          - name: kafka-provisioner-config
+            mountPath: /etc/config-provisioner
+      volumes:
+        - name: kafka-provisioner-config
+          configMap:
+            name: kafka-provisioner-config

config/provisioners/kafka/kafka-provisioner-config.yaml

@@ -0,0 +1,24 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kafka-provisioner-config
+  namespace: knative-eventing
+data:
+  # Name of the provisioner that this controller represents
+  cluster-provisioner-name: kafka
+  # Broker URL's for the provisioner
+  brokers: kafkabroker.kafka:9092

config/provisioners/kafka/service-account.yaml

@@ -0,0 +1,19 @@
+# Copyright 2018 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: kafka-provisioner
+  namespace: knative-eventing

pkg/apis/eventing/v1alpha1/channel_types.go

@@ -117,6 +117,16 @@ func (cs *ChannelStatus) GetCondition(t duckv1alpha1.ConditionType) *duckv1alpha
 	return chanCondSet.Manage(cs).GetCondition(t)
 }
 
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (cs *ChannelStatus) InitializeConditions() {
+	chanCondSet.Manage(cs).InitializeConditions()
+}
+
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (cs *ChannelStatus) MarkAsNotProvisioned(reason, messageFormat string, messageA ...interface{}) {
+	chanCondSet.Manage(cs).MarkFalse(ChannelConditionProvisioned, reason, messageFormat, messageA...)
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // ChannelList is a collection of Channels.

pkg/apis/eventing/v1alpha1/channel_types_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -80,3 +81,89 @@ func TestChannelGetCondition(t *testing.T) {
 		})
 	}
 }
+
+func TestChannelInitializeConditions(t *testing.T) {
+	tests := []struct {
+		name string
+		cs   *ChannelStatus
+		want *ChannelStatus
+	}{{
+		name: "empty",
+		cs:   &ChannelStatus{},
+		want: &ChannelStatus{
+			Conditions: []duckv1alpha1.Condition{{
+				Type:   ChannelConditionProvisioned,
+				Status: corev1.ConditionUnknown,
+			}, {
+				Type:   ChannelConditionReady,
+				Status: corev1.ConditionUnknown,
+			}},
+		},
+	}, {
+		name: "one false",
+		cs: &ChannelStatus{
+			Conditions: []duckv1alpha1.Condition{{
+				Type:   ChannelConditionProvisioned,
+				Status: corev1.ConditionFalse,
+			}},
+		},
+		want: &ChannelStatus{
+			Conditions: []duckv1alpha1.Condition{{
+				Type:   ChannelConditionProvisioned,
+				Status: corev1.ConditionFalse,
+			}, {
+				Type:   ChannelConditionReady,
+				Status: corev1.ConditionUnknown,
+			}},
+		},
+	}, {
+		name: "one true",
+		cs: &ChannelStatus{
+			Conditions: []duckv1alpha1.Condition{{
+				Type:   ChannelConditionProvisioned,
+				Status: corev1.ConditionTrue,
+			}},
+		},
+		want: &ChannelStatus{
+			Conditions: []duckv1alpha1.Condition{{
+				Type:   ChannelConditionProvisioned,
+				Status: corev1.ConditionTrue,
+			}, {
+				Type:   ChannelConditionReady,
+				Status: corev1.ConditionUnknown,
+			}}},
+	},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			test.cs.InitializeConditions()
+			ignore := cmpopts.IgnoreFields(duckv1alpha1.Condition{}, "LastTransitionTime")
+			if diff := cmp.Diff(test.want, test.cs, ignore); diff != "" {
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestChannelStatus_MarkAsNotProvisioned(t *testing.T) {
+	cs := &ChannelStatus{}
+	cs.InitializeConditions()
+	want := &ChannelStatus{
+		Conditions: []duckv1alpha1.Condition{{
+			Type:    ChannelConditionProvisioned,
+			Status:  corev1.ConditionFalse,
+			Reason:  "Not Provisioned",
+			Message: "testing",
+		}, {
+			Type:    ChannelConditionReady,
+			Status:  corev1.ConditionFalse,
+			Reason:  "Not Provisioned",
+			Message: "testing",
+		}}}
+	ignore := cmpopts.IgnoreFields(duckv1alpha1.Condition{}, "LastTransitionTime")
+	cs.MarkAsNotProvisioned("Not Provisioned", "testing")
+	if diff := cmp.Diff(want, cs, ignore); diff != "" {
+		t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+	}
+}

pkg/apis/eventing/v1alpha1/cluster_provisioner_types.go

@@ -66,7 +66,7 @@ type ClusterProvisionerSpec struct {
 	Reconciles metav1.GroupKind `json:"reconciles"`
 }
 
-var cProvCondSet = duckv1alpha1.NewLivingConditionSet()
+var cProvCondSet = duckv1alpha1.NewLivingConditionSet(ClusterProvisionerConditionProvisionerReady)
 
 // ClusterProvisionerStatus is the status for a ClusterProvisioner resource
 type ClusterProvisionerStatus struct {
@@ -82,6 +82,15 @@ type ClusterProvisionerStatus struct {
 	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
 }
 
+const (
+
+	// ClusterProvisionerConditionReady has status True when all subconditions below have been set to True.
+	ClusterProvisionerConditionReady = duckv1alpha1.ConditionReady
+
+	// ClusterProvisionerConditionProvisionerReady has status True when the provisioner is ready
+	ClusterProvisionerConditionProvisionerReady duckv1alpha1.ConditionType = "ProvisionerReady"
+)
+
 // GetCondition returns the condition currently associated with the given type, or nil.
 func (ps *ClusterProvisionerStatus) GetCondition(t duckv1alpha1.ConditionType) *duckv1alpha1.Condition {
 	return cProvCondSet.Manage(ps).GetCondition(t)
@@ -92,6 +101,16 @@ func (ps *ClusterProvisionerStatus) IsReady() bool {
 	return cProvCondSet.Manage(ps).IsHappy()
 }
 
+// MarkProvisionerReady sets the condition that the provisioner is ready to provision backing resource.
+func (ps *ClusterProvisionerStatus) MarkProvisionerReady() {
+	cProvCondSet.Manage(ps).MarkTrue(ClusterProvisionerConditionProvisionerReady)
+}
+
+// MarkProvisionerNotReady sets the condition that the provisioner is not ready to provision backing resource.
+func (ps *ClusterProvisionerStatus) MarkProvisionerNotReady(reason, messageFormat string, messageA ...interface{}) {
+	cProvCondSet.Manage(ps).MarkFalse(ClusterProvisionerConditionProvisionerReady, reason, messageFormat, messageA...)
+}
+
 // InitializeConditions sets relevant unset conditions to Unknown state.
 func (ps *ClusterProvisionerStatus) InitializeConditions() {
 	cProvCondSet.Manage(ps).InitializeConditions()

pkg/apis/eventing/v1alpha1/cluster_provisioner_types_test.go

@@ -68,7 +68,26 @@ func TestClusterProvisionerStatusIsReady(t *testing.T) {
 			}},
 		},
 		want: false,
-	}}
+	}, {
+		name: "mark provisioner ready",
+		ps: func() *ClusterProvisionerStatus {
+			ps := &ClusterProvisionerStatus{}
+			ps.InitializeConditions()
+			ps.MarkProvisionerReady()
+			return ps
+		}(),
+		want: true,
+	},
+		{
+			name: "mark provisioner not ready",
+			ps: func() *ClusterProvisionerStatus {
+				ps := &ClusterProvisionerStatus{}
+				ps.InitializeConditions()
+				ps.MarkProvisionerNotReady("Not Ready", "testing")
+				return ps
+			}(),
+			want: false,
+		}}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {