Lapi detailed metrics

cmd/crowdsec/crowdsec.go

@@ -140,6 +140,19 @@
 	})
 	outputWg.Wait()
 
+	mp := NewMetricsProvider(
+		apiClient,
+		lpMetricsDefaultInterval,
+		log.WithField("service", "lpmetrics"),
+		[]string{},
+		datasources,
+		hub,
+	)
+
+	lpMetricsTomb.Go(func() error {
+		return mp.Run(context.Background(), &lpMetricsTomb)
+	})
+
 	if cConfig.Prometheus != nil && cConfig.Prometheus.Enabled {
 		aggregated := false
 		if cConfig.Prometheus.Level == configuration.CFG_METRICS_AGGREGATE {

cmd/crowdsec/lpmetrics.go

@@ -0,0 +1,182 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"gopkg.in/tomb.v2"
+
+	"github.com/crowdsecurity/go-cs-lib/ptr"
+	"github.com/crowdsecurity/go-cs-lib/trace"
+	"github.com/crowdsecurity/go-cs-lib/version"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition"
+	"github.com/crowdsecurity/crowdsec/pkg/apiclient"
+	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
+	"github.com/crowdsecurity/crowdsec/pkg/fflag"
+	"github.com/crowdsecurity/crowdsec/pkg/models"
+)
+
+const lpMetricsDefaultInterval = 30 * time.Minute
+
+// MetricsProvider collects metrics from the LP and sends them to the LAPI
+type MetricsProvider struct {
+	apic     *apiclient.ApiClient
+	interval time.Duration
+	static   staticMetrics
+	logger   *logrus.Entry
+}
+
+type staticMetrics struct {
+	osName         string
+	osVersion      string
+	startupTS      int64
+	featureFlags   []string
+	consoleOptions []string
+	datasourceMap  map[string]int64
+	hubState       models.HubItems
+}
+
+func getHubState(hub *cwhub.Hub) models.HubItems {
+	ret := models.HubItems{}
+
+	for _, itemType := range cwhub.ItemTypes {
+		ret[itemType] = []models.HubItem{}
+		items, _ := hub.GetInstalledItemsByType(itemType)
+		cwhub.SortItemSlice(items)
+
+		for _, item := range items {
+			status := "official"
+			if item.State.IsLocal() {
+				status = "custom"
+			}
+			if item.State.Tainted {
+				status = "tainted"
+			}
+			ret[itemType] = append(ret[itemType], models.HubItem{
+				Name:    item.Name,
+				Status:  status,
+				Version: item.Version,
+			})
+		}
+	}
+
+	return ret
+}
+
+// newStaticMetrics is called when the process starts, or reloads the configuration
+func newStaticMetrics(consoleOptions []string, datasources []acquisition.DataSource, hub *cwhub.Hub) staticMetrics {
+	datasourceMap := map[string]int64{}
+
+	for _, ds := range datasources {
+		datasourceMap[ds.GetName()] += 1
+	}
+
+	osName, osVersion := version.DetectOS()
+
+	return staticMetrics{
+		osName:         osName,
+		osVersion:      osVersion,
+		startupTS:      time.Now().UTC().Unix(),
+		featureFlags:   fflag.Crowdsec.GetEnabledFeatures(),
+		consoleOptions: consoleOptions,
+		datasourceMap:  datasourceMap,
+		hubState:       getHubState(hub),
+	}
+}
+
+func NewMetricsProvider(apic *apiclient.ApiClient, interval time.Duration, logger *logrus.Entry,
+	consoleOptions []string, datasources []acquisition.DataSource, hub *cwhub.Hub) *MetricsProvider {
+	return &MetricsProvider{
+		apic:     apic,
+		interval: interval,
+		logger:   logger,
+		static:   newStaticMetrics(consoleOptions, datasources, hub),
+	}
+}
+
+func (m *MetricsProvider) metricsPayload() *models.AllMetrics {
+	os := &models.OSversion{
+		Name:    ptr.Of(m.static.osName),
+		Version: ptr.Of(m.static.osVersion),
+	}
+
+	base := models.BaseMetrics{
+		UtcStartupTimestamp: ptr.Of(m.static.startupTS),
+		Os:                  os,
+		Version:             ptr.Of(version.String()),
+		FeatureFlags:        m.static.featureFlags,
+		Metrics:             make([]*models.DetailedMetrics, 0),
+	}
+
+	met := &models.LogProcessorsMetrics{
+		BaseMetrics: base,
+		Datasources: m.static.datasourceMap,
+		HubItems:    m.static.hubState,
+	}
+
+	met.Metrics = append(met.Metrics, &models.DetailedMetrics{
+		Meta: &models.MetricsMeta{
+			UtcNowTimestamp:   ptr.Of(time.Now().Unix()),
+			WindowSizeSeconds: ptr.Of(int64(m.interval.Seconds())),
+		},
+		Items: make([]*models.MetricsDetailItem, 0),
+	})
+
+	return &models.AllMetrics{
+		LogProcessors: []*models.LogProcessorsMetrics{met},
+	}
+}
+
+func (m *MetricsProvider) Run(ctx context.Context, myTomb *tomb.Tomb) error {
+	defer trace.CatchPanic("crowdsec/MetricsProvider.Run")
+
+	if m.interval == time.Duration(0) {
+		return nil
+	}
+
+	met := m.metricsPayload()
+
+	ticker := time.NewTicker(1) // Send on start
+
+	for {
+		select {
+		case <-ticker.C:
+			ctxTime, cancel := context.WithTimeout(ctx, 10*time.Second)
+			defer cancel()
+
+			_, resp, err := m.apic.UsageMetrics.Add(ctxTime, met)
+			switch {
+			case errors.Is(err, context.DeadlineExceeded):
+				m.logger.Warnf("timeout sending lp metrics")
+				ticker.Reset(m.interval)
+				continue
+			case err != nil && resp != nil && resp.Response.StatusCode == http.StatusNotFound:
+				m.logger.Warnf("metrics endpoint not found, older LAPI?")
+				ticker.Reset(m.interval)
+				continue
+			case err != nil:
+				m.logger.Warnf("failed to send lp metrics: %s", err)
+				ticker.Reset(m.interval)
+				continue
+			}
+
+			if resp.Response.StatusCode != http.StatusCreated {
+				m.logger.Warnf("failed to send lp metrics: %s", resp.Response.Status)
+				ticker.Reset(m.interval)
+				continue
+			}
+
+			ticker.Reset(m.interval)
+
+			m.logger.Tracef("lp usage metrics sent")
+		case <-myTomb.Dying():
+			ticker.Stop()
+			return nil
+		}
+	}
+}

cmd/crowdsec/main.go

@@ -29,28 +29,29 @@ import (
 )
 
 var (
-	/*tombs for the parser, buckets and outputs.*/
-	acquisTomb   tomb.Tomb
-	parsersTomb  tomb.Tomb
-	bucketsTomb  tomb.Tomb
-	outputsTomb  tomb.Tomb
-	apiTomb      tomb.Tomb
-	crowdsecTomb tomb.Tomb
-	pluginTomb   tomb.Tomb
+	// tombs for the parser, buckets and outputs.
+	acquisTomb    tomb.Tomb
+	parsersTomb   tomb.Tomb
+	bucketsTomb   tomb.Tomb
+	outputsTomb   tomb.Tomb
+	apiTomb       tomb.Tomb
+	crowdsecTomb  tomb.Tomb
+	pluginTomb    tomb.Tomb
+	lpMetricsTomb tomb.Tomb
 
 	flags *Flags
 
-	/*the state of acquisition*/
+	// the state of acquisition
 	dataSources []acquisition.DataSource
-	/*the state of the buckets*/
+	// the state of the buckets
 	holders []leakybucket.BucketFactory
 	buckets *leakybucket.Buckets
 
 	inputLineChan   chan types.Event
 	inputEventChan  chan types.Event
 	outputEventChan chan types.Event // the buckets init returns its own chan that is used for multiplexing
-	/*settings*/
-	lastProcessedItem time.Time /*keep track of last item timestamp in time-machine. it is used to GC buckets when we dump them.*/
+	// settings
+	lastProcessedItem time.Time // keep track of last item timestamp in time-machine. it is used to GC buckets when we dump them.
 	pluginBroker      csplugin.PluginBroker
 )
 
@@ -307,7 +308,7 @@ func LoadConfig(configFile string, disableAgent bool, disableAPI bool, quiet boo
 		if cConfig.API != nil && cConfig.API.Server != nil {
 			cConfig.API.Server.OnlineClient = nil
 		}
-		/*if the api is disabled as well, just read file and exit, don't daemonize*/
+		// if the api is disabled as well, just read file and exit, don't daemonize
 		if cConfig.DisableAPI {
 			cConfig.Common.Daemonize = false
 		}

cmd/crowdsec/serve.go

@@ -60,6 +60,7 @@
 	apiTomb = tomb.Tomb{}
 	crowdsecTomb = tomb.Tomb{}
 	pluginTomb = tomb.Tomb{}
+	lpMetricsTomb = tomb.Tomb{}
 
 	cConfig, err := LoadConfig(flags.ConfigFile, flags.DisableAgent, flags.DisableAPI, false)
 	if err != nil {
@@ -179,6 +180,15 @@
 		log.Warningf("Outputs didn't finish in time, some events may have not been flushed")
 	}
 
+	lpMetricsTomb.Kill(nil)
+
+	if err := lpMetricsTomb.Wait(); err != nil {
+		log.Warningf("Metrics returned error : %s", err)
+		reterr = err
+	}
+
+	log.Debugf("metrics are done")
+
 	// He's dead, Jim.
 	crowdsecTomb.Kill(nil)
 
@@ -322,6 +332,7 @@
 	apiTomb = tomb.Tomb{}
 	crowdsecTomb = tomb.Tomb{}
 	pluginTomb = tomb.Tomb{}
+	lpMetricsTomb = tomb.Tomb{}
 
 	ctx := context.TODO()
 

go.mod

@@ -13,7 +13,6 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/agext/levenshtein v1.2.3
 	github.com/alexliesenfeld/health v0.8.0
-	github.com/antonmedv/expr v1.15.3
 	github.com/appleboy/gin-jwt/v2 v2.9.2
 	github.com/aws/aws-lambda-go v1.47.0
 	github.com/aws/aws-sdk-go v1.52.0
@@ -33,6 +32,7 @@ require (
 	github.com/dghubble/sling v1.4.2
 	github.com/docker/docker v24.0.9+incompatible
 	github.com/docker/go-connections v0.4.0
+	github.com/expr-lang/expr v1.16.9
 	github.com/fatih/color v1.16.0
 	github.com/fsnotify/fsnotify v1.7.0
 	github.com/gin-gonic/gin v1.9.1
@@ -111,7 +111,6 @@ require (
 	github.com/creack/pty v1.1.18 // indirect
 	github.com/docker/distribution v2.8.2+incompatible // indirect
 	github.com/docker/go-units v0.5.0 // indirect
-	github.com/expr-lang/expr v1.16.9 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect

go.sum

@@ -39,8 +39,6 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
 github.com/alexliesenfeld/health v0.8.0 h1:lCV0i+ZJPTbqP7LfKG7p3qZBl5VhelwUFCIVWl77fgk=
 github.com/alexliesenfeld/health v0.8.0/go.mod h1:TfNP0f+9WQVWMQRzvMUjlws4ceXKEL3WR+6Hp95HUFc=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
-github.com/antonmedv/expr v1.15.3 h1:q3hOJZNvLvhqE8OHBs1cFRdbXFNKuA+bHmRaI+AmRmI=
-github.com/antonmedv/expr v1.15.3/go.mod h1:0E/6TxnOlRNp81GMzX9QfDPAmHo2Phg00y4JUv1ihsE=
 github.com/apparentlymart/go-textseg/v13 v13.0.0 h1:Y+KvPE1NYz0xl601PVImeQfFyEy6iT90AvPUL1NNfNw=
 github.com/apparentlymart/go-textseg/v13 v13.0.0/go.mod h1:ZK2fH7c4NqDTLtiYLvIkEghdlcqw7yxLeM89kiTRPUo=
 github.com/appleboy/gin-jwt/v2 v2.9.2 h1:GeS3lm9mb9HMmj7+GNjYUtpp3V1DAQ1TkUFa5poiZ7Y=

pkg/apiclient/client.go

@@ -39,6 +39,7 @@ type ApiClient struct {
 	Metrics        *MetricsService
 	Signal         *SignalService
 	HeartBeat      *HeartBeatService
+	UsageMetrics   *UsageMetricsService
 }
 
 func (a *ApiClient) GetClient() *http.Client {
@@ -108,6 +109,7 @@ func NewClient(config *Config) (*ApiClient, error) {
 	c.Signal = (*SignalService)(&c.common)
 	c.DecisionDelete = (*DecisionDeleteService)(&c.common)
 	c.HeartBeat = (*HeartBeatService)(&c.common)
+	c.UsageMetrics = (*UsageMetricsService)(&c.common)
 
 	return c, nil
 }
@@ -144,6 +146,7 @@ func NewDefaultClient(URL *url.URL, prefix string, userAgent string, client *htt
 	c.Signal = (*SignalService)(&c.common)
 	c.DecisionDelete = (*DecisionDeleteService)(&c.common)
 	c.HeartBeat = (*HeartBeatService)(&c.common)
+	c.UsageMetrics = (*UsageMetricsService)(&c.common)
 
 	return c, nil
 }

pkg/apiclient/client_test.go

@@ -348,5 +348,5 @@ func TestNewClientBadAnswer(t *testing.T) {
 		URL:           apiURL,
 		VersionPrefix: "v1",
 	}, &http.Client{})
-	cstest.RequireErrorContains(t, err, "invalid body: invalid character 'b' looking for beginning of value")
+	cstest.RequireErrorContains(t, err, "invalid body: bad")
 }

pkg/apiclient/resperr.go

@@ -34,12 +34,18 @@
 
 	data, err := io.ReadAll(r.Body)
 	if err != nil || len(data) == 0 {
-		ret.Message = ptr.Of(fmt.Sprintf("http code %d, no error message", r.StatusCode))
+		ret.Message = ptr.Of(fmt.Sprintf("http code %d, no response body", r.StatusCode))
 		return ret
 	}
 
-	if err := json.Unmarshal(data, ret); err != nil {
-		return fmt.Errorf("http code %d, invalid body: %w", r.StatusCode, err)
+	switch r.StatusCode {
+	case http.StatusUnprocessableEntity:
+		ret.Message = ptr.Of(fmt.Sprintf("http code %d, invalid request: %s", r.StatusCode, string(data)))
+	default:
+		if err := json.Unmarshal(data, ret); err != nil {
+			ret.Message = ptr.Of(fmt.Sprintf("http code %d, invalid body: %s", r.StatusCode, string(data)))
+			return ret
+		}
 	}
 
 	return ret