Trigger status depends on importer status

config/200-source-resolvers-clusterrole.yaml

@@ -0,0 +1,79 @@
+# 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.
+
+# Use this aggregated ClusterRole when you need readonly access to "Sources"
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: source-resolver
+  labels:
+    eventing.knative.dev/release: devel
+aggregationRule:
+  clusterRoleSelectors:
+  - matchLabels:
+      duck.knative.dev/sourceObserver: "true"
+rules: [] # Rules are automatically filled in by the controller manager.
+
+---
+
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: eventing-source-resolver
+  labels:
+    eventing.knative.dev/release: devel
+    duck.knative.dev/sourceObserver: "true"
+# Do not use this role directly. These rules will be added to the "source-resolver" role.
+# For resources "awssqssources", "camelsources", "kafkasources", "githubsources",
+# These should eventually be moved to config directories in https://github.com/knative/eventing-contrib.
+# TODO After https://github.com/knative/eventing-contrib/issues/583 gets done, remove "awssqssources", "camelsources", "kafkasources", "githubsources" in resources and modified config/201-clusterrolebinding.yaml
+rules:
+- apiGroups:
+  - "sources.eventing.knative.dev"
+  resources:
+    - "cronjobsources"
+    - "containersources"
+    - "apiserversources"
+    - "awssqssources"
+    - "camelsources"
+    - "kafkasources"
+    - "githubsources"
+  verbs:
+  - get
+  - list
+  - watch
+---
+
+# These should eventually be moved to config directories in https://github.com/google/knative-gcp.
+# TODO After https://github.com/google/knative-gcp/issues/252 gets done, remove this ClusterRole and modified config/201-clusterrolebinding.yaml
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: knative-gcp-source-resolver
+  labels:
+    eventing.knative.dev/release: devel
+    duck.knative.dev/sourceObserver: "true"
+# Do not use this role directly. These rules will be added to the "source-resolver" role.
+rules:
+  - apiGroups:
+      - "pubsub.cloud.run"
+      - "events.cloud.run"
+    resources:
+      - "pullsubscriptions"
+      - "storages"
+    verbs:
+      - get
+      - list
+      - watch
+---

config/201-clusterrolebinding.yaml

@@ -46,6 +46,23 @@ roleRef:
 
 ---
 
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: eventing-controller-source-resolver
+  labels:
+    eventing.knative.dev/release: devel
+subjects:
+  - kind: ServiceAccount
+    name: eventing-controller
+    namespace: knative-eventing
+roleRef:
+  kind: ClusterRole
+  name: source-resolver
+  apiGroup: rbac.authorization.k8s.io
+
+---
+
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:

pkg/apis/eventing/v1alpha1/trigger_lifecycle.go

@@ -17,11 +17,12 @@
 package v1alpha1
 
 import (
+	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
 	messagingv1alpha1 "knative.dev/eventing/pkg/apis/messaging/v1alpha1"
 	"knative.dev/pkg/apis"
 )
 
