Merge pull request #150 from odacremolbap/task/make-redis-tracking-id…

…-opt-in

Redis: make redis tracking id opt-in

README.md

@@ -227,6 +227,7 @@ redis.tls-skip-verify     | REDIS_TLS_SKIP_VERIFY           | false | TLS skippi
 redis.tls-ca-certificate  | REDIS_TLS_CA_CERTIFICATE        | | TLS CA certificate used to connect to Redis.
 redis.tls-certificate     | REDIS_TLS_CERTIFICATE           | | TLS certificate used to authenticate with Redis.
 redis.tls-key             | REDIS_TLS_KEY                   | | TLS key used to authenticate with Redis.
+redis.tracking-id-enabled | REDIS_TRACKING_ID_ENABLED       | false | Adds the Redis ID for the event as `triggermeshbackendid` CloudEvents attribute.
 redis.stream              | REDIS_STREAM                    | triggermesh | Stream name that stores the broker's CloudEvents.
 redis.group               | REDIS_GROUP                     | default | Redis stream consumer group name.
 redis.stream-max-len      | REDIS_STREAM_MAX_LEN            | 1000 | Limit the number of items in a stream by trimming it. Set to 0 for unlimited.

pkg/backend/impl/redis/cmd.go

@@ -26,7 +26,8 @@ type RedisArgs struct {
 	// Instance at the Redis stream consumer group. Copied from the InstanceName at the global args.
 	Instance string `kong:"-"`
 
-	StreamMaxLen int `help:"Limit the number of items in a stream by trimming it. Set to 0 for unlimited." env:"STREAM_MAX_LEN" default:"1000"`
+	StreamMaxLen      int  `help:"Limit the number of items in a stream by trimming it. Set to 0 for unlimited." env:"STREAM_MAX_LEN" default:"1000"`
+	TrackingIDEnabled bool `help:"Enables adding Redis ID as a CloudEvent attribute." env:"TRACKING_ID_ENABLED" default:"false"`
 }
 
 func (ra *RedisArgs) Validate() error {

pkg/backend/impl/redis/redis.go

@@ -226,6 +226,8 @@ func (s *redis) Subscribe(name string, ccb backend.ConsumerDispatcher) error {
 		name:     name,
 		group:    group,
 
+		trackingEnabled: s.args.TrackingIDEnabled,
+
 		// caller's callback for dispatching events from Redis.
 		ccbDispatch: ccb,
 

pkg/backend/impl/redis/subscription.go

@@ -27,6 +27,8 @@ type subscription struct {
 	name     string
 	group    string
 
+	trackingEnabled bool
+
 	// caller's callback for dispatching events from Redis.
 	ccbDispatch backend.ConsumerDispatcher
 
@@ -132,20 +134,20 @@ func (s *subscription) start() {
 					continue
 				}
 
-				if err = ce.Context.SetExtension(BackendIDAttribute, msg.ID); err != nil {
-					s.logger.Errorw(fmt.Sprintf("could not set %s attributes for the Redis message %s. Tracking will not be possible.", BackendIDAttribute, msg.ID),
-						zap.Error(err))
+				if s.trackingEnabled {
+					if err = ce.Context.SetExtension(BackendIDAttribute, msg.ID); err != nil {
+						s.logger.Errorw(fmt.Sprintf("could not set %s attributes for the Redis message %s. Tracking will not be possible.", BackendIDAttribute, msg.ID),
+							zap.Error(err))
+					}
 				}
 
-				go func() {
+				go func(msgID string) {
 					s.ccbDispatch(ce)
-					id := ce.Extensions()[BackendIDAttribute].(string)
-
-					if err := s.ack(id); err != nil {
-						s.logger.Errorw(fmt.Sprintf("could not ACK the Redis message %s containing CloudEvent %s", id, ce.Context.GetID()),
+					if err := s.ack(msgID); err != nil {
+						s.logger.Errorw(fmt.Sprintf("could not ACK the Redis message %s containing CloudEvent %s", msgID, ce.Context.GetID()),
 							zap.Error(err))
 					}
-				}()
+				}(msg.ID)
 
 				// If we are processing pending messages the ACK might take a
 				// while to be sent. We need to set the message ID so that the