diff --git a/examples/otel-receiver/main.go b/examples/otel-receiver/main.go new file mode 100644 index 0000000..9edfa07 --- /dev/null +++ b/examples/otel-receiver/main.go @@ -0,0 +1,49 @@ +package main + +import ( + "log" + "net/http" + "os" + "strings" + + raindrop "github.com/raindrop-ai/go" +) + +// This example starts an HTTP server that receives OTLP/HTTP JSON traces +// and forwards them to Raindrop. +// +// Usage: +// RAINDROP_WRITE_KEY=rk_... go run . +// +// Then configure your OTLP trace source to send to: +// http://your-server:8090/v1/traces + +func main() { + writeKey := strings.TrimSpace(os.Getenv("RAINDROP_WRITE_KEY")) + if writeKey == "" { + log.Fatal("RAINDROP_WRITE_KEY is required") + } + + client, err := raindrop.New( + raindrop.WithWriteKey(writeKey), + raindrop.WithServiceName("my-go-proxy"), + ) + if err != nil { + log.Fatal(err) + } + defer client.Close() + + mux := http.NewServeMux() + mux.Handle("/v1/traces", client.OTLPHandler()) + + addr := envOrDefault("ADDR", ":8090") + log.Printf("OTLP receiver listening on %s", addr) + log.Fatal(http.ListenAndServe(addr, mux)) +} + +func envOrDefault(key, fallback string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return fallback +} diff --git a/examples/otel-spans/main.go b/examples/otel-spans/main.go new file mode 100644 index 0000000..6f4aff6 --- /dev/null +++ b/examples/otel-spans/main.go @@ -0,0 +1,244 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" + "time" + + raindrop "github.com/raindrop-ai/go" + otelattribute "go.opentelemetry.io/otel/attribute" + otelcodes "go.opentelemetry.io/otel/codes" + sdkresource "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + oteltrace "go.opentelemetry.io/otel/trace" +) + +// OpenAI-compatible chat types (works with any OpenAI-compatible provider). + +type chatMessage struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type chatRequest struct { + Model string `json:"model"` + Messages []chatMessage `json:"messages"` + MaxTokens int `json:"max_tokens,omitempty"` +} + +type chatResponse struct { + ID string `json:"id"` + Model string `json:"model"` + Choices []chatChoice `json:"choices"` + Usage *chatUsage `json:"usage"` +} + +type chatChoice struct { + Message chatMessage `json:"message"` +} + +type chatUsage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + +type weatherResult struct { + Forecast string `json:"forecast"` + TemperatureF int `json:"temperature_f"` +} + +func main() { + writeKey := strings.TrimSpace(os.Getenv("RAINDROP_WRITE_KEY")) + if writeKey == "" { + log.Fatal("RAINDROP_WRITE_KEY is required") + } + + llmAPIKey := strings.TrimSpace(os.Getenv("LLM_API_KEY")) + if llmAPIKey == "" { + log.Fatal("LLM_API_KEY is required") + } + + llmEndpoint := envOrDefault("LLM_ENDPOINT", "https://api.openai.com/v1/chat/completions") + model := envOrDefault("LLM_MODEL", "gpt-4o-mini") + userID := "user-123" + convoID := "conv-123" + prompt := "Plan a calm Saturday morning in San Francisco." + + client, err := raindrop.New( + raindrop.WithWriteKey(writeKey), + ) + if err != nil { + log.Fatal(err) + } + defer client.Close() + + ctx := context.Background() + httpClient := &http.Client{Timeout: 30 * time.Second} + tp := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(client.OTelSpanExporter()), + sdktrace.WithResource(sdkresource.NewSchemaless( + otelattribute.String("service.name", "otel-spans-example"), + otelattribute.String("service.version", raindrop.Version), + )), + ) + defer func() { _ = tp.Shutdown(ctx) }() + + tracer := tp.Tracer("github.com/raindrop-ai/go/examples/otel-spans") + interaction := client.Begin(ctx, raindrop.BeginOptions{ + UserID: userID, + Event: "chat_message", + Input: prompt, + ConvoID: convoID, + }) + + _, weatherSpan := tracer.Start(ctx, "weather_lookup", + oteltrace.WithAttributes(raindrop.OTelToolAttributes("weather_lookup", map[string]any{"location": "San Francisco"}, nil, map[string]any{ + "event_id": interaction.EventID(), + })...), + ) + weather, err := lookupWeather("San Francisco") + if err != nil { + weatherSpan.RecordError(err) + weatherSpan.SetStatus(otelcodes.Error, err.Error()) + weatherSpan.End() + log.Fatal(err) + } + weatherSpan.SetAttributes(raindrop.OTelToolAttributes("weather_lookup", nil, weather, nil)...) + weatherSpan.End() + + llmCtx, llmSpan := tracer.Start(ctx, "chat.completion", + oteltrace.WithSpanKind(oteltrace.SpanKindClient), + oteltrace.WithAttributes( + otelattribute.String("ai.telemetry.metadata.raindrop.eventId", interaction.EventID()), + otelattribute.String("gen_ai.request.model", model), + ), + ) + completion, err := createChatCompletion(llmCtx, httpClient, llmEndpoint, llmAPIKey, model, prompt, weather) + if err != nil { + llmSpan.RecordError(err) + llmSpan.SetStatus(otelcodes.Error, err.Error()) + llmSpan.End() + log.Fatal(err) + } + if completion.Usage != nil { + llmSpan.SetAttributes( + otelattribute.String("gen_ai.response.model", completion.Model), + otelattribute.Int("gen_ai.usage.prompt_tokens", completion.Usage.PromptTokens), + otelattribute.Int("gen_ai.usage.completion_tokens", completion.Usage.CompletionTokens), + ) + } + llmSpan.End() + + reply, err := assistantText(completion) + if err != nil { + log.Fatal(err) + } + + properties := map[string]any{} + if completion.Usage != nil { + properties["gen_ai.usage.prompt_tokens"] = completion.Usage.PromptTokens + properties["gen_ai.usage.completion_tokens"] = completion.Usage.CompletionTokens + } + + if err := interaction.Finish(raindrop.FinishOptions{ + Output: reply, + Model: completion.Model, + Properties: properties, + }); err != nil { + log.Fatal(err) + } + + fmt.Println(reply) +} + +func createChatCompletion(ctx context.Context, httpClient *http.Client, endpoint, apiKey, model, prompt string, weather weatherResult) (chatResponse, error) { + reqBody := chatRequest{ + Model: model, + Messages: []chatMessage{ + { + Role: "system", + Content: fmt.Sprintf("You are a helpful local planner. Weather context: %s, %dF.", weather.Forecast, weather.TemperatureF), + }, + { + Role: "user", + Content: prompt, + }, + }, + MaxTokens: 300, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return chatResponse{}, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return chatResponse{}, err + } + req.Header.Set("Authorization", "Bearer "+apiKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := httpClient.Do(req) + if err != nil { + return chatResponse{}, err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return chatResponse{}, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return chatResponse{}, fmt.Errorf("chat completion failed: %s: %s", resp.Status, strings.TrimSpace(string(respBody))) + } + + var decoded chatResponse + if err := json.Unmarshal(respBody, &decoded); err != nil { + return chatResponse{}, err + } + return decoded, nil +} + +func assistantText(response chatResponse) (string, error) { + if len(response.Choices) == 0 { + return "", errors.New("provider returned no choices") + } + + text := strings.TrimSpace(response.Choices[0].Message.Content) + if text == "" { + return "", errors.New("provider returned an empty assistant message") + } + return text, nil +} + +func lookupWeather(location string) (weatherResult, error) { + switch location { + case "San Francisco": + return weatherResult{ + Forecast: "sunny", + TemperatureF: 65, + }, nil + default: + return weatherResult{ + Forecast: "mild", + TemperatureF: 68, + }, nil + } +} + +func envOrDefault(key, fallback string) string { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + return fallback +} diff --git a/go.mod b/go.mod index 7738b8c..0109fb3 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,17 @@ module github.com/raindrop-ai/go go 1.21 + +require ( + go.opentelemetry.io/otel v1.28.0 + go.opentelemetry.io/otel/sdk v1.28.0 + go.opentelemetry.io/otel/trace v1.28.0 +) + +require ( + github.com/go-logr/logr v1.4.2 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/google/uuid v1.6.0 // indirect + go.opentelemetry.io/otel/metric v1.28.0 // indirect + golang.org/x/sys v0.21.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..bff4aba --- /dev/null +++ b/go.sum @@ -0,0 +1,27 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +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= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +go.opentelemetry.io/otel v1.28.0 h1:/SqNcYk+idO0CxKEUOtKQClMK/MimZihKYMruSMViUo= +go.opentelemetry.io/otel v1.28.0/go.mod h1:q68ijF8Fc8CnMHKyzqL6akLO46ePnjkgfIMIjUIX9z4= +go.opentelemetry.io/otel/metric v1.28.0 h1:f0HGvSl1KRAU1DLgLGFjrwVyismPlnuU6JD6bOeuA5Q= +go.opentelemetry.io/otel/metric v1.28.0/go.mod h1:Fb1eVBFZmLVTMb6PPohq3TO9IIhUisDsbJoL/+uQW4s= +go.opentelemetry.io/otel/sdk v1.28.0 h1:b9d7hIry8yZsgtbmM0DKyPWMMUMlK9NEKuIG4aBqWyE= +go.opentelemetry.io/otel/sdk v1.28.0/go.mod h1:oYj7ClPUA7Iw3m+r7GeEjz0qckQRJK2B8zjcZEfu7Pg= +go.opentelemetry.io/otel/trace v1.28.0 h1:GhQ9cUuQGmNDd5BTCP2dAvv75RdMxEfTmYejp+lkx9g= +go.opentelemetry.io/otel/trace v1.28.0/go.mod h1:jPyXzNPg6da9+38HEwElrQiHlVMTnVfM3/yv2OlIHaI= +golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/otel.go b/otel.go new file mode 100644 index 0000000..ebbde07 --- /dev/null +++ b/otel.go @@ -0,0 +1,360 @@ +package raindrop + +import ( + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "sort" + + otelattribute "go.opentelemetry.io/otel/attribute" + otelcodes "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + oteltrace "go.opentelemetry.io/otel/trace" +) + +const maxOTLPBodyBytes = 5 << 20 // 5 MB + +type OTelSpanExporter struct { + client *Client +} + +func (c *Client) OTelSpanExporter() *OTelSpanExporter { + return &OTelSpanExporter{client: c} +} + +func (e *OTelSpanExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error { + if e == nil || e.client == nil || !e.client.enabled || len(spans) == 0 { + return nil + } + if err := e.client.ensureOpen(); err != nil { + return err + } + + payload := buildOTelExportTraceServiceRequest(spans, e.client.serviceName, e.client.version) + if len(payload.ResourceSpans) == 0 { + return nil + } + return e.client.transport.postJSON(ctx, "traces", payload) +} + +func (e *OTelSpanExporter) Shutdown(context.Context) error { + return nil +} + +// OTLPHandler returns an http.Handler that accepts OTLP/HTTP JSON trace +// payloads and forwards them to Raindrop. Mount it at your preferred path +// (typically "/v1/traces"). +func (c *Client) OTLPHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if c == nil || !c.enabled { + _, _ = io.Copy(io.Discard, r.Body) + w.WriteHeader(http.StatusOK) + return + } + if err := c.ensureOpen(); err != nil { + http.Error(w, "service unavailable", http.StatusServiceUnavailable) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxOTLPBodyBytes)) + if err != nil { + http.Error(w, "failed to read body", http.StatusBadRequest) + return + } + + var req exportTraceServiceRequest + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, "invalid OTLP JSON", http.StatusBadRequest) + return + } + if len(req.ResourceSpans) == 0 { + w.WriteHeader(http.StatusOK) + return + } + + defaults := defaultResourceAttributes(c.serviceName, c.version) + for i := range req.ResourceSpans { + req.ResourceSpans[i].Resource.Attributes = mergeOTLPAttributes( + defaults, + req.ResourceSpans[i].Resource.Attributes, + ) + } + + if err := c.transport.postJSON(r.Context(), "traces", req); err != nil { + c.debugLog("OTLPHandler: failed to forward traces", "error", err) + http.Error(w, "failed to forward traces", http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusOK) + }) +} + +func OTelToolAttributes(name string, input any, output any, properties map[string]any) []otelattribute.KeyValue { + attrs := make([]otelattribute.KeyValue, 0, 2+len(properties)+2) + attrs = append(attrs, otelattribute.String("traceloop.span.kind", "tool")) + if name != "" { + attrs = append(attrs, otelattribute.String("traceloop.entity.name", name)) + } + if input != nil { + attrs = append(attrs, otelattribute.String("traceloop.entity.input", stringifyValue(input))) + } + if output != nil { + attrs = append(attrs, otelattribute.String("traceloop.entity.output", stringifyValue(output))) + } + attrs = append(attrs, otelToolPropertyAttributes(properties)...) + return attrs +} + +func buildOTelExportTraceServiceRequest(spans []sdktrace.ReadOnlySpan, defaultServiceName, defaultServiceVersion string) exportTraceServiceRequest { + resourceSpansList := make([]resourceSpans, 0, len(spans)) + + for _, span := range spans { + converted, ok := convertOTelSpan(span) + if !ok { + continue + } + + resourceAttrs := []otlpKeyValue(nil) + if resource := span.Resource(); resource != nil { + resourceAttrs = convertOTelKeyValues(resource.Attributes()) + } + resourceAttrs = mergeOTLPAttributes( + defaultResourceAttributes(defaultServiceName, defaultServiceVersion), + resourceAttrs, + ) + + scopeInfo := span.InstrumentationScope() + scopeName := scopeInfo.Name + if scopeName == "" { + scopeName = defaultServiceName + } + scopeVersion := scopeInfo.Version + if scopeVersion == "" { + scopeVersion = defaultServiceVersion + } + + resourceSpansList = append(resourceSpansList, resourceSpans{ + Resource: resource{Attributes: resourceAttrs}, + ScopeSpans: []scopeSpans{ + { + Scope: scope{ + Name: scopeName, + Version: scopeVersion, + }, + Spans: []otlpSpan{converted}, + }, + }, + }) + } + + return exportTraceServiceRequest{ResourceSpans: resourceSpansList} +} + +func convertOTelSpan(span sdktrace.ReadOnlySpan) (otlpSpan, bool) { + spanContext := span.SpanContext() + if !spanContext.IsValid() { + return otlpSpan{}, false + } + + attributes := convertOTelKeyValues(span.Attributes()) + if kind := span.SpanKind().String(); kind != "" && kind != "unspecified" { + attributes = append(attributes, otlpKeyValue{ + Key: "otel.span.kind", + Value: otlpAnyValue{StringValue: kind}, + }) + } + + parent := span.Parent() + + return otlpSpan{ + TraceID: encodeOTelTraceID(spanContext.TraceID()), + SpanID: encodeOTelSpanID(spanContext.SpanID()), + ParentSpanID: encodeOTelParentSpanID(parent), + Name: span.Name(), + StartTimeUnixNano: unixNanoString(span.StartTime()), + EndTimeUnixNano: unixNanoString(span.EndTime()), + Attributes: attributes, + Status: convertOTelStatus(span.Status()), + }, true +} + +func convertOTelStatus(status sdktrace.Status) *otlpStatus { + switch status.Code { + case otelcodes.Error: + return &otlpStatus{ + Code: SpanStatusError, + Message: status.Description, + } + case otelcodes.Ok: + return &otlpStatus{Code: SpanStatusOK} + default: + return &otlpStatus{Code: SpanStatusUnset} + } +} + +func convertOTelKeyValues(attrs []otelattribute.KeyValue) []otlpKeyValue { + if len(attrs) == 0 { + return nil + } + + converted := make([]otlpKeyValue, 0, len(attrs)) + for _, attr := range attrs { + if !attr.Valid() { + continue + } + converted = append(converted, otlpKeyValue{ + Key: string(attr.Key), + Value: convertOTelValue(attr.Value), + }) + } + return converted +} + +func convertOTelValue(value otelattribute.Value) otlpAnyValue { + switch value.Type() { + case otelattribute.BOOL: + boolValue := value.AsBool() + return otlpAnyValue{BoolValue: &boolValue} + case otelattribute.INT64: + return otlpAnyValue{IntValue: stringifyValue(value.AsInt64())} + case otelattribute.FLOAT64: + floatValue := value.AsFloat64() + return otlpAnyValue{DoubleValue: &floatValue} + case otelattribute.STRING: + return otlpAnyValue{StringValue: value.AsString()} + case otelattribute.BOOLSLICE: + values := make([]otlpAnyValue, 0, len(value.AsBoolSlice())) + for _, item := range value.AsBoolSlice() { + boolValue := item + values = append(values, otlpAnyValue{BoolValue: &boolValue}) + } + return otlpAnyValue{ArrayValue: &otlpArrayValue{Values: values}} + case otelattribute.INT64SLICE: + values := make([]otlpAnyValue, 0, len(value.AsInt64Slice())) + for _, item := range value.AsInt64Slice() { + values = append(values, otlpAnyValue{IntValue: stringifyValue(item)}) + } + return otlpAnyValue{ArrayValue: &otlpArrayValue{Values: values}} + case otelattribute.FLOAT64SLICE: + values := make([]otlpAnyValue, 0, len(value.AsFloat64Slice())) + for _, item := range value.AsFloat64Slice() { + floatValue := item + values = append(values, otlpAnyValue{DoubleValue: &floatValue}) + } + return otlpAnyValue{ArrayValue: &otlpArrayValue{Values: values}} + case otelattribute.STRINGSLICE: + values := make([]otlpAnyValue, 0, len(value.AsStringSlice())) + for _, item := range value.AsStringSlice() { + values = append(values, otlpAnyValue{StringValue: item}) + } + return otlpAnyValue{ArrayValue: &otlpArrayValue{Values: values}} + default: + return otlpAnyValue{StringValue: value.Emit()} + } +} + +func otelToolPropertyAttributes(properties map[string]any) []otelattribute.KeyValue { + if len(properties) == 0 { + return nil + } + + keys := make([]string, 0, len(properties)) + for key := range properties { + if key != "" && properties[key] != nil { + keys = append(keys, key) + } + } + sort.Strings(keys) + + attrs := make([]otelattribute.KeyValue, 0, len(keys)) + for _, key := range keys { + attrKey := "traceloop.association.properties." + key + switch typed := properties[key].(type) { + case string: + attrs = append(attrs, otelattribute.String(attrKey, typed)) + case bool: + attrs = append(attrs, otelattribute.Bool(attrKey, typed)) + case int: + attrs = append(attrs, otelattribute.Int(attrKey, typed)) + case int8: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case int16: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case int32: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case int64: + attrs = append(attrs, otelattribute.Int64(attrKey, typed)) + case uint: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case uint8: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case uint16: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case uint32: + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + case uint64: + if typed <= uint64(^uint64(0)>>1) { + attrs = append(attrs, otelattribute.Int64(attrKey, int64(typed))) + } else { + attrs = append(attrs, otelattribute.String(attrKey, stringifyValue(typed))) + } + case float32: + attrs = append(attrs, otelattribute.Float64(attrKey, float64(typed))) + case float64: + attrs = append(attrs, otelattribute.Float64(attrKey, typed)) + case []string: + attrs = append(attrs, otelattribute.StringSlice(attrKey, typed)) + default: + attrs = append(attrs, otelattribute.String(attrKey, stringifyValue(typed))) + } + } + + return attrs +} + +func mergeOTLPAttributes(base []otlpKeyValue, overlay []otlpKeyValue) []otlpKeyValue { + if len(base) == 0 && len(overlay) == 0 { + return nil + } + + valuesByKey := make(map[string]otlpAnyValue, len(base)+len(overlay)) + for _, attr := range base { + valuesByKey[attr.Key] = attr.Value + } + for _, attr := range overlay { + valuesByKey[attr.Key] = attr.Value + } + + keys := make([]string, 0, len(valuesByKey)) + for key := range valuesByKey { + keys = append(keys, key) + } + sort.Strings(keys) + + merged := make([]otlpKeyValue, 0, len(keys)) + for _, key := range keys { + merged = append(merged, otlpKeyValue{Key: key, Value: valuesByKey[key]}) + } + return merged +} + +func encodeOTelTraceID(id oteltrace.TraceID) string { + return base64.StdEncoding.EncodeToString(id[:]) +} + +func encodeOTelSpanID(id oteltrace.SpanID) string { + return base64.StdEncoding.EncodeToString(id[:]) +} + +func encodeOTelParentSpanID(parent oteltrace.SpanContext) string { + if !parent.IsValid() { + return "" + } + return encodeOTelSpanID(parent.SpanID()) +} diff --git a/otlp.go b/otlp.go index 6447f41..6fd0f0d 100644 --- a/otlp.go +++ b/otlp.go @@ -151,6 +151,13 @@ func unixNanoString(at time.Time) string { return strconv.FormatInt(at.UnixNano(), 10) } +func defaultResourceAttributes(serviceName, serviceVersion string) []otlpKeyValue { + return []otlpKeyValue{ + {Key: "service.name", Value: otlpAnyValue{StringValue: serviceName}}, + {Key: "service.version", Value: otlpAnyValue{StringValue: serviceVersion}}, + } +} + func buildExportTraceServiceRequest(spans []otlpSpan, serviceName, serviceVersion string) exportTraceServiceRequest { return exportTraceServiceRequest{ ResourceSpans: []resourceSpans{ diff --git a/raindrop_test.go b/raindrop_test.go index 37cb34c..ae19e0d 100644 --- a/raindrop_test.go +++ b/raindrop_test.go @@ -1,6 +1,7 @@ package raindrop import ( + "bytes" "context" "encoding/json" "io" @@ -1040,3 +1041,159 @@ func newTestClient(t *testing.T, endpoint string, opts ...Option) *Client { } return client } + +func validOTLPPayload() exportTraceServiceRequest { + return exportTraceServiceRequest{ + ResourceSpans: []resourceSpans{{ + Resource: resource{Attributes: []otlpKeyValue{ + {Key: "service.name", Value: otlpAnyValue{StringValue: "upstream-llm"}}, + }}, + ScopeSpans: []scopeSpans{{ + Scope: scope{Name: "upstream-llm", Version: "1.0"}, + Spans: []otlpSpan{{ + TraceID: "dGVzdHRyYWNlaWQx", + SpanID: "dGVzdHNwYW4x", + Name: "chat_completion", + StartTimeUnixNano: "1000000000", + EndTimeUnixNano: "2000000000", + }}, + }}, + }}, + } +} + +func TestOTLPHandlerForwardsValidPayload(t *testing.T) { + var received exportTraceServiceRequest + var gotRequest atomic.Bool + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/traces" { + t.Errorf("unexpected path: %s", r.URL.Path) + } + gotRequest.Store(true) + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &received); err != nil { + t.Errorf("unmarshal: %v", err) + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + client := newTestClient(t, server.URL+"/") + defer func() { _ = client.Close() }() + + payload, _ := json.Marshal(validOTLPPayload()) + req := httptest.NewRequest(http.MethodPost, "/v1/traces", bytes.NewReader(payload)) + rec := httptest.NewRecorder() + client.OTLPHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if !gotRequest.Load() { + t.Fatal("upstream never received the request") + } + if len(received.ResourceSpans) != 1 { + t.Fatalf("expected 1 resourceSpans, got %d", len(received.ResourceSpans)) + } + // Verify incoming service.name was preserved (overlay wins over defaults) + attrs := received.ResourceSpans[0].Resource.Attributes + var foundServiceName string + for _, attr := range attrs { + if attr.Key == "service.name" { + foundServiceName = attr.Value.StringValue + } + } + if foundServiceName != "upstream-llm" { + t.Fatalf("expected service.name=upstream-llm, got %q", foundServiceName) + } +} + +func TestOTLPHandlerRejectsInvalidJSON(t *testing.T) { + var gotRequest atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotRequest.Store(true) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + client := newTestClient(t, server.URL+"/") + defer func() { _ = client.Close() }() + + req := httptest.NewRequest(http.MethodPost, "/v1/traces", strings.NewReader("not json")) + rec := httptest.NewRecorder() + client.OTLPHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d", rec.Code) + } + if gotRequest.Load() { + t.Fatal("upstream should not have received a request") + } +} + +func TestOTLPHandlerRejectsWrongMethod(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("upstream should not have received a request") + })) + defer server.Close() + + client := newTestClient(t, server.URL+"/") + defer func() { _ = client.Close() }() + + req := httptest.NewRequest(http.MethodGet, "/v1/traces", nil) + rec := httptest.NewRecorder() + client.OTLPHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("expected 405, got %d", rec.Code) + } +} + +func TestOTLPHandlerNoopWhenDisabled(t *testing.T) { + var gotRequest atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotRequest.Store(true) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + client := newTestClient(t, server.URL+"/", WithWriteKey("")) + defer func() { _ = client.Close() }() + + payload, _ := json.Marshal(validOTLPPayload()) + req := httptest.NewRequest(http.MethodPost, "/v1/traces", bytes.NewReader(payload)) + rec := httptest.NewRecorder() + client.OTLPHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if gotRequest.Load() { + t.Fatal("upstream should not have received a request when disabled") + } +} + +func TestOTLPHandlerEmptyResourceSpans(t *testing.T) { + var gotRequest atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotRequest.Store(true) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + client := newTestClient(t, server.URL+"/") + defer func() { _ = client.Close() }() + + payload, _ := json.Marshal(exportTraceServiceRequest{ResourceSpans: []resourceSpans{}}) + req := httptest.NewRequest(http.MethodPost, "/v1/traces", bytes.NewReader(payload)) + rec := httptest.NewRecorder() + client.OTLPHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if gotRequest.Load() { + t.Fatal("upstream should not have received a request for empty payload") + } +}