Repository: honeycombio/refinery (go)

Issue:
Refinery returns GRPC errors, but doesn't log them
**Is your feature request related to a problem? Please describe.**

Customer was sending batches of data larger than Refinery was configured to accept. While it did return errors to the sender, it didn't log anything. This is because Refinery uses Husky to accept OTLP traffic, and Husky doesn't have a model for callbacks or telemetry today.

**Describe the solution you'd like**

* 

- [x] Enable Husky to have an error callback or some other mechanism for returning a log message
- [ ] * Hook it into Refinery so that Refinery can log it (and maybe add a metric for GRPC incoming errors as well)

**Describe alternatives you've considered**

The _status quo_

**Additional context**

Inspired by a support ticket.


Pull request:
feat: Add basic telemetry to event, batch and OTLP endpoints
## Which problem is this PR solving?

We don't currently have any logging or telemetry for our endpoints that receive data (spans & logs). This PR adds tracing  to the following endpoints using OTel instrumentation libraries. It also configures husky to set appropriate validation error attributes.

Endpoints:
- /event
- /batch
- OTel trace (http & OTLP)
- OTel logs (http & OTLP)

Closes the following issue:
- Closes #1054 

## Short description of the changes
- Configure husky's telemetry callback to set validation attributes on existing spans
- Inject the already configured OTel Tracer into the Router
- Add otelhttp & otelgrpc instrumentation libraries and wire up
- Update tests to set a `NoopTracer` in the router

Example HTTP span
<img width="559" alt="image" src="https://github.com/user-attachments/assets/5cbba2f2-4925-45df-9e15-9489767bf944">

Example gRPC span
<img width="530" alt="image" src="https://github.com/user-attachments/assets/6c1a356c-ed07-410b-9678-1d3ed5d14c1f">

