Capture metrics with more granular labels

Author: hslatman

appsec/appsec.go

@@ -29,6 +29,7 @@ import (
 	"github.com/hslatman/caddy-crowdsec-bouncer/crowdsec"
 	"github.com/hslatman/caddy-crowdsec-bouncer/internal/bouncer"
 	"github.com/hslatman/caddy-crowdsec-bouncer/internal/httputils"
+	"github.com/hslatman/caddy-crowdsec-bouncer/internal/servername"
 )
 
 func init() {
@@ -87,11 +88,14 @@ func (h *Handler) Cleanup() error {
 // ServeHTTP is the Caddy handler for serving HTTP requests.
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
 	var (
-		ctx = r.Context()
-		ip  netip.Addr
+		ctx    = r.Context()
+		ip     netip.Addr
+		server = servername.FromContext(ctx)
 	)
 
 	ctx, ip = httputils.EnsureIP(ctx)
+	defer h.crowdsec.IncrementProcessedRequests(server, ip.Is6())
+
 	r = r.WithContext(ctx)
 	if err := h.crowdsec.CheckRequest(ctx, r); err != nil {
 		a := &bouncer.AppSecError{}
@@ -102,10 +106,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
 		switch a.Action {
 		case "allow":
 			// nothing to do
+			h.crowdsec.IncrementBlockedRequests(server, "appsec", "bypass", ip.Is6()) // TODO: properly set the action that was performed
 		case "log":
 			h.logger.Info("appsec rule triggered", zap.String("ip", ip.String()), zap.String("action", a.Action))
+			h.crowdsec.IncrementBlockedRequests(server, "appsec", "log", ip.Is6()) // TODO: properly set the action that was performed
 		default:
-			return httputils.WriteResponse(w, h.logger, a.Action, ip.String(), a.Duration, a.StatusCode)
+			if err := httputils.WriteResponse(w, h.logger, a.Action, ip.String(), a.Duration, a.StatusCode); err != nil {
+				h.crowdsec.IncrementBlockedRequests(server, "appsec", a.Action, ip.Is6()) // TODO: properly set the action that was performed
+				return err
+			}
+
+			h.crowdsec.IncrementBlockedRequests(server, "appsec", a.Action, ip.Is6()) // TODO: properly set the action that was performed
+			return nil
 		}
 	}
 

crowdsec/crowdsec.go

@@ -313,6 +313,14 @@ func (c *CrowdSec) CheckRequest(ctx context.Context, r *http.Request) error {
 	return c.bouncer.CheckRequest(ctx, r)
 }
 
+func (c *CrowdSec) IncrementProcessedRequests(server string, isIPv6 bool) {
+	c.bouncer.IncrementProcessedRequests(server, isIPv6)
+}
+
+func (c *CrowdSec) IncrementBlockedRequests(server, origin, remediation string, isIPv6 bool) {
+	c.bouncer.IncrementBlockedRequests(server, origin, remediation, isIPv6)
+}
+
 func (c *CrowdSec) metricsInterval() time.Duration {
 	return time.Duration(c.MetricsInterval)
 }

crowdsec/crowdsec_test.go

@@ -205,16 +205,13 @@ func TestCrowdSecStreamingBouncerRuntime(t *testing.T) {
 	require.NoError(t, err)
 
 	wg := &sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		// simulate request coming in and stopping the server from another goroutine
-		defer wg.Done()
+	wg.Go(func() {
 
 		// wait a little bit of time to let the go-cs-bouncer do _some_ work,
 		// before it properly returns; seems to hang otherwise on b.wg.Wait().
 		time.Sleep(100 * time.Millisecond)
 
-		// simulate a lookup
+		// simulate request coming in and stopping the server from another goroutine
 		allowed, decision, err := c.IsAllowed(netip.MustParseAddr("127.0.0.1"))
 		assert.NoError(t, err)
 		assert.Nil(t, decision)
@@ -225,7 +222,7 @@ func TestCrowdSecStreamingBouncerRuntime(t *testing.T) {
 
 		err = c.Cleanup()
 		require.NoError(t, err)
-	}()
+	})
 
 	// wait for the stop and cleanup process
 	wg.Wait()

go.mod

@@ -9,11 +9,12 @@ require (
 	github.com/crowdsecurity/go-cs-lib v0.0.15
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
-	github.com/hslatman/ipstore v0.4.0
+	github.com/hslatman/ipstore v0.5.0
 	github.com/jarcoal/httpmock v1.4.1
 	github.com/mholt/caddy-l4 v0.0.0-20231016112149-a362a1fbf652
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c
 	github.com/prometheus/client_golang v1.23.0
+	github.com/prometheus/client_model v0.6.2
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.9.1
 	github.com/stretchr/testify v1.11.1
@@ -137,7 +138,6 @@ require (
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.65.0 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
 	github.com/quic-go/qpack v0.5.1 // indirect

go.sum

@@ -288,8 +288,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.5.0/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpg
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 h1:X5VWvz21y3gzm9Nw/kaUeku/1+uBhcekkmy4IkffJww=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1/go.mod h1:Zanoh4+gvIgluNqcfMVTJueD4wSS5hT7zTt4Mrutd90=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hslatman/ipstore v0.4.0 h1:s9HAmDwJ5gdpD1heI24XqDdWaZJvZfp7DERimtV50uw=
-github.com/hslatman/ipstore v0.4.0/go.mod h1:IauyP66Q5kk1mfjysfAfa2Y4AnN7tFLYJlYHhkPlEPY=
+github.com/hslatman/ipstore v0.5.0 h1:pFbh8OKWOCttHwMvjtF9O02vPKI5HSHlhRk5hlO82rk=
+github.com/hslatman/ipstore v0.5.0/go.mod h1:IauyP66Q5kk1mfjysfAfa2Y4AnN7tFLYJlYHhkPlEPY=
 github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI=
 github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=

http/http.go

@@ -29,6 +29,7 @@ import (
 	_ "github.com/hslatman/caddy-crowdsec-bouncer/appsec" // always include AppSec module when HTTP is added
 	"github.com/hslatman/caddy-crowdsec-bouncer/crowdsec"
 	"github.com/hslatman/caddy-crowdsec-bouncer/internal/httputils"
+	"github.com/hslatman/caddy-crowdsec-bouncer/internal/servername"
 )
 
 func init() {
@@ -86,11 +87,14 @@ func (h *Handler) Cleanup() error {
 // ServeHTTP is the Caddy handler for serving HTTP requests.
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
 	var (
-		ctx = r.Context()
-		ip  netip.Addr
+		ctx    = r.Context()
+		ip     netip.Addr
+		server = servername.FromContext(ctx)
 	)
 
 	ctx, ip = httputils.EnsureIP(ctx)
+	defer h.crowdsec.IncrementProcessedRequests(server, ip.Is6())
+
 	isAllowed, decision, err := h.crowdsec.IsAllowed(ip)
 	if err != nil {
 		return err // TODO: return error here? Or just log it and continue serving
@@ -104,8 +108,16 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
 		typ := *decision.Type
 		value := *decision.Value
 		duration := *decision.Duration
+		origin := *decision.Origin
+
+		if err := httputils.WriteResponse(w, h.logger, typ, value, duration, 0); err != nil {
+			h.crowdsec.IncrementBlockedRequests(server, origin, typ, ip.Is6()) // TODO: properly set the action that was performed
+			return err
+		}
 
-		return httputils.WriteResponse(w, h.logger, typ, value, duration, 0)
+		h.crowdsec.IncrementBlockedRequests(server, origin, typ, ip.Is6()) // TODO: properly set the action that was performed
+
+		return nil
 	}
 
 	// Continue down the handler stack

internal/bouncer/bouncer.go

@@ -241,11 +241,6 @@ func (b *Bouncer) Shutdown() error {
 // IsAllowed checks if an IP is allowed or not
 func (b *Bouncer) IsAllowed(ip netip.Addr, forceLive bool) (bool, *models.Decision, error) {
 	isAllowed, decision, err := b.isAllowed(ip, forceLive)
-
-	if !isAllowed {
-		blockedRequestsCounter.Inc()
-	}
-
 	return isAllowed, decision, err
 }
 
@@ -276,6 +271,23 @@ func (b *Bouncer) CheckRequest(ctx context.Context, r *http.Request) error {
 	return b.appsec.checkRequest(ctx, r)
 }
 
+func toIPType(isIPv6 bool) (ipType string) {
+	ipType = "ipv4"
+	if isIPv6 {
+		ipType = "ipv6"
+	}
+
+	return
+}
+
+func (b *Bouncer) IncrementProcessedRequests(server string, isIPv6 bool) {
+	processedRequestsCounter.With(prometheus.Labels{"server": server, "ip_type": toIPType(isIPv6)}).Inc() // TODO: test whether these globals are OK to use
+}
+
+func (b *Bouncer) IncrementBlockedRequests(server, origin, remediation string, isIPv6 bool) {
+	blockedRequestsCounter.With(prometheus.Labels{"server": server, "origin": origin, "remediation": remediation, "ip_type": toIPType(isIPv6)}).Inc() // TODO: test whether these globals are OK to use
+}
+
 func generateInstanceID(t time.Time) (string, error) {
 	r := rand.New(rand.NewSource(t.Unix()))
 	b := [4]byte{}

internal/bouncer/decisions.go

@@ -132,26 +132,46 @@ func (b *Bouncer) retrieveDecision(ip netip.Addr, forceLive bool) (*models.Decis
 	return nil, nil
 }
 
+type ipType string
+
+const (
+	ipv4 ipType = "ipv4"
+	ipv6 ipType = "ipv6"
+)
+
 func (b *Bouncer) recalculateAndRecordDecisionCounts() {
 	// initialize map, so that these origins are always
 	// known and recorded
-	m := map[string]float64{
-		"CAPI":             0,
-		"crowdsec":         0,
-		"cscli":            0,
-		"cscli-import":     0,
-		"console":          0,
-		"appsec":           0,
-		"remediation_sync": 0,
+	m := map[string]map[ipType]int{
+		"CAPI":             {ipv4: 0, ipv6: 0},
+		"crowdsec":         {ipv4: 0, ipv6: 0},
+		"cscli":            {ipv4: 0, ipv6: 0},
+		"cscli-import":     {ipv4: 0, ipv6: 0},
+		"console":          {ipv4: 0, ipv6: 0},
+		"appsec":           {ipv4: 0, ipv6: 0},
+		"remediation_sync": {ipv4: 0, ipv6: 0},
 	}
 
-	for _, v := range b.store.store.All() {
+	for prefix, v := range b.store.store.All() {
 		origin := *v.Origin
-		m[origin] += 1
+		isIPv6 := prefix.Bits() > 32
+		n, ok := m[origin]
+		if !ok {
+			n = map[ipType]int{ipv4: 0, ipv6: 0}
+			m[origin] = n
+		}
+
+		if isIPv6 {
+			n[ipv6] += 1
+		} else {
+			n[ipv4] += 1
+		}
 	}
 
-	for origin, count := range m {
-		// update the count of active decisions
-		activeDecisionsGauge.With(map[string]string{"origin": origin}).Set(count)
+	// update the count of active decisions per origin and IP type
+	for origin, tuple := range m {
+		for ipType, count := range tuple {
+			activeDecisionsGauge.With(map[string]string{"origin": origin, "ip_type": string(ipType)}).Set(float64(count))
+		}
 	}
 }

internal/bouncer/metrics.go

@@ -53,15 +53,15 @@ var (
 	activeDecisionsGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Name: string(activeDecisionsName),
 		Help: "The current number of active decisions",
-	}, []string{"origin"}) // TODO: additional labels, similar to firewall bouncer?
+	}, []string{"origin", "ip_type"})
 	blockedRequestsCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: string(blockedRequestsCounterName),
 		Help: "The total number of requests blocked",
-	}, []string{"origin", "remediation"})
-	processedRequestsCounter = prometheus.NewCounter(prometheus.CounterOpts{
+	}, []string{"server", "origin", "remediation", "ip_type"})
+	processedRequestsCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: string(processedRequestsCounterName),
 		Help: "The total number of requests handled",
-	})
+	}, []string{"server", "ip_type"})
 
 	// TODO: referencing the global metrics from csbouncer may not be the right
 	// thing to do with how the CrowdSec module operates as part of Caddy. On
@@ -113,32 +113,34 @@ func newMetricsProvider(client *apiclient.ApiClient, metricsRegistry, caddyMetri
 			Name:         "active_decisions",
 			Unit:         "ip",
 			Collector:    activeDecisionsGauge,
-			LabelKeys:    []string{"origin"},
+			LabelKeys:    []string{"origin", "ip_type"},
 			LastValueMap: nil, // absolute value
 			KeyFunc: func(labels []*model.LabelPair) string {
-				return getLabelValue(labels, "origin")
+				return getLabelValue(labels, "origin") + getLabelValue(labels, "ip_type")
 			},
 			SendToLAPI: true,
 		},
 		blockedRequestsCounterName: {
 			Name:         "dropped",
 			Unit:         "request",
 			Collector:    blockedRequestsCounter,
-			LabelKeys:    []string{"origin", "remediation"},
+			LabelKeys:    []string{"server", "origin", "remediation", "ip_type"},
 			LastValueMap: make(map[string]float64),
 			KeyFunc: func(labels []*model.LabelPair) string {
-				return getLabelValue(labels, "origin") + getLabelValue(labels, "remediation")
+				return getLabelValue(labels, "server") + getLabelValue(labels, "origin") + getLabelValue(labels, "remediation") + getLabelValue(labels, "ip_type")
 			},
 			SendToLAPI: true,
 		},
 		processedRequestsCounterName: {
 			Name:         "processed",
 			Unit:         "request",
 			Collector:    processedRequestsCounter,
-			LabelKeys:    []string{},
+			LabelKeys:    []string{"server", "ip_type"},
 			LastValueMap: make(map[string]float64),
-			KeyFunc:      func([]*model.LabelPair) string { return "" },
-			SendToLAPI:   true,
+			KeyFunc: func(labels []*model.LabelPair) string {
+				return getLabelValue(labels, "server") + getLabelValue(labels, "ip_type")
+			},
+			SendToLAPI: true,
 		},
 		totalBouncerCallsName: {
 			Name:         "bouncer_calls",
@@ -198,7 +200,7 @@ func newMetricsProvider(client *apiclient.ApiClient, metricsRegistry, caddyMetri
 			Name:    &osName,
 			Version: &osVersion,
 		},
-		bouncerFeatureFlags: []string{}, // not used in bouncers
+		bouncerFeatureFlags: []string{}, // not used in bouncers?
 		logger:              logger.With(zap.String("instance_id", instanceID)),
 		instanceID:          instanceID,
 	}
@@ -229,7 +231,7 @@ func (m *metricsProvider) metricsPayload(now time.Time) (metrics *models.AllMetr
 	metrics = &models.AllMetrics{
 		RemediationComponents: []*models.RemediationComponentsMetrics{
 			{
-				Name: userAgentName, // TODO: verify this is OK to use as-is
+				Name: userAgentName,
 				Type: m.bouncerType,
 				BaseMetrics: models.BaseMetrics{
 					Os:                  &m.bouncerOS,

internal/servername/context.go

@@ -0,0 +1,41 @@
+package servername
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
+	l4 "github.com/mholt/caddy-l4/layer4"
+)
+
+const unknownServerName = "UNKNOWN"
+
+// FromContext extracts the current server name from the
+// [context.Context]. Returns "UNKNOWN" string if none is available.
+func FromContext(ctx context.Context) string {
+	srv, ok := ctx.Value(caddyhttp.ServerCtxKey).(*caddyhttp.Server)
+	if !ok || srv == nil {
+		return ""
+	}
+
+	if srv.Name() == "" {
+		return unknownServerName
+	}
+
+	return srv.Name()
+}
+
+// FromConnection extracts the current server name from the
+// [l4.Connection]. Returns "UNKNOWN" string if none is available.
+func FromConnection(cx *l4.Connection) string {
+	server := FromContext(cx.Context) // TODO: change layer4 to also have a server name; they're named
+	if server == "" {
+		server = cx.LocalAddr().String()
+	}
+
+	if server == "" {
+		return unknownServerName
+	}
+
+	return fmt.Sprintf("layer4-%s", server)
+}