Broker, Trigger, and Namespace Controllers

b.yaml

@@ -0,0 +1,11 @@
+apiVersion: eventing.knative.dev/v1alpha1
+kind: Broker
+metadata:
+  name: default
+spec:
+  channelTemplate:
+    provisioner:
+      apiVersion: eventing.knative.dev/v1alpha1
+      kind: ClusterChannelProvisioner
+      name: gcp-pubsub
+

b2.yaml

@@ -0,0 +1,5 @@
+apiVersion: eventing.knative.dev/v1alpha1
+kind: Broker
+metadata:
+  name: default
+spec: {}

cmd/broker/filter/kodata/HEAD

@@ -0,0 +1 @@
+../../../../.git/HEAD

cmd/broker/filter/kodata/LICENSE

@@ -0,0 +1 @@
+../../../../LICENSE

cmd/broker/filter/kodata/VENDOR-LICENSE

@@ -0,0 +1 @@
+../../../../third_party/VENDOR-LICENSE

cmd/broker/filter/main.go

@@ -0,0 +1,69 @@
+/*
+ * 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 main
+
+import (
+	"flag"
+
+	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
+	"github.com/knative/eventing/pkg/broker"
+	"github.com/knative/eventing/pkg/provisioners"
+	"github.com/knative/pkg/signals"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+func main() {
+	logConfig := provisioners.NewLoggingConfig()
+	logConfig.LoggingLevel["provisioner"] = zapcore.DebugLevel
+	logger := provisioners.NewProvisionerLoggerFromConfig(logConfig).Desugar()
+	defer logger.Sync()
+
+	flag.Parse()
+
+	logger.Info("Starting...")
+
+	mgr, err := manager.New(config.GetConfigOrDie(), manager.Options{})
+	if err != nil {
+		logger.Fatal("Error starting up.", zap.Error(err))
+	}
+
+	// Add custom types to this array to get them into the manager's scheme.
+	eventingv1alpha1.AddToScheme(mgr.GetScheme())
+
+	// We are running both the receiver (takes messages in from the cluster and writes them to
+	// PubSub) and the dispatcher (takes messages in PubSub and sends them in cluster) in this
+	// binary.
+
+	_, runnable := broker.New(logger, mgr.GetClient())
+	err = mgr.Add(runnable)
+	if err != nil {
+		logger.Fatal("Unable to start the receivers runnable", zap.Error(err), zap.Any("runnable", runnable))
+	}
+
+	// set up signals so we handle the first shutdown signal gracefully
+	stopCh := signals.SetupSignalHandler()
+
+	// Start blocks forever.
+	logger.Info("Manager starting...")
+	err = mgr.Start(stopCh)
+	if err != nil {
+		logger.Fatal("Manager.Start() returned an error", zap.Error(err))
+	}
+}

cmd/broker/ingress/kodata/HEAD

@@ -0,0 +1 @@
+../../../../.git/HEAD

cmd/broker/ingress/kodata/LICENSE

@@ -0,0 +1 @@
+../../../../LICENSE

cmd/broker/ingress/kodata/VENDOR-LICENSE

@@ -0,0 +1 @@
+../../../../third_party/VENDOR-LICENSE

cmd/broker/ingress/main.go

@@ -0,0 +1,144 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
+	"github.com/knative/eventing/pkg/provisioners"
+	"github.com/knative/pkg/signals"
+	"go.uber.org/zap"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+var (
+	readTimeout  = 1 * time.Minute
+	writeTimeout = 1 * time.Minute
+)
+
+func main() {
+	logConfig := provisioners.NewLoggingConfig()
+	logger := provisioners.NewProvisionerLoggerFromConfig(logConfig).Desugar()
+	defer logger.Sync()
+	flag.Parse()
+
+	logger.Info("Starting...")
+
+	mgr, err := manager.New(config.GetConfigOrDie(), manager.Options{})
+	if err != nil {
+		logger.Fatal("Error starting up.", zap.Error(err))
+	}
+
+	// Add custom types to this array to get them into the manager's scheme.
+	err = eventingv1alpha1.AddToScheme(mgr.GetScheme())
+	if err != nil {
+		logger.Fatal("Unable to add scheme", zap.Error(err))
+	}
+
+	c := getRequiredEnv("CHANNEL")
+
+	h := NewHandler(logger, c)
+
+	s := &http.Server{
+		Addr:         ":8080",
+		Handler:      h,
+		ErrorLog:     zap.NewStdLog(logger),
+		ReadTimeout:  readTimeout,
+		WriteTimeout: writeTimeout,
+	}
+
+	// Start both the manager (which notices ConfigMap changes) and the HTTP server.
+	var g errgroup.Group
+	g.Go(func() error {
+		// set up signals so we handle the first shutdown signal gracefully
+		stopCh := signals.SetupSignalHandler()
+		// Start blocks forever, so run it in a goroutine.
+		return mgr.Start(stopCh)
+	})
+	logger.Info("Ingress Listening...", zap.String("Address", s.Addr))
+	g.Go(s.ListenAndServe)
+	err = g.Wait()
+	if err != nil {
+		logger.Error("HTTP server failed.", zap.Error(err))
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), writeTimeout)
+	defer cancel()
+	s.Shutdown(ctx)
+}
+
+func getRequiredEnv(envKey string) string {
+	val, defined := os.LookupEnv(envKey)
+	if !defined {
+		log.Fatalf("required environment variable not defined '%s'", envKey)
+	}
+	return val
+}
+
+// http.Handler that takes a single request in and fans it out to N other servers.
+type Handler struct {
+	receiver    *provisioners.MessageReceiver
+	dispatcher  *provisioners.MessageDispatcher
+	destination string
+
+	logger *zap.Logger
+}
+
+// NewHandler creates a new fanout.Handler.
+func NewHandler(logger *zap.Logger, destination string) *Handler {
+	handler := &Handler{
+		logger:      logger,
+		dispatcher:  provisioners.NewMessageDispatcher(logger.Sugar()),
+		destination: fmt.Sprintf("http://%s", destination),
+	}
+	// The receiver function needs to point back at the handler itself, so set it up after
+	// initialization.
+	handler.receiver = provisioners.NewMessageReceiver(createReceiverFunction(handler), logger.Sugar())
+
+	return handler
+}
+
+func createReceiverFunction(f *Handler) func(provisioners.ChannelReference, *provisioners.Message) error {
+	return func(_ provisioners.ChannelReference, m *provisioners.Message) error {
+		// TODO Filter.
+		return f.dispatch(m)
+	}
+}
+
+func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	f.receiver.HandleRequest(w, r)
+}
+
+// dispatch takes the request, fans it out to each subscription in f.config. If all the fanned out
+// requests return successfully, then return nil. Else, return an error.
+func (f *Handler) dispatch(msg *provisioners.Message) error {
+	err := f.dispatcher.DispatchMessage(msg, f.destination, "", provisioners.DispatchDefaults{})
+	if err != nil {
+		f.logger.Error("Error dispatching message", zap.String("destination", f.destination))
+	}
+	return err
+}

cmd/controller/main.go

@@ -21,9 +21,13 @@ import (
 	"flag"
 	"log"
 	"net/http"
+	"os"
 	"time"
 
+	"github.com/knative/eventing/pkg/reconciler/v1alpha1/namespace"
+	"github.com/knative/eventing/pkg/reconciler/v1alpha1/broker"
 	"github.com/knative/eventing/pkg/reconciler/v1alpha1/subscription"
+	"github.com/knative/eventing/pkg/reconciler/v1alpha1/trigger"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -110,7 +114,7 @@ func main() {
 		eventingv1alpha1.AddToScheme,
 	}
 	for _, schemeFunc := range schemeFuncs {
-		if err := schemeFunc(mgr.GetScheme()); err != nil {
+		if err = schemeFunc(mgr.GetScheme()); err != nil {
 			logger.Fatalf("Error adding type to manager's scheme: %v", err)
 		}
 	}
@@ -119,6 +123,9 @@ func main() {
 	// manager run it.
 	providers := []ProvideFunc{
 		subscription.ProvideController,
+		broker.ProvideController(logger.Desugar(), getRequiredEnv("INGRESS_IMAGE"), getRequiredEnv("INGRESS_SERVICE_ACCOUNT"), getRequiredEnv("FILTER_IMAGE"), getRequiredEnv("FILTER_SERVICE_ACCOUNT")),
+		trigger.ProvideController(logger.Desugar()),
+		namespace.ProvideController(logger.Desugar()),
 	}
 	for _, provider := range providers {
 		if _, err := provider(mgr); err != nil {
@@ -188,3 +195,11 @@ func getLoggingConfigOrDie() map[string]string {
 		return cm
 	}
 }
+
+func getRequiredEnv(envKey string) string {
+	val, defined := os.LookupEnv(envKey)
+	if !defined {
+		log.Fatalf("required environment variable not defined '%s'", envKey)
+	}
+	return val
+}

cmd/webhook/main.go

@@ -96,14 +96,17 @@ func main() {
 		Options: options,
 		Handlers: map[schema.GroupVersionKind]webhook.GenericCRD{
 			// For group eventing.knative.dev,
+			eventingv1alpha1.SchemeGroupVersion.WithKind("Broker"):                    &eventingv1alpha1.Broker{},
 			eventingv1alpha1.SchemeGroupVersion.WithKind("Channel"):                   &eventingv1alpha1.Channel{},
 			eventingv1alpha1.SchemeGroupVersion.WithKind("ClusterChannelProvisioner"): &eventingv1alpha1.ClusterChannelProvisioner{},
 			eventingv1alpha1.SchemeGroupVersion.WithKind("Subscription"):              &eventingv1alpha1.Subscription{},
+			eventingv1alpha1.SchemeGroupVersion.WithKind("Trigger"):                   &eventingv1alpha1.Trigger{},
 		},
 		Logger: logger,
 	}
 	if err != nil {
 		logger.Fatal("Failed to create the admission controller", zap.Error(err))
 	}
-	controller.Run(stopCh)
+	err = controller.Run(stopCh)
+	logger.Errorw("Webhook stopping", zap.Error(err))
 }

config/300-broker.yaml

@@ -0,0 +1,38 @@
+# 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: brokers.eventing.knative.dev
+spec:
+  group: eventing.knative.dev
+  version: v1alpha1
+  names:
+    kind: Broker
+    plural: brokers
+    singular: broker
+    categories:
+    - all
+    - knative
+    - eventing
+  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"