Broker controller new infrastructure

cmd/controller/main.go

@@ -17,22 +17,16 @@ limitations under the License.
 package main
 
 import (
-	"context"
 	"flag"
 	"log"
-	"net/http"
 	"os"
-	"time"
 
-	"k8s.io/apimachinery/pkg/runtime"
 	kubeinformers "k8s.io/client-go/informers"
-	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
-	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	informers "github.com/knative/eventing/pkg/client/informers/externalversions"
 	"github.com/knative/eventing/pkg/logconfig"
 	"github.com/knative/eventing/pkg/logging"
@@ -42,35 +36,19 @@ import (
 	"github.com/knative/eventing/pkg/reconciler/subscription"
 	"github.com/knative/eventing/pkg/reconciler/trigger"
 	"github.com/knative/eventing/pkg/reconciler/v1alpha1/broker"
-	istiov1alpha3 "github.com/knative/pkg/apis/istio/v1alpha3"
 	"github.com/knative/pkg/configmap"
 	kncontroller "github.com/knative/pkg/controller"
 	"github.com/knative/pkg/logging/logkey"
 	"github.com/knative/pkg/signals"
-	"github.com/knative/pkg/system"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 	controllerruntime "sigs.k8s.io/controller-runtime/pkg/client/config"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 )
 
-const (
-	metricsScrapeAddr = ":9090"
-	metricsScrapePath = "/metrics"
-)
-
 var (
 	hardcodedLoggingConfig bool
 )
 
-// SchemeFunc adds types to a Scheme.
-type SchemeFunc func(*runtime.Scheme) error
-
-// ProvideFunc adds a controller to a Manager.
-type ProvideFunc func(manager.Manager, *zap.Logger) (controller.Controller, error)
-
 func main() {
 	flag.Parse()
 	logf.SetLogger(logf.ZapLogger(false))
@@ -87,16 +65,9 @@ func main() {
 		logger.Fatalf("Error building kubeconfig: %v", err)
 	}
 
-	go startPkgController(stopCh, cfg, logger, atomicLevel)
-	go startControllerRuntime(stopCh, cfg, logger, atomicLevel)
-	<-stopCh
-}
-
-func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.SugaredLogger, atomicLevel zap.AtomicLevel) {
-	logger = logger.With(zap.String("controller/impl", "pkg"))
 	logger.Info("Starting the controller")
 
-	const numControllers = 4
+	const numControllers = 5
 	cfg.QPS = numControllers * rest.DefaultQPS
 	cfg.Burst = numControllers * rest.DefaultBurst
 	opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh)
@@ -114,6 +85,7 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 	serviceInformer := kubeInformerFactory.Core().V1().Services()
 	namespaceInformer := kubeInformerFactory.Core().V1().Namespaces()
 	configMapInformer := kubeInformerFactory.Core().V1().ConfigMaps()
+	deploymentInformer := kubeInformerFactory.Apps().V1().Deployments()
 
 	// Build all of our controllers, with the clients constructed above.
 	// Add new controllers to this array.
@@ -139,6 +111,20 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 			brokerInformer,
 			serviceInformer,
 		),
+		broker.NewController(
+			opt,
+			brokerInformer,
+			subscriptionInformer,
+			channelInformer,
+			serviceInformer,
+			deploymentInformer,
+			broker.ReconcilerArgs{
+				IngressImage:              getRequiredEnv("BROKER_INGRESS_IMAGE"),
+				IngressServiceAccountName: getRequiredEnv("BROKER_INGRESS_SERVICE_ACCOUNT"),
+				FilterImage:               getRequiredEnv("BROKER_FILTER_IMAGE"),
+				FilterServiceAccountName:  getRequiredEnv("BROKER_FILTER_SERVICE_ACCOUNT"),
+			},
+		),
 	}
 	if len(controllers) != numControllers {
 		logger.Fatalf("Number of controllers and QPS settings mismatch: %d != %d", len(controllers), numControllers)
@@ -165,89 +151,15 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 		configMapInformer.Informer(),
 		serviceInformer.Informer(),
 		namespaceInformer.Informer(),
+		deploymentInformer.Informer(),
 	); err != nil {
 		logger.Fatalf("Failed to start informers: %v", err)
 	}
 
 	// Start all of the controllers.
 	logger.Info("Starting controllers.")
 	go kncontroller.StartAll(stopCh, controllers...)
-}
-
-// TODO: remove after done integrating all controllers.
-func startControllerRuntime(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.SugaredLogger, atomicLevel zap.AtomicLevel) {
-	logger = logger.With(zap.String("controller/impl", "cr"))
-	logger.Info("Starting the controller")
-
-	kubeClient, err := kubernetes.NewForConfig(cfg)
-	if err != nil {
-		logger.Fatalf("Error building kubernetes clientset: %v", err)
-	}
-
-	// Watch the logging config map and dynamically update logging levels.
-	configMapWatcher := configmap.NewInformedWatcher(kubeClient, system.Namespace())
-	configMapWatcher.Watch(logconfig.ConfigMapName(), logging.UpdateLevelFromConfigMap(logger, atomicLevel, logconfig.Controller))
-	if err = configMapWatcher.Start(stopCh); err != nil {
-		logger.Fatalf("Failed to start controller config map watcher: %v", err)
-	}
-
-	// Setup a Manager
-	mgr, err := manager.New(cfg, manager.Options{})
-	if err != nil {
-		logger.Fatalf("Failed to create manager: %v", err)
-	}
-
-	// Add custom types to this array to get them into the manager's scheme.
-	schemeFuncs := []SchemeFunc{
-		istiov1alpha3.AddToScheme,
-		eventingv1alpha1.AddToScheme,
-	}
-	for _, schemeFunc := range schemeFuncs {
-		if err = schemeFunc(mgr.GetScheme()); err != nil {
-			logger.Fatalf("Error adding type to manager's scheme: %v", err)
-		}
-	}
-
-	// Add each controller's ProvideController func to this list to have the
-	// manager run it.
-	providers := []ProvideFunc{
-		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"),
-			}),
-	}
-	for _, provider := range providers {
-		if _, err = provider(mgr, logger.Desugar()); err != nil {
-			logger.Fatalf("Error adding controller to manager: %v", err)
-		}
-	}
-
-	// Start the Manager
-	go func() {
-		if localErr := mgr.Start(stopCh); localErr != nil {
-			logger.Fatalf("Error starting manager: %v", localErr)
-		}
-	}()
-
-	// Start the endpoint that Prometheus scraper talks to
-	srv := &http.Server{Addr: metricsScrapeAddr}
-	http.Handle(metricsScrapePath, promhttp.Handler())
-	go func() {
-		logger.Infof("Starting metrics listener at %s", metricsScrapeAddr)
-		if localErr := srv.ListenAndServe(); localErr != nil {
-			logger.Infof("HTTPserver: ListenAndServe() finished with error: %s", localErr)
-		}
-	}()
-
 	<-stopCh
-
-	// Close the http server gracefully
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	srv.Shutdown(ctx)
 }
 
 func init() {