Pipeline Implementation

cmd/controller/main.go

@@ -35,6 +35,7 @@ import (
 	"github.com/knative/eventing/pkg/reconciler/channel"
 	"github.com/knative/eventing/pkg/reconciler/eventtype"
 	"github.com/knative/eventing/pkg/reconciler/namespace"
+	"github.com/knative/eventing/pkg/reconciler/pipeline"
 	"github.com/knative/eventing/pkg/reconciler/subscription"
 	"github.com/knative/eventing/pkg/reconciler/trigger"
 	"github.com/knative/pkg/configmap"
@@ -81,7 +82,7 @@ func main() {
 
 	logger.Info("Starting the controller")
 
-	const numControllers = 6
+	const numControllers = 7
 	cfg.QPS = numControllers * rest.DefaultQPS
 	cfg.Burst = numControllers * rest.DefaultBurst
 	opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh)
@@ -97,6 +98,9 @@ func main() {
 	brokerInformer := eventingInformerFactory.Eventing().V1alpha1().Brokers()
 	eventTypeInformer := eventingInformerFactory.Eventing().V1alpha1().EventTypes()
 
+	// Messaging
+	pipelineInformer := eventingInformerFactory.Messaging().V1alpha1().Pipelines()
+
 	// Kube
 	serviceInformer := kubeInformerFactory.Core().V1().Services()
 	namespaceInformer := kubeInformerFactory.Core().V1().Namespaces()
@@ -165,6 +169,12 @@ func main() {
 			eventTypeInformer,
 			brokerInformer,
 		),
+		pipeline.NewController(
+			opt,
+			pipelineInformer,
+			addressableInformer,
+			subscriptionInformer,
+		),
 	}
 	// This line asserts at compile time that the length of controllers is equal to numControllers.
 	// It is based on https://go101.org/article/tips.html#assert-at-compile-time, which notes that
@@ -191,6 +201,7 @@ func main() {
 		subscriptionInformer.Informer(),
 		triggerInformer.Informer(),
 		eventTypeInformer.Informer(),
+		pipelineInformer.Informer(),
 		// Kube
 		configMapInformer.Informer(),
 		serviceInformer.Informer(),

config/200-controller-clusterrole.yaml

@@ -76,6 +76,16 @@ rules:
     verbs:
       - "update"
 
+  # Our own resources and statuses we care about.
+  - apiGroups:
+      - "messaging.knative.dev"
+    resources:
+      - "pipelines"
+      - "pipelines/status"
+      - "inmemorychannels"
+      - "inmemorychannels/status"
+    verbs: *everything
+
   # Source resources and statuses we care about.
   - apiGroups:
       - "sources.eventing.knative.dev"

config/300-pipeline.yaml

@@ -0,0 +1,96 @@
+# Copyright 2019 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.
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: pipelines.messaging.knative.dev
+  labels:
+    knative.dev/crd-install: "true"
+spec:
+  group: messaging.knative.dev
+  version: v1alpha1
+  names:
+    kind: Pipeline
+    plural: pipelines
+    singular: pipeline
+    categories:
+    - all
+    - knative
+    - eventing
+    - messaging
+  scope: Namespaced
+  subresources:
+    status: {}
+  additionalPrinterColumns:
+    - name: Ready
+      type: string
+      JSONPath: ".status.conditions[?(@.type==\"Ready\")].status"
+    - name: Reason
+      type: string
+      JSONPath: ".status.conditions[?(@.type==\"Ready\")].reason"
+    - name: Age
+      type: date
+      JSONPath: .metadata.creationTimestamp
+  validation:
+    openAPIV3Schema:
+      properties:
+        spec:
+          required:
+            - steps
+            - channelTemplate
+          properties:
+            steps:
+              type: array
+              items:
+                type: object
+                properties:
+                  dnsName:
+                    type: string
+                    minLength: 1
+                  uri:
+                    type: string
+                    minLength: 1
+                  ref:
+                    type: object
+                    required:
+                      - apiVersion
+                      - kind
+                      - name
+                    properties:
+                      apiVersion:
+                        type: string
+                        minLength: 1
+                      kind:
+                        type: string
+                        minLength: 1
+                      name:
+                        type: string
+                        minLength: 1
+                      namespace:
+                        type: string
+                        maxLength: 0
+            channelTemplate:
+              type: object
+              required:
+                - apiVersion
+                - kind
+              properties:
+                apiVersion:
+                  type: string
+                  minLength: 1
+                kind:
+                  type: string
+                  minLength: 1
+                spec:
+                  type: object

pkg/apis/eventing/v1alpha1/register.go

@@ -51,12 +51,12 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&ChannelList{},
 		&ClusterChannelProvisioner{},
 		&ClusterChannelProvisionerList{},
+		&EventType{},
+		&EventTypeList{},
 		&Subscription{},
 		&SubscriptionList{},
 		&Trigger{},
 		&TriggerList{},
-		&EventType{},
-		&EventTypeList{},
 	)
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil

pkg/apis/eventing/v1alpha1/subscribable_channelable_validation.go

@@ -31,10 +31,10 @@ func isChannelEmpty(f corev1.ObjectReference) bool {
 
 // Valid if it is a valid object reference.
 func isValidChannel(f corev1.ObjectReference) *apis.FieldError {
-	return isValidObjectReference(f)
+	return IsValidObjectReference(f)
 }
 
-func isValidObjectReference(f corev1.ObjectReference) *apis.FieldError {
+func IsValidObjectReference(f corev1.ObjectReference) *apis.FieldError {
 	return checkRequiredObjectReferenceFields(f).
 		Also(checkDisallowedObjectReferenceFields(f))
 }

pkg/apis/eventing/v1alpha1/subscribable_channelable_validation_test.go

@@ -141,7 +141,7 @@ func TestIsValidObjectReference(t *testing.T) {
 			for _, fe := range test.want {
 				allWanted = allWanted.Also(fe)
 			}
-			got := isValidObjectReference(test.ref)
+			got := IsValidObjectReference(test.ref)
 			if diff := cmp.Diff(allWanted.Error(), got.Error()); diff != "" {
 				t.Errorf("%s: validation (-want, +got) = %v", test.name, diff)
 			}

pkg/apis/eventing/v1alpha1/subscription_validation.go

@@ -52,7 +52,7 @@ func (ss *SubscriptionSpec) Validate(ctx context.Context) *apis.FieldError {
 	}
 
 	if !missingSubscriber {
-		if fe := isValidSubscriberSpec(*ss.Subscriber); fe != nil {
+		if fe := IsValidSubscriberSpec(*ss.Subscriber); fe != nil {
 			errs = errs.Also(fe.ViaField("subscriber"))
 		}
 	}
@@ -78,7 +78,7 @@ func isSubscriberSpecNilOrEmpty(s *SubscriberSpec) bool {
 	return false
 }
 
-func isValidSubscriberSpec(s SubscriberSpec) *apis.FieldError {
+func IsValidSubscriberSpec(s SubscriberSpec) *apis.FieldError {
 	var errs *apis.FieldError
 
 	fieldsSet := make([]string, 0, 0)
@@ -99,7 +99,7 @@ func isValidSubscriberSpec(s SubscriberSpec) *apis.FieldError {
 
 	// If Ref given, check the fields.
 	if s.Ref != nil && !equality.Semantic.DeepEqual(s.Ref, &corev1.ObjectReference{}) {
-		fe := isValidObjectReference(*s.Ref)
+		fe := IsValidObjectReference(*s.Ref)
 		if fe != nil {
 			errs = errs.Also(fe.ViaField("ref"))
 		}
@@ -112,7 +112,7 @@ func isReplyStrategyNilOrEmpty(r *ReplyStrategy) bool {
 }
 
 func isValidReply(r ReplyStrategy) *apis.FieldError {
-	if fe := isValidObjectReference(*r.Channel); fe != nil {
+	if fe := IsValidObjectReference(*r.Channel); fe != nil {
 		return fe.ViaField("channel")
 	}
 	return nil

pkg/apis/eventing/v1alpha1/subscription_validation_test.go

@@ -625,7 +625,7 @@ func TestValidgetValidSubscriber(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got := isValidSubscriberSpec(test.s)
+			got := IsValidSubscriberSpec(test.s)
 			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
 				t.Errorf("%s: isValidSubscriber (-want, +got) = %v", test.name, diff)
 			}

pkg/apis/eventing/v1alpha1/trigger_validation.go

@@ -47,7 +47,7 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	if isSubscriberSpecNilOrEmpty(ts.Subscriber) {
 		fe := apis.ErrMissingField("subscriber")
 		errs = errs.Also(fe)
-	} else if fe := isValidSubscriberSpec(*ts.Subscriber); fe != nil {
+	} else if fe := IsValidSubscriberSpec(*ts.Subscriber); fe != nil {
 		errs = errs.Also(fe.ViaField("subscriber"))
 	}
 

pkg/apis/messaging/v1alpha1/pipeline_defaults.go

@@ -0,0 +1,27 @@
+/*
+Copyright 2019 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 v1alpha1
+
+import "context"
+
+func (s *Pipeline) SetDefaults(ctx context.Context) {
+	s.Spec.SetDefaults(ctx)
+}
+
+func (ss *PipelineSpec) SetDefaults(ctx context.Context) {
+	// TODO anything?
+}