feat(ingest)!: bound the decoded telemetry held in flight

cmd/fanout/main.go

@@ -132,6 +132,7 @@ func main() {
 	defer q.Close()
 
 	writer := telemetrystore.NewWriter(repository, cfg.IngestBatchSize)
+	writer.SetInFlightBudget(cfg.IngestMaxInFlightBytes)
 	writerResult := make(chan error, 1)
 	go func() {
 		err := writer.Run(ctx)

fanout.example.yaml

@@ -19,6 +19,9 @@ server:
 ingest:
   default_namespace: default # FANOUT_DEFAULT_NAMESPACE
   batch_size: 50000 # FANOUT_INGEST_BATCH_SIZE
+  # Decoded telemetry held across concurrent requests before new ones are
+  # refused with a retryable status. 0 disables the ceiling.
+  max_in_flight_bytes: 268435456 # FANOUT_INGEST_MAX_IN_FLIGHT_BYTES
 
 storage:
   data_dir: ./data # FANOUT_DATA_DIR

internal/config/config.go

@@ -20,12 +20,18 @@ const maxIngestBatchSize = 50_000
 // Elapsed-time settings use time.Duration with unit-bearing values.
 type Config struct {
 	// Addr binds the browser, API, MCP, OTLP/gRPC and OTLP/HTTP listener.
-	Addr            string        `koanf:"server.addr" env:"FANOUT_ADDR" default:":7520"`
-	DataDir         string        `koanf:"storage.data_dir" env:"FANOUT_DATA_DIR" default:"./data"`
-	IngestBatchSize int           `koanf:"ingest.batch_size" env:"FANOUT_INGEST_BATCH_SIZE" default:"50000"`
-	RollupInterval  time.Duration `koanf:"storage.rollup_interval" env:"FANOUT_ROLLUP_INTERVAL" default:"1m"`
-	MCPEnabled      bool          `koanf:"mcp.enabled" env:"FANOUT_MCP_ENABLED" default:"true"`
-	RetentionDays   int           `koanf:"storage.retention_days" env:"FANOUT_RETENTION_DAYS" default:"30"`
+	Addr            string `koanf:"server.addr" env:"FANOUT_ADDR" default:":7520"`
+	DataDir         string `koanf:"storage.data_dir" env:"FANOUT_DATA_DIR" default:"./data"`
+	IngestBatchSize int    `koanf:"ingest.batch_size" env:"FANOUT_INGEST_BATCH_SIZE" default:"50000"`
+	// IngestMaxInFlightBytes caps the decoded telemetry the process will hold
+	// across concurrent requests. Concurrency is HTTP/2 streams times
+	// connections and each handler holds its batch until the commit is
+	// durable, so without a ceiling a burst is bounded only by memory. Zero is
+	// unbounded, which is the behaviour that predates this setting.
+	IngestMaxInFlightBytes int           `koanf:"ingest.max_in_flight_bytes" env:"FANOUT_INGEST_MAX_IN_FLIGHT_BYTES" default:"268435456"`
+	RollupInterval         time.Duration `koanf:"storage.rollup_interval" env:"FANOUT_ROLLUP_INTERVAL" default:"1m"`
+	MCPEnabled             bool          `koanf:"mcp.enabled" env:"FANOUT_MCP_ENABLED" default:"true"`
+	RetentionDays          int           `koanf:"storage.retention_days" env:"FANOUT_RETENTION_DAYS" default:"30"`
 	// MaintenanceInterval controls Parquet retention and compaction, and
 	// query-cache checkpointing.
 	MaintenanceInterval time.Duration `koanf:"storage.maintenance_interval" env:"FANOUT_MAINTENANCE_INTERVAL" default:"1h"`

internal/ingest/server.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"math"
@@ -26,6 +27,8 @@ import (
 	"github.com/labstack/fanout/internal/config"
 	"github.com/labstack/fanout/internal/telemetry"
 	telemetrystore "github.com/labstack/fanout/internal/telemetry/store"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 type batchSubmitter interface {
@@ -128,11 +131,22 @@ func (s *Server) exportTraces(ctx context.Context, req *collectortrace.ExportTra
 		}
 	}
 	if err := s.submitter.Submit(ctx, batch); err != nil {
-		return nil, err
+		return nil, ingestStatusError(err)
 	}
 	return &collectortrace.ExportTraceServiceResponse{}, nil
 }
 
+// ingestStatusError maps a refusal to a status a sender knows how to act on.
+// RESOURCE_EXHAUSTED is what the OTLP specification tells an exporter to back
+// off and retry on, which is the entire point of shedding rather than dying:
+// the data is not lost, it arrives a moment later.
+func ingestStatusError(err error) error {
+	if errors.Is(err, telemetrystore.ErrIngestOverBudget) {
+		return status.Error(codes.ResourceExhausted, err.Error())
+	}
+	return err
+}
+
 // ---- Logs ingest ----
 
 func (ls *logsService) Export(ctx context.Context, req *collectorlogs.ExportLogsServiceRequest) (*collectorlogs.ExportLogsServiceResponse, error) {
@@ -178,7 +192,7 @@ func (s *Server) exportLogs(ctx context.Context, req *collectorlogs.ExportLogsSe
 		}
 	}
 	if err := s.submitter.Submit(ctx, batch); err != nil {
-		return nil, err
+		return nil, ingestStatusError(err)
 	}
 	return &collectorlogs.ExportLogsServiceResponse{}, nil
 }
@@ -329,7 +343,7 @@ func (s *Server) exportMetrics(ctx context.Context, req *collectormetrics.Export
 		}
 	}
 	if err := s.submitter.Submit(ctx, batch); err != nil {
-		return nil, err
+		return nil, ingestStatusError(err)
 	}
 	return &collectormetrics.ExportMetricsServiceResponse{}, nil
 }

