feat: Add OTLP log endpoints (gRPC & HTTP)

collect/mockCollector.go

@@ -0,0 +1,49 @@
+package collect
+
+import (
+	"github.com/honeycombio/refinery/types"
+)
+
+type MockCollector struct {
+	Spans chan *types.Span
+}
+
+func NewMockCollector() *MockCollector {
+	return &MockCollector{
+		Spans: make(chan *types.Span, 100),
+	}
+}
+
+func (m *MockCollector) AddSpan(span *types.Span) error {
+	m.Spans <- span
+	return nil
+}
+
+func (m *MockCollector) AddSpanFromPeer(span *types.Span) error {
+	m.Spans <- span
+	return nil
+}
+
+func (m *MockCollector) GetStressedSampleRate(traceID string) (rate uint, keep bool, reason string) {
+	return 0, false, ""
+}
+
+func (m *MockCollector) ProcessSpanImmediately(sp *types.Span, keep bool, sampleRate uint, reason string) {
+	m.Spans <- sp
+}
+
+func (m *MockCollector) Stressed() bool {
+	return false
+}
+
+func (m *MockCollector) Flush() {
+	for {
+		select {
+		case <-m.Spans:
+		default:
+			return
+		}
+	}
+}
+
+var _ Collector = (*MockCollector)(nil)

route/otlp_logs.go

@@ -0,0 +1,77 @@
+package route
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+
+	huskyotlp "github.com/honeycombio/husky/otlp"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+)
+
+func (r *Router) postOTLPLogs(w http.ResponseWriter, req *http.Request) {
+	ri := huskyotlp.GetRequestInfoFromHttpHeaders(req.Header)
+
+	if err := ri.ValidateLogsHeaders(); err != nil {
+		if errors.Is(err, huskyotlp.ErrInvalidContentType) {
+			r.handlerReturnWithError(w, ErrInvalidContentType, err)
+		} else {
+			r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusUnauthorized})
+		}
+		return
+	}
+
+	if !r.Config.IsAPIKeyValid(ri.ApiKey) {
+		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: fmt.Sprintf("api key %s not found in list of authorized keys", ri.ApiKey), HTTPStatusCode: http.StatusUnauthorized})
+		return
+	}
+
+	result, err := huskyotlp.TranslateLogsRequestFromReader(req.Context(), 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, ri.ApiKey); err != nil {
+		r.handleOTLPFailureResponse(w, req, huskyotlp.OTLPError{Message: err.Error(), HTTPStatusCode: http.StatusInternalServerError})
+		return
+	}
+
+	_ = huskyotlp.WriteOtlpHttpTraceSuccessResponse(w, req)
+}
+
+type LogsServer struct {
+	router *Router
+	collectorlogs.UnimplementedLogsServiceServer
+}
+
+func NewLogsServer(router *Router) *LogsServer {
+	logsServer := LogsServer{router: router}
+	return &logsServer
+}
+
+func (l *LogsServer) Export(ctx context.Context, req *collectorlogs.ExportLogsServiceRequest) (*collectorlogs.ExportLogsServiceResponse, error) {
+	ri := huskyotlp.GetRequestInfoFromGrpcMetadata(ctx)
+	if err := ri.ValidateLogsHeaders(); err != nil {
+		return nil, huskyotlp.AsGRPCError(err)
+	}
+
+	if !l.router.Config.IsAPIKeyValid(ri.ApiKey) {
+		return nil, status.Error(codes.Unauthenticated, fmt.Sprintf("api key %s not found in list of authorized keys", ri.ApiKey))
+	}
+
+	result, err := huskyotlp.TranslateLogsRequest(ctx, req, ri)
+	if err != nil {
+		return nil, huskyotlp.AsGRPCError(err)
+	}
+
+	if err := l.router.processOTLPRequest(ctx, result.Batches, ri.ApiKey); err != nil {
+		return nil, huskyotlp.AsGRPCError(err)
+	}
+
+	return &collectorlogs.ExportLogsServiceResponse{}, nil
+}