[prometheusremotewriteexporter] reduce allocations in createAttributes #35184
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| # Use this changelog template to create an entry for release notes. | ||
|
|
||
| # One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' | ||
| change_type: 'breaking' | ||
|
|
||
| # The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver) | ||
| component: prometheusremotewriteexporter | ||
|
|
||
| # A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). | ||
| note: reduce allocations in createAttributes | ||
|
|
||
| # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. | ||
| issues: [35184] | ||
|
|
||
| # (Optional) One or more lines of additional information to render under the primary note. | ||
| # These lines will be padded with 2 spaces and then inserted directly into the document. | ||
| # Use pipe (|) for multiline entries. | ||
| subtext: | ||
|
|
||
| # If your change doesn't affect end users or the exported elements of any package, | ||
| # you should instead start your pull request title with [chore] or use the "Skip Changelog" label. | ||
| # Optional: The change log or logs in which this entry should be included. | ||
| # e.g. '[user]' or '[user, api]' | ||
| # Include 'user' if the change is relevant to end users. | ||
| # Include 'api' if there is a change to a library API. | ||
| # Default: '[user]' | ||
| change_logs: [] | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,6 +11,7 @@ import ( | |
| "slices" | ||
| "sort" | ||
| "strconv" | ||
| "strings" | ||
| "time" | ||
| "unicode/utf8" | ||
|
|
||
|
|
@@ -96,46 +97,40 @@ var seps = []byte{'\xff'} | |
| // createAttributes creates a slice of Prometheus Labels with OTLP attributes and pairs of string values. | ||
| // Unpaired string values are ignored. String pairs overwrite OTLP labels if collisions happen and | ||
| // if logOnOverwrite is true, the overwrite is logged. Resulting label names are sanitized. | ||
| func (c *PrometheusConverter) createAttributes(resource pcommon.Resource, attributes pcommon.Map, externalLabels map[string]string, | ||
| ignoreAttrs []string, logOnOverwrite bool, extras ...string) []prompb.Label { | ||
| resourceAttrs := resource.Attributes() | ||
| serviceName, haveServiceName := resourceAttrs.Get(conventions.AttributeServiceName) | ||
| instance, haveInstanceID := resourceAttrs.Get(conventions.AttributeServiceInstanceID) | ||
|
|
||
| // map ensures no duplicate label name | ||
| l := c.labelsMap | ||
| clear(l) | ||
|
Comment on lines
+116
to
+117
There was a problem hiding this comment. Could we also clear There was a problem hiding this comment. For the same reason as https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/35184/files/928529a1cf587e8e5b29bd4880f2c36157eb8194#r1829677356 we want to isolate the contents of this map between calls to There was a problem hiding this comment. That makes sense, thank you! Could we add a comment explaining it? |
||
|
|
||
| // store duplicate labels separately in a throwaway map | ||
| // assuming this is the less common case | ||
| collisions := make(map[string][]string) | ||
|
|
||
| // XXX: Should we always drop service namespace/service name/service instance ID from the labels | ||
| // (as they get mapped to other Prometheus labels)? | ||
| attributes.Range(func(key string, value pcommon.Value) bool { | ||
| if !slices.Contains(ignoreAttrs, key) { | ||
| var finalKey = prometheustranslator.NormalizeLabel(key) | ||
| if _, alreadyExists := l[finalKey]; alreadyExists { | ||
| collisions[finalKey] = append(collisions[finalKey], value.AsString()) | ||
| } else { | ||
| l[finalKey] = value.AsString() | ||
| } | ||
| } | ||
| return true | ||
| }) | ||
|
|
||
| for key, values := range collisions { | ||
| values = append(values, l[key]) | ||
| // Ensure attributes are sorted by key for consistent merging of keys which | ||
| // collide when sanitized. | ||
| sort.Strings(values) | ||
| l[key] = strings.Join(values, ";") | ||
| } | ||
|
|
||
| // Map service.name + service.namespace to job | ||
|
|
@@ -175,12 +170,12 @@ func createAttributes(resource pcommon.Resource, attributes pcommon.Map, externa | |
| l[name] = extras[i+1] | ||
| } | ||
|
|
||
| startIndex := len(c.labels) | ||
| for k, v := range l { | ||
| c.labels = append(c.labels, prompb.Label{Name: k, Value: v}) | ||
| } | ||
|
|
||
| return c.labels[startIndex:] | ||
|
There was a problem hiding this comment. Is there any possibility that for k, v := range l {
c.labels = append(c.labels, prompb.Label{Name: k, Value: v})
}
return c.labelsThere was a problem hiding this comment.
Here is how it works: For example, if X1...X4 are labels from series X and Y1...Y3 are labels from series Y, then the backing array of There was a problem hiding this comment. Perfect, thank you for the explanation :) Now what I'm thinking is if we have tests that assure concurrency works. Mostly to make sure we don't break the non-thread-safe promise by accident |
||
| } | ||
|
|
||
| // isValidAggregationTemporality checks whether an OTel metric has a valid | ||
|
|
@@ -200,12 +195,12 @@ func isValidAggregationTemporality(metric pmetric.Metric) bool { | |
| return false | ||
| } | ||
|
|
||
| func (c *PrometheusConverter) addHistogramDataPoints(dataPoints pmetric.HistogramDataPointSlice, | ||
| resource pcommon.Resource, settings Settings, baseName string) { | ||
| for x := 0; x < dataPoints.Len(); x++ { | ||
| pt := dataPoints.At(x) | ||
| timestamp := convertTimeStamp(pt.Timestamp()) | ||
| baseLabels := c.createAttributes(resource, pt.Attributes(), settings.ExternalLabels, nil, false) | ||
|
|
||
| // If the sum is unset, it indicates the _sum metric point should be | ||
| // omitted | ||
|
|
@@ -383,12 +378,12 @@ func maxTimestamp(a, b pcommon.Timestamp) pcommon.Timestamp { | |
| return b | ||
| } | ||
|
|
||
| func (c *PrometheusConverter) addSummaryDataPoints(dataPoints pmetric.SummaryDataPointSlice, resource pcommon.Resource, | ||
| settings Settings, baseName string) { | ||
| for x := 0; x < dataPoints.Len(); x++ { | ||
| pt := dataPoints.At(x) | ||
| timestamp := convertTimeStamp(pt.Timestamp()) | ||
| baseLabels := c.createAttributes(resource, pt.Attributes(), settings.ExternalLabels, nil, false) | ||
|
|
||
| // treat sum as a sample in an individual TimeSeries | ||
| sum := &prompb.Sample{ | ||
|
|
@@ -456,7 +451,7 @@ func createLabels(name string, baseLabels []prompb.Label, extras ...string) []pr | |
|
|
||
| // getOrCreateTimeSeries returns the time series corresponding to the label set if existent, and false. | ||
| // Otherwise it creates a new one and returns that, and true. | ||
| func (c *PrometheusConverter) getOrCreateTimeSeries(lbls []prompb.Label) (*prompb.TimeSeries, bool) { | ||
| h := timeSeriesSignature(lbls) | ||
| ts := c.unique[h] | ||
| if ts != nil { | ||
|
|
@@ -492,7 +487,7 @@ func (c *prometheusConverter) getOrCreateTimeSeries(lbls []prompb.Label) (*promp | |
| // addTimeSeriesIfNeeded adds a corresponding time series if it doesn't already exist. | ||
| // If the time series doesn't already exist, it gets added with startTimestamp for its value and timestamp for its timestamp, | ||
| // both converted to milliseconds. | ||
| func (c *PrometheusConverter) addTimeSeriesIfNeeded(lbls []prompb.Label, startTimestamp pcommon.Timestamp, timestamp pcommon.Timestamp) { | ||
| ts, created := c.getOrCreateTimeSeries(lbls) | ||
| if created { | ||
| ts.Samples = []prompb.Sample{ | ||
|
|
@@ -506,7 +501,7 @@ func (c *prometheusConverter) addTimeSeriesIfNeeded(lbls []prompb.Label, startTi | |
| } | ||
|
|
||
| // addResourceTargetInfo converts the resource to the target info metric. | ||
| func addResourceTargetInfo(resource pcommon.Resource, settings Settings, timestamp pcommon.Timestamp, converter *PrometheusConverter) { | ||
| if settings.DisableTargetInfo || timestamp == 0 { | ||
| return | ||
| } | ||
|
|
@@ -534,7 +529,7 @@ func addResourceTargetInfo(resource pcommon.Resource, settings Settings, timesta | |
| name = settings.Namespace + "_" + name | ||
| } | ||
|
|
||
| labels := converter.createAttributes(resource, attributes, settings.ExternalLabels, identifyingAttrs, false, model.MetricNameLabel, name) | ||
| haveIdentifier := false | ||
| for _, l := range labels { | ||
| if l.Name == model.JobLabel || l.Name == model.InstanceLabel { | ||
|
|
||
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.
That's an enhancement, right? Not a breaking change
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.
@dashpole suggested this should be considered breaking, unless I misread the comment #35184 (comment).
I agree it's more of an enhancement though it is technically changing a public function (FromMetrics).
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.
FromMetricsdoesn't need to change to achieve this, does it? Rather than storing a converter pool in the PRW exporter it could be stored in the translator package and used withinFromMetricssince a single invocation of that function encompasses a complete usage of the converter. Alternately, we can complete the effort that was started to formalize that interface and add aReset()method to it, if required.I think that should probably happen separately from optimization efforts, though. It already seems like there are several different threads being pulled at in this area. Can we work through what this API should look like to enable those desired changes, change the API, then deal with concurrency within the exporter and optimization within the translator independently?