Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Don't drop instance and job labels in the Prometheus Remote Write Exporter #2979

Merged
merged 1 commit into from
Apr 22, 2021
Merged

Don't drop instance and job labels in the Prometheus Remote Write Exporter #2979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
30 changes: 17 additions & 13 deletions exporter/prometheusremotewriteexporter/exporter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import (
"go.opentelemetry.io/collector/consumer/pdata"
"go.opentelemetry.io/collector/internal"
otlp "go.opentelemetry.io/collector/internal/data/protogen/metrics/v1"
resourcev1 "go.opentelemetry.io/collector/internal/data/protogen/resource/v1"
"go.opentelemetry.io/collector/internal/version"
)

Expand Down Expand Up @@ -107,6 +108,8 @@ func (prwe *PrwExporter) PushMetrics(ctx context.Context, md pdata.Metrics) erro
if resourceMetric == nil {
continue
}

resource := resourceMetric.Resource
// TODO: add resource attributes as labels, probably in next PR
rakyll marked this conversation as resolved.
Show resolved Hide resolved
for _, instrumentationMetrics := range resourceMetric.InstrumentationLibraryMetrics {
if instrumentationMetrics == nil {
Expand All @@ -124,20 +127,21 @@ func (prwe *PrwExporter) PushMetrics(ctx context.Context, md pdata.Metrics) erro
errs = append(errs, consumererror.Permanent(errors.New("invalid temporality and type combination")))
continue
}

// handle individual metric based on type
switch metric.Data.(type) {
case *otlp.Metric_DoubleSum, *otlp.Metric_IntSum, *otlp.Metric_DoubleGauge, *otlp.Metric_IntGauge:
if err := prwe.handleScalarMetric(tsMap, metric); err != nil {
if err := prwe.handleScalarMetric(tsMap, resource, metric); err != nil {
dropped++
errs = append(errs, consumererror.Permanent(err))
}
case *otlp.Metric_DoubleHistogram, *otlp.Metric_IntHistogram:
if err := prwe.handleHistogramMetric(tsMap, metric); err != nil {
if err := prwe.handleHistogramMetric(tsMap, resource, metric); err != nil {
dropped++
errs = append(errs, consumererror.Permanent(err))
}
case *otlp.Metric_DoubleSummary:
if err := prwe.handleSummaryMetric(tsMap, metric); err != nil {
if err := prwe.handleSummaryMetric(tsMap, resource, metric); err != nil {
dropped++
errs = append(errs, consumererror.Permanent(err))
}
Expand Down Expand Up @@ -184,36 +188,36 @@ func validateAndSanitizeExternalLabels(externalLabels map[string]string) (map[st
// handleScalarMetric processes data points in a single OTLP scalar metric by adding the each point as a Sample into
// its corresponding TimeSeries in tsMap.
// tsMap and metric cannot be nil, and metric must have a non-nil descriptor
func (prwe *PrwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error {
func (prwe *PrwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries, resource resourcev1.Resource, metric *otlp.Metric) error {
switch metric.Data.(type) {
// int points
case *otlp.Metric_DoubleGauge:
if metric.GetDoubleGauge().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetDoubleGauge().GetDataPoints() {
addSingleDoubleDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleDoubleDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
case *otlp.Metric_IntGauge:
if metric.GetIntGauge().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetIntGauge().GetDataPoints() {
addSingleIntDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleIntDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
case *otlp.Metric_DoubleSum:
if metric.GetDoubleSum().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetDoubleSum().GetDataPoints() {
addSingleDoubleDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleDoubleDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
case *otlp.Metric_IntSum:
if metric.GetIntSum().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetIntSum().GetDataPoints() {
addSingleIntDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleIntDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
}
return nil
Expand All @@ -222,21 +226,21 @@ func (prwe *PrwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries,
// handleHistogramMetric processes data points in a single OTLP histogram metric by mapping the sum, count and each
// bucket of every data point as a Sample, and adding each Sample to its corresponding TimeSeries.
// tsMap and metric cannot be nil.
func (prwe *PrwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error {
func (prwe *PrwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeries, resource resourcev1.Resource, metric *otlp.Metric) error {
switch metric.Data.(type) {
case *otlp.Metric_IntHistogram:
if metric.GetIntHistogram().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetIntHistogram().GetDataPoints() {
addSingleIntHistogramDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleIntHistogramDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
case *otlp.Metric_DoubleHistogram:
if metric.GetDoubleHistogram().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetDoubleHistogram().GetDataPoints() {
addSingleDoubleHistogramDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleDoubleHistogramDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
}
return nil
Expand All @@ -245,12 +249,12 @@ func (prwe *PrwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeri
// handleSummaryMetric processes data points in a single OTLP summary metric by mapping the sum, count and each
// quantile of every data point as a Sample, and adding each Sample to its corresponding TimeSeries.
// tsMap and metric cannot be nil.
func (prwe *PrwExporter) handleSummaryMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error {
func (prwe *PrwExporter) handleSummaryMetric(tsMap map[string]*prompb.TimeSeries, resource resourcev1.Resource, metric *otlp.Metric) error {
if metric.GetDoubleSummary().GetDataPoints() == nil {
return fmt.Errorf("nil data point. %s is dropped", metric.GetName())
}
for _, pt := range metric.GetDoubleSummary().GetDataPoints() {
addSingleDoubleSummaryDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels)
addSingleDoubleSummaryDataPoint(pt, resource, metric, prwe.namespace, tsMap, prwe.externalLabels)
}
return nil
}
Expand Down
63 changes: 44 additions & 19 deletions exporter/prometheusremotewriteexporter/helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,13 @@ import (
"time"
"unicode"

"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/prompb"

"go.opentelemetry.io/collector/consumer/pdata"
common "go.opentelemetry.io/collector/internal/data/protogen/common/v1"
otlp "go.opentelemetry.io/collector/internal/data/protogen/metrics/v1"
resourcev1 "go.opentelemetry.io/collector/internal/data/protogen/resource/v1"
)

const (
Expand Down Expand Up @@ -125,7 +127,7 @@ func timeSeriesSignature(metric *otlp.Metric, labels *[]prompb.Label) string {
// createLabelSet creates a slice of Cortex Label with OTLP labels and paris of string values.
// Unpaired string value is ignored. String pairs overwrites OTLP labels if collision happens, and the overwrite is
// logged. Resultant label names are sanitized.
func createLabelSet(labels []common.StringKeyValue, externalLabels map[string]string, extras ...string) []prompb.Label {
func createLabelSet(resource resourcev1.Resource, labels []common.StringKeyValue, externalLabels map[string]string, extras ...string) []prompb.Label {
// map ensures no duplicate label name
l := map[string]prompb.Label{}

Expand All @@ -137,6 +139,15 @@ func createLabelSet(labels []common.StringKeyValue, externalLabels map[string]st
}
}

for _, attr := range resource.Attributes {
if isUsefulResourceAttribute(attr) {
l[attr.Key] = prompb.Label{
Name: sanitize(attr.Key),
Value: attr.Value.GetStringValue(), // TODO(jbd): Decide what to do with non-string attributes.
}
}
}

for _, lb := range labels {
l[lb.Key] = prompb.Label{
Name: sanitize(lb.Key),
Expand Down Expand Up @@ -171,6 +182,20 @@ func createLabelSet(labels []common.StringKeyValue, externalLabels map[string]st
return s
}

func isUsefulResourceAttribute(attr common.KeyValue) bool {
// TODO(jbd): Allow users to configure what other resource
// attributes to be included.
// Decide what to do with non-string attributes.
// We should always output "job" and "instance".
switch attr.Key {
case model.InstanceLabel:
return true
case model.JobLabel:
return true
}
return false
}

// getPromMetricName creates a Prometheus metric name by attaching namespace prefix, and _total suffix for Monotonic
// metrics.
func getPromMetricName(metric *otlp.Metric, ns string) string {
Expand Down Expand Up @@ -296,14 +321,14 @@ func getTypeString(metric *otlp.Metric) string {

// addSingleDoubleDataPoint converts the metric value stored in pt to a Prometheus sample, and add the sample
// to its corresponding time series in tsMap
func addSingleDoubleDataPoint(pt *otlp.DoubleDataPoint, metric *otlp.Metric, namespace string,
func addSingleDoubleDataPoint(pt *otlp.DoubleDataPoint, resource resourcev1.Resource, metric *otlp.Metric, namespace string,
tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) {
if pt == nil {
return
}
// create parameters for addSample
name := getPromMetricName(metric, namespace)
labels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, name)
labels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, name)
sample := &prompb.Sample{
Value: pt.Value,
// convert ns to ms
Expand All @@ -314,14 +339,14 @@ func addSingleDoubleDataPoint(pt *otlp.DoubleDataPoint, metric *otlp.Metric, nam

// addSingleIntDataPoint converts the metric value stored in pt to a Prometheus sample, and add the sample
// to its corresponding time series in tsMap
func addSingleIntDataPoint(pt *otlp.IntDataPoint, metric *otlp.Metric, namespace string,
func addSingleIntDataPoint(pt *otlp.IntDataPoint, resource resourcev1.Resource, metric *otlp.Metric, namespace string,
tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) {
if pt == nil {
return
}
// create parameters for addSample
name := getPromMetricName(metric, namespace)
labels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, name)
labels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, name)
sample := &prompb.Sample{
Value: float64(pt.Value),
// convert ns to ms
Expand All @@ -332,7 +357,7 @@ func addSingleIntDataPoint(pt *otlp.IntDataPoint, metric *otlp.Metric, namespace

// addSingleIntHistogramDataPoint converts pt to 2 + min(len(ExplicitBounds), len(BucketCount)) + 1 samples. It
// ignore extra buckets if len(ExplicitBounds) > len(BucketCounts)
func addSingleIntHistogramDataPoint(pt *otlp.IntHistogramDataPoint, metric *otlp.Metric, namespace string,
func addSingleIntHistogramDataPoint(pt *otlp.IntHistogramDataPoint, resource resourcev1.Resource, metric *otlp.Metric, namespace string,
tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) {
if pt == nil {
return
Expand All @@ -346,15 +371,15 @@ func addSingleIntHistogramDataPoint(pt *otlp.IntHistogramDataPoint, metric *otlp
Timestamp: time,
}

sumlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
sumlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
addSample(tsMap, sum, sumlabels, metric)

// treat count as a sample in an individual TimeSeries
count := &prompb.Sample{
Value: float64(pt.GetCount()),
Timestamp: time,
}
countlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
countlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
addSample(tsMap, count, countlabels, metric)

// cumulative count for conversion to cumulative histogram
Expand All @@ -371,7 +396,7 @@ func addSingleIntHistogramDataPoint(pt *otlp.IntHistogramDataPoint, metric *otlp
Timestamp: time,
}
boundStr := strconv.FormatFloat(bound, 'f', -1, 64)
labels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, boundStr)
labels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, boundStr)
addSample(tsMap, bucket, labels, metric)
}
// add le=+Inf bucket
Expand All @@ -380,13 +405,13 @@ func addSingleIntHistogramDataPoint(pt *otlp.IntHistogramDataPoint, metric *otlp
Value: float64(cumulativeCount),
Timestamp: time,
}
infLabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, pInfStr)
infLabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, pInfStr)
addSample(tsMap, infBucket, infLabels, metric)
}

// addSingleDoubleHistogramDataPoint converts pt to 2 + min(len(ExplicitBounds), len(BucketCount)) + 1 samples. It
// ignore extra buckets if len(ExplicitBounds) > len(BucketCounts)
func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, metric *otlp.Metric, namespace string,
func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, resource resourcev1.Resource, metric *otlp.Metric, namespace string,
tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) {
if pt == nil {
return
Expand All @@ -400,15 +425,15 @@ func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, metric
Timestamp: time,
}

sumlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
sumlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
addSample(tsMap, sum, sumlabels, metric)

// treat count as a sample in an individual TimeSeries
count := &prompb.Sample{
Value: float64(pt.GetCount()),
Timestamp: time,
}
countlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
countlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
addSample(tsMap, count, countlabels, metric)

// cumulative count for conversion to cumulative histogram
Expand All @@ -425,7 +450,7 @@ func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, metric
Timestamp: time,
}
boundStr := strconv.FormatFloat(bound, 'f', -1, 64)
labels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, boundStr)
labels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, boundStr)
addSample(tsMap, bucket, labels, metric)
}
// add le=+Inf bucket
Expand All @@ -434,12 +459,12 @@ func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, metric
Value: float64(cumulativeCount),
Timestamp: time,
}
infLabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, pInfStr)
infLabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, pInfStr)
addSample(tsMap, infBucket, infLabels, metric)
}

// addSingleDoubleSummaryDataPoint converts pt to len(QuantileValues) + 2 samples.
func addSingleDoubleSummaryDataPoint(pt *otlp.DoubleSummaryDataPoint, metric *otlp.Metric, namespace string,
func addSingleDoubleSummaryDataPoint(pt *otlp.DoubleSummaryDataPoint, resource resourcev1.Resource, metric *otlp.Metric, namespace string,
tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) {
if pt == nil {
return
Expand All @@ -453,15 +478,15 @@ func addSingleDoubleSummaryDataPoint(pt *otlp.DoubleSummaryDataPoint, metric *ot
Timestamp: time,
}

sumlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
sumlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+sumStr)
addSample(tsMap, sum, sumlabels, metric)

// treat count as a sample in an individual TimeSeries
count := &prompb.Sample{
Value: float64(pt.GetCount()),
Timestamp: time,
}
countlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
countlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName+countStr)
addSample(tsMap, count, countlabels, metric)

// process each percentile/quantile
Expand All @@ -471,7 +496,7 @@ func addSingleDoubleSummaryDataPoint(pt *otlp.DoubleSummaryDataPoint, metric *ot
Timestamp: time,
}
percentileStr := strconv.FormatFloat(qt.GetQuantile(), 'f', -1, 64)
qtlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName, quantileStr, percentileStr)
qtlabels := createLabelSet(resource, pt.GetLabels(), externalLabels, nameStr, baseName, quantileStr, percentileStr)
addSample(tsMap, quantile, qtlabels, metric)
}
}
Expand Down
Loading