diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index cbf19b84..21774fa1 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -42,6 +42,8 @@ type OtelFormat struct { trapLog *OtelLogger logTee chan string metrics *OtelMetrics + warnFull map[string]bool + registry go_metrics.Registry } const ( @@ -68,7 +70,7 @@ func init() { } type OtelMetrics struct { - ExportDrops go_metrics.Counter + ExportDrops map[string]go_metrics.Meter } /* @@ -88,8 +90,10 @@ func NewFormat(ctx context.Context, log logger.Underlying, cfg *ktranslate.OtelF config: cfg, inputs: map[string]chan OtelData{}, logTee: logTee, + warnFull: map[string]bool{}, + registry: registry, metrics: &OtelMetrics{ - ExportDrops: go_metrics.GetOrRegisterCounter("otel_export_drops", registry), + ExportDrops: map[string]go_metrics.Meter{}, }, } @@ -176,7 +180,7 @@ func NewFormat(ctx context.Context, log logger.Underlying, cfg *ktranslate.OtelF jf.trapLog = ol otelm = otel.Meter("ktranslate") - jf.Infof("Running exporting via %s to %s. Blocking: %v", cfg.Protocol, cfg.Endpoint, cfg.NoBlockExport) + jf.Infof("Running exporting via %s to %s. NoBlockExport: %v", cfg.Protocol, cfg.Endpoint, cfg.NoBlockExport) return jf, nil } @@ -231,7 +235,25 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { select { case ch <- m: default: - f.metrics.ExportDrops.Inc(1) + if !f.warnFull[m.Name] { + f.mux.RUnlock() + f.mux.Lock() + firstWarn := !f.warnFull[m.Name] + if firstWarn { + f.warnFull[m.Name] = true + f.metrics.ExportDrops[m.Name] = go_metrics.GetOrRegisterMeter(fmt.Sprintf("otel_export_drops^name=%s^force=true", m.Name), f.registry) + } + f.mux.Unlock() + f.mux.RLock() + if firstWarn { + f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) + } else { + f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) + } + f.metrics.ExportDrops[m.Name].Mark(1) + } else { + f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) + } } } else { ch <- m @@ -292,7 +314,25 @@ func (f *OtelFormat) Rollup(rolls []rollup.Rollup) (*kt.Output, error) { select { case ch <- m: default: - f.metrics.ExportDrops.Inc(1) + if !f.warnFull[m.Name] { + f.mux.RUnlock() + f.mux.Lock() + firstWarn := !f.warnFull[m.Name] + if firstWarn { + f.warnFull[m.Name] = true + f.metrics.ExportDrops[m.Name] = go_metrics.GetOrRegisterMeter(fmt.Sprintf("otel_export_drops^name=%s^force=true", m.Name), f.registry) + } + f.mux.Unlock() + f.mux.RLock() + if firstWarn { + f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) + } else { + f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) + } + f.metrics.ExportDrops[m.Name].Mark(1) + } else { + f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) + } } } else { ch <- m