Merge pull request #4 from ChenglongLiu/dev0415

Merge upstream master

Author: ChenglongLiu

Makefile

@@ -0,0 +1,28 @@
+.PHONY: build
+build: tidy ## Build the CLI
+	go build
+
+build-image: ## Build the Docker image
+	docker build -t kubernetes-event-exporter .
+
+.PHONY: fmt
+fmt: ## Run go fmt against code
+	gofmt -s -l -w .
+
+.PHONY: vet
+vet: ## Run go vet against code
+	go vet ./...
+
+tidy: ## Run go mod tidy
+	go mod tidy
+
+test: tidy ## Run tests
+	go test -cover -mod=mod -v ./...
+
+clean: ## Delete go.sum and clean mod cache
+	go clean -modcache
+	rm go.sum
+
+.PHONY: help
+help: ## Display this help.
+	@cat $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n  make \033[36m<target>\033[0m\n"} /^[a-zA-Z_0-9-]+:.*?##/ { printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2 } /^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } '

deploy/00-roles.yaml

@@ -25,3 +25,6 @@ rules:
 - apiGroups: ["*"]
   resources: ["*"]
   verbs: ["get", "watch", "list"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["*"]

event-exporter.dockerignore

@@ -0,0 +1,5 @@
+# Ignore everything
+*
+
+# Include pkg directory
+!pkg/

main.go

@@ -34,7 +34,7 @@ func main() {
 
 	configBytes = []byte(os.ExpandEnv(string(configBytes)))
 
-	cfg, err := setup.ParseConfigFromBites(configBytes)
+	cfg, err := setup.ParseConfigFromBytes(configBytes)
 	if err != nil {
 		log.Fatal().Msg(err.Error())
 	}
@@ -64,6 +64,8 @@ func main() {
 
 	cfg.SetDefaults()
 
+	log.Info().Msgf("Starting with config: %#v", cfg)
+
 	if err := cfg.Validate(); err != nil {
 		log.Fatal().Err(err).Msg("config validation failed")
 	}
@@ -91,45 +93,61 @@ func main() {
 
 	w := kube.NewEventWatcher(kubecfg, cfg.Namespace, cfg.MaxEventAgeSeconds, metricsStore, onEvent, cfg.OmitLookup, cfg.CacheSize)
 
-	ctx, cancel := context.WithCancel(context.Background())
-	leaderLost := make(chan bool)
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+
 	if cfg.LeaderElection.Enabled {
+		var wasLeader bool
+		log.Info().Msg("leader election enabled")
+
+		onStoppedLeading := func(ctx context.Context) {
+			select {
+			case <-ctx.Done():
+				log.Info().Msg("Context was cancelled, stopping leader election loop")
+			default:
+				log.Info().Msg("Lost the leader lease, stopping leader election loop")
+			}
+		}
+
 		l, err := kube.NewLeaderElector(cfg.LeaderElection.LeaderElectionID, kubecfg,
+			// this method gets called when this instance becomes the leader
 			func(_ context.Context) {
-				log.Info().Msg("leader election got")
+				wasLeader = true
+				log.Info().Msg("leader election won")
 				w.Start()
 			},
+			// this method gets called when the leader election loop is closed
+			// either due to context cancellation or due to losing the leader lease
 			func() {
-				log.Error().Msg("leader election lost")
-				leaderLost <- true
+				onStoppedLeading(ctx)
+			},
+			func(identity string) {
+				log.Info().Msg("new leader observed: " + identity)
 			},
 		)
 		if err != nil {
 			log.Fatal().Err(err).Msg("create leaderelector failed")
 		}
-		go l.Run(ctx)
+
+		// Run returns if either the context is canceled or client stopped holding the leader lease
+		l.Run(ctx)
+
+		// We get here either because we lost the leader lease or the context was canceled.
+		// In either case we want to stop the event watcher and exit.
+		// However, if we were the leader, we wait leaseDuration seconds before stopping
+		// so that we don't lose events until the next leader is elected. The new leader
+		// will only be elected after leaseDuration seconds.
+		if wasLeader {
+			log.Info().Msgf("waiting leaseDuration seconds before stopping: %s", kube.GetLeaseDuration())
+			time.Sleep(kube.GetLeaseDuration())
+		}
 	} else {
+		log.Info().Msg("leader election disabled")
 		w.Start()
+		<-ctx.Done()
 	}
 
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
-
-	gracefulExit := func() {
-		defer close(c)
-		defer close(leaderLost)
-		cancel()
-		w.Stop()
-		engine.Stop()
-		log.Info().Msg("Exiting")
-	}
-
-	select {
-	case sig := <-c:
-		log.Info().Str("signal", sig.String()).Msg("Received signal to exit")
-		gracefulExit()
-	case <-leaderLost:
-		log.Warn().Msg("Leader election lost")
-		gracefulExit()
-	}
+	log.Info().Msg("Received signal to exit. Stopping.")
+	w.Stop()
+	engine.Stop()
 }

pkg/kube/leaderelection.go

@@ -28,6 +28,10 @@ const (
 	defaultRetryPeriod      = 2 * time.Second
 )
 
+func GetLeaseDuration() time.Duration {
+	return defaultLeaseDuration
+}
+
 // NewResourceLock creates a new config map resource lock for use in a leader
 // election loop
 func newResourceLock(config *rest.Config, leaderElectionID string) (resourcelock.Interface, error) {
@@ -53,7 +57,7 @@ func newResourceLock(config *rest.Config, leaderElectionID string) (resourcelock
 		return nil, err
 	}
 
-	return resourcelock.New(resourcelock.ConfigMapsLeasesResourceLock,
+	return resourcelock.New(resourcelock.LeasesResourceLock,
 		leaderElectionNamespace,
 		leaderElectionID,
 		client.CoreV1(),
@@ -82,7 +86,7 @@ func getInClusterNamespace() (string, error) {
 }
 
 // NewLeaderElector return  a leader elector object using client-go
-func NewLeaderElector(leaderElectionID string, config *rest.Config, startFunc func(context.Context), stopFunc func()) (*leaderelection.LeaderElector, error) {
+func NewLeaderElector(leaderElectionID string, config *rest.Config, startFunc func(context.Context), stopFunc func(), newLeaderFunc func(string)) (*leaderelection.LeaderElector, error) {
 	resourceLock, err := newResourceLock(config, leaderElectionID)
 	if err != nil {
 		return &leaderelection.LeaderElector{}, err
@@ -96,6 +100,7 @@ func NewLeaderElector(leaderElectionID string, config *rest.Config, startFunc fu
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: startFunc,
 			OnStoppedLeading: stopFunc,
+			OnNewLeader:      newLeaderFunc,
 		},
 	})
 	return l, err

pkg/kube/watcher.go

@@ -1,6 +1,7 @@
 package kube
 
 import (
+	"sync"
 	"time"
 
 	"github.com/resmoio/kubernetes-event-exporter/pkg/metrics"
@@ -19,6 +20,7 @@ var startUpTime = time.Now()
 type EventHandler func(event *EnhancedEvent)
 
 type EventWatcher struct {
+	wg                  sync.WaitGroup
 	informer            cache.SharedInformer
 	stopper             chan struct{}
 	objectMetadataCache ObjectMetadataProvider
@@ -135,12 +137,16 @@ func (e *EventWatcher) OnDelete(obj interface{}) {
 }
 
 func (e *EventWatcher) Start() {
-	go e.informer.Run(e.stopper)
+	e.wg.Add(1)
+	go func() {
+		defer e.wg.Done()
+		e.informer.Run(e.stopper)
+	}()
 }
 
 func (e *EventWatcher) Stop() {
-	e.stopper <- struct{}{}
 	close(e.stopper)
+	e.wg.Wait()
 }
 
 func (e *EventWatcher) setStartUpTime(time time.Time) {

pkg/setup/setup.go

@@ -8,7 +8,7 @@ import (
 	"github.com/resmoio/kubernetes-event-exporter/pkg/exporter"
 )
 
-func ParseConfigFromBites(configBytes []byte) (exporter.Config, error) {
+func ParseConfigFromBytes(configBytes []byte) (exporter.Config, error) {
 	var config exporter.Config
 	err := yaml.Unmarshal(configBytes, &config)
 	if err != nil {

pkg/setup/setup_test.go

@@ -7,14 +7,14 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func Test_ParseConfigFromBites_ExampleConfigIsCorrect(t *testing.T) {
+func Test_ParseConfigFromBytes_ExampleConfigIsCorrect(t *testing.T) {
 	configBytes, err := os.ReadFile("../../config.example.yaml")
 	if err != nil {
 		assert.NoError(t, err, "cannot read config file: "+err.Error())
 		return
 	}
 
-	config, err := ParseConfigFromBites(configBytes)
+	config, err := ParseConfigFromBytes(configBytes)
 
 	assert.NoError(t, err)
 	assert.NotEmpty(t, config.LogLevel)
@@ -26,26 +26,26 @@ func Test_ParseConfigFromBites_ExampleConfigIsCorrect(t *testing.T) {
 	assert.Equal(t, 10, len(config.Receivers))
 }
 
-func Test_ParseConfigFromBites_NoErrors(t *testing.T) {
+func Test_ParseConfigFromBytes_NoErrors(t *testing.T) {
 	configBytes := []byte(`
 logLevel: info
 logFormat: json
 `)
 
-	config, err := ParseConfigFromBites(configBytes)
+	config, err := ParseConfigFromBytes(configBytes)
 
 	assert.NoError(t, err)
 	assert.Equal(t, "info", config.LogLevel)
 	assert.Equal(t, "json", config.LogFormat)
 }
 
-func Test_ParseConfigFromBites_ErrorWhenCurlyBracesNotEscaped(t *testing.T) {
+func Test_ParseConfigFromBytes_ErrorWhenCurlyBracesNotEscaped(t *testing.T) {
 	configBytes := []byte(`
 logLevel: {{info}}
 logFormat: json
 `)
 
-	config, err := ParseConfigFromBites(configBytes)
+	config, err := ParseConfigFromBytes(configBytes)
 
 	expectedErrorLine := ">  2 | logLevel: {{info}}"
 	expectedErrorSuggestion := "Need to wrap values with special characters in quotes"
@@ -56,26 +56,26 @@ logFormat: json
 	assert.Equal(t, "", config.LogFormat)
 }
 
-func Test_ParseConfigFromBites_OkWhenCurlyBracesEscaped(t *testing.T) {
+func Test_ParseConfigFromBytes_OkWhenCurlyBracesEscaped(t *testing.T) {
 	configBytes := []byte(`
 logLevel: "{{info}}"
 logFormat: json
 `)
 
-	config, err := ParseConfigFromBites(configBytes)
+	config, err := ParseConfigFromBytes(configBytes)
 
 	assert.Nil(t, err)
 	assert.Equal(t, "{{info}}", config.LogLevel)
 	assert.Equal(t, "json", config.LogFormat)
 }
 
-func Test_ParseConfigFromBites_ErrorErrorNotWithCurlyBraces(t *testing.T) {
+func Test_ParseConfigFromBytes_ErrorErrorNotWithCurlyBraces(t *testing.T) {
 	configBytes := []byte(`
 logLevelNotYAMLErrorError
 logFormat: json
 `)
 
-	config, err := ParseConfigFromBites(configBytes)
+	config, err := ParseConfigFromBytes(configBytes)
 
 	expectedErrorLine := ">  2 | logLevelNotYAMLErrorError"
 	expectedErrorSuggestion := "Need to wrap values with special characters in quotes"