From ba956938af42bf7347676ff71d39b7d58c86cd14 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Sun, 20 Sep 2026 15:43:07 -0700 Subject: [PATCH] feat(ingest)!: bound the decoded telemetry held in flight Every other change in this series makes one commit cost less. None of them makes the process bounded. Ingest concurrency is HTTP/2 streams times connections -- net/http allows 250 streams per connection by default and nothing here lowers it -- and each handler holds its decoded batch and its proto until the commit is durably acknowledged. Nothing counted how much was in flight at once, so a large enough burst was still an out-of-memory kill however cheap an individual batch became. Submit now charges a request against a byte ceiling on entry and releases it on return. That is the right place: both transports funnel through it, and it already blocks until durable acknowledgement, so the ledger measures exactly what the process is holding rather than what arrived on the wire. A refusal returns RESOURCE_EXHAUSTED over gRPC and the existing 503 over HTTP. Both are what the OTLP specification tells an exporter to back off and retry on, so a shed request is delayed rather than lost. The fanout_ingest_shed_total counter makes that visible before it becomes a support question. An empty ledger admits any single request however large: one request is already one atomic batch, and refusing the only thing in flight would shed load the process is not short of. The ceiling exists to stop requests accumulating, not to reject a large one. Also bounds a group-commit batch by payload. maxGroupBatchRows caps rows, but a row is whatever the sender put in it: 50,000 bare log lines and 50,000 spans carrying kilobytes of attributes are the same count and differ by orders of magnitude in what the commit holds. Breaking: a deployment under sustained overload now receives retryable refusals where it previously received acceptance and, eventually, a killed process. ingest.max_in_flight_bytes defaults to 256 MiB; zero restores the old unbounded behaviour exactly. --- cmd/fanout/main.go | 1 + fanout.example.yaml | 3 + internal/config/config.go | 18 ++- internal/ingest/server.go | 20 +++- internal/metrics/metrics.go | 13 ++ internal/telemetry/store/admission_test.go | 75 ++++++++++++ internal/telemetry/store/batch_bytes_test.go | 50 ++++++++ internal/telemetry/store/writer.go | 113 ++++++++++++++++-- .../docs/reference/settings/ingest.mdx | 7 ++ 9 files changed, 283 insertions(+), 17 deletions(-) create mode 100644 internal/telemetry/store/admission_test.go create mode 100644 internal/telemetry/store/batch_bytes_test.go diff --git a/cmd/fanout/main.go b/cmd/fanout/main.go index 57b91fa7..6642c740 100644 --- a/cmd/fanout/main.go +++ b/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) diff --git a/fanout.example.yaml b/fanout.example.yaml index 182f7879..7b6a8053 100644 --- a/fanout.example.yaml +++ b/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 diff --git a/internal/config/config.go b/internal/config/config.go index 43f3beae..8271aaca 100644 --- a/internal/config/config.go +++ b/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"` diff --git a/internal/ingest/server.go b/internal/ingest/server.go index cc34276b..488b2d1d 100644 --- a/internal/ingest/server.go +++ b/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 } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index a855c9f2..ee75a3cc 100644 --- a/internal/metrics/metrics.go +++ b/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) diff --git a/internal/telemetry/store/admission_test.go b/internal/telemetry/store/admission_test.go new file mode 100644 index 00000000..3e9a1a93 --- /dev/null +++ b/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) + } +} diff --git a/internal/telemetry/store/batch_bytes_test.go b/internal/telemetry/store/batch_bytes_test.go new file mode 100644 index 00000000..3cd437a3 --- /dev/null +++ b/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") + } +} diff --git a/internal/telemetry/store/writer.go b/internal/telemetry/store/writer.go index 6c84252d..f3da1c98 100644 --- a/internal/telemetry/store/writer.go +++ b/internal/telemetry/store/writer.go @@ -7,6 +7,7 @@ import ( "log/slog" "runtime" "sync" + "sync/atomic" "time" "github.com/google/uuid" @@ -29,13 +30,15 @@ type batchCommitter interface { } type Writer struct { - repository batchCommitter - batchSize int - retryDelay func(int) time.Duration - groupWindow time.Duration - shutdownGrace time.Duration - done chan struct{} - submissions chan submission + inFlightBudget int + inFlightBytes atomic.Int64 + repository batchCommitter + batchSize int + retryDelay func(int) time.Duration + groupWindow time.Duration + shutdownGrace time.Duration + done chan struct{} + submissions chan submission } type submission struct { @@ -48,6 +51,43 @@ type commitJob struct { acks []chan error } +// ErrIngestOverBudget is returned when accepting a request would take the +// process past the bytes it is willing to hold in flight. Callers map it to a +// retryable status so a sender backs off and resends: shedding load is the +// point, and a dropped request would defeat it. +var ErrIngestOverBudget = errors.New("telemetry ingest is over its in-flight byte budget") + +// reserve charges a batch against the in-flight budget, or refuses it. +// +// An empty ledger admits anything, however large. One request is one atomic +// batch and refusing the only thing in flight would shed load the process is +// not actually short of -- the budget exists to stop requests accumulating, +// not to reject a single large one. A budget of zero is unbounded, so an +// unconfigured deployment behaves exactly as it did before. +func (w *Writer) reserve(batch Batch) error { + if w.inFlightBudget <= 0 { + return nil + } + cost := int64(batchBytes(batch)) + for { + current := w.inFlightBytes.Load() + if current > 0 && current > int64(w.inFlightBudget)-cost { + metrics.RecordIngestShed() + return ErrIngestOverBudget + } + if w.inFlightBytes.CompareAndSwap(current, current+cost) { + return nil + } + } +} + +func (w *Writer) release(batch Batch) { + if w.inFlightBudget <= 0 { + return + } + w.inFlightBytes.Add(-int64(batchBytes(batch))) +} + func NewWriter(repository *Repository, batchSize int) *Writer { return &Writer{ repository: repository, batchSize: batchSize, groupWindow: groupAdmissionWindow, @@ -55,6 +95,10 @@ func NewWriter(repository *Repository, batchSize int) *Writer { } } +// SetInFlightBudget caps the decoded telemetry the writer will hold across +// concurrent requests. Zero is unbounded. +func (w *Writer) SetInFlightBudget(bytes int) { w.inFlightBudget = bytes } + func (w *Writer) Wait() { <-w.done } // Submit returns after every row in the request belongs to a durably published @@ -65,6 +109,13 @@ func (w *Writer) Submit(ctx context.Context, batch Batch) error { if batchRows(batch) == 0 { return nil } + // Charged here rather than at the transport because both transports funnel + // through this call, and it already blocks until durable acknowledgement -- + // so the ledger measures exactly the bytes the process is holding. + if err := w.reserve(batch); err != nil { + return err + } + defer w.release(batch) request := submission{batch: batch, ack: make(chan error, 1)} select { case w.submissions <- request: @@ -188,16 +239,20 @@ drained: batch := Batch{ID: uuid.NewString()} group := make([]submission, 0, len(requests)) - rows := 0 + rows, groupBytes := 0, 0 for len(requests) > 0 { next := requests[0] nextRows := batchRows(next.batch) if len(group) > 0 && rows+nextRows > limit { break } + if !groupBatchFits(groupBytes, next.batch, len(group)) { + break + } requests = requests[1:] group = append(group, next) rows += nextRows + groupBytes += batchBytes(next.batch) batch.Spans = append(batch.Spans, next.batch.Spans...) batch.Logs = append(batch.Logs, next.batch.Logs...) batch.Metrics = append(batch.Metrics, next.batch.Metrics...) @@ -293,6 +348,48 @@ func (w *Writer) commitJob(ctx context.Context, job commitJob) error { return nil } +// maxGroupBatchBytes bounds a group-commit batch by payload rather than row +// count. maxGroupBatchRows caps rows, but a row is whatever the sender put in +// it: 50,000 bare log lines and 50,000 spans carrying kilobytes of attributes +// are the same count and differ by orders of magnitude in what the commit +// holds. The row ceiling alone therefore bounds nothing on a fat-row stream, +// which is precisely the stream that gets the process killed. +const maxGroupBatchBytes = 64 << 20 + +// batchBytes approximates what a batch will cost to hold. It counts the JSON +// columns and the free-form strings, which carry effectively all of the +// variable size; the fixed-width fields are already bounded by the row count. +func batchBytes(batch Batch) int { + total := 0 + for i := range batch.Spans { + span := &batch.Spans[i] + total += len(span.ResourceJSON) + len(span.AttributesJSON) + len(span.EventsJSON) + len(span.LinksJSON) + + len(span.Name) + len(span.StatusMsg) + len(span.TraceID) + len(span.SpanID) + } + for i := range batch.Logs { + log := &batch.Logs[i] + total += len(log.ResourceJSON) + len(log.AttributesJSON) + len(log.Body) + len(log.BodyTemplate) + } + for i := range batch.Metrics { + metric := &batch.Metrics[i] + total += len(metric.ResourceJSON) + len(metric.AttributesJSON) + len(metric.ExemplarsJSON) + + len(metric.HistBoundsJSON) + len(metric.HistCountsJSON) + len(metric.Name) + len(metric.Description) + } + return total +} + +// groupBatchFits reports whether next can join a group already holding +// groupBytes. An empty group admits anything: one request is already one +// atomic directory, and refusing it would drop data rather than batch it more +// carefully -- the same reason the row ceiling lets a lone oversized request +// through. +func groupBatchFits(groupBytes int, next Batch, groupLen int) bool { + if groupLen == 0 { + return true + } + return groupBytes <= maxGroupBatchBytes-batchBytes(next) +} + func batchRows(batch Batch) int { return len(batch.Spans) + len(batch.Logs) + len(batch.Metrics) } diff --git a/site/src/content/docs/reference/settings/ingest.mdx b/site/src/content/docs/reference/settings/ingest.mdx index a6e7b7fc..39cc5e1b 100644 --- a/site/src/content/docs/reference/settings/ingest.mdx +++ b/site/src/content/docs/reference/settings/ingest.mdx @@ -20,3 +20,10 @@ as a refusal to start rather than as a default nobody chose. |---|---|---|---| | `ingest.batch_size` | `FANOUT_INGEST_BATCH_SIZE` | integer | `50000` | | `ingest.default_namespace` | `FANOUT_DEFAULT_NAMESPACE` | string | `default` | +| `ingest.max_in_flight_bytes` | `FANOUT_INGEST_MAX_IN_FLIGHT_BYTES` | integer | `268435456` | + +## Notes + +### `ingest.max_in_flight_bytes` + +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.