Merge branch 'main' into s3exporter-canned-acl

Author: andrzej-stencel

.chloggen/add-cloud-provider-attribute.yaml

@@ -0,0 +1,28 @@
+# 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: awsfirehosereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `cloud.provider`, `aws.log.group.names` and `aws.log.stream.names` resource attributes to logs and set scope name and version of logs and metrics ingested by awsfirehosereceiver.
+
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37968]
+
+# (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: [user]

.chloggen/fix-testbed-div-by-zero.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: testbed
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix batch interval calculation to avoid possible division by zero
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38084]
+
+# (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: [api]

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/unmarshaler.go

@@ -13,10 +13,13 @@ import (
 
 	jsoniter "github.com/json-iterator/go"
 	"github.com/klauspost/compress/gzip"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
 )
 
 const (
@@ -30,15 +33,16 @@ var errInvalidRecords = errors.New("record format invalid")
 
 // Unmarshaler for the CloudWatch Log JSON record format.
 type Unmarshaler struct {
-	logger   *zap.Logger
-	gzipPool sync.Pool
+	logger    *zap.Logger
+	buildInfo component.BuildInfo
+	gzipPool  sync.Pool
 }
 
 var _ plog.Unmarshaler = (*Unmarshaler)(nil)
 
 // NewUnmarshaler creates a new instance of the Unmarshaler.
-func NewUnmarshaler(logger *zap.Logger) *Unmarshaler {
-	return &Unmarshaler{logger: logger}
+func NewUnmarshaler(logger *zap.Logger, buildInfo component.BuildInfo) *Unmarshaler {
+	return &Unmarshaler{logger: logger, buildInfo: buildInfo}
 }
 
 // UnmarshalLogs deserializes the given record as CloudWatch Logs events
@@ -120,10 +124,19 @@ func (u *Unmarshaler) UnmarshalLogs(compressedRecord []byte) (plog.Logs, error)
 	for resourceKey, logRecords := range byResource {
 		rl := logs.ResourceLogs().AppendEmpty()
 		resourceAttrs := rl.Resource().Attributes()
+		resourceAttrs.PutStr(conventions.AttributeCloudProvider, conventions.AttributeCloudProviderAWS)
 		resourceAttrs.PutStr(conventions.AttributeCloudAccountID, resourceKey.owner)
+		resourceAttrs.PutEmptySlice(conventions.AttributeAWSLogGroupNames).AppendEmpty().SetStr(resourceKey.logGroup)
+		resourceAttrs.PutEmptySlice(conventions.AttributeAWSLogStreamNames).AppendEmpty().SetStr(resourceKey.logStream)
+		// Deprecated: [v0.121.0] Use `conventions.AttributeAWSLogGroupNames` instead
 		resourceAttrs.PutStr(attributeAWSCloudWatchLogGroupName, resourceKey.logGroup)
+		// Deprecated: [v0.121.0] Use `conventions.AttributeAWSLogStreamNames` instead
 		resourceAttrs.PutStr(attributeAWSCloudWatchLogStreamName, resourceKey.logStream)
-		logRecords.MoveAndAppendTo(rl.ScopeLogs().AppendEmpty().LogRecords())
+
+		sl := rl.ScopeLogs().AppendEmpty()
+		sl.Scope().SetName(metadata.ScopeName)
+		sl.Scope().SetVersion(u.buildInfo.Version)
+		logRecords.MoveAndAppendTo(sl.LogRecords())
 	}
 	return logs, nil
 }

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/unmarshaler_test.go

@@ -10,46 +10,62 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
 )
 
 func TestType(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 	require.Equal(t, TypeStr, unmarshaler.Type())
 }
 
 func TestUnmarshal(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 	testCases := map[string]struct {
-		filename          string
-		wantResourceCount int
-		wantLogCount      int
-		wantErr           error
+		filename               string
+		wantResourceCount      int
+		wantLogCount           int
+		wantErr                error
+		wantResourceLogGroups  [][]string
+		wantResourceLogStreams [][]string
 	}{
 		"WithMultipleRecords": {
-			filename:          "multiple_records",
-			wantResourceCount: 1,
-			wantLogCount:      2,
+			filename:               "multiple_records",
+			wantResourceCount:      1,
+			wantLogCount:           2,
+			wantResourceLogGroups:  [][]string{{"test"}},
+			wantResourceLogStreams: [][]string{{"test"}},
 		},
 		"WithSingleRecord": {
-			filename:          "single_record",
-			wantResourceCount: 1,
-			wantLogCount:      1,
+			filename:               "single_record",
+			wantResourceCount:      1,
+			wantLogCount:           1,
+			wantResourceLogGroups:  [][]string{{"test"}},
+			wantResourceLogStreams: [][]string{{"test"}},
 		},
 		"WithInvalidRecords": {
 			filename: "invalid_records",
 			wantErr:  errInvalidRecords,
 		},
 		"WithSomeInvalidRecords": {
-			filename:          "some_invalid_records",
-			wantResourceCount: 1,
-			wantLogCount:      2,
+			filename:               "some_invalid_records",
+			wantResourceCount:      1,
+			wantLogCount:           2,
+			wantResourceLogGroups:  [][]string{{"test"}},
+			wantResourceLogStreams: [][]string{{"test"}},
 		},
 		"WithMultipleResources": {
-			filename:          "multiple_resources",
-			wantResourceCount: 3,
-			wantLogCount:      6,
+			filename:               "multiple_resources",
+			wantResourceCount:      3,
+			wantLogCount:           6,
+			wantResourceLogGroups:  nil, // not checking log group names because logs are unordered
+			wantResourceLogStreams: nil, // not checking log stream names because logs are unordered
 		},
 	}
 	for name, testCase := range testCases {
@@ -72,7 +88,18 @@ func TestUnmarshal(t *testing.T) {
 				for i := 0; i < got.ResourceLogs().Len(); i++ {
 					rm := got.ResourceLogs().At(i)
 					require.Equal(t, 1, rm.ScopeLogs().Len())
+					attrs := rm.Resource().Attributes()
+					assertString(t, attrs, conventions.AttributeCloudProvider, "aws")
+					assertString(t, attrs, conventions.AttributeCloudAccountID, "123")
+					if testCase.wantResourceLogGroups != nil {
+						assertStringArray(t, attrs, conventions.AttributeAWSLogGroupNames, testCase.wantResourceLogGroups[i])
+					}
+					if testCase.wantResourceLogStreams != nil {
+						assertStringArray(t, attrs, conventions.AttributeAWSLogStreamNames, testCase.wantResourceLogStreams[i])
+					}
 					ilm := rm.ScopeLogs().At(0)
+					assert.Equal(t, metadata.ScopeName, ilm.Scope().Name())
+					assert.Equal(t, component.NewDefaultBuildInfo().Version, ilm.Scope().Version())
 					gotLogCount += ilm.LogRecords().Len()
 				}
 				require.Equal(t, testCase.wantLogCount, gotLogCount)
@@ -82,7 +109,7 @@ func TestUnmarshal(t *testing.T) {
 }
 
 func TestLogTimestamp(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 	record, err := os.ReadFile(filepath.Join(".", "testdata", "single_record"))
 	require.NoError(t, err)
 
@@ -114,3 +141,24 @@ func gzipData(data []byte) ([]byte, error) {
 	}
 	return b.Bytes(), nil
 }
+
+func assertString(t *testing.T, m pcommon.Map, key, expected string) {
+	t.Helper()
+
+	v, ok := m.Get(key)
+	require.True(t, ok)
+	assert.Equal(t, expected, v.AsRaw())
+}
+
+func assertStringArray(t *testing.T, m pcommon.Map, key string, expected []string) {
+	t.Helper()
+
+	v, ok := m.Get(key)
+	require.True(t, ok)
+	s := v.Slice().AsRaw()
+	vAsStrings := make([]string, len(s))
+	for i, v := range s {
+		vAsStrings[i] = v.(string)
+	}
+	assert.ElementsMatch(t, expected, vAsStrings)
+}

receiver/awsfirehosereceiver/internal/unmarshaler/cwmetricstream/unmarshaler.go

@@ -11,10 +11,13 @@ import (
 	"time"
 
 	jsoniter "github.com/json-iterator/go"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
 )
 
 const (
@@ -33,13 +36,15 @@ var errInvalidRecords = errors.New("record format invalid")
 // https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-metric-streams-formats-json.html
 type Unmarshaler struct {
 	logger *zap.Logger
+
+	buildInfo component.BuildInfo
 }
 
 var _ pmetric.Unmarshaler = (*Unmarshaler)(nil)
 
 // NewUnmarshaler creates a new instance of the Unmarshaler.
-func NewUnmarshaler(logger *zap.Logger) *Unmarshaler {
-	return &Unmarshaler{logger}
+func NewUnmarshaler(logger *zap.Logger, buildInfo component.BuildInfo) *Unmarshaler {
+	return &Unmarshaler{logger, buildInfo}
 }
 
 // UnmarshalMetrics deserializes the record in CloudWatch Metric Stream JSON
@@ -122,6 +127,8 @@ func (u Unmarshaler) UnmarshalMetrics(record []byte) (pmetric.Metrics, error) {
 		rm := metrics.ResourceMetrics().AppendEmpty()
 		setResourceAttributes(resourceKey, rm.Resource())
 		scopeMetrics := rm.ScopeMetrics().AppendEmpty()
+		scopeMetrics.Scope().SetName(metadata.ScopeName)
+		scopeMetrics.Scope().SetVersion(u.buildInfo.Version)
 		for _, metric := range metricsMap {
 			metric.MoveTo(scopeMetrics.Metrics().AppendEmpty())
 		}

receiver/awsfirehosereceiver/internal/unmarshaler/cwmetricstream/unmarshaler_test.go

@@ -10,10 +10,13 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
 )
 
 const (
@@ -24,12 +27,12 @@ const (
 )
 
 func TestType(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 	require.Equal(t, TypeStr, unmarshaler.Type())
 }
 
 func TestUnmarshal(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 	testCases := map[string]struct {
 		filename           string
 		wantResourceCount  int
@@ -93,7 +96,7 @@ func TestUnmarshal(t *testing.T) {
 }
 
 func TestUnmarshal_SingleRecord(t *testing.T) {
-	unmarshaler := NewUnmarshaler(zap.NewNop())
+	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())
 
 	record, err := os.ReadFile(filepath.Join("testdata", "single_record"))
 	require.NoError(t, err)
@@ -112,6 +115,8 @@ func TestUnmarshal_SingleRecord(t *testing.T) {
 	assert.Equal(t, conventions.AttributeCloudProviderAWS, cloudProvider.Str())
 	require.Equal(t, 1, rm.ScopeMetrics().Len())
 	sm := rm.ScopeMetrics().At(0)
+	assert.Equal(t, metadata.ScopeName, sm.Scope().Name())
+	assert.Equal(t, component.NewDefaultBuildInfo().Version, sm.Scope().Version())
 
 	require.Equal(t, 1, sm.Metrics().Len())
 	metric := sm.Metrics().At(0)

receiver/awsfirehosereceiver/internal/unmarshaler/otlpmetricstream/unmarshaler.go

@@ -8,9 +8,12 @@ import (
 	"fmt"
 
 	"github.com/gogo/protobuf/proto"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
 )
 
 const (
@@ -25,14 +28,15 @@ var errInvalidOTLPFormatStart = errors.New("unable to decode data length from me
 // More details can be found at:
 // https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-metric-streams-formats-opentelemetry-100.html
 type Unmarshaler struct {
-	logger *zap.Logger
+	logger    *zap.Logger
+	buildInfo component.BuildInfo
 }
 
 var _ pmetric.Unmarshaler = (*Unmarshaler)(nil)
 
 // NewUnmarshaler creates a new instance of the Unmarshaler.
-func NewUnmarshaler(logger *zap.Logger) *Unmarshaler {
-	return &Unmarshaler{logger}
+func NewUnmarshaler(logger *zap.Logger, buildInfo component.BuildInfo) *Unmarshaler {
+	return &Unmarshaler{logger, buildInfo}
 }
 
 // UnmarshalMetrics deserializes the recordsas a length-delimited sequence of
@@ -52,6 +56,14 @@ func (u Unmarshaler) UnmarshalMetrics(record []byte) (pmetric.Metrics, error) {
 		if err != nil {
 			return pmetric.Metrics{}, fmt.Errorf("unable to unmarshal input: %w", err)
 		}
+		for i := 0; i < req.Metrics().ResourceMetrics().Len(); i++ {
+			rm := req.Metrics().ResourceMetrics().At(i)
+			for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+				sm := rm.ScopeMetrics().At(j)
+				sm.Scope().SetName(metadata.ScopeName)
+				sm.Scope().SetVersion(u.buildInfo.Version)
+			}
+		}
 		req.Metrics().ResourceMetrics().MoveAndAppendTo(md.ResourceMetrics())
 	}