internal/metrics/metrics.go

@@ -61,6 +61,16 @@ var (
 		Help: "Total flush operations",
 	}, []string{"signal"})
 
+	// IngestShedTotal counts requests refused because accepting them would
+	// take the process past the bytes it is willing to hold in flight. A
+	// non-zero value is the system protecting itself, not a fault -- but a
+	// rising one means senders are being asked to retry, which is worth
+	// seeing before it becomes a support question.
+	IngestShedTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "fanout_ingest_shed_total",
+		Help: "Telemetry requests refused because ingest was over its in-flight byte budget",
+	})
+
 	FlushDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Name:    "fanout_flush_duration_seconds",
 		Help:    "Flush duration in seconds",
@@ -288,6 +298,9 @@ func RecordIngest(signal string, count int) {
 }
 
 // RecordFlush records a flush event
+// RecordIngestShed counts one refused telemetry request.
+func RecordIngestShed() { IngestShedTotal.Inc() }
+
 func RecordFlush(signal string, durationSec float64) {
 	FlushTotal.WithLabelValues(signal).Inc()
 	FlushDuration.WithLabelValues(signal).Observe(durationSec)

internal/telemetry/store/admission_test.go

@@ -0,0 +1,75 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+// Everything else in this package makes one commit cost less. None of it makes
+// the process bounded: concurrency is HTTP/2 streams times connections, each
+// handler holding its decoded batch until the commit is durably acknowledged,
+// and nothing counts how much is in flight at once. A large enough burst is
+// still an out-of-memory kill however cheap each batch became.
+//
+// Submit is where that is countable -- both transports funnel through it, and
+// it already blocks until durable acknowledgement, so charging it on entry and
+// releasing it on return measures exactly the bytes the process is holding.
+func TestSubmitRefusesWorkBeyondTheInFlightBudget(t *testing.T) {
+	w := NewWriter(nil, 1000)
+	w.inFlightBudget = 4096
+
+	big := Batch{Spans: []telemetry.Span{{AttributesJSON: strings.Repeat("x", 8192)}}}
+	if got := w.reserve(big); got != nil {
+		t.Fatalf("a lone request must be admitted however large, got %v", got)
+	}
+	// Now over budget on a single oversized reservation.
+	small := Batch{Spans: []telemetry.Span{{AttributesJSON: strings.Repeat("x", 16)}}}
+	if err := w.reserve(small); !errors.Is(err, ErrIngestOverBudget) {
+		t.Errorf("reserve while over budget = %v, want ErrIngestOverBudget", err)
+	}
+
+	w.release(big)
+	if err := w.reserve(small); err != nil {
+		t.Errorf("reserve after the budget was released = %v, want nil", err)
+	}
+	w.release(small)
+	if left := w.inFlightBytes.Load(); left != 0 {
+		t.Errorf("in-flight bytes = %d after every release, want 0", left)
+	}
+}
+
+// A budget of zero means unbounded, so an operator who has not configured one
+// gets exactly the behaviour they had before.
+func TestSubmitBudgetOfZeroAdmitsEverything(t *testing.T) {
+	w := NewWriter(nil, 1000)
+	w.inFlightBudget = 0
+	huge := Batch{Spans: []telemetry.Span{{AttributesJSON: strings.Repeat("x", 1<<20)}}}
+	for range 64 {
+		if err := w.reserve(huge); err != nil {
+			t.Fatalf("unbounded budget refused a request: %v", err)
+		}
+	}
+}
+
+// Refusal has to reach the caller as a retryable condition, not a silent drop:
+// an OTLP exporter that is told the collector is out of capacity backs off and
+// resends, which is the whole point of shedding rather than dying.
+func TestSubmitOverBudgetIsReportedToTheCaller(t *testing.T) {
+	w := NewWriter(nil, 1000)
+	w.inFlightBudget = 1
+	batch := Batch{Spans: []telemetry.Span{{AttributesJSON: strings.Repeat("x", 4096)}}}
+	if err := w.reserve(batch); err != nil {
+		t.Fatalf("first reservation must be admitted: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	err := w.Submit(ctx, batch)
+	if !errors.Is(err, ErrIngestOverBudget) {
+		t.Errorf("Submit over budget = %v, want ErrIngestOverBudget", err)
+	}
+}

internal/telemetry/store/batch_bytes_test.go

@@ -0,0 +1,50 @@
+package store
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+// Rows are a poor proxy for what a batch costs. maxGroupBatchRows caps a group
+// at 50,000 rows, but a row is whatever the sender put in it: 50,000 bare log
+// lines and 50,000 spans carrying kilobytes of attributes each are the same
+// number and differ by orders of magnitude in what the commit has to hold. The
+// row ceiling alone therefore bounds nothing that matters on a fat-row stream.
+func TestBatchBytesMeasuresThePayloadNotTheRowCount(t *testing.T) {
+	thin := Batch{Spans: make([]telemetry.Span, 100)}
+	fat := Batch{Spans: make([]telemetry.Span, 100)}
+	for i := range fat.Spans {
+		fat.Spans[i].AttributesJSON = strings.Repeat("x", 4096)
+	}
+
+	if batchRows(thin) != batchRows(fat) {
+		t.Fatalf("fixture is wrong: row counts must match to make the point")
+	}
+	thinBytes, fatBytes := batchBytes(thin), batchBytes(fat)
+	if fatBytes <= thinBytes {
+		t.Errorf("batchBytes: fat %d, thin %d; the payload must be what is counted", fatBytes, thinBytes)
+	}
+	if fatBytes < 100*4096 {
+		t.Errorf("batchBytes = %d, want at least the %d bytes of attributes", fatBytes, 100*4096)
+	}
+}
+
+// The byte ceiling has to admit a single oversized request, for the same
+// reason the row ceiling does: one request is already one atomic directory,
+// and refusing it would drop data rather than batch it more carefully.
+func TestGroupBatchBytesAdmitsALoneOversizedRequest(t *testing.T) {
+	huge := Batch{Spans: make([]telemetry.Span, 1)}
+	huge.Spans[0].AttributesJSON = strings.Repeat("x", maxGroupBatchBytes*2)
+
+	if got := batchBytes(huge); got <= maxGroupBatchBytes {
+		t.Fatalf("fixture is wrong: batchBytes = %d, want more than the %d ceiling", got, maxGroupBatchBytes)
+	}
+	if !groupBatchFits(0, huge, 0) {
+		t.Error("an empty group must admit any single request, however large")
+	}
+	if groupBatchFits(maxGroupBatchBytes, huge, 1) {
+		t.Error("a non-empty group must not accept a request that takes it past the ceiling")
+	}
+}