Fix for #1109 [BUG] eventing controller and source-controller fall off main() on shutdown signal

cmd/controller/main.go

@@ -21,6 +21,8 @@ import (
 	"log"
 	"os"
 
+	"github.com/knative/eventing/pkg/reconciler/eventtype"
+
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/rest"
 
@@ -67,7 +69,7 @@ func main() {
 
 	logger.Info("Starting the controller")
 
-	const numControllers = 5
+	const numControllers = 6
 	cfg.QPS = numControllers * rest.DefaultQPS
 	cfg.Burst = numControllers * rest.DefaultBurst
 	opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh)
@@ -80,6 +82,7 @@ func main() {
 	channelInformer := eventingInformerFactory.Eventing().V1alpha1().Channels()
 	subscriptionInformer := eventingInformerFactory.Eventing().V1alpha1().Subscriptions()
 	brokerInformer := eventingInformerFactory.Eventing().V1alpha1().Brokers()
+	eventTypeInformer := eventingInformerFactory.Eventing().V1alpha1().EventTypes()
 
 	// Kube
 	serviceInformer := kubeInformerFactory.Core().V1().Services()
@@ -125,6 +128,11 @@ func main() {
 				FilterServiceAccountName:  getRequiredEnv("BROKER_FILTER_SERVICE_ACCOUNT"),
 			},
 		),
+		eventtype.NewController(
+			opt,
+			eventTypeInformer,
+			brokerInformer,
+		),
 	}
 	if len(controllers) != numControllers {
 		logger.Fatalf("Number of controllers and QPS settings mismatch: %d != %d", len(controllers), numControllers)
@@ -147,6 +155,7 @@ func main() {
 		channelInformer.Informer(),
 		subscriptionInformer.Informer(),
 		triggerInformer.Informer(),
+		eventTypeInformer.Informer(),
 		// Kube
 		configMapInformer.Informer(),
 		serviceInformer.Informer(),
@@ -158,8 +167,7 @@ func main() {
 
 	// Start all of the controllers.
 	logger.Info("Starting controllers.")
-	go kncontroller.StartAll(stopCh, controllers...)
-	<-stopCh
+	kncontroller.StartAll(stopCh, controllers...)
 }
 
 func init() {

cmd/sources-controller/main.go

@@ -121,9 +121,7 @@ func main() {
 
 	// Start all of the controllers.
 	logger.Info("Starting controllers.")
-	go kncontroller.StartAll(stopCh, controllers...)
-
-	<-stopCh
+	kncontroller.StartAll(stopCh, controllers...)
 }
 
 func setupLogger() (*zap.SugaredLogger, zap.AtomicLevel) {

cmd/webhook/main.go

@@ -101,6 +101,7 @@ func main() {
 			eventingv1alpha1.SchemeGroupVersion.WithKind("ClusterChannelProvisioner"): &eventingv1alpha1.ClusterChannelProvisioner{},
 			eventingv1alpha1.SchemeGroupVersion.WithKind("Subscription"):              &eventingv1alpha1.Subscription{},
 			eventingv1alpha1.SchemeGroupVersion.WithKind("Trigger"):                   &eventingv1alpha1.Trigger{},
+			eventingv1alpha1.SchemeGroupVersion.WithKind("EventType"):                 &eventingv1alpha1.EventType{},
 		},
 		Logger: logger,
 	}

config/200-controller-clusterrole.yaml

@@ -70,6 +70,8 @@ rules:
       - "subscriptions/status"
       - "triggers"
       - "triggers/status"
+      - "eventtypes"
+      - "eventtypes/status"
     verbs: *everything
 
   # Source resources and statuses we care about.

config/200-webhook-clusterrole.yaml

@@ -79,6 +79,8 @@ rules:
       - "subscriptions/status"
       - "triggers"
       - "triggers/status"
+      - "eventtypes"
+      - "eventtypes/status"
     verbs:
       - "get"
       - "list"

config/300-eventtype.yaml

@@ -0,0 +1,53 @@
+# 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: eventtypes.eventing.knative.dev
+spec:
+  group: eventing.knative.dev
+  version: v1alpha1
+  names:
+    kind: EventType
+    plural: eventtypes
+    singular: eventtype
+    categories:
+      - all
+      - knative
+      - eventing
+  scope: Namespaced
+  subresources:
+    status: {}
+  additionalPrinterColumns:
+    - name: Type
+      type: string
+      JSONPath: ".spec.type"
+    - name: Source
+      type: string
+      JSONPath: ".spec.source"
+    - name: Schema
+      type: string
+      JSONPath: ".spec.schema"
+    - name: Broker
+      type: string
+      JSONPath: ".spec.broker"
+    - name: Description
+      type: string
+      JSONPath: ".spec.description"
+    - name: Ready
+      type: string
+      JSONPath: ".status.conditions[?(@.type==\"Ready\")].status"
+    - name: Reason
+      type: string
+      JSONPath: ".status.conditions[?(@.type==\"Ready\")].reason"

docs/registry/example_eventtype.yaml

@@ -0,0 +1,49 @@
+# 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.
+
+---
+
+# This EventType creates an event type with type 'com.github.pull_request', source
+# 'github.com', for the 'default' Broker.
+
+apiVersion: eventing.knative.dev/v1alpha1
+kind: EventType
+metadata:
+  name: com.github.pullrequest
+spec:
+  type: com.github.pull_request
+  source: github.com
+  broker: default
+  description: "GitHub Pull Request"
+
+---
+
+# This Trigger matches all events of type 'com.github.pull_request' and source
+# 'github.com', that are sent to the 'default' Broker.
+
+apiVersion: eventing.knative.dev/v1alpha1
+kind: Trigger
+metadata:
+  name: filtering-event-type
+spec:
+  filter:
+    sourceAndType:
+      type: com.github.pull_request
+      source: github.com
+  broker: default
+  subscriber:
+    ref:
+      apiVersion: serving.knative.dev/v1alpha1
+      kind: Service
+      name: message-dumper

pkg/apis/eventing/v1alpha1/eventtype_defaults.go

@@ -0,0 +1,29 @@
+/*
+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
+
+    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 (et *EventType) SetDefaults(ctx context.Context) {
+	et.Spec.SetDefaults(ctx)
+}
+
+func (ets *EventTypeSpec) SetDefaults(ctx context.Context) {
+	if ets.Broker == "" {
+		ets.Broker = "default"
+	}
+}

pkg/apis/eventing/v1alpha1/eventtype_defaults_test.go

@@ -0,0 +1,83 @@
+/*
+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"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestEventTypeDefaults(t *testing.T) {
+	testCases := map[string]struct {
+		initial  EventType
+		expected EventType
+	}{
+		"nil spec": {
+			initial: EventType{},
+			expected: EventType{
+				Spec: EventTypeSpec{
+					Broker: "default",
+				},
+			},
+		},
+		"broker empty": {
+			initial: EventType{
+				Spec: EventTypeSpec{
+					Type:   "test-type",
+					Source: "test-source",
+					Broker: "",
+					Schema: "test-schema",
+				},
+			},
+			expected: EventType{
+				Spec: EventTypeSpec{
+					Type:   "test-type",
+					Source: "test-source",
+					Broker: "default",
+					Schema: "test-schema",
+				},
+			},
+		},
+		"broker not set": {
+			initial: EventType{
+				Spec: EventTypeSpec{
+					Type:   "test-type",
+					Source: "test-source",
+					Schema: "test-schema",
+				},
+			},
+			expected: EventType{
+				Spec: EventTypeSpec{
+					Type:   "test-type",
+					Source: "test-source",
+					Broker: "default",
+					Schema: "test-schema",
+				},
+			},
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			tc.initial.SetDefaults(context.TODO())
+			if diff := cmp.Diff(tc.expected, tc.initial); diff != "" {
+				t.Fatalf("Unexpected defaults (-want, +got): %s", diff)
+			}
+		})
+	}
+}

pkg/apis/eventing/v1alpha1/eventtype_lifecycle.go

@@ -0,0 +1,58 @@
+/*
+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 duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
+
+var eventTypeCondSet = duckv1alpha1.NewLivingConditionSet(EventTypeConditionBrokerExists, EventTypeConditionBrokerReady)
+
+const (
+	EventTypeConditionReady                                   = duckv1alpha1.ConditionReady
+	EventTypeConditionBrokerExists duckv1alpha1.ConditionType = "BrokerExists"
+	EventTypeConditionBrokerReady  duckv1alpha1.ConditionType = "BrokerReady"
+)
+
+// GetCondition returns the condition currently associated with the given type, or nil.
+func (et *EventTypeStatus) GetCondition(t duckv1alpha1.ConditionType) *duckv1alpha1.Condition {
+	return eventTypeCondSet.Manage(et).GetCondition(t)
+}
+
+// IsReady returns true if the resource is ready overall.
+func (et *EventTypeStatus) IsReady() bool {
+	return eventTypeCondSet.Manage(et).IsHappy()
+}
+
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (et *EventTypeStatus) InitializeConditions() {
+	eventTypeCondSet.Manage(et).InitializeConditions()
+}
+
+func (et *EventTypeStatus) MarkBrokerExists() {
+	eventTypeCondSet.Manage(et).MarkTrue(EventTypeConditionBrokerExists)
+}
+
+func (et *EventTypeStatus) MarkBrokerDoesNotExist() {
+	eventTypeCondSet.Manage(et).MarkFalse(EventTypeConditionBrokerExists, "BrokerDoesNotExist", "Broker does not exist")
+}
+
+func (et *EventTypeStatus) MarkBrokerReady() {
+	eventTypeCondSet.Manage(et).MarkTrue(EventTypeConditionBrokerReady)
+}
+
+func (et *EventTypeStatus) MarkBrokerNotReady() {
+	eventTypeCondSet.Manage(et).MarkFalse(EventTypeConditionBrokerReady, "BrokerNotReady", "Broker is not ready")
+}