logging export drops for otel now (#912)

* logging export drops for otel now

* only logging at warn level once for metric drop

* adding lock to avoid race

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Adding a metric for dropping otel metrics

* Cleaning up misleading log line

---------

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: i3149

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