Outputs dirty locking

outputs/client.go

@@ -105,7 +105,8 @@ type Client struct {
 	DogstatsdClient         *statsd.Client
 	GCPTopicClient          *pubsub.Topic
 	GCPCloudFunctionsClient *gcpfunctions.CloudFunctionsClient
-	httpClientLock          sync.Mutex
+	// FIXME: this lock requires a per-output usage lock currently if headers are used -- needs to be refactored
+	httpClientLock sync.Mutex
 
 	GCSStorageClient  *storage.Client
 	KafkaProducer     *kafka.Writer
@@ -139,8 +140,6 @@ func NewClient(outputType string, defaultEndpointURL string, mutualTLSEnabled bo
 
 // Post sends event (payload) to Output.
 func (c *Client) Post(payload interface{}) error {
-	c.httpClientLock.Lock()
-	defer c.httpClientLock.Unlock()
 	// defer + recover to catch panic if output doesn't respond
 	defer func() {
 		if err := recover(); err != nil {
@@ -290,7 +289,5 @@ func (c *Client) BasicAuth(username, password string) {
 
 // AddHeader adds an HTTP Header to the Client.
 func (c *Client) AddHeader(key, value string) {
-	c.httpClientLock.Lock()
-	defer c.httpClientLock.Unlock()
 	c.HeaderList = append(c.HeaderList, Header{Key: key, Value: value})
 }

outputs/cliq.go

@@ -156,6 +156,8 @@ func newCliqPayload(falcopayload types.FalcoPayload, config *types.Configuration
 func (c *Client) CliqPost(falcopayload types.FalcoPayload) {
 	c.Stats.Cliq.Add(Total, 1)
 
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	c.AddHeader(ContentTypeHeaderKey, "application/json")
 	err := c.Post(newCliqPayload(falcopayload, c.Config))
 	if err != nil {

outputs/elasticsearch.go

@@ -34,6 +34,8 @@ func (c *Client) ElasticsearchPost(falcopayload types.FalcoPayload) {
 
 	c.EndpointURL = endpointURL
 	if c.Config.Elasticsearch.Username != "" && c.Config.Elasticsearch.Password != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.BasicAuth(c.Config.Elasticsearch.Username, c.Config.Elasticsearch.Password)
 	}
 

outputs/fission.go

@@ -81,6 +81,8 @@ func (c *Client) FissionCall(falcopayload types.FalcoPayload) {
 		}
 		log.Printf("[INFO]  : %s - Function Response : %v\n", Fission, string(rawbody))
 	} else {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.AddHeader(FissionEventIDKey, uuid.New().String())
 		c.ContentType = FissionContentType
 

outputs/gcpcloudrun.go

@@ -11,6 +11,8 @@ func (c *Client) CloudRunFunctionPost(falcopayload types.FalcoPayload) {
 	c.Stats.GCPCloudRun.Add(Total, 1)
 
 	if c.Config.GCP.CloudRun.JWT != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.AddHeader(AuthorizationHeaderKey, "Bearer "+c.Config.GCP.CloudRun.JWT)
 	}
 

outputs/gotify.go

@@ -92,6 +92,8 @@ func (c *Client) GotifyPost(falcopayload types.FalcoPayload) {
 	c.Stats.Gotify.Add(Total, 1)
 
 	if c.Config.Gotify.Token != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.AddHeader("X-Gotify-Key", c.Config.Gotify.Token)
 	}
 

outputs/grafana.go

@@ -56,7 +56,8 @@ func newGrafanaPayload(falcopayload types.FalcoPayload, config *types.Configurat
 func (c *Client) GrafanaPost(falcopayload types.FalcoPayload) {
 	c.Stats.Grafana.Add(Total, 1)
 	c.ContentType = GrafanaContentType
-
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	c.AddHeader("Authorization", "Bearer "+c.Config.Grafana.APIKey)
 
 	err := c.Post(newGrafanaPayload(falcopayload, c.Config))

outputs/influxdb.go

@@ -34,6 +34,8 @@ func newInfluxdbPayload(falcopayload types.FalcoPayload, config *types.Configura
 func (c *Client) InfluxdbPost(falcopayload types.FalcoPayload) {
 	c.Stats.Influxdb.Add(Total, 1)
 
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	c.AddHeader("Accept", "application/json")
 
 	if c.Config.Influxdb.Token != "" {

outputs/kubeless.go

@@ -80,6 +80,8 @@ func (c *Client) KubelessCall(falcopayload types.FalcoPayload) {
 		}
 		log.Printf("[INFO]  : Kubeless - Function Response : %v\n", string(rawbody))
 	} else {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.AddHeader(KubelessEventIDKey, uuid.New().String())
 		c.AddHeader(KubelessEventTypeKey, KubelessEventTypeValue)
 		c.AddHeader(KubelessEventNamespaceKey, c.Config.Kubeless.Namespace)

outputs/loki.go

@@ -63,10 +63,14 @@ func (c *Client) LokiPost(falcopayload types.FalcoPayload) {
 	c.Stats.Loki.Add(Total, 1)
 	c.ContentType = LokiContentType
 	if c.Config.Loki.Tenant != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.AddHeader("X-Scope-OrgID", c.Config.Loki.Tenant)
 	}
 
 	if c.Config.Loki.User != "" && c.Config.Loki.APIKey != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.BasicAuth(c.Config.Loki.User, c.Config.Loki.APIKey)
 	}
 

outputs/nodered.go

@@ -11,6 +11,8 @@ import (
 func (c *Client) NodeRedPost(falcopayload types.FalcoPayload) {
 	c.Stats.NodeRed.Add(Total, 1)
 
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	if c.Config.NodeRed.User != "" && c.Config.NodeRed.Password != "" {
 		c.AddHeader("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(c.Config.NodeRed.User+":"+c.Config.NodeRed.Password)))
 	}

outputs/opsgenie.go

@@ -59,6 +59,8 @@ func newOpsgeniePayload(falcopayload types.FalcoPayload, config *types.Configura
 // OpsgeniePost posts event to OpsGenie
 func (c *Client) OpsgeniePost(falcopayload types.FalcoPayload) {
 	c.Stats.Opsgenie.Add(Total, 1)
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	c.AddHeader(AuthorizationHeaderKey, "GenieKey "+c.Config.Opsgenie.APIKey)
 
 	err := c.Post(newOpsgeniePayload(falcopayload, c.Config))

outputs/spyderbat.go

@@ -199,6 +199,8 @@ func NewSpyderbatClient(config *types.Configuration, stats *types.Statistics, pr
 func (c *Client) SpyderbatPost(falcopayload types.FalcoPayload) {
 	c.Stats.Spyderbat.Add(Total, 1)
 
+	c.httpClientLock.Lock()
+	defer c.httpClientLock.Unlock()
 	c.AddHeader("Authorization", "Bearer "+c.Config.Spyderbat.APIKey)
 	c.AddHeader("Content-Encoding", "gzip")
 

outputs/webhook.go

@@ -11,6 +11,8 @@ func (c *Client) WebhookPost(falcopayload types.FalcoPayload) {
 	c.Stats.Webhook.Add(Total, 1)
 
 	if len(c.Config.Webhook.CustomHeaders) != 0 {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		for i, j := range c.Config.Webhook.CustomHeaders {
 			c.AddHeader(i, j)
 		}

outputs/zincsearch.go

@@ -12,6 +12,8 @@ func (c *Client) ZincsearchPost(falcopayload types.FalcoPayload) {
 	c.Stats.Zincsearch.Add(Total, 1)
 
 	if c.Config.Zincsearch.Username != "" && c.Config.Zincsearch.Password != "" {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
 		c.BasicAuth(c.Config.Zincsearch.Username, c.Config.Zincsearch.Password)
 	}