Skip to content

Commit

Permalink
feat(quickwit_output): implement the quickwit output
Browse files Browse the repository at this point in the history
Signed-off-by: Idriss Neumann <idriss.neumann@comwork.io>
  • Loading branch information
idrissneumann committed Jan 24, 2024
1 parent d2194e1 commit d85dc1e
Show file tree
Hide file tree
Showing 11 changed files with 335 additions and 0 deletions.
1 change: 1 addition & 0 deletions README.md
Expand Up @@ -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

Expand Down
11 changes: 11 additions & 0 deletions config.go
Expand Up @@ -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)},
Expand Down Expand Up @@ -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.IndexVersion", "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", "")
Expand Down Expand Up @@ -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)
Expand Down
12 changes: 12 additions & 0 deletions config_example.yaml
Expand Up @@ -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)
Expand Down
Binary file added docs/outputs/images/grafana_quickwit.png
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
50 changes: 50 additions & 0 deletions 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)
4 changes: 4 additions & 0 deletions handlers.go
Expand Up @@ -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)
}
Expand Down
25 changes: 25 additions & 0 deletions main.go
Expand Up @@ -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
Expand Down Expand Up @@ -95,6 +96,7 @@ var (
config *types.Configuration
stats *types.Statistics
promStats *types.PromStatistics
initClientArgs *types.InitClientArgs

regPromLabels *regexp.Regexp
)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
5 changes: 5 additions & 0 deletions outputs/client.go
Expand Up @@ -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?)://.*`)
Expand Down
202 changes: 202 additions & 0 deletions 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.IndexVersion,
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", "output_fields"},
},
}

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()
}
1 change: 1 addition & 0 deletions stats.go
Expand Up @@ -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"),
Expand Down

0 comments on commit d85dc1e

Please sign in to comment.