clean up cloudevents, make it look like this codebase. Add config exa…

…mple

Signed-off-by: Scott Nichols <[email protected]>

.gitignore

@@ -18,3 +18,5 @@ falcosidekick
 .idea
 *.swp
 /hack/tools/bin/*
+
+tmp/

README.md

@@ -377,6 +377,9 @@ The *env vars* "match" field names in *yaml file with this structure (**take car
 * **WEBHOOK_ADDRESS** : "" # Webhook address, if not empty, Webhook output is *enabled*
 * **WEBHOOK_CUSTOMHEADERS** : a list of comma separated custom headers to add, syntax is "key:value,key:value"
 * **WEBHOOK_MINIMUMPRIORITY** : minimum priority of event for using this output, order is `emergency|alert|critical|error|warning|notice|informational|debug or "" (default)`
+* **CLOUDEVENTS_ADDRESS** : "" # Webhook address, if not empty, Webhook output is *enabled*
+* **CLOUDEVENTS_EXTENSIONS** : a list of comma separated extensions to add, syntax is "key:value,key:value"
+* **CLOUDEVENTS_MINIMUMPRIORITY** : minimum priority of event for using this output, order is `emergency|alert|critical|error|warning|notice|informational|debug or "" (default)`
 * **AZURE_EVENTHUB_NAME**: Name of the Hub, if not empty, EventHub is *enabled*
 * **AZURE_EVENTHUB_NAMESPACE**: Name of the space the Hub is in
 * **AZURE_EVENTHUB_MINIMUMPRIORITY**: minimum priority of event for using this output, order is `emergency|alert|critical|error|warning|notice|informational|debug or "" (default)`

config.go

@@ -19,6 +19,7 @@ func getConfig() *types.Configuration {
 	c := &types.Configuration{
 		Customfields: make(map[string]string),
 		Webhook:      types.WebhookOutputConfig{CustomHeaders: make(map[string]string)},
+		CloudEvents:  types.CloudEventsOutputConfig{Extensions: make(map[string]string)},
 	}
 
 	configFile := kingpin.Flag("config-file", "config file").Short('c').ExistingFile()
@@ -152,6 +153,7 @@ func getConfig() *types.Configuration {
 
 	v.GetStringMapString("customfields")
 	v.GetStringMapString("Webhook.CustomHeaders")
+	v.GetStringMapString("CloudEvents.Extensions")
 	v.Unmarshal(c)
 
 	if value, present := os.LookupEnv("CUSTOMFIELDS"); present {
@@ -174,6 +176,16 @@ func getConfig() *types.Configuration {
 		}
 	}
 
+	if value, present := os.LookupEnv("CLOUDEVENTS_EXTENSIONS"); present {
+		customfields := strings.Split(value, ",")
+		for _, label := range customfields {
+			tagkeys := strings.Split(label, ":")
+			if len(tagkeys) == 2 {
+				c.CloudEvents.Extensions[tagkeys[0]] = tagkeys[1]
+			}
+		}
+	}
+
 	if c.ListenPort == 0 || c.ListenPort > 65536 {
 		log.Fatalf("[ERROR] : Bad port number\n")
 	}

config_example.yaml

@@ -118,6 +118,12 @@ webhook:
   #   key: value
   # minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|error|warning|notice|informational|debug or "" (default)
 
+cloudevents:
+# address: "" # CloudEvents consumer http address, if not empty, CloudEvents output is enabled
+# extensions: # Extensions to add in the outbound Event, useful for routing
+#   key: value
+# minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|error|warning|notice|informational|debug or "" (default)
+
 azure:
   eventHub:
     name: "" # Name of the Hub, if not empty, EventHub is enabled

outputs/client.go

@@ -45,19 +45,18 @@ var ErrClientCreation = errors.New("Client creation Error")
 
 // Client communicates with the different API.
 type Client struct {
-	OutputType      string
-	EndpointURL     *url.URL
-	Config          *types.Configuration
-	Stats           *types.Statistics
-	PromStats       *types.PromStatistics
-	AWSSession      *session.Session
-	StatsdClient    *statsd.Client
-	DogstatsdClient *statsd.Client
-	GCPTopicClient  *pubsub.Topic
-	KafkaProducer   *kafka.Conn
-	PagerdutyClient *pagerduty.Client
-
-	ce cloudevents.Client
+	OutputType        string
+	EndpointURL       *url.URL
+	Config            *types.Configuration
+	Stats             *types.Statistics
+	PromStats         *types.PromStatistics
+	AWSSession        *session.Session
+	StatsdClient      *statsd.Client
+	DogstatsdClient   *statsd.Client
+	GCPTopicClient    *pubsub.Topic
+	KafkaProducer     *kafka.Conn
+	PagerdutyClient   *pagerduty.Client
+	CloudEventsClient cloudevents.Client
 }
 
 // NewClient returns a new output.Client for accessing the different API.

outputs/cloudevents.go

@@ -2,10 +2,8 @@ package outputs
 
 import (
 	"context"
-	"log"
-	"strings"
-
 	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"log"
 
 	"github.com/falcosecurity/falcosidekick/types"
 )
@@ -14,30 +12,35 @@ import (
 func (c *Client) CloudEvents(falcopayload types.FalcoPayload) {
 	c.Stats.CloudEvents.Add(Total, 1)
 
-	if c.ce == nil {
+	if c.CloudEventsClient == nil {
 		client, err := cloudevents.NewDefaultClient()
 		if err != nil {
 			go c.CountMetric(Outputs, 1, []string{"output:cloudevents", "status:error"})
 			log.Println("[ERROR] : CloudEvents - NewDefaultClient", err)
 			return
 		}
-		c.ce = client
+		c.CloudEventsClient = client
 	}
 
 	ctx := cloudevents.ContextWithTarget(context.Background(), c.EndpointURL.String())
 
 	event := cloudevents.NewEvent()
 	event.SetTime(falcopayload.Time)
 	event.SetSource("falco.org") // TODO: this should have some info on the falco server that made the event.
-	event.SetType(toEventType(falcopayload.Rule))
+	event.SetType("falco.rule.output.v1")
 	event.SetExtension("priority", falcopayload.Priority)
 	event.SetExtension("rule", falcopayload.Rule)
 
+	// Set Extensions.
+	for k, v := range c.Config.CloudEvents.Extensions {
+		event.SetExtension(k, v)
+	}
+
 	if err := event.SetData(cloudevents.ApplicationJSON, falcopayload); err != nil {
 		log.Println("[ERROR] : CloudEvents, failed to set data", err)
 	}
 
-	if result := c.ce.Send(ctx, event); cloudevents.IsUndelivered(result) {
+	if result := c.CloudEventsClient.Send(ctx, event); cloudevents.IsUndelivered(result) {
 		go c.CountMetric(Outputs, 1, []string{"output:cloudevents", "status:error"})
 		c.Stats.Webhook.Add(Error, 1)
 		c.PromStats.Outputs.With(map[string]string{"destination": "cloudevents", "status": Error}).Inc()
@@ -51,16 +54,3 @@ func (c *Client) CloudEvents(falcopayload types.FalcoPayload) {
 	c.PromStats.Outputs.With(map[string]string{"destination": "cloudevents", "status": OK}).Inc()
 	log.Println("[INFO] : CloudEvents - Send OK")
 }
-
-func toEventType(rule string) string {
-	rule = strings.ToLower(rule)
-
-	// TODO: this could be improved but I do not know what is possible, I am assuming it is a free form entry.
-	for _, old := range []string{".", ":", ","} {
-		rule = strings.ReplaceAll(rule, old, "")
-	}
-
-	rule = strings.ReplaceAll(rule, " ", ".")
-
-	return rule
-}

types/types.go

@@ -201,7 +201,8 @@ type WebhookOutputConfig struct {
 // CloudEventsOutputConfig represents parameters for CloudEvents
 type CloudEventsOutputConfig struct {
 	Address         string
-	MinimumPriority string // likely an anti-pattern.
+	Extensions      map[string]string
+	MinimumPriority string
 }
 
 type statsdOutputConfig struct {