Add support for usage metrics (#48)

.github/workflows/build-binary-package.yml

@@ -26,7 +26,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: 1.21.8
+        go-version: 1.23
 
     - name: Build all versions
       run: |

.github/workflows/tests.yml

@@ -20,7 +20,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: 1.21.8
+        go-version: 1.23
 
     - name: Set up Node
       uses: actions/setup-node@v4

.github/workflows/tests_deb.yml

@@ -24,7 +24,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: 1.21.8
+        go-version: 1.23
 
     - name: Cache virtualenvs
       id: cache-pipenv

Dockerfile

@@ -1,4 +1,4 @@
-ARG GOVERSION=1.21.5
+ARG GOVERSION=1.23
 
 # Stage 1: Build stage
 FROM golang:${GOVERSION}-alpine AS build

cmd/root.go

@@ -11,25 +11,145 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/crowdsecurity/crowdsec/pkg/apiclient"
 	"github.com/crowdsecurity/crowdsec/pkg/models"
 	csbouncer "github.com/crowdsecurity/go-cs-bouncer"
+	"github.com/crowdsecurity/go-cs-lib/ptr"
 	"github.com/crowdsecurity/go-cs-lib/version"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	io_prometheus_client "github.com/prometheus/client_model/go"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/crowdsecurity/crowdsec-cloudflare-worker-bouncer/pkg/cfg"
 	cf "github.com/crowdsecurity/crowdsec-cloudflare-worker-bouncer/pkg/cloudflare"
+	"github.com/crowdsecurity/crowdsec-cloudflare-worker-bouncer/pkg/metrics"
 )
 
 const (
 	DEFAULT_CONFIG_PATH = "/etc/crowdsec/bouncers/crowdsec-cloudflare-worker-bouncer.yaml"
 	name                = "crowdsec-cloudflare-worker-bouncer"
 )
 
