fix(push): Add guard clauses to prevent negative counter values (#17056)

Signed-off-by: Jordan Rushing <rushing.jordan@gmail.com>
Co-authored-by: Trevor Whitney <trevorjwhitney@gmail.com>

Author: JordanRushing

pkg/distributor/http.go

@@ -68,6 +68,11 @@ func (d *Distributor) pushHandler(w http.ResponseWriter, r *http.Request, pushRe
 				// Add empty values for retention_hours and policy labels since we don't have
 				// that information for request body too large errors
 				validation.DiscardedBytes.WithLabelValues(validation.RequestBodyTooLarge, tenantID, "", "").Add(float64(r.ContentLength))
+			} else {
+				level.Error(logger).Log(
+					"msg", "negative content length observed",
+					"tenantID", tenantID,
+					"contentLength", r.ContentLength)
 			}
 			errorWriter(w, err.Error(), http.StatusRequestEntityTooLarge, logger)
 			return

pkg/loghttp/push/push.go

@@ -159,8 +159,19 @@ func ParseRequest(logger log.Logger, userID string, maxRecvMsgSize int, r *http.
 	for policyName, retentionToSizeMapping := range pushStats.LogLinesBytes {
 		for retentionPeriod, size := range retentionToSizeMapping {
 			retentionHours := RetentionPeriodToString(retentionPeriod)
-			bytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
-			bytesReceivedStats.Inc(size)
+			// Add guard clause to prevent negative values from being passed to Prometheus counters
+			if size > 0 {
+				bytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
+				bytesReceivedStats.Inc(size)
+			} else {
+				level.Error(logger).Log(
+					"msg", "negative log lines bytes received",
+					"userID", userID,
+					"retentionHours", retentionHours,
+					"isAggregatedMetric", isAggregatedMetric,
+					"policyName", policyName,
+					"size", size)
+			}
 			entriesSize += size
 		}
 	}
@@ -169,10 +180,21 @@ func ParseRequest(logger log.Logger, userID string, maxRecvMsgSize int, r *http.
 		for retentionPeriod, size := range retentionToSizeMapping {
 			retentionHours := RetentionPeriodToString(retentionPeriod)
 
-			structuredMetadataBytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
-			bytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
-			bytesReceivedStats.Inc(size)
-			structuredMetadataBytesReceivedStats.Inc(size)
+			// Add guard clause to prevent negative values from being passed to Prometheus counters
+			if size > 0 {
+				structuredMetadataBytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
+				bytesIngested.WithLabelValues(userID, retentionHours, isAggregatedMetric, policyName).Add(float64(size))
+				bytesReceivedStats.Inc(size)
+				structuredMetadataBytesReceivedStats.Inc(size)
+			} else {
+				level.Error(logger).Log(
+					"msg", "negative structured metadata bytes received",
+					"userID", userID,
+					"retentionHours", retentionHours,
+					"isAggregatedMetric", isAggregatedMetric,
+					"policyName", policyName,
+					"size", size)
+			}
 
 			entriesSize += size
 			structuredMetadataSize += size

pkg/loghttp/push/push_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	kitlog "github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/stretchr/testify/assert"
@@ -23,6 +24,7 @@ import (
 
 	"github.com/grafana/dskit/flagext"
 
+	"github.com/grafana/loki/v3/pkg/logproto"
 	util_log "github.com/grafana/loki/v3/pkg/util/log"
 )
 
@@ -573,6 +575,80 @@ func TestRetentionPeriodToString(t *testing.T) {
 	}
 }
 
+// TestNegativeSizeHandling tests that the code handles negative size values
+// properly without causing a panic when incrementing Prometheus counters.
+func TestNegativeSizeHandling(t *testing.T) {
+	// Reset metrics for accurate testing
+	structuredMetadataBytesIngested.Reset()
+	bytesIngested.Reset()
+	linesIngested.Reset()
+
+	// Create a custom request parser that will generate negative sizes
+	var mockParser RequestParser = func(_ string, _ *http.Request, _ Limits, _ int, _ UsageTracker, _ StreamResolver, _ bool, _ kitlog.Logger) (*logproto.PushRequest, *Stats, error) {
+		// Create a minimal valid request
+		req := &logproto.PushRequest{
+			Streams: []logproto.Stream{
+				{
+					Labels: `{foo="bar"}`,
+					Entries: []logproto.Entry{
+						{
+							Timestamp: time.Now(),
+							Line:      "test line",
+						},
+					},
+				},
+			},
+		}
+
+		// Create stats with negative sizes to test our guard clauses
+		stats := NewPushStats()
+		policy := ""
+		retention := time.Hour
+
+		// Set up negative sizes in both maps
+		stats.LogLinesBytes[policy] = make(map[time.Duration]int64)
+		stats.LogLinesBytes[policy][retention] = -100
+
+		stats.StructuredMetadataBytes[policy] = make(map[time.Duration]int64)
+		stats.StructuredMetadataBytes[policy][retention] = -200
+
+		return req, stats, nil
+	}
+
+	// Create a mock request
+	request := httptest.NewRequest("POST", "/loki/api/v1/push", strings.NewReader("{}"))
+	request.Header.Add("Content-Type", "application/json")
+
+	// Use a mock stream resolver to ensure consistent results
+	streamResolver := newMockStreamResolver("fake", &fakeLimits{})
+
+	// This should not panic with our guard clauses in place
+	_, err := ParseRequest(
+		util_log.Logger,
+		"fake",
+		100<<20,
+		request,
+		&fakeLimits{},
+		mockParser,
+		NewMockTracker(),
+		streamResolver,
+		false,
+	)
+
+	// No error should be returned
+	require.NoError(t, err)
+
+	// Check that the metrics were not incremented for negative values
+	userID := "fake"
+	isAggregatedMetric := "false"
+	policy := ""
+
+	// Verify no counters were incremented since all sizes were negative
+	// This test passes if no panic occurred and the counters remain at 0
+	require.Equal(t, float64(0), testutil.ToFloat64(bytesIngested.WithLabelValues(userID, "1", isAggregatedMetric, policy)))
+	require.Equal(t, float64(0), testutil.ToFloat64(structuredMetadataBytesIngested.WithLabelValues(userID, "1", isAggregatedMetric, policy)))
+}
+
 type fakeLimits struct {
 	enabled         bool
 	labels          []string