Loki integration #2

.github/workflows/go-tests.yml

@@ -108,6 +108,18 @@ jobs:
           --health-timeout 10s
           --health-retries 5
 
+      loki:
+        image: grafana/loki:2.8.0
+        ports:
+          - "3100:3100"
+        options: >-
+          --name=loki1
+          --health-cmd "wget -q -O -  http://localhost:3100/ready | grep 'ready'"
+          --health-interval 30s
+          --health-timeout 10s
+          --health-retries 5
+          --health-start-period 30s
+
     steps:
 
     - name: Check out CrowdSec repository

go.mod

@@ -49,6 +49,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/google/winops v0.0.0-20230712152054-af9b550d0601
 	github.com/goombaio/namegenerator v0.0.0-20181006234301-989e774b106e
+	github.com/gorilla/websocket v1.5.0
 	github.com/hashicorp/go-hclog v1.5.0
 	github.com/hashicorp/go-plugin v1.4.10
 	github.com/hashicorp/go-version v1.2.1
@@ -87,6 +88,7 @@ require (
 	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
+	gotest.tools/v3 v3.5.0
 	k8s.io/apiserver v0.27.3
 )
 
@@ -126,7 +128,6 @@ require (
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/hashicorp/hcl/v2 v2.13.0 // indirect
 	github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb // indirect
 	github.com/huandu/xstrings v1.3.2 // indirect
@@ -201,7 +202,6 @@ require (
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
-	gotest.tools/v3 v3.5.0 // indirect
 	k8s.io/api v0.27.3 // indirect
 	k8s.io/apimachinery v0.27.3 // indirect
 	k8s.io/klog/v2 v2.90.1 // indirect

pkg/acquisition/acquisition.go

@@ -25,6 +25,7 @@
 	kafkaacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/kafka"
 	kinesisacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/kinesis"
 	k8sauditacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/kubernetesaudit"
+	lokiacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/loki"
 	s3acquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/s3"
 	syslogacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/syslog"
 	wineventlogacquisition "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/wineventlog"
@@ -36,7 +37,7 @@
 
 type DataSourceUnavailableError struct {
 	Name string
-	Err error
+	Err  error
 }
 
 func (e *DataSourceUnavailableError) Error() string {
@@ -47,7 +48,6 @@
 	return e.Err
 }
 
-
 // The interface each datasource must implement
 type DataSource interface {
 	GetMetrics() []prometheus.Collector                                 // Returns pointers to metrics that are managed by the module
@@ -74,6 +74,7 @@
 	"wineventlog": func() DataSource { return &wineventlogacquisition.WinEventLogSource{} },
 	"kafka":       func() DataSource { return &kafkaacquisition.KafkaSource{} },
 	"k8s-audit":   func() DataSource { return &k8sauditacquisition.KubernetesAuditSource{} },
+	"loki":        func() DataSource { return &lokiacquisition.LokiSource{} },
 	"s3":          func() DataSource { return &s3acquisition.S3Source{} },
 }
 
@@ -322,7 +323,7 @@
 
 	for i := 0; i < len(sources); i++ {
 		subsrc := sources[i] //ensure its a copy
 		log.Debugf("starting one source %d/%d ->> %T", i, len(sources), subsrc)
 
 		AcquisTomb.Go(func() error {
 			defer trace.CatchPanic("crowdsec/acquis")

pkg/acquisition/modules/loki/entry.go

@@ -0,0 +1,60 @@
+package loki
+
+import (
+	"encoding/json"
+	"strconv"
+	"time"
+)
+
+type Entry struct {
+	Timestamp time.Time
+	Line      string
+}
+
+func (e *Entry) UnmarshalJSON(b []byte) error {
+	var values []string
+	err := json.Unmarshal(b, &values)
+	if err != nil {
+		return err
+	}
+	t, err := strconv.Atoi(values[0])
+	if err != nil {
+		return err
+	}
+	e.Timestamp = time.Unix(int64(t), 0)
+	e.Line = values[1]
+	return nil
+}
+
+type Stream struct {
+	Stream  map[string]string `json:"stream"`
+	Entries []Entry           `json:"values"`
+}
+
+type DroppedEntry struct {
+	Labels    map[string]string `json:"labels"`
+	Timestamp time.Time         `json:"timestamp"`
+}
+
+type Tail struct {
+	Streams        []Stream       `json:"streams"`
+	DroppedEntries []DroppedEntry `json:"dropped_entries"`
+}
+
+// LokiQuery GET response.
+// See https://grafana.com/docs/loki/latest/api/#get-lokiapiv1query
+type LokiQuery struct {
+	Status string `json:"status"`
+	Data   Data   `json:"data"`
+}
+
+type Data struct {
+	ResultType string         `json:"resultType"`
+	Result     []StreamResult `json:"result"` // Warning, just stream value is handled
+	Stats      interface{}    `json:"stats"`  // Stats is boring, just ignore it
+}
+
+type StreamResult struct {
+	Stream map[string]string `json:"stream"`
+	Values []Entry           `json:"values"`
+}