API + Reconciler for CronJobSource.

cmd/controller/main.go

@@ -104,12 +104,15 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(opt.KubeClientSet, opt.ResyncPeriod)
 	eventingInformerFactory := informers.NewSharedInformerFactory(opt.EventingClientSet, opt.ResyncPeriod)
 
+	// Eventing
 	triggerInformer := eventingInformerFactory.Eventing().V1alpha1().Triggers()
 	channelInformer := eventingInformerFactory.Eventing().V1alpha1().Channels()
 	subscriptionInformer := eventingInformerFactory.Eventing().V1alpha1().Subscriptions()
 	brokerInformer := eventingInformerFactory.Eventing().V1alpha1().Brokers()
-	coreServiceInformer := kubeInformerFactory.Core().V1().Services()
-	coreNamespaceInformer := kubeInformerFactory.Core().V1().Namespaces()
+
+	// Kube
+	serviceInformer := kubeInformerFactory.Core().V1().Services()
+	namespaceInformer := kubeInformerFactory.Core().V1().Namespaces()
 	configMapInformer := kubeInformerFactory.Core().V1().ConfigMaps()
 
 	// Build all of our controllers, with the clients constructed above.
@@ -122,7 +125,7 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 		),
 		namespace.NewController(
 			opt,
-			coreNamespaceInformer,
+			namespaceInformer,
 		),
 		channel.NewController(
 			opt,
@@ -134,7 +137,7 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 			channelInformer,
 			subscriptionInformer,
 			brokerInformer,
-			coreServiceInformer,
+			serviceInformer,
 		),
 	}
 	if len(controllers) != numControllers {
@@ -153,13 +156,15 @@ func startPkgController(stopCh <-chan struct{}, cfg *rest.Config, logger *zap.Su
 	logger.Info("Starting informers.")
 	if err := kncontroller.StartInformers(
 		stopCh,
+		// Eventing
+		brokerInformer.Informer(),
+		channelInformer.Informer(),
 		subscriptionInformer.Informer(),
-		configMapInformer.Informer(),
-		coreNamespaceInformer.Informer(),
 		triggerInformer.Informer(),
-		channelInformer.Informer(),
-		brokerInformer.Informer(),
-		coreServiceInformer.Informer(),
+		// Kube
+		configMapInformer.Informer(),
+		serviceInformer.Informer(),
+		namespaceInformer.Informer(),
 	); err != nil {
 		logger.Fatalf("Failed to start informers: %v", err)
 	}

cmd/cronjob_receive_adapter/main.go

@@ -0,0 +1,74 @@
+/*
+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"
+
+	"github.com/knative/eventing/pkg/adapter/cronjobevents"
+	"github.com/knative/pkg/signals"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"golang.org/x/net/context"
+)
+
+const (
+	// Environment variable container schedule.
+	envSchedule = "SCHEDULE"
+
+	// Environment variable containing data.
+	envData = "DATA"
+
+	// Sink for messages.
+	envSinkURI = "SINK_URI"
+)
+
+func getRequiredEnv(envKey string) string {
+	val, defined := os.LookupEnv(envKey)
+	if !defined {
+		log.Fatalf("required environment variable not defined %q", envKey)
+	}
+	return val
+}
+
+func main() {
+	flag.Parse()
+
+	ctx := context.Background()
+	logCfg := zap.NewProductionConfig()
+	logCfg.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
+	logger, err := logCfg.Build()
+	if err != nil {
+		log.Fatalf("Unable to create logger: %v", err)
+	}
+
+	adapter := &cronjobevents.Adapter{
+		Schedule: getRequiredEnv(envSchedule),
+		Data:     getRequiredEnv(envData),
+		SinkURI:  getRequiredEnv(envSinkURI),
+	}
+
+	logger.Info("Starting Receive Adapter", zap.Reflect("adapter", adapter))
+
+	stopCh := signals.SetupSignalHandler()
+
+	if err := adapter.Start(ctx, stopCh); err != nil {
+		logger.Fatal("Failed to start adapter", zap.Error(err))
+	}
+}

cmd/sources-controller/kodata/HEAD

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

cmd/sources-controller/kodata/LICENSE

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

cmd/sources-controller/kodata/VENDOR-LICENSE

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

cmd/sources-controller/main.go

@@ -0,0 +1,163 @@
+/*
+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"
+	"k8s.io/client-go/tools/clientcmd"
+	"log"
+
+	// 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"
+
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/rest"
+
+	informers "github.com/knative/eventing/pkg/client/informers/externalversions"
+	"github.com/knative/eventing/pkg/logconfig"
+	"github.com/knative/eventing/pkg/logging"
+	"github.com/knative/eventing/pkg/reconciler"
+	"github.com/knative/eventing/pkg/reconciler/cronjobsource"
+	"github.com/knative/pkg/configmap"
+	kncontroller "github.com/knative/pkg/controller"
+	"github.com/knative/pkg/logging/logkey"
+	"github.com/knative/pkg/signals"
+	"go.uber.org/zap"
+)
+
+var (
+	hardcodedLoggingConfig = flag.Bool("hardCodedLoggingConfig", false, "If true, use the hard coded logging config. It is intended to be used only when debugging outside a Kubernetes cluster.")
+	masterURL              = flag.String("master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+	kubeconfig             = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
+)
+
+func main() {
+	flag.Parse()
+
+	logger, atomicLevel := setupLogger()
+	defer logger.Sync()
+	logger = logger.With(zap.String(logkey.ControllerType, logconfig.SourcesController))
+
+	// set up signals so we handle the first shutdown signal gracefully
+	stopCh := signals.SetupSignalHandler()
+
+	cfg, err := clientcmd.BuildConfigFromFlags(*masterURL, *kubeconfig)
+	if err != nil {
+		logger.Fatalw("Error building kubeconfig", zap.Error(err))
+	}
+
+	logger = logger.With(zap.String("controller/impl", "pkg"))
+	logger.Info("Starting the controller")
+
+	const numControllers = 1
+	cfg.QPS = numControllers * rest.DefaultQPS
+	cfg.Burst = numControllers * rest.DefaultBurst
+	opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh)
+
+	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(opt.KubeClientSet, opt.ResyncPeriod)
+	eventingInformerFactory := informers.NewSharedInformerFactory(opt.EventingClientSet, opt.ResyncPeriod)
+
+	// Eventing
+	cronjobsourceInformer := eventingInformerFactory.Sources().V1alpha1().CronJobSources()
+
+	// Kube
+	deploymentInformer := kubeInformerFactory.Apps().V1().Deployments()
+
+	// Build all of our controllers, with the clients constructed above.
+	// Add new controllers to this array.
+	// You also need to modify numControllers above to match this.
+	controllers := []*kncontroller.Impl{
+		cronjobsource.NewController(
+			opt,
+			cronjobsourceInformer,
+			deploymentInformer,
+		),
+	}
+	if len(controllers) != numControllers {
+		logger.Fatalf("Number of controllers and QPS settings mismatch: %d != %d", len(controllers), numControllers)
+	}
+
+	// Watch the logging config map and dynamically update logging levels.
+	opt.ConfigMapWatcher.Watch(logconfig.ConfigMapName(), logging.UpdateLevelFromConfigMap(logger, atomicLevel, logconfig.SourcesController))
+	// TODO: Watch the observability config map and dynamically update metrics exporter.
+	//opt.ConfigMapWatcher.Watch(metrics.ObservabilityConfigName, metrics.UpdateExporterFromConfigMap(component, logger))
+	if err := opt.ConfigMapWatcher.Start(stopCh); err != nil {
+		logger.Fatalw("failed to start configuration manager", zap.Error(err))
+	}
+
+	// Start all of the informers and wait for them to sync.
+	logger.Info("Starting informers.")
+	if err := kncontroller.StartInformers(
+		stopCh,
+		// Eventing
+		cronjobsourceInformer.Informer(),
+		// Kube
+		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...)
+
+	<-stopCh
+}
+
+func setupLogger() (*zap.SugaredLogger, zap.AtomicLevel) {
+	// Set up our logger.
+	loggingConfigMap := getLoggingConfigOrDie()
+	loggingConfig, err := logging.NewConfigFromMap(loggingConfigMap)
+	if err != nil {
+		log.Fatalf("Error parsing logging configuration: %v", err)
+	}
+	return logging.NewLoggerFromConfig(loggingConfig, logconfig.Controller)
+}
+
+func getLoggingConfigOrDie() map[string]string {
+	if hardcodedLoggingConfig != nil && *hardcodedLoggingConfig {
+		return map[string]string{
+			"loglevel.controller": "info",
+			"zap-logger-config": `
+				{
+					"level": "info",
+					"development": false,
+					"outputPaths": ["stdout"],
+					"errorOutputPaths": ["stderr"],
+					"encoding": "json",
+					"encoderConfig": {
+					"timeKey": "ts",
+					"levelKey": "level",
+					"nameKey": "logger",
+					"callerKey": "caller",
+					"messageKey": "msg",
+					"stacktraceKey": "stacktrace",
+					"lineEnding": "",
+					"levelEncoder": "",
+					"timeEncoder": "iso8601",
+					"durationEncoder": "",
+					"callerEncoder": ""
+				}`,
+		}
+	} else {
+		cm, err := configmap.Load("/etc/config-logging")
+		if err != nil {
+			log.Fatalf("Error loading logging configuration: %v", err)
+		}
+		return cm
+	}
+}