feat: Kafka EventBus #2502

dfarr · 2023-03-03T23:29:14Z

Kafka EventBus

An implementation of EventBus leveraging Kafka. Supports simple conditions, complex conditions (ones containing 'and' clauses), trigger condition resets, and both at-most-once and at-least-once trigger semantics. In contrast to existing EventBus implementations, Sensors are horizontally scalable (EventSources are sometimes horizontally scalable).

The implementation uses three topics, an event topic that EventSources produce to and Sensors consume from. A trigger topic that is used to maintain state and ensure resiliency for triggers with complex conditions. And finally an action topic used to decouple event processing from trigger actions. The event topic is shared between all EventSources and Sensors wired up to the same EventBus, whereas the trigger and action topics are specific to a single Sensor.

Default naming convention

topic	name
event	`{namespace}-{eventbus}`
trigger	`{namespace}-{eventbus}-{sensor}-trigger`
action	`{namespace}-{eventbus}-{sensor}-action`

High level architecture

For simple trigger conditions (such as t2) we can skip the trigger topic and publish messages directly to the action topic. For complex trigger conditions (such as t1) we use the trigger topic to hold on to events until the trigger condition is satisfied.

Please see my blog post for in-depth implementation details. The post poses a dilemma with the Kafka implementation that occurs under the following scenario. Imagine a Sensor with the following two triggers.

t1: a && b
t2: c && d

Assume the events below are received in the following order. Furthermore, assume all topics {event, trigger, action} contain only a single partition.

{a, c, d}

Trigger t2 will be invoked once, however, because all messages land on the same partition (there is only one) the dilemma is that we cannot bump our consumer’s offset as we need to hold on to the a event in order to maintain resiliency. But, imagine a restart occurs and we re-consume all three events starting with event a. How can we ensure that we do not re-invoke t2? To solve this problem we opted to maintain metadata alongside the offset.

The metadata maintains a mapping of triggers to offsets, for example:

{
  "t1": 0,
  "t2": 3
}

If ever a restart occurs and we start consuming from offset 0, however, any message pertaining to trigger t2 are skipped until we get to offset 3 and therefore no errant trigger action is invoked.

Checklist:

My organization is added to USERS.md.

dfarr · 2023-03-03T23:39:45Z

controllers/eventsource/resource.go

-	envs = append(envs, envVars...)
-	deploymentSpec.Template.Spec.Containers[0].Env = envs
+	// secrets
+	volSecrets, volSecretMounts := common.VolumesFromSecretsOrConfigMaps(common.SecretKeySelectorType, secretObjs...)


When the eventBus type is kafka, we need can use the common.VolumesFromSecretsOrConfigMaps function to attach the tls and sasl secrets. They need to be attached at the same time as the secrets in the sensor/eventsource so that any secrets with the same name will be deduplicated.

dfarr · 2023-03-03T23:43:04Z

test/e2e/testdata/es-durable-consumer.yaml

@@ -3,6 +3,8 @@ kind: EventSource
 metadata:
  name: e2e-durable-consumer
 spec:
+  template:
+    serviceAccountName: argo-events-sa


Added the argo-events-sa service account to tests as the kafka eventsource requires a kubernetes lease and the default service account does not have permission to create these objects in the ci tests

dfarr · 2023-03-03T23:44:05Z

.github/workflows/ci.yaml

@@ -123,6 +123,7 @@ jobs:
        include:
          - driver: stan
          - driver: jetstream
+          - driver: kafka


This executes all e2e tests against kafka as well

dfarr · 2023-03-03T23:45:10Z

eventbus/common/interface.go

@@ -42,5 +42,9 @@ type EventSourceDriver interface {

 type SensorDriver interface {
 	Initialize() error
-	Connect(triggerName string, dependencyExpression string, deps []Dependency) (TriggerConnection, error)
+	Connect(ctx context.Context,


Added context and atLeastOnce boolean to this interface, the stan and jetstream implementation just ignore these values

juliev0 · 2023-03-15T18:35:30Z

common/leaderelection/leaderelection.go

@@ -40,7 +40,7 @@ type LeaderCallbacks struct {

 func NewElector(ctx context.Context, eventBusConfig eventbusv1alpha1.BusConfig, clusterName string, clusterSize int, namespace string, leasename string, hostname string) (Elector, error) {
 	switch {
-	case strings.ToLower(os.Getenv(common.EnvVarLeaderElection)) == "k8s":
+	case eventBusConfig.Kafka != nil || strings.ToLower(os.Getenv(common.EnvVarLeaderElection)) == "k8s":
 		return newKubernetesElector(namespace, leasename, hostname)


So, if Kafka is defined we still do leader election here? (even though it's "master/master")

We still need the leader election for the eventsource. For example if an eventsource uses the calendar we need active/passive so that both pods aren't emitting calendar events at the same time.

Got it. And I see it's not being called on the Sensor side.

juliev0 · 2023-03-17T17:45:43Z

I assume you'll be adding docs for Kafka after this PR, right?

controllers/eventsource/resource.go

pkg/apis/eventbus/v1alpha1/kafka_eventbus.go

dfarr · 2023-03-17T18:11:49Z

We created a small test framework to verify correctness of the kafka eventbus under failure. The framework has a notion of chaos that randomly deletes one of the sensor pods on a schedule (every 30s, 1m, etc). In addition to the chaos we ran a few iterations of the tests by tweaking the following dimensions:

number of input events
number of dependencies
number of triggers (all possible permutations of the dependencies joined with the && operator)
replicas (for simplicity, replicas == (kafka) partitions)
semantics (either "at least once" or "at most once")

Under all scenarios tested we were able to verify correctness. Please note that this does not mean the trigger is always invoked, when "at least once" is specified we can verify that trigger invocations conform to this semantic and same for "at most once".

Attached are the results.
kafka-eventbus-tests.xlsx

eventbus/kafka/sensor/kafka_sensor.go

juliev0 · 2023-03-17T19:08:35Z

eventbus/kafka/sensor/kafka_sensor.go

+			s.Logger.Errorw("Kafka error", zap.Error(err))
+			return
+		}
+	}


That's interesting that all the logic is in the Sensor and not in the KafkaTriggerConnection - did it fit better to put it here?

(the Trigger-processing part that is...I know the Events are generic and all)

This implementation is top down, which is in contrast to how it works for jetstream. When an event is consumed from a kafka topic, the 3 handler functions (Event, Trigger, Action) are invoked depending on which topic the event was consumed from. These functions then use the Triggers map (part of the KafkaSensor struct) to send these messages to each trigger that requires it.

To maintain seperation, I try to keep all state in the TriggerConnection struct (the events) and the Sensor only interacts with the trigger connections by invoking the interface functions (which is a superset of the TriggerConnection interface).

Thanks. I've looked at it further since my comment. It's all very elegant.

Thank you :)

dfarr · 2023-03-17T19:41:27Z

I assume you'll be adding docs for Kafka after this PR, right?

Yes! Actually let me add at least the minimum docs to this PR and if we need more in depth details I'm happy to follow up with another PR.

juliev0 · 2023-03-17T20:26:14Z

eventbus/kafka/sensor/kafka_sensor.go

+	for _, trigger := range s.triggers {
+		offset = trigger.Offset(msg.Partition, offset)
+	}
+


can this handler be called by two threads at the same time or no? if so, any issues related to both threads executing lines 347-349 at the same time?

No, luckily this is not possible by construction. We use the kafka message key to look up the trigger here, a key always maps to only one partition and all messages pertaining to the same partition are processed sequentially (and by one pod). That means we don't have to worry about concurrency in our stateful code (mostly the KafkaTriggerConnection). This is also how we enable scaling - all events that pertain to a specific trigger will land on the same partition and are guaranteed to be processed by the same pod.

Right, I guess my confusion was that we were looking at other triggers outside of our own on line 347

I think I see what you're saying. The Offset() function only goes through events on this same partition, and all messages pertaining to the same partition are processed sequentially.

eventbus/kafka/sensor/kafka_handler.go

juliev0 · 2023-03-17T21:12:57Z

I assume you'll be adding docs for Kafka after this PR, right?

Yes! Actually let me add at least the minimum docs to this PR and if we need more in depth details I'm happy to follow up with another PR.

I'm fine if you want to get this in first before the docs - up to you (and Derek)

Makefile

common/leaderelection/leaderelection.go

eventbus/kafka/base/utils.go

pkg/apis/eventbus/v1alpha1/kafka_eventbus.go

sensors/listener.go

sensors/triggers/kafka/kafka.go

test/manifests/kafka/kafka.yaml