mtping now uses bucket-based leader election (#4009)

* prepare pingsource controller for codegen

* add more tests

* prepare pingsource controller for codegen

* prepare pingsource controller for codegen

* prepare pingsource controller for codegen

* ha buckets

* rebasing

* remove pingconfig

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)
 }

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

@@ -27,8 +27,8 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/robfig/cron/v3"
 	"go.uber.org/zap"
+
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 
 	"knative.dev/eventing/pkg/adapter/v2"
@@ -70,14 +70,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/mtping/adapter_test.go

@@ -32,16 +32,12 @@ import (
 	"knative.dev/pkg/logging"
 	rectesting "knative.dev/pkg/reconciler/testing"
 
-	pkgadapter "knative.dev/eventing/pkg/adapter/v2"
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
 )
 
 func TestStartStopAdapter(t *testing.T) {
 	ctx, _ := rectesting.SetupFakeContext(t)
 	ctx, cancel := context.WithCancel(ctx)
-	cmw := pkgadapter.SetupConfigMapWatchOrDie(ctx, "component", "test-ns")
-	ctx = pkgadapter.WithConfigMapWatcher(ctx, cmw)
-
 	envCfg := NewEnvConfig()
 
 	ce := adaptertest.NewTestClient()

pkg/adapter/mtping/controller.go

@@ -65,6 +65,7 @@ func NewController(ctx context.Context, adapter adapter.Adapter) *controller.Imp
 	//		},
 	//	}
 	//}
+
 	impl := pingsourcereconciler.NewImpl(ctx, r)
 
 	logging.FromContext(ctx).Info("Setting up event handlers")

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/adapter/v2/config_watcher_test.go → pkg/adapter/v2/context_test.go

@@ -20,23 +20,34 @@ import (
 	"context"
 	"testing"
 
+	"knative.dev/pkg/controller"
+
 	_ "knative.dev/pkg/client/injection/kube/client/fake"
-	rectesting "knative.dev/pkg/reconciler/testing"
 )
 
-func TestConfigMapWatcherEnabled(t *testing.T) {
-	ctx := WithConfigMapWatcherEnabled(context.TODO())
-	if !IsConfigMapWatcherEnabled(ctx) {
-		t.Error("expected config watcher to be enabled")
+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")
 	}
 }
 
-func TestConfigMapWatcher(t *testing.T) {
-	ctx, _ := rectesting.SetupFakeContext(t)
-	watcher := SetupConfigMapWatchOrDie(ctx, "component", "test-ns")
-	ctx = WithConfigMapWatcher(ctx, watcher)
+func TestWithHAEnabled(t *testing.T) {
+	ctx := context.Background()
+	ctx = WithHAEnabled(ctx)
+	if !IsHAEnabled(ctx) {
+		t.Error("Expected HA to be enabled")
+	}
 
-	if ConfigMapWatcherFromContext(ctx) == nil {
-		t.Error("expected config watcher, got nothing")
+	ctx = withHADisabledFlag(ctx)
+	if !IsHAEnabled(ctx) {
+		t.Error("Expected HA to be enabled")
 	}
+	if !isHADisabledFlag(ctx) {
+		t.Error("Expected HA to be disabled via commandline flag")
+	}
+
 }