Reimplement prometheus receiver (#572)

Author: fivesheep

cmd/ocagent/main.go

@@ -168,7 +168,7 @@ func runOCAgent() {
 
 	// If the Prometheus receiver is enabled, then run it.
 	if agentConfig.PrometheusReceiverEnabled() {
-		promDoneFn, err := runPrometheusReceiver(viperCfg, commonMetricsSink, asyncErrorChan)
+		promDoneFn, err := runPrometheusReceiver(logger, viperCfg, commonMetricsSink, asyncErrorChan)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -318,8 +318,8 @@ func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, next consumer.
 	return doneFn, nil
 }
 
-func runPrometheusReceiver(v *viper.Viper, next consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
-	pmr, err := prometheusreceiver.New(v.Sub("receivers.prometheus"), next)
+func runPrometheusReceiver(logger *zap.Logger, v *viper.Viper, next consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	pmr, err := prometheusreceiver.New(logger, v.Sub("receivers.prometheus"), next)
 	if err != nil {
 		return nil, err
 	}

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b // indirect
 	github.com/census-instrumentation/opencensus-proto v0.2.0
+	github.com/go-kit/kit v0.8.0
 	github.com/gogo/googleapis v1.2.0 // indirect
 	github.com/golang/protobuf v1.3.1
 	github.com/google/go-cmp v0.3.0
@@ -26,11 +27,11 @@ require (
 	github.com/opentracing/opentracing-go v1.1.0 // indirect
 	github.com/openzipkin/zipkin-go v0.1.6
 	github.com/orijtech/prometheus-go-metrics-exporter v0.0.3-0.20190313163149-b321c5297f60
-	github.com/orijtech/promreceiver v0.0.6
 	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/pkg/errors v0.8.0
 	github.com/prashantv/protectmem v0.0.0-20171002184600-e20412882b3a // indirect
 	github.com/prometheus/client_golang v0.9.2
+	github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
 	github.com/prometheus/procfs v0.0.0-20190117184657-bf6a532e95b1
 	github.com/prometheus/prometheus v0.0.0-20190131111325-62e591f928dd
 	github.com/rs/cors v1.6.0

receiver/prometheusreceiver/README.md

@@ -1,4 +1,4 @@
-// Copyright 2018, OpenCensus Authors
+// Copyright 2019, OpenCensus Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

receiver/prometheusreceiver/doc.go

@@ -0,0 +1,79 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+	"errors"
+	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/scrape"
+	"go.uber.org/zap"
+	"sync"
+)
+
+// test helpers
+
+var zapLogger *zap.Logger
+var testLogger *zap.SugaredLogger
+
+func init() {
+	zl, _ := zap.NewDevelopment()
+	zapLogger = zl
+	testLogger = zapLogger.Sugar()
+}
+
+type mockMetadataCache struct {
+	data map[string]scrape.MetricMetadata
+}
+
+func (m *mockMetadataCache) Metadata(metricName string) (scrape.MetricMetadata, bool) {
+	mm, ok := m.data[metricName]
+	return mm, ok
+}
+
+func (m *mockMetadataCache) SharedLabels() labels.Labels {
+	return labels.FromStrings("__scheme__", "http")
+}
+
+func newMockConsumer() *mockConsumer {
+	return &mockConsumer{
+		Metrics: make(chan *data.MetricsData, 1),
+	}
+}
+
+type mockConsumer struct {
+	Metrics    chan *data.MetricsData
+	consumOnce sync.Once
+}
+
+func (m *mockConsumer) ConsumeMetricsData(ctx context.Context, md data.MetricsData) error {
+	m.consumOnce.Do(func() {
+		m.Metrics <- &md
+	})
+	return nil
+}
+
+type mockMetadataSvc struct {
+	caches map[string]*mockMetadataCache
+}
+
+func (mm *mockMetadataSvc) Get(job, instance string) (MetadataCache, error) {
+	if mc, ok := mm.caches[job+"_"+instance]; ok {
+		return mc, nil
+	}
+
+	return nil, errors.New("cache not found")
+}

receiver/prometheusreceiver/internal/internal_test.go

@@ -0,0 +1,43 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import "go.uber.org/zap"
+import gokitLog "github.com/go-kit/kit/log"
+
+// NewZapToGokitLogAdapter create an adapter for zap.Logger to gokitLog.Logger
+func NewZapToGokitLogAdapter(logger *zap.Logger) gokitLog.Logger {
+	// need to skip two levels in order to get the correct caller
+	// one for this method, the other for gokitLog
+	logger = logger.WithOptions(zap.AddCallerSkip(2))
+	return &zapToGokitLogAdapter{l: logger.Sugar()}
+}
+
+type zapToGokitLogAdapter struct {
+	l *zap.SugaredLogger
+}
+
+func (w *zapToGokitLogAdapter) Log(keyvals ...interface{}) error {
+	if len(keyvals)%2 == 0 {
+		// expecting key value pairs, the number of items need to be even
+		w.l.Infow("", keyvals...)
+	} else {
+		// in case something goes wrong
+		w.l.Info(keyvals...)
+	}
+	return nil
+}
+
+var _ gokitLog.Logger = (*zapToGokitLogAdapter)(nil)

receiver/prometheusreceiver/internal/logger.go

@@ -0,0 +1,66 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"errors"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/scrape"
+)
+
+// MetadataService is an adapter to scrapeManager and provide only the functionality which is needed
+type MetadataService interface {
+	Get(job, instance string) (MetadataCache, error)
+}
+
+// MetadataCache is an adapter to prometheus' scrape.Target  and provide only the functionality which is needed
+type MetadataCache interface {
+	Metadata(metricName string) (scrape.MetricMetadata, bool)
+	SharedLabels() labels.Labels
+}
+
+type mService struct {
+	sm *scrape.Manager
+}
+
+func (t *mService) Get(job, instance string) (MetadataCache, error) {
+	targetGroup, ok := t.sm.TargetsAll()[job]
+	if !ok {
+		return nil, errors.New("unable to find a target group with job=" + job)
+	}
+
+	// from the same targetGroup, instance is not going to be duplicated
+	for _, target := range targetGroup {
+		if target.DiscoveredLabels().Get(model.AddressLabel) == instance {
+			return &mCache{target}, nil
+		}
+	}
+
+	return nil, errors.New("unable to find a target with job=" + job + ", and instance=" + instance)
+}
+
+// adapter to get metadata from scrape.Target
+type mCache struct {
+	t *scrape.Target
+}
+
+func (m *mCache) Metadata(metricName string) (scrape.MetricMetadata, bool) {
+	return m.t.Metadata(metricName)
+}
+
+func (m *mCache) SharedLabels() labels.Labels {
+	return m.t.DiscoveredLabels()
+}