Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 45 additions & 5 deletions pkg/formats/otel/otel.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ type OtelFormat struct {
trapLog *OtelLogger
logTee chan string
metrics *OtelMetrics
warnFull map[string]bool
Comment thread
i3149 marked this conversation as resolved.
registry go_metrics.Registry
}

const (
Expand All @@ -68,7 +70,7 @@ func init() {
}

type OtelMetrics struct {
ExportDrops go_metrics.Counter
ExportDrops map[string]go_metrics.Meter
}

/*
Expand All @@ -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{},
},
}

Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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)
}
Comment thread
Copilot marked this conversation as resolved.
}
} else {
ch <- m
Expand Down Expand Up @@ -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
Expand Down
Loading