Repository: honeycombio/refinery
Pull request #279 (2021-07-01 20:35:51)

Add support for OTLP over HTTP/protobuf
Adds support for ingesting OTLP requests over HTTP with protobuf data. Changs in this PR:

- Move the existing gRPC export handler and helpers to to otlp_trace.go and tests into otlp_trace_test.go
- Add HTTP handler `postOTLP` that retrieves honeycomb headers, decodes the request body into a OTLP ExportTraceRequest and hands to internal function to process request
- Update OTLP grpc Export handler to retrieve honeycomb headers from metadata and hand request to new internal function to process
- Adds tests to verify rejection behaviour for non-protobuf based requests to postOTLP and test to pass request to postOTLP handler endpoint

MikeGoldsmith: Most recent commits was to add support for gzip/zstd encoding as part of the parsing process.

Patch:
diff --git a/route/errors.go b/route/errors.go
index 5e347ec..ff697d9 100644
--- a/route/errors.go
+++ b/route/errors.go
@@ -33,6 +33,7 @@ var (
 	ErrUpstreamUnavailable = handlerError{nil, "upstream target unavailable", http.StatusServiceUnavailable, true, true}
 	ErrReqToEvent          = handlerError{nil, "failed to parse event", http.StatusBadRequest, false, true}
 	ErrBatchToEvent        = handlerError{nil, "failed to parse event within batch", http.StatusBadRequest, false, true}
+	ErrInvalidContentType  = handlerError{nil, "invalid content-type - only 'application/protobuf' is supported", http.StatusNotImplemented, false, true}
 )
 
 func (r *Router) handlerReturnWithError(w http.ResponseWriter, he handlerError, err error) {
diff --git a/route/otlp_trace.go b/route/otlp_trace.go
new file mode 100644
index 0000000..dca390e
--- /dev/null
+++ b/route/otlp_trace.go
@@ -0,0 +1,404 @@
+package route
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/honeycombio/refinery/types"
+	"github.com/klauspost/compress/zstd"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	collectortrace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/collector/trace/v1"
+	common "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/common/v1"
+	trace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/trace/v1"
+)
+
+func (router *Router) postOTLP(w http.ResponseWriter, req *http.Request) {
+	defer req.Body.Close()
+	contentType := req.Header.Get("content-type")
+	if contentType != "application/protobuf" && contentType != "application/x-protobuf" {
+		router.handlerReturnWithError(w, ErrInvalidContentType, errors.New("invalid content-type"))
+		return
+	}
+
+	apiKey, datasetName, err := getAPIKeyDatasetAndTokenFromHttpHeaders(req)
+	if err != nil {
+		router.handlerReturnWithError(w, ErrAuthNeeded, err)
+		return
+	}
+
+	request, cleanup, err := parseOTLPBody(req, router.zstdDecoders)
+	defer cleanup()
+	if err != nil {
+		router.handlerReturnWithError(w, ErrPostBody, err)
+	}
+
+	if err := processTraceRequest(req.Context(), router, request, apiKey, datasetName); err != nil {
+		router.handlerReturnWithError(w, ErrUpstreamFailed, err)
+	}
+}
+
+func (router *Router) Export(ctx context.Context, req *collectortrace.ExportTraceServiceRequest) (*collectortrace.ExportTraceServiceResponse, error) {
+	apiKey, datasetName, err := getAPIKeyDatasetAndTokenFromMetadata(ctx)
+	if err != nil {
+		return nil, status.Error(codes.Unauthenticated, err.Error())
+	}
+
+	if err := processTraceRequest(ctx, router, req, apiKey, datasetName); err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	return &collectortrace.ExportTraceServiceResponse{}, nil
+}
+
+func processTraceRequest(
+	ctx context.Context,
+	router *Router,
+	request *collectortrace.ExportTraceServiceRequest,
+	apiKey string,
+	datasetName string) error {
+
+	var requestID types.RequestIDContextKey
+	debugLog := router.iopLogger.Debug().WithField("request_id", requestID)
+
+	apiHost, err := router.Config.GetHoneycombAPI()
+	if err != nil {
+		router.Logger.Error().Logf("Unable to retrieve APIHost from config while processing OTLP batch")
+		return err
+	}
+
+	for _, resourceSpan := range request.ResourceSpans {
+		resourceAttrs := make(map[string]interface{})
+
+		if resourceSpan.Resource != nil {
+			addAttributesToMap(resourceAttrs, resourceSpan.Resource.Attributes)
+		}
+
+		for _, librarySpan := range resourceSpan.InstrumentationLibrarySpans {
+			library := librarySpan.InstrumentationLibrary
+			if library != nil {
+				if len(library.Name) > 0 {
+					resourceAttrs["library.name"] = library.Name
+				}
+				if len(library.Version) > 0 {
+					resourceAttrs["library.version"] = library.Version
+				}
+			}
+
+			for _, span := range librarySpan.GetSpans() {
+				traceID := bytesToTraceID(span.TraceId)
+				spanID := hex.EncodeToString(span.SpanId)
+				timestamp := time.Unix(0, int64(span.StartTimeUnixNano)).UTC()
+
+				eventAttrs := map[string]interface{}{
+					"trace.trace_id": traceID,
+					"trace.span_id":  spanID,
+					"type":           getSpanKind(span.Kind),
+					"name":           span.Name,
+					"duration_ms":    float64(span.EndTimeUnixNano-span.StartTimeUnixNano) / float64(time.Millisecond),
+					"status_code":    int32(getSpanStatusCode(span.Status)),
+				}
+				if span.ParentSpanId != nil {
+					eventAttrs["trace.parent_id"] = hex.EncodeToString(span.ParentSpanId)
+				}
+				if getSpanStatusCode(span.Status) == trace.Status_STATUS_CODE_ERROR {
+					eventAttrs["error"] = true
+				}
+				if span.Status != nil && len(span.Status.Message) > 0 {
+					eventAttrs["status_message"] = span.Status.Message
+				}
+				if span.Attributes != nil {
+					addAttributesToMap(eventAttrs, span.Attributes)
+				}
+
+				sampleRate, err := getSampleRateFromAttributes(eventAttrs)
+				if err != nil {
+					debugLog.WithField("error", err.Error()).WithField("sampleRate", eventAttrs["sampleRate"]).Logf("error parsing sampleRate")
+				}
+
+				// copy resource attributes to event attributes
+				for k, v := range resourceAttrs {
+					eventAttrs[k] = v
+				}
+
+				events := make([]*types.Event, 0, 1+len(span.Events)+len(span.Links))
+				events = append(events, &types.Event{
+					Context:    ctx,
+					APIHost:    apiHost,
+					APIKey:     apiKey,
+					Dataset:    datasetName,
+					SampleRate: uint(sampleRate),
+					Timestamp:  timestamp,
+					Data:       eventAttrs,
+				})
+
+				for _, sevent := range span.Events {
+					timestamp := time.Unix(0, int64(sevent.TimeUnixNano)).UTC()
+					attrs := map[string]interface{}{
+						"trace.trace_id":       traceID,
+						"trace.parent_id":      spanID,
+						"name":                 sevent.Name,
+						"parent_name":          span.Name,
+						"meta.annotation_type": "span_event",
+					}
+
+					if sevent.Attributes != nil {
+						addAttributesToMap(attrs, sevent.Attributes)
+					}
+					for k, v := range resourceAttrs {
+						attrs[k] = v
+					}
+					sampleRate, err := getSampleRateFromAttributes(attrs)
+					if err != nil {
+						debugLog.
+							WithField("error", err.Error()).
+							WithField("sampleRate", attrs["sampleRate"]).
+							Logf("error parsing sampleRate")
+					}
+					events = append(events, &types.Event{
+						Context:    ctx,
+						APIHost:    apiHost,
+						APIKey:     apiKey,
+						Dataset:    datasetName,
+						SampleRate: uint(sampleRate),
+						Timestamp:  timestamp,
+						Data:       attrs,
+					})
+				}
+
+				for _, slink := range span.Links {
+					attrs := map[string]interface{}{
+						"trace.trace_id":       traceID,
+						"trace.parent_id":      spanID,
+						"trace.link.trace_id":  bytesToTraceID(slink.TraceId),
+						"trace.link.span_id":   hex.EncodeToString(slink.SpanId),
+						"parent_name":          span.Name,
+						"meta.annotation_type": "link",
+					}
+
+					if slink.Attributes != nil {
+						addAttributesToMap(attrs, slink.Attributes)
+					}
+					for k, v := range resourceAttrs {
+						attrs[k] = v
+					}
+					sampleRate, err := getSampleRateFromAttributes(attrs)
+					if err != nil {
+						debugLog.
+							WithField("error", err.Error()).
+							WithField("sampleRate", attrs["sampleRate"]).
+							Logf("error parsing sampleRate")
+					}
+					events = append(events, &types.Event{
+						Context:    ctx,
+						APIHost:    apiHost,
+						APIKey:     apiKey,
+						Dataset:    datasetName,
+						SampleRate: uint(sampleRate),
+						Timestamp:  time.Time{}, //links don't have timestamps, so use empty time
+						Data:       attrs,
+					})
+				}
+
+				for _, evt := range events {
+					err = router.processEvent(evt, requestID)
+					if err != nil {
+						router.Logger.Error().Logf("Error processing event: " + err.Error())
+					}
+				}
+
+			}
+		}
+	}
+
+	return nil
+}
+
+func addAttributesToMap(attrs map[string]interface{}, attributes []*common.KeyValue) {
+	for _, attr := range attributes {
+		if attr.Key == "" {
+			continue
+		}
+		switch attr.Value.Value.(type) {
+		case *common.AnyValue_StringValue:
+			attrs[attr.Key] = attr.Value.GetStringValue()
+		case *common.AnyValue_BoolValue:
+			attrs[attr.Key] = attr.Value.GetBoolValue()
+		case *common.AnyValue_DoubleValue:
+			attrs[attr.Key] = attr.Value.GetDoubleValue()
+		case *common.AnyValue_IntValue:
+			attrs[attr.Key] = attr.Value.GetIntValue()
+		}
+	}
+}
+
+func getSpanKind(kind trace.Span_SpanKind) string {
+	switch kind {
+	case trace.Span_SPAN_KIND_CLIENT:
+		return "client"
+	case trace.Span_SPAN_KIND_SERVER:
+		return "server"
+	case trace.Span_SPAN_KIND_PRODUCER:
+		return "producer"
+	case trace.Span_SPAN_KIND_CONSUMER:
+		return "consumer"
+	case trace.Span_SPAN_KIND_INTERNAL:
+		return "internal"
+	case trace.Span_SPAN_KIND_UNSPECIFIED:
+		fallthrough
+	default:
+		return "unspecified"
+	}
+}
+
+// bytesToTraceID returns an ID suitable for use for spans and traces. Before
+// encoding the bytes as a hex string, we want to handle cases where we are
+// given 128-bit IDs with zero padding, e.g. 0000000000000000f798a1e7f33c8af6.
+// To do this, we borrow a strategy from Jaeger [1] wherein we split the byte
+// sequence into two parts. The leftmost part could contain all zeros. We use
+// that to determine whether to return a 64-bit hex encoded string or a 128-bit
+// one.
+//
+// [1]: https://github.com/jaegertracing/jaeger/blob/cd19b64413eca0f06b61d92fe29bebce1321d0b0/model/ids.go#L81
+func bytesToTraceID(traceID []byte) string {
+	// binary.BigEndian.Uint64() does a bounds check on traceID which will
+	// cause a panic if traceID is fewer than 8 bytes. In this case, we don't
+	// need to check for zero padding on the high part anyway, so just return a
+	// hex string.
+	if len(traceID) < traceIDShortLength {
+		return fmt.Sprintf("%x", traceID)
+	}
+	var low uint64
+	if len(traceID) == traceIDLongLength {
+		low = binary.BigEndian.Uint64(traceID[traceIDShortLength:])
+		if high := binary.BigEndian.Uint64(traceID[:traceIDShortLength]); high != 0 {
+			return fmt.Sprintf("%016x%016x", high, low)
+		}
+	} else {
+		low = binary.BigEndian.Uint64(traceID)
+	}
+
+	return fmt.Sprintf("%016x", low)
+}
+
+// getSpanStatusCode checks the value of both the deprecated code and code fields
+// on the span status and using the rules specified in the backward compatibility
+// notes in the protobuf definitions. See:
+//
+// https://github.com/open-telemetry/opentelemetry-proto/blob/59c488bfb8fb6d0458ad6425758b70259ff4a2bd/opentelemetry/proto/trace/v1/trace.proto#L230
+func getSpanStatusCode(status *trace.Status) trace.Status_StatusCode {
+	if status == nil {
+		return trace.Status_STATUS_CODE_UNSET
+	}
+	if status.Code == trace.Status_STATUS_CODE_UNSET {
+		if status.DeprecatedCode == trace.Status_DEPRECATED_STATUS_CODE_OK {
+			return trace.Status_STATUS_CODE_UNSET
+		}
+		return trace.Status_STATUS_CODE_ERROR
+	}
+	return status.Code
+}
+
+func getAPIKeyDatasetAndTokenFromMetadata(ctx context.Context) (
+	apiKey string,
+	datasetName string,
+	err error) {
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		apiKey = getValueFromMetadata(md, "x-honeycomb-team")
+		datasetName = getValueFromMetadata(md, "x-honeycomb-dataset")
+	}
+
+	if err := validateHeaders(apiKey, datasetName); err != nil {
+		return "", "", err
+	}
+	return apiKey, datasetName, nil
+}
+
+func getValueFromMetadata(md metadata.MD, key string) string {
+	if vals := md.Get(key); len(vals) > 0 {
+		return vals[0]
+	}
+	return ""
+}
+
+func getAPIKeyDatasetAndTokenFromHttpHeaders(r *http.Request) (
+	apiKey string,
+	datasetName string,
+	err error) {
+	apiKey = r.Header.Get("x-honeycomb-team")
+	datasetName = r.Header.Get("x-honeycomb-dataset")
+
+	if err := validateHeaders(apiKey, datasetName); err != nil {
+		return "", "", err
+	}
+	return apiKey, datasetName, nil
+}
+
+func validateHeaders(apiKey string, datasetName string) error {
+	if apiKey == "" {
+		return errors.New("missing x-honeycomb-team header")
+	}
+	if datasetName == "" {
+		return errors.New("missing x-honeycomb-team header")
+	}
+	return nil
+}
+
+func parseOTLPBody(r *http.Request, zstdDecoders chan *zstd.Decoder) (request *collectortrace.ExportTraceServiceRequest, cleanup func(), err error) {
+	cleanup = func() { /* empty cleanup */ }
+	bodyBytes, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return nil, cleanup, err
+	}
+	bodyReader := bytes.NewReader(bodyBytes)
+
+	var reader io.Reader
+	switch r.Header.Get("Content-Encoding") {
+	case "gzip":
+		var err error
+		reader, err = gzip.NewReader(bodyReader)
+		if err != nil {
+			return nil, cleanup, err
+		}
+	case "zstd":
+		zReader := <-zstdDecoders
+		cleanup = func() {
+			zReader.Reset(nil)
+			zstdDecoders <- zReader
+		}
+
+		err = zReader.Reset(bodyReader)
+		if err != nil {
+			return nil, cleanup, err
+		}
+
+		reader = zReader
+	default:
+		reader = bodyReader
+	}
+
+	bytes, err := ioutil.ReadAll(reader)
+	if err != nil {
+		return nil, cleanup, err
+	}
+
+	otlpRequet := &collectortrace.ExportTraceServiceRequest{}
+	err = proto.Unmarshal(bytes, otlpRequet)
+	if err != nil {
+		return nil, cleanup, err
+	}
+
+	return otlpRequet, cleanup, nil
+}
diff --git a/route/route.go b/route/route.go
index 437a7c1..6597318 100644
--- a/route/route.go
+++ b/route/route.go
@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
-	"encoding/binary"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -15,7 +13,6 @@ import (
 	"net"
 	"net/http"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 
@@ -39,8 +36,6 @@ import (
 	"github.com/honeycombio/refinery/types"
 
 	collectortrace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/collector/trace/v1"
-	common "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/common/v1"
-	trace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/trace/v1"
 )
 
 const (
@@ -161,6 +156,13 @@ func (r *Router) LnS(incomingOrPeer string) {
 	authedMuxxer.HandleFunc("/events/{datasetName}", r.event).Name("event")
 	authedMuxxer.HandleFunc("/batch/{datasetName}", r.batch).Name("batch")
 
+	// require an auth header for OTLP requests
+	otlpMuxxer := muxxer.PathPrefix("/v1/").Methods("POST").Subrouter()
+	otlpMuxxer.Use(r.apiKeyChecker)
+
+	// handle OTLP trace requests
+	otlpMuxxer.HandleFunc("/traces", r.postOTLP).Name("otlp")
+
 	// pass everything else through unmolested
 	muxxer.PathPrefix("/").HandlerFunc(r.proxy).Name("proxy")
 
@@ -386,190 +388,6 @@ func (r *Router) batch(w http.ResponseWriter, req *http.Request) {
 	w.Write(response)
 }
 
-func (r *Router) Export(ctx context.Context, req *collectortrace.ExportTraceServiceRequest) (*collectortrace.ExportTraceServiceResponse, error) {
-
-	md, ok := metadata.FromIncomingContext(ctx)
-	if !ok {
-		r.Logger.Error().Logf("Unable to retreive metadata from OTLP request.")
-		return &collectortrace.ExportTraceServiceResponse{}, nil
-	}
-
-	// requestID is used to track a requst as it moves between refinery nodes (peers)
-	// the OTLP handler only receives incoming (not peer) requests for now so will be empty here
-	var requestID types.RequestIDContextKey
-	debugLog := r.iopLogger.Debug().WithField("request_id", requestID)
-
-	apiKey, dataset := getAPIKeyAndDatasetFromMetadata(md)
-	if apiKey == "" {
-		r.Logger.Error().Logf("Received OTLP request without Honeycomb APIKey header")
-		return &collectortrace.ExportTraceServiceResponse{}, nil
-	}
-	if dataset == "" {
-		r.Logger.Error().Logf("Received OTLP request without Honeycomb dataset header")
-		return &collectortrace.ExportTraceServiceResponse{}, nil
-	}
-
-	apiHost, err := r.Config.GetHoneycombAPI()
-	if err != nil {
-		r.Logger.Error().Logf("Unable to retrieve APIHost from config while processing OTLP batch")
-		return &collectortrace.ExportTraceServiceResponse{}, nil
-	}
-
-	var grpcRequestEncoding string
-	if val := md.Get("grpc-accept-encoding"); val != nil {
-		grpcRequestEncoding = strings.Join(val, ",")
-	}
-
-	for _, resourceSpan := range req.ResourceSpans {
-		resourceAttrs := make(map[string]interface{})
-
-		if resourceSpan.Resource != nil {
-			addAttributesToMap(resourceAttrs, resourceSpan.Resource.Attributes)
-		}
-
-		for _, librarySpan := range resourceSpan.InstrumentationLibrarySpans {
-			library := librarySpan.InstrumentationLibrary
-			if library != nil {
-				if len(library.Name) > 0 {
-					resourceAttrs["library.name"] = library.Name
-				}
-				if len(library.Version) > 0 {
-					resourceAttrs["library.version"] = library.Version
-				}
-			}
-
-			for _, span := range librarySpan.GetSpans() {
-				traceID := bytesToTraceID(span.TraceId)
-				spanID := hex.EncodeToString(span.SpanId)
-				timestamp := time.Unix(0, int64(span.StartTimeUnixNano)).UTC()
-
-				eventAttrs := map[string]interface{}{
-					"trace.trace_id": traceID,
-					"trace.span_id":  spanID,
-					"type":           getSpanKind(span.Kind),
-					"name":           span.Name,
-					"duration_ms":    float64(span.EndTimeUnixNano-span.StartTimeUnixNano) / float64(time.Millisecond),
-					"status_code":    int32(r.getSpanStatusCode(span.Status)),
-				}
-				if span.ParentSpanId != nil {
-					eventAttrs["trace.parent_id"] = hex.EncodeToString(span.ParentSpanId)
-				}
-				if r.getSpanStatusCode(span.Status) == trace.Status_STATUS_CODE_ERROR {
-					eventAttrs["error"] = true
-				}
-				if span.Status != nil && len(span.Status.Message) > 0 {
-					eventAttrs["status_message"] = span.Status.Message
-				}
-				if span.Attributes != nil {
-					addAttributesToMap(eventAttrs, span.Attributes)
-				}
-				if grpcRequestEncoding != "" {
-					eventAttrs["grpc_request_encoding"] = grpcRequestEncoding
-				}
-
-				sampleRate, err := getSampleRateFromAttributes(eventAttrs)
-				if err != nil {
-					debugLog.WithField("error", err.Error()).WithField("sampleRate", eventAttrs["sampleRate"]).Logf("error parsing sampleRate")
-				}
-
-				// copy resource attributes to event attributes
-				for k, v := range resourceAttrs {
-					eventAttrs[k] = v
-				}
-
-				events := make([]*types.Event, 0, 1+len(span.Events)+len(span.Links))
-				events = append(events, &types.Event{
-					Context:    ctx,
-					APIHost:    apiHost,
-					APIKey:     apiKey,
-					Dataset:    dataset,
-					SampleRate: uint(sampleRate),
-					Timestamp:  timestamp,
-					Data:       eventAttrs,
-				})
-
-				for _, sevent := range span.Events {
-					timestamp := time.Unix(0, int64(sevent.TimeUnixNano)).UTC()
-					attrs := map[string]interface{}{
-						"trace.trace_id":       traceID,
-						"trace.parent_id":      spanID,
-						"name":                 sevent.Name,
-						"parent_name":          span.Name,
-						"meta.annotation_type": "span_event",
-					}
-
-					if sevent.Attributes != nil {
-						addAttributesToMap(attrs, sevent.Attributes)
-					}
-					for k, v := range resourceAttrs {
-						attrs[k] = v
-					}
-					sampleRate, err := getSampleRateFromAttributes(attrs)
-					if err != nil {
-						debugLog.
-							WithField("error", err.Error()).
-							WithField("sampleRate", attrs["sampleRate"]).
-							Logf("error parsing sampleRate")
-					}
-					events = append(events, &types.Event{
-						Context:    ctx,
-						APIHost:    apiHost,
-						APIKey:     apiKey,
-						Dataset:    dataset,
-						SampleRate: uint(sampleRate),
-						Timestamp:  timestamp,
-						Data:       attrs,
-					})
-				}
-
-				for _, slink := range span.Links {
-					attrs := map[string]interface{}{
-						"trace.trace_id":       traceID,
-						"trace.parent_id":      spanID,
-						"trace.link.trace_id":  bytesToTraceID(slink.TraceId),
-						"trace.link.span_id":   hex.EncodeToString(slink.SpanId),
-						"parent_name":          span.Name,
-						"meta.annotation_type": "link",
-					}
-
-					if slink.Attributes != nil {
-						addAttributesToMap(attrs, slink.Attributes)
-					}
-					for k, v := range resourceAttrs {
-						attrs[k] = v
-					}
-					sampleRate, err := getSampleRateFromAttributes(attrs)
-					if err != nil {
-						debugLog.
-							WithField("error", err.Error()).
-							WithField("sampleRate", attrs["sampleRate"]).
-							Logf("error parsing sampleRate")
-					}
-					events = append(events, &types.Event{
-						Context:    ctx,
-						APIHost:    apiHost,
-						APIKey:     apiKey,
-						Dataset:    dataset,
-						SampleRate: uint(sampleRate),
-						Timestamp:  time.Time{}, //links don't have timestamps, so use empty time
-						Data:       attrs,
-					})
-				}
-
-				for _, evt := range events {
-					err = r.processEvent(evt, requestID)
-					if err != nil {
-						r.Logger.Error().Logf("Error processing event: " + err.Error())
-					}
-				}
-
-			}
-		}
-	}
-
-	return &collectortrace.ExportTraceServiceResponse{}, nil
-}
-
 func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 	debugLog := r.iopLogger.Debug().
 		WithField("request_id", reqID).
@@ -806,91 +624,6 @@ func getFirstValueFromMetadata(key string, md metadata.MD) string {
 	return ""
 }
 
-func addAttributesToMap(attrs map[string]interface{}, attributes []*common.KeyValue) {
-	for _, attr := range attributes {
-		if attr.Key == "" {
-			continue
-		}
-		switch attr.Value.Value.(type) {
-		case *common.AnyValue_StringValue:
-			attrs[attr.Key] = attr.Value.GetStringValue()
-		case *common.AnyValue_BoolValue:
-			attrs[attr.Key] = attr.Value.GetBoolValue()
-		case *common.AnyValue_DoubleValue:
-			attrs[attr.Key] = attr.Value.GetDoubleValue()
-		case *common.AnyValue_IntValue:
-			attrs[attr.Key] = attr.Value.GetIntValue()
-		}
-	}
-}
-
-func getSpanKind(kind trace.Span_SpanKind) string {
-	switch kind {
-	case trace.Span_SPAN_KIND_CLIENT:
-		return "client"
-	case trace.Span_SPAN_KIND_SERVER:
-		return "server"
-	case trace.Span_SPAN_KIND_PRODUCER:
-		return "producer"
-	case trace.Span_SPAN_KIND_CONSUMER:
-		return "consumer"
-	case trace.Span_SPAN_KIND_INTERNAL:
-		return "internal"
-	case trace.Span_SPAN_KIND_UNSPECIFIED:
-		fallthrough
-	default:
-		return "unspecified"
-	}
-}
-
-// bytesToTraceID returns an ID suitable for use for spans and traces. Before
-// encoding the bytes as a hex string, we want to handle cases where we are
-// given 128-bit IDs with zero padding, e.g. 0000000000000000f798a1e7f33c8af6.
-// To do this, we borrow a strategy from Jaeger [1] wherein we split the byte
-// sequence into two parts. The leftmost part could contain all zeros. We use
-// that to determine whether to return a 64-bit hex encoded string or a 128-bit
-// one.
-//
-// [1]: https://github.com/jaegertracing/jaeger/blob/cd19b64413eca0f06b61d92fe29bebce1321d0b0/model/ids.go#L81
-func bytesToTraceID(traceID []byte) string {
-	// binary.BigEndian.Uint64() does a bounds check on traceID which will
-	// cause a panic if traceID is fewer than 8 bytes. In this case, we don't
-	// need to check for zero padding on the high part anyway, so just return a
-	// hex string.
-	if len(traceID) < traceIDShortLength {
-		return fmt.Sprintf("%x", traceID)
-	}
-	var low uint64
-	if len(traceID) == traceIDLongLength {
-		low = binary.BigEndian.Uint64(traceID[traceIDShortLength:])
-		if high := binary.BigEndian.Uint64(traceID[:traceIDShortLength]); high != 0 {
-			return fmt.Sprintf("%016x%016x", high, low)
-		}
-	} else {
-		low = binary.BigEndian.Uint64(traceID)
-	}
-
-	return fmt.Sprintf("%016x", low)
-}
-
-// getSpanStatusCode checks the value of both the deprecated code and code fields
-// on the span status and using the rules specified in the backward compatibility
-// notes in the protobuf definitions. See:
-//
-// https://github.com/open-telemetry/opentelemetry-proto/blob/59c488bfb8fb6d0458ad6425758b70259ff4a2bd/opentelemetry/proto/trace/v1/trace.proto#L230
-func (r *Router) getSpanStatusCode(status *trace.Status) trace.Status_StatusCode {
-	if status == nil {
-		return trace.Status_STATUS_CODE_UNSET
-	}
-	if status.Code == trace.Status_STATUS_CODE_UNSET {
-		if status.DeprecatedCode == trace.Status_DEPRECATED_STATUS_CODE_OK {
-			return trace.Status_STATUS_CODE_UNSET
-		}
-		return trace.Status_STATUS_CODE_ERROR
-	}
-	return status.Code
-}
-
 func getSampleRateFromAttributes(attrs map[string]interface{}) (int, error) {
 	var sampleRateKey string
 	if attrs["sampleRate"] != nil {


Test patch:
diff --git a/route/otlp_trace_test.go b/route/otlp_trace_test.go
new file mode 100644
index 0000000..2354b29
--- /dev/null
+++ b/route/otlp_trace_test.go
@@ -0,0 +1,348 @@
+package route
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/hex"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/honeycombio/refinery/config"
+	collectortrace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/collector/trace/v1"
+	common "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/common/v1"
+	trace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/trace/v1"
+	"github.com/honeycombio/refinery/logger"
+	"github.com/honeycombio/refinery/metrics"
+	"github.com/honeycombio/refinery/transmit"
+	"github.com/klauspost/compress/zstd"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/metadata"
+)
+
+func TestOTLPHandler(t *testing.T) {
+	md := metadata.New(map[string]string{"x-honeycomb-team": "meow", "x-honeycomb-dataset": "ds"})
+	ctx := metadata.NewIncomingContext(context.Background(), md)
+
+	mockMetrics := metrics.MockMetrics{}
+	mockMetrics.Start()
+	mockTransmission := &transmit.MockTransmission{}
+	mockTransmission.Start()
+	decoders, err := makeDecoders(1)
+	if err != nil {
+		t.Error(err)
+	}
+	router := &Router{
+		Config:               &config.MockConfig{},
+		Metrics:              &mockMetrics,
+		UpstreamTransmission: mockTransmission,
+		iopLogger: iopLogger{
+			Logger:         &logger.MockLogger{},
+			incomingOrPeer: "incoming",
+		},
+		Logger:       &logger.MockLogger{},
+		zstdDecoders: decoders,
+	}
+
+	conf := &config.MockConfig{
+		GetSendDelayVal:    0,
+		GetTraceTimeoutVal: 60 * time.Second,
+		GetSamplerTypeVal:  &config.DeterministicSamplerConfig{SampleRate: 1},
+		SendTickerVal:      2 * time.Millisecond,
+		GetInMemoryCollectorCacheCapacityVal: config.InMemoryCollectorCacheCapacity{
+			CacheCapacity: 100,
+			MaxAlloc:      100,
+		},
+	}
+
+	t.Run("span with status", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: helperOTLPRequestSpansWithStatus(),
+				}},
+			}},
+		}
+		_, err := router.Export(ctx, req)
+		if err != nil {
+			t.Errorf(`Unexpected error: %s`, err)
+		}
+		assert.Equal(t, 2, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+
+	t.Run("span without status", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: helperOTLPRequestSpansWithoutStatus(),
+				}},
+			}},
+		}
+		_, err := router.Export(ctx, req)
+		if err != nil {
+			t.Errorf(`Unexpected error: %s`, err)
+		}
+		assert.Equal(t, 2, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+
+	// TODO: (MG) figuure out how we can test JSON created from OTLP requests
+	// Below is example, but requires significant usage of collector, sampler, conf, etc
+	t.Run("creates events for span events", func(t *testing.T) {
+		t.Skip("need additional work to support inspecting outbound JSON")
+
+		traceID := []byte{0, 0, 0, 0, 1}
+		spanID := []byte{1, 0, 0, 0, 0}
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: []*trace.Span{{
+						TraceId: traceID,
+						SpanId:  spanID,
+						Name:    "span_with_event",
+						Events: []*trace.Span_Event{{
+							TimeUnixNano: 12345,
+							Name:         "span_link",
+							Attributes: []*common.KeyValue{{
+								Key: "event_attr_key", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "event_attr_val"}},
+							}},
+						}},
+					}},
+				}},
+			}},
+		}
+		_, err := router.Export(ctx, req)
+		if err != nil {
+			t.Errorf(`Unexpected error: %s`, err)
+		}
+
+		time.Sleep(conf.SendTickerVal * 2)
+
+		mockTransmission.Mux.Lock()
+		assert.Equal(t, 2, len(mockTransmission.Events))
+
+		spanEvent := mockTransmission.Events[0]
+		// assert.Equal(t, time.Unix(0, int64(12345)).UTC(), spanEvent.Timestamp)
+		assert.Equal(t, bytesToTraceID(traceID), spanEvent.Data["trace.trace_id"])
+		assert.Equal(t, hex.EncodeToString(spanID), spanEvent.Data["trace.span_id"])
+		assert.Equal(t, "span_link", spanEvent.Data["span.name"])
+		assert.Equal(t, "span_with_event", spanEvent.Data["parent.name"])
+		assert.Equal(t, "span_event", spanEvent.Data["meta.annotation_type"])
+		assert.Equal(t, "event_attr_key", spanEvent.Data["event_attr_val"])
+		mockTransmission.Mux.Unlock()
+		mockTransmission.Flush()
+	})
+
+	t.Run("creates events for span links", func(t *testing.T) {
+		t.Skip("need additional work to support inspecting outbound JSON")
+
+		traceID := []byte{0, 0, 0, 0, 1}
+		spanID := []byte{1, 0, 0, 0, 0}
+		linkTraceID := []byte{0, 0, 0, 0, 2}
+		linkSpanID := []byte{2, 0, 0, 0, 0}
+
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: []*trace.Span{{
+						Name:    "span_with_link",
+						TraceId: traceID,
+						SpanId:  spanID,
+						Links: []*trace.Span_Link{{
+							TraceId:    traceID,
+							SpanId:     spanID,
+							TraceState: "link_trace_state",
+							Attributes: []*common.KeyValue{{
+								Key: "link_attr_key", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "link_attr_val"}},
+							}},
+						}},
+					}},
+				}},
+			}},
+		}
+		_, err := router.Export(ctx, req)
+		if err != nil {
+			t.Errorf(`Unexpected error: %s`, err)
+		}
+
+		time.Sleep(conf.SendTickerVal * 2)
+		assert.Equal(t, 2, len(mockTransmission.Events))
+
+		spanLink := mockTransmission.Events[1]
+		assert.Equal(t, bytesToTraceID(traceID), spanLink.Data["trace.trace_id"])
+		assert.Equal(t, hex.EncodeToString(spanID), spanLink.Data["trace.span_id"])
+		assert.Equal(t, bytesToTraceID(linkTraceID), spanLink.Data["trace.link.trace_id"])
+		assert.Equal(t, hex.EncodeToString(linkSpanID), spanLink.Data["trace.link.span_id"])
+		assert.Equal(t, "link", spanLink.Data["meta.annotation_type"])
+		assert.Equal(t, "link_attr_val", spanLink.Data["link_attr_key"])
+		mockTransmission.Flush()
+	})
+
+	t.Run("can receive OTLP over HTTP/protobuf", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: helperOTLPRequestSpansWithStatus(),
+				}},
+			}},
+		}
+		body, err := proto.Marshal(req)
+		if err != nil {
+			t.Error(err)
+		}
+
+		request, _ := http.NewRequest("POST", "/v1/traces", strings.NewReader(string(body)))
+		request.Header = http.Header{}
+		request.Header.Set("content-type", "application/protobuf")
+		request.Header.Set("x-honeycomb-team", "apikey")
+		request.Header.Set("x-honeycomb-dataset", "dataset")
+
+		w := httptest.NewRecorder()
+		router.postOTLP(w, request)
+		assert.Equal(t, w.Code, http.StatusOK)
+
+		assert.Equal(t, 2, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+
+	t.Run("can receive OTLP over HTTP/protobuf with gzip encoding", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: helperOTLPRequestSpansWithStatus(),
+				}},
+			}},
+		}
+		body, err := proto.Marshal(req)
+		if err != nil {
+			t.Error(err)
+		}
+
+		buf := new(bytes.Buffer)
+		writer := gzip.NewWriter(buf)
+		writer.Write(body)
+		writer.Close()
+		if err != nil {
+			t.Error(err)
+		}
+
+		request, _ := http.NewRequest("POST", "/v1/traces", strings.NewReader(buf.String()))
+		request.Header = http.Header{}
+		request.Header.Set("content-type", "application/protobuf")
+		request.Header.Set("content-encoding", "gzip")
+		request.Header.Set("x-honeycomb-team", "apikey")
+		request.Header.Set("x-honeycomb-dataset", "dataset")
+
+		w := httptest.NewRecorder()
+		router.postOTLP(w, request)
+		assert.Equal(t, w.Code, http.StatusOK)
+
+		assert.Equal(t, 2, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+
+	t.Run("can receive OTLP over HTTP/protobuf with zstd encoding", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+					Spans: helperOTLPRequestSpansWithStatus(),
+				}},
+			}},
+		}
+		body, err := proto.Marshal(req)
+		if err != nil {
+			t.Error(err)
+		}
+
+		buf := new(bytes.Buffer)
+		writer, err := zstd.NewWriter(buf)
+		if err != nil {
+			t.Error(err)
+		}
+		writer.Write(body)
+		writer.Close()
+		if err != nil {
+			t.Error(err)
+		}
+
+		request, _ := http.NewRequest("POST", "/v1/traces", strings.NewReader(buf.String()))
+		request.Header = http.Header{}
+		request.Header.Set("content-type", "application/protobuf")
+		request.Header.Set("content-encoding", "zstd")
+		request.Header.Set("x-honeycomb-team", "apikey")
+		request.Header.Set("x-honeycomb-dataset", "dataset")
+
+		w := httptest.NewRecorder()
+		router.postOTLP(w, request)
+		assert.Equal(t, w.Code, http.StatusOK)
+
+		assert.Equal(t, 2, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+
+	t.Run("rejects OTLP over HTTP/JSON ", func(t *testing.T) {
+		request, _ := http.NewRequest("POST", "/v1/traces", strings.NewReader("{}"))
+		request.Header = http.Header{}
+		request.Header.Set("content-type", "application/json")
+		request.Header.Set("x-honeycomb-team", "apikey")
+		request.Header.Set("x-honeycomb-dataset", "dataset")
+
+		w := httptest.NewRecorder()
+		router.postOTLP(w, request)
+		assert.Equal(t, w.Code, http.StatusNotImplemented)
+		assert.Equal(t, `{"source":"refinery","error":"invalid content-type - only 'application/protobuf' is supported"}`, string(w.Body.String()))
+
+		assert.Equal(t, 0, len(mockTransmission.Events))
+		mockTransmission.Flush()
+	})
+}
+
+func helperOTLPRequestSpansWithoutStatus() []*trace.Span {
+	now := time.Now()
+	return []*trace.Span{
+		{
+			StartTimeUnixNano: uint64(now.UnixNano()),
+			Events: []*trace.Span_Event{
+				{
+					TimeUnixNano: uint64(now.UnixNano()),
+					Attributes: []*common.KeyValue{
+						{
+							Key: "attribute_key",
+							Value: &common.AnyValue{
+								Value: &common.AnyValue_StringValue{StringValue: "attribute_value"},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func helperOTLPRequestSpansWithStatus() []*trace.Span {
+	now := time.Now()
+	return []*trace.Span{
+		{
+			StartTimeUnixNano: uint64(now.UnixNano()),
+			Events: []*trace.Span_Event{
+				{
+					TimeUnixNano: uint64(now.UnixNano()),
+					Attributes: []*common.KeyValue{
+						{
+							Key: "attribute_key",
+							Value: &common.AnyValue{
+								Value: &common.AnyValue_StringValue{StringValue: "attribute_value"},
+							},
+						},
+					},
+				},
+			},
+			Status: &trace.Status{Code: trace.Status_STATUS_CODE_OK},
+		},
+	}
+}
diff --git a/route/route_test.go b/route/route_test.go
index 1522d7d..2080396 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -3,8 +3,6 @@ package route
 import (
 	"bytes"
 	"compress/gzip"
-	"context"
-	"encoding/hex"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -18,13 +16,9 @@ import (
 	"github.com/facebookgo/inject"
 	"github.com/honeycombio/refinery/collect"
 	"github.com/honeycombio/refinery/config"
-	collectortrace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/collector/trace/v1"
-	common "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/common/v1"
-	trace "github.com/honeycombio/refinery/internal/opentelemetry-proto-gen/trace/v1"
 	"github.com/honeycombio/refinery/logger"
 	"github.com/honeycombio/refinery/metrics"
 	"github.com/honeycombio/refinery/transmit"
-	"github.com/stretchr/testify/assert"
 
 	"github.com/gorilla/mux"
 	"github.com/honeycombio/refinery/sharder"
@@ -399,160 +393,6 @@ func TestDebugTrace(t *testing.T) {
 	}
 }
 
-func TestOTLPHandler(t *testing.T) {
-	md := metadata.New(map[string]string{"x-honeycomb-team": "meow", "x-honeycomb-dataset": "ds"})
-	ctx := metadata.NewIncomingContext(context.Background(), md)
-
-	mockMetrics := metrics.MockMetrics{}
-	mockMetrics.Start()
-	mockTransmission := &transmit.MockTransmission{}
-	mockTransmission.Start()
-	router := &Router{
-		Config:               &config.MockConfig{},
-		Metrics:              &mockMetrics,
-		UpstreamTransmission: mockTransmission,
-		iopLogger: iopLogger{
-			Logger:         &logger.NullLogger{},
-			incomingOrPeer: "incoming",
-		},
-	}
-
-	conf := &config.MockConfig{
-		GetSendDelayVal:    0,
-		GetTraceTimeoutVal: 60 * time.Second,
-		GetSamplerTypeVal:  &config.DeterministicSamplerConfig{SampleRate: 1},
-		SendTickerVal:      2 * time.Millisecond,
-		GetInMemoryCollectorCacheCapacityVal: config.InMemoryCollectorCacheCapacity{
-			CacheCapacity: 100,
-			MaxAlloc:      100,
-		},
-	}
-
-	t.Run("span with status", func(t *testing.T) {
-		req := &collectortrace.ExportTraceServiceRequest{
-			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
-					Spans: helperOTLPRequestSpansWithStatus(),
-				}},
-			}},
-		}
-		_, err := router.Export(ctx, req)
-		if err != nil {
-			t.Errorf(`Unexpected error: %s`, err)
-		}
-		assert.Equal(t, 2, len(mockTransmission.Events))
-		mockTransmission.Flush()
-	})
-
-	t.Run("span without status", func(t *testing.T) {
-		req := &collectortrace.ExportTraceServiceRequest{
-			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
-					Spans: helperOTLPRequestSpansWithoutStatus(),
-				}},
-			}},
-		}
-		_, err := router.Export(ctx, req)
-		if err != nil {
-			t.Errorf(`Unexpected error: %s`, err)
-		}
-		assert.Equal(t, 2, len(mockTransmission.Events))
-		mockTransmission.Flush()
-	})
-
-	// TODO: (MG) figuure out how we can test JSON created from OTLP requests
-	// Below is example, but requires significant usage of collector, sampler, conf, etc
-	t.Run("creates events for span events", func(t *testing.T) {
-		t.Skip("need additional work to support inspecting outbound JSON")
-
-		traceID := []byte{0, 0, 0, 0, 1}
-		spanID := []byte{1, 0, 0, 0, 0}
-		req := &collectortrace.ExportTraceServiceRequest{
-			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
-					Spans: []*trace.Span{{
-						TraceId: traceID,
-						SpanId:  spanID,
-						Name:    "span_with_event",
-						Events: []*trace.Span_Event{{
-							TimeUnixNano: 12345,
-							Name:         "span_link",
-							Attributes: []*common.KeyValue{{
-								Key: "event_attr_key", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "event_attr_val"}},
-							}},
-						}},
-					}},
-				}},
-			}},
-		}
-		_, err := router.Export(ctx, req)
-		if err != nil {
-			t.Errorf(`Unexpected error: %s`, err)
-		}
-
-		time.Sleep(conf.SendTickerVal * 2)
-
-		mockTransmission.Mux.Lock()
-		assert.Equal(t, 2, len(mockTransmission.Events))
-
-		spanEvent := mockTransmission.Events[0]
-		// assert.Equal(t, time.Unix(0, int64(12345)).UTC(), spanEvent.Timestamp)
-		assert.Equal(t, bytesToTraceID(traceID), spanEvent.Data["trace.trace_id"])
-		assert.Equal(t, hex.EncodeToString(spanID), spanEvent.Data["trace.span_id"])
-		assert.Equal(t, "span_link", spanEvent.Data["span.name"])
-		assert.Equal(t, "span_with_event", spanEvent.Data["parent.name"])
-		assert.Equal(t, "span_event", spanEvent.Data["meta.annotation_type"])
-		assert.Equal(t, "event_attr_key", spanEvent.Data["event_attr_val"])
-		mockTransmission.Mux.Unlock()
-		mockTransmission.Flush()
-	})
-
-	t.Run("creates events for span links", func(t *testing.T) {
-		t.Skip("need additional work to support inspecting outbound JSON")
-
-		traceID := []byte{0, 0, 0, 0, 1}
-		spanID := []byte{1, 0, 0, 0, 0}
-		linkTraceID := []byte{0, 0, 0, 0, 2}
-		linkSpanID := []byte{2, 0, 0, 0, 0}
-
-		req := &collectortrace.ExportTraceServiceRequest{
-			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
-					Spans: []*trace.Span{{
-						Name:    "span_with_link",
-						TraceId: traceID,
-						SpanId:  spanID,
-						Links: []*trace.Span_Link{{
-							TraceId:    traceID,
-							SpanId:     spanID,
-							TraceState: "link_trace_state",
-							Attributes: []*common.KeyValue{{
-								Key: "link_attr_key", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "link_attr_val"}},
-							}},
-						}},
-					}},
-				}},
-			}},
-		}
-		_, err := router.Export(ctx, req)
-		if err != nil {
-			t.Errorf(`Unexpected error: %s`, err)
-		}
-
-		time.Sleep(conf.SendTickerVal * 2)
-		assert.Equal(t, 2, len(mockTransmission.Events))
-
-		spanLink := mockTransmission.Events[1]
-		assert.Equal(t, bytesToTraceID(traceID), spanLink.Data["trace.trace_id"])
-		assert.Equal(t, hex.EncodeToString(spanID), spanLink.Data["trace.span_id"])
-		assert.Equal(t, bytesToTraceID(linkTraceID), spanLink.Data["trace.link.trace_id"])
-		assert.Equal(t, hex.EncodeToString(linkSpanID), spanLink.Data["trace.link.span_id"])
-		assert.Equal(t, "link", spanLink.Data["meta.annotation_type"])
-		assert.Equal(t, "link_attr_val", spanLink.Data["link_attr_key"])
-		mockTransmission.Flush()
-	})
-}
-
 func TestDependencyInjection(t *testing.T) {
 	var g inject.Graph
 	err := g.Provide(
@@ -575,51 +415,6 @@ func TestDependencyInjection(t *testing.T) {
 	}
 }
 
-func helperOTLPRequestSpansWithoutStatus() []*trace.Span {
-	now := time.Now()
-	return []*trace.Span{
-		{
-			StartTimeUnixNano: uint64(now.UnixNano()),
-			Events: []*trace.Span_Event{
-				{
-					TimeUnixNano: uint64(now.UnixNano()),
-					Attributes: []*common.KeyValue{
-						{
-							Key: "attribute_key",
-							Value: &common.AnyValue{
-								Value: &common.AnyValue_StringValue{StringValue: "attribute_value"},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-}
-
-func helperOTLPRequestSpansWithStatus() []*trace.Span {
-	now := time.Now()
-	return []*trace.Span{
-		{
-			StartTimeUnixNano: uint64(now.UnixNano()),
-			Events: []*trace.Span_Event{
-				{
-					TimeUnixNano: uint64(now.UnixNano()),
-					Attributes: []*common.KeyValue{
-						{
-							Key: "attribute_key",
-							Value: &common.AnyValue{
-								Value: &common.AnyValue_StringValue{StringValue: "attribute_value"},
-							},
-						},
-					},
-				},
-			},
-			Status: &trace.Status{Code: trace.Status_STATUS_CODE_OK},
-		},
-	}
-}
-
 type TestSharder struct{}
 
 func (s *TestSharder) MyShard() sharder.Shard { return nil }