From c091bfe201fad8ff7f34122dd95e01d04476909d Mon Sep 17 00:00:00 2001 From: Ian Pye Date: Thu, 25 Jun 2026 08:21:11 -0700 Subject: [PATCH 1/6] logging export drops for otel now --- pkg/formats/otel/otel.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index cbf19b84..20474d07 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -232,6 +232,7 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { case ch <- m: default: f.metrics.ExportDrops.Inc(1) + f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) } } else { ch <- m From df6f222092c7434728aaa31f7957b4512cfd183f Mon Sep 17 00:00:00 2001 From: Ian Pye Date: Thu, 25 Jun 2026 08:30:11 -0700 Subject: [PATCH 2/6] only logging at warn level once for metric drop --- pkg/formats/otel/otel.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index 20474d07..dff0c7e2 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -42,6 +42,7 @@ type OtelFormat struct { trapLog *OtelLogger logTee chan string metrics *OtelMetrics + warnFull map[string]bool } const ( @@ -88,6 +89,7 @@ func NewFormat(ctx context.Context, log logger.Underlying, cfg *ktranslate.OtelF config: cfg, inputs: map[string]chan OtelData{}, logTee: logTee, + warnFull: map[string]bool{}, metrics: &OtelMetrics{ ExportDrops: go_metrics.GetOrRegisterCounter("otel_export_drops", registry), }, @@ -232,7 +234,12 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { case ch <- m: default: f.metrics.ExportDrops.Inc(1) - f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) + if !f.warnFull[m.Name] { + f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) + f.warnFull[m.Name] = true + } else { + f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) + } } } else { ch <- m From 87e4c1e21bba10bf6385fcbb0ec0c2a823c4e3b5 Mon Sep 17 00:00:00 2001 From: Ian Pye Date: Mon, 29 Jun 2026 10:15:00 -0700 Subject: [PATCH 3/6] adding lock to avoid race --- pkg/formats/otel/otel.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index dff0c7e2..375142ea 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -236,7 +236,11 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { f.metrics.ExportDrops.Inc(1) if !f.warnFull[m.Name] { f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) + f.mux.RUnlock() + f.mux.Lock() f.warnFull[m.Name] = true + f.mux.Unlock() + f.mux.RLock() } else { f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) } From 79cebb61a0fef42e7c3f288e45bfb56283bb9408 Mon Sep 17 00:00:00 2001 From: Ian Date: Mon, 29 Jun 2026 13:08:26 -0700 Subject: [PATCH 4/6] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- pkg/formats/otel/otel.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index 375142ea..b2955199 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -235,12 +235,19 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { default: f.metrics.ExportDrops.Inc(1) if !f.warnFull[m.Name] { - f.Warnf("OTEL channel full, dropping sample for metric=%s", m.Name) f.mux.RUnlock() f.mux.Lock() - f.warnFull[m.Name] = true + firstWarn := !f.warnFull[m.Name] + if firstWarn { + f.warnFull[m.Name] = true + } 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) + } } else { f.Debugf("OTEL channel full, dropping sample for metric=%s", m.Name) } From 589d39646979af7f9cb2a46c37edf818b088a4eb Mon Sep 17 00:00:00 2001 From: Ian Pye Date: Mon, 29 Jun 2026 13:25:53 -0700 Subject: [PATCH 5/6] Adding a metric for dropping otel metrics --- pkg/formats/otel/otel.go | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index b2955199..61ab0378 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -43,6 +43,7 @@ type OtelFormat struct { logTee chan string metrics *OtelMetrics warnFull map[string]bool + registry go_metrics.Registry } const ( @@ -69,7 +70,7 @@ func init() { } type OtelMetrics struct { - ExportDrops go_metrics.Counter + ExportDrops map[string]go_metrics.Meter } /* @@ -90,8 +91,9 @@ func NewFormat(ctx context.Context, log logger.Underlying, cfg *ktranslate.OtelF 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{}, }, } @@ -233,13 +235,13 @@ 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() @@ -248,6 +250,7 @@ func (f *OtelFormat) To(msgs []*kt.JCHF, serBuf []byte) (*kt.Output, error) { } 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) } @@ -311,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 From 2e8de1ba5527bfb4a4a5e7504701a44f17155d6d Mon Sep 17 00:00:00 2001 From: Ian Pye Date: Mon, 29 Jun 2026 13:33:04 -0700 Subject: [PATCH 6/6] Cleaning up misleading log long --- pkg/formats/otel/otel.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/formats/otel/otel.go b/pkg/formats/otel/otel.go index 61ab0378..21774fa1 100644 --- a/pkg/formats/otel/otel.go +++ b/pkg/formats/otel/otel.go @@ -180,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 }