Patch:
diff --git a/cmd/refinery/main.go b/cmd/refinery/main.go
index 74d3c7f..3194df7 100644
--- a/cmd/refinery/main.go
+++ b/cmd/refinery/main.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"net/http"
@@ -10,6 +11,7 @@ import (
 	"syscall"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel/trace/noop"
 	_ "go.uber.org/automaxprocs"
@@ -19,6 +21,7 @@ import (
 	"github.com/facebookgo/inject"
 	"github.com/facebookgo/startstop"
 	"github.com/google/uuid"
+	"github.com/honeycombio/husky"
 	libhoney "github.com/honeycombio/libhoney-go"
 	"github.com/honeycombio/libhoney-go/transmission"
 	"github.com/jonboulle/clockwork"
@@ -237,8 +240,13 @@ func main() {
 	if c.GetOTelTracingConfig().Enabled {
 		// let's set up some OTel tracing
 		tracer, shutdown = otelutil.SetupTracing(c.GetOTelTracingConfig(), resourceLib, resourceVer)
-	}
 
+		// add telemetry callback so husky can enrich spans with attributes
+		husky.AddTelemetryAttributeFunc = func(ctx context.Context, key string, value any) {
+			span := trace.SpanFromContext(ctx)
+			span.SetAttributes(attribute.String(key, value.(string)))
+		}
+	}
 	defer shutdown()
 
 	// we need to include all the metrics types so we can inject them in case they're needed
diff --git a/go.mod b/go.mod
index 1abea9e..d38e7c6 100644
--- a/go.mod
+++ b/go.mod
@@ -33,6 +33,8 @@ require (
 	github.com/stretchr/testify v1.9.0
 	github.com/tidwall/gjson v1.18.0
 	github.com/vmihailenco/msgpack/v5 v5.4.1
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0
 	go.opentelemetry.io/otel v1.32.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.32.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.32.0
@@ -51,8 +53,6 @@ require (
 	gopkg.in/yaml.v3 v3.0.1
 )
 
-require golang.org/x/mod v0.22.0 // indirect
-
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
@@ -63,6 +63,7 @@ require (
 	github.com/facebookgo/limitgroup v0.0.0-20150612190941-6abd8d71ec01 // indirect
 	github.com/facebookgo/muster v0.0.0-20150708232844-fd3d7953fd52 // indirect
 	github.com/facebookgo/structtag v0.0.0-20150214074306-217e25fb9691 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/snappy v0.0.4
@@ -80,6 +81,7 @@ require (
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/mod v0.22.0 // indirect
 	golang.org/x/net v0.31.0 // indirect
 	golang.org/x/sys v0.27.0 // indirect
 	golang.org/x/text v0.20.0 // indirect
diff --git a/go.sum b/go.sum
index 7195379..c568d3c 100644
--- a/go.sum
+++ b/go.sum
@@ -45,6 +45,8 @@ github.com/facebookgo/structtag v0.0.0-20150214074306-217e25fb9691 h1:KnnwHN59Jx
 github.com/facebookgo/structtag v0.0.0-20150214074306-217e25fb9691/go.mod h1:sKLL1iua/0etWfo/nPCmyz+v2XDMXy+Ho53W7RAuZNY=
 github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 h1:7HZCaLC5+BZpmbhCOZJ293Lz68O7PYrF2EzeiFMwCLk=
 github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4/go.mod h1:5tD+neXqOorC30/tWg0LCSkrqj/AR6gu8yY8/fpw1q0=
+github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
+github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -144,6 +146,10 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
 github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 h1:yMkBS9yViCc7U7yeLzJPM2XizlfdVvBRSmsQDWu6qc0=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0/go.mod h1:n8MR6/liuGB5EmTETUBeU5ZgqMOlqKRxUaqPQBOANZ8=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 h1:UP6IpuHFkUgOQL9FFQFrZ+5LiwhhYRbi7VZSIx6Nj5s=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0/go.mod h1:qxuZLtbq5QDtdeSHsS7bcf6EH6uO6jUAgk764zd3rhM=
 go.opentelemetry.io/otel v1.32.0 h1:WnBN+Xjcteh0zdk01SVqV55d/m62NJLJdIyb4y/WO5U=
 go.opentelemetry.io/otel v1.32.0/go.mod h1:00DCVSB0RQcnzlwyTfqtxSm+DRr9hpYrHjNGiBHVQIg=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.32.0 h1:t/Qur3vKSkUCcDVaSumWF2PKHt85pc7fRvFuoVT8qFU=
diff --git a/route/otlp_logs.go b/route/otlp_logs.go
index 14ff31a..2a95821 100644
--- a/route/otlp_logs.go
+++ b/route/otlp_logs.go
@@ -6,6 +6,7 @@ import (
 	"net/http"
 
 	huskyotlp "github.com/honeycombio/husky/otlp"
+	"github.com/honeycombio/refinery/internal/otelutil"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -13,6 +14,9 @@ import (
 )
 
 func (r *Router) postOTLPLogs(w http.ResponseWriter, req *http.Request) {
+	ctx, span := otelutil.StartSpan(req.Context(), r.Tracer, "postOTLPLogs")
+	defer span.End()
+
 	ri := huskyotlp.GetRequestInfoFromHttpHeaders(req.Header)
 
 	if err := ri.ValidateLogsHeaders(); err != nil {
@@ -31,13 +35,13 @@ func (r *Router) postOTLPLogs(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	result, err := huskyotlp.TranslateLogsRequestFromReader(req.Context(), req.Body, ri)
+	result, err := huskyotlp.TranslateLogsRequestFromReader(ctx, req.Body, ri)
 	if err != nil {
 		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusInternalServerError})
 		return
 	}
 
-	if err := r.processOTLPRequest(req.Context(), result.Batches, keyToUse, ri.UserAgent); err != nil {
+	if err := r.processOTLPRequest(ctx, result.Batches, keyToUse, ri.UserAgent); err != nil {
 		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusInternalServerError})
 		return
 	}
@@ -56,6 +60,9 @@ func NewLogsServer(router *Router) *LogsServer {
 }
 
 func (l *LogsServer) Export(ctx context.Context, req *collectorlogs.ExportLogsServiceRequest) (*collectorlogs.ExportLogsServiceResponse, error) {
+	ctx, span := otelutil.StartSpan(ctx, l.router.Tracer, "ExportOTLPLogs")
+	defer span.End()
+
 	ri := huskyotlp.GetRequestInfoFromGrpcMetadata(ctx)
 	if err := ri.ValidateLogsHeaders(); err != nil {
 		return nil, huskyotlp.AsGRPCError(err)
diff --git a/route/otlp_trace.go b/route/otlp_trace.go
index f430c96..56614f0 100644
--- a/route/otlp_trace.go
+++ b/route/otlp_trace.go
@@ -6,6 +6,7 @@ import (
 	"net/http"
 
 	huskyotlp "github.com/honeycombio/husky/otlp"
+	"github.com/honeycombio/refinery/internal/otelutil"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -13,8 +14,10 @@ import (
 )
 
 func (r *Router) postOTLPTrace(w http.ResponseWriter, req *http.Request) {
-	ri := huskyotlp.GetRequestInfoFromHttpHeaders(req.Header)
+	ctx, span := otelutil.StartSpan(req.Context(), r.Tracer, "postOTLPTrace")
+	defer span.End()
 
+	ri := huskyotlp.GetRequestInfoFromHttpHeaders(req.Header)
 	if err := ri.ValidateTracesHeaders(); err != nil {
 		if errors.Is(err, huskyotlp.ErrInvalidContentType) {
 			r.handleOTLPFailureResponse(w, req, huskyotlp.ErrInvalidContentType)
@@ -32,13 +35,13 @@ func (r *Router) postOTLPTrace(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	result, err := huskyotlp.TranslateTraceRequestFromReader(req.Context(), req.Body, ri)
+	result, err := huskyotlp.TranslateTraceRequestFromReader(ctx, req.Body, ri)
 	if err != nil {
 		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusInternalServerError})
 		return
 	}
 
-	if err := r.processOTLPRequest(req.Context(), result.Batches, keyToUse, ri.UserAgent); err != nil {
+	if err := r.processOTLPRequest(ctx, result.Batches, keyToUse, ri.UserAgent); err != nil {
 		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusInternalServerError})
 		return
 	}
@@ -57,6 +60,9 @@ func NewTraceServer(router *Router) *TraceServer {
 }
 
 func (t *TraceServer) Export(ctx context.Context, req *collectortrace.ExportTraceServiceRequest) (*collectortrace.ExportTraceServiceResponse, error) {
+	ctx, span := otelutil.StartSpan(ctx, t.router.Tracer, "ExportOTLPTrace")
+	defer span.End()
+
 	ri := huskyotlp.GetRequestInfoFromGrpcMetadata(ctx)
 	if err := ri.ValidateTracesHeaders(); err != nil {
 		return nil, huskyotlp.AsGRPCError(err)
diff --git a/route/route.go b/route/route.go
index d1a31f3..6e784c9 100644
--- a/route/route.go
+++ b/route/route.go
@@ -44,6 +44,9 @@ import (
 	"github.com/honeycombio/refinery/transmit"
 	"github.com/honeycombio/refinery/types"
 
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel/trace"
 	collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1"
 	collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1"
 )
@@ -70,6 +73,7 @@ type Router struct {
 	Sharder              sharder.Sharder       `inject:""`
 	Collector            collect.Collector     `inject:""`
 	Metrics              metrics.Metrics       `inject:"genericMetrics"`
+	Tracer               trace.Tracer          `inject:"tracer"`
 
 	// version is set on startup so that the router may answer HTTP requests for
 	// the version
@@ -181,8 +185,9 @@ func (r *Router) LnS(incomingOrPeer string) {
 	authedMuxxer.Use(r.apiKeyProcessor)
 
 	// handle events and batches
-	authedMuxxer.HandleFunc("/events/{datasetName}", r.event).Name("event")
-	authedMuxxer.HandleFunc("/batch/{datasetName}", r.batch).Name("batch")
+	// Adds the OpenTelemetry instrumentation to the handler to enable tracing
+	authedMuxxer.Handle("/events/{datasetName}", otelhttp.NewHandler(http.HandlerFunc(r.event), "handle_event")).Name("event")
+	authedMuxxer.Handle("/batch/{datasetName}", otelhttp.NewHandler(http.HandlerFunc(r.batch), "handle_batch")).Name("batch")
 
 	// require an auth header for OTLP requests
 	r.AddOTLPMuxxer(muxxer)
@@ -225,6 +230,8 @@ func (r *Router) LnS(incomingOrPeer string) {
 				Time:                  time.Duration(grpcConfig.KeepAlive),
 				Timeout:               time.Duration(grpcConfig.KeepAliveTimeout),
 			}),
+			// Add the OpenTelemetry interceptor to the gRPC server to enable tracing
+			grpc.StatsHandler(otelgrpc.NewServerHandler()),
 		}
 		r.grpcServer = grpc.NewServer(serverOpts...)
 
@@ -366,6 +373,7 @@ func (r *Router) event(w http.ResponseWriter, req *http.Request) {
 	r.Metrics.Increment(r.incomingOrPeer + "_router_event")
 	defer req.Body.Close()
 
+	ctx := req.Context()
 	bodyReader, err := r.getMaybeCompressedBody(req)
 	if err != nil {
 		r.handlerReturnWithError(w, ErrPostBody, err)
@@ -378,14 +386,14 @@ func (r *Router) event(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	ev, err := r.requestToEvent(req, reqBod)
+	ev, err := r.requestToEvent(ctx, req, reqBod)
 	if err != nil {
 		r.handlerReturnWithError(w, ErrReqToEvent, err)
 		return
 	}
 	addIncomingUserAgent(ev, getUserAgentFromRequest(req))
 
-	reqID := req.Context().Value(types.RequestIDContextKey{})
+	reqID := ctx.Value(types.RequestIDContextKey{})
 	err = r.processEvent(ev, reqID)
 	if err != nil {
 		r.handlerReturnWithError(w, ErrReqToEvent, err)
@@ -393,7 +401,7 @@ func (r *Router) event(w http.ResponseWriter, req *http.Request) {
 	}
 }
 
-func (r *Router) requestToEvent(req *http.Request, reqBod []byte) (*types.Event, error) {
+func (r *Router) requestToEvent(ctx context.Context, req *http.Request, reqBod []byte) (*types.Event, error) {
 	// get necessary bits out of the incoming event
 	apiKey := req.Header.Get(types.APIKeyHeader)
 	if apiKey == "" {
@@ -424,7 +432,7 @@ func (r *Router) requestToEvent(req *http.Request, reqBod []byte) (*types.Event,
 	}
 
 	return &types.Event{
-		Context:     req.Context(),
+		Context:     ctx,
 		APIHost:     apiHost,
 		APIKey:      apiKey,
 		Dataset:     dataset,
@@ -439,7 +447,8 @@ func (r *Router) batch(w http.ResponseWriter, req *http.Request) {
 	r.Metrics.Increment(r.incomingOrPeer + "_router_batch")
 	defer req.Body.Close()
 
-	reqID := req.Context().Value(types.RequestIDContextKey{})
+	ctx := req.Context()
+	reqID := ctx.Value(types.RequestIDContextKey{})
 	debugLog := r.iopLogger.Debug().WithField("request_id", reqID)
 
 	bodyReader, err := r.getMaybeCompressedBody(req)
@@ -483,7 +492,7 @@ func (r *Router) batch(w http.ResponseWriter, req *http.Request) {
 	batchedResponses := make([]*BatchResponse, 0, len(batchedEvents))
 	for _, bev := range batchedEvents {
 		ev := &types.Event{
-			Context:     req.Context(),
+			Context:     ctx,
 			APIHost:     apiHost,
 			APIKey:      apiKey,
 			Dataset:     dataset,


Test patch:
diff --git a/route/otlp_logs_test.go b/route/otlp_logs_test.go
index d5bfa34..51f3c82 100644
--- a/route/otlp_logs_test.go
+++ b/route/otlp_logs_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 	collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1"
 	common "go.opentelemetry.io/proto/otlp/common/v1"
 	logs "go.opentelemetry.io/proto/otlp/logs/v1"
@@ -67,6 +68,7 @@ func TestLogsOTLPHandler(t *testing.T) {
 		},
 		Collector:      mockCollector,
 		incomingOrPeer: "incoming",
+		Tracer:         noop.Tracer{},
 	}
 	logsServer := NewLogsServer(router)
 
diff --git a/route/otlp_trace_test.go b/route/otlp_trace_test.go
index abcea1c..6176813 100644
--- a/route/otlp_trace_test.go
+++ b/route/otlp_trace_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 	collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1"
 	common "go.opentelemetry.io/proto/otlp/common/v1"
 	resource "go.opentelemetry.io/proto/otlp/resource/v1"
@@ -74,6 +75,7 @@ func TestOTLPHandler(t *testing.T) {
 		Logger:           &logger.MockLogger{},
 		zstdDecoders:     decoders,
 		environmentCache: newEnvironmentCache(time.Second, nil),
+		Tracer:           noop.Tracer{},
 	}
 
 	t.Run("span with status", func(t *testing.T) {
diff --git a/route/route_test.go b/route/route_test.go
index f6ab3e6..8ab45cf 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -350,6 +350,7 @@ func TestOTLPRequest(t *testing.T) {
 		},
 		Logger:           &logger.MockLogger{},
 		environmentCache: newEnvironmentCache(time.Second, nil),
+		Tracer:           noop.Tracer{},
 	}
 
 	muxxer := mux.NewRouter()