add support for Link. (open-telemetry#80)

* add support for Link.

* add AddLink to mockSpan

* update api documentation.

Author: rghetia

api/trace/api.go

@@ -75,6 +75,13 @@ type Span interface {
 	// IsRecordingEvents returns true if the span is active and recording events is enabled.
 	IsRecordingEvents() bool
 
+	// AddLink adds a link to the span.
+	AddLink(link Link)
+
+	// Link creates a link between this span and the other span specified by the SpanContext.
+	// It then adds the newly created Link to the span.
+	Link(sc core.SpanContext, attrs ...core.KeyValue)
+
 	// SpancContext returns span context of the span. Return SpanContext is usable
 	// even after the span is finished.
 	SpanContext() core.SpanContext
@@ -129,6 +136,22 @@ const (
 	FollowsFromRelationship
 )
 
+// Link is used to establish relationship between two spans within the same Trace or
+// across different Traces. Few examples of Link usage.
+//   1. Batch Processing: A batch of elements may contain elements associated with one
+//      or more traces/spans. Since there can only be one parent SpanContext, Link is
+//      used to keep reference to SpanContext of all elements in the batch.
+//   2. Public Endpoint: A SpanContext in incoming client request on a public endpoint
+//      is untrusted from service provider perspective. In such case it is advisable to
+//      start a new trace with appropriate sampling decision.
+//      However, it is desirable to associate incoming SpanContext to new trace initiated
+//      on service provider side so two traces (from Client and from Service Provider) can
+//      be correlated.
+type Link struct {
+	core.SpanContext
+	Attributes []core.KeyValue
+}
+
 // Start starts a new span using registered global tracer.
 func Start(ctx context.Context, name string, opts ...SpanOption) (context.Context, Span) {
 	return GlobalTracer().Start(ctx, name, opts...)

api/trace/current_test.go

@@ -107,3 +107,11 @@ func (mockSpan) Tracer() trace.Tracer {
 // Event does nothing.
 func (mockSpan) AddEvent(ctx context.Context, msg string, attrs ...core.KeyValue) {
 }
+
+// AddLink does nothing.
+func (mockSpan) AddLink(link trace.Link) {
+}
+
+// Link does nothing.
+func (mockSpan) Link(sc core.SpanContext, attrs ...core.KeyValue) {
+}

api/trace/noop_span.go

@@ -78,3 +78,11 @@ func (NoopSpan) AddEvent(ctx context.Context, msg string, attrs ...core.KeyValue
 // SetName does nothing.
 func (NoopSpan) SetName(name string) {
 }
+
+// AddLink does nothing.
+func (NoopSpan) AddLink(link Link) {
+}
+
+// Link does nothing.
+func (NoopSpan) Link(sc core.SpanContext, attrs ...core.KeyValue) {
+}

experimental/streaming/sdk/span.go

@@ -128,3 +128,9 @@ func (sp *span) SetName(name string) {
 		String: name,
 	})
 }
+
+func (sp *span) AddLink(link apitrace.Link) {
+}
+
+func (sp *span) Link(sc core.SpanContext, attrs ...core.KeyValue) {
+}

sdk/trace/export.go

@@ -22,6 +22,7 @@ import (
 	"google.golang.org/grpc/codes"
 
 	"go.opentelemetry.io/api/core"
+	apitrace "go.opentelemetry.io/api/trace"
 )
 
 // Exporter is a type for functions that receive sampled trace spans.
@@ -87,6 +88,7 @@ type SpanData struct {
 	// The values of Attributes each have type string, bool, or int64.
 	Attributes               map[string]interface{}
 	MessageEvents            []Event
+	Links                    []apitrace.Link
 	Status                   codes.Code
 	HasRemoteParent          bool
 	DroppedAttributeCount    int

sdk/trace/span.go

@@ -195,6 +195,38 @@ func (s *span) SetName(name string) {
 	makeSamplingDecision(data)
 }
 
+// AddLink implements Span interface. Specified link is added to the span.
+// If the total number of links associated with the span exceeds the limit
+// then the oldest link is removed to create space for the link being added.
+func (s *span) AddLink(link apitrace.Link) {
+	if !s.IsRecordingEvents() {
+		return
+	}
+	s.addLink(link)
+}
+
+// Link implements Span interface. It is similar to AddLink but it excepts
+// SpanContext and attributes as arguments instead of Link. It first creates
+// a Link object and then adds to the span.
+func (s *span) Link(sc core.SpanContext, attrs ...core.KeyValue) {
+	if !s.IsRecordingEvents() {
+		return
+	}
+	attrsCopy := attrs
+	if attrs != nil {
+		attrsCopy = make([]core.KeyValue, len(attrs))
+		copy(attrsCopy, attrs)
+	}
+	link := apitrace.Link{SpanContext: sc, Attributes: attrsCopy}
+	s.addLink(link)
+}
+
+func (s *span) addLink(link apitrace.Link) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.links.add(link)
+}
+
 // makeSpanData produces a SpanData representing the current state of the span.
 // It requires that s.data is non-nil.
 func (s *span) makeSpanData() *SpanData {
@@ -210,9 +242,21 @@ func (s *span) makeSpanData() *SpanData {
 		sd.MessageEvents = s.interfaceArrayToMessageEventArray()
 		sd.DroppedMessageEventCount = s.messageEvents.droppedCount
 	}
+	if len(s.links.queue) > 0 {
+		sd.Links = s.interfaceArrayToLinksArray()
+		sd.DroppedLinkCount = s.links.droppedCount
+	}
 	return &sd
 }
 
+func (s *span) interfaceArrayToLinksArray() []apitrace.Link {
+	linkArr := make([]apitrace.Link, 0)
+	for _, value := range s.links.queue {
+		linkArr = append(linkArr, value.(apitrace.Link))
+	}
+	return linkArr
+}
+
 func (s *span) interfaceArrayToMessageEventArray() []Event {
 	messageEventArr := make([]Event, 0)
 	for _, value := range s.messageEvents.queue {

sdk/trace/trace_test.go

@@ -316,6 +316,118 @@ func TestEventsOverLimit(t *testing.T) {
 	}
 }
 
+func TestAddLinks(t *testing.T) {
+	span := startSpan()
+	k1v1 := key.New("key1").String("value1")
+	k2v2 := key.New("key2").String("value2")
+
+	sc1 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x1}, SpanID: 0x3}
+	sc2 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x2}, SpanID: 0x3}
+
+	link1 := apitrace.Link{SpanContext: sc1, Attributes: []core.KeyValue{k1v1}}
+	link2 := apitrace.Link{SpanContext: sc2, Attributes: []core.KeyValue{k2v2}}
+	span.AddLink(link1)
+	span.AddLink(link2)
+
+	got, err := endSpan(span)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := &SpanData{
+		SpanContext: core.SpanContext{
+			TraceID:      tid,
+			TraceOptions: 0x1,
+		},
+		ParentSpanID:    sid,
+		Name:            "span0",
+		HasRemoteParent: true,
+		Links: []apitrace.Link{
+			{SpanContext: sc1, Attributes: []core.KeyValue{k1v1}},
+			{SpanContext: sc2, Attributes: []core.KeyValue{k2v2}},
+		},
+	}
+	if diff := cmp.Diff(got, want, cmp.AllowUnexported(Event{})); diff != "" {
+		t.Errorf("AddLink: -got +want %s", diff)
+	}
+}
+
+func TestLinks(t *testing.T) {
+	span := startSpan()
+	k1v1 := key.New("key1").String("value1")
+	k2v2 := key.New("key2").String("value2")
+	k3v3 := key.New("key3").String("value3")
+
+	sc1 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x1}, SpanID: 0x3}
+	sc2 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x2}, SpanID: 0x3}
+
+	span.Link(sc1, key.New("key1").String("value1"))
+	span.Link(sc2,
+		key.New("key2").String("value2"),
+		key.New("key3").String("value3"),
+	)
+	got, err := endSpan(span)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := &SpanData{
+		SpanContext: core.SpanContext{
+			TraceID:      tid,
+			TraceOptions: 0x1,
+		},
+		ParentSpanID:    sid,
+		Name:            "span0",
+		HasRemoteParent: true,
+		Links: []apitrace.Link{
+			{SpanContext: sc1, Attributes: []core.KeyValue{k1v1}},
+			{SpanContext: sc2, Attributes: []core.KeyValue{k2v2, k3v3}},
+		},
+	}
+	if diff := cmp.Diff(got, want, cmp.AllowUnexported(Event{})); diff != "" {
+		t.Errorf("Link: -got +want %s", diff)
+	}
+}
+
+func TestLinksOverLimit(t *testing.T) {
+	cfg := Config{MaxLinksPerSpan: 2}
+	ApplyConfig(cfg)
+	sc1 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x1}, SpanID: 0x3}
+	sc2 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x2}, SpanID: 0x3}
+	sc3 := core.SpanContext{TraceID: core.TraceID{High: 0x1, Low: 0x3}, SpanID: 0x3}
+
+	span := startSpan()
+	k2v2 := key.New("key2").String("value2")
+	k3v3 := key.New("key3").String("value3")
+
+	span.Link(sc1, key.New("key1").String("value1"))
+	span.Link(sc2, key.New("key2").String("value2"))
+	span.Link(sc3, key.New("key3").String("value3"))
+
+	got, err := endSpan(span)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := &SpanData{
+		SpanContext: core.SpanContext{
+			TraceID:      tid,
+			TraceOptions: 0x1,
+		},
+		ParentSpanID: sid,
+		Name:         "span0",
+		Links: []apitrace.Link{
+			{SpanContext: sc2, Attributes: []core.KeyValue{k2v2}},
+			{SpanContext: sc3, Attributes: []core.KeyValue{k3v3}},
+		},
+		DroppedLinkCount: 1,
+		HasRemoteParent:  true,
+	}
+	if diff := cmp.Diff(got, want, cmp.AllowUnexported(Event{})); diff != "" {
+		t.Errorf("Link over limit: -got +want %s", diff)
+	}
+}
+
 func TestSetSpanName(t *testing.T) {
 	want := "SpanName-1"
 	_, span := apitrace.GlobalTracer().Start(context.Background(), want,