pluggable leader election strategy

config/core/configmaps/leader-election.yaml

@@ -21,7 +21,7 @@ metadata:
     eventing.knative.dev/release: devel
     knative.dev/example-checksum: "43edfe90"
   annotations:
-    knative.dev/example-checksum: 0694ae46
+    knative.dev/example-checksum: 882dea41
 data:
   _example: |
     ################################
@@ -63,4 +63,5 @@ data:
     # - broker-controller
     # - inmemorychannel-dispatcher
     # - inmemorychannel-controller
+    # - pingsource-mt-adapter
     enabledComponents: "controller,broker-controller,inmemorychannel-dispatcher,inmemorychannel-controller,webhook,pingsource-mt-adapter"

pkg/adapter/v2/main.go

@@ -23,11 +23,12 @@ import (
 	"net/http"
 	"time"
 
+	"knative.dev/eventing/pkg/leaderelection"
+
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
-	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/profiling"
@@ -112,25 +113,13 @@ func MainWithContext(ctx context.Context, component string, ector EnvConfigConst
 	// Configuring the adapter
 	adapter := ctor(ctx, env, eventsClient)
 
-	run := func(ctx context.Context) {
-		logger.Info("Starting Receive Adapter", zap.Any("adapter", adapter))
-
-		if err := adapter.Start(ctx); err != nil {
-			logger.Warn("Start returned an error", zap.Error(err))
-		}
-	}
-
-	leConfig, err := env.GetLeaderElectionConfig()
+	// Build the leader elector
+	elector, err := leaderelection.BuildAdapterElector(ctx, adapter)
 	if err != nil {
-		logger.Error("Error loading the leader election configuration", zap.Error(err))
+		logger.Fatal("Error creating the adapter elector", zap.Error(err))
 	}
 
-	if leConfig.LeaderElect {
-		sharedmain.RunLeaderElected(ctx, logger, run, *leConfig)
-	} else {
-		logger.Infof("%v will not run in leader-elected mode", component)
-		run(ctx)
-	}
+	elector.Run(ctx)
 }
 
 func flush(logger *zap.SugaredLogger) {

pkg/adapter/v2/main_message.go

@@ -23,7 +23,7 @@ import (
 	"net/http"
 	"time"
 
-	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/eventing/pkg/leaderelection"
 
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/stats/view"
@@ -116,23 +116,11 @@ func MainMessageAdapterWithContext(ctx context.Context, component string, ector
 	// Configuring the adapter
 	adapter := ctor(ctx, env, httpBindingsSender, reporter)
 
-	run := func(ctx context.Context) {
-		logger.Info("Starting Receive MessageAdapter", zap.Any("adapter", adapter))
-
-		if err := adapter.Start(ctx); err != nil {
-			logger.Warn("start returned an error", zap.Error(err))
-		}
-	}
-
-	leConfig, err := env.GetLeaderElectionConfig()
+	// Build the leader elector
+	elector, err := leaderelection.BuildAdapterElector(ctx, adapter)
 	if err != nil {
-		logger.Error("Error loading the leader election configuration", zap.Error(err))
+		logger.Fatal("Error creating the adapter elector", zap.Error(err))
 	}
 
-	if leConfig.LeaderElect {
-		sharedmain.RunLeaderElected(ctx, logger, run, *leConfig)
-	} else {
-		logger.Infof("%v will not run in leader-elected mode", component)
-		run(ctx)
-	}
+	elector.Run(ctx)
 }

pkg/leaderelection/context.go

@@ -0,0 +1,161 @@
+/*
+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 leaderelection
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/tools/record"
+
+	"knative.dev/pkg/controller"
+	pkgleaderelection "knative.dev/pkg/leaderelection"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+)
+
+// WithStandardLeaderElectorBuilder infuses a context with the ability to build
+// LeaderElectors with the provided component configuration acquiring resource
+// locks via the provided kubernetes client
+func WithStandardLeaderElectorBuilder(ctx context.Context, kc kubernetes.Interface, cc pkgleaderelection.ComponentConfig) context.Context {
+	return context.WithValue(ctx, builderKey{}, &standardBuilder{
+		kc:  kc,
+		lec: cc,
+	})
+}
+
+// Elector is the interface for running a leader elector.
+type Elector interface {
+	Run(context.Context)
+}
+
+// Adapter is the interface for running adapter
+type Adapter interface {
+	Start(ctx context.Context) error
+}
+
+// BuildAdapterElector builds a leaderelection.LeaderElector for the given adapter
+// using a builder added to the context via WithXXLeaderElectorBuilder.
+func BuildAdapterElector(ctx context.Context, adapter Adapter) (Elector, error) {
+	if val := ctx.Value(builderKey{}); val != nil {
+		switch builder := val.(type) {
+		case *standardBuilder:
+			// Only use the standard elector is leader election is enabled
+			if builder.lec.LeaderElect {
+				return builder.BuildElector(ctx, adapter)
+			}
+		}
+	}
+
+	return &unopposedElector{
+		adapter: adapter,
+	}, nil
+}
+
+type builderKey struct{}
+
+type standardBuilder struct {
+	kc  kubernetes.Interface
+	lec pkgleaderelection.ComponentConfig
+}
+
+func (b *standardBuilder) BuildElector(ctx context.Context, adapter Adapter) (Elector, error) {
+	logger := logging.FromContext(ctx)
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcaster()
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&typedcorev1.EventSinkImpl{Interface: b.kc.CoreV1().Events(system.Namespace())}),
+		}
+		recorder = eventBroadcaster.NewRecorder(
+			scheme.Scheme, corev1.EventSource{Component: b.lec.Component})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+		}()
+	}
+
+	// Create a unique identifier so that two controllers on the same host don't
+	// race.
+	id, err := pkgleaderelection.UniqueID()
+	if err != nil {
+		logger.Fatalw("Failed to get unique ID for leader election", zap.Error(err))
+	}
+	logger.Infof("%v will run in leader-elected mode with id %v", b.lec.Component, id)
+
+	// rl is the resource used to hold the leader election lock.
+	rl, err := resourcelock.New(b.lec.ResourceLock,
+		system.Namespace(), // use namespace we are running in
+		b.lec.Component,    // component is used as the resource name
+		b.kc.CoreV1(),
+		b.kc.CoordinationV1(),
+		resourcelock.ResourceLockConfig{
+			Identity:      id,
+			EventRecorder: recorder,
+		})
+	if err != nil {
+		logger.Fatalw("Error creating lock", zap.Error(err))
+	}
+
+	return leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:          rl,
+		LeaseDuration: b.lec.LeaseDuration,
+		RenewDeadline: b.lec.RenewDeadline,
+		RetryPeriod:   b.lec.RetryPeriod,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				logger.Infof("%q has started leading %q", rl.Identity(), b.lec.Component)
+				err := adapter.Start(ctx)
+				if err != nil {
+					logger.Fatalw("The adapter failed to start", zap.Error(err))
+				}
+			},
+			OnStoppedLeading: func() {
+				logger.Infof("%q has stopped leading %q", rl.Identity(), b.lec.Component)
+			},
+		},
+		ReleaseOnCancel: true,
+		// TODO: use health check watchdog, knative/pkg#1048
+		Name: b.lec.Component,
+	})
+}
+
+// unopposedElector runs the adapter without needing to be elected.
+type unopposedElector struct {
+	adapter Adapter
+}
+
+// Run implements Elector
+func (ue *unopposedElector) Run(ctx context.Context) {
+	err := ue.adapter.Start(ctx)
+	if err != nil {
+		logging.FromContext(ctx).Warn("Start returned an error", zap.Error(err))
+	}
+}

pkg/leaderelection/context_test.go

@@ -0,0 +1,117 @@
+// +build !race
+// TODO(https://github.com/kubernetes/kubernetes/issues/90952): Remove the above.
+
+/*
+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 leaderelection
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekube "k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+	"knative.dev/pkg/leaderelection"
+	_ "knative.dev/pkg/system/testing"
+)
+
+type fakeAdapter struct{}
+
+func (a *fakeAdapter) Start(ctx context.Context) error {
+	<-ctx.Done()
+	return nil
+}
+
+func TestWithBuilder(t *testing.T) {
+	cc := leaderelection.ComponentConfig{
+		Component:     "component",
+		LeaderElect:   true,
+		ResourceLock:  "leases",
+		LeaseDuration: 15 * time.Second,
+		RenewDeadline: 10 * time.Second,
+		RetryPeriod:   2 * time.Second,
+	}
+	kc := fakekube.NewSimpleClientset()
+	ctx := context.Background()
+	a := &fakeAdapter{}
+
+	created := make(chan struct{})
+	kc.PrependReactor("create", "leases",
+		func(action ktesting.Action) (bool, runtime.Object, error) {
+			close(created)
+			return false, nil, nil
+		},
+	)
+
+	updated := make(chan struct{})
+	kc.PrependReactor("update", "leases",
+		func(action ktesting.Action) (bool, runtime.Object, error) {
+			// Only close update once.
+			select {
+			case <-updated:
+			default:
+				close(updated)
+			}
+			return false, nil, nil
+		},
+	)
+
+	if le, err := BuildAdapterElector(ctx, a); err != nil {
+		t.Errorf("BuildElector() = %v, wanted an unopposedElector", err)
+	} else if _, ok := le.(*unopposedElector); !ok {
+		t.Errorf("BuildElector() = %T, wanted an unopposedElector", le)
+	}
+
+	ctx = WithStandardLeaderElectorBuilder(ctx, kc, cc)
+
+	le, err := BuildAdapterElector(ctx, a)
+	if err != nil {
+		t.Errorf("BuildElector() = %v", err)
+	}
+
+	// We shouldn't see leases until we Run the elector.
+	select {
+	case <-created:
+		t.Error("Got created, want no actions.")
+	case <-updated:
+		t.Error("Got updated, want no actions.")
+	default:
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go le.Run(ctx)
+
+	select {
+	case <-created:
+		// We expect the lease to be created.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for lease creation.")
+	}
+
+	// Cancelling the context should case us to give up leadership.
+	cancel()
+
+	select {
+	case <-updated:
+		// We expect the lease to be updated.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for lease update.")
+	}
+}