- add the ability to accept partial spans that will be upserted in pe…

…rsistence See jaegertracing#729 for details on the concept Signed-off-by: Martin Foerster <martin@atroo.de>
phal0r · Mar 8, 2018 · 368dbd6 · 368dbd6
1 parent dbd5db7
commit 368dbd6
Show file tree

Hide file tree

Showing 21 changed files with 224 additions and 20 deletions.
diff --git a/model/converter/json/fixtures/ui_01.json b/model/converter/json/fixtures/ui_01.json
@@ -32,7 +32,8 @@
         }
       ],
       "processID": "p1",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     },
     {
       "traceID": "1",
@@ -70,7 +71,8 @@
       ],
       "logs": [],
       "processID": "p1",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     },
     {
       "traceID": "1",
@@ -88,7 +90,8 @@
       "tags": [],
       "logs": [],
       "processID": "p2",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     },
     {
       "traceID": "1",
@@ -118,7 +121,8 @@
       "processID": "p2",
       "warnings": [
         "some span warning"
-      ]
+      ],
+      "incomplete": false
     },
     {
       "traceID": "1",
@@ -138,7 +142,8 @@
       "processID": "p2",
       "warnings": [
         "some span warning"
-      ]
+      ],
+      "incomplete": false
     }
   ],
   "processes": {

diff --git a/model/converter/json/to_domain.go b/model/converter/json/to_domain.go
@@ -74,6 +74,7 @@ func (td toDomain) spanToDomain(dbSpan *json.Span) (*model.Span, error) {
 		Tags:          tags,
 		Logs:          logs,
 		Process:       process,
+		Incomplete:    dbSpan.Incomplete,
 	}
 	return span, nil
 }

diff --git a/model/converter/thrift/jaeger/fixtures/model_01.json b/model/converter/thrift/jaeger/fixtures/model_01.json
@@ -6,6 +6,7 @@
     "operationName":"get",
     "startTime":"2017-01-26T16:46:31.639875-05:00",
     "duration":22938000,
+    "type": "jaeger",
     "tags":[
       {
         "key":"http.url",

diff --git a/model/converter/thrift/jaeger/fixtures/model_02.json b/model/converter/thrift/jaeger/fixtures/model_02.json
@@ -6,6 +6,7 @@
     "operationName":"get",
     "startTime":"2017-01-26T16:46:31.639875-05:00",
     "duration":22938000,
+    "type": "jaeger",
     "tags":[
       {
         "key":"peer.service",
@@ -17,4 +18,4 @@
       "serviceName":"api"
     }
   }
-]
+]
diff --git a/model/converter/thrift/jaeger/fixtures/model_03.json b/model/converter/thrift/jaeger/fixtures/model_03.json
@@ -6,6 +6,7 @@
     "operationName":"get",
     "startTime":"2017-01-26T16:46:31.639875-05:00",
     "duration":22938000,
+    "type": "jaeger",
     "tags":[
       {
         "key":"http.url",
@@ -97,6 +98,7 @@
     "operationName":"get",
     "startTime":"2017-01-26T16:46:31.639875-05:00",
     "duration":22938000,
+    "type": "jaeger",
     "references": [
       {
         "refType": "child-of",

diff --git a/model/converter/thrift/jaeger/fixtures/thrift_batch_01.json b/model/converter/thrift/jaeger/fixtures/thrift_batch_01.json
@@ -27,6 +27,7 @@
       "operationName": "get",
       "startTime": 1485467191639875,
       "duration": 22938,
+      "type": "jaeger",
       "tags": [
         {
           "key": "http.url",

diff --git a/model/converter/thrift/jaeger/fixtures/thrift_batch_02.json b/model/converter/thrift/jaeger/fixtures/thrift_batch_02.json
@@ -11,6 +11,7 @@
       "operationName": "get",
       "startTime": 1485467191639875,
       "duration": 22938,
+      "type": "jaeger",
       "tags": [
         {
           "key": "peer.service",

diff --git a/model/converter/thrift/jaeger/to_domain.go b/model/converter/thrift/jaeger/to_domain.go
@@ -70,6 +70,8 @@ func (td toDomain) transformSpan(jSpan *jaeger.Span, mProcess *model.Process) *m
 		Tags:          tags,
 		Logs:          td.getLogs(jSpan.Logs),
 		Process:       mProcess,
+		Incomplete: 	 jSpan.Incomplete,
+		Type:					 model.JaegerSpanType,
 	}
 }
 

diff --git a/model/converter/thrift/zipkin/fixtures/jaeger_01.json b/model/converter/thrift/zipkin/fixtures/jaeger_01.json
@@ -5,6 +5,7 @@
       "spanID": "2",
       "operationName": "test-general-conversion",
       "startTime": "2017-01-26T16:46:31.639875-05:00",
+      "type": "zipkin",
       "process": {
         "serviceName": "service-x"
       },
@@ -34,6 +35,7 @@
       "spanID": "2",
       "operationName": "some-operation",
       "startTime": "1970-01-01T00:00:00-00:00",
+      "type": "zipkin",
       "tags": [
         {
           "key": "peer.service",
@@ -66,6 +68,7 @@
       "parentSpanID": "2",
       "operationName": "some-operation",
       "startTime": "1970-01-01T00:00:00-00:00",
+      "type": "zipkin",
       "tags": [
         {
           "key": "peer.service",

diff --git a/model/converter/thrift/zipkin/fixtures/jaeger_02.json b/model/converter/thrift/zipkin/fixtures/jaeger_02.json
@@ -5,6 +5,7 @@
       "spanID": "2",
       "operationName": "test-process-tags",
       "startTime": "1970-01-01T00:00:00-00:00",
+      "type": "zipkin",
       "process": {
         "serviceName": "service-x",
         "tags": [

diff --git a/model/converter/thrift/zipkin/fixtures/jaeger_03.json b/model/converter/thrift/zipkin/fixtures/jaeger_03.json
@@ -6,6 +6,7 @@
       "flags": 2,
       "operationName": "test-custom-tags",
       "startTime": "1970-01-01T00:00:00-00:00",
+      "type": "zipkin",
       "process": {
         "serviceName": "service-x"
       },

diff --git a/model/converter/thrift/zipkin/to_domain.go b/model/converter/thrift/zipkin/to_domain.go
@@ -143,6 +143,7 @@ func (td toDomain) transformSpan(zSpan *zipkincore.Span) []*model.Span {
 		Duration:      model.MicrosecondsAsDuration(uint64(zSpan.GetDuration())),
 		Tags:          tags,
 		Logs:          td.getLogs(zSpan.Annotations),
+		Type:					 model.ZipkinSpanType,
 	}}
 
 	cs := td.findAnnotation(zSpan, zipkincore.CLIENT_SEND)

diff --git a/model/hash.go b/model/hash.go
@@ -28,9 +28,9 @@ type Hashable interface {
 
 // HashCode calcualtes a FNV-1a hash code for a Hashable object.
 func HashCode(o Hashable) (uint64, error) {
-	h := fnv.New64a()
-	if err := o.Hash(h); err != nil {
-		return 0, err
-	}
-	return h.Sum64(), nil
+		h := fnv.New64a()
+		if err := o.Hash(h); err != nil {
+			return 0, err
+		}
+		return h.Sum64(), nil
 }
diff --git a/model/json/fixture-actual.json b/model/json/fixture-actual.json
@@ -0,0 +1,105 @@
+{
+  "traceID": "abc0",
+  "spans": [
+    {
+      "traceID": "abc0",
+      "spanID": "abc0",
+      "operationName": "root-span",
+      "references": null,
+      "startTime": 1000,
+      "duration": 500,
+      "tags": null,
+      "logs": null,
+      "processID": "p1",
+      "warnings": null,
+      "incomplete": false
+    },
+    {
+      "traceID": "abc0",
+      "spanID": "123",
+      "operationName": "span1",
+      "references": [
+        {
+          "refType": "CHILD_OF",
+          "traceID": "abc0",
+          "spanID": "abc0"
+        }
+      ],
+      "startTime": 1000,
+      "duration": 500,
+      "tags": [
+        {
+          "key": "error",
+          "type": "bool",
+          "value": true
+        },
+        {
+          "key": "int64",
+          "type": "int64",
+          "value": 123
+        },
+        {
+          "key": "float64",
+          "type": "float64",
+          "value": 123.567
+        },
+        {
+          "key": "binary",
+          "type": "binary",
+          "value": "AQ=="
+        }
+      ],
+      "logs": [
+        {
+          "timestamp": 1400,
+          "fields": [
+            {
+              "key": "error",
+              "type": "string",
+              "value": "something bad happened"
+            }
+          ]
+        }
+      ],
+      "processID": "p2",
+      "warnings": null,
+      "incomplete": false
+    },
+    {
+      "traceID": "abc0",
+      "spanID": "567",
+      "operationName": "span2",
+      "references": [
+        {
+          "refType": "FOLLOWS_FROM",
+          "traceID": "abc0",
+          "spanID": "abc0"
+        }
+      ],
+      "startTime": 1000,
+      "duration": 500,
+      "tags": null,
+      "logs": null,
+      "processID": "p2",
+      "warnings": null,
+      "incomplete": false
+    }
+  ],
+  "processes": {
+    "p1": {
+      "serviceName": "service_1",
+      "tags": null
+    },
+    "p2": {
+      "serviceName": "service_2",
+      "tags": [
+        {
+          "key": "host",
+          "type": "string",
+          "value": "google.com"
+        }
+      ]
+    }
+  },
+  "warnings": null
+}
diff --git a/model/json/fixture.json b/model/json/fixture.json
@@ -11,7 +11,8 @@
       "tags": null,
       "logs": null,
       "processID": "p1",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     },
     {
       "traceID": "abc0",
@@ -61,7 +62,8 @@
         }
       ],
       "processID": "p2",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     },
     {
       "traceID": "abc0",
@@ -79,7 +81,8 @@
       "tags": null,
       "logs": null,
       "processID": "p2",
-      "warnings": null
+      "warnings": null,
+      "incomplete": false
     }
   ],
   "processes": {

diff --git a/model/json/model.go b/model/json/model.go
@@ -79,6 +79,7 @@ type Span struct {
 	ProcessID     ProcessID   `json:"processID"`
 	Process       *Process    `json:"process,omitempty"`
 	Warnings      []string    `json:"warnings"`
+	Incomplete		bool 				`json:"incomplete"`
 }
 
 // Reference is a reference from one span to another

diff --git a/model/span.go b/model/span.go
@@ -30,6 +30,9 @@ const (
 	sampledFlag = Flags(1)
 	// debugFlag is the bit set in Flags in order to define a span as a debug span
 	debugFlag = Flags(2)
+
+	JaegerSpanType string = "jaeger"
+	ZipkinSpanType string = "zipkin"
 )
 
 // TraceID is a random 128bit identifier for a trace
@@ -58,14 +61,29 @@ type Span struct {
 	Logs          []Log         `json:"logs,omitempty"`
 	Process       *Process      `json:"process"`
 	Warnings      []string      `json:"warnings,omitempty"`
+	Incomplete		bool 					`json:"incomplete"`
+	Type					string				`json:"type,omitempty"`
+}
+
+type JaegerSpanHash struct {
+	TraceID TraceID
+ 	SpanID	SpanID
 }
 
 // Hash implements Hash from Hashable.
 func (s *Span) Hash(w io.Writer) (err error) {
 	// gob is not the most efficient way, but it ensures we don't miss any fields.
 	// See BenchmarkSpanHash in span_test.go
 	enc := gob.NewEncoder(w)
-	return enc.Encode(s)
+
+	//with supporting incomplete spans the hash needs to be unique for a combination
+	//of spanId and traceId, so the final representation of the same span
+	//will overwrite the incomplete version in cassandra
+	if s.Type == JaegerSpanType {
+		return enc.Encode(JaegerSpanHash{s.TraceID, s.SpanID})
+	} else {
+		return enc.Encode(s)
+	}
 }
 
 // HasSpanKind returns true if the span has a `span.kind` tag set to `kind`.

diff --git a/plugin/storage/es/spanstore/writer.go b/plugin/storage/es/spanstore/writer.go
@@ -126,7 +126,7 @@ func (s *SpanWriter) WriteSpan(span *model.Span) error {
 	if err := s.createIndex(spanIndexName, spanMapping, jsonSpan); err != nil {
 		return err
 	}
-	return s.writeSpan(spanIndexName, jsonSpan)
+	return s.writeSpan(spanIndexName, jsonSpan, span.Type)
 }
 
 func indexNames(span *model.Span) (string, string) {
@@ -171,10 +171,15 @@ func (s *SpanWriter) writeService(indexName string, jsonSpan *jModel.Span) error
 	return s.serviceWriter(indexName, jsonSpan)
 }
 
-func (s *SpanWriter) writeSpan(indexName string, jsonSpan *jModel.Span) error {
+func (s *SpanWriter) writeSpan(indexName string, jsonSpan *jModel.Span, jaegerSpanType string) error {
 	start := time.Now()
 	elasticSpan := Span{Span: jsonSpan, StartTimeMillis: jsonSpan.StartTime / 1000} // Microseconds to milliseconds
-	_, err := s.client.Index().Index(indexName).Type(spanType).BodyJson(&elasticSpan).Do(s.ctx)
+	var err error
+	if jaegerSpanType == model.JaegerSpanType {
+		_, err = s.client.Index().Index(indexName).Type(spanType).Id(string(jsonSpan.TraceID) + string(jsonSpan.SpanID)).BodyJson(&elasticSpan).Do(s.ctx)
+	} else {
+		_, 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)