[Fix] Restore ES metrics

internal/storage/elasticsearch/config/config.go

@@ -235,9 +235,13 @@
 		After(func(id int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
 			start, ok := m.Load(id)
 			if !ok {
-				return
+				m.Delete(id)
+			} else {
+				start = time.Now()
 			}
-			m.Delete(id)
+
+			duration := time.Since(start.(time.Time))
+			var successCount, failureCount int
 
 			// log individual errors, note that err might be false and these errors still present
 			if response != nil && response.Errors {
@@ -246,11 +250,27 @@
 						if val.Error != nil {
 							logger.Error("Elasticsearch part of bulk request failed", zap.String("map-key", key),
 								zap.Reflect("response", val))
+						} else {
+							successCount++
 						}
 					}
 				}
+			} else if response != nil {
+				// All succeeded
+				successCount = len(response.Items)
 			}
 
+			sm.Inserts.Inc(int64(successCount))
+			sm.Errors.Inc(int64(failureCount))
+
+			if err != nil || failureCount > 0 {
+				sm.LatencyErr.Record(duration)
+			} else {
+				sm.LatencyOk.Record(duration)
+			}
+
+			sm.Emit(err, duration)
+
 			sm.Emit(err, time.Since(start.(time.Time)))
 			if err != nil {
 				var failed int

internal/storage/v1/elasticsearch/spanstore/writer.go

@@ -16,6 +16,7 @@ import (
 	es "github.com/jaegertracing/jaeger/internal/storage/elasticsearch"
 	cfg "github.com/jaegertracing/jaeger/internal/storage/elasticsearch/config"
 	"github.com/jaegertracing/jaeger/internal/storage/elasticsearch/dbmodel"
+	"github.com/jaegertracing/jaeger/internal/storage/v1/api/spanstore/spanstoremetrics"
 )
 
 const (
@@ -25,13 +26,18 @@ const (
 	indexCacheTTLDefault   = 48 * time.Hour
 )
 
+type spanWriterMetrics struct {
+	indexCreate *spanstoremetrics.WriteMetrics
+}
+
 type serviceWriter func(string, *dbmodel.Span)
 
 // SpanWriter is a wrapper around elastic.Client
 type SpanWriter struct {
 	client func() es.Client
 	logger *zap.Logger
 	// indexCache       cache.Cache
+	writerMetrics    spanWriterMetrics
 	serviceWriter    serviceWriter
 	spanServiceIndex spanAndServiceIndexFn
 }
@@ -41,7 +47,7 @@ type CoreSpanWriter interface {
 	// CreateTemplates creates index templates.
 	CreateTemplates(spanTemplate, serviceTemplate string, indexPrefix cfg.IndexPrefix) error
 	// WriteSpan writes a span and its corresponding service:operation in ElasticSearch
-	WriteSpan(spanStartTime time.Time, span *dbmodel.Span)
+	WriteSpan(spanStartTime time.Time, span *dbmodel.Span) error
 	// Close closes CoreSpanWriter
 	Close() error
 }
@@ -80,8 +86,11 @@ func NewSpanWriter(p SpanWriterParams) *SpanWriter {
 
 	serviceOperationStorage := NewServiceOperationStorage(p.Client, p.Logger, serviceCacheTTL)
 	return &SpanWriter{
-		client:           p.Client,
-		logger:           p.Logger,
+		client: p.Client,
+		logger: p.Logger,
+		writerMetrics: spanWriterMetrics{
+			indexCreate: spanstoremetrics.NewWriter(p.MetricsFactory, "index_create"),
+		},
 		serviceWriter:    serviceOperationStorage.Write,
 		spanServiceIndex: getSpanAndServiceIndexFn(p, writeAliasSuffix),
 	}
@@ -119,13 +128,17 @@ func getSpanAndServiceIndexFn(p SpanWriterParams, writeAlias string) spanAndServ
 }
 
 // WriteSpan writes a span and its corresponding service:operation in ElasticSearch
-func (s *SpanWriter) WriteSpan(spanStartTime time.Time, span *dbmodel.Span) {
+func (s *SpanWriter) WriteSpan(spanStartTime time.Time, span *dbmodel.Span) error {
+	s.writerMetrics.indexCreate.Attempts.Inc(1)
+
 	spanIndexName, serviceIndexName := s.spanServiceIndex(spanStartTime)
 	if serviceIndexName != "" {
 		s.writeService(serviceIndexName, span)
 	}
 	s.writeSpan(spanIndexName, span)
 	s.logger.Debug("Wrote span to ES index", zap.String("index", spanIndexName))
+
+	return nil
 }
 
 // Close closes SpanWriter

internal/storage/v1/elasticsearch/spanstore/writerv1.go

@@ -31,8 +31,7 @@ func (s *SpanWriterV1) CreateTemplates(spanTemplate, serviceTemplate string, ind
 // WriteSpan writes a span and its corresponding service:operation in ElasticSearch
 func (s *SpanWriterV1) WriteSpan(_ context.Context, span *model.Span) error {
 	jsonSpan := s.spanConverter.FromDomainEmbedProcess(span)
-	s.spanWriter.WriteSpan(span.StartTime, jsonSpan)
-	return nil
+	return s.spanWriter.WriteSpan(span.StartTime, jsonSpan)
 }
 
 // Close closes SpanWriter

internal/storage/v1/elasticsearch/spanstore/writerv1_test.go

@@ -21,7 +21,7 @@ func TestSpanWriterV1_WriteSpan(t *testing.T) {
 	}
 	converter := NewFromDomain(true, []string{}, "-")
 	writerV1 := &SpanWriterV1{spanWriter: coreWriter, spanConverter: converter}
-	coreWriter.On("WriteSpan", s.StartTime, converter.FromDomainEmbedProcess(s))
+	coreWriter.On("WriteSpan", s.StartTime, converter.FromDomainEmbedProcess(s)).Return(nil)
 	err := writerV1.WriteSpan(context.Background(), s)
 	require.NoError(t, err)
 }