[receiver/splunkhec] Drop events when nested indexed fields are present

.chloggen/handle_nested_map_splunkhec_receiver.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receiver/splunkhec
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Return 400 status code when nested indextime fields are present
+
+# One or more tracking issues related to the change
+issues: [17308]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

receiver/splunkhecreceiver/receiver.go

@@ -51,7 +51,7 @@ const (
 	responseErrInternalServerError    = "Internal Server Error"
 	responseErrUnsupportedMetricEvent = "Unsupported metric event"
 	responseErrUnsupportedLogEvent    = "Unsupported log event"
-
+	responseErrHandlingIndexedFields  = `{"text":"Error in handling indexed fields","code":15,"invalid-event-number":%d}`
 	// Centralizing some HTTP and related string constants.
 	gzipEncoding              = "gzip"
 	httpContentEncodingHeader = "Content-Encoding"
@@ -341,6 +341,13 @@ func (r *splunkReceiver) handleReq(resp http.ResponseWriter, req *http.Request)
 			r.failRequest(ctx, resp, http.StatusBadRequest, errUnmarshalBodyRespBody, len(events), err)
 			return
 		}
+
+		for _, v := range msg.Fields {
+			if !isFlatJSONField(v) {
+				r.failRequest(ctx, resp, http.StatusBadRequest, []byte(fmt.Sprintf(responseErrHandlingIndexedFields, len(events))), len(events), nil)
+				return
+			}
+		}
 		if msg.IsMetric() {
 			if r.metricsConsumer == nil {
 				r.failRequest(ctx, resp, http.StatusBadRequest, errUnsupportedMetricEvent, len(events), err)
@@ -458,3 +465,18 @@ func initJSONResponse(s string) []byte {
 	}
 	return respBody
 }
+
+func isFlatJSONField(field interface{}) bool {
+	switch value := field.(type) {
+	case map[string]interface{}:
+		return false
+	case []interface{}:
+		for _, v := range value {
+			switch v.(type) {
+			case map[string]interface{}, []interface{}:
+				return false
+			}
+		}
+	}
+	return true
+}

receiver/splunkhecreceiver/receiver_test.go

@@ -28,6 +28,7 @@ import (
 	"testing"
 	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
@@ -1004,6 +1005,77 @@ func Test_splunkhecreceiver_handleHealthPath(t *testing.T) {
 	assert.Equal(t, 200, resp.StatusCode)
 }
 
+func Test_splunkhecreceiver_handle_nested_fields(t *testing.T) {
+	tests := []struct {
+		name    string
+		field   interface{}
+		success bool
+	}{
+		{
+			name:    "map",
+			field:   map[string]interface{}{},
+			success: false,
+		},
+		{
+			name:    "flat_array",
+			field:   []interface{}{1, 2, 3},
+			success: true,
+		},
+		{
+			name:    "nested_array",
+			field:   []interface{}{1, []interface{}{1, 2}},
+			success: false,
+		},
+		{
+			name: "array_of_map",
+			field: []interface{}{
+				map[string]interface{}{
+					"key": "value",
+				},
+			},
+			success: false,
+		},
+		{
+			name:    "int",
+			field:   int(0),
+			success: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			config := createDefaultConfig().(*Config)
+			sink := new(consumertest.LogsSink)
+			rcv, err := newLogsReceiver(receivertest.NewNopCreateSettings(), *config, sink)
+			assert.NoError(t, err)
+
+			r := rcv.(*splunkReceiver)
+			assert.NoError(t, r.Start(context.Background(), componenttest.NewNopHost()))
+			defer func() {
+				assert.NoError(t, r.Shutdown(context.Background()))
+			}()
+			currentTime := float64(time.Now().UnixNano()) / 1e6
+			event := buildSplunkHecMsg(currentTime, 3)
+			event.Fields["nested_map"] = tt.field
+			msgBytes, err := jsoniter.Marshal(event)
+			require.NoError(t, err)
+			req := httptest.NewRequest("POST", "http://localhost/services/collector", bytes.NewReader(msgBytes))
+
+			w := httptest.NewRecorder()
+			r.handleReq(w, req)
+
+			if tt.success {
+				assert.Equal(t, http.StatusOK, w.Code)
+				assert.Equal(t, 1, sink.LogRecordCount())
+			} else {
+				assert.Equal(t, http.StatusBadRequest, w.Code)
+				assert.Equal(t, fmt.Sprintf(responseErrHandlingIndexedFields, 0), w.Body.String())
+			}
+
+		})
+	}
+}
+
 func BenchmarkHandleReq(b *testing.B) {
 	config := createDefaultConfig().(*Config)
 	config.Endpoint = "localhost:0"