feat: implement otlp prom exporter #24158
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,171 @@ | ||||||||||||||||||||
| package telemetry | ||||||||||||||||||||
|
|
||||||||||||||||||||
| import ( | ||||||||||||||||||||
| "context" | ||||||||||||||||||||
| "encoding/base64" | ||||||||||||||||||||
| "fmt" | ||||||||||||||||||||
| "log" | ||||||||||||||||||||
| "math" | ||||||||||||||||||||
| "time" | ||||||||||||||||||||
|
|
||||||||||||||||||||
| "github.com/prometheus/client_golang/prometheus" | ||||||||||||||||||||
| dto "github.com/prometheus/client_model/go" | ||||||||||||||||||||
| "go.opentelemetry.io/otel" | ||||||||||||||||||||
| "go.opentelemetry.io/otel/attribute" | ||||||||||||||||||||
| "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp" | ||||||||||||||||||||
| otmetric "go.opentelemetry.io/otel/metric" | ||||||||||||||||||||
| "go.opentelemetry.io/otel/sdk/metric" | ||||||||||||||||||||
| "go.opentelemetry.io/otel/sdk/resource" | ||||||||||||||||||||
| semconv "go.opentelemetry.io/otel/semconv/v1.21.0" | ||||||||||||||||||||
| ) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| const meterName = "cosmos-sdk-otlp-exporter" | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func StartOtlpExporter(cfg Config) { | ||||||||||||||||||||
|
There was a problem hiding this comment. lets have a comment on what this does and how it works |
||||||||||||||||||||
| ctx := context.Background() | ||||||||||||||||||||
|
|
||||||||||||||||||||
| exporter, err := otlpmetrichttp.New(ctx, | ||||||||||||||||||||
| otlpmetrichttp.WithEndpoint(cfg.OtlpCollectorEndpoint), | ||||||||||||||||||||
| otlpmetrichttp.WithURLPath(cfg.OtlpCollectorMetricsURLPath), | ||||||||||||||||||||
| otlpmetrichttp.WithHeaders(map[string]string{ | ||||||||||||||||||||
| "Authorization": "Basic " + formatBasicAuth(cfg.OtlpUser, cfg.OtlpToken), | ||||||||||||||||||||
| }), | ||||||||||||||||||||
|
Comment on lines
+30
to
+32
There was a problem hiding this comment. This feels brittle in general. Couldn't some collectors expect auth in Http headers, some in grpc request directly, some without authz entirely? |
||||||||||||||||||||
| ) | ||||||||||||||||||||
| if err != nil { | ||||||||||||||||||||
| log.Fatalf("OTLP exporter setup failed: %v", err) | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
|
Comment on lines
+34
to
+37
There was a problem hiding this comment. Avoid
- if err != nil {
- log.Fatalf("OTLP exporter setup failed: %v", err)
- }
+ if err != nil {
+ return fmt.Errorf("OTLP exporter setup failed: %w", err)
+ }…and bubble it up to the caller as noted in the 📝 Committable suggestion
Suggested change
|
||||||||||||||||||||
| res, _ := resource.New(ctx, resource.WithAttributes( | ||||||||||||||||||||
| semconv.ServiceName(cfg.OtlpServiceName), | ||||||||||||||||||||
| )) | ||||||||||||||||||||
|
Comment on lines
+38
to
+40
There was a problem hiding this comment. do we need to check the ignored error here? if not, can we comment why we are able to ignore it? |
||||||||||||||||||||
|
|
||||||||||||||||||||
|
Comment on lines
+38
to
+41
There was a problem hiding this comment. 🛠️ Refactor suggestion Handle and log the error returned by The second return value is currently discarded. If the resource cannot be created, the exporter will run with incomplete metadata. -res, _ := resource.New(ctx, resource.WithAttributes(
- semconv.ServiceName(cfg.OtlpServiceName),
-))
+res, rErr := resource.New(ctx, resource.WithAttributes(
+ semconv.ServiceName(cfg.OtlpServiceName),
+))
+if rErr != nil {
+ return fmt.Errorf("failed to initialise OTLP resource: %w", rErr)
+}📝 Committable suggestion
Suggested change
|
||||||||||||||||||||
| meterProvider := metric.NewMeterProvider( | ||||||||||||||||||||
| metric.WithReader(metric.NewPeriodicReader(exporter, | ||||||||||||||||||||
| metric.WithInterval(cfg.OtlpPushInterval))), | ||||||||||||||||||||
| metric.WithResource(res), | ||||||||||||||||||||
| ) | ||||||||||||||||||||
| otel.SetMeterProvider(meterProvider) | ||||||||||||||||||||
| meter := otel.Meter(meterName) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| gauges := make(map[string]otmetric.Float64Gauge) | ||||||||||||||||||||
| histograms := make(map[string]otmetric.Float64Histogram) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| go func() { | ||||||||||||||||||||
|
There was a problem hiding this comment. how does this behave when shutting a node down? Do we want any kind of graceful shutdown here? |
||||||||||||||||||||
| for { | ||||||||||||||||||||
| if err := scrapePrometheusMetrics(ctx, meter, gauges, histograms); err != nil { | ||||||||||||||||||||
| log.Printf("error scraping metrics: %v", err) | ||||||||||||||||||||
| } | ||||||||||||||||||||
| time.Sleep(cfg.OtlpPushInterval) | ||||||||||||||||||||
|
Comment on lines
+54
to
+58
There was a problem hiding this comment. small nit
Suggested change
|
||||||||||||||||||||
| } | ||||||||||||||||||||
| }() | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func scrapePrometheusMetrics(ctx context.Context, meter otmetric.Meter, gauges map[string]otmetric.Float64Gauge, histograms map[string]otmetric.Float64Histogram) error { | ||||||||||||||||||||
| metricFamilies, err := prometheus.DefaultGatherer.Gather() | ||||||||||||||||||||
| if err != nil { | ||||||||||||||||||||
| log.Printf("failed to gather prometheus metrics: %v", err) | ||||||||||||||||||||
| return err | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| for _, mf := range metricFamilies { | ||||||||||||||||||||
| name := mf.GetName() | ||||||||||||||||||||
| for _, m := range mf.Metric { | ||||||||||||||||||||
| switch mf.GetType() { | ||||||||||||||||||||
| case dto.MetricType_GAUGE: | ||||||||||||||||||||
| recordGauge(ctx, meter, gauges, name, mf.GetHelp(), m.Gauge.GetValue(), nil) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| case dto.MetricType_COUNTER: | ||||||||||||||||||||
| recordGauge(ctx, meter, gauges, name, mf.GetHelp(), m.Counter.GetValue(), nil) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| case dto.MetricType_HISTOGRAM: | ||||||||||||||||||||
| recordHistogram(ctx, meter, histograms, name, mf.GetHelp(), m.Histogram) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| case dto.MetricType_SUMMARY: | ||||||||||||||||||||
| recordSummary(ctx, meter, gauges, name, mf.GetHelp(), m.Summary) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| default: | ||||||||||||||||||||
| continue | ||||||||||||||||||||
| } | ||||||||||||||||||||
| } | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| return nil | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func recordGauge(ctx context.Context, meter otmetric.Meter, gauges map[string]otmetric.Float64Gauge, name, help string, val float64, attrs []attribute.KeyValue) { | ||||||||||||||||||||
| g, ok := gauges[name] | ||||||||||||||||||||
| if !ok { | ||||||||||||||||||||
| var err error | ||||||||||||||||||||
| g, err = meter.Float64Gauge(name, otmetric.WithDescription(help)) | ||||||||||||||||||||
| if err != nil { | ||||||||||||||||||||
| log.Printf("failed to create gauge %q: %v", name, err) | ||||||||||||||||||||
| return | ||||||||||||||||||||
| } | ||||||||||||||||||||
| gauges[name] = g | ||||||||||||||||||||
| } | ||||||||||||||||||||
| g.Record(ctx, val, otmetric.WithAttributes(attrs...)) | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func recordHistogram(ctx context.Context, meter otmetric.Meter, histograms map[string]otmetric.Float64Histogram, name, help string, h *dto.Histogram) { | ||||||||||||||||||||
| boundaries := make([]float64, 0, len(h.Bucket)-1) // excluding +Inf | ||||||||||||||||||||
| bucketCounts := make([]uint64, 0, len(h.Bucket)) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| for _, bucket := range h.Bucket { | ||||||||||||||||||||
| if math.IsInf(bucket.GetUpperBound(), +1) { | ||||||||||||||||||||
| continue // Skip +Inf bucket boundary explicitly | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
Comment on lines
+114
to
+116
There was a problem hiding this comment. why do we skip this? i know it says +Inf boundary, but as someone who is unfamiliar with OTL, i am not sure what the significance is |
||||||||||||||||||||
| boundaries = append(boundaries, bucket.GetUpperBound()) | ||||||||||||||||||||
| bucketCounts = append(bucketCounts, bucket.GetCumulativeCount()) | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| hist, ok := histograms[name] | ||||||||||||||||||||
| if !ok { | ||||||||||||||||||||
| var err error | ||||||||||||||||||||
| hist, err = meter.Float64Histogram( | ||||||||||||||||||||
| name, | ||||||||||||||||||||
| otmetric.WithDescription(help), | ||||||||||||||||||||
| otmetric.WithExplicitBucketBoundaries(boundaries...), | ||||||||||||||||||||
| ) | ||||||||||||||||||||
| if err != nil { | ||||||||||||||||||||
| log.Printf("failed to create histogram %s: %v", name, err) | ||||||||||||||||||||
| return | ||||||||||||||||||||
| } | ||||||||||||||||||||
| histograms[name] = hist | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| prevCount := uint64(0) | ||||||||||||||||||||
| for i, count := range bucketCounts { | ||||||||||||||||||||
| countInBucket := count - prevCount | ||||||||||||||||||||
| prevCount = count | ||||||||||||||||||||
|
|
||||||||||||||||||||
| // Explicitly record the mid-point of the bucket as approximation: | ||||||||||||||||||||
| var value float64 | ||||||||||||||||||||
| if i == 0 { | ||||||||||||||||||||
| value = boundaries[0] / 2.0 | ||||||||||||||||||||
Check noticeCode scanning / CodeQL Floating point arithmetic Note |
||||||||||||||||||||
| } else { | ||||||||||||||||||||
| value = (boundaries[i-1] + boundaries[i]) / 2.0 | ||||||||||||||||||||
Check noticeCode scanning / CodeQL Floating point arithmetic Note Check noticeCode scanning / CodeQL Floating point arithmetic Note |
||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| // Record `countInBucket` number of observations explicitly (approximation): | ||||||||||||||||||||
| for j := uint64(0); j < countInBucket; j++ { | ||||||||||||||||||||
| hist.Record(ctx, value) | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
Comment on lines
+150
to
+152
There was a problem hiding this comment.
Suggested change
|
||||||||||||||||||||
| } | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func recordSummary(ctx context.Context, meter otmetric.Meter, gauges map[string]otmetric.Float64Gauge, name, help string, s *dto.Summary) { | ||||||||||||||||||||
| recordGauge(ctx, meter, gauges, name+"_sum", help+" (summary sum)", s.GetSampleSum(), nil) | ||||||||||||||||||||
| recordGauge(ctx, meter, gauges, name+"_count", help+" (summary count)", float64(s.GetSampleCount()), nil) | ||||||||||||||||||||
|
|
||||||||||||||||||||
| for _, q := range s.Quantile { | ||||||||||||||||||||
| attrs := []attribute.KeyValue{ | ||||||||||||||||||||
| attribute.String("quantile", fmt.Sprintf("%v", q.GetQuantile())), | ||||||||||||||||||||
| } | ||||||||||||||||||||
| recordGauge(ctx, meter, gauges, name, help+" (summary quantile)", q.GetValue(), attrs) | ||||||||||||||||||||
| } | ||||||||||||||||||||
| } | ||||||||||||||||||||
|
|
||||||||||||||||||||
| func formatBasicAuth(username, token string) string { | ||||||||||||||||||||
| auth := username + ":" + token | ||||||||||||||||||||
| return base64.StdEncoding.EncodeToString([]byte(auth)) | ||||||||||||||||||||
| } | ||||||||||||||||||||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🛠️ Refactor suggestion
Exporter is started without lifecycle management or error propagation
StartOtlpExporter(1) blocks fatal‑logging on failure and (2) launches a goroutine that never stops.Starting it here means:
graceDuration, tests, etc.).telemetry.Newregisters the Prom sink.Recommend returning a cancel/cleanup func and wiring it into the existing
errgroup, then starting aftertelemetry.New: