[awsemfexporter] Support gauge to cloudwatch histogram convertion in EMF exporter #371
+247
−60
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,6 +5,7 @@ package awsemfexporter // import "github.com/open-telemetry/opentelemetry-collec | |
|
|
||
| import ( | ||
| "encoding/json" | ||
| "math" | ||
| "strings" | ||
|
|
||
| "go.opentelemetry.io/collector/pdata/pmetric" | ||
|
|
@@ -41,50 +42,20 @@ func addToGroupedMetric( | |
| return nil | ||
| } | ||
|
|
||
| filteredDps := filterAndCalculateDps(dps, pmd.Name(), metadata, config, calculators) | ||
|
|
||
| if shouldCompactMetrics(pmd, config) { | ||
| compactedMetrics, labels, updatedMetadata := compactGaugeMetrics(filteredDps, metadata, patternReplaceSucceeded, config) | ||
| if compactedMetrics != nil { | ||
| upsertGroupedMetric(groupedMetrics, updatedMetadata, labels, pmd.Name(), compactedMetrics, translateUnit(pmd, descriptor), config.logger) | ||
| } | ||
| } else { | ||
| for i, dp := range filteredDps { | ||
| labels := enrichLabels(dp, config) | ||
| metadata = replacePatternsIfNeeded(metadata, labels, config, patternReplaceSucceeded) | ||
| if dp.timestampMs > 0 { | ||
| metadata.timestampMs = dp.timestampMs | ||
| } | ||
| // Extra params to use when grouping metrics | ||
| if metadata.metricDataType != pmetric.MetricTypeSummary || !config.DetailedMetrics { | ||
| // Summary metrics can be split into separate datapoints when using DetailedMetrics, but we still want to group | ||
|
|
@@ -99,28 +70,10 @@ func addToGroupedMetric( | |
| metadata.metricDataType = pmetric.MetricTypeSum | ||
| } | ||
| } | ||
| upsertGroupedMetric(groupedMetrics, metadata, labels, dp.name, dp.value, translateUnit(pmd, descriptor), config.logger) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
|
|
@@ -224,3 +177,147 @@ func translateUnit(metric pmetric.Metric, descriptor map[string]MetricDescriptor | |
| } | ||
| return unit | ||
| } | ||
|
|
||
| func shouldCompactMetrics(pmd pmetric.Metric, config *Config) bool { | ||
| if pmd.Type() != pmetric.MetricTypeGauge { | ||
| return false | ||
| } | ||
| // Check if the current metric is in the GaugeMetricsToCompact list | ||
| for _, name := range config.GaugeMetricsToCompact { | ||
| if name == pmd.Name() { | ||
| return true | ||
| } | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func filterAndCalculateDps(dps dataPoints, metricName string, metadata cWMetricMetadata, config *Config, calculators *emfCalculators) []dataPoint { | ||
| var result []dataPoint | ||
| for i := 0; i < dps.Len(); i++ { | ||
| if isStale, attrs := dps.IsStaleNaNInf(i); isStale { | ||
| if config != nil && config.logger != nil { | ||
| config.logger.Debug("dropped metric with nan value", | ||
| zap.String("metric.name", metricName), | ||
| zap.Any("metric.attributes", attrs)) | ||
| } | ||
| continue | ||
| } | ||
| calculated, retained := dps.CalculateDeltaDatapoints(i, metadata.instrumentationScopeName, config.DetailedMetrics, calculators) | ||
| if retained { | ||
| result = append(result, calculated...) | ||
| } | ||
| } | ||
| return result | ||
| } | ||
|
|
||
| func enrichLabels(dp dataPoint, config *Config) map[string]string { | ||
| labels := dp.labels | ||
| if metricType, ok := labels["Type"]; ok { | ||
| if (metricType == "Pod" || metricType == "Container") && config.EKSFargateContainerInsightsEnabled { | ||
| addKubernetesWrapper(labels) | ||
| } | ||
| } | ||
| return labels | ||
| } | ||
|
|
||
| func replacePatternsIfNeeded(metadata cWMetricMetadata, labels map[string]string, config *Config, patternReplaceSucceeded bool) cWMetricMetadata { | ||
| if !patternReplaceSucceeded { | ||
| if strings.Contains(metadata.logGroup, "undefined") { | ||
| metadata.logGroup, _ = replacePatterns(config.LogGroupName, labels, config.logger) | ||
| } | ||
| if strings.Contains(metadata.logStream, "undefined") { | ||
| metadata.logStream, _ = replacePatterns(config.LogStreamName, labels, config.logger) | ||
| } | ||
| } | ||
| return metadata | ||
| } | ||
|
|
||
| func upsertGroupedMetric( | ||
| groupedMetrics map[any]*groupedMetric, | ||
| metadata cWMetricMetadata, | ||
| labels map[string]string, | ||
| metricName string, | ||
| metricVal any, | ||
| unit string, | ||
| logger *zap.Logger, | ||
| ) { | ||
| metric := &metricInfo{value: metricVal, unit: unit} | ||
| groupKey := aws.NewKey(metadata.groupedMetricMetadata, labels) | ||
|
|
||
| if _, ok := groupedMetrics[groupKey]; ok { | ||
| // if MetricName already exists in metrics map, print warning log | ||
| if _, ok := groupedMetrics[groupKey].metrics[metricName]; ok { | ||
| logger.Warn("Duplicate metric found", zap.String("Name", metricName), zap.Any("Labels", labels)) | ||
| } else { | ||
| groupedMetrics[groupKey].metrics[metricName] = metric | ||
| } | ||
| } else { | ||
| groupedMetrics[groupKey] = &groupedMetric{ | ||
| labels: labels, | ||
| metrics: map[string]*metricInfo{metricName: metric}, | ||
| metadata: metadata, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // compactGaugeMetrics converts a collection of gauge data points into a compact representation, e.g. values and counts. | ||
| func compactGaugeMetrics( | ||
| dps []dataPoint, | ||
| metadata cWMetricMetadata, | ||
| patternReplaceSucceeded bool, | ||
| config *Config, | ||
| ) (compactedMetrics *cWMetricHistogram, labels map[string]string, updatedMetadata cWMetricMetadata) { | ||
| var values []float64 | ||
| var timestampMs int64 | ||
|
|
||
| // Extract float values from data points and find the latest timestamp | ||
| for _, dp := range dps { | ||
| if dp.timestampMs > timestampMs { | ||
| timestampMs = dp.timestampMs | ||
| } | ||
| if v, ok := dp.value.(float64); ok { | ||
| values = append(values, v) | ||
| labels = enrichLabels(dp, config) | ||
| } | ||
| } | ||
|
|
||
| if len(values) == 0 { | ||
| return nil, nil, metadata | ||
| } | ||
|
|
||
| updatedMetadata = replacePatternsIfNeeded(metadata, labels, config, patternReplaceSucceeded) | ||
| updatedMetadata.metricDataType = pmetric.MetricTypeGauge | ||
| updatedMetadata.timestampMs = timestampMs | ||
|
|
||
| compactedMetrics = &cWMetricHistogram{ | ||
| Values: []float64{}, | ||
| Counts: []float64{}, | ||
| Count: uint64(len(values)), | ||
| Sum: 0, | ||
| Min: math.MaxFloat64, | ||
| Max: -math.MaxFloat64, | ||
| } | ||
|
|
||
| // Calculate sum, min, max and count frequencies for each unique value | ||
| countMap := make(map[float64]float64) | ||
| for _, v := range values { | ||
| compactedMetrics.Sum += v | ||
| if v < compactedMetrics.Min { | ||
| compactedMetrics.Min = v | ||
| } | ||
| if v > compactedMetrics.Max { | ||
| compactedMetrics.Max = v | ||
| } | ||
| countMap[v]++ | ||
|
There was a problem hiding this comment. Correct me if I'm wrong, but if this is the core of the aggregation logic, then this isn't really creating a histogram. It's aggregating to a series of value/count pairs for each unique floating point value. Considering the precision of float64, I dont think we'll get much aggregation out of it. If we wanted to create an actual histogram that aggregates a range of datapoints, we'd need define bucket where each bucket represents a range of values, store all of the incoming datapoints into those buckets, and then convert the buckets to values/counts. There was a problem hiding this comment. You are correct, in fact, we are aggregating to cloudwatch histogram instead of opentelemetry histogram. For cw histogram, it's exactly in the formact of values and counts, and cloudwatch backend would do the calcluation for percentile values e.g. P90. |
||
| } | ||
|
|
||
| // Pre-allocate slices to avoid multiple allocations during append | ||
| compactedMetrics.Values = make([]float64, 0, len(countMap)) | ||
| compactedMetrics.Counts = make([]float64, 0, len(countMap)) | ||
| for val, cnt := range countMap { | ||
| compactedMetrics.Values = append(compactedMetrics.Values, val) | ||
| compactedMetrics.Counts = append(compactedMetrics.Counts, cnt) | ||
| } | ||
|
|
||
| return compactedMetrics, labels, updatedMetadata | ||
| } | ||
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
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.
iiuc it's not really compacting gauges but converting to CWHistogram. The name could be more aligned with what it actually does like
convertGaugesToCWHistogramor somethingThere 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.
Previously I named it as
buildHistogram, but from Rick's perspective, it should be a kind of compaction or deduplication, so renamed all occurrence ofHistogramtoCompact. Pasted the comment from Rick for reference: