Add startTimeMillis field to JSON Spans submitted to ElasticSearch (#491

)

* Add startTimeMillis field to JSON Spans submitted to ElasticSearch, and adjust mapping to match the new field.

plugin/storage/es/spanstore/schema.go

@@ -60,6 +60,10 @@ var (
 		    "startTime":{
 		       "type":"long"
 		    },
+		    "startTimeMillis":{
+		       "type":"date",
+		       "format":"epoch_millis"
+		    },
 		    "duration":{
 		       "type":"long"
 		    },

plugin/storage/es/spanstore/writer.go

@@ -65,6 +65,15 @@ type Service struct {
 	OperationName string `json:"operationName"`
 }
 
+// Span adds a StartTimeMillis field to the standard JSON span.
+// ElasticSearch does not support a UNIX Epoch timestamp in microseconds,
+// so Jaeger maps StartTime to a 'long' type. This extra StartTimeMillis field
+// works around this issue, enabling timerange queries.
+type Span struct {
+	*jModel.Span
+	StartTimeMillis uint64 `json:"startTimeMillis"`
+}
+
 // NewSpanWriter creates a new SpanWriter for use
 func NewSpanWriter(
 	client es.Client,
@@ -164,7 +173,8 @@ func (s *SpanWriter) writeService(indexName string, jsonSpan *jModel.Span) error
 
 func (s *SpanWriter) writeSpan(indexName string, jsonSpan *jModel.Span) error {
 	start := time.Now()
-	_, err := s.client.Index().Index(indexName).Type(spanType).BodyJson(jsonSpan).Do(s.ctx)
+	elasticSpan := Span{Span: jsonSpan, StartTimeMillis: jsonSpan.StartTime / 1000} // Microseconds to milliseconds
+	_, err := s.client.Index().Index(indexName).Type(spanType).BodyJson(&elasticSpan).Do(s.ctx)
 	s.writerMetrics.spans.Emit(err, time.Since(start))
 	if err != nil {
 		return s.logError(jsonSpan, err, "Failed to insert span", s.logger)

plugin/storage/es/spanstore/writer_test.go

@@ -194,7 +194,7 @@ func TestSpanWriter_WriteSpan(t *testing.T) {
 				indexServicePut.On("Do", mock.AnythingOfType("*context.emptyCtx")).Return(nil, testCase.servicePutError)
 
 				indexSpanPut.On("Id", mock.AnythingOfType("string")).Return(indexSpanPut)
-				indexSpanPut.On("BodyJson", mock.AnythingOfType("*json.Span")).Return(indexSpanPut)
+				indexSpanPut.On("BodyJson", mock.AnythingOfType("*spanstore.Span")).Return(indexSpanPut)
 				indexSpanPut.On("Do", mock.AnythingOfType("*context.emptyCtx")).Return(nil, testCase.spanPutError)
 
 				w.client.On("IndexExists", stringMatcher(spanIndexName)).Return(spanExistsService)
@@ -346,7 +346,7 @@ func TestWriteSpanInternal(t *testing.T) {
 		indexName := "jaeger-1995-04-21"
 		indexService.On("Index", stringMatcher(indexName)).Return(indexService)
 		indexService.On("Type", stringMatcher(spanType)).Return(indexService)
-		indexService.On("BodyJson", mock.AnythingOfType("*json.Span")).Return(indexService)
+		indexService.On("BodyJson", mock.AnythingOfType("*spanstore.Span")).Return(indexService)
 		indexService.On("Do", mock.AnythingOfType("*context.emptyCtx")).Return(&elastic.IndexResponse{}, nil)
 
 		w.client.On("Index").Return(indexService)
@@ -368,7 +368,7 @@ func TestWriteSpanInternalError(t *testing.T) {
 		indexName := "jaeger-1995-04-21"
 		indexService.On("Index", stringMatcher(indexName)).Return(indexService)
 		indexService.On("Type", stringMatcher(spanType)).Return(indexService)
-		indexService.On("BodyJson", mock.AnythingOfType("*json.Span")).Return(indexService)
+		indexService.On("BodyJson", mock.AnythingOfType("*spanstore.Span")).Return(indexService)
 		indexService.On("Do", mock.AnythingOfType("*context.emptyCtx")).Return(nil, errors.New("span insertion error"))
 
 		w.client.On("Index").Return(indexService)