Broker, Trigger, and Namespace Controllers

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,85 @@
+/*
+ * 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 main
+
+import (
+	"flag"
+	"log"
+	"os"
+
+	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"
+)
+
+const (
+	NAMESPACE = "NAMESPACE"
+)
+
+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{
+		Namespace: getRequiredEnv(NAMESPACE),
+	})
+	if err != nil {
+		logger.Fatal("Error starting up.", zap.Error(err))
+	}
+
+	if err = eventingv1alpha1.AddToScheme(mgr.GetScheme()); err != nil {
+		logger.Fatal("Unable to add eventingv1alpha1 scheme", zap.Error(err))
+	}
+
+	// We are running both the receiver (takes messages in from the Broker) and the dispatcher (send
+	// the messages to the triggers' subscribers) 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))
+	}
+	logger.Info("Exiting...")
+}
+
+func getRequiredEnv(envKey string) string {
+	val, defined := os.LookupEnv(envKey)
+	if !defined {
+		log.Fatalf("required environment variable not defined '%s'", envKey)
+	}
+	return val
+}

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,158 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	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 (
+	port = 8080
+
+	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))
+	}
+
+	if err = eventingv1alpha1.AddToScheme(mgr.GetScheme()); err != nil {
+		logger.Fatal("Unable to add eventingv1alpha1 scheme", zap.Error(err))
+	}
+
+	c := getRequiredEnv("CHANNEL")
+
+	h := NewHandler(logger, c)
+
+	s := &http.Server{
+		Addr:         fmt.Sprintf(":%d", port),
+		Handler:      h,
+		ErrorLog:     zap.NewStdLog(logger),
+		ReadTimeout:  readTimeout,
+		WriteTimeout: writeTimeout,
+	}
+
+	err = mgr.Add(&runnableServer{
+		logger: logger,
+		s:      s,
+	})
+	if err != nil {
+		logger.Fatal("Unable to add runnableServer", zap.Error(err))
+	}
+
+	// Set up signals so we handle the first shutdown signal gracefully.
+	stopCh := signals.SetupSignalHandler()
+	// Start blocks forever.
+	if err = mgr.Start(stopCh); err != nil {
+		logger.Error("manager.Start() returned an error", zap.Error(err))
+	}
+	logger.Info("Exiting...")
+
+	ctx, cancel := context.WithTimeout(context.Background(), writeTimeout)
+	defer cancel()
+	if err = s.Shutdown(ctx); err != nil {
+		logger.Error("Shutdown returned an error", zap.Error(err))
+	}
+}
+
+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 sends it out to a single destination.
+type Handler struct {
+	receiver    *provisioners.MessageReceiver
+	dispatcher  *provisioners.MessageDispatcher
+	destination string
+
+	logger *zap.Logger
+}
+
+// NewHandler creates a new ingress.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)
+	}
+}
+
+// http.Handler interface.
+func (f *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	f.receiver.HandleRequest(w, r)
+}
+
+// dispatch takes the request, and sends it out the f.destination. If the dispatched
+// request returns 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
+}
+
+// runnableServer is a small wrapper around http.Server so that it matches the manager.Runnable
+// interface.
+type runnableServer struct {
+	logger *zap.Logger
+	s      *http.Server
+}
+
+func (r *runnableServer) Start(<-chan struct{}) error {
+	r.logger.Info("Ingress Listening...", zap.String("Address", r.s.Addr))
+	return r.s.ListenAndServe()
+}

cmd/controller/main.go

@@ -21,10 +21,14 @@ import (
 	"flag"
 	"log"
 	"net/http"
+	"os"
 	"time"
 
+	"github.com/knative/eventing/pkg/reconciler/v1alpha1/broker"
 	"github.com/knative/eventing/pkg/reconciler/v1alpha1/channel"
+	"github.com/knative/eventing/pkg/reconciler/v1alpha1/namespace"
 	"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"
@@ -111,7 +115,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)
 		}
 	}
@@ -121,17 +125,26 @@ func main() {
 	providers := []ProvideFunc{
 		subscription.ProvideController,
 		channel.ProvideController,
+		broker.ProvideController(
+			broker.ReconcilerArgs{
+				IngressImage:              getRequiredEnv("BROKER_INGRESS_IMAGE"),
+				IngressServiceAccountName: getRequiredEnv("BROKER_INGRESS_SERVICE_ACCOUNT"),
+				FilterImage:               getRequiredEnv("BROKER_FILTER_IMAGE"),
+				FilterServiceAccountName:  getRequiredEnv("BROKER_FILTER_SERVICE_ACCOUNT"),
+			}),
+		trigger.ProvideController,
+		namespace.ProvideController,
 	}
 	for _, provider := range providers {
-		if _, err := provider(mgr, logger.Desugar()); err != nil {
+		if _, err = provider(mgr, logger.Desugar()); err != nil {
 			logger.Fatalf("Error adding controller to manager: %v", err)
 		}
 	}
 
 	// Start the Manager
 	go func() {
-		if err := mgr.Start(stopCh); err != nil {
-			logger.Fatalf("Error starting manager: %v", err)
+		if localErr := mgr.Start(stopCh); localErr != nil {
+			logger.Fatalf("Error starting manager: %v", localErr)
 		}
 	}()
 
@@ -140,8 +153,8 @@ func main() {
 	http.Handle(metricsScrapePath, promhttp.Handler())
 	go func() {
 		logger.Infof("Starting metrics listener at %s", metricsScrapeAddr)
-		if err := srv.ListenAndServe(); err != nil {
-			logger.Infof("Httpserver: ListenAndServe() finished with error: %s", err)
+		if localErr := srv.ListenAndServe(); localErr != nil {
+			logger.Infof("Httpserver: ListenAndServe() finished with error: %s", localErr)
 		}
 	}()
 
@@ -190,3 +203,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

@@ -53,19 +53,19 @@ func main() {
 	defer logger.Sync()
 	logger = logger.With(zap.String(logkey.ControllerType, logconfig.Webhook))
 
-	logger.Info("Starting the Eventing Webhook")
+	logger.Infow("Starting the Eventing Webhook")
 
 	// set up signals so we handle the first shutdown signal gracefully
 	stopCh := signals.SetupSignalHandler()
 
 	clusterConfig, err := rest.InClusterConfig()
 	if err != nil {
-		logger.Fatal("Failed to get in cluster config", zap.Error(err))
+		logger.Fatalw("Failed to get in cluster config", zap.Error(err))
 	}
 
 	kubeClient, err := kubernetes.NewForConfig(clusterConfig)
 	if err != nil {
-		logger.Fatal("Failed to get the client set", zap.Error(err))
+		logger.Fatalw("Failed to get the client set", zap.Error(err))
 	}
 
 	// Watch the logging config map and dynamically update logging levels.
@@ -96,14 +96,19 @@ 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))
+		logger.Fatalw("Failed to create the admission controller", zap.Error(err))
 	}
-	controller.Run(stopCh)
+	if err = controller.Run(stopCh); err != nil {
+		logger.Errorw("controller.Run() failed", zap.Error(err))
+	}
+	logger.Infow("Webhook stopping")
 }