feat(quickwit_output): implement the quickwit output

README.md

@@ -111,6 +111,7 @@ Follow the links to get the configuration of each output.
 - [**Zincsearch**](https://github.com/falcosecurity/falcosidekick/blob/master/docs/outputs//zincsearch.md)
 - [**OpenObserve**](https://github.com/falcosecurity/falcosidekick/blob/master/docs/outputs/openobserve.md)
 - [**SumoLogic**](https://github.com/falcosecurity/falcosidekick/blob/master/docs/outputs/sumologic.md)
+- [**Quickwit**](https://github.com/falcosecurity/falcosidekick/blob/master/docs/outputs/quickwit.md)
 
 ### Object Storage
 

config.go

@@ -44,6 +44,7 @@ func getConfig() *types.Configuration {
 		Grafana:         types.GrafanaOutputConfig{CustomHeaders: make(map[string]string)},
 		Loki:            types.LokiOutputConfig{CustomHeaders: make(map[string]string)},
 		Elasticsearch:   types.ElasticsearchOutputConfig{CustomHeaders: make(map[string]string)},
+		Quickwit:        types.QuickwitOutputConfig{CustomHeaders: make(map[string]string)},
 		OpenObserve:     types.OpenObserveConfig{CustomHeaders: make(map[string]string)},
 		Webhook:         types.WebhookOutputConfig{CustomHeaders: make(map[string]string)},
 		Alertmanager:    types.AlertmanagerOutputConfig{ExtraLabels: make(map[string]string), ExtraAnnotations: make(map[string]string), CustomSeverityMap: make(map[types.PriorityType]string)},
@@ -148,6 +149,15 @@ func getConfig() *types.Configuration {
 	v.SetDefault("Elasticsearch.Username", "")
 	v.SetDefault("Elasticsearch.Password", "")
 
+	v.SetDefault("Quickwit.HostPort", "")
+	v.SetDefault("Quickwit.Index", "falco")
+	v.SetDefault("Quickwit.ApiEndpoint", "api/v1")
+	v.SetDefault("Quickwit.Version", "0.7")
+	v.SetDefault("Quickwit.AutoCreateIndex", false)
+	v.SetDefault("Quickwit.MinimumPriority", "")
+	v.SetDefault("Quickwit.MutualTls", false)
+	v.SetDefault("Quickwit.CheckCert", true)
+
 	v.SetDefault("Influxdb.HostPort", "")
 	v.SetDefault("Influxdb.Database", "falco")
 	v.SetDefault("Influxdb.Organization", "")
@@ -714,6 +724,7 @@ func getConfig() *types.Configuration {
 	c.Alertmanager.MinimumPriority = checkPriority(c.Alertmanager.MinimumPriority)
 	c.Alertmanager.DropEventDefaultPriority = checkPriority(c.Alertmanager.DropEventDefaultPriority)
 	c.Elasticsearch.MinimumPriority = checkPriority(c.Elasticsearch.MinimumPriority)
+	c.Quickwit.MinimumPriority = checkPriority(c.Quickwit.MinimumPriority)
 	c.Influxdb.MinimumPriority = checkPriority(c.Influxdb.MinimumPriority)
 	c.Loki.MinimumPriority = checkPriority(c.Loki.MinimumPriority)
 	c.SumoLogic.MinimumPriority = checkPriority(c.SumoLogic.MinimumPriority)

config_example.yaml

@@ -95,6 +95,18 @@ elasticsearch:
   # customHeaders: # Custom headers to add in POST, useful for Authentication
   #   key: value
 
+quickwit:
+  # hostport: "" # http(s)://{domain or ip}:{port}, if not empty, Quickwit output is enabled
+  # apiendpoint: "/api/v1"
+  # index: "falco" # index (default: falco)
+  # version: "0.7"
+  # autocreateindex: false # create the index mapping if true and if the index doesn't already exists
+  # customHeaders: # Custom headers to add in POST, useful for Authentication
+  #   key: value
+  # mutualtls: false # if true, checkcert flag will be ignored (server cert will always be checked)
+  # checkcert: true # check if ssl certificate of the output is valid (default: true)
+  # minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|error|warning|notice|informational|debug or "" (default)
+
 influxdb:
   # hostport: "" # http://{domain or ip}:{port}, if not empty, Influxdb output is enabled
   # database: "falco" # Influxdb database (api v1 only) (default: falco)

docs/outputs/quickwit.md

@@ -0,0 +1,50 @@
+# Quickwit
+
+- **Category**: Logs
+- **Website**: https://quickwit.io/
+
+## Table of content
+
+- [Quickwit](#quickwit)
+  - [Table of content](#table-of-content)
+  - [Configuration](#configuration)
+  - [Example of config.yaml](#example-of-configyaml)
+  - [Screenshots](#screenshots)
+
+## Configuration
+
+| Setting                         | Env var                         | Default value    | Description                                                                                                                            |
+| ------------------------------- | ------------------------------- | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------- |
+| `quickwit.hosport`              | `QUICKWIT_HOSTPORT`             |                  | http://{domain or ip}:{port}, if not empty, Quickwit output is **enabled**                                                             |
+| `quickwit.apiendpoint`          | `QUICKWIT_APIENDPOINT`          | `api/v1`        | API endpoint (containing the API version, overideable in case of quickwit behind a reverse proxy with URL rewriting)                   |
+| `quickwit.index`                | `QUICKWIT_INDEX`                | `falco`          | Index                                                                                                                                  |
+| `quickwit.version`              | `QUICKWIT_VERSION`              | `0.7`            | Version of quickwit                                                                                                                    |
+| `quickwit.autocreateindex`      | `QUICKWIT_AUTOCREATEINDEX`      | `false`          | Autocreate a `falco` index mapping if it doesn't exists                                                                                |
+| `quickwit.customheaders`        | `QUICKWIT_CUSTOMHEADERS`        |                  | Custom headers to add in POST, useful for Authentication                                                                               |
+| `quickwit.mutualtls`            | `QUICKWIT_MUTUALTLS`            | `false`          | Authenticate to the output with TLS, if true, checkcert flag will be ignored (server cert will always be checked)                      |
+| `quickwit.checkcert`            | `QUICKWIT_CHECKCERT`            | `true`           | Check if ssl certificate of the output is valid                                                                                        |
+| `quickwit.minimumpriority`      | `QUICKWIT_MINIMUMPRIORITY`      | `""` (= `debug`) | Minimum priority of event for using this output, order is `emergency,alert,critical,error,warning,notice,informational,debug or ""` |
+
+> **Note**
+The Env var values override the settings from yaml file.
+
+## Example of config.yaml
+
+```yaml
+quickwit:
+  # hostport: ""
+  # apiendpoint: "/api/v1"
+  # index: "falco"
+  # version: "0.7"
+  # autocreateindex: false
+  # customHeaders:
+  #   key: value
+  # mutualtls: false
+  # checkcert: true
+  # minimumpriority: ""
+```
+
+## Screenshots
+
+With Grafana:
+![Grafana example](images/grafana_quickwit.png)

handlers.go

@@ -235,6 +235,10 @@ func forwardEvent(falcopayload types.FalcoPayload) {
 		go elasticsearchClient.ElasticsearchPost(falcopayload)
 	}
 
+	if config.Quickwit.HostPort != "" && (falcopayload.Priority >= types.Priority(config.Quickwit.MinimumPriority) || falcopayload.Rule == testRule) {
+		go quickwitClient.QuickwitPost(falcopayload)
+	}
+
 	if config.Influxdb.HostPort != "" && (falcopayload.Priority >= types.Priority(config.Influxdb.MinimumPriority) || falcopayload.Rule == testRule) {
 		go influxdbClient.InfluxdbPost(falcopayload)
 	}

main.go

@@ -50,6 +50,7 @@ var (
 	discordClient       *outputs.Client
 	alertmanagerClient  *outputs.Client
 	elasticsearchClient *outputs.Client
+	quickwitClient      *outputs.Client
 	influxdbClient      *outputs.Client
 	lokiClient          *outputs.Client
 	sumologicClient     *outputs.Client
@@ -95,6 +96,7 @@ var (
 	config                        *types.Configuration
 	stats                         *types.Statistics
 	promStats                     *types.PromStatistics
+	initClientArgs                *types.InitClientArgs
 
 	regPromLabels *regexp.Regexp
 )
@@ -123,6 +125,13 @@ func init() {
 		DogstatsdClient: dogstatsdClient,
 	}
 
+	initClientArgs = &types.InitClientArgs{
+		Config:          config,
+		Stats:           stats,
+		DogstatsdClient: dogstatsdClient,
+		PromStats:       promStats,
+	}
+
 	if config.Statsd.Forwarder != "" {
 		var err error
 		statsdClient, err = outputs.NewStatsdClient("StatsD", config, stats)
@@ -235,6 +244,22 @@ func init() {
 		}
 	}
 
+	if config.Quickwit.HostPort != "" {
+		var err error
+
+		endpointUrl := fmt.Sprintf("%s/%s/%s/ingest", config.Quickwit.HostPort, config.Quickwit.ApiEndpoint, config.Quickwit.Index)
+		quickwitClient, err = outputs.InitClient("Quickwit", endpointUrl, config.Quickwit.MutualTLS, config.Quickwit.CheckCert, *initClientArgs)
+		if err == nil && config.Quickwit.AutoCreateIndex {
+			err = quickwitClient.AutoCreateQuickwitIndex(*initClientArgs)
+		}
+
+		if err != nil {
+			config.Quickwit.HostPort = ""
+		} else {
+			outputs.EnabledOutputs = append(outputs.EnabledOutputs, "Quickwit")
+		}
+	}
+
 	if config.Loki.HostPort != "" {
 		var err error
 		lokiClient, err = outputs.NewClient("Loki", config.Loki.HostPort+config.Loki.Endpoint, config.Loki.MutualTLS, config.Loki.CheckCert, config, stats, promStats, statsdClient, dogstatsdClient)

outputs/client.go

@@ -143,6 +143,11 @@ type Client struct {
 	RedisClient       *redis.Client
 }
 
+// InitClient returns a new output.Client for accessing the different API.
+func InitClient(outputType string, defaultEndpointURL string, mutualTLSEnabled bool, checkCert bool, params types.InitClientArgs) (*Client, error) {
+	return NewClient(outputType, defaultEndpointURL, mutualTLSEnabled, checkCert, params.Config, params.Stats, params.PromStats, params.StatsdClient, params.DogstatsdClient)
+}
+
 // NewClient returns a new output.Client for accessing the different API.
 func NewClient(outputType string, defaultEndpointURL string, mutualTLSEnabled bool, checkCert bool, config *types.Configuration, stats *types.Statistics, promStats *types.PromStatistics, statsdClient, dogstatsdClient *statsd.Client) (*Client, error) {
 	reg := regexp.MustCompile(`(http|nats)(s?)://.*`)

outputs/quickwit.go

@@ -0,0 +1,202 @@
+// SPDX-License-Identifier: Apache-2.0
+/*
+Copyright (C) 2023 The Falco Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package outputs
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/falcosecurity/falcosidekick/types"
+)
+
+type QuickwitDynamicMapping struct {
+	Description string `json:"description"`
+	Fast        bool   `json:"fast"`
+	ExpendDots  bool   `json:"expand_dots"`
+	Indexed     bool   `json:"indexed"`
+	Record      string `json:"record"`
+	Stored      bool   `json:"stored"`
+	Tokenizer   string `json:"tokenizer"`
+}
+
+type QuickwitFieldMapping struct {
+	Name string `json:"name"`
+	Type string `json:"type"`
+	Fast bool   `json:"fast"`
+}
+
+type QuickwitSearchSettings struct {
+	DefaultSearchFields []string `json:"default_search_fields"`
+}
+
+type QuickwitDocMapping struct {
+	DynamicMapping QuickwitDynamicMapping `json:"dynamic_mapping"`
+	FieldMappings  []QuickwitFieldMapping `json:"field_mappings"`
+	Mode           string                 `json:"mode"`
+	StoreSource    bool                   `json:"store_source"`
+	TimestampField string                 `json:"timestamp_field"`
+}
+
+type QuickwitMappingPayload struct {
+	Id             string                 `json:"index_id"`
+	Version        string                 `json:"version"`
+	SearchSettings QuickwitSearchSettings `json:"search_settings"`
+	DocMapping     QuickwitDocMapping     `json:"doc_mapping"`
+}
+
+func (c *Client) checkQuickwitIndexAlreadyExists(args types.InitClientArgs) bool {
+	config := args.Config.Quickwit
+
+	endpointUrl := fmt.Sprintf("%s/%s/indexes/%s/describe", config.HostPort, config.ApiEndpoint, config.Index)
+	quickwitCheckClient, err := InitClient("QuickwitCheckAlreadyExists", endpointUrl, config.MutualTLS, config.CheckCert, args)
+	if err != nil {
+		return false
+	}
+
+	if nil != quickwitCheckClient.sendRequest("GET", "") {
+		return false
+	}
+
+	return true
+}
+
+func (c *Client) AutoCreateQuickwitIndex(args types.InitClientArgs) error {
+	config := args.Config.Quickwit
+
+	if c.checkQuickwitIndexAlreadyExists(args) {
+		return nil
+	}
+
+	endpointUrl := fmt.Sprintf("%s/%s/indexes", config.HostPort, config.ApiEndpoint)
+	quickwitInitClient, err := InitClient("QuickwitInit", endpointUrl, config.MutualTLS, config.CheckCert, args)
+	if err != nil {
+		return err
+	}
+
+	mapping := &QuickwitMappingPayload{
+		Id:      config.Index,
+		Version: config.Version,
+		DocMapping: QuickwitDocMapping{
+			Mode:           "dynamic",
+			StoreSource:    true,
+			TimestampField: "time",
+			DynamicMapping: QuickwitDynamicMapping{
+				Description: "Falco",
+				Fast:        true,
+				ExpendDots:  true,
+				Indexed:     true,
+				Stored:      true,
+				Record:      "basic",
+				Tokenizer:   "raw",
+			},
+			FieldMappings: []QuickwitFieldMapping{
+				{
+					Name: "time",
+					Type: "datetime",
+					Fast: true,
+				},
+				{
+					Name: "uuid",
+					Type: "text",
+					Fast: true,
+				},
+				{
+					Name: "hostname",
+					Type: "text",
+					Fast: true,
+				},
+				{
+					Name: "priority",
+					Type: "text",
+					Fast: true,
+				},
+				{
+					Name: "source",
+					Type: "text",
+					Fast: true,
+				},
+				{
+					Name: "output",
+					Type: "text",
+				},
+				{
+					Name: "rule",
+					Type: "text",
+					Fast: true,
+				},
+				{
+					Name: "tags",
+					Type: "array<text>",
+					Fast: true,
+				},
+				{
+					Name: "output_fields",
+					Type: "json",
+					Fast: true,
+				},
+			},
+		},
+		SearchSettings: QuickwitSearchSettings{
+			DefaultSearchFields: []string{"rule", "source", "output", "priority", "hostname", "tags"},
+		},
+	}
+
+	if args.Config.Debug {
+		log.Printf("[DEBUG] : Quickwit - mapping: %#v\n", mapping)
+	}
+
+	err = quickwitInitClient.Post(mapping)
+
+	// This error means it's an http 400 (meaning the index already exists, so no need to throw an error)
+	if err != nil && err.Error() == "header missing" {
+		return nil
+	}
+
+	return err
+}
+
+func (c *Client) QuickwitPost(falcopayload types.FalcoPayload) {
+	c.Stats.Quickwit.Add(Total, 1)
+
+	if len(c.Config.Quickwit.CustomHeaders) != 0 {
+		c.httpClientLock.Lock()
+		defer c.httpClientLock.Unlock()
+		for i, j := range c.Config.Quickwit.CustomHeaders {
+			c.AddHeader(i, j)
+		}
+	}
+
+	if c.Config.Debug {
+		log.Printf("[DEBUG] : Quickwit - ingesting payload: %v\n", falcopayload)
+	}
+
+	err := c.Post(falcopayload)
+
+	if err != nil {
+		go c.CountMetric(Outputs, 1, []string{"output:quickwit", "status:error"})
+		c.Stats.Quickwit.Add(Error, 1)
+		c.PromStats.Outputs.With(map[string]string{"destination": "quickwit", "status": Error}).Inc()
+		log.Printf("[ERROR] : Quickwit - %v\n", err.Error())
+		return
+	}
+
+	// Setting the success status
+	go c.CountMetric(Outputs, 1, []string{"output:quickwit", "status:ok"})
+	c.Stats.Quickwit.Add(OK, 1)
+	c.PromStats.Outputs.With(map[string]string{"destination": "quickwit", "status": OK}).Inc()
+}

stats.go

@@ -48,6 +48,7 @@ func getInitStats() *types.Statistics {
 		Discord:           getOutputNewMap("discord"),
 		Alertmanager:      getOutputNewMap("alertmanager"),
 		Elasticsearch:     getOutputNewMap("elasticsearch"),
+		Quickwit:          getOutputNewMap("quickwit"),
 		Loki:              getOutputNewMap("loki"),
 		SumoLogic:         getOutputNewMap("sumologic"),
 		Nats:              getOutputNewMap("nats"),