implemented support for incomplete spans

- span types have a new attribute incomplete
- spans without given incomplete attribute will default to false
- backwards compatible with old clients
- breaking change for es and cassandra schemas
- generated thrift types are up to date in this pr
- detailled concept described in jaegertracing#729
- based on pull request jaegertracing#728

Signed-off-by: Martin Foerster <martin@atroo.de>

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": {

model/converter/json/from_domain.go

@@ -70,6 +70,7 @@ func (fd fromDomain) convertSpanInternal(span *model.Span) json.Span {
 		Duration:      model.DurationAsMicroseconds(span.Duration),
 		Tags:          fd.convertKeyValuesFunc(span.Tags),
 		Logs:          fd.convertLogs(span.Logs),
+		Incomplete:    span.Incomplete,
 	}
 }
 

model/converter/json/to_domain.go

@@ -75,6 +75,7 @@ func (td toDomain) spanToDomain(dbSpan *json.Span) (*model.Span, error) {
 		Tags:          tags,
 		Logs:          logs,
 		Process:       process,
+		Incomplete:    dbSpan.Incomplete,
 	}
 	return span, nil
 }

model/converter/thrift/jaeger/fixtures/model_02.json

@@ -16,5 +16,24 @@
     "process":{
       "serviceName":"api"
     }
+  },
+  {
+    "traceID":"52969a8955571a3f",
+    "spanID":"647d98",
+    "parentSpanID":"68c4e3",
+    "operationName":"get",
+    "startTime":"2017-01-26T16:46:31.639875-05:00",
+    "duration":22938000,
+    "tags":[
+      {
+        "key":"peer.service",
+        "vType":"binary",
+        "vBlob":"AAAAAAAAMDk="
+      }
+    ],
+    "process":{
+      "serviceName":"api"
+    },
+    "incomplete": true
   }
-]
+]

model/converter/thrift/jaeger/fixtures/thrift_batch_02.json

@@ -18,6 +18,22 @@
           "vBinary": "AAAAAAAAMDk="
         }
       ]
+    },
+    {
+      "traceIdLow": 5951113872249657919,
+      "spanId": 6585752,
+      "parentSpanId": 6866147,
+      "operationName": "get",
+      "startTime": 1485467191639875,
+      "duration": 22938,
+      "tags": [
+        {
+          "key": "peer.service",
+          "vType": "BINARY",
+          "vBinary": "AAAAAAAAMDk="
+        }
+      ],
+      "incomplete": true
     }
   ]
 }

model/converter/thrift/jaeger/from_domain.go

@@ -151,6 +151,7 @@ func (d domainToJaegerTransformer) transformSpan(span *model.Span) *jaeger.Span
 		Duration:      int64(model.DurationAsMicroseconds(span.Duration)),
 		Tags:          tags,
 		Logs:          logs,
+		Incomplete:    &span.Incomplete,
 	}
 	return jaegerSpan
 }

model/converter/thrift/jaeger/to_domain.go

@@ -70,9 +70,17 @@ func (td toDomain) transformSpan(jSpan *jaeger.Span, mProcess *model.Process) *m
 		Tags:          tags,
 		Logs:          td.getLogs(jSpan.Logs),
 		Process:       mProcess,
+		Incomplete:    td.checkIncomplete(jSpan.Incomplete),
 	}
 }
 
+func (td toDomain) checkIncomplete(incomplete *bool) bool {
+	if incomplete == nil {
+		return false
+	}
+	return *incomplete
+}
+
 func (td toDomain) getReferences(jRefs []*jaeger.SpanRef) []model.SpanRef {
 	if len(jRefs) == 0 {
 		return nil

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),
+		Incomplete:    false,
 	}}
 
 	cs := td.findAnnotation(zSpan, zipkincore.CLIENT_SEND)

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": {

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

model/span.go

@@ -57,6 +57,7 @@ type Span struct {
 	Logs          []Log         `json:"logs,omitempty"`
 	Process       *Process      `json:"process"`
 	Warnings      []string      `json:"warnings,omitempty"`
+	Incomplete    bool          `json:"incomplete"`
 }
 
 // Hash implements Hash from Hashable.

plugin/storage/cassandra/schema/v001.cql.tmpl

@@ -64,11 +64,12 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.traces (
     logs            list<frozen<log>>,
     refs            list<frozen<span_ref>>,
     process         frozen<process>,
+    incomplete      boolean,
     PRIMARY KEY (trace_id, span_id, span_hash)
 )
     WITH compaction = {
-        'compaction_window_size': '1', 
-        'compaction_window_unit': 'HOURS', 
+        'compaction_window_size': '1',
+        'compaction_window_unit': 'HOURS',
         'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
@@ -83,7 +84,7 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.service_names (
     WITH compaction = {
         'min_threshold': '4',
         'max_threshold': '32',
-        'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy' 
+        'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
     AND default_time_to_live = ${trace_ttl}
@@ -114,8 +115,8 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.service_operation_index (
     PRIMARY KEY ((service_name, operation_name), start_time)
 ) WITH CLUSTERING ORDER BY (start_time DESC)
     AND compaction = {
-        'compaction_window_size': '1', 
-        'compaction_window_unit': 'HOURS', 
+        'compaction_window_size': '1',
+        'compaction_window_unit': 'HOURS',
         'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
@@ -131,8 +132,8 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.service_name_index (
     PRIMARY KEY ((service_name, bucket), start_time)
 ) WITH CLUSTERING ORDER BY (start_time DESC)
     AND compaction = {
-        'compaction_window_size': '1', 
-        'compaction_window_unit': 'HOURS', 
+        'compaction_window_size': '1',
+        'compaction_window_unit': 'HOURS',
         'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
@@ -150,8 +151,8 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.duration_index (
     PRIMARY KEY ((service_name, operation_name, bucket), duration, start_time, trace_id)
 ) WITH CLUSTERING ORDER BY (duration DESC, start_time DESC)
     AND compaction = {
-        'compaction_window_size': '1', 
-        'compaction_window_unit': 'HOURS', 
+        'compaction_window_size': '1',
+        'compaction_window_unit': 'HOURS',
         'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
@@ -172,8 +173,8 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.tag_index (
 )
     WITH CLUSTERING ORDER BY (start_time DESC)
     AND compaction = {
-        'compaction_window_size': '1', 
-        'compaction_window_unit': 'HOURS', 
+        'compaction_window_size': '1',
+        'compaction_window_unit': 'HOURS',
         'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy'
     }
     AND dclocal_read_repair_chance = 0.0
@@ -202,6 +203,6 @@ CREATE TABLE IF NOT EXISTS ${keyspace}.dependencies (
     }
     AND default_time_to_live = ${dependencies_ttl};
 
-CREATE CUSTOM INDEX ON ${keyspace}.dependencies (ts_index) 
-    USING 'org.apache.cassandra.index.sasi.SASIIndex' 
+CREATE CUSTOM INDEX ON ${keyspace}.dependencies (ts_index)
+    USING 'org.apache.cassandra.index.sasi.SASIIndex'
     WITH OPTIONS = {'mode': 'SPARSE'};

plugin/storage/es/spanstore/schema.go

@@ -70,6 +70,9 @@ var (
 		    "flags":{
 		       "type":"integer"
 		    },
+        "incomplete":{
+            "type":"boolean"
+        },
 		    "logs":{
 		       "properties":{
 			  "timestamp":{