[encoding/googlecloudlogentry] Add encoding format to gcp encoding ext (open-telemetry#43759)

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

Add generic `encoding.format` to GCP encoding extension per the
discussion at
open-telemetry#43320

| **GCP Log Type** | **`cloud.resource_id`** | **new `encoding.format`
value** |
| ---|---|---|
| Activity Logs | one of `cloudaudit.googleapis.com%2Factivity`,
`cloudaudit.googleapis.com%2Fdata_access`,
`cloudaudit.googleapis.com%2Fpolicy`, or
`cloudaudit.googleapis.com%2Fsystem_event` | `gcp.auditlog` |
| VPC Flow Logs | one of `networkmanagement.googleapis.com%2Fvpc_flows`
or `compute.googleapis.com%2Fvpc_flows` | `gcp.vpcflow` |

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

Fixes
open-telemetry#43320

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

Unit tests added and updated

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

README updated

---------

Co-authored-by: Andrew Wilkins <[email protected]>

Author: joecompute

.chloggen/add-encoding-format-to-gcp-encoding-ext.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: extension/googlecloudlogentry_encoding
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add encoding.format attribute to GCP encoding extension to identify the source format.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43320]
+
+# (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: []

extension/encoding/googlecloudlogentryencodingextension/README.md

@@ -114,6 +114,41 @@ The severity is mapped from [Google Cloud Log Severity](https://cloud.google.com
 
 Currently, these are the log types that are specifically parsed into log record attributes.
 
+## Log Format Identification
+
+A subset of logs processed by this extension are automatically tagged with an `encoding.format` attribute at the scope level to identify the source format. This allows you to easily filter and route logs based on their Google Cloud service origin.
+
+The pattern used is `gcp.<format_name>`.
+
+Examples:
+- Audit Logs: `encoding.format: "gcp.auditlog"`
+- VPC Flow Logs: `encoding.format: "gcp.vpcflow"`
+
+### How encoding.format is determined
+
+The `encoding.format` attribute is automatically determined based on the log type extracted from the `logName` field. The extension uses the following logic:
+
+1. **Parse the logName**: The extension extracts the log type from the `logName` field.
+
+For example, `projects/my-project/logs/cloudaudit.googleapis.com%2Fsystem_event` is identified as a system event log via the log type suffix `cloudaudit.googleapis.com%2Fsystem_event`.
+
+2. **Map log type to format**: The extension maps specific log types to their corresponding encoding formats (`encoding.format`):
+   - Audit logs (activity, data access, system event, policy): `gcp.auditlog`
+   - VPC flow logs (network management-sourced and compute-sourced VPC flow logs): `gcp.vpcflow`
+
+3. **Set the attribute**: For recognized log types, the `encoding.format` attribute is set as an attribute of the `scope` field in the OTEL output log, allowing for flexible filtering and routing.
+
+For unrecognized log types, no `encoding.format` attribute is set.
+
+## Format Values
+
+The following format values are supported in the `googlecloudlogentryencodingextension` to identify different Google Cloud log types:
+
+| **GCP Log Type** | **Format Value** | **Description** |
+|------------------|------------------|-----------------|
+| Audit Logs | `auditlog` | Google Cloud audit logs (activity, data access, system event, policy) |
+| VPC Flow Logs | `vpcflow` | Virtual Private Cloud flow log records |
+
 ### Cloud Audit Logs
 
 See the struct of the Cloud Audit Log payload in [AuditLog](https://cloud.google.com/logging/docs/reference/audit/auditlog/rest/Shared.Types/AuditLog). The fields are mapped this way in the extension:

extension/encoding/googlecloudlogentryencodingextension/extension.go

@@ -61,8 +61,9 @@ func (ex *ext) handleLogLine(logs plog.Logs, logLine []byte) error {
 
 	rl := logs.ResourceLogs().AppendEmpty()
 	r := rl.Resource()
-	logRecord := rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
-	if err := handleLogEntryFields(r.Attributes(), logRecord, log, ex.config); err != nil {
+	scopeLogs := rl.ScopeLogs().AppendEmpty()
+
+	if err := handleLogEntryFields(r.Attributes(), scopeLogs, log, ex.config); err != nil {
 		return fmt.Errorf("failed to handle log entry: %w", err)
 	}
 

extension/encoding/googlecloudlogentryencodingextension/extension_test.go

@@ -5,6 +5,7 @@ package googlecloudlogentryencodingextension
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"testing"
 
@@ -13,6 +14,9 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/pdata/plog"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/auditlog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/constants"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/vpcflowlog"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
 )
@@ -212,3 +216,101 @@ func TestPayloads(t *testing.T) {
 		})
 	}
 }
+
+func TestEncodingFormatScopeAttributeUnknownLogs(t *testing.T) {
+	// note that testing for known log types is already done in TestPayloads,
+	// so we don't have tests here for known log types.
+	tests := []struct {
+		name           string
+		logName        string
+		expectedFormat string
+	}{
+		{
+			name:           "unknown log type",
+			logName:        "projects/test-project/logs/unknown-log-type",
+			expectedFormat: "",
+		},
+		{
+			name:           "generic log type",
+			logName:        "projects/test-project/logs/generic-log",
+			expectedFormat: "",
+		},
+	}
+
+	extension := newTestExtension(t, Config{})
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			// Create a minimal log entry that won't trigger payload processing errors
+			logLine := fmt.Appendf(nil, `{"logName": "%s", "timestamp": "2024-05-05T10:31:19.45570687Z", "textPayload": "test message"}`, tt.logName)
+
+			logs, err := extension.UnmarshalLogs(logLine)
+			require.NoError(t, err)
+
+			require.Equal(t, 1, logs.ResourceLogs().Len())
+			resourceLogs := logs.ResourceLogs().At(0)
+			require.Equal(t, 1, resourceLogs.ScopeLogs().Len())
+			scopeLogs := resourceLogs.ScopeLogs().At(0)
+
+			scopeAttrs := scopeLogs.Scope().Attributes()
+
+			_, exists := scopeAttrs.Get(constants.FormatIdentificationTag)
+			require.False(t, exists, "encoding.format attribute should not exist for unknown log types")
+		})
+	}
+}
+
+func TestGetEncodingFormatFunction(t *testing.T) {
+	tests := []struct {
+		name           string
+		logType        string
+		expectedFormat string
+	}{
+		{
+			name:           "audit log activity",
+			logType:        auditlog.ActivityLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log data access",
+			logType:        auditlog.DataAccessLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log system event",
+			logType:        auditlog.SystemEventLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log policy",
+			logType:        auditlog.PolicyLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "vpc flow log network management",
+			logType:        vpcflowlog.NetworkManagementNameSuffix,
+			expectedFormat: constants.GCPFormatVPCFlowLog,
+		},
+		{
+			name:           "vpc flow log compute",
+			logType:        vpcflowlog.ComputeNameSuffix,
+			expectedFormat: constants.GCPFormatVPCFlowLog,
+		},
+		{
+			name:           "unknown log type",
+			logType:        "unknown-log-type",
+			expectedFormat: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			result := getEncodingFormat(tt.logType)
+			require.Equal(t, tt.expectedFormat, result)
+		})
+	}
+}

extension/encoding/googlecloudlogentryencodingextension/internal/constants/format.go

@@ -0,0 +1,16 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package constants // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/constants"
+
+const (
+	FormatAuditLog   = "auditlog"
+	FormatVPCFlowLog = "vpcflow"
+
+	FormatIdentificationTag = "encoding.format"
+
+	// GCP-specific format prefixes
+	GCPFormatPrefix     = "gcp."
+	GCPFormatAuditLog   = GCPFormatPrefix + FormatAuditLog
+	GCPFormatVPCFlowLog = GCPFormatPrefix + FormatVPCFlowLog
+)

extension/encoding/googlecloudlogentryencodingextension/log_entry.go

@@ -19,6 +19,7 @@ import (
 	ltype "google.golang.org/genproto/googleapis/logging/type"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/auditlog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/constants"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/shared"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/vpcflowlog"
 )
@@ -62,6 +63,22 @@ const (
 	gcpAppHubWorkloadCriticalityTypeField = "workload.criticality_type"
 )
 
+// getEncodingFormat maps GCP log types to encoding format values
+func getEncodingFormat(logType string) string {
+	switch logType {
+	case auditlog.ActivityLogNameSuffix,
+		auditlog.DataAccessLogNameSuffix,
+		auditlog.SystemEventLogNameSuffix,
+		auditlog.PolicyLogNameSuffix:
+		return constants.GCPFormatAuditLog
+	case vpcflowlog.NetworkManagementNameSuffix,
+		vpcflowlog.ComputeNameSuffix:
+		return constants.GCPFormatVPCFlowLog
+	default:
+		return ""
+	}
+}
+
 // See: https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry
 type logEntry struct {
 	ProtoPayload gojson.RawMessage `json:"protoPayload"`
@@ -457,18 +474,18 @@ func handleLogNameField(logName string, resourceAttr pcommon.Map) (string, error
 	}
 }
 
