Set aws.vpc.flow.start timestamp field to ISO8601 format

.chloggen/chore_set-timestamp-field-to-iso8601-format.yaml

@@ -0,0 +1,31 @@
+# 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. filelogreceiver)
+component: extension/awslogs_encoding
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add feature gate to set aws.vpc.flow.start timestamp field to ISO8601 format
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43392]
+
+# (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: |
+  Feature gate ID: extension.awslogsencoding.vpcflow.start.iso8601
+  When enabled, the aws.vpc.flow.start field will be formatted as an ISO-8601 string 
+  instead of a Unix timestamp integer in seconds since epoch. Default behavior remains unchanged for backward compatibility.
+  Enable with: --feature-gates=extension.awslogsencoding.vpcflow.start.iso8601
+
+# 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/awslogsencodingextension/README.md

@@ -111,6 +111,31 @@ The following format values are supported in the `awslogsencodingextension` to i
 
 If you're using the old format values you should update the encoding extension configuration with the new format values.
 
+## Feature Gates
+
+### VPC Flow Log Start Field ISO-8601 Format
+
+**Feature Gate ID**: `extension.awslogsencoding.vpcflow.start.iso8601`
+
+**Stage**: Alpha
+
+**Description**: When enabled, the `aws.vpc.flow.start` field will be formatted as an ISO-8601 string instead of a Unix timestamp integer in seconds since epoch.
+
+**Default**: Disabled (legacy behavior)
+
+#### Behavior
+
+| **Feature Gate State** | **Field Type** | **Format** | **Example** |
+|------------------------|----------------|------------|-------------|
+| **Disabled (Default)** | `int64` | Unix seconds since epoch | `1609459200` |
+| **Enabled** | `string` | ISO-8601 with milliseconds | `"2021-01-01T00:00:00.000Z"` |
+
+#### Enabling the Feature Gate
+
+**Command Line:**
+```bash
+--feature-gates=extension.awslogsencoding.vpcflow.start.iso8601
+```
 
 ## Produced Records per Format
 

extension/encoding/awslogsencodingextension/extension.go

@@ -13,6 +13,7 @@ import (
 	"github.com/klauspost/compress/gzip"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension"
+	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.uber.org/zap"
 
@@ -33,6 +34,17 @@ const (
 	parquetEncoding = "parquet"
 )
 
+var vpcFlowStartISO8601FormatFeatureGate *featuregate.Gate
+
+func init() {
+	vpcFlowStartISO8601FormatFeatureGate = featuregate.GlobalRegistry().MustRegister(
+		constants.VPCFlowStartISO8601FormatID,
+		featuregate.StageAlpha,
+		featuregate.WithRegisterDescription("When enabled, aws.vpc.flow.start field will be formatted as ISO-8601 string instead of seconds since epoch integer."),
+		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/43390"),
+	)
+}
+
 var _ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
 
 type encodingExtension struct {
@@ -72,6 +84,7 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 			fileFormat,
 			settings.BuildInfo,
 			settings.Logger,
+			vpcFlowStartISO8601FormatFeatureGate.IsEnabled(),
 		)
 		return &encodingExtension{
 			unmarshaler: unmarshaler,

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

@@ -22,4 +22,7 @@ const (
 	FileFormatPlainText     = "plain-text"
 	FileFormatParquet       = "parquet"
 	FormatIdentificationTag = "encoding.format"
+
+	// Feature gate for VPC flow start field ISO-8601 format
+	VPCFlowStartISO8601FormatID = "extension.awslogsencoding.vpcflow.start.iso8601"
 )

extension/encoding/awslogsencodingextension/internal/unmarshaler/vpc-flow-log/testdata/valid_vpc_flow_log_expected_iso8601.yaml

@@ -0,0 +1,32 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: cloud.provider
+          value:
+            stringValue: aws
+        - key: cloud.account.id
+          value:
+            stringValue: "12345678910"
+    scopeLogs:
+      - logRecords:
+          - attributes:
+              - key: aws.vpc.flow.log.version
+                value:
+                  intValue: "2"
+              - key: network.interface.name
+                value:
+                  stringValue: eni-0eb1e4178af74336c
+              - key: aws.vpc.flow.start
+                value:
+                  stringValue: "2025-03-21T15:14:49.000Z"
+              - key: aws.vpc.flow.status
+                value:
+                  stringValue: NODATA
+            body: {}
+            timeUnixNano: "1742570142000000000"
+        scope:
+          attributes:
+            - key: encoding.format
+              value:
+                stringValue: aws.vpcflow
+          name: github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension

extension/encoding/awslogsencodingextension/internal/unmarshaler/vpc-flow-log/unmarshaler.go

@@ -34,12 +34,16 @@ type vpcFlowLogUnmarshaler struct {
 
 	buildInfo component.BuildInfo
 	logger    *zap.Logger
+
+	// Whether VPC flow start field should use ISO-8601 format
+	vpcFlowStartISO8601FormatEnabled bool
 }
 
 func NewVPCFlowLogUnmarshaler(
 	format string,
 	buildInfo component.BuildInfo,
 	logger *zap.Logger,
+	vpcFlowStartISO8601FormatEnabled bool,
 ) (unmarshaler.AWSUnmarshaler, error) {
 	switch format {
 	case constants.FileFormatParquet:
@@ -54,9 +58,10 @@ func NewVPCFlowLogUnmarshaler(
 		)
 	}
 	return &vpcFlowLogUnmarshaler{
-		fileFormat: format,
-		buildInfo:  buildInfo,
-		logger:     logger,
+		fileFormat:                       format,
+		buildInfo:                        buildInfo,
+		logger:                           logger,
+		vpcFlowStartISO8601FormatEnabled: vpcFlowStartISO8601FormatEnabled,
 	}, nil
 }
 
@@ -179,7 +184,7 @@ func (v *vpcFlowLogUnmarshaler) addToLogs(
 			continue
 		}
 
-		found, err := handleField(field, value, record, addr, key)
+		found, err := v.handleField(field, value, record, addr, key)
 		if err != nil {
 			return err
 		}
@@ -242,7 +247,7 @@ func (v *vpcFlowLogUnmarshaler) handleAddresses(addr *address, record plog.LogRe
 // adds its value to the resourceKey or puts the
 // field and its value in the attributes map. If the
 // field is not recognized, it returns false.
-func handleField(
+func (v *vpcFlowLogUnmarshaler) handleField(
 	field string,
 	value string,
 	record plog.LogRecord,
@@ -340,8 +345,17 @@ func handleField(
 			return false, err
 		}
 	case "start":
-		if err := addNumber(field, value, "aws.vpc.flow.start"); err != nil {
-			return false, err
+		unixSeconds, err := getNumber(value)
+		if err != nil {
+			return true, fmt.Errorf("value %s for field %s does not correspond to a valid timestamp", value, field)
+		}
+		if v.vpcFlowStartISO8601FormatEnabled {
+			// New behavior: ISO-8601 format
+			timestamp := time.Unix(unixSeconds, 0).UTC()
+			record.Attributes().PutStr("aws.vpc.flow.start", timestamp.Format("2006-01-02T15:04:05.000Z"))
+		} else {
+			// Legacy behavior: Unix timestamp as integer
+			record.Attributes().PutInt("aws.vpc.flow.start", unixSeconds)
 		}
 	case "end":
 		unixSeconds, err := getNumber(value)

extension/encoding/awslogsencodingextension/internal/unmarshaler/vpc-flow-log/unmarshaler_test.go

@@ -51,26 +51,35 @@ func TestUnmarshalLogs_PlainText(t *testing.T) {
 		reader               io.Reader
 		logsExpectedFilename string
 		expectedErr          string
+		featureGateEnabled   bool
 	}{
 		"valid_vpc_flow_log": {
 			reader:               readAndCompressLogFile(t, dir, "valid_vpc_flow_log.log"),
 			logsExpectedFilename: "valid_vpc_flow_log_expected.yaml",
+			featureGateEnabled:   false,
+		},
+		"valid_vpc_flow_log_iso8601": {
+			reader:               readAndCompressLogFile(t, dir, "valid_vpc_flow_log.log"),
+			logsExpectedFilename: "valid_vpc_flow_log_expected_iso8601.yaml",
+			featureGateEnabled:   true,
 		},
 		"vpc_flow_log_with_more_fields_than_allowed": {
-			reader:      readAndCompressLogFile(t, dir, "vpc_flow_log_too_few_fields.log"),
-			expectedErr: "log line has less fields than the ones expected",
+			reader:             readAndCompressLogFile(t, dir, "vpc_flow_log_too_few_fields.log"),
+			expectedErr:        "log line has less fields than the ones expected",
+			featureGateEnabled: false,
 		},
 		"vpc_flow_log_with_less_fields_than_required": {
-			reader:      readAndCompressLogFile(t, dir, "vpc_flow_log_too_many_fields.log"),
-			expectedErr: "log line has more fields than the ones expected",
+			reader:             readAndCompressLogFile(t, dir, "vpc_flow_log_too_many_fields.log"),
+			expectedErr:        "log line has more fields than the ones expected",
+			featureGateEnabled: false,
 		},
 	}
 
-	u, err := NewVPCFlowLogUnmarshaler(constants.FileFormatPlainText, component.BuildInfo{}, zap.NewNop())
-	require.NoError(t, err)
-
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
+			u, err := NewVPCFlowLogUnmarshaler(constants.FileFormatPlainText, component.BuildInfo{}, zap.NewNop(), test.featureGateEnabled)
+			require.NoError(t, err)
+
 			logs, err := u.UnmarshalAWSLogs(test.reader)
 
 			if test.expectedErr != "" {