-var triggerCondSet = apis.NewLivingConditionSet(TriggerConditionBroker, TriggerConditionSubscribed)
+var triggerCondSet = apis.NewLivingConditionSet(TriggerConditionBroker, TriggerConditionSubscribed, TriggerConditionDependency)
 
 const (
 	// TriggerConditionReady has status True when all subconditions below have been set to True.
@@ -31,6 +32,8 @@ const (
 
 	TriggerConditionSubscribed apis.ConditionType = "Subscribed"
 
+	TriggerConditionDependency apis.ConditionType = "Dependency"
+
 	// TriggerAnyFilter Constant to represent that we should allow anything.
 	TriggerAnyFilter = ""
 )
@@ -85,3 +88,28 @@ func (ts *TriggerStatus) MarkNotSubscribed(reason, messageFormat string, message
 func (ts *TriggerStatus) MarkSubscriptionNotOwned(sub *messagingv1alpha1.Subscription) {
 	triggerCondSet.Manage(ts).MarkFalse(TriggerConditionSubscribed, "SubscriptionNotOwned", "Subscription %q is not owned by this Trigger.", sub.Name)
 }
+
+func (ts *TriggerStatus) MarkDependencySucceeded() {
+	triggerCondSet.Manage(ts).MarkTrue(TriggerConditionDependency)
+}
+
+func (ts *TriggerStatus) MarkDependencyFailed(reason, messageFormat string, messageA ...interface{}) {
+	triggerCondSet.Manage(ts).MarkFalse(TriggerConditionDependency, reason, messageFormat, messageA...)
+}
+
+func (ts *TriggerStatus) MarkDependencyUnknown(reason, messageFormat string, messageA ...interface{}) {
+	triggerCondSet.Manage(ts).MarkUnknown(TriggerConditionDependency, reason, messageFormat, messageA...)
+}
+
+func (ts *TriggerStatus) PropagateDependencyStatus(ks *duckv1alpha1.KResource) {
+	kc := ks.Status.GetCondition(duckv1alpha1.ConditionReady)
+	if kc != nil && kc.IsTrue() {
+		ts.MarkDependencySucceeded()
+	} else {
+		msg := "nil"
+		if kc != nil {
+			msg = kc.Message
+		}
+		ts.MarkDependencyFailed("DependencyNotReady", "Dependency is not ready: %s", msg)
+	}
+}

pkg/apis/eventing/v1alpha1/trigger_lifecycle_test.go

@@ -41,6 +41,11 @@ var (
 		Status: corev1.ConditionTrue,
 	}
 
+	triggerConditionDependency = apis.Condition{
+		Type:   TriggerConditionDependency,
+		Status: corev1.ConditionTrue,
+	}
+
 	triggerConditionSubscribed = apis.Condition{
 		Type:   TriggerConditionSubscribed,
 		Status: corev1.ConditionFalse,
@@ -71,6 +76,7 @@ func TestTriggerGetCondition(t *testing.T) {
 				Conditions: []apis.Condition{
 					triggerConditionBroker,
 					triggerConditionSubscribed,
+					triggerConditionDependency,
 				},
 			},
 		},
@@ -83,6 +89,7 @@ func TestTriggerGetCondition(t *testing.T) {
 				Conditions: []apis.Condition{
 					triggerConditionBroker,
 					triggerConditionSubscribed,
+					triggerConditionDependency,
 				},
 			},
 		},
@@ -124,13 +131,17 @@ func TestTriggerInitializeConditions(t *testing.T) {
 				Conditions: []apis.Condition{{
 					Type:   TriggerConditionBroker,
 					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionDependency,
+					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionReady,
 					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionSubscribed,
 					Status: corev1.ConditionUnknown,
-				}},
+				},
+				},
 			},
 		},
 	}, {
@@ -149,12 +160,17 @@ func TestTriggerInitializeConditions(t *testing.T) {
 					Type:   TriggerConditionBroker,
 					Status: corev1.ConditionFalse,
 				}, {
-					Type:   TriggerConditionReady,
-					Status: corev1.ConditionUnknown,
-				}, {
-					Type:   TriggerConditionSubscribed,
+					Type:   TriggerConditionDependency,
 					Status: corev1.ConditionUnknown,
-				}},
+				},
+					{
+						Type:   TriggerConditionReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   TriggerConditionSubscribed,
+						Status: corev1.ConditionUnknown,
+					},
+				},
 			},
 		},
 	}, {
@@ -172,13 +188,17 @@ func TestTriggerInitializeConditions(t *testing.T) {
 				Conditions: []apis.Condition{{
 					Type:   TriggerConditionBroker,
 					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionDependency,
+					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionReady,
 					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionSubscribed,
 					Status: corev1.ConditionTrue,
-				}},
+				},
+				},
 			},
 		},
 	}}