-func handlePayload(logType string, log logEntry, logRecord plog.LogRecord, cfg Config) error {
-	switch logType {
-	case auditlog.ActivityLogNameSuffix,
-		auditlog.DataAccessLogNameSuffix,
-		auditlog.SystemEventLogNameSuffix,
-		auditlog.PolicyLogNameSuffix:
+func handlePayload(encodingFormat string, log logEntry, logRecord plog.LogRecord, scope pcommon.InstrumentationScope, cfg Config) error {
+	switch encodingFormat {
+	case constants.GCPFormatAuditLog:
+		// Add encoding.format to scope attributes for audit logs
+		scope.Attributes().PutStr(constants.FormatIdentificationTag, encodingFormat)
 		if err := auditlog.ParsePayloadIntoAttributes(log.ProtoPayload, logRecord.Attributes()); err != nil {
 			return fmt.Errorf("failed to parse audit log proto payload: %w", err)
 		}
 		return nil
-	case vpcflowlog.NetworkManagementNameSuffix,
-		vpcflowlog.ComputeNameSuffix:
+	case constants.GCPFormatVPCFlowLog:
+		// Add encoding.format to scope attributes for VPC flow logs
+		scope.Attributes().PutStr(constants.FormatIdentificationTag, encodingFormat)
 		if err := vpcflowlog.ParsePayloadIntoAttributes(log.JSONPayload, logRecord.Attributes()); err != nil {
 			return fmt.Errorf("failed to parse VPC flow log JSON payload: %w", err)
 		}
@@ -496,7 +513,10 @@ func handlePayload(logType string, log logEntry, logRecord plog.LogRecord, cfg C
 
 // handleLogEntryFields will place each entry of logEntry as either an attribute of the log,
 // or as part of the log body, in case of payload.
-func handleLogEntryFields(resourceAttributes pcommon.Map, logRecord plog.LogRecord, log logEntry, cfg Config) error {
+func handleLogEntryFields(resourceAttributes pcommon.Map, scopeLogs plog.ScopeLogs, log logEntry, cfg Config) error {
+	logRecord := scopeLogs.LogRecords().AppendEmpty()
+	scope := scopeLogs.Scope()
+
 	ts := log.Timestamp
 	if ts == nil {
 		return errors.New("missing timestamp")
@@ -509,11 +529,14 @@ func handleLogEntryFields(resourceAttributes pcommon.Map, logRecord plog.LogReco
 
 	shared.PutStr(string(semconv.LogRecordUIDKey), log.InsertID, logRecord.Attributes())
 
+	// Handle log name, get type and encoding format
 	logType, errLogName := handleLogNameField(log.LogName, resourceAttributes)
 	if errLogName != nil {
 		return fmt.Errorf("failed to handle log name field: %w", errLogName)
 	}
-	if err := handlePayload(logType, log, logRecord, cfg); err != nil {
+	encodingFormat := getEncodingFormat(logType)
+
+	if err := handlePayload(encodingFormat, log, logRecord, scope, cfg); err != nil {
 		return fmt.Errorf("failed to handle payload field: %w", err)
 	}
 

extension/encoding/googlecloudlogentryencodingextension/log_entry_test.go

@@ -14,6 +14,9 @@ import (
 	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
 	ltype "google.golang.org/genproto/googleapis/logging/type"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/auditlog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/constants"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/googlecloudlogentryencodingextension/internal/vpcflowlog"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
 )
@@ -382,17 +385,76 @@ func TestHandleLogEntryFields(t *testing.T) {
 	logs := plog.NewLogs()
 	resourceLogs := logs.ResourceLogs().AppendEmpty()
 	resource := resourceLogs.Resource()
-	logRecord := resourceLogs.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+	scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
 	cfg := *createDefaultConfig().(*Config)
 
-	err = handleLogEntryFields(resource.Attributes(), logRecord, l, cfg)
+	// add the rest of the log entry fields
+	err = handleLogEntryFields(resource.Attributes(), scopeLogs, l, cfg)
 	require.NoError(t, err)
 
 	expected, err := golden.ReadLogs("testdata/log_entry_expected.yaml")
 	require.NoError(t, err)
 	require.NoError(t, plogtest.CompareLogs(expected, logs))
 }
 
+func TestGetEncodingFormat(t *testing.T) {
+	tests := []struct {
+		name           string
+		logType        string
+		expectedFormat string
+	}{
+		{
+			name:           "audit log activity",
+			logType:        auditlog.ActivityLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log data access",
+			logType:        auditlog.DataAccessLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log system event",
+			logType:        auditlog.SystemEventLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "audit log policy",
+			logType:        auditlog.PolicyLogNameSuffix,
+			expectedFormat: constants.GCPFormatAuditLog,
+		},
+		{
+			name:           "vpc flow log network management",
+			logType:        vpcflowlog.NetworkManagementNameSuffix,
+			expectedFormat: constants.GCPFormatVPCFlowLog,
+		},
+		{
+			name:           "vpc flow log compute",
+			logType:        vpcflowlog.ComputeNameSuffix,
+			expectedFormat: constants.GCPFormatVPCFlowLog,
+		},
+		{
+			name:           "unknown log type",
+			logType:        "unknown-log-type",
+			expectedFormat: "",
+		},
+		{
+			name:           "empty log type",
+			logType:        "",
+			expectedFormat: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			result := getEncodingFormat(tt.logType)
+			require.Equal(t, tt.expectedFormat, result)
+		})
+	}
+}
+
 func TestHandleJSONPayload(t *testing.T) {
 	tests := []struct {
 		name        string