Support AWS ELB Access logs

extension/encoding/awslogsencodingextension/README.md

@@ -18,8 +18,13 @@ This extension unmarshals logs encoded in formats produced by AWS services, incl
    - Parquet support still to be added.
  - [S3 access log records](https://docs.aws.amazon.com/AmazonS3/latest/userguide/LogFormat.html).
  - [AWS CloudTrail logs](https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-log-file-examples.html).
+- ELB access logs:
+   - [Classic Load Balancer (CLB)](https://docs.aws.amazon.com/elasticloadbalancing/latest/classic/access-log-collection.html)
+   - [Application Load Balancer (ALB)](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-access-logs.html)
+   - [Network Load Balancer (NLB)](https://docs.aws.amazon.com/elasticloadbalancing/latest/network/load-balancer-access-logs.html)
  - (More to be added later.)
 
+
 Example for Amazon CloudWatch Logs Subscription Filters:
 ```yaml
 extensions:
@@ -57,6 +62,13 @@ extensions:
     format: cloudtrail_log
 ```
 
+Example for ELB access logs:
+```yaml
+extensions:
+  awslogs_encoding/elb_access_log:
+    format: elb_access_log
+```
+
 #### VPC flow log record fields
 
 [VPC flow log record fields](https://docs.aws.amazon.com/vpc/latest/userguide/flow-log-records.html#flow-logs-fields) are mapped this way in the resulting OpenTelemetry log:
@@ -213,5 +225,62 @@ extensions:
 | `userIdentity.userId`                 | `user.id`                                                     |
 | `userIdentity.userName`               | `user.name`                                                   |
 
-
 All request parameters and response elements are included directly as nested maps in the attributes, preserving their original structure.
+
+#### ELB Access Log Fields
+
+ELB access log record fields are mapped this way in the resulting OpenTelemetry log:
+
+##### Application Load Balancer (ALB)
+
+
+| **AWS Field**         | **OpenTelemetry Field**                                                                 |
+|-----------------------|---------------------------------------------------------------------------------------------|
+| Provider              | `cloud.provider`                                                                          |
+| Resource              | `cloud.resource_id`                                                                       |
+| Protocol              | `network.protocol.name`<br> `network.protocol.version`                                   |
+| Client IP             | `client.address`                                                                          |
+| Request Method        | `http.request.method`                                                                     |
+| Request URL           | `url.full`                                                                                |
+| Client Port           | `client.port`                                                                             |
+| Request Size          | `http.request.size`                                                                       |
+| Response Size         | `http.response.size`                                                                      |
+| ELB Status            | `aws.elb.status.code`                                                                     |
+| TLS Version           | `tls.protocol.version`                                                                    |
+| Cipher Suite          | `tls.cipher`                                                                              |
+
+##### Network Load Balancer (NLB)
+
+
+| **AWS Field**         | **OpenTelemetry Field(s)**                                                                 |
+|-----------------------|---------------------------------------------------------------------------------------------|
+| Provider              | `cloud.provider`                                                                          |
+| Resource              | `cloud.resource_id`                                                                       |
+| Protocol              | `network.protocol.name`<br>`network.protocol.version`                                   |
+| Client IP             | `client.address`                                                                          |
+| Client Port           | `client.port`                                                                             |
+| Request Size          | `http.request.size`                                                                       |
+| Response Size         | `http.response.size`                                                                      |
+| Listener ARN          | `tls.listener.resource_id`                                                                |
+| TLS Version           | `tls.protocol.version`                                                                    |
+| Cipher Suite          | `tls.cipher`                                                                              |
+
+
+##### Classic Load Balancer (CLB)
+
+
+| **AWS Field**         | **OpenTelemetry Field(s)**                                                                 |
+|-----------------------|---------------------------------------------------------------------------------------------|
+| Provider              | `cloud.provider`                                                                          |
+| Resource              | `cloud.resource_id`                                                                       |
+| Client IP             | `client.address`                                                                          |
+| Request Method        | `http.request.method`                                                                     |
+| Request URL           | `url.full`                                                                                |
+| Protocol              | `network.protocol.name`<br> `network.protocol.version`                                   |
+| Client Port           | `client.port`                                                                             |
+| Request Size          | `http.request.size`                                                                       |
+| Response Size         | `http.response.size`                                                                      |
+| TLS Version           | `tls.protocol.version`                                                                    |
+| Cipher Suite          | `tls.cipher`                                                                              |
+| ELB Status            | `aws.elb.status.code`                                                                     |
+| Backend Status        | `aws.elb.backend.status.code`                                                             |

extension/encoding/awslogsencodingextension/config.go

@@ -18,6 +18,7 @@ const (
 	formatS3AccessLog                      = "s3_access_log"
 	formatWAFLog                           = "waf_log"
 	formatCloudTrailLog                    = "cloudtrail_log"
+	formatELBAccessLog                     = "elb_access_log"
 
 	fileFormatPlainText = "plain-text"
 	fileFormatParquet   = "parquet"
@@ -29,6 +30,8 @@ var (
 		formatVPCFlowLog,
 		formatS3AccessLog,
 		formatWAFLog,
+		formatCloudTrailLog,
+		formatELBAccessLog,
 	}
 	supportedVPCFlowLogFileFormat = []string{fileFormatPlainText, fileFormatParquet}
 )
@@ -42,6 +45,7 @@ type Config struct {
 	// - s3_access_log
 	// - waf_log
 	// - cloudtrail_log
+	// - elb_access_log
 	//
 	Format string `mapstructure:"format"`
 
@@ -73,6 +77,7 @@ func (cfg *Config) Validate() error {
 	case formatS3AccessLog: // valid
 	case formatWAFLog: // valid
 	case formatCloudTrailLog: // valid
+	case formatELBAccessLog: // valid
 	default:
 		errs = append(errs, fmt.Errorf("unsupported format %q, expected one of %q", cfg.Format, supportedLogFormats))
 	}

extension/encoding/awslogsencodingextension/config_test.go

@@ -94,6 +94,15 @@ func TestLoadConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "elb_access_log"),
+			expected: &Config{
+				Format: formatELBAccessLog,
+				VPCFlowLogConfig: VPCFlowLogConfig{
+					FileFormat: fileFormatPlainText,
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

extension/encoding/awslogsencodingextension/extension.go

@@ -18,6 +18,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding"
 	awsunmarshaler "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler"
 	cloudtraillog "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/cloudtraillog"
+	elbaccesslogs "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/elb-access-log"
 	s3accesslog "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/s3-access-log"
 	subscriptionfilter "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/subscription-filter"
 	vpcflowlog "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/vpc-flow-log"
@@ -75,6 +76,11 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 			unmarshaler: cloudtraillog.NewCloudTrailLogUnmarshaler(settings.BuildInfo),
 			format:      formatCloudTrailLog,
 		}, nil
+	case formatELBAccessLog:
+		return &encodingExtension{
+			unmarshaler: elbaccesslogs.NewELBAccessLogUnmarshaler(settings.BuildInfo),
+			format:      formatELBAccessLog,
+		}, nil
 	default:
 		// Format will have been validated by Config.Validate,
 		// so we'll only get here if we haven't handled a valid
@@ -115,6 +121,16 @@ func (e *encodingExtension) getReaderFromFormat(buf []byte) (string, io.Reader,
 		return gzipEncoding, reader, err
 	case formatS3AccessLog:
 		return bytesEncoding, bytes.NewReader(buf), nil
+	case formatELBAccessLog:
+		// Check if the data is compressed
+		// Network LB and Applicayion LB store compressed files.
+		// Classic LB stores plain text files.
+		if len(buf) > 2 && buf[0] == 0x1f && buf[1] == 0x8b {
+			reader, err := e.getGzipReader(buf)
+			return gzipEncoding, reader, err
+		} else {
+			return bytesEncoding, bytes.NewReader(buf), nil
+		}
 	case formatVPCFlowLog:
 		switch e.vpcFormat {
 		case fileFormatParquet:

extension/encoding/awslogsencodingextension/extension_test.go

@@ -65,6 +65,15 @@ func TestNew_WAFLog(t *testing.T) {
 	require.ErrorContains(t, err, `failed to get reader for "waf_log" logs`)
 }
 
+func TestNew_ELBAcessLog(t *testing.T) {
+	e, err := newExtension(&Config{Format: formatELBAccessLog}, extensiontest.NewNopSettings(extensiontest.NopType))
+	require.NoError(t, err)
+	require.NotNil(t, e)
+
+	_, err = e.UnmarshalLogs([]byte("invalid"))
+	require.ErrorContains(t, err, `failed to get reader for "elb_access_log" logs`)
+}
+
 func TestNew_Unimplemented(t *testing.T) {
 	e, err := newExtension(&Config{Format: "invalid"}, extensiontest.NewNopSettings(extensiontest.NopType))
 	require.Error(t, err)

extension/encoding/awslogsencodingextension/internal/unmarshaler/elb-access-log/benchmark_test.go

@@ -0,0 +1,67 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package elbaccesslogs
+
+import (
+	"bytes"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.uber.org/zap"
+)
+
+type elbBenchmarkCase struct {
+	name     string
+	filename string
+	nLogs    int
+}
+
+var elbCases = []elbBenchmarkCase{
+	{"ALB_1_log", "./testdata/alb_al_valid_logs.log", 1},
+	{"ALB_1000_logs", "./testdata/alb_al_valid_logs.log", 1_000},
+	{"CLB_1_log", "./testdata/clb_al_valid_logs.log", 1},
+	{"CLB_1000_logs", "./testdata/clb_al_valid_logs.log", 1_000},
+	{"NLB_1_log", "./testdata/nlb_al_valid_logs.log", 1},
+	{"NLB_1000_logs", "./testdata/nlb_al_valid_logs.log", 1_000},
+}
+
+// createELBAccessLogContent reads data from a given elb log file
+// to get a valid log line. It will add this log line to a byte array
+// and append this line as many times as it takes to reach the desire
+// number of logs defined in nLogs. Each log record is defined in one
+// line. A new line means a new log record.
+func createELBAccessLogContent(b *testing.B, filename string, nLogs int) []byte {
+	// get a log line from the testdata
+	data, err := os.ReadFile(filename)
+	require.NoError(b, err)
+
+	size := len(data) + 1 // + "\n"
+	buf := bytes.NewBuffer(make([]byte, 0, nLogs*size))
+	for i := 0; i < nLogs; i++ {
+		buf.Write(data)
+		if i != nLogs-1 {
+			buf.Write([]byte("\n"))
+		}
+	}
+
+	return buf.Bytes()
+}
+
+func BenchmarkUnmarshalAWSLogs(b *testing.B) {
+	u := &elbAccessLogUnmarshaler{
+		buildInfo: component.BuildInfo{},
+		logger:    zap.NewNop(),
+	}
+	for _, bc := range elbCases {
+		data := createELBAccessLogContent(b, bc.filename, bc.nLogs)
+		b.Run(bc.name, func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				_, _ = u.UnmarshalAWSLogs(bytes.NewReader(data))
+			}
+		})
+	}
+}