+type metricsHandler struct {
+	cfManagers []*cf.CloudflareAccountManager
+}
+
+func getLabelValue(labels []*io_prometheus_client.LabelPair, key string) string {
+
+	for _, label := range labels {
+		if label.GetName() == key {
+			return label.GetValue()
+		}
+	}
+
+	return ""
+}
+
+func (m *metricsHandler) metricsUpdater(met *models.RemediationComponentsMetrics, updateInterval time.Duration) {
+	for _, manager := range m.cfManagers {
+		err := manager.UpdateMetrics()
+		if err != nil {
+			log.Errorf("unable to update metrics for account %s: %s", manager.AccountCfg.Name, err)
+		}
+	}
+
+	promMetrics, err := prometheus.DefaultGatherer.Gather()
+
+	if err != nil {
+		log.Errorf("unable to gather prometheus metrics: %s", err)
+		return
+	}
+
+	met.Metrics = append(met.Metrics, &models.DetailedMetrics{
+		Meta: &models.MetricsMeta{
+			UtcNowTimestamp:   ptr.Of(time.Now().Unix()),
+			WindowSizeSeconds: ptr.Of(int64(updateInterval.Seconds())),
+		},
+		Items: make([]*models.MetricsDetailItem, 0),
+	})
+
+	for _, metricFamily := range promMetrics {
+		for _, metric := range metricFamily.GetMetric() {
+			switch metricFamily.GetName() {
+			case metrics.ActiveDecisionsMetricName:
+				//We send the absolute value, as it makes no sense to try to sum them crowdsec side
+				labels := metric.GetLabel()
+				value := metric.GetGauge().GetValue()
+				origin := getLabelValue(labels, "origin")
+				ipType := getLabelValue(labels, "ip_type")
+				account := getLabelValue(labels, "account")
+				remediation := getLabelValue(labels, "remediation")
+				log.Debugf("Sending active decisions for %s %s %s %s| current value: %f", origin, ipType, remediation, account, value)
+				met.Metrics[0].Items = append(met.Metrics[0].Items, &models.MetricsDetailItem{
+					Name:  ptr.Of("active_decisions"),
+					Value: ptr.Of(value),
+					Labels: map[string]string{
+						"origin":      origin,
+						"ip_type":     ipType,
+						"account":     account,
+						"remediation": remediation,
+					},
+					Unit: ptr.Of("ip"),
+				})
+			case metrics.BlockedRequestMetricName:
+				labels := metric.GetLabel()
+				value := metric.GetGauge().GetValue()
+				origin := getLabelValue(labels, "origin")
+				ipType := getLabelValue(labels, "ip_type")
+				account := getLabelValue(labels, "account")
+				remediation := getLabelValue(labels, "remediation")
+				key := origin + ipType + account + remediation
+				log.Debugf("Sending dropped bytes for %s %s %s %s %f | current value: %f | previous value: %f\n", origin, ipType, remediation, account, value-metrics.LastBlockedRequestValue[key], value, metrics.LastBlockedRequestValue[key])
+				met.Metrics[0].Items = append(met.Metrics[0].Items, &models.MetricsDetailItem{
+					Name:  ptr.Of("dropped"),
+					Value: ptr.Of(value - metrics.LastBlockedRequestValue[key]),
+					Labels: map[string]string{
+						"origin":      origin,
+						"ip_type":     ipType,
+						"account":     account,
+						"remediation": remediation,
+					},
+					Unit: ptr.Of("request"),
+				})
+				metrics.LastBlockedRequestValue[key] = value
+			case metrics.ProcessedRequestMetricName:
+				labels := metric.GetLabel()
+				value := metric.GetGauge().GetValue()
+				ipType := getLabelValue(labels, "ip_type")
+				account := getLabelValue(labels, "account")
+				key := ipType + account
+				log.Debugf("Sending processed packets for %s %s %f | current value: %f | previous value: %f\n", ipType, account, value-metrics.LastProcessedRequestValue[key], value, metrics.LastProcessedRequestValue[key])
+				met.Metrics[0].Items = append(met.Metrics[0].Items, &models.MetricsDetailItem{
+					Name:  ptr.Of("processed"),
+					Value: ptr.Of(value - metrics.LastProcessedRequestValue[key]),
+					Labels: map[string]string{
+						"ip_type": ipType,
+						"account": account,
+					},
+					Unit: ptr.Of("request"),
+				})
+				metrics.LastProcessedRequestValue[key] = value
+			}
+		}
+	}
+}
+
+func (m *metricsHandler) computeMetricsHandler(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		for _, manager := range m.cfManagers {
+			err := manager.UpdateMetrics()
+			if err != nil {
+				log.Errorf("unable to update metrics for account %s: %s", manager.AccountCfg.Name, err)
+			}
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
 func cleanUp(managers []*cf.CloudflareAccountManager, c context.CancelFunc, ctx context.Context) {
 	var g errgroup.Group
 	c()
@@ -38,7 +158,7 @@ func cleanUp(managers []*cf.CloudflareAccountManager, c context.CancelFunc, ctx
 		manager := m
 		manager.Ctx = context.Background()
 		g.Go(func() error {
-			return manager.CleanUpExistingWorkers()
+			return manager.CleanUpExistingWorkers(false)
 		})
 	}
 	if err := g.Wait(); err != nil {
@@ -152,7 +272,7 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 		CAPath:   conf.CrowdSecConfig.CAPath,
 	}
 
-	if (testConfig != nil && *testConfig) || (setupOnly == nil || (setupOnly != nil && !*setupOnly)) || (deleteOnly == nil || (deleteOnly != nil && !*deleteOnly)) {
+	if (testConfig != nil && *testConfig) || (setupOnly == nil || !*setupOnly) || (deleteOnly == nil || !*deleteOnly) {
 		if err := csLAPI.Init(); err != nil {
 			return fmt.Errorf("unable to initialize crowdsec bouncer: %w", err)
 		}
@@ -172,7 +292,7 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 	for _, cfManager := range cfManagers {
 		manager := cfManager
 		g.Go(func() error {
-			err := manager.CleanUpExistingWorkers()
+			err := manager.CleanUpExistingWorkers(true)
 			if err != nil {
 				return fmt.Errorf("unable to cleanup existing workers: %w for account %s", err, manager.AccountCfg.Name)
 			}
@@ -210,6 +330,8 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 		})
 	}
 
+	defer cleanUp(cfManagers, cancel, ctx)
+
 	g.Go(func() error {
 		return HandleSignals(ctx)
 	})
@@ -219,14 +341,28 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 		return fmt.Errorf("crowdsec bouncer stopped")
 	})
 
+	mHandler := metricsHandler{
+		cfManagers: cfManagers,
+	}
+
+	metricsProvider, err := csbouncer.NewMetricsProvider(csLAPI.APIClient, name, mHandler.metricsUpdater, log.StandardLogger())
+	if err != nil {
+		return fmt.Errorf("unable to create metrics provider: %w", err)
+	}
+
+	g.Go(func() error {
+		return metricsProvider.Run(ctx)
+	})
+
+	prometheus.MustRegister(csbouncer.TotalLAPICalls, csbouncer.TotalLAPIError, metrics.CloudflareAPICallsByAccount, metrics.TotalKeysByAccount,
+		metrics.TotalActiveDecisions, metrics.TotalBlockedRequests, metrics.TotalProcessedRequests)
 	if conf.PrometheusConfig.Enabled {
-		prometheus.MustRegister(csbouncer.TotalLAPICalls, csbouncer.TotalLAPIError, cf.CloudflareAPICallsByAccount, cf.TotalKeysByAccount)
 		g.Go(func() error {
-			http.Handle("/metrics", promhttp.Handler())
+			http.Handle("/metrics", mHandler.computeMetricsHandler(promhttp.Handler()))
 			return http.ListenAndServe(net.JoinHostPort(conf.PrometheusConfig.ListenAddress, conf.PrometheusConfig.ListenPort), nil)
 		})
 	}
-	defer cleanUp(cfManagers, cancel, ctx)
+
 	for {
 		select {
 		case <-ctx.Done():

cmd/root_test.go

@@ -120,7 +120,7 @@ func runInfraTest(t *testing.T, m *cf.CloudflareAccountManager) error {
 }
 
 func runCleanUpTest(t *testing.T, m *cf.CloudflareAccountManager) error {
-	if err := m.CleanUpExistingWorkers(); err != nil {
+	if err := m.CleanUpExistingWorkers(false); err != nil {
 		return err
 	}
 	api, err := apiFromManager(m)

go.mod

@@ -1,11 +1,11 @@
 module github.com/crowdsecurity/crowdsec-cloudflare-worker-bouncer
 
-go 1.21
+go 1.23
 
 require (
-	github.com/crowdsecurity/crowdsec v1.6.1
-	github.com/crowdsecurity/go-cs-bouncer v0.0.13
-	github.com/prometheus/client_golang v1.19.0
+	github.com/crowdsecurity/crowdsec v1.6.3
+	github.com/crowdsecurity/go-cs-bouncer v0.0.14
+	github.com/prometheus/client_golang v1.20.5
 	github.com/sirupsen/logrus v1.9.3
 	github.com/whuang8/redactrus v1.0.2
 	golang.org/x/sync v0.8.0
@@ -16,8 +16,11 @@ require (
 	github.com/antonmedv/expr v1.15.5 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blackfireio/osinfo v1.0.5 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/fatih/color v1.16.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/expr-lang/expr v1.16.9 // indirect
+	github.com/fatih/color v1.18.0 // indirect
 	github.com/go-openapi/analysis v0.23.0 // indirect
 	github.com/go-openapi/errors v0.22.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
@@ -28,32 +31,34 @@ require (
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/go-openapi/validate v0.24.0 // indirect
 	github.com/goccy/go-json v0.10.3 // indirect
-	github.com/goccy/go-yaml v1.11.3 // indirect
+	github.com/goccy/go-yaml v1.12.0 // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
-	github.com/prometheus/common v0.53.0 // indirect
-	github.com/prometheus/procfs v0.14.0 // indirect
-	go.mongodb.org/mongo-driver v1.15.0 // indirect
-	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
-	google.golang.org/protobuf v1.33.0 // indirect
+	github.com/prometheus/common v0.60.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
+	go.mongodb.org/mongo-driver v1.17.1 // indirect
+	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
+	google.golang.org/protobuf v1.35.1 // indirect
 	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
 
 require (
 	github.com/cloudflare/cloudflare-go v0.103.0
-	github.com/crowdsecurity/go-cs-lib v0.0.10
+	github.com/crowdsecurity/go-cs-lib v0.0.15
 	github.com/google/go-querystring v1.1.0 // indirect
-	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/sys v0.25.0 // indirect
-	golang.org/x/text v0.18.0 // indirect
+	golang.org/x/net v0.30.0 // indirect
+	golang.org/x/sys v0.26.0 // indirect
+	golang.org/x/text v0.19.0 // indirect
 	golang.org/x/time v0.6.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1
 )