[receiver/libhoney] Fix msgpack response object (#40414)

Author: mterhar

.chloggen/libhoney-resonponse-encoding.yaml

@@ -0,0 +1,8 @@
+change_type: bug_fix
+component: libhoneyreceiver
+note: Fix response encoding for msgpack
+issues: [40413]
+change_logs: [user]
+subtext: |
+  The libhoneyreceiver now correctly encodes the response for msgpack.
+  This fixes the issue where the response was not being encoded correctly.

receiver/libhoneyreceiver/encoder/encoder.go

@@ -16,7 +16,7 @@ import (
 const (
 	PbContentType      = "application/x-protobuf"
 	JSONContentType    = "application/json"
-	MsgpackContentType = "application/x-msgpack"
+	MsgpackContentType = "application/msgpack"
 )
 
 var (

receiver/libhoneyreceiver/internal/libhoneyevent/libhoneyevent.go

@@ -104,18 +104,18 @@ func (l *LibhoneyEvent) SignalType(logger zap.Logger) string {
 				case "link":
 					return "span_link"
 				}
-				logger.Warn("invalid annotation type", zap.String("meta.annotation_type", atype.(string)))
+				logger.Debug("invalid annotation type", zap.String("meta.annotation_type", atype.(string)))
 				return "span"
 			}
 			return "span"
 		case "log":
 			return "log"
 		default:
-			logger.Warn("invalid meta.signal_type", zap.String("meta.signal_type", sig.(string)))
+			logger.Debug("invalid meta.signal_type", zap.String("meta.signal_type", sig.(string)))
 			return "log"
 		}
 	}
-	logger.Warn("missing meta.signal_type and meta.annotation_type")
+	logger.Debug("missing meta.signal_type and meta.annotation_type")
 	return "log"
 }
 
@@ -251,7 +251,7 @@ func (l *LibhoneyEvent) ToPLogRecord(newLog *plog.LogRecord, alreadyUsedFields *
 		case bool:
 			newLog.Attributes().PutBool(k, v)
 		default:
-			logger.Warn("Span data type issue", zap.Int64("timestamp", timeNs), zap.String("key", k))
+			logger.Debug("Span data type issue", zap.Int64("timestamp", timeNs), zap.String("key", k))
 		}
 	}
 	return nil
@@ -378,7 +378,7 @@ func (l *LibhoneyEvent) ToPTraceSpan(newSpan *ptrace.Span, alreadyUsedFields *[]
 		case bool:
 			newSpan.Attributes().PutBool(k, v)
 		default:
-			logger.Warn("Span data type issue", zap.String("trace.trace_id", newSpan.TraceID().String()), zap.String("trace.span_id", newSpan.SpanID().String()), zap.String("key", k))
+			logger.Debug("Span data type issue", zap.String("trace.trace_id", newSpan.TraceID().String()), zap.String("trace.span_id", newSpan.SpanID().String()), zap.String("key", k))
 		}
 	}
 	return nil

receiver/libhoneyreceiver/internal/parser/parser.go

@@ -52,16 +52,16 @@ func ToPdata(dataset string, lhes []libhoneyevent.LibhoneyEvent, cfg libhoneyeve
 	} // seed a default
 
 	alreadyUsedFields := []string{cfg.Resources.ServiceName, cfg.Scopes.LibraryName, cfg.Scopes.LibraryVersion}
-	alreadyUsedFields = append(alreadyUsedFields, cfg.Attributes.Name,
+	alreadyUsedTraceFields := []string{
+		cfg.Attributes.Name,
 		cfg.Attributes.TraceID, cfg.Attributes.ParentID, cfg.Attributes.SpanID,
 		cfg.Attributes.Error, cfg.Attributes.SpanKind,
-	)
-	alreadyUsedFields = append(alreadyUsedFields, cfg.Attributes.DurationFields...)
+	}
 
 	for _, lhe := range lhes {
 		parentID, err := lhe.GetParentID(cfg.Attributes.ParentID)
 		if err != nil {
-			logger.Warn("parent id not found")
+			logger.Debug("parent id not found")
 		}
 
 		action := lhe.SignalType(logger)
@@ -70,6 +70,8 @@ func ToPdata(dataset string, lhes []libhoneyevent.LibhoneyEvent, cfg libhoneyeve
 			spanService, _ := lhe.GetService(cfg, &foundServices, dataset)
 			spanScopeKey, _ := lhe.GetScope(cfg, &foundScopes, spanService) // adds a new found scope if needed
 			newSpan := foundScopes.Scope[spanScopeKey].ScopeSpans.AppendEmpty()
+			alreadyUsedFields = append(alreadyUsedFields, alreadyUsedTraceFields...)
+			alreadyUsedFields = append(alreadyUsedFields, cfg.Attributes.DurationFields...)
 			err := lhe.ToPTraceSpan(&newSpan, &alreadyUsedFields, cfg, logger)
 			if err != nil {
 				logger.Warn("span could not be converted from libhoney to ptrace", zap.String("span.object", lhe.DebugString()))
@@ -160,7 +162,7 @@ func addSpanEventsToSpan(sp ptrace.Span, events []libhoneyevent.LibhoneyEvent, a
 			case bool:
 				newEvent.Attributes().PutBool(lkey, lval)
 			default:
-				logger.Warn("SpanEvent data type issue",
+				logger.Debug("SpanEvent data type issue",
 					zap.String("trace.trace_id", sp.TraceID().String()),
 					zap.String("trace.span_id", sp.SpanID().String()),
 					zap.String("key", lkey))
@@ -176,7 +178,7 @@ func addSpanLinksToSpan(sp ptrace.Span, links []libhoneyevent.LibhoneyEvent, alr
 		if linkTraceStr, ok := spl.Data["trace.link.trace_id"]; ok {
 			tidByteArray, err := hex.DecodeString(linkTraceStr.(string))
 			if err != nil {
-				logger.Warn("span link invalid",
+				logger.Debug("span link invalid",
 					zap.String("missing.attribute", "trace.link.trace_id"),
 					zap.String("span link contents", spl.DebugString()))
 				continue
@@ -186,7 +188,7 @@ func addSpanLinksToSpan(sp ptrace.Span, links []libhoneyevent.LibhoneyEvent, alr
 			}
 			newLink.SetTraceID(pcommon.TraceID(tidByteArray))
 		} else {
-			logger.Warn("span link missing attributes",
+			logger.Debug("span link missing attributes",
 				zap.String("missing.attribute", "trace.link.trace_id"),
 				zap.String("span link contents", spl.DebugString()))
 			continue
@@ -195,7 +197,7 @@ func addSpanLinksToSpan(sp ptrace.Span, links []libhoneyevent.LibhoneyEvent, alr
 		if linkSpanStr, ok := spl.Data["trace.link.span_id"]; ok {
 			sidByteArray, err := hex.DecodeString(linkSpanStr.(string))
 			if err != nil {
-				logger.Warn("span link invalid",
+				logger.Debug("span link invalid",
 					zap.String("missing.attribute", "trace.link.span_id"),
 					zap.String("span link contents", spl.DebugString()))
 				continue
@@ -205,7 +207,7 @@ func addSpanLinksToSpan(sp ptrace.Span, links []libhoneyevent.LibhoneyEvent, alr
 			}
 			newLink.SetSpanID(pcommon.SpanID(sidByteArray))
 		} else {
-			logger.Warn("span link missing attributes",
+			logger.Debug("span link missing attributes",
 				zap.String("missing.attribute", "trace.link.span_id"),
 				zap.String("span link contents", spl.DebugString()))
 			continue
@@ -234,7 +236,7 @@ func addSpanLinksToSpan(sp ptrace.Span, links []libhoneyevent.LibhoneyEvent, alr
 			case bool:
 				newLink.Attributes().PutBool(lkey, lval)
 			default:
-				logger.Warn("SpanLink data type issue",
+				logger.Debug("SpanLink data type issue",
 					zap.String("trace.trace_id", sp.TraceID().String()),
 					zap.String("trace.span_id", sp.SpanID().String()),
 					zap.String("key", lkey))

receiver/libhoneyreceiver/internal/response/response.go

@@ -0,0 +1,26 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package response // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/response"
+
+import "net/http"
+
+type ResponseInBatch struct {
+	ErrorStr string `json:"error,omitempty"`
+	Status   int    `json:"status,omitempty"`
+}
+
+func MakeResponse(eventErrs []int) []ResponseInBatch {
+	if len(eventErrs) == 0 {
+		return []ResponseInBatch{{Status: http.StatusAccepted}}
+	}
+
+	responses := make([]ResponseInBatch, len(eventErrs))
+	for i, eventErr := range eventErrs {
+		responses[i] = ResponseInBatch{
+			ErrorStr: "error",
+			Status:   eventErr,
+		}
+	}
+	return responses
+}

receiver/libhoneyreceiver/receiver.go

@@ -15,6 +15,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/vmihailenco/msgpack/v5"
 	"go.opentelemetry.io/collector/component"
@@ -26,8 +27,10 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/errorutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/encoder"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/eventtime"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/libhoneyevent"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/parser"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/response"
 )
 
 type libhoneyReceiver struct {
@@ -226,6 +229,21 @@ func (r *libhoneyReceiver) handleEvent(resp http.ResponseWriter, req *http.Reque
 		if err != nil {
 			r.settings.Logger.Info("messagepack decoding failed")
 		}
+		// Post-process msgpack events to ensure timestamps are set
+		for i := range libhoneyevents {
+			if libhoneyevents[i].MsgPackTimestamp == nil {
+				if libhoneyevents[i].Time != "" {
+					// Parse the time string and set MsgPackTimestamp
+					propertime := eventtime.GetEventTime(libhoneyevents[i].Time)
+					libhoneyevents[i].MsgPackTimestamp = &propertime
+				} else {
+					// No time field, use current time
+					tnow := time.Now()
+					libhoneyevents[i].MsgPackTimestamp = &tnow
+					libhoneyevents[i].Time = eventtime.GetEventTimeDefaultString()
+				}
+			}
+		}
 		if len(libhoneyevents) > 0 {
 			r.settings.Logger.Debug("Decoding with msgpack worked", zap.Time("timestamp.first.msgpacktimestamp", *libhoneyevents[0].MsgPackTimestamp), zap.String("timestamp.first.time", libhoneyevents[0].Time))
 			r.settings.Logger.Debug("event zero", zap.String("event.data", libhoneyevents[0].DebugString()))
@@ -238,6 +256,8 @@ func (r *libhoneyReceiver) handleEvent(resp http.ResponseWriter, req *http.Reque
 		if len(libhoneyevents) > 0 {
 			r.settings.Logger.Debug("Decoding with json worked", zap.Time("timestamp.first.msgpacktimestamp", *libhoneyevents[0].MsgPackTimestamp), zap.String("timestamp.first.time", libhoneyevents[0].Time))
 		}
+	default:
+		r.settings.Logger.Info("unsupported content type", zap.String("content-type", req.Header.Get("Content-Type")))
 	}
 
 	otlpLogs, otlpTraces := parser.ToPdata(dataset, libhoneyevents, r.cfg.FieldMapConfig, *r.settings.Logger)
@@ -261,8 +281,28 @@ func (r *libhoneyReceiver) handleEvent(resp http.ResponseWriter, req *http.Reque
 		return
 	}
 
-	noErrors := []byte(`{"errors":[]}`)
-	writeResponse(resp, enc.ContentType(), http.StatusOK, noErrors)
+	// return clean response if no errors above
+	noErrors := response.MakeResponse([]int{})
+
+	var responseBody []byte
+	var contentType string
+
+	switch enc.ContentType() {
+	case encoder.MsgpackContentType:
+		// For msgpack requests, return msgpack response
+		responseBody, err = msgpack.Marshal(noErrors)
+		contentType = encoder.MsgpackContentType
+	default:
+		// For JSON requests, return JSON response
+		responseBody, err = json.Marshal(noErrors)
+		contentType = encoder.JSONContentType
+	}
+
+	if err != nil {
+		errorutil.HTTPError(resp, err)
+		return
+	}
+	writeResponse(resp, contentType, http.StatusOK, responseBody)
 }
 
 func readContentType(resp http.ResponseWriter, req *http.Request) (encoder.Encoder, bool) {

receiver/libhoneyreceiver/receiver_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/libhoneyevent"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/libhoneyreceiver/internal/response"
 )
 
 func TestNewLibhoneyReceiver(t *testing.T) {
@@ -89,11 +90,12 @@ func TestLibhoneyReceiver_Start(t *testing.T) {
 func TestLibhoneyReceiver_HandleEvent(t *testing.T) {
 	now := time.Now()
 	tests := []struct {
-		name           string
-		events         []libhoneyevent.LibhoneyEvent
-		contentType    string
-		expectedStatus int
-		wantError      bool
+		name             string
+		events           []libhoneyevent.LibhoneyEvent
+		contentType      string
+		expectedStatus   int
+		expectedResponse []response.ResponseInBatch
+		wantError        bool
 	}{
 		{
 			name: "valid_json_event",
@@ -109,6 +111,9 @@ func TestLibhoneyReceiver_HandleEvent(t *testing.T) {
 			},
 			contentType:    "application/json",
 			expectedStatus: http.StatusOK,
+			expectedResponse: []response.ResponseInBatch{
+				{Status: http.StatusAccepted},
+			},
 		},
 		{
 			name: "valid_msgpack_event",
@@ -126,11 +131,12 @@ func TestLibhoneyReceiver_HandleEvent(t *testing.T) {
 			expectedStatus: http.StatusOK,
 		},
 		{
-			name:           "invalid_content_type",
-			events:         []libhoneyevent.LibhoneyEvent{},
-			contentType:    "text/plain",
-			expectedStatus: http.StatusUnsupportedMediaType,
-			wantError:      true,
+			name:             "invalid_content_type",
+			events:           []libhoneyevent.LibhoneyEvent{},
+			contentType:      "text/plain",
+			expectedStatus:   http.StatusUnsupportedMediaType,
+			wantError:        true,
+			expectedResponse: nil,
 		},
 	}
 
@@ -168,6 +174,13 @@ func TestLibhoneyReceiver_HandleEvent(t *testing.T) {
 				assert.Eventually(t, func() bool {
 					return sink.LogRecordCount() > 0
 				}, time.Second, 10*time.Millisecond)
+				assert.Equal(t, tt.contentType, resp.Header.Get("Content-Type"))
+			}
+			if tt.expectedResponse != nil {
+				var actualResponse []response.ResponseInBatch
+				err = json.NewDecoder(resp.Body).Decode(&actualResponse)
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedResponse, actualResponse)
 			}
 		})
 	}