@@ -201,6 +221,8 @@ func TestTriggerIsReady(t *testing.T) {
 		markVirtualServiceExists    bool
 		subscriptionOwned           bool
 		subscriptionStatus          *messagingv1alpha1.SubscriptionStatus
+		dependencyAnnotationExists  bool
+		dependencyStatusReady       bool
 		wantReady                   bool
 	}{{
 		name:                        "all happy",
@@ -209,6 +231,7 @@ func TestTriggerIsReady(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionOwned:           true,
 		subscriptionStatus:          TestHelper.ReadySubscriptionStatus(),
+		dependencyAnnotationExists:  false,
 		wantReady:                   true,
 	}, {
 		name:                        "broker sad",
@@ -217,6 +240,7 @@ func TestTriggerIsReady(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionOwned:           true,
 		subscriptionStatus:          TestHelper.ReadySubscriptionStatus(),
+		dependencyAnnotationExists:  false,
 		wantReady:                   false,
 	}, {
 		name:                        "subscribed sad",
@@ -225,6 +249,7 @@ func TestTriggerIsReady(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionOwned:           true,
 		subscriptionStatus:          TestHelper.NotReadySubscriptionStatus(),
+		dependencyAnnotationExists:  false,
 		wantReady:                   false,
 	}, {
 		name:                        "subscription not owned",
@@ -233,16 +258,30 @@ func TestTriggerIsReady(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionOwned:           false,
 		subscriptionStatus:          TestHelper.ReadySubscriptionStatus(),
+		dependencyAnnotationExists:  false,
 		wantReady:                   false,
 	}, {
-		name:                        "all sad",
-		brokerStatus:                TestHelper.NotReadyBrokerStatus(),
-		markKubernetesServiceExists: false,
-		markVirtualServiceExists:    false,
-		subscriptionOwned:           false,
-		subscriptionStatus:          TestHelper.NotReadySubscriptionStatus(),
+		name:                        "dependency not ready",
+		brokerStatus:                TestHelper.ReadyBrokerStatus(),
+		markKubernetesServiceExists: true,
+		markVirtualServiceExists:    true,
+		subscriptionOwned:           true,
+		subscriptionStatus:          TestHelper.ReadySubscriptionStatus(),
+		dependencyAnnotationExists:  true,
+		dependencyStatusReady:       false,
 		wantReady:                   false,
-	}}
+	},
+		{
+			name:                        "all sad",
+			brokerStatus:                TestHelper.NotReadyBrokerStatus(),
+			markKubernetesServiceExists: false,
+			markVirtualServiceExists:    false,
+			subscriptionOwned:           false,
+			subscriptionStatus:          TestHelper.NotReadySubscriptionStatus(),
+			dependencyAnnotationExists:  true,
+			dependencyStatusReady:       false,
+			wantReady:                   false,
+		}}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			ts := &TriggerStatus{}
@@ -254,6 +293,11 @@ func TestTriggerIsReady(t *testing.T) {
 			} else if test.subscriptionStatus != nil {
 				ts.PropagateSubscriptionStatus(test.subscriptionStatus)
 			}
+			if test.dependencyAnnotationExists && !test.dependencyStatusReady {
+				ts.MarkDependencyFailed("Dependency is not ready", "Dependency is not ready")
+			} else {
+				ts.MarkDependencySucceeded()
+			}
 			got := ts.IsReady()
 			if test.wantReady != got {
 				t.Errorf("unexpected readiness: want %v, got %v", test.wantReady, got)

pkg/apis/eventing/v1alpha1/trigger_types.go

@@ -27,6 +27,12 @@ import (
 	"knative.dev/pkg/webhook"
 )
 
+const (
+	// DependencyAnnotation is the annotation key used to mark the sources that the Trigger depends on.
+	// This will be used when the kn client creates an importer and trigger pair for the user such that the trigger only receives events produced by the paired importer.
+	DependencyAnnotation = "knative.dev/dependency"
+)
+
 // +genclient
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object