receiver

package main

import (
	"context"
	"encoding/json"
	"log"
	"time"

	"contrib.go.opencensus.io/exporter/ocagent"
	"go.opencensus.io/trace"

	"github.com/census-instrumentation/opencensus-service/data"
	"github.com/census-instrumentation/opencensus-service/receiver"
	"github.com/census-instrumentation/opencensus-service/receiver/opencensus"
)

func main() {
	// This is what the cmd/ocagent code would look like this.
	// A trace receiver as per the trace receiver
	// configs that have been parsed.
	tr, err := opencensus.New("localhost:55678")
	if err != nil {
		log.Fatalf("Failed to create trace receiver: %v", err)
	}

	// The agent will combine all trace receivers like this.
	trl := []receiver.TraceReceiver{tr}

	// Once we have the span receiver which will connect to the
	// various exporter pipeline i.e. *tracepb.Span->OpenCensus.SpanData
	lsr := new(logSpanSink)
	for _, tr := range trl {
		if err := tr.StartTraceReception(context.Background(), lsr); err != nil {
			log.Fatalf("Failed to start trace receiver: %v", err)
		}
	}

	// Before exiting, stop all the trace receivers
	defer func() {
		for _, tr := range trl {
			_ = tr.StopTraceReception(context.Background())
		}
	}()
	log.Println("Done starting the trace receiver")
	// We are done with the agent-core

	// Now this code would exist in the client application e.g. client code.
	// Create the agent exporter
	oce, err := ocagent.NewExporter(ocagent.WithInsecure())
	if err != nil {
		log.Fatalf("Failed to create ocagent exporter: %v", err)
	}
	defer oce.Stop()

	// Register it as a trace exporter
	trace.RegisterExporter(oce)
	// For demo purposes we are always sampling
	trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})

	log.Println("Starting loop")
	ctx, span := trace.StartSpan(context.Background(), "ClientLibrarySpan")
	for i := 0; i < 10; i++ {
		_, span := trace.StartSpan(ctx, "ChildSpan")
		span.Annotatef([]trace.Attribute{
			trace.StringAttribute("type", "Child"),
			trace.Int64Attribute("i", int64(i)),
		}, "This is an annotation")
		<-time.After(100 * time.Millisecond)
		span.End()
		oce.Flush()
	}
	span.End()

	<-time.After(400 * time.Millisecond)
	oce.Flush()
	<-time.After(5 * time.Second)
}

type logSpanSink int

var _ receiver.TraceReceiverSink = (*logSpanSink)(nil)

func (lsr *logSpanSink) ReceiveTraceData(ctx context.Context, td data.TraceData) (*receiver.TraceReceiverAcknowledgement, error) {
	spansBlob, _ := json.MarshalIndent(td.Spans, " ", "  ")
	log.Printf("\n****\nNode: %#v\nSpans: %s\n****\n", td.Node, spansBlob)

	return &receiver.TraceReceiverAcknowledgement{SavedSpans: uint64(len(td.Spans))}, nil
}

Output: