Report observed egress hosts via the metrics pipeline

internal/handlers/egress_allowlist.go

@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/elazarl/goproxy"
@@ -19,6 +20,18 @@ const (
 	egressEnforceExperiment = "proxy_egress_enforce"
 )
 
+// egressHostMetric is the metric emitted for every observed outbound host. The
+// backend logs its raw request_host tag for allowlist discovery and buckets the
+// host before forwarding to Datadog to keep tag cardinality low.
+const egressHostMetric = "egress_host"
+
+// MetricSender emits a metric for each observed outbound host, reusing the
+// proxy's existing metrics collector (buffering, flushing, retries, and
+// job-lifecycle handling) instead of a dedicated reporting pipeline.
+type MetricSender interface {
+	SendMetric(name string, metricType string, value float64, additionalTags map[string]string) error
+}
+
 // EgressAllowlistHandler filters outbound requests against a per-job allowlist
 // of non-hostile domains. In observe mode it only logs non-allowlisted hosts;
 // in enforce mode it drops them with a 403. When neither flag is set it allows
@@ -27,14 +40,17 @@ type EgressAllowlistHandler struct {
 	observe bool
 	enforce bool
 	allowed []string
+	metrics MetricSender
 }
 
 // NewEgressAllowlistHandler builds the allowlist from the always-allowed GitHub
 // infrastructure domains, the union of every ecosystem's default registry hosts,
 // and the job's dynamic hosts (configured registries and OIDC token-exchange
 // endpoints derived from cfg.Credentials). The observe/enforce toggles are
-// driven by job experiments.
-func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings) *EgressAllowlistHandler {
+// driven by job experiments. The metric sender, when non-nil, receives an
+// observation for every host (with its allowlisted status) for reporting to the
+// backend.
+func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings, metricSender MetricSender) *EgressAllowlistHandler {
 	allowed := append([]string(nil), githubInfraDomains...)
 	allowed = append(allowed, allEcosystemDomains...)
 	allowed = append(allowed, dynamicHosts(cfg.Credentials)...)
@@ -43,6 +59,7 @@ func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings)
 		observe: cfg.Experiments.Enabled(egressObserveExperiment),
 		enforce: cfg.Experiments.Enabled(egressEnforceExperiment),
 		allowed: allowed,
+		metrics: metricSender,
 	}
 }
 
@@ -53,19 +70,40 @@ func (h *EgressAllowlistHandler) HandleRequest(req *http.Request, proxyCtx *gopr
 	}
 
 	host := helpers.GetHost(req)
-	if host == "" || h.isAllowed(host) {
+	if host == "" {
 		return req, nil
 	}
 
-	if h.observe {
-		logging.RequestLogf(proxyCtx, "* egress not allowlisted %s", host)
-	}
-	if h.enforce {
-		return req, goproxy.NewResponse(req, goproxy.ContentTypeText, http.StatusForbidden, "Forbidden")
+	allowed := h.isAllowed(host)
+
+	// Record the observation here, at the point of the allowlist decision, so
+	// that enforce-blocked hosts are captured before the 403 short-circuits the
+	// request chain (the downstream metrics handler would never see them).
+	h.recordHost(host, allowed)
+
+	if !allowed {
+		if h.observe {
+			logging.RequestLogf(proxyCtx, "* egress not allowlisted %s", host)
+		}
+		if h.enforce {
+			return req, goproxy.NewResponse(req, goproxy.ContentTypeText, http.StatusForbidden, "Forbidden")
+		}
 	}
 	return req, nil
 }
 
+func (h *EgressAllowlistHandler) recordHost(host string, allowed bool) {
+	if h.metrics == nil {
+		return
+	}
+	// package_manager is added by the collector's default tags. request_host is
+	// the raw host; the backend buckets it before emitting to Datadog.
+	_ = h.metrics.SendMetric(egressHostMetric, "increment", 1, map[string]string{
+		"request_host": host,
+		"allowlisted":  strconv.FormatBool(allowed),
+	})
+}
+
 func (h *EgressAllowlistHandler) isAllowed(host string) bool {
 	// Normalize an absolute DNS name (trailing dot) so exact matches treat
 	// "registry.npmjs.org." as equivalent to "registry.npmjs.org", consistent

internal/handlers/egress_allowlist_test.go

@@ -6,12 +6,13 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/dependabot/proxy/internal/config"
 )
 
 func newEgressHandler(observe, enforce bool, packageManager string) *EgressAllowlistHandler {
-	return NewEgressAllowlistHandler(egressCfg(observe, enforce), config.ProxyEnvSettings{PackageManager: packageManager})
+	return NewEgressAllowlistHandler(egressCfg(observe, enforce), config.ProxyEnvSettings{PackageManager: packageManager}, nil)
 }
 
 // egressCfg builds a Config whose experiments toggle the egress observe/enforce
@@ -30,7 +31,7 @@ func egressCfg(observe, enforce bool) *config.Config {
 func newEgressHandlerWithCreds(creds config.Credentials) *EgressAllowlistHandler {
 	cfg := egressCfg(false, true)
 	cfg.Credentials = creds
-	return NewEgressAllowlistHandler(cfg, config.ProxyEnvSettings{})
+	return NewEgressAllowlistHandler(cfg, config.ProxyEnvSettings{}, nil)
 }
 
 // egressResult runs HandleRequest and returns the response (nil means allowed).
@@ -129,6 +130,59 @@ func TestEgressAllowlist_SuffixEntryAllowsSubdomain(t *testing.T) {
 	assert.Nil(t, egressResult(t, h, "https://europe-docker.pkg.dev/v2/project/image"), "artifact registry subdomain allowed")
 }
 
+// fakeMetricSender captures the metrics emitted by the egress handler.
+type fakeMetricSender struct {
+	metrics []sentMetric
+}
+
+type sentMetric struct {
+	name string
+	tags map[string]string
+}
+
+func (s *fakeMetricSender) SendMetric(name string, _ string, _ float64, additionalTags map[string]string) error {
+	s.metrics = append(s.metrics, sentMetric{name: name, tags: additionalTags})
+	return nil
+}
+
+func TestEgressAllowlist_RecordsObservedHosts(t *testing.T) {
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(true, false), config.ProxyEnvSettings{}, sender)
+
+	egressResult(t, h, "https://registry.npmjs.org/left-pad")
+	egressResult(t, h, "https://evil.com/steal")
+
+	assert.Equal(t, []sentMetric{
+		{name: egressHostMetric, tags: map[string]string{"request_host": "registry.npmjs.org", "allowlisted": "true"}},
+		{name: egressHostMetric, tags: map[string]string{"request_host": "evil.com", "allowlisted": "false"}},
+	}, sender.metrics)
+}
+
+// TestEgressAllowlist_RecordsEnforceBlockedHosts verifies that a host blocked in
+// enforce mode is still recorded, since the observation happens before the 403
+// short-circuits the request chain.
+func TestEgressAllowlist_RecordsEnforceBlockedHosts(t *testing.T) {
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(false, true), config.ProxyEnvSettings{}, sender)
+
+	resp := egressResult(t, h, "https://evil.com/steal")
+	require.NotNil(t, resp, "enforce blocks the host")
+	assert.Equal(t, http.StatusForbidden, resp.StatusCode)
+
+	assert.Equal(t, []sentMetric{
+		{name: egressHostMetric, tags: map[string]string{"request_host": "evil.com", "allowlisted": "false"}},
+	}, sender.metrics, "blocked host is recorded despite the 403")
+}
+
+func TestEgressAllowlist_DoesNotRecordWhenDisabled(t *testing.T) {
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(false, false), config.ProxyEnvSettings{}, sender)
+
+	egressResult(t, h, "https://evil.com/steal")
+
+	assert.Empty(t, sender.metrics, "fail-open mode records nothing")
+}
+
 func TestEgressAllowlist_UnknownOrEmptyPackageManagerStillGetsUnion(t *testing.T) {
 	// The allowlist does not depend on PACKAGE_MANAGER: an unknown or empty
 	// value still yields GitHub infra + the full ecosystem union.

internal/metrics/collector_client.go

@@ -21,7 +21,7 @@ type CollectorClient struct {
 	JobID               string
 	MetricsBuffer       []map[string]any
 	BufferMutex         sync.Mutex
-	MaxBufferSize       int
+	MaxSeriesPerMetric  int
 	FlushTicker         *time.Ticker
 	estimatedBufferSize int
 	closeCh             chan struct{}
@@ -43,12 +43,19 @@ type Client interface {
 func New(envSettings config.ProxyEnvSettings, apiClient apiclient.ClientInterface) *CollectorClient {
 
 	instance = &CollectorClient{
-		APIClient:           apiClient,
-		APIEndpoint:         envSettings.APIEndpoint,
-		DefaultTags:         map[string]string{"package_manager": envSettings.PackageManager, "grouped_update": envSettings.GroupedUpdate},
-		JobID:               envSettings.JobID,
-		MetricsBuffer:       make([]map[string]any, 0),
-		MaxBufferSize:       1000,
+		APIClient:     apiClient,
+		APIEndpoint:   envSettings.APIEndpoint,
+		DefaultTags:   map[string]string{"package_manager": envSettings.PackageManager, "grouped_update": envSettings.GroupedUpdate},
+		JobID:         envSettings.JobID,
+		MetricsBuffer: make([]map[string]any, 0),
+		// MaxSeriesPerMetric bounds the number of distinct series buffered per
+		// metric name between flushes. It is applied per name (rather than as a
+		// single shared cap) so a high-cardinality metric such as egress_host,
+		// whose request_host tag is a raw hostname, cannot fill the buffer and
+		// starve the ordinary request/response metrics. 500 is comfortably above
+		// the distinct-host and bucketed-tag counts either metric produces in a
+		// one-minute flush window.
+		MaxSeriesPerMetric:  500,
 		FlushTicker:         time.NewTicker(1 * time.Minute),
 		estimatedBufferSize: 0,
 		closeCh:             make(chan struct{}),
@@ -88,22 +95,29 @@ func (c *CollectorClient) canSendMetrics() bool {
 	return c.APIEndpoint != ""
 }
 
-func (c *CollectorClient) flushBuffer() {
-	// To avoid sending metrics during smoke tests in CI build
-	if !c.canSendMetrics() {
-		logrus.Info("Skipping sending metrics because api endpoint is empty")
-		return
+// drainLocked snapshots the buffered metrics, clears the buffer, and resets the
+// running size estimate. The caller MUST already hold BufferMutex. Resetting the
+// estimate here (on every drain) is what keeps it in sync with the buffer;
+// previously flushBuffer cleared the slice but left estimatedBufferSize growing
+// forever, which eventually tripped the payload-size branch in SendMetric.
+func (c *CollectorClient) drainLocked() []map[string]any {
+	if len(c.MetricsBuffer) == 0 {
+		return nil
 	}
+	batch := c.MetricsBuffer
+	c.MetricsBuffer = make([]map[string]any, 0)
+	c.estimatedBufferSize = 0
+	return batch
+}
 
-	c.BufferMutex.Lock()
-	if len(c.MetricsBuffer) == 0 {
-		c.BufferMutex.Unlock()
+// sendBatch marshals a drained batch and posts it to the API. It performs
+// blocking network I/O and MUST be called without holding BufferMutex.
+func (c *CollectorClient) sendBatch(batch []map[string]any) {
+	if len(batch) == 0 {
 		return
 	}
-	jsonData, err := json.Marshal(map[string]any{"data": c.MetricsBuffer})
-	c.MetricsBuffer = c.MetricsBuffer[:0] // Reset buffer
-	c.BufferMutex.Unlock()
 
+	jsonData, err := json.Marshal(map[string]any{"data": batch})
 	if err != nil {
 		logrus.Errorln("Error marshaling metrics data:", err)
 		return
@@ -122,6 +136,21 @@ func (c *CollectorClient) flushBuffer() {
 	}
 }
 
+func (c *CollectorClient) flushBuffer() {
+	// To avoid sending metrics during smoke tests in CI build. Checked before
+	// draining so the buffer is preserved when there is no endpoint to post to.
+	if !c.canSendMetrics() {
+		logrus.Info("Skipping sending metrics because api endpoint is empty")
+		return
+	}
+
+	c.BufferMutex.Lock()
+	batch := c.drainLocked()
+	c.BufferMutex.Unlock()
+
+	c.sendBatch(batch)
+}
+
 func (c *CollectorClient) SendMetric(name string, metricType string, value float64, additionalTags map[string]string) error {
 	// This check is in place to prevent the transmission of metrics initiated by smoke tests
 	prefixedName := "dependabot.job_proxy." + name
@@ -131,31 +160,60 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 	maps.Copy(combinedTags, c.DefaultTags)
 	maps.Copy(combinedTags, additionalTags)
 
+	// A batch drained because the payload-size limit was reached is sent after
+	// the mutex is released. This defer is registered before the Unlock defer so
+	// that, LIFO, the unlock runs first: we never perform network I/O or re-lock
+	// BufferMutex while already holding it (which previously deadlocked when the
+	// size-triggered path called flushBuffer).
+	var toSend []map[string]any
+	defer func() { c.sendBatch(toSend) }()
+
 	c.BufferMutex.Lock()
 	defer c.BufferMutex.Unlock()
 
-	// Check for existing metric and aggregate if possible
+	// Look for an existing series to aggregate into, and count how many series
+	// already exist for this metric name. Metrics are only aggregated when their
+	// name, type, AND tags all match, so distinct tag sets (e.g. different
+	// request_host values) are kept as separate series rather than being merged
+	// under whichever series was buffered first.
+	sameNameCount := 0
 	for i, existingMetric := range c.MetricsBuffer {
-		if existingMetric["metric"] == prefixedName && existingMetric["type"] == metricType {
+		if existingMetric["metric"] != prefixedName {
+			continue
+		}
+		sameNameCount++
+		existingTags, _ := existingMetric["tags"].(map[string]string)
+		if existingMetric["type"] == metricType && maps.Equal(existingTags, combinedTags) {
 			if metricType == "increment" {
-				if existingValue, ok := existingMetric["value"].(float64); ok {
-					c.MetricsBuffer[i]["value"] = existingValue + value
-				} else {
+				existingValue, ok := existingMetric["value"].(float64)
+				if !ok {
 					return fmt.Errorf("type assertion failed for metric value")
 				}
+				c.MetricsBuffer[i]["value"] = existingValue + value
 				return nil
 			}
 			if metricType == "distribution" {
-				if existingValues, ok := existingMetric["values"].([]float64); ok {
-					c.MetricsBuffer[i]["values"] = append(existingValues, value)
-				} else {
+				existingValues, ok := existingMetric["values"].([]float64)
+				if !ok {
 					return fmt.Errorf("type assertion failed for metric values")
 				}
+				c.MetricsBuffer[i]["values"] = append(existingValues, value)
 				return nil
 			}
 		}
 	}
 
+	// Bound the number of distinct series per metric name. High cardinality tags
+	// (e.g. raw request_host from the egress handler) could otherwise grow the
+	// buffer without limit. The cap is applied per metric name so a flood of
+	// egress observations can never starve the ordinary request/response
+	// metrics. Once a metric reaches its cap, new series for it are dropped until
+	// the next flush clears the buffer; series already buffered continue to
+	// aggregate above.
+	if sameNameCount >= c.MaxSeriesPerMetric {
+		return nil
+	}
+
 	// Create new metric data
 	metricData := map[string]any{
 		"metric": prefixedName,
@@ -177,14 +235,14 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 	}
 	estimatedSize := len(data)
 
-	// Check if adding this metric exceeds the maximum payload size
-	if c.estimatedBufferSize+estimatedSize < MaxPayloadSize {
-		c.MetricsBuffer = append(c.MetricsBuffer, metricData)
-		c.estimatedBufferSize += estimatedSize
-	} else {
-		c.flushBuffer()
-		c.estimatedBufferSize = len(data)
-		c.MetricsBuffer = append(c.MetricsBuffer, metricData)
+	// If adding this series would exceed the maximum payload size, drain the
+	// current buffer first (posted after the mutex is released) so the new series
+	// starts a fresh payload. Only drain when there is an endpoint to post to, so
+	// buffered metrics are not discarded during smoke tests.
+	if c.canSendMetrics() && c.estimatedBufferSize+estimatedSize >= MaxPayloadSize {
+		toSend = c.drainLocked()
 	}
+	c.MetricsBuffer = append(c.MetricsBuffer, metricData)
+	c.estimatedBufferSize += estimatedSize
 	return nil
 }