feat: Allow to base64 decode messages from Azure Queue Storage (#2627)

Signed-off-by: eduardodbr <eduardodbr@hotmail.com>

api/jsonschema/schema.json

@@ -941,6 +941,10 @@
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
           "description": "ConnectionString is the connection string to access Azure Queue Storage. If this fields is not provided it will try to access via Azure AD with StorageAccountName."
         },
+        "decodeMessage": {
+          "description": "DecodeMessage specifies if all the messages should be base64 decoded. If set to true the decoding is done before the evaluation of JSONBody",
+          "type": "boolean"
+        },
         "dlq": {
           "description": "DLQ specifies if a dead-letter queue is configured for messages that can't be processed successfully. If set to true, messages with invalid payload won't be acknowledged to allow to forward them farther to the dead-letter queue. The default value is false.",
           "type": "boolean"

eventsources/sources/azurequeuestorage/start.go

@@ -2,6 +2,7 @@ package azurequeuestorage
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"time"
@@ -126,11 +127,23 @@ func (el *EventListener) processMessage(message *azqueue.DequeuedMessage, dispat
 		InsertionTime: *message.InsertionTime,
 		Metadata:      el.AzureQueueStorageEventSource.Metadata,
 	}
+	body := []byte(*message.MessageText)
+	if el.AzureQueueStorageEventSource.DecodeMessage {
+		rawDecodedText, err := base64.RawURLEncoding.DecodeString(*message.MessageText)
+		if err != nil {
+			log.Errorw("failed to base64 decode message...", zap.Error(err))
+			el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
+			if !el.AzureQueueStorageEventSource.DLQ {
+				ack()
+			}
+			return
+		}
+		body = rawDecodedText
+	}
 	if el.AzureQueueStorageEventSource.JSONBody {
-		body := []byte(*message.MessageText)
 		data.Body = (*json.RawMessage)(&body)
 	} else {
-		data.Body = []byte(*message.MessageText)
+		data.Body = body
 	}
 	eventBytes, err := json.Marshal(data)
 	if err != nil {

examples/event-sources/azure-queue-storage.yaml

@@ -10,6 +10,8 @@ spec:
       # jsonBody specifies that all event body payload coming from this
       # source will be JSON
       jsonBody: true
+      # DecodeMessage specifies if all the messages from AQS should be base64 decoded
+      decodeMessage: false
       # connection string contains information about K8s secret that stores the connection string
       connectionString:
         # Key within the K8s secret whose corresponding value (must be base64 encoded) is access key

pkg/apis/eventsource/v1alpha1/types.go

@@ -1223,6 +1223,10 @@ type AzureQueueStorageEventSource struct {
 	// Filter
 	// +optional
 	Filter *EventSourceFilter `json:"filter,omitempty" protobuf:"bytes,7,opt,name=filter"`
+	// DecodeMessage specifies if all the messages should be base64 decoded.
+	// If set to true the decoding is done before the evaluation of JSONBody
+	// +optional
+	DecodeMessage bool `json:"decodeMessage,omitempty" protobuf:"bytes,8,opt,name=decodeMessage"`
 }
 
 // StripeEventSource describes the event source for stripe webhook notifications