Merge branch 'main' into otel-respect-data-stream

.github/workflows/bench-diff.yml

@@ -0,0 +1,56 @@
+name: bench-diff
+on:
+  push:
+  pull_request_target:
+    types: [labeled, synchronize]
+
+permissions:
+  contents: read
+
+jobs:
+  run-benchdiff:
+    runs-on: ubuntu-latest
+    # Support for branches, non-forked-prs or forked-prs with a label
+    if: |
+      contains(github.event.pull_request.labels.*.name, 'safe-to-test') ||
+      ! startsWith(github.event_name, 'pull_request') ||
+      (startsWith(github.event_name, 'pull_request') && github.event.pull_request.head.repo.fork == false)
+    permissions:
+      checks: write
+    steps:
+    - name: Get sha commit
+      id: event
+      run: |
+        echo "DEBUG: event type(${{ github.event_name }})"
+        ref="${{ github.sha }}"
+        if [ "${{ startsWith(github.event_name, 'pull_request') }}" = "true" ] ; then
+          ref="${{ github.event.pull_request.head.sha }}"
+        fi
+        echo "ref=${ref}" >> "$GITHUB_OUTPUT"
+    - uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+        persist-credentials: false
+        ref: ${{ steps.event.outputs.ref }}
+    - uses: actions/setup-go@v5
+      with:
+        go-version-file: go.mod
+    # Version: https://github.com/WillAbides/benchdiff-action/releases/tag/v0.3.5
+    - uses: WillAbides/benchdiff-action@990b4c50b5420b485bf87e42c9f18234eba76fbc
+      id: benchdiff
+      with:
+        benchdiff_version: 0.9.1
+        status_sha: ${{ steps.event.outputs.ref }}
+        status_name: benchdiff-result
+        status_on_degraded: neutral
+        # See https://github.com/WillAbides/benchdiff
+        benchdiff_args: |
+          --base-ref=origin/main
+          --cpu=1,2
+          --count=5
+          --warmup-count=1
+          --warmup-time=10ms
+          --benchtime=100ms
+          --tolerance=20
+          --benchmem
+          --debug

.github/workflows/ci.yml

@@ -36,59 +36,3 @@ jobs:
         cache: true
     - name: Run tests
       run: make test
-
-  run-benchdiff:
-    runs-on: ubuntu-latest
-    permissions:
-      checks: write
-    steps:
-    - uses: actions/checkout@v4
-      with:
-        fetch-depth: 0
-    - uses: actions/setup-go@v5
-      with:
-        go-version-file: go.mod
-    # Version: https://github.com/WillAbides/benchdiff-action/releases/tag/v0.3.3
-    - uses: WillAbides/benchdiff-action@990b4c50b5420b485bf87e42c9f18234eba76fbc
-      ## As long as we cannot use write permissions on forked pull requests, then let's avoid failing
-      continue-on-error: true
-      id: benchdiff
-      with:
-        benchdiff_version: 0.9.1
-        status_sha: ${{ github.sha }}
-        status_name: benchdiff-result
-        status_on_degraded: neutral
-        # See https://github.com/WillAbides/benchdiff
-        benchdiff_args: |
-          --base-ref=origin/main
-          --cpu=1,2
-          --count=5
-          --warmup-count=1
-          --warmup-time=10ms
-          --benchtime=100ms
-          --tolerance=20
-          --benchmem
-          --debug
-
-    - name: Summary
-      run: |
-        echo "${{ steps.benchdiff.outputs.benchstat_output }}" > benchdiff-report.md
-        {
-          echo "## Benchdiff Results"
-          echo ""
-          echo "HEAD: ${{ steps.benchdiff.outputs.head_sha }}"
-          echo "Base: ${{ steps.benchdiff.outputs.base_sha }}"
-          echo "Degraded: ${{ steps.benchdiff.outputs.degraded_result }}"
-          echo ""
-          echo "<details>"
-          echo "<summary>Results</summary>"
-          echo ""
-          echo "${{ steps.benchdiff.outputs.benchstat_output }}"
-          echo ""
-          echo "</details>"
-        } >> $GITHUB_STEP_SUMMARY
-
-    - uses: actions/upload-artifact@v4
-      with:
-        name: benchdiff-report
-        path: benchdiff-report.md

.github/workflows/label-elastic-pull-requests.yml

@@ -0,0 +1,29 @@
+name: "label-elastic-pull-requests"
+
+on:
+  pull_request_target:
+    types: [opened]
+
+jobs:
+  safe-to-test:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Check team membership for user
+      uses: elastic/[email protected]
+      id: checkUserMember
+      with:
+        username: ${{ github.actor }}
+        team: 'apm'
+        GITHUB_TOKEN: ${{ secrets.APM_TECH_USER_TOKEN }}
+    - name: Add safe-to-test label
+      uses: actions/github-script@v7
+      if: steps.checkUserMember.outputs.isTeamMember == 'true'
+      with:
+        github-token: ${{ secrets.APM_TECH_USER_TOKEN }}
+        script: |
+          github.rest.issues.addLabels({
+            issue_number: context.issue.number,
+            owner: context.repo.owner,
+            repo: context.repo.repo,
+            labels: ["safe-to-test"]
+          })

.github/workflows/opentelemetry.yml

@@ -1,12 +1,17 @@
 ---
+# Look up results at https://ela.st/oblt-ci-cd-stats
+# There will be one service per GitHub repository, including the org name, and one Transaction per Workflow.
 name: OpenTelemetry Export Trace
 
 on:
   workflow_run:
-    workflows:
-      - ci
+    workflows: [ "*" ]
     types: [completed]
 
+permissions:
+  contents: read
+  actions: read
+
 jobs:
   otel-export-trace:
     runs-on: ubuntu-latest

input/elasticapm/internal/modeldecoder/v2/transaction_test.go

@@ -561,6 +561,24 @@ func TestDecodeMapToTransactionModel(t *testing.T) {
 			}, event.Transaction.Message)
 		})
 
+		t.Run("messaging_without_destination", func(t *testing.T) {
+			var input transaction
+			var event modelpb.APMEvent
+			modeldecodertest.SetStructValues(&input, modeldecodertest.DefaultValues())
+			input.Type.Reset()
+			attrs := map[string]interface{}{
+				"messaging.system":    "kafka",
+				"messaging.operation": "publish",
+			}
+			input.OTel.Attributes = attrs
+			input.OTel.SpanKind.Reset()
+
+			mapToTransactionModel(&input, &event)
+			assert.Equal(t, "messaging", event.Transaction.Type)
+			assert.Equal(t, "CONSUMER", event.Span.Kind)
+			assert.Nil(t, event.Transaction.Message)
+		})
+
 		t.Run("network", func(t *testing.T) {
 			attrs := map[string]interface{}{
 				"network.connection.type":    "cell",

input/otlp/traces.go

@@ -257,7 +257,7 @@ func TranslateTransaction(
 	)
 
 	var isHTTP, isRPC, isMessaging bool
-	var message modelpb.Message
+	var messagingQueueName string
 
 	var samplerType, samplerParam pcommon.Value
 	attributes.Range(func(kDots string, v pcommon.Value) bool {
@@ -416,8 +416,14 @@ func TranslateTransaction(
 
 			// messaging.*
 			case "message_bus.destination", semconv.AttributeMessagingDestination:
-				message.QueueName = stringval
 				isMessaging = true
+				messagingQueueName = stringval
+			case semconv.AttributeMessagingSystem:
+				isMessaging = true
+				modelpb.Labels(event.Labels).Set(k, stringval)
+			case semconv.AttributeMessagingOperation:
+				isMessaging = true
+				modelpb.Labels(event.Labels).Set(k, stringval)
 
 			// rpc.*
 			//
@@ -513,7 +519,12 @@ func TranslateTransaction(
 		event.Url = modelpb.ParseURL(httpURL, httpHost, httpScheme)
 	}
 	if isMessaging {
-		event.Transaction.Message = &message
+		// Overwrite existing event.Transaction.Message
+		event.Transaction.Message = nil
+		if messagingQueueName != "" {
+			event.Transaction.Message = modelpb.MessageFromVTPool()
+			event.Transaction.Message.QueueName = messagingQueueName
+		}
 	}
 
 	if event.Client == nil && event.Source != nil {

input/otlp/traces_test.go

@@ -609,20 +609,56 @@ func TestRPCSpan(t *testing.T) {
 }
 
 func TestMessagingTransaction(t *testing.T) {
-	event := transformTransactionWithAttributes(t, map[string]interface{}{
-		"messaging.destination": "myQueue",
-	}, func(s ptrace.Span) {
-		s.SetKind(ptrace.SpanKindConsumer)
-		// Set parentID to imply this isn't the root, but
-		// kind==Consumer should still force the span to be translated
-		// as a transaction.
-		s.SetParentSpanID(pcommon.SpanID{3})
-	})
-	assert.Equal(t, "messaging", event.Transaction.Type)
-	assert.Empty(t, event.Labels)
-	assert.Equal(t, &modelpb.Message{
-		QueueName: "myQueue",
-	}, event.Transaction.Message)
+	for _, tc := range []struct {
+		attrs map[string]interface{}
+
+		expectedLabels     map[string]*modelpb.LabelValue
+		expectedTxnMessage *modelpb.Message
+	}{
+		{
+			attrs: map[string]interface{}{
+				"messaging.destination": "myQueue",
+			},
+			expectedLabels: nil,
+			expectedTxnMessage: &modelpb.Message{
+				QueueName: "myQueue",
+			},
+		},
+		{
+			attrs: map[string]interface{}{
+				"messaging.system": "kafka",
+			},
+			expectedLabels: map[string]*modelpb.LabelValue{
+				"messaging_system": {Value: "kafka"},
+			},
+			expectedTxnMessage: nil,
+		},
+		{
+			attrs: map[string]interface{}{
+				"messaging.operation": "publish",
+			},
+			expectedLabels: map[string]*modelpb.LabelValue{
+				"messaging_operation": {Value: "publish"},
+			},
+			expectedTxnMessage: nil,
+		},
+	} {
+		tcName, err := json.Marshal(tc.attrs)
+		require.NoError(t, err)
+		t.Run(string(tcName), func(t *testing.T) {
+			event := transformTransactionWithAttributes(t, tc.attrs, func(s ptrace.Span) {
+				s.SetKind(ptrace.SpanKindConsumer)
+				// Set parentID to imply this isn't the root, but
+				// kind==Consumer should still force the span to be translated
+				// as a transaction.
+				s.SetParentSpanID(pcommon.SpanID{3})
+			})
+			assert.Equal(t, "messaging", event.Transaction.Type)
+			assert.Equal(t, tc.expectedLabels, event.Labels)
+			assert.Equal(t, tc.expectedTxnMessage, event.Transaction.Message)
+		})
+	}
+
 }
 
 func TestMessagingSpan(t *testing.T) {