Merge branch 'main' into flaky-wait-items

.chloggen/resourcedetectionprocessor-profiles.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. filelogreceiver)
+component: resourcedetectionprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Introduce support for Profiles signal type."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35980]
+
+# (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: []

.chloggen/stanza-sync-log-emitter.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: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Synchronous handling of entries passed from the log emitter to the receiver adapter
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35453]
+
+# (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: []

exporter/datadogexporter/go.mod

@@ -352,6 +352,7 @@ require (
 	go.opentelemetry.io/collector/pdata/testdata v0.113.0 // indirect
 	go.opentelemetry.io/collector/pipeline v0.113.0 // indirect
 	go.opentelemetry.io/collector/pipeline/pipelineprofiles v0.113.0 // indirect
+	go.opentelemetry.io/collector/processor/processorhelper/processorhelperprofiles v0.113.0 // indirect
 	go.opentelemetry.io/collector/processor/processorprofiles v0.113.0 // indirect
 	go.opentelemetry.io/collector/processor/processortest v0.113.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.113.0 // indirect

pkg/stanza/adapter/converter.go

@@ -155,44 +155,46 @@ func (c *Converter) workerLoop() {
 	defer c.wg.Done()
 
 	for entries := range c.workerChan {
+		// Send plogs directly to flushChan
+		c.flushChan <- ConvertEntries(entries)
+	}
+}
 
-		resourceHashToIdx := make(map[uint64]int)
-		scopeIdxByResource := make(map[uint64]map[string]int)
+func ConvertEntries(entries []*entry.Entry) plog.Logs {
+	resourceHashToIdx := make(map[uint64]int)
+	scopeIdxByResource := make(map[uint64]map[string]int)
 
-		pLogs := plog.NewLogs()
-		var sl plog.ScopeLogs
+	pLogs := plog.NewLogs()
+	var sl plog.ScopeLogs
 
-		for _, e := range entries {
-			resourceID := HashResource(e.Resource)
-			var rl plog.ResourceLogs
+	for _, e := range entries {
+		resourceID := HashResource(e.Resource)
+		var rl plog.ResourceLogs
 
-			resourceIdx, ok := resourceHashToIdx[resourceID]
-			if !ok {
-				resourceHashToIdx[resourceID] = pLogs.ResourceLogs().Len()
+		resourceIdx, ok := resourceHashToIdx[resourceID]
+		if !ok {
+			resourceHashToIdx[resourceID] = pLogs.ResourceLogs().Len()
 
-				rl = pLogs.ResourceLogs().AppendEmpty()
-				upsertToMap(e.Resource, rl.Resource().Attributes())
+			rl = pLogs.ResourceLogs().AppendEmpty()
+			upsertToMap(e.Resource, rl.Resource().Attributes())
 
-				scopeIdxByResource[resourceID] = map[string]int{e.ScopeName: 0}
+			scopeIdxByResource[resourceID] = map[string]int{e.ScopeName: 0}
+			sl = rl.ScopeLogs().AppendEmpty()
+			sl.Scope().SetName(e.ScopeName)
+		} else {
+			rl = pLogs.ResourceLogs().At(resourceIdx)
+			scopeIdxInResource, ok := scopeIdxByResource[resourceID][e.ScopeName]
+			if !ok {
+				scopeIdxByResource[resourceID][e.ScopeName] = rl.ScopeLogs().Len()
 				sl = rl.ScopeLogs().AppendEmpty()
 				sl.Scope().SetName(e.ScopeName)
 			} else {
-				rl = pLogs.ResourceLogs().At(resourceIdx)
-				scopeIdxInResource, ok := scopeIdxByResource[resourceID][e.ScopeName]
-				if !ok {
-					scopeIdxByResource[resourceID][e.ScopeName] = rl.ScopeLogs().Len()
-					sl = rl.ScopeLogs().AppendEmpty()
-					sl.Scope().SetName(e.ScopeName)
-				} else {
-					sl = pLogs.ResourceLogs().At(resourceIdx).ScopeLogs().At(scopeIdxInResource)
-				}
+				sl = pLogs.ResourceLogs().At(resourceIdx).ScopeLogs().At(scopeIdxInResource)
 			}
-			convertInto(e, sl.LogRecords().AppendEmpty())
 		}
-
-		// Send plogs directly to flushChan
-		c.flushChan <- pLogs
+		convertInto(e, sl.LogRecords().AppendEmpty())
 	}
+	return pLogs
 }
 
 func (c *Converter) flushLoop() {

pkg/stanza/adapter/factory.go

@@ -46,14 +46,30 @@ func createLogsReceiver(logReceiverType LogReceiverType) rcvr.CreateLogsFunc {
 
 		operators := append([]operator.Config{inputCfg}, baseCfg.Operators...)
 
+		obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
+			ReceiverID:             params.ID,
+			ReceiverCreateSettings: params,
+		})
+		if err != nil {
+			return nil, err
+		}
+		rcv := &receiver{
+			set:       params.TelemetrySettings,
+			id:        params.ID,
+			consumer:  consumerretry.NewLogs(baseCfg.RetryOnFailure, params.Logger, nextConsumer),
+			obsrecv:   obsrecv,
+			storageID: baseCfg.StorageID,
+		}
+
 		var emitterOpts []helper.EmitterOption
 		if baseCfg.maxBatchSize > 0 {
 			emitterOpts = append(emitterOpts, helper.WithMaxBatchSize(baseCfg.maxBatchSize))
 		}
 		if baseCfg.flushInterval > 0 {
 			emitterOpts = append(emitterOpts, helper.WithFlushInterval(baseCfg.flushInterval))
 		}
-		emitter := helper.NewLogEmitter(params.TelemetrySettings, emitterOpts...)
+
+		emitter := helper.NewLogEmitter(params.TelemetrySettings, rcv.consumeEntries, emitterOpts...)
 		pipe, err := pipeline.Config{
 			Operators:     operators,
 			DefaultOutput: emitter,
@@ -62,27 +78,9 @@ func createLogsReceiver(logReceiverType LogReceiverType) rcvr.CreateLogsFunc {
 			return nil, err
 		}
 
-		var converterOpts []converterOption
-		if baseCfg.numWorkers > 0 {
-			converterOpts = append(converterOpts, withWorkerCount(baseCfg.numWorkers))
-		}
-		converter := NewConverter(params.TelemetrySettings, converterOpts...)
-		obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
-			ReceiverID:             params.ID,
-			ReceiverCreateSettings: params,
-		})
-		if err != nil {
-			return nil, err
-		}
-		return &receiver{
-			set:       params.TelemetrySettings,
-			id:        params.ID,
-			pipe:      pipe,
-			emitter:   emitter,
-			consumer:  consumerretry.NewLogs(baseCfg.RetryOnFailure, params.Logger, nextConsumer),
-			converter: converter,
-			obsrecv:   obsrecv,
-			storageID: baseCfg.StorageID,
-		}, nil
+		rcv.emitter = emitter
+		rcv.pipe = pipe
+
+		return rcv, nil
 	}
 }

pkg/stanza/adapter/integration_test.go

@@ -27,7 +27,7 @@ import (
 func createNoopReceiver(nextConsumer consumer.Logs) (*receiver, error) {
 	set := componenttest.NewNopTelemetrySettings()
 	set.Logger = zap.NewNop()
-	emitter := helper.NewLogEmitter(set)
+
 	pipe, err := pipeline.Config{
 		Operators: []operator.Config{
 			{
@@ -48,15 +48,18 @@ func createNoopReceiver(nextConsumer consumer.Logs) (*receiver, error) {
 		return nil, err
 	}
 
-	return &receiver{
-		set:       set,
-		id:        component.MustNewID("testReceiver"),
-		pipe:      pipe,
-		emitter:   emitter,
-		consumer:  nextConsumer,
-		converter: NewConverter(componenttest.NewNopTelemetrySettings()),
-		obsrecv:   obsrecv,
-	}, nil
+	rcv := &receiver{
+		set:      set,
+		id:       component.MustNewID("testReceiver"),
+		pipe:     pipe,
+		consumer: nextConsumer,
+		obsrecv:  obsrecv,
+	}
+
+	emitter := helper.NewLogEmitter(set, rcv.consumeEntries)
+
+	rcv.emitter = emitter
+	return rcv, nil
 }
 
 // BenchmarkEmitterToConsumer serves as a benchmark for entries going from the emitter to consumer,