ha buckets

cmd/mtping/main.go

@@ -37,7 +37,7 @@ func main() {
 	// which under the cover delays the release of the lease.
 	ctx := mtping.NewDelayingContext(sctx, mtping.GetNoShutDownAfterValue())
 
-	ctx = adapter.WithInjectorEnabled(ctx)
+	ctx = adapter.WithController(ctx, mtping.NewController)
 	ctx = adapter.WithHAEnabled(ctx)
 	adapter.MainWithContext(ctx, component, mtping.NewEnvConfig, mtping.NewAdapter)
 }

config/core/deployments/pingsource-mt-adapter.yaml

@@ -21,7 +21,7 @@ metadata:
     eventing.knative.dev/release: devel
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
-  replicas: 0
+  replicas: 3
   selector:
     matchLabels:
       eventing.knative.dev/source: ping-source-controller

docs/source/receive-adapter.md

@@ -83,12 +83,12 @@ handle more than one source instance at a time, either all source instances in
 one namespace or all source instances in the cluster.
 
 In order to support multi-resources per receive adapter, you need to enable the
-ConfigMap watcher feature, as follows:
+Controller watcher feature, as follows:
 
 ```go
 func main() {
   ctx := signals.NewContext()
-  ctx = adapter.WithInjectorEnabled(ctx)
+  ctx = adapter.WithController(ctx, youradapter.NewController)
   ctx = adapter.WithConfigMapWatcherEnabled(ctx)
   adapter.MainWithContext(ctx, "yourcomponent",
     youradapter.NewEnvConfig,

pkg/adapter/mtping/adapter.go

@@ -30,7 +30,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 
 	"knative.dev/eventing/pkg/adapter/v2"
@@ -72,14 +71,9 @@ func NewAdapter(ctx context.Context, _ adapter.EnvConfigAccessor, ceClient cloud
 
 // Start implements adapter.Adapter
 func (a *mtpingAdapter) Start(ctx context.Context) error {
-	ctrl := NewController(ctx, a)
-
-	a.logger.Info("Starting controllers...")
-	go controller.StartAll(ctx, ctrl)
-
-	defer a.runner.Stop()
 	a.logger.Info("Starting job runner...")
 	a.runner.Start(ctx.Done())
+	defer a.runner.Stop()
 
 	a.logger.Infof("runner stopped")
 	return nil

pkg/adapter/v2/config.go

@@ -179,18 +179,20 @@ func (e *EnvConfig) GetCloudEventOverrides() (*duckv1.CloudEventOverrides, error
 
 func (e *EnvConfig) GetLeaderElectionConfig() (*kle.ComponentConfig, error) {
 	if e.LeaderElectionConfigJson == "" {
-		return defaultLeaderElectionConfig(), nil
+		return e.defaultLeaderElectionConfig(), nil
 	}
 
 	var config kle.ComponentConfig
 	if err := json.Unmarshal([]byte(e.LeaderElectionConfigJson), &config); err != nil {
-		return defaultLeaderElectionConfig(), err
+		return e.defaultLeaderElectionConfig(), err
 	}
+	config.Component = e.Component
 	return &config, nil
 }
 
-func defaultLeaderElectionConfig() *kle.ComponentConfig {
+func (e *EnvConfig) defaultLeaderElectionConfig() *kle.ComponentConfig {
 	return &kle.ComponentConfig{
+		Component:     e.Component,
 		Buckets:       1,
 		LeaseDuration: 15 * time.Second,
 		RenewDeadline: 10 * time.Second,

pkg/adapter/v2/context.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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 adapter
+
+import (
+	"context"
+)
+
+type haEnabledKey struct{}
+
+// WithHAEnabled signals to MainWithContext that it should set up an appropriate leader elector for this component.
+func WithHAEnabled(ctx context.Context) context.Context {
+	return context.WithValue(ctx, haEnabledKey{}, struct{}{})
+}
+
+// IsHAEnabled checks the context for the desire to enable leader elector.
+func IsHAEnabled(ctx context.Context) bool {
+	val := ctx.Value(haEnabledKey{})
+	return val != nil
+}
+
+type haDisabledFlagKey struct{}
+
+// withHADisabledFlag signals to MainWithConfig that it should not set up an appropriate leader elector for this component.
+func withHADisabledFlag(ctx context.Context) context.Context {
+	return context.WithValue(ctx, haDisabledFlagKey{}, struct{}{})
+}
+
+// isHADisabledFlag checks the context for the desired to disable leader elector.
+func isHADisabledFlag(ctx context.Context) bool {
+	val := ctx.Value(haDisabledFlagKey{})
+	return val != nil
+}
+
+type controllerKey struct{}
+
+// WithController signals to MainWithContext that it should
+// create and configure a controller notifying the adapter
+// when a resource is ready and removed
+func WithController(ctx context.Context, ctor ControllerConstructor) context.Context {
+	return context.WithValue(ctx, controllerKey{}, ctor)
+}
+
+// ControllerFromContext gets the controller constructor from the context
+func ControllerFromContext(ctx context.Context) ControllerConstructor {
+	value := ctx.Value(controllerKey{})
+	if value == nil {
+		return nil
+	}
+	return value.(ControllerConstructor)
+}

pkg/leaderelection/leader_election.go → pkg/adapter/v2/context_test.go

@@ -14,10 +14,23 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package leaderelection
+package adapter
 
 import (
-	kle "knative.dev/pkg/leaderelection"
+	"context"
+	"testing"
+
+	"knative.dev/pkg/controller"
+
+	_ "knative.dev/pkg/client/injection/kube/client/fake"
 )
 
-var ValidateConfig = kle.NewConfigFromConfigMap
+func TestWithController(t *testing.T) {
+	ctx := WithController(context.TODO(), func(ctx context.Context, adapter Adapter) *controller.Impl {
+		return nil
+	})
+
+	if ControllerFromContext(ctx) == nil {
+		t.Error("expected non-nil controller constructor")
+	}
+}