Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
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
Prev Previous commit
Next Next commit
built success
  • Loading branch information
huyan0 committed Aug 27, 2020
commit ac208279a38db8cfca5d08cdc3cb931820e0a3d5
2 changes: 0 additions & 2 deletions exporter/prometheusremotewriteexporter/DESIGN.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@


# **OpenTelemetry Collector Prometheus Remote Write/Cortex Exporter Design**

Authors: @huyan0, @danielbang907
Expand Down
23 changes: 11 additions & 12 deletions exporter/prometheusremotewriteexporter/exporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import (
"bufio"
"bytes"
"context"
"github.com/pkg/errors"
"errors"
"io"
"net/http"
"net/url"
Expand All @@ -35,7 +35,6 @@ import (
otlp "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/metrics/v1"
)


// prwExporter converts OTLP metrics to Prometheus remote write TimeSeries and sends them to a remote endpoint
type prwExporter struct {
namespace string
Expand All @@ -50,12 +49,12 @@ type prwExporter struct {
func newPrwExporter(namespace string, endpoint string, client *http.Client) (*prwExporter, error) {

if client == nil {
return nil, errors.Errorf("http client cannot be nil")
return nil, errors.New("http client cannot be nil")
}

endpointURL, err := url.ParseRequestURI(endpoint)
if err != nil {
return nil, errors.Errorf("invalid endpoint")
return nil, errors.New("invalid endpoint")
}

return &prwExporter{
Expand Down Expand Up @@ -84,7 +83,7 @@ func (prwe *prwExporter) pushMetrics(ctx context.Context, md pdata.Metrics) (int
defer prwe.wg.Done()
select {
case <-prwe.closeChan:
return pdatautil.MetricCount(md), errors.Errorf("shutdown has been called")
return pdatautil.MetricCount(md), errors.New("shutdown has been called")
default:
tsMap := map[string]*prompb.TimeSeries{}
dropped := 0
Expand Down Expand Up @@ -126,7 +125,7 @@ func (prwe *prwExporter) pushMetrics(ctx context.Context, md pdata.Metrics) (int
}

if dropped != 0 {
return dropped, errors.Errorf(strings.Join(errs, "\n"))
return dropped, errors.New(strings.Join(errs, "\n"))
}

return 0, nil
Expand All @@ -145,7 +144,7 @@ func (prwe *prwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries,
// int points
case otlp.MetricDescriptor_MONOTONIC_INT64, otlp.MetricDescriptor_INT64:
if metric.Int64DataPoints == nil {
return errors.Errorf("nil data point field in metric" + metric.GetMetricDescriptor().Name)
return errors.New("nil data point field in metric" + metric.GetMetricDescriptor().Name)
}

for _, pt := range metric.Int64DataPoints {
Expand All @@ -166,7 +165,7 @@ func (prwe *prwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries,
// double points
case otlp.MetricDescriptor_MONOTONIC_DOUBLE, otlp.MetricDescriptor_DOUBLE:
if metric.DoubleDataPoints == nil {
return errors.Errorf("nil data point field in metric" + metric.GetMetricDescriptor().Name)
return errors.New("nil data point field in metric" + metric.GetMetricDescriptor().Name)
}
for _, pt := range metric.DoubleDataPoints {

Expand All @@ -183,7 +182,7 @@ func (prwe *prwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries,
return nil
}

return errors.Errorf("invalid metric type: wants int or double data points")
return errors.New("invalid metric type: wants int or double data points")
}

// handleHistogramMetric processes data points in a single OTLP histogram metric by mapping the sum, count and each
Expand All @@ -192,7 +191,7 @@ func (prwe *prwExporter) handleScalarMetric(tsMap map[string]*prompb.TimeSeries,
func (prwe *prwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error {

if metric.HistogramDataPoints == nil {
return errors.Errorf("invalid metric type: wants histogram points")
return errors.New("invalid metric type: wants histogram points")
}

for _, pt := range metric.HistogramDataPoints {
Expand Down Expand Up @@ -251,7 +250,7 @@ func (prwe *prwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeri
func (prwe *prwExporter) handleSummaryMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error {

if metric.SummaryDataPoints == nil {
return errors.Errorf("invalid metric type: wants summary points")
return errors.New("invalid metric type: wants summary points")
}

for _, pt := range metric.SummaryDataPoints {
Expand Down Expand Up @@ -342,7 +341,7 @@ func (prwe *prwExporter) export(ctx context.Context, tsMap map[string]*prompb.Ti
if scanner.Scan() {
line = scanner.Text()
}
err = errors.Errorf("server returned HTTP status %s: %s", httpResp.Status, line)
err = errors.New("server returned HTTP status: " + httpResp.Status + ", " + line)
}
return err
}
2 changes: 0 additions & 2 deletions exporter/prometheusremotewriteexporter/exporter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ package prometheusremotewriteexporter

import (
"context"

"io/ioutil"
"net/http"
"net/http/httptest"
Expand All @@ -34,7 +33,6 @@ import (
"go.opentelemetry.io/collector/config/confighttp"
"go.opentelemetry.io/collector/config/configmodels"
"go.opentelemetry.io/collector/consumer/pdata"

"go.opentelemetry.io/collector/consumer/pdatautil"
"go.opentelemetry.io/collector/exporter/exporterhelper"
"go.opentelemetry.io/collector/internal/data"
Expand Down
2 changes: 0 additions & 2 deletions exporter/prometheusremotewriteexporter/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,6 @@ func createDefaultConfig() configmodels.Exporter {
},
Namespace: "",


TimeoutSettings: exporterhelper.CreateDefaultTimeoutSettings(),
RetrySettings: exporterhelper.CreateDefaultRetrySettings(),
QueueSettings: qs,
Expand All @@ -91,7 +90,6 @@ func createDefaultConfig() configmodels.Exporter {
WriteBufferSize: 512 * 1024,
Timeout: exporterhelper.CreateDefaultTimeoutSettings().Timeout,
Headers: map[string]string{},

},
}
}
6 changes: 2 additions & 4 deletions exporter/prometheusremotewriteexporter/helper.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
package prometheusremotewriteexporter

import (
"github.com/pkg/errors"
"errors"
"log"
"sort"
"strings"
Expand All @@ -29,7 +29,6 @@ import (
)

const (

nameStr = "__name__"
sumStr = "_sum"
countStr = "_count"
Expand Down Expand Up @@ -189,12 +188,11 @@ func getPromMetricName(desc *otlp.MetricDescriptor, ns string) string {
return sanitize(b.String())
}


// Simple helper function that takes the <Signature String - *TimeSeries> map
// and creates a WriteRequest from the struct -- can move to the helper.go file
func wrapTimeSeries(tsMap map[string]*prompb.TimeSeries) (*prompb.WriteRequest, error) {
if len(tsMap) == 0 {
return nil, errors.Errorf("invalid TsMap: cannot be empty map")
return nil, errors.New("invalid TsMap: cannot be empty map")
}
TsArray := []prompb.TimeSeries{}
for _, v := range tsMap {
Expand Down
15 changes: 5 additions & 10 deletions exporter/prometheusremotewriteexporter/testutil_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import (
"github.com/prometheus/prometheus/prompb"

"go.opentelemetry.io/collector/internal/data"

commonpb "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/common/v1"
otlp "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/metrics/v1"
)
Expand All @@ -31,12 +30,10 @@ type combination struct {
}

var (

time1 = uint64(time.Now().UnixNano())
time2 = uint64(time.Date(1970, 1, 0, 0, 0, 0, 0, time.UTC).UnixNano())
msTime1 = int64(time1 / uint64(int64(time.Millisecond)/int64(time.Nanosecond)))
msTime2 = int64(time2 / uint64(int64(time.Millisecond)/int64(time.Nanosecond)))
testHeaders = map[string]string{"headerOne": "value1"}
time1 = uint64(time.Now().UnixNano())
time2 = uint64(time.Date(1970, 1, 0, 0, 0, 0, 0, time.UTC).UnixNano())
msTime1 = int64(time1 / uint64(int64(time.Millisecond)/int64(time.Nanosecond)))
msTime2 = int64(time2 / uint64(int64(time.Millisecond)/int64(time.Nanosecond)))

typeInt64 = "INT64"
typeMonotonicInt64 = "MONOTONIC_INT64"
Expand All @@ -59,7 +56,6 @@ var (
dirty1 = "%"
dirty2 = "?"


intVal1 int64 = 1
intVal2 int64 = 2
floatVal1 = 1.0
Expand All @@ -77,7 +73,6 @@ var (
ns1 = "test_ns"
name1 = "valid_single_int_point"


monotonicInt64Comb = 0
monotonicDoubleComb = 1
histogramComb = 2
Expand Down Expand Up @@ -201,7 +196,7 @@ func getSummaryDataPoint(labels []*commonpb.StringKeyValue, ts uint64, sum float
// Prometheus TimeSeries
func getPromLabels(lbs ...string) []prompb.Label {
pbLbs := prompb.Labels{
Labels: []prompb.Label{},
Labels: []prompb.Label{},
}
for i := 0; i < len(lbs); i += 2 {
pbLbs.Labels = append(pbLbs.Labels, getLabel(lbs[i], lbs[i+1]))
Expand Down