Add storage metrics to OTEL, metrics by span service name

cmd/opentelemetry/app/exporter/elasticsearchexporter/exporter.go

@@ -26,7 +26,7 @@ import (
 // new creates Elasticsearch exporter/storage.
 func new(ctx context.Context, config *Config, params component.ExporterCreateParams) (component.TraceExporter, error) {
 	esCfg := config.GetPrimary()
-	w, err := newEsSpanWriter(*esCfg, params.Logger, false)
+	w, err := newEsSpanWriter(*esCfg, params.Logger, false, config.Name())
 	if err != nil {
 		return nil, err
 	}

cmd/opentelemetry/app/exporter/elasticsearchexporter/integration_test.go

@@ -32,13 +32,11 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/internal/esclient"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/internal/reader/es/esdependencyreader"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/internal/reader/es/esspanreader"
-	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/pkg/es/config"
 	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/es"
 	"github.com/jaegertracing/jaeger/plugin/storage/es/spanstore/dbmodel"
 	"github.com/jaegertracing/jaeger/plugin/storage/integration"
-	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
 const (
@@ -49,6 +47,8 @@ const (
 	indexPrefix     = "integration-test"
 	tagKeyDeDotChar = "@"
 	maxSpanAge      = time.Hour * 72
+	numShards       = 5
+	numReplicas     = 0
 )
 
 type IntegrationTest struct {
@@ -57,37 +57,21 @@ type IntegrationTest struct {
 	logger *zap.Logger
 }
 
-type storageWrapper struct {
-	writer *esSpanWriter
-}
-
-var _ spanstore.Writer = (*storageWrapper)(nil)
-
-func (s storageWrapper) WriteSpan(ctx context.Context, span *model.Span) error {
-	// This fails because there is no binary tag type in OTEL and also OTEL span's status code is always created
-	//traces := jaegertranslator.ProtoBatchesToInternalTraces([]*model.Batch{{Process: span.Process, Spans: []*model.Span{span}}})
-	//_, err := s.writer.WriteTraces(context.Background(), traces)
-	converter := dbmodel.FromDomain{}
-	dbSpan := converter.FromDomainEmbedProcess(span)
-	_, err := s.writer.writeSpans(ctx, []*dbmodel.Span{dbSpan})
-	return err
-}
-
 func (s *IntegrationTest) initializeES(allTagsAsFields bool) error {
 	s.logger, _ = testutils.NewLogger()
 
 	s.initSpanstore(allTagsAsFields)
 	s.CleanUp = func() error {
-		return s.esCleanUp()
+		return s.esCleanUp(allTagsAsFields)
 	}
 	s.Refresh = s.esRefresh
-	s.esCleanUp()
+	s.esCleanUp(allTagsAsFields)
 	// TODO: remove this flag after ES support returning spanKind when get operations
 	s.NotSupportSpanKindWithOperation = true
 	return nil
 }
 
-func (s *IntegrationTest) esCleanUp() error {
+func (s *IntegrationTest) esCleanUp(allTagsAsFields bool) error {
 	request, err := http.NewRequest(http.MethodDelete, fmt.Sprintf("%s/*", esURL), strings.NewReader(""))
 	if err != nil {
 		return err
@@ -96,7 +80,12 @@ func (s *IntegrationTest) esCleanUp() error {
 	if err != nil {
 		return err
 	}
-	return response.Body.Close()
+	err = response.Body.Close()
+	if err != nil {
+		return err
+	}
+	// initialize writer, it caches service names
+	return s.initSpanstore(allTagsAsFields)
 }
 
 func (s *IntegrationTest) initSpanstore(allTagsAsFields bool) error {
@@ -107,18 +96,19 @@ func (s *IntegrationTest) initSpanstore(allTagsAsFields bool) error {
 			AllAsFields: allTagsAsFields,
 		},
 	}
-	w, err := newEsSpanWriter(cfg, s.logger, false)
+	w, err := newEsSpanWriter(cfg, s.logger, false, "")
 	if err != nil {
 		return err
 	}
 	esVersion := uint(w.esClientVersion())
-	spanMapping, serviceMapping := es.GetSpanServiceMappings(5, 1, esVersion)
+	spanMapping, serviceMapping := es.GetSpanServiceMappings(numShards, numReplicas, esVersion)
 	err = w.CreateTemplates(context.Background(), spanMapping, serviceMapping)
 	if err != nil {
 		return err
 	}
-	s.SpanWriter = storageWrapper{
-		writer: w,
+	s.SpanWriter = singleSpanWriter{
+		writer:    w,
+		converter: dbmodel.NewFromDomain(allTagsAsFields, []string{}, tagKeyDeDotChar),
 	}
 
 	elasticsearchClient, err := esclient.NewElasticsearchClient(cfg, s.logger)
@@ -133,7 +123,7 @@ func (s *IntegrationTest) initSpanstore(allTagsAsFields bool) error {
 	})
 	s.SpanReader = reader
 
-	depMapping := es.GetDependenciesMappings(1, 0, esVersion)
+	depMapping := es.GetDependenciesMappings(numShards, numReplicas, esVersion)
 	depStore := esdependencyreader.NewDependencyStore(elasticsearchClient, s.logger, indexPrefix)
 	if err := depStore.CreateTemplates(depMapping); err != nil {
 		return nil

cmd/opentelemetry/app/exporter/elasticsearchexporter/spanstore.go

@@ -23,12 +23,15 @@ import (
 	"strings"
 	"time"
 
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/exporter/elasticsearchexporter/esmodeltranslator"
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/exporter/storagemetrics"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/internal/esclient"
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/pkg/cache"
@@ -47,6 +50,7 @@ const (
 // esSpanWriter holds components required for ES span writer
 type esSpanWriter struct {
 	logger           *zap.Logger
+	nameTag          tag.Mutator
 	client           esclient.ElasticsearchClient
 	serviceCache     cache.Cache
 	spanIndexName    indexNameProvider
@@ -56,7 +60,7 @@ type esSpanWriter struct {
 }
 
 // newEsSpanWriter creates new instance of esSpanWriter
-func newEsSpanWriter(params config.Configuration, logger *zap.Logger, archive bool) (*esSpanWriter, error) {
+func newEsSpanWriter(params config.Configuration, logger *zap.Logger, archive bool, name string) (*esSpanWriter, error) {
 	client, err := esclient.NewElasticsearchClient(params, logger)
 	if err != nil {
 		return nil, err
@@ -66,6 +70,8 @@ func newEsSpanWriter(params config.Configuration, logger *zap.Logger, archive bo
 		return nil, err
 	}
 	return &esSpanWriter{
+		logger:           logger,
+		nameTag:          tag.Insert(storagemetrics.TagExporterName(), name),
 		client:           client,
 		spanIndexName:    newIndexNameProvider(spanIndexBaseName, params.IndexPrefix, params.UseReadWriteAliases, archive),
 		serviceIndexName: newIndexNameProvider(serviceIndexBaseName, params.IndexPrefix, params.UseReadWriteAliases, archive),
@@ -106,7 +112,7 @@ func (w *esSpanWriter) WriteTraces(ctx context.Context, traces pdata.Traces) (in
 func (w *esSpanWriter) writeSpans(ctx context.Context, spans []*dbmodel.Span) (int, error) {
 	buffer := &bytes.Buffer{}
 	// mapping for bulk operation to span
-	bulkOperations := make([]bulkItem, len(spans))
+	var bulkOperations []bulkItem
 	var errs []error
 	dropped := 0
 	for _, span := range spans {
@@ -136,14 +142,17 @@ func (w *esSpanWriter) writeSpans(ctx context.Context, spans []*dbmodel.Span) (i
 		errs = append(errs, err)
 		return len(spans), componenterror.CombineErrors(errs)
 	}
-	droppedFromResponse := w.handleResponse(res, bulkOperations)
+	droppedFromResponse := w.handleResponse(ctx, res, bulkOperations)
 	dropped += droppedFromResponse
 	return dropped, componenterror.CombineErrors(errs)
 }
 
-func (w *esSpanWriter) handleResponse(blk *esclient.BulkResponse, operationToSpan []bulkItem) int {
+func (w *esSpanWriter) handleResponse(ctx context.Context, blk *esclient.BulkResponse, operationToSpan []bulkItem) int {
 	numErrors := 0
+	storedSpans := map[string]int64{}
+	notStoredSpans := map[string]int64{}
 	for i, d := range blk.Items {
+		bulkOp := operationToSpan[i]
 		if d.Index.Status > 201 {
 			numErrors++
 			w.logger.Error("Part of the bulk request failed",
@@ -154,15 +163,29 @@ func (w *esSpanWriter) handleResponse(blk *esclient.BulkResponse, operationToSpa
 				zap.String("error.cause.reason", d.Index.Error.Cause.Reason))
 			// TODO return an error or a struct that indicates which spans should be retried
 			// https://github.com/open-telemetry/opentelemetry-collector/issues/990
+			if !bulkOp.isService {
+				notStoredSpans[bulkOp.span.Process.ServiceName] = notStoredSpans[bulkOp.span.Process.ServiceName] + 1
+			}
 		} else {
 			// passed
-			bulkOp := operationToSpan[i]
-			if bulkOp.isService {
+			if !bulkOp.isService {
+				storedSpans[bulkOp.span.Process.ServiceName] = storedSpans[bulkOp.span.Process.ServiceName] + 1
+			} else {
 				cacheKey := hashCode(bulkOp.span.Process.ServiceName, bulkOp.span.OperationName)
 				w.serviceCache.Put(cacheKey, cacheKey)
 			}
 		}
 	}
+	for k, v := range notStoredSpans {
+		ctx, _ := tag.New(ctx,
+			tag.Insert(storagemetrics.TagServiceName(), k), w.nameTag)
+		stats.Record(ctx, storagemetrics.StatSpansNotStoredCount().M(v))
+	}
+	for k, v := range storedSpans {
+		ctx, _ := tag.New(ctx,
+			tag.Insert(storagemetrics.TagServiceName(), k), w.nameTag)
+		stats.Record(ctx, storagemetrics.StatSpansStoredCount().M(v))
+	}
 	return numErrors
 }
 

cmd/opentelemetry/app/exporter/elasticsearchexporter/spanstore_test.go

@@ -0,0 +1,67 @@
+// Copyright (c) 2020 The Jaeger 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 elasticsearchexporter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opencensus.io/stats/view"
+	"go.uber.org/zap"
+
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/exporter/storagemetrics"
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry/app/internal/esclient"
+	"github.com/jaegertracing/jaeger/pkg/es/config"
+	"github.com/jaegertracing/jaeger/plugin/storage/es/spanstore/dbmodel"
+)
+
+func TestMetrics(t *testing.T) {
+	w, err := newEsSpanWriter(config.Configuration{Servers: []string{"localhost:9200"}, Version: 6}, zap.NewNop(), false, "elasticsearch")
+	require.NoError(t, err)
+	response := &esclient.BulkResponse{}
+	response.Items = []esclient.BulkResponseItem{
+		{Index: esclient.BulkIndexResponse{Status: 200}},
+		{Index: esclient.BulkIndexResponse{Status: 500}},
+		{Index: esclient.BulkIndexResponse{Status: 200}},
+		{Index: esclient.BulkIndexResponse{Status: 500}},
+	}
+	blkItms := []bulkItem{
+		{isService: true, span: &dbmodel.Span{}},
+		{isService: true, span: &dbmodel.Span{}},
+		{span: &dbmodel.Span{Process: dbmodel.Process{ServiceName: "foo"}}},
+		{span: &dbmodel.Span{Process: dbmodel.Process{ServiceName: "foo"}}},
+	}
+
+	views := storagemetrics.MetricViews()
+	require.NoError(t, view.Register(views...))
+	defer view.Unregister(views...)
+
+	errs := w.handleResponse(context.Background(), response, blkItms)
+	assert.Equal(t, 2, errs)
+
+	viewData, err := view.RetrieveData(storagemetrics.StatSpansStoredCount().Name())
+	require.NoError(t, err)
+	require.Equal(t, 1, len(viewData))
+	distData := viewData[0].Data.(*view.SumData)
+	assert.Equal(t, float64(1), distData.Value)
+
+	viewData, err = view.RetrieveData(storagemetrics.StatSpansNotStoredCount().Name())
+	require.NoError(t, err)
+	require.Equal(t, 1, len(viewData))
+	distData = viewData[0].Data.(*view.SumData)
+	assert.Equal(t, float64(1), distData.Value)
+}

cmd/opentelemetry/app/exporter/elasticsearchexporter/storagefactory.go

@@ -16,6 +16,7 @@ package elasticsearchexporter
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
@@ -35,18 +36,20 @@ const archiveNamespace = "es-archive"
 
 // StorageFactory implements storage.Factory and storage.ArchiveFactory
 type StorageFactory struct {
-	Options *es.Options
+	options *es.Options
+	name    string
 	logger  *zap.Logger
 }
 
 var _ storage.Factory = (*StorageFactory)(nil)
 var _ storage.ArchiveFactory = (*StorageFactory)(nil)
 
 // NewStorageFactory creates StorageFactory
-func NewStorageFactory(opts *es.Options, logger *zap.Logger) *StorageFactory {
+func NewStorageFactory(opts *es.Options, logger *zap.Logger, name string) *StorageFactory {
 	return &StorageFactory{
-		Options: opts,
+		options: opts,
 		logger:  logger,
+		name:    name,
 	}
 }
 
@@ -58,8 +61,8 @@ func (s *StorageFactory) Initialize(_ metrics.Factory, logger *zap.Logger) error
 
 // CreateSpanWriter creates spanstore.Writer
 func (s *StorageFactory) CreateSpanWriter() (spanstore.Writer, error) {
-	cfg := s.Options.GetPrimary()
-	writer, err := newEsSpanWriter(*cfg, s.logger, false)
+	cfg := s.options.GetPrimary()
+	writer, err := newEsSpanWriter(*cfg, s.logger, false, s.name)
 	if err != nil {
 		return nil, err
 	}
@@ -75,7 +78,7 @@ func (s *StorageFactory) CreateSpanWriter() (spanstore.Writer, error) {
 
 // CreateSpanReader creates spanstore.Reader
 func (s *StorageFactory) CreateSpanReader() (spanstore.Reader, error) {
-	cfg := s.Options.GetPrimary()
+	cfg := s.options.GetPrimary()
 	client, err := esclient.NewElasticsearchClient(*cfg, s.logger)
 	if err != nil {
 		return nil, err
@@ -92,7 +95,7 @@ func (s *StorageFactory) CreateSpanReader() (spanstore.Reader, error) {
 
 // CreateDependencyReader creates dependencystore.Reader
 func (s *StorageFactory) CreateDependencyReader() (dependencystore.Reader, error) {
-	cfg := s.Options.GetPrimary()
+	cfg := s.options.GetPrimary()
 	client, err := esclient.NewElasticsearchClient(*cfg, s.logger)
 	if err != nil {
 		return nil, err
@@ -102,7 +105,7 @@ func (s *StorageFactory) CreateDependencyReader() (dependencystore.Reader, error
 
 // CreateArchiveSpanReader creates archive spanstore.Reader
 func (s *StorageFactory) CreateArchiveSpanReader() (spanstore.Reader, error) {
-	cfg := s.Options.Get(archiveNamespace)
+	cfg := s.options.Get(archiveNamespace)
 	client, err := esclient.NewElasticsearchClient(*cfg, s.logger)
 	if err != nil {
 		return nil, err
@@ -119,8 +122,8 @@ func (s *StorageFactory) CreateArchiveSpanReader() (spanstore.Reader, error) {
 
 // CreateArchiveSpanWriter creates archive spanstore.Writer
 func (s *StorageFactory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
-	cfg := s.Options.Get(archiveNamespace)
-	writer, err := newEsSpanWriter(*cfg, s.logger, true)
+	cfg := s.options.Get(archiveNamespace)
+	writer, err := newEsSpanWriter(*cfg, s.logger, true, fmt.Sprintf("%s/%s", s.name, archiveNamespace))
 	if err != nil {
 		return nil, err
 	}