add fuzz tests to multiple receivers and processors (#35715)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Adds fuzz tests for receiver handlers and processing of metrics, logs
and traces. For receivers, the fuzz test will create an HTTP request
with the fuzzers testcase as its body. For processors, the fuzzer will
create random logs, metrics or traces and pass it to a routine that
processes the data - for the most part either a
`processLogs`/`processMetrics`/`processTraces` method or
`ConsumeLogs`/`ConsumeTraces` which will then invoke the internal
`process` routine.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Signed-off-by: Adam Korczynski <adam@adalogics.com>
Co-authored-by: Pablo Baeyens <pbaeyens31+github@gmail.com>

processor/groupbyattrsprocessor/fuzz_test.go

@@ -0,0 +1,59 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package groupbyattrsprocessor
+
+import (
+	"context"
+	"testing"
+
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func FuzzProcessTraces(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &ptrace.JSONUnmarshaler{}
+		traces, err := ju.UnmarshalTraces(data)
+		if err != nil {
+			return
+		}
+		gap, err := createGroupByAttrsProcessor(processortest.NewNopSettings(), []string{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _ = gap.processTraces(context.Background(), traces)
+	})
+}
+
+func FuzzProcessLogs(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &plog.JSONUnmarshaler{}
+		logs, err := ju.UnmarshalLogs(data)
+		if err != nil {
+			return
+		}
+		gap, err := createGroupByAttrsProcessor(processortest.NewNopSettings(), []string{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _ = gap.processLogs(context.Background(), logs)
+	})
+}
+
+func FuzzProcessMetrics(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &pmetric.JSONUnmarshaler{}
+		metrics, err := ju.UnmarshalMetrics(data)
+		if err != nil {
+			return
+		}
+		gap, err := createGroupByAttrsProcessor(processortest.NewNopSettings(), []string{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _ = gap.processMetrics(context.Background(), metrics)
+	})
+}

processor/logdedupprocessor/fuzz_test.go

@@ -0,0 +1,31 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package logdedupprocessor
+
+import (
+	"context"
+	"testing"
+
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func FuzzConsumeLogs(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &plog.JSONUnmarshaler{}
+		logs, err := ju.UnmarshalLogs(data)
+		if err != nil {
+			return
+		}
+		sink := new(consumertest.LogsSink)
+		set := processortest.NewNopSettings()
+		cfg := &Config{}
+		lp, err := newProcessor(cfg, sink, set)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = lp.ConsumeLogs(context.Background(), logs)
+	})
+}

processor/probabilisticsamplerprocessor/fuzz_test.go

@@ -0,0 +1,49 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package probabilisticsamplerprocessor
+
+import (
+	"context"
+	"testing"
+
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func FuzzConsumeTraces(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &ptrace.JSONUnmarshaler{}
+		traces, err := ju.UnmarshalTraces(data)
+		if err != nil {
+			return
+		}
+		sink := new(consumertest.TracesSink)
+		set := processortest.NewNopSettings()
+		cfg := &Config{}
+		tsp, err := newTracesProcessor(context.Background(), set, cfg, sink)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = tsp.ConsumeTraces(context.Background(), traces)
+	})
+}
+
+func FuzzConsumeLogs(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &plog.JSONUnmarshaler{}
+		logs, err := ju.UnmarshalLogs(data)
+		if err != nil {
+			return
+		}
+		nextConsumer := consumertest.NewNop()
+		cfg := &Config{}
+		lp, err := newLogsProcessor(context.Background(), processortest.NewNopSettings(), nextConsumer, cfg)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = lp.ConsumeLogs(context.Background(), logs)
+	})
+}

processor/sumologicprocessor/fuzz_test.go

@@ -0,0 +1,84 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sumologicprocessor
+
+import (
+	"testing"
+
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func FuzzProcessTraces(f *testing.F) {
+	f.Fuzz(func(_ *testing.T, data []byte, processorType uint8) {
+		ju := &ptrace.JSONUnmarshaler{}
+		traces, err := ju.UnmarshalTraces(data)
+		if err != nil {
+			return
+		}
+		switch int(processorType) % 4 {
+		case 0:
+			proc := &aggregateAttributesProcessor{}
+			_ = proc.processTraces(traces)
+		case 1:
+			proc := &cloudNamespaceProcessor{}
+			_ = proc.processTraces(traces)
+		case 2:
+			proc := &NestingProcessor{}
+			_ = proc.processTraces(traces)
+		case 3:
+			proc := &translateAttributesProcessor{}
+			_ = proc.processTraces(traces)
+		}
+	})
+}
+
+func FuzzProcessLogs(f *testing.F) {
+	f.Fuzz(func(_ *testing.T, data []byte, processorType uint8) {
+		ju := &plog.JSONUnmarshaler{}
+		logs, err := ju.UnmarshalLogs(data)
+		if err != nil {
+			return
+		}
+		switch int(processorType) % 4 {
+		case 0:
+			proc := &aggregateAttributesProcessor{}
+			_ = proc.processLogs(logs)
+		case 1:
+			proc := &cloudNamespaceProcessor{}
+			_ = proc.processLogs(logs)
+		case 2:
+			proc := &NestingProcessor{}
+			_ = proc.processLogs(logs)
+		case 3:
+			proc := &translateAttributesProcessor{}
+			_ = proc.processLogs(logs)
+		}
+	})
+}
+
+func FuzzProcessMetrics(f *testing.F) {
+	f.Fuzz(func(_ *testing.T, data []byte, processorType uint8) {
+		ju := &pmetric.JSONUnmarshaler{}
+		metrics, err := ju.UnmarshalMetrics(data)
+		if err != nil {
+			return
+		}
+		switch int(processorType) % 4 {
+		case 0:
+			proc := &aggregateAttributesProcessor{}
+			_ = proc.processMetrics(metrics)
+		case 1:
+			proc := &cloudNamespaceProcessor{}
+			_ = proc.processMetrics(metrics)
+		case 2:
+			proc := &NestingProcessor{}
+			_ = proc.processMetrics(metrics)
+		case 3:
+			proc := &translateAttributesProcessor{}
+			_ = proc.processMetrics(metrics)
+		}
+	})
+}

processor/tailsamplingprocessor/fuzz_test.go

@@ -0,0 +1,31 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package tailsamplingprocessor
+
+import (
+	"context"
+	"testing"
+
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func FuzzConsumeTraces(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ju := &ptrace.JSONUnmarshaler{}
+		traces, err := ju.UnmarshalTraces(data)
+		if err != nil {
+			return
+		}
+		sink := new(consumertest.TracesSink)
+		set := processortest.NewNopSettings()
+		cfg := &Config{}
+		tsp, err := newTracesProcessor(context.Background(), set, sink, *cfg)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = tsp.ConsumeTraces(context.Background(), traces)
+	})
+}

receiver/cloudflarereceiver/fuzz_test.go

@@ -0,0 +1,45 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cloudflarereceiver
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"go.opentelemetry.io/collector/config/configtls"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+)
+
+func FuzzHandleReq(f *testing.F) {
+	f.Fuzz(func(t *testing.T, reqBody []byte, gZip bool) {
+		req, err := http.NewRequest(http.MethodPost, "http://example.com", bytes.NewReader(reqBody))
+		if err != nil {
+			t.Skip()
+		}
+		req.Header.Add(secretHeaderName, "abc123")
+		req.Header.Add("Content-Type", "text/plain; charset=utf-8")
+		if gZip {
+			req.Header.Add("Content-Encoding", "gzip")
+		}
+		consumer := &consumertest.LogsSink{}
+
+		r := newReceiver(t, &Config{
+			Logs: LogsConfig{
+				Endpoint:       "localhost:0",
+				Secret:         "abc123",
+				TimestampField: "MyTimestamp",
+				Attributes: map[string]string{
+					"ClientIP": "http_request.client_ip",
+				},
+				TLS: &configtls.ServerConfig{},
+			},
+		},
+			consumer,
+		)
+		rec := httptest.NewRecorder()
+		r.handleRequest(rec, req)
+	})
+}

receiver/lokireceiver/internal/fuzz_test.go

@@ -0,0 +1,28 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal
+
+import (
+	"bytes"
+	"net/http"
+	"testing"
+)
+
+func FuzzParseRequest(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte, headerType uint8) {
+		req, err := http.NewRequest(http.MethodPost, "http://example.com", bytes.NewReader(data))
+		if err != nil {
+			t.Skip()
+		}
+		switch int(headerType) % 3 {
+		case 0:
+			req.Header.Add("Content-Encoding", "snappy")
+		case 1:
+			req.Header.Add("Content-Encoding", "gzip")
+		case 2:
+			req.Header.Add("Content-Encoding", "deflat")
+		}
+		_, _ = ParseRequest(req)
+	})
+}

receiver/mongodbatlasreceiver/fuzz_test.go

@@ -0,0 +1,35 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package mongodbatlasreceiver
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+	"go.uber.org/zap/zaptest"
+)
+
+func FuzzHandleReq(f *testing.F) {
+	f.Fuzz(func(t *testing.T, reqBody []byte, payloadSigHeader string) {
+		req, err := http.NewRequest(http.MethodPost, "http://example.com", bytes.NewReader(reqBody))
+		if err != nil {
+			t.Skip()
+		}
+		req.Header.Add(signatureHeaderName, payloadSigHeader)
+		consumer := &consumertest.LogsSink{}
+
+		set := receivertest.NewNopSettings()
+		set.Logger = zaptest.NewLogger(t)
+		ar, err := newAlertsReceiver(set, &Config{Alerts: AlertConfig{Secret: "some_secret"}}, consumer)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		ar.handleRequest(rec, req)
+	})
+}