Run all controllers in the same process

controller-runtime implements its own command wrapper, but it can still
run in the same process as other controllers. This moves them to a
single deployment and removes the existing cmd/controller-manager
binary.

cmd/controller-manager/main.go → cmd/controller/controller-runtime-main.go

@@ -16,31 +16,16 @@ limitations under the License.
 package main
 
 import (
-	"flag"
-	"log"
-
 	buildv1alpha1 "github.com/knative/build/pkg/apis/build/v1alpha1"
 	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
 	feedsv1alpha1 "github.com/knative/eventing/pkg/apis/feeds/v1alpha1"
 	flowsv1alpha1 "github.com/knative/eventing/pkg/apis/flows/v1alpha1"
 	"github.com/knative/eventing/pkg/controller/feed"
 	"github.com/knative/eventing/pkg/controller/flow"
-	"github.com/knative/eventing/pkg/logconfig"
-
-	"github.com/knative/pkg/configmap"
-	"github.com/knative/pkg/logging"
-	"github.com/knative/pkg/logging/logkey"
-
 	istiov1alpha3 "github.com/knative/serving/pkg/apis/istio/v1alpha3"
 	servingv1alpha1 "github.com/knative/serving/pkg/apis/serving/v1alpha1"
-	"go.uber.org/zap"
-
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
-	"k8s.io/client-go/rest"
-
-	"github.com/knative/eventing/pkg/system"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -54,51 +39,16 @@ type SchemeFunc func(*runtime.Scheme) error
 // ProvideFunc adds a controller to a Manager.
 type ProvideFunc func(manager.Manager) (controller.Controller, error)
 
-func main() {
-	flag.Parse()
-
-	// Read the logging config and setup a logger.
-	cm, err := configmap.Load("/etc/config-logging")
-	if err != nil {
-		log.Fatalf("Error loading logging configuration: %v", err)
-	}
-	cfg, err := logging.NewConfigFromMap(cm, logconfig.ControllerManager)
-	if err != nil {
-		log.Fatalf("Error parsing logging configuration: %v", err)
-	}
-	logger, atomicLevel := logging.NewLoggerFromConfig(cfg, logconfig.ControllerManager)
-	defer logger.Sync()
-	logger = logger.With(zap.String(logkey.ControllerType, logconfig.ControllerManager))
-
-	logger.Info("Starting the Controller Manager")
-
-	// This tells controller-runtime to use zap to log internal messages.
+// controllerRuntimeStart runs controllers written for controller-runtime. It's
+// intended to be called from main(). Any controllers migrated to use
+// controller-runtime should move their initialization to this function.
+func controllerRuntimeStart() error {
 	logf.SetLogger(logf.ZapLogger(false))
 
-	// 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", err)
-	}
-
-	kubeClient, err := kubernetes.NewForConfig(clusterConfig)
-	if err != nil {
-		logger.Fatal("Failed to get the client set", err)
-	}
-
-	// Watch the logging config map and dynamically update logging levels.
-	configMapWatcher := configmap.NewDefaultWatcher(kubeClient, system.Namespace)
-	configMapWatcher.Watch(logconfig.ConfigName, logging.UpdateLevelFromConfigMap(logger, atomicLevel, logconfig.ControllerManager, logconfig.ControllerManager))
-	if err = configMapWatcher.Start(stopCh); err != nil {
-		logger.Fatalf("failed to start controller manager configmap watcher: %v", err)
-	}
-
 	// Setup a Manager
 	mrg, err := manager.New(config.GetConfigOrDie(), manager.Options{})
 	if err != nil {
-		log.Fatal(err)
+		return err
 	}
 
 	// Add custom types to this array to get them into the manager's scheme.
@@ -121,12 +71,11 @@ func main() {
 		flow.ProvideController,
 	}
 
-	// TODO(n3wscott): Send the logger to the controllers.
 	for _, provider := range providers {
 		if _, err := provider(mrg); err != nil {
-			log.Fatal(err)
+			return err
 		}
 	}
 
-	log.Fatal(mrg.Start(stopCh))
+	return mrg.Start(signals.SetupSignalHandler())
 }

cmd/controller/main.go

@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"flag"
+	"log"
 	"net/http"
 	"time"
 
@@ -47,7 +48,6 @@ import (
 	"github.com/knative/pkg/logging/logkey"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
-	"log"
 )
 
 const (
@@ -122,7 +122,8 @@ func main() {
 		logger.Fatalf("failed to start controller config map watcher: %v", err)
 	}
 
-	// Add new controllers here.
+	// Add new controllers here, except controllers that use controller-runtime.
+	// Those should be added to controller-runtime-main.go.
 	ctors := []controller.Constructor{
 		bus.NewController,
 		clusterbus.NewController,
@@ -152,6 +153,13 @@ func main() {
 		}(ctrlr)
 	}
 
+	// Start the controller-runtime controllers.
+	go func() {
+		if err := controllerRuntimeStart(); err != nil {
+			logger.Fatalf("Error running controller-runtime controllers: %v", err)
+		}
+	}()
+
 	// Start the endpoint that Prometheus scraper talks to
 	srv := &http.Server{Addr: metricsScrapeAddr}
 	http.Handle(metricsScrapePath, promhttp.Handler())
@@ -171,6 +179,7 @@ func main() {
 }
 
 func init() {
-	flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
-	flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+	// These are commented because they're also defined by controller-runtime.
+	// flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
+	// flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
 }