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.