From f27f130a2d2717db61f89e22a778fd4b9ae6c0c0 Mon Sep 17 00:00:00 2001 From: Derek Wang Date: Tue, 6 Apr 2021 17:05:07 -0700 Subject: [PATCH] feat: EventSource and Sensor HA without extra RBAC Signed-off-by: Derek Wang --- common/leaderelection/leaderelection.go | 158 ++++++++++++++++++ controllers/eventsource/controller.go | 14 -- controllers/eventsource/resource.go | 27 +-- controllers/sensor/controller.go | 14 -- controllers/sensor/resource.go | 9 +- docs/eventsources/ha.md | 25 --- docs/sensors/ha.md | 24 --- docs/service-accounts.md | 10 +- eventsources/eventing.go | 60 ++----- go.mod | 4 +- go.sum | 8 +- .../rbac/argo-events-cluster-role.yaml | 6 - manifests/install.yaml | 6 - manifests/namespace-install.yaml | 6 - .../rbac/argo-events-role.yaml | 6 - sensors/listener.go | 57 ++----- test/util/util.go | 75 ++++----- 17 files changed, 240 insertions(+), 269 deletions(-) create mode 100644 common/leaderelection/leaderelection.go diff --git a/common/leaderelection/leaderelection.go b/common/leaderelection/leaderelection.go new file mode 100644 index 0000000000..cd364970e3 --- /dev/null +++ b/common/leaderelection/leaderelection.go @@ -0,0 +1,158 @@ +package leaderelection + +import ( + "context" + + "github.com/fsnotify/fsnotify" + "github.com/nats-io/graft" + nats "github.com/nats-io/nats.go" + "github.com/pkg/errors" + "github.com/spf13/viper" + "go.uber.org/zap" + + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/common/logging" + eventbusdriver "github.com/argoproj/argo-events/eventbus/driver" + apicommon "github.com/argoproj/argo-events/pkg/apis/common" + eventbusv1alpha1 "github.com/argoproj/argo-events/pkg/apis/eventbus/v1alpha1" +) + +type Elector interface { + RunOrDie(context.Context, LeaderCallbacks) +} + +type LeaderCallbacks struct { + OnStartedLeading func(context.Context) + OnStoppedLeading func() +} + +func NewEventBusElector(ctx context.Context, eventBusConfig eventbusv1alpha1.BusConfig, clusterName string, clusterSize int) (Elector, error) { + logger := logging.FromContext(ctx) + var eventBusType apicommon.EventBusType + var eventBusAuth *eventbusv1alpha1.AuthStrategy + if eventBusConfig.NATS != nil { + eventBusType = apicommon.EventBusNATS + eventBusAuth = eventBusConfig.NATS.Auth + } else { + return nil, errors.New("invalid event bus") + } + var auth *eventbusdriver.Auth + cred := &eventbusdriver.AuthCredential{} + if eventBusAuth == nil || *eventBusAuth == eventbusv1alpha1.AuthStrategyNone { + auth = &eventbusdriver.Auth{ + Strategy: eventbusv1alpha1.AuthStrategyNone, + } + } else { + v := viper.New() + v.SetConfigName("auth") + v.SetConfigType("yaml") + v.AddConfigPath(common.EventBusAuthFileMountPath) + err := v.ReadInConfig() + if err != nil { + return nil, errors.Errorf("failed to load auth.yaml. err: %+v", err) + } + err = v.Unmarshal(cred) + if err != nil { + logger.Errorw("failed to unmarshal auth.yaml", zap.Error(err)) + return nil, err + } + v.WatchConfig() + v.OnConfigChange(func(e fsnotify.Event) { + logger.Info("eventbus auth config file changed.") + err = v.Unmarshal(cred) + if err != nil { + logger.Errorw("failed to unmarshal auth.yaml after reloading", zap.Error(err)) + } + }) + auth = &eventbusdriver.Auth{ + Strategy: *eventBusAuth, + Crendential: cred, + } + } + var elector Elector + switch eventBusType { + case apicommon.EventBusNATS: + elector = &natsEventBusElector{ + clusterName: clusterName, + size: clusterSize, + url: eventBusConfig.NATS.URL, + auth: auth, + } + default: + return nil, errors.New("invalid eventbus type") + } + return elector, nil +} + +type natsEventBusElector struct { + clusterName string + size int + url string + auth *eventbusdriver.Auth +} + +func (e *natsEventBusElector) RunOrDie(ctx context.Context, callbacks LeaderCallbacks) { + log := logging.FromContext(ctx) + ci := graft.ClusterInfo{Name: e.clusterName, Size: e.size} + opts := &nats.DefaultOptions + opts.Url = e.url + if e.auth.Strategy == eventbusv1alpha1.AuthStrategyToken { + opts.Token = e.auth.Crendential.Token + } + rpc, err := graft.NewNatsRpc(opts) + if err != nil { + log.Fatalw("failed to new Nats Rpc", zap.Error(err)) + } + errChan := make(chan error) + stateChangeChan := make(chan graft.StateChange) + handler := graft.NewChanHandler(stateChangeChan, errChan) + node, err := graft.New(ci, handler, rpc, "/tmp/graft.log") + if err != nil { + log.Fatalw("failed to new a node", zap.Error(err)) + } + defer node.Close() + + cctx, cancel := context.WithCancel(ctx) + if node.State() == graft.LEADER { + log.Info("I'm the LEADER, starting ...") + go callbacks.OnStartedLeading(cctx) + } else { + log.Info("Not the LEADER, stand by ...") + } + + handleStateChange := func(sc graft.StateChange) { + switch sc.To { + case graft.LEADER: + log.Info("I'm the LEADER, starting ...") + go callbacks.OnStartedLeading(cctx) + case graft.FOLLOWER, graft.CANDIDATE: + log.Infof("Becoming a %v, stand by ...", sc.To) + if sc.From == graft.LEADER { + cancel() + callbacks.OnStoppedLeading() + cctx, cancel = context.WithCancel(ctx) + } + case graft.CLOSED: + if sc.From == graft.LEADER { + cancel() + callbacks.OnStoppedLeading() + } + log.Fatal("Leader elector connection was CLOSED") + default: + log.Fatalf("Unknown state: %s", sc.To) + } + } + + for { + select { + case <-ctx.Done(): + log.Info("exiting...") + cancel() + return + case sc := <-stateChangeChan: + handleStateChange(sc) + case err := <-errChan: + log.Errorw("Error happened", zap.Error(err)) + } + } +} diff --git a/controllers/eventsource/controller.go b/controllers/eventsource/controller.go index 4d85b73263..767c177046 100644 --- a/controllers/eventsource/controller.go +++ b/controllers/eventsource/controller.go @@ -4,7 +4,6 @@ import ( "context" "go.uber.org/zap" - coordinationv1 "k8s.io/api/coordination/v1" "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" @@ -71,9 +70,6 @@ func (r *reconciler) reconcile(ctx context.Context, eventSource *v1alpha1.EventS log.Info("deleting eventsource") if controllerutil.ContainsFinalizer(eventSource, finalizerName) { // Finalizer logic should be added here. - if err := r.finalize(ctx, eventSource); err != nil { - return err - } controllerutil.RemoveFinalizer(eventSource, finalizerName) } return nil @@ -97,16 +93,6 @@ func (r *reconciler) reconcile(ctx context.Context, eventSource *v1alpha1.EventS return Reconcile(r.client, args, log) } -func (r *reconciler) finalize(ctx context.Context, eventSource *v1alpha1.EventSource) error { - // Clean up Lease objects if there's any - if err := r.client.DeleteAllOf(ctx, &coordinationv1.Lease{}, - client.InNamespace(eventSource.Namespace), - client.MatchingFields{"metadata.name": "eventsource-" + eventSource.Name}); err != nil { - return err - } - return nil -} - func (r *reconciler) needsUpdate(old, new *v1alpha1.EventSource) bool { if old == nil { return true diff --git a/controllers/eventsource/resource.go b/controllers/eventsource/resource.go index 0533a2a961..341246188f 100644 --- a/controllers/eventsource/resource.go +++ b/controllers/eventsource/resource.go @@ -20,8 +20,6 @@ import ( "github.com/argoproj/argo-events/common" controllerscommon "github.com/argoproj/argo-events/controllers/common" - "github.com/argoproj/argo-events/eventsources" - apicommon "github.com/argoproj/argo-events/pkg/apis/common" eventbusv1alpha1 "github.com/argoproj/argo-events/pkg/apis/eventbus/v1alpha1" "github.com/argoproj/argo-events/pkg/apis/eventsource/v1alpha1" ) @@ -221,9 +219,14 @@ func buildDeployment(args *AdaptorArgs, eventBus *eventbusv1alpha1.EventBus) (*a }, }, }) + emptyDirVolName := "tmp-volume" + volumes = append(volumes, corev1.Volume{ + Name: emptyDirVolName, VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}, + }) deploymentSpec.Template.Spec.Volumes = volumes volumeMounts := deploymentSpec.Template.Spec.Containers[0].VolumeMounts volumeMounts = append(volumeMounts, corev1.VolumeMount{Name: "auth-volume", MountPath: common.EventBusAuthFileMountPath}) + volumeMounts = append(volumeMounts, corev1.VolumeMount{Name: emptyDirVolName, MountPath: "/tmp"}) deploymentSpec.Template.Spec.Containers[0].VolumeMounts = volumeMounts } } else { @@ -332,26 +335,6 @@ func buildDeploymentSpec(args *AdaptorArgs) (*appv1.DeploymentSpec, error) { spec.Template.Spec.PriorityClassName = args.EventSource.Spec.Template.PriorityClassName spec.Template.Spec.Priority = args.EventSource.Spec.Template.Priority } - allEventTypes := eventsources.GetEventingServers(args.EventSource, nil) - recreateTypes := make(map[apicommon.EventSourceType]bool) - for _, esType := range apicommon.RecreateStrategyEventSources { - recreateTypes[esType] = true - } - recreates := 0 - for eventType := range allEventTypes { - if _, ok := recreateTypes[eventType]; ok { - recreates++ - break - } - } - if recreates > 0 && replicas == 1 { - // For those event types, if there's only 1 replica, use recreate strategy. - // If replicas > 1, which means HA is available for them, rolling update strategy - // is better. - spec.Strategy = appv1.DeploymentStrategy{ - Type: appv1.RecreateDeploymentStrategyType, - } - } return spec, nil } diff --git a/controllers/sensor/controller.go b/controllers/sensor/controller.go index f563deaf81..93c735c7f2 100644 --- a/controllers/sensor/controller.go +++ b/controllers/sensor/controller.go @@ -20,7 +20,6 @@ import ( "context" "go.uber.org/zap" - coordinationv1 "k8s.io/api/coordination/v1" "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" @@ -87,9 +86,6 @@ func (r *reconciler) reconcile(ctx context.Context, sensor *v1alpha1.Sensor) err log.Info("deleting sensor") if controllerutil.ContainsFinalizer(sensor, finalizerName) { // Finalizer logic should be added here. - if err := r.finalize(ctx, sensor); err != nil { - return err - } controllerutil.RemoveFinalizer(sensor, finalizerName) } return nil @@ -113,16 +109,6 @@ func (r *reconciler) reconcile(ctx context.Context, sensor *v1alpha1.Sensor) err return Reconcile(r.client, args, log) } -func (r *reconciler) finalize(ctx context.Context, sensor *v1alpha1.Sensor) error { - // Clean up Lease objects if there's any - if err := r.client.DeleteAllOf(ctx, &coordinationv1.Lease{}, - client.InNamespace(sensor.Namespace), - client.MatchingFields{"metadata.name": "sensor-" + sensor.Name}); err != nil { - return err - } - return nil -} - func (r *reconciler) needsUpdate(old, new *v1alpha1.Sensor) bool { if old == nil { return true diff --git a/controllers/sensor/resource.go b/controllers/sensor/resource.go index cd733186f1..6aa4634b40 100644 --- a/controllers/sensor/resource.go +++ b/controllers/sensor/resource.go @@ -185,9 +185,14 @@ func buildDeployment(args *AdaptorArgs, eventBus *eventbusv1alpha1.EventBus) (*a }, }, }) + emptyDirVolName := "tmp-volume" + volumes = append(volumes, corev1.Volume{ + Name: emptyDirVolName, VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}, + }) deploymentSpec.Template.Spec.Volumes = volumes volumeMounts := deploymentSpec.Template.Spec.Containers[0].VolumeMounts volumeMounts = append(volumeMounts, corev1.VolumeMount{Name: "auth-volume", MountPath: common.EventBusAuthFileMountPath}) + volumeMounts = append(volumeMounts, corev1.VolumeMount{Name: emptyDirVolName, MountPath: "/tmp"}) deploymentSpec.Template.Spec.Containers[0].VolumeMounts = volumeMounts } } else { @@ -270,10 +275,6 @@ func buildDeploymentSpec(args *AdaptorArgs) (*appv1.DeploymentSpec, error) { MatchLabels: args.Labels, }, Replicas: &replicas, - Strategy: appv1.DeploymentStrategy{ - // Event bus does not allow multiple clients with same clientID to connect to the server at the same time. - Type: appv1.RecreateDeploymentStrategyType, - }, Template: corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Labels: podTemplateLabels, diff --git a/docs/eventsources/ha.md b/docs/eventsources/ha.md index a761a14de5..dde7cc56da 100644 --- a/docs/eventsources/ha.md +++ b/docs/eventsources/ha.md @@ -51,31 +51,6 @@ old one is gone. - Calendar - Generic -### RBAC - -To achieve `Active-Passive` strategy for these EventSources, a Service Account -with extra RBAC settings is needed. The Service Account needs to be bound to a -Role like following, and specified in the spec through -`spec.template.serviceAccountName`. - -```yaml -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: lease-role -rules: - - apiGroups: - - coordination.k8s.io - resources: - - leases - resourceNames: - - eventsource-{event-source-name} - verbs: - - "*" -``` - -**NOTE: This is not requried if `spec.replicas = 1`.** - ## More Check [this](../dr_ha_recommendations.md) out to learn more information about diff --git a/docs/sensors/ha.md b/docs/sensors/ha.md index 9779517024..3507344de6 100644 --- a/docs/sensors/ha.md +++ b/docs/sensors/ha.md @@ -9,30 +9,6 @@ elected to be active if the old one is gone. **Please DO NOT manually scale up the replicas, that might cause unexpected behaviors!** -## RBAC - -To achieve HA for Sensor Pods, a Service Account with extra RBAC settings is -needed. The Service Account needs to be bound to a Role like following, and -specified in the spec through `spec.template.serviceAccountName`. - -```yaml -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: lease-role -rules: - - apiGroups: - - coordination.k8s.io - resources: - - leases - resourceNames: - - sensor-{sensor-name) - verbs: - - "*" -``` - -**NOTE: This is not requried if `spec.replicas = 1`.** - ## More Check [this](../dr_ha_recommendations.md) out to learn more information about diff --git a/docs/service-accounts.md b/docs/service-accounts.md index 40a79fb9ed..49d13653ff 100644 --- a/docs/service-accounts.md +++ b/docs/service-accounts.md @@ -4,10 +4,9 @@ A `Service Account` can be specified in the EventSource object with `spec.template.serviceAccountName`, however it is not needed for all the -EventSource types except `resource`, unless you want to achieve -[HA](eventsources/ha.md) for some of them. For a `resource` EventSource, you -need to specify a Service Accout and give it `list` and `watch` permissions for -the resource being watched. +EventSource types except `resource`. For a `resource` EventSource, you need to +specify a Service Accout and give it `list` and `watch` permissions for the +resource being watched. For example, if you want to watch actions on `Deployment` objects, you need to: @@ -31,8 +30,7 @@ For example, if you want to watch actions on `Deployment` objects, you need to: A `Service Account` also can be specified in a Sensor object via `spec.template.serviceAccountName`, this is only needed when `k8s` trigger or -`argoWorkflow` trigger is defined in the Sensor object, or you want to run the -Sensor with [HA](sensors/ha.md). +`argoWorkflow` trigger is defined in the Sensor object. The sensor examples provided by us use `argo-events-sa` service account to execute the triggers, but it has more permissions than needed, and you may want diff --git a/eventsources/eventing.go b/eventsources/eventing.go index e2f76c9303..eb5c7b52e9 100644 --- a/eventsources/eventing.go +++ b/eventsources/eventing.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "math/rand" - "os" "strings" "sync" "time" @@ -14,12 +13,9 @@ import ( "github.com/google/uuid" "github.com/pkg/errors" "go.uber.org/zap" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/tools/leaderelection" - "k8s.io/client-go/tools/leaderelection/resourcelock" "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/common/leaderelection" "github.com/argoproj/argo-events/common/logging" "github.com/argoproj/argo-events/eventbus" eventbusdriver "github.com/argoproj/argo-events/eventbus/driver" @@ -284,54 +280,24 @@ func (e *EventSourceAdaptor) Start(ctx context.Context) error { // EventSource object use the same type of deployment strategy break } - if !isRecreatType || e.eventSource.Spec.GetReplicas() == 1 { + if !isRecreatType { return e.run(ctx, servers) } - kubeConfig, _ := os.LookupEnv(common.EnvVarKubeConfig) - restConfig, err := common.GetClientConfig(kubeConfig) + custerName := fmt.Sprintf("%s-eventsource-%s", e.eventSource.Namespace, e.eventSource.Name) + elector, err := leaderelection.NewEventBusElector(ctx, *e.eventBusConfig, custerName, int(e.eventSource.Spec.GetReplicas())) if err != nil { - return errors.Wrapf(err, "failed to get a K8s rest config for the event source %s", e.eventSource.Name) + log.Errorw("failed to get an elector", zap.Error(err)) + return err } - kubeClient := kubernetes.NewForConfigOrDie(restConfig) - - leaseName := "eventsource-" + e.eventSource.Name - lock := &resourcelock.LeaseLock{ - LeaseMeta: metav1.ObjectMeta{ - Name: leaseName, - Namespace: e.eventSource.Namespace, - }, - Client: kubeClient.CoordinationV1(), - LockConfig: resourcelock.ResourceLockConfig{ - Identity: e.hostname, + elector.RunOrDie(ctx, leaderelection.LeaderCallbacks{ + OnStartedLeading: func(ctx context.Context) { + if err := e.run(ctx, servers); err != nil { + log.Errorw("failed to start", zap.Error(err)) + } }, - } - - ctx, cancel := context.WithCancel(ctx) - leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ - Lock: lock, - ReleaseOnCancel: true, - LeaseDuration: 60 * time.Second, - RenewDeadline: 15 * time.Second, - RetryPeriod: 5 * time.Second, - Callbacks: leaderelection.LeaderCallbacks{ - OnStartedLeading: func(ctx context.Context) { - if err := e.run(ctx, servers); err != nil { - log.Errorw("failed to start", zap.Error(err)) - cancel() - } - }, - OnStoppedLeading: func() { - log.Infof("leader lost: %s", e.hostname) - cancel() - }, - OnNewLeader: func(identity string) { - if identity == e.hostname { - log.Infof("I am the new leader: %s", identity) - return - } - log.Infof("stand by, new leader elected: %s", identity) - }, + OnStoppedLeading: func() { + log.Infof("leader lost: %s", e.hostname) }, }) diff --git a/go.mod b/go.mod index 0cee9cdbb0..1dc4f855c8 100644 --- a/go.mod +++ b/go.mod @@ -52,9 +52,9 @@ require ( github.com/mitchellh/reflectwalk v1.0.1 // indirect github.com/nats-io/gnatsd v1.4.1 // indirect github.com/nats-io/go-nats v1.7.2 + github.com/nats-io/graft v0.0.0-20200605173148-348798afea05 github.com/nats-io/nats-streaming-server v0.17.0 // indirect - github.com/nats-io/nats.go v1.9.1 - github.com/nats-io/nkeys v0.1.4 // indirect + github.com/nats-io/nats.go v1.10.0 github.com/nats-io/stan.go v0.6.0 github.com/nicksnyder/go-i18n v1.10.1-0.20190510212457-b280125b035a // indirect github.com/nsqio/go-nsq v1.0.8 diff --git a/go.sum b/go.sum index 89c7f8928a..74442e470f 100644 --- a/go.sum +++ b/go.sum @@ -739,16 +739,20 @@ github.com/nats-io/gnatsd v1.4.1 h1:RconcfDeWpKCD6QIIwiVFcvForlXpWeJP7i5/lDLy44= github.com/nats-io/gnatsd v1.4.1/go.mod h1:nqco77VO78hLCJpIcVfygDP2rPGfsEHkGTUk94uh5DQ= github.com/nats-io/go-nats v1.7.2 h1:cJujlwCYR8iMz5ofZSD/p2WLW8FabhkQ2lIEVbSvNSA= github.com/nats-io/go-nats v1.7.2/go.mod h1:+t7RHT5ApZebkrQdnn6AhQJmhJJiKAvJUio1PiiCtj0= +github.com/nats-io/graft v0.0.0-20200605173148-348798afea05 h1:wF/dApMICOCM+/c/1dpFxooYGwmSUvclQMT9CRjnEbM= +github.com/nats-io/graft v0.0.0-20200605173148-348798afea05/go.mod h1:idnzXeCwCx69FMg+R0DyD4/OhrF1A+v3BqF5xSz+tS4= github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg= github.com/nats-io/jwt v0.3.2 h1:+RB5hMpXUUA2dfxuhBTEkMOrYmM+gKIZYS1KjSostMI= github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU= github.com/nats-io/nats-server/v2 v2.1.2/go.mod h1:Afk+wRZqkMQs/p45uXdrVLuab3gwv3Z8C4HTBu8GD/k= -github.com/nats-io/nats-server/v2 v2.1.4 h1:BILRnsJ2Yb/fefiFbBWADpViGF69uh4sxe8poVDQ06g= github.com/nats-io/nats-server/v2 v2.1.4/go.mod h1:Jw1Z28soD/QasIA2uWjXyM9El1jly3YwyFOuR8tH1rg= +github.com/nats-io/nats-server/v2 v2.1.7 h1:jCoQwDvRYJy3OpOTHeYfvIPLP46BMeDmH7XEJg/r42I= +github.com/nats-io/nats-server/v2 v2.1.7/go.mod h1:rbRrRE/Iv93O/rUvZ9dh4NfT0Cm9HWjW/BqOWLGgYiE= github.com/nats-io/nats-streaming-server v0.17.0 h1:eYhSmjRmRsCYNsoUshmZ+RgKbhq6B+7FvMHXo3M5yMs= github.com/nats-io/nats-streaming-server v0.17.0/go.mod h1:ewPBEsmp62Znl3dcRsYtlcfwudxHEdYMtYqUQSt4fE0= -github.com/nats-io/nats.go v1.9.1 h1:ik3HbLhZ0YABLto7iX80pZLPw/6dx3T+++MZJwLnMrQ= github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzEE/Zbp4w= +github.com/nats-io/nats.go v1.10.0 h1:L8qnKaofSfNFbXg0C5F71LdjPRnmQwSsA4ukmkt1TvY= +github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE= github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w= github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w= github.com/nats-io/nkeys v0.1.4 h1:aEsHIssIk6ETN5m2/MD8Y4B2X7FfXrBAUdkyRvbVYzA= diff --git a/manifests/cluster-install/rbac/argo-events-cluster-role.yaml b/manifests/cluster-install/rbac/argo-events-cluster-role.yaml index 284abfd3a7..5cc567b6bb 100644 --- a/manifests/cluster-install/rbac/argo-events-cluster-role.yaml +++ b/manifests/cluster-install/rbac/argo-events-cluster-role.yaml @@ -89,9 +89,3 @@ rules: - update - patch - delete - - apiGroups: - - coordination.k8s.io - resources: - - leases - verbs: - - "*" \ No newline at end of file diff --git a/manifests/install.yaml b/manifests/install.yaml index bf597e815e..1558173928 100644 --- a/manifests/install.yaml +++ b/manifests/install.yaml @@ -297,12 +297,6 @@ rules: - update - patch - delete -- apiGroups: - - coordination.k8s.io - resources: - - leases - verbs: - - '*' --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/manifests/namespace-install.yaml b/manifests/namespace-install.yaml index b1f4412706..9a48ad4734 100644 --- a/manifests/namespace-install.yaml +++ b/manifests/namespace-install.yaml @@ -202,12 +202,6 @@ rules: - update - patch - delete -- apiGroups: - - coordination.k8s.io - resources: - - leases - verbs: - - '*' --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding diff --git a/manifests/namespace-install/rbac/argo-events-role.yaml b/manifests/namespace-install/rbac/argo-events-role.yaml index 11542abe65..f893e36435 100644 --- a/manifests/namespace-install/rbac/argo-events-role.yaml +++ b/manifests/namespace-install/rbac/argo-events-role.yaml @@ -75,9 +75,3 @@ rules: - update - patch - delete - - apiGroups: - - coordination.k8s.io - resources: - - leases - verbs: - - "*" \ No newline at end of file diff --git a/sensors/listener.go b/sensors/listener.go index d1ccdb4a99..a039d46f5f 100644 --- a/sensors/listener.go +++ b/sensors/listener.go @@ -31,10 +31,9 @@ import ( "github.com/pkg/errors" "go.uber.org/zap" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/tools/leaderelection" - "k8s.io/client-go/tools/leaderelection/resourcelock" "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/common/leaderelection" "github.com/argoproj/argo-events/common/logging" "github.com/argoproj/argo-events/eventbus" eventbusdriver "github.com/argoproj/argo-events/eventbus/driver" @@ -65,51 +64,23 @@ func (sensorCtx *SensorContext) getGroupAndClientID(depExpression string) (strin } func (sensorCtx *SensorContext) Start(ctx context.Context) error { - if sensorCtx.sensor.Spec.GetReplicas() == 1 { - return sensorCtx.listenEvents(ctx) + log := logging.FromContext(ctx) + custerName := fmt.Sprintf("%s-sensor-%s", sensorCtx.sensor.Namespace, sensorCtx.sensor.Name) + elector, err := leaderelection.NewEventBusElector(ctx, *sensorCtx.eventBusConfig, custerName, int(sensorCtx.sensor.Spec.GetReplicas())) + if err != nil { + log.Errorw("failed to get an elector", zap.Error(err)) + return err } - - leaseName := "sensor-" + sensorCtx.sensor.Name - lock := &resourcelock.LeaseLock{ - LeaseMeta: metav1.ObjectMeta{ - Name: leaseName, - Namespace: sensorCtx.sensor.Namespace, + elector.RunOrDie(ctx, leaderelection.LeaderCallbacks{ + OnStartedLeading: func(ctx context.Context) { + if err := sensorCtx.listenEvents(ctx); err != nil { + log.Errorw("failed to start", zap.Error(err)) + } }, - Client: sensorCtx.kubeClient.CoordinationV1(), - LockConfig: resourcelock.ResourceLockConfig{ - Identity: sensorCtx.hostname, - }, - } - - log := logging.FromContext(ctx) - ctx, cancel := context.WithCancel(ctx) - leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ - Lock: lock, - ReleaseOnCancel: true, - LeaseDuration: 60 * time.Second, - RenewDeadline: 15 * time.Second, - RetryPeriod: 5 * time.Second, - Callbacks: leaderelection.LeaderCallbacks{ - OnStartedLeading: func(ctx context.Context) { - if err := sensorCtx.listenEvents(ctx); err != nil { - log.Errorw("failed to start", zap.Error(err)) - cancel() - } - }, - OnStoppedLeading: func() { - log.Infof("leader lost: %s", sensorCtx.hostname) - cancel() - }, - OnNewLeader: func(identity string) { - if identity == sensorCtx.hostname { - log.Infof("I am the new leader: %s", identity) - return - } - log.Infof("stand by, new leader elected: %s", identity) - }, + OnStoppedLeading: func() { + log.Infof("leader lost: %s", sensorCtx.hostname) }, }) - return nil } diff --git a/test/util/util.go b/test/util/util.go index 4d287ddbce..32e3dc0a07 100644 --- a/test/util/util.go +++ b/test/util/util.go @@ -8,9 +8,7 @@ import ( "time" appsv1 "k8s.io/api/apps/v1" - coordinationv1 "k8s.io/api/coordination/v1" corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" @@ -253,28 +251,8 @@ func EventSourcePodLogContains(ctx context.Context, kubeClient kubernetes.Interf if err != nil { return false, fmt.Errorf("error getting event source pod name: %w", err) } - podName := podList.Items[0].GetName() - if len(podList.Items) > 1 { - // HA - var l *coordinationv1.Lease - for { - l, err = kubeClient.CoordinationV1().Leases(namespace).Get(ctx, "eventsource-"+eventSourceName, metav1.GetOptions{}) - if err != nil { - if !apierrors.IsNotFound(err) { - return false, fmt.Errorf("error getting event source Lease information: %w", err) - } else if err != nil && apierrors.IsNotFound(err) { - time.Sleep(2 * time.Second) - continue - } - } - break - } - podName = *l.Spec.HolderIdentity - } - fmt.Printf("EventSource POD name: %s\n", podName) - cctx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() - return podLogContains(cctx, kubeClient, namespace, podName, regex) + + return PodsLogContains(ctx, kubeClient, namespace, regex, podList, timeout), nil } func SensorPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, sensorName, regex string, timeout time.Duration) (bool, error) { @@ -283,28 +261,41 @@ func SensorPodLogContains(ctx context.Context, kubeClient kubernetes.Interface, if err != nil { return false, fmt.Errorf("error getting sensor pod name: %w", err) } - podName := podList.Items[0].GetName() - if len(podList.Items) > 1 { - // HA - var l *coordinationv1.Lease - for { - l, err = kubeClient.CoordinationV1().Leases(namespace).Get(ctx, "sensor-"+sensorName, metav1.GetOptions{}) + + return PodsLogContains(ctx, kubeClient, namespace, regex, podList, timeout), nil +} + +func PodsLogContains(ctx context.Context, kubeClient kubernetes.Interface, namespace, regex string, podList *corev1.PodList, timeout time.Duration) bool { + cctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + errChan := make(chan error) + resultChan := make(chan bool) + for _, p := range podList.Items { + go func(podName string) { + fmt.Printf("Watching POD: %s\n", podName) + contains, err := podLogContains(cctx, kubeClient, namespace, podName, regex) if err != nil { - if !apierrors.IsNotFound(err) { - return false, fmt.Errorf("error getting sensor Lease information: %w", err) - } else if err != nil && apierrors.IsNotFound(err) { - time.Sleep(2 * time.Second) - continue - } + errChan <- err + return } - break + if contains { + resultChan <- true + } + }(p.Name) + } + + for { + select { + case <-cctx.Done(): + return false + case result := <-resultChan: + if result { + return true + } + case err := <-errChan: + fmt.Printf("error: %v", err) } - podName = *l.Spec.HolderIdentity } - fmt.Printf("Sensor POD name: %s\n", podName) - cctx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() - return podLogContains(cctx, kubeClient, namespace, podName, regex) } func podLogContains(ctx context.Context, client kubernetes.Interface, namespace, podName, regex string) (bool, error) {