[pkg/translator/azure] Decode "incorrect" JSON without failing completely

.chloggen/azuretranslator-allow-numeric-values.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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: azuretranslator
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Allow numeric fields to use a String or Integer representation in JSON.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [28650]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

pkg/translator/azure/resourcelogs_to_logs.go

@@ -5,6 +5,7 @@ package azure // import "github.com/open-telemetry/opentelemetry-collector-contr
 
 import (
 	"bytes"
+	"encoding/json"
 	"strconv"
 
 	jsoniter "github.com/json-iterator/go"
@@ -54,11 +55,11 @@ type azureLogRecord struct {
 	ResultType        *string      `json:"resultType"`
 	ResultSignature   *string      `json:"resultSignature"`
 	ResultDescription *string      `json:"resultDescription"`
-	DurationMs        *string      `json:"durationMs"`
+	DurationMs        *json.Number `json:"durationMs"`
 	CallerIPAddress   *string      `json:"callerIpAddress"`
 	CorrelationID     *string      `json:"correlationId"`
 	Identity          *interface{} `json:"identity"`
-	Level             *string      `json:"Level"`
+	Level             *json.Number `json:"Level"`
 	Location          *string      `json:"location"`
 	Properties        *interface{} `json:"properties"`
 }
@@ -112,7 +113,7 @@ func (r ResourceLogsUnmarshaler) UnmarshalLogs(buf []byte) (plog.Logs, error) {
 			if log.Level != nil {
 				severity := asSeverity(*log.Level)
 				lr.SetSeverityNumber(severity)
-				lr.SetSeverityText(*log.Level)
+				lr.SetSeverityText(log.Level.String())
 			}
 
 			if err := lr.Attributes().FromRaw(extractRawAttributes(log)); err != nil {
@@ -138,8 +139,8 @@ func asTimestamp(s string) (pcommon.Timestamp, error) {
 // asSeverity converts the Azure log level to equivalent
 // OpenTelemetry severity numbers. If the log level is not
 // valid, then the 'Unspecified' value is returned.
-func asSeverity(s string) plog.SeverityNumber {
-	switch s {
+func asSeverity(number json.Number) plog.SeverityNumber {
+	switch number.String() {
 	case "Informational":
 		return plog.SeverityNumberInfo
 	case "Warning":
@@ -149,6 +150,11 @@ func asSeverity(s string) plog.SeverityNumber {
 	case "Critical":
 		return plog.SeverityNumberFatal
 	default:
+		var levelNumber, _ = number.Int64()
+		if levelNumber > 0 {
+			return plog.SeverityNumber(levelNumber)
+		}
+
 		return plog.SeverityNumberUnspecified
 	}
 }
@@ -159,7 +165,7 @@ func extractRawAttributes(log azureLogRecord) map[string]interface{} {
 	attrs[azureCategory] = log.Category
 	setIf(attrs, azureCorrelationID, log.CorrelationID)
 	if log.DurationMs != nil {
-		duration, err := strconv.ParseInt(*log.DurationMs, 10, 64)
+		duration, err := strconv.ParseInt(log.DurationMs.String(), 10, 64)
 		if err == nil {
 			attrs[azureDuration] = duration
 		}

pkg/translator/azure/resourcelogs_to_logs_test.go

@@ -4,6 +4,7 @@
 package azure // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure"
 
 import (
+	"encoding/json"
 	"os"
 	"path/filepath"
 	"testing"
@@ -126,6 +127,46 @@ var maximumLogRecord2 = func() []plog.LogRecord {
 	return append(records, lr, lr2)
 }()
 
+var badLevelLogRecord = func() plog.LogRecord {
+	lr := plog.NewLogs().ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+	ts, _ := asTimestamp("2023-10-26T14:22:43.3416357Z")
+	lr.SetTimestamp(ts)
+	lr.SetSeverityNumber(plog.SeverityNumberTrace4)
+	lr.SetSeverityText("4")
+	guid := "128bc026-5ead-40c7-8853-ebb32bc077a3"
+
+	lr.Attributes().PutStr(azureOperationName, "Microsoft.ApiManagement/GatewayLogs")
+	lr.Attributes().PutStr(azureCategory, "GatewayLogs")
+	lr.Attributes().PutStr(azureCorrelationID, guid)
+	lr.Attributes().PutStr(azureResultType, "Succeeded")
+	lr.Attributes().PutInt(azureDuration, 243)
+	lr.Attributes().PutStr(conventions.AttributeNetSockPeerAddr, "13.14.15.16")
+	lr.Attributes().PutStr(conventions.AttributeCloudRegion, "West US")
+	lr.Attributes().PutStr(conventions.AttributeCloudProvider, conventions.AttributeCloudProviderAzure)
+
+	m := lr.Attributes().PutEmptyMap(azureProperties)
+	m.PutStr("method", "GET")
+	m.PutStr("url", "https://api.azure-api.net/sessions")
+	m.PutDouble("backendResponseCode", 200)
+	m.PutDouble("responseCode", 200)
+	m.PutDouble("responseSize", 102945)
+	m.PutStr("cache", "none")
+	m.PutDouble("backendTime", 54)
+	m.PutDouble("requestSize", 632)
+	m.PutStr("apiId", "demo-api")
+	m.PutStr("operationId", "GetSessions")
+	m.PutStr("apimSubscriptionId", "master")
+	m.PutDouble("clientTime", 190)
+	m.PutStr("clientProtocol", "HTTP/1.1")
+	m.PutStr("backendProtocol", "HTTP/1.1")
+	m.PutStr("apiRevision", "1")
+	m.PutStr("clientTlsVersion", "1.2")
+	m.PutStr("backendMethod", "GET")
+	m.PutStr("backendUrl", "https://api.azurewebsites.net/sessions")
+	return lr
+}()
+
 func TestAsTimestamp(t *testing.T) {
 	timestamp := "2022-11-11T04:48:27.6767145Z"
 	nanos, err := asTimestamp(timestamp)
@@ -149,7 +190,7 @@ func TestAsSeverity(t *testing.T) {
 
 	for input, expected := range tests {
 		t.Run(input, func(t *testing.T) {
-			assert.Equal(t, expected, asSeverity(input))
+			assert.Equal(t, expected, asSeverity(json.Number(input)))
 		})
 	}
 }
@@ -176,8 +217,8 @@ func TestSetIf(t *testing.T) {
 }
 
 func TestExtractRawAttributes(t *testing.T) {
-	badDuration := "invalid"
-	goodDuration := "1234"
+	badDuration := json.Number("invalid")
+	goodDuration := json.Number("1234")
 
 	tenantID := "tenant.id"
 	operationVersion := "operation.version"
@@ -186,7 +227,7 @@ func TestExtractRawAttributes(t *testing.T) {
 	resultDescription := "result.description"
 	callerIPAddress := "127.0.0.1"
 	correlationID := "edb70d1a-eec2-4b4c-b2f4-60e3510160ee"
-	level := "Informational"
+	level := json.Number("Informational")
 	location := "location"
 
 	identity := interface{}("someone")
@@ -321,6 +362,15 @@ func TestUnmarshalLogs(t *testing.T) {
 	maximumLogRecord2[0].CopyTo(lr)
 	maximumLogRecord2[1].CopyTo(lr2)
 
+	expectedBadLevel := plog.NewLogs()
+	resourceLogs = expectedBadLevel.ResourceLogs().AppendEmpty()
+	resourceLogs.Resource().Attributes().PutStr(azureResourceID, "/RESOURCE_ID")
+	scopeLogs = resourceLogs.ScopeLogs().AppendEmpty()
+	scopeLogs.Scope().SetName("otelcol/azureresourcelogs")
+	scopeLogs.Scope().SetVersion(testBuildInfo.Version)
+	lr = scopeLogs.LogRecords().AppendEmpty()
+	badLevelLogRecord.CopyTo(lr)
+
 	tests := []struct {
 		file     string
 		expected plog.Logs
@@ -337,6 +387,10 @@ func TestUnmarshalLogs(t *testing.T) {
 			file:     "log-maximum.json",
 			expected: expectedMaximum,
 		},
+		{
+			file:     "log-bad-level.json",
+			expected: expectedBadLevel,
+		},
 	}
 
 	sut := &ResourceLogsUnmarshaler{

pkg/translator/azure/testdata/log-bad-level.json

@@ -0,0 +1,39 @@
+{
+    "records": [
+        {
+            "DeploymentVersion": "0.40.16708.0",
+            "Level": 4,
+            "isRequestSuccess": true,
+            "time": "2023-10-26T14:22:43.3416357Z",
+            "operationName": "Microsoft.ApiManagement/GatewayLogs",
+            "category": "GatewayLogs",
+            "durationMs": 243,
+            "callerIpAddress": "13.14.15.16",
+            "correlationId": "128bc026-5ead-40c7-8853-ebb32bc077a3",
+            "location": "West US",
+            "properties": {
+                "method": "GET",
+                "url": "https://api.azure-api.net/sessions",
+                "backendResponseCode": 200,
+                "responseCode": 200,
+                "responseSize": 102945,
+                "cache": "none",
+                "backendTime": 54,
+                "requestSize": 632,
+                "apiId": "demo-api",
+                "operationId": "GetSessions",
+                "apimSubscriptionId": "master",
+                "clientTime": 190,
+                "clientProtocol": "HTTP/1.1",
+                "backendProtocol": "HTTP/1.1",
+                "apiRevision": "1",
+                "clientTlsVersion": "1.2",
+                "backendMethod": "GET",
+                "backendUrl": "https://api.azurewebsites.net/sessions"
+            },
+            "resourceId": "/RESOURCE_ID",
+            "resultType": "Succeeded",
+            "truncated": 0
+        }
+    ]
+}

receiver/azureeventhubreceiver/eventhubhandler.go

@@ -161,6 +161,7 @@ func (h *eventhubHandler) newMessageHandler(ctx context.Context, event *eventhub
 
 	err := h.dataConsumer.consume(ctx, event)
 	if err != nil {
+		h.settings.Logger.Error("error decoding message", zap.Error(err))
 		return err
 	}