Allow configurable elasticsearch index date format

pkg/es/config/config.go

@@ -50,6 +50,7 @@ type Configuration struct {
 	TagsFilePath      string
 	AllTagsAsFields   bool
 	TagDotReplacement string
+	IndexDateFormat   string
 	TLS               TLSConfig
 }
 
@@ -64,6 +65,7 @@ type TLSConfig struct {
 // ClientBuilder creates new es.Client
 type ClientBuilder interface {
 	NewClient(logger *zap.Logger, metricsFactory metrics.Factory) (es.Client, error)
+	GetIndexDateFormat() string
 	GetNumShards() int64
 	GetNumReplicas() int64
 	GetMaxSpanAge() time.Duration
@@ -168,6 +170,14 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 	if c.BulkFlushInterval == 0 {
 		c.BulkFlushInterval = source.BulkFlushInterval
 	}
+	if c.IndexDateFormat == "" {
+		c.IndexDateFormat = source.IndexDateFormat
+	}
+}
+
+// GetIndexDateFormat returns the Elasticsearch index name date format
+func (c *Configuration) GetIndexDateFormat() string {
+	return c.IndexDateFormat
 }
 
 // GetNumShards returns number of shards from Configuration

plugin/storage/es/factory.go

@@ -85,6 +85,7 @@ func (f *Factory) CreateSpanReader() (spanstore.Reader, error) {
 		MaxSpanAge:        cfg.GetMaxSpanAge(),
 		IndexPrefix:       cfg.GetIndexPrefix(),
 		TagDotReplacement: cfg.GetTagDotReplacement(),
+		DateFormat:        cfg.GetIndexDateFormat(),
 	}), nil
 }
 
@@ -99,6 +100,7 @@ func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) {
 			return nil, err
 		}
 	}
+	// add format
 	return esSpanStore.NewSpanWriter(esSpanStore.SpanWriterParams{Client: f.primaryClient,
 		Logger:            f.logger,
 		MetricsFactory:    f.metricsFactory,
@@ -108,6 +110,7 @@ func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) {
 		AllTagsAsFields:   f.primaryConfig.GetAllTagsAsFields(),
 		TagKeysAsFields:   tags,
 		TagDotReplacement: f.primaryConfig.GetTagDotReplacement(),
+		DateFormat:        f.primaryConfig.GetIndexDateFormat(),
 	}), nil
 }
 

plugin/storage/es/options.go

@@ -46,6 +46,7 @@ const (
 	suffixTagsAsFieldsAll   = suffixTagsAsFields + ".all"
 	suffixTagsFile          = suffixTagsAsFields + ".config-file"
 	suffixTagDeDotChar      = suffixTagsAsFields + ".dot-replacement"
+	suffixIndexDateFortmat  = ".index.date-format"
 )
 
 // TODO this should be moved next to config.Configuration struct (maybe ./flags package)
@@ -85,6 +86,7 @@ func NewOptions(primaryNamespace string, otherNamespaces ...string) *Options {
 				BulkActions:       1000,
 				BulkFlushInterval: time.Millisecond * 200,
 				TagDotReplacement: "@",
+				IndexDateFormat:   "2006-01-02",
 			},
 			servers:   "http://127.0.0.1:9200",
 			namespace: primaryNamespace,
@@ -188,6 +190,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixTagDeDotChar,
 		nsConfig.TagDotReplacement,
 		"(experimental) The character used to replace dots (\".\") in tag keys stored as object fields.")
+	flagSet.String(
+		nsConfig.namespace+suffixIndexDateFortmat,
+		nsConfig.IndexDateFormat,
+		"The date format of the Elasticsearch index names suffix, using Golang date format - https://golang.org/pkg/time/#example_Time_Format")
 }
 
 // InitFromViper initializes Options with properties from viper
@@ -219,6 +225,7 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
 	cfg.AllTagsAsFields = v.GetBool(cfg.namespace + suffixTagsAsFieldsAll)
 	cfg.TagsFilePath = v.GetString(cfg.namespace + suffixTagsFile)
 	cfg.TagDotReplacement = v.GetString(cfg.namespace + suffixTagDeDotChar)
+	cfg.IndexDateFormat = v.GetString(cfg.namespace + suffixIndexDateFortmat)
 }
 
 // GetPrimary returns primary configuration.

plugin/storage/es/spanstore/reader.go

@@ -96,6 +96,7 @@ type SpanReader struct {
 	spanIndexPrefix         string
 	serviceIndexPrefix      string
 	spanConverter           dbmodel.ToDomain
+	dateFormat              string
 }
 
 // SpanReaderParams holds constructor params for NewSpanReader
@@ -107,6 +108,7 @@ type SpanReaderParams struct {
 	serviceOperationStorage *ServiceOperationStorage
 	IndexPrefix             string
 	TagDotReplacement       string
+	DateFormat              string
 }
 
 // NewSpanReader returns a new SpanReader with a metrics.
@@ -128,6 +130,7 @@ func newSpanReader(p SpanReaderParams) *SpanReader {
 		spanIndexPrefix:         p.IndexPrefix + spanIndex,
 		serviceIndexPrefix:      p.IndexPrefix + serviceIndex,
 		spanConverter:           dbmodel.NewToDomain(p.TagDotReplacement),
+		dateFormat:              p.DateFormat,
 	}
 }
 
@@ -176,12 +179,12 @@ func (s *SpanReader) unmarshalJSONSpan(esSpanRaw *elastic.SearchHit) (*dbmodel.S
 // Returns the array of indices that we need to query, based on query params
 func (s *SpanReader) indicesForTimeRange(indexName string, startTime time.Time, endTime time.Time) []string {
 	var indices []string
-	firstIndex := indexWithDate(indexName, startTime)
-	currentIndex := indexWithDate(indexName, endTime)
+	firstIndex := indexWithDate(indexName, startTime, s.dateFormat)
+	currentIndex := indexWithDate(indexName, endTime, s.dateFormat)
 	for currentIndex != firstIndex {
 		indices = append(indices, currentIndex)
 		endTime = endTime.Add(-24 * time.Hour)
-		currentIndex = indexWithDate(indexName, endTime)
+		currentIndex = indexWithDate(indexName, endTime, s.dateFormat)
 	}
 	return append(indices, firstIndex)
 }

plugin/storage/es/spanstore/reader_test.go

@@ -97,6 +97,7 @@ func withSpanReader(fn func(r *spanReaderTest)) {
 			MaxSpanAge:        0,
 			IndexPrefix:       "",
 			TagDotReplacement: "@",
+			DateFormat:        "2006-01-02",
 		}),
 	}
 	fn(r)
@@ -320,24 +321,24 @@ func TestSpanReaderFindIndices(t *testing.T) {
 			startTime: today.Add(-time.Millisecond),
 			endTime:   today,
 			expected: []string{
-				indexWithDate(spanIndex, today),
+				indexWithDate(spanIndex, today, "2006-01-02"),
 			},
 		},
 		{
 			startTime: today.Add(-13 * time.Hour),
 			endTime:   today,
 			expected: []string{
-				indexWithDate(spanIndex, today),
-				indexWithDate(spanIndex, yesterday),
+				indexWithDate(spanIndex, today, "2006-01-02"),
+				indexWithDate(spanIndex, yesterday, "2006-01-02"),
 			},
 		},
 		{
 			startTime: today.Add(-48 * time.Hour),
 			endTime:   today,
 			expected: []string{
-				indexWithDate(spanIndex, today),
-				indexWithDate(spanIndex, yesterday),
-				indexWithDate(spanIndex, twoDaysAgo),
+				indexWithDate(spanIndex, today, "2006-01-02"),
+				indexWithDate(spanIndex, yesterday, "2006-01-02"),
+				indexWithDate(spanIndex, twoDaysAgo, "2006-01-02"),
 			},
 		},
 	}
@@ -351,7 +352,7 @@ func TestSpanReaderFindIndices(t *testing.T) {
 
 func TestSpanReader_indexWithDate(t *testing.T) {
 	withSpanReader(func(r *spanReaderTest) {
-		actual := indexWithDate(spanIndex, time.Date(1995, time.April, 21, 4, 21, 19, 95, time.UTC))
+		actual := indexWithDate(spanIndex, time.Date(1995, time.April, 21, 4, 21, 19, 95, time.UTC), "2006-01-02")
 		assert.Equal(t, "jaeger-span-1995-04-21", actual)
 	})
 }

plugin/storage/es/spanstore/writer.go

@@ -16,6 +16,7 @@ package spanstore
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"strings"
 	"time"
@@ -58,6 +59,7 @@ type SpanWriter struct {
 	spanIndexPrefix    string
 	serviceIndexPrefix string
 	spanConverter      dbmodel.FromDomain
+	dateFormat         string // used for elasticsearch rolling index name suffix
 }
 
 // SpanWriterParams holds constructor parameters for NewSpanWriter
@@ -71,6 +73,7 @@ type SpanWriterParams struct {
 	AllTagsAsFields   bool
 	TagKeysAsFields   []string
 	TagDotReplacement string
+	DateFormat        string
 }
 
 // NewSpanWriter creates a new SpanWriter for use
@@ -86,9 +89,10 @@ func NewSpanWriter(p SpanWriterParams) *SpanWriter {
 		p.IndexPrefix += ":"
 	}
 	return &SpanWriter{
-		ctx:    ctx,
-		client: p.Client,
-		logger: p.Logger,
+		ctx:        ctx,
+		client:     p.Client,
+		logger:     p.Logger,
+		dateFormat: p.DateFormat,
 		writerMetrics: spanWriterMetrics{
 			indexCreate: storageMetrics.NewWriteMetrics(p.MetricsFactory, "index_create"),
 		},
@@ -109,9 +113,8 @@ func NewSpanWriter(p SpanWriterParams) *SpanWriter {
 
 // WriteSpan writes a span and its corresponding service:operation in ElasticSearch
 func (s *SpanWriter) WriteSpan(span *model.Span) error {
-	spanIndexName := indexWithDate(s.spanIndexPrefix, span.StartTime)
-	serviceIndexName := indexWithDate(s.serviceIndexPrefix, span.StartTime)
-
+	serviceIndexName := indexWithDate(s.serviceIndexPrefix, span.StartTime, s.dateFormat)
+	spanIndexName := indexWithDate(s.spanIndexPrefix, span.StartTime, s.dateFormat)
 	jsonSpan := s.spanConverter.FromDomainEmbedProcess(span)
 
 	if err := s.createIndex(serviceIndexName, serviceMapping, jsonSpan); err != nil {
@@ -130,9 +133,11 @@ func (s *SpanWriter) Close() error {
 	return s.client.Close()
 }
 
-func indexWithDate(indexPrefix string, date time.Time) string {
-	spanDate := date.UTC().Format("2006-01-02")
-	return indexPrefix + spanDate
+func indexWithDate(indexPrefix string, date time.Time, dateFormat string) string {
+	spanDate := date.UTC().Format(dateFormat)
+	r := indexPrefix + spanDate
+	fmt.Println(r)
+	return r
 }
 
 func (s *SpanWriter) createIndex(indexName string, mapping string, jsonSpan *dbmodel.Span) error {

plugin/storage/es/spanstore/writer_test.go

@@ -48,7 +48,7 @@ func withSpanWriter(fn func(w *spanWriterTest)) {
 		client:    client,
 		logger:    logger,
 		logBuffer: logBuffer,
-		writer:    NewSpanWriter(SpanWriterParams{Client: client, Logger: logger, MetricsFactory: metricsFactory}),
+		writer:    NewSpanWriter(SpanWriterParams{Client: client, Logger: logger, MetricsFactory: metricsFactory, DateFormat: "2006-01-02"}),
 	}
 	fn(w)
 }
@@ -75,6 +75,12 @@ func TestNewSpanWriterIndexPrefix(t *testing.T) {
 	}
 }
 
+func TestWithDateFormat(t *testing.T) {
+	withSpanWriter(func(w *spanWriterTest) {
+		assert.Equal(t, "2006-01-02", w.writer.dateFormat, "Span writer index date format")
+	})
+}
+
 func TestClientClose(t *testing.T) {
 	withSpanWriter(func(w *spanWriterTest) {
 		w.client.On("Close").Return(nil)
@@ -221,8 +227,9 @@ func TestSpanIndexName(t *testing.T) {
 	span := &model.Span{
 		StartTime: date,
 	}
-	spanIndexName := indexWithDate(spanIndex, span.StartTime)
-	serviceIndexName := indexWithDate(serviceIndex, span.StartTime)
+
+	spanIndexName := indexWithDate(spanIndex, span.StartTime, "2006-01-02")
+	serviceIndexName := indexWithDate(serviceIndex, span.StartTime, "2006-01-02")
 	assert.Equal(t, "jaeger-span-1995-04-21", spanIndexName)
 	assert.Equal(t, "jaeger-service-1995-04-21", serviceIndexName)
 }

plugin/storage/integration/elasticsearch_test.go

@@ -42,6 +42,7 @@ const (
 	password        = "changeme" // the elasticsearch default password
 	indexPrefix     = "integration-test"
 	tagKeyDeDotChar = "@"
+	dateFormat      = "2006-01-02"
 )
 
 type ESStorageIntegration struct {
@@ -95,6 +96,7 @@ func (s *ESStorageIntegration) initSpanstore(allTagsAsFields bool) {
 			IndexPrefix:       indexPrefix,
 			AllTagsAsFields:   allTagsAsFields,
 			TagDotReplacement: tagKeyDeDotChar,
+			DateFormat:        dateFormat,
 		})
 	s.SpanReader = spanstore.NewSpanReader(spanstore.SpanReaderParams{
 		Client:            client,
@@ -103,6 +105,7 @@ func (s *ESStorageIntegration) initSpanstore(allTagsAsFields bool) {
 		IndexPrefix:       indexPrefix,
 		MaxSpanAge:        72 * time.Hour,
 		TagDotReplacement: tagKeyDeDotChar,
+		DateFormat:        dateFormat,
 	})
 }