Kafka EventBus #10

dfarr · 2023-01-11T20:10:12Z

Implement Kafka EventBus

controllers/eventbus/controller.go

AalokAhluwalia · 2023-02-28T17:55:42Z

eventbus/driver.go

@@ -35,12 +37,12 @@ func GetEventSourceDriver(ctx context.Context, eventBusConfig eventbusv1alpha1.B

 	var eventBusType apicommon.EventBusType
 	switch {
-	case eventBusConfig.NATS != nil && eventBusConfig.JetStream != nil:


Might be a good check to keep in to make sure multiple specs aren't defined.

But permutations increase exponentially, so it quickly gets out of control

1. eventBusConfig.NATS != nil && eventBusConfig.JetStream != nil 2. eventBusConfig.NATS != nil && eventBusConfig.KafkaStream != nil 3. eventBusConfig.JetStream != nil && eventBusConfig.KafkaStream != nil 4. eventBusConfig.NATS != nil && eventBusConfig.JetStream != nil && eventBusConfig.KafkaStream != nil

Yuck!

AalokAhluwalia · 2023-02-28T17:58:30Z

eventbus/driver.go

@@ -79,12 +83,12 @@ func GetSensorDriver(ctx context.Context, eventBusConfig eventbusv1alpha1.BusCon

 	var eventBusType apicommon.EventBusType
 	switch {
-	case eventBusConfig.NATS != nil && eventBusConfig.JetStream != nil:


Same comment as above

AalokAhluwalia · 2023-02-28T18:28:02Z

eventbus/kafka/base/kafka.go

+	// consumer config
+	config.Consumer.IsolationLevel = sarama.ReadCommitted
+	config.Consumer.Offsets.AutoCommit.Enable = false
+


Shouldn't this Config method also set the ConsumerGroup.GroupName?

GroupName is specified on the consumer, not the config. You can see this here

AalokAhluwalia · 2023-02-28T18:47: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":


In sensors/listener.go you mention kafka doesn't require leader election? If sources don't do leader election also, maybe return an error mentioning that in this function

Eventsources still need a leader election (under some circumstances, it's a little confusing). So unfortunately we still need this for the event source.

Here's the logic that determines if the eventsource needs a leader election or not. The important thing to note is there are circumstances where an eventsource with a kafka eventbus requires a leader election.

AalokAhluwalia · 2023-03-01T17:52:13Z

eventbus/kafka/sensor/kafka_sensor.go

+func NewKafkaSensor(kafkaConfig *eventbusv1alpha1.KafkaConfig, sensor *sensorv1alpha1.Sensor, hostname string, logger *zap.SugaredLogger) *KafkaSensor {
+	topics := &Topics{
+		event:   kafkaConfig.Topic,
+		trigger: fmt.Sprintf("%s.%s.%s", kafkaConfig.Topic, sensor.Name, "trigger"),


For the trigger topics, same comment as before, should it also include the namespace to include the possibility that there are two sensors in different namespaces with the same name.

Yes definitely! The namespace is already included in the topic name itself here

But I think we should think through the option of the user specifying the topic names manually - do you think users may want to provide the trigger and action topic name manually too?

Personally I feel it's sufficient to (optionally) have the user specify the eventbus {event} topic and then derive the {trigger, action} topic names from the name of this topic (as I do currently)

AalokAhluwalia · 2023-03-01T17:58:27Z

eventbus/kafka/sensor/kafka_sensor.go

+	}
+
+	if _, ok := s.triggers[triggerName]; !ok {
+		expr, err := govaluate.NewEvaluableExpression(strings.ReplaceAll(depExpression, "-", "\\-"))


What is line 165 NewEvaluableExpression(strings.ReplaceAll for ? Does the argo events code do this in other places?

This creates a new evaluatable expression from the string that's provided in the trigger's condition field. It's the same library as used today in jetstream here.

AalokAhluwalia · 2023-03-01T18:00:49Z

eventbus/kafka/sensor/kafka_sensor.go

+	defer s.Disconnect()
+
+	for {
+		if len(s.triggers) != len(s.sensor.Spec.Triggers) || !s.triggers.Ready() {


Is it possible that the triggers would never be ready and this would loop forever. Would it be helpful to add a max wait duration?

Doesn't hurt, I can add one!

So I implemented this and then I realized, even if we waited a max duration and return, the function would just be re-invoked in sensors/listener.go anyways. Furthermore, I realized that Connect and Subscribe are guaranteed to be called for each trigger (and if this happens then the if condition will be satisfied) so I don't think this is actually necessary.

eventbus/kafka/sensor/kafka_sensor.go

AalokAhluwalia · 2023-03-01T18:09:01Z

eventbus/kafka/sensor/kafka_sensor.go

+	return messages, offset, nil
+}
+
+func (s *KafkaSensor) Action(msg *sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func()) {


Looks like for event/action/trigger the second return value is always msg.offset + 1. Maybe remove this as a return value if it's always the same

It's not always the same, the trigger is the odd one out, it is set to the smallest required offset across all triggers. That value is set here

AalokAhluwalia · 2023-03-01T18:15:55Z

And then just one genreal question for the above triggering logic. Suppose you have the condition (a&&b) || (b&&c) and event b has already come in so the expression looks like (a && true) || (true && c). What will happen if in the next batch, messages a and c come. Will there be two triggers?

dfarr · 2023-03-01T22:21:57Z

And then just one genreal question for the above triggering logic. Suppose you have the condition (a&&b) || (b&&c) and event b has already come in so the expression looks like (a && true) || (true && c). What will happen if in the next batch, messages a and c come. Will there be two triggers?

The ordering of messages in a batch is guaranteed (thanks kafka) so the a event would satisfy the trigger (a message to the action topic would be published), the triggers state would be reset, and then the first message added to the fresh state of the trigger would be the c event.

So stepping through this (the number preceding the event represents the offset)

batch 1:
0: b

state:
(a && true) || (true && c)

transaction:
offset: 0
messages: []

batch 2:
1: a
2: c

state:
(a && b) || (b && true)

transaction:
offset: 2
messages: [action]