[processor/coralogix] Add transactions feature to Coralogix processor (#40868)

#### Description
Add APM transactions support to the Coralogix exporter.

Reference:
https://coralogix.com/docs/user-guides/apm/features/transactions/

#### Link to tracking issue
Fixes #40863

#### Testing
- Added some tests
- Manually tested


![image](https://github.com/user-attachments/assets/663dfb01-f9a5-4b98-a432-78e39e75a814)

![image](https://github.com/user-attachments/assets/cb47e894-7025-4fab-8c40-4020c2551c95)

![image](https://github.com/user-attachments/assets/f3cf3169-840e-4930-9575-861f7eef65e3)

![image](https://github.com/user-attachments/assets/e759f541-a958-4c2f-b5c0-edfd1bf32dce)

![image](https://github.com/user-attachments/assets/d1096714-9a89-4c34-bf45-d38e6bd55cc7)

![image](https://github.com/user-attachments/assets/68d1fb76-0861-4e89-8690-ee1f53ca850f)

[Trace exported from Coralogix
backend](https://github.com/user-attachments/files/20975849/transactions.csv).


`values.yaml` (note I changed the image to one built for this test):
```yaml
opentelemetry-agent:
  service:
    enabled: true
  extraConfig:
    processors:
      groupbytrace:
        wait_duration: 30s

      coralogix:
        transactions:
          enabled: true
    service:
      pipelines:
        traces:
          processors: [groupbytrace, coralogix]
```

#### Documentation
Added some documentation in the `README.md` file.

Signed-off-by: Israel Blancas <[email protected]>

Author: iblancasa

.chloggen/40863.yaml

@@ -0,0 +1,36 @@
+# 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: coralogixprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add transactions feature
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40863]
+
+# (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: |
+  The transactions feature enables tracking of distributed transactions across microservices in a distributed system. 
+  It provides end-to-end visibility into request flows by correlating spans across different services, allowing 
+  developers to understand the complete journey of a request through their microservices architecture. This 
+  feature is particularly useful for identifying performance bottlenecks, debugging issues, and monitoring 
+  the health of distributed applications.
+
+  More information:
+    https://coralogix.com/docs/user-guides/apm/features/transactions
+
+
+# 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]

processor/coralogixprocessor/README.md

@@ -17,73 +17,57 @@
 
 The Coralogix processor adds attributes to spans that enable features in Coralogix.
 
-## Features
-
-### DB Statement Blueprints
+## Configuration
 
-This feature enables the processor to create blueprints from SQL queries, this means replacing any variables with `?`.
-The blueprint is also hashed to be able to be used with the spanmetrics connector.
-Long queries can be an issue when being stored in certain metric stores.
-Blueprints alleviate this problem by using the hash as the identifying dimension on the metric, which enables
-users to query metrics by blueprints.
+- `transactions`:
+  - `enabled` (`false` by default): enables the transactions feature from the Coralogix processor (more information below).
 
-The added attributes are `db.statement.blueprint` and `db.statement.blueprint.id`.
+## Features
 
-* `db.statement.blueprint` contains the blueprinted version of the statement, we require them to be sent to Coralogix to
-  display your blueprinted statement
-* `db.statement.blueprint.id` contains a hash of the statement, this way we can add it as a dimension in the spanmetrics
-  connector and use it to query your blueprints.
-* `sampling.priority` if enabled contains the value 100 for new blueprints, further explanation below.
+### Transactions
 
-#### Sampling
+The Transactions feature (originally called "Service Flows") is Coralogix's extension of OpenTelemetry instrumentation that breaks down each transaction into segments and aggregates their performance over time. It provides visibility into how each segment within a service contributes to overall transaction performance.
 
-If sampling is enabled then it stores the found blueprints in an in-memory cache to be able to send only new blueprints
-that haven't been seen yet.
-This only adds an attribute to the span named `sampling.priority`, if the blueprint is new then the sampling priority
-will be `100`.
+More information in the [official docs](https://coralogix.com/docs/user-guides/apm/features/transactions).
 
-Using this key it's possible to use either
-the [Tail Sampler](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/tailsamplingprocessor)
-or
-the [Probabilistic Sampler](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/probabilisticsamplerprocessor)
-to only send new blueprints to Coralogix.
-If sampling is not enabled it won't cache anything and the `sampling.priority` attribute won't be added.
+#### How It Works
 
-The cache is limited by the `max_cache_size_mib` configuration, if the cache is full it will remove the oldest entries
-to make space for new ones.
-The cache stores hashes of the queries, each hash is 8 bytes, so the number of maximum cache entries is calculated
-by `max_cache_size_mib * 1024 * 1024 / 8`.
+The processor automatically identifies the root span within each trace and applies transaction attributes to all spans in that trace:
 
-## Config
+1. **Root Span Identification**: The processor finds the span with no parent span ID (or whose parent is not in the current trace) and marks it as the root span.
 
-* `db_statement_blueprints`
-    * `sampling`:
-        * `enabled`: (default: `false`) If enabled, adds the attribute `sampling.priority` with a value of `100` to spans with new
-          blueprints.
-          Refer to the [Sampling section](#sampling) for more information.
-        * `max_cache_size_mib` (default: `1024`) The size of the cache in mebibytes to store seen blueprints hashes.
+2. **Attribute Application**: All spans in the trace receive the following attributes:
+   - `cgx.transaction`: Set to the name of the root span
+   - `cgx.transaction.root`: Set to `true` for the root span only
 
-### Basic Setup
+3. **Trace State Updates**: For client spans (not server or consumer spans), the processor also updates the trace state with `cgx_transaction=<root-span-name>` to propagate transaction information across service boundaries.
 
-This setup is without sampling meaning no `sampling.priority` attribute will be added to spans.
-The cache will be disabled.
+#### Configuration
 
 ```yaml
 processors:
   coralogix:
-    db_statement_blueprints:
+    transactions:
+      enabled: true
 ```
 
-### With Sampling Config
+**Note**: The transactions feature requires the `groupbytrace` processor to be configured before the `coralogix` processor in your pipeline to work properly. This ensures that all spans from the same trace are processed together.
 
-This setup will enable the cache to store seen blueprints and add the `sampling.priority` attribute to spans with new
-blueprints.
+```yaml
+processors:
+  groupbytrace:
+    wait_duration: 5s
+    num_traces: 1000
+  coralogix:
+    transactions:
+      enabled: true
+```
+
+### Basic Setup
 
 ```yaml
 processors:
   coralogix:
-    db_statement_blueprints:
-      sampling:
-        enabled: true
-        max_cache_size_mib: 1024 #1GiB
-  ```
+    transactions:
+      enabled: true
+```

processor/coralogixprocessor/config.go

@@ -3,27 +3,12 @@
 
 package coralogixprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/coralogixprocessor"
 
-import "errors"
-
-type samplingConfig struct {
-	enabled         bool  `mapstructure:"enabled"`
-	maxCacheSizeMib int64 `mapstructure:"max_cache_size_mib"`
-}
-
-type databaseBlueprintsConfig struct {
-	sampling samplingConfig `mapstructure:"sampling"`
+// TransactionsConfig holds configuration for transactions.
+type TransactionsConfig struct {
+	Enabled bool     `mapstructure:"enabled"`
+	_       struct{} // prevents unkeyed literal initialization
 }
 
 type Config struct {
-	databaseBlueprintsConfig `mapstructure:"database_blueprints_config"`
-}
-
-func (c *Config) Validate() error {
-	if c.sampling.enabled && c.sampling.maxCacheSizeMib <= 0 {
-		return errors.New("max_cache_size_mib must be a positive integer")
-	}
-	if c.sampling.enabled && c.sampling.maxCacheSizeMib != 0 {
-		return errors.New("max_cache_size_mib can only be defined when sampling is enabled")
-	}
-	return nil
+	TransactionsConfig `mapstructure:"transactions"`
 }

processor/coralogixprocessor/factory.go

@@ -22,7 +22,11 @@ func NewFactory() processor.Factory {
 }
 
 func createDefaultConfig() component.Config {
-	return &Config{}
+	return &Config{
+		TransactionsConfig: TransactionsConfig{
+			Enabled: false,
+		},
+	}
 }
 
 func createTracesProcessor(

processor/coralogixprocessor/go.mod

@@ -14,6 +14,7 @@ require (
 	go.opentelemetry.io/collector/processor/processorhelper v0.129.1-0.20250630174123-18b3b578b0b3
 	go.opentelemetry.io/collector/processor/processortest v0.129.1-0.20250630174123-18b3b578b0b3
 	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.0
 )
 
 require (
@@ -52,7 +53,6 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.37.0 // indirect
 	go.opentelemetry.io/otel/trace v1.37.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.3 // indirect
 	golang.org/x/net v0.39.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect

processor/coralogixprocessor/internal/transactions/span_tree.go

@@ -0,0 +1,78 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package transactions // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/coralogixprocessor/internal/transactions"
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+type spanNode struct {
+	span     ptrace.Span
+	children []*spanNode
+}
+
+// buildSpanTree constructs a hierarchical tree of spans
+func buildSpanTree(spans []ptrace.Span, logger *zap.Logger) *spanNode {
+	spanMap := make(map[pcommon.SpanID]*spanNode)
+	var rootSpan *spanNode
+	var orphanedSpans []*spanNode
+
+	for _, span := range spans {
+		node := &spanNode{span: span}
+		spanMap[span.SpanID()] = node
+
+		if span.ParentSpanID().IsEmpty() {
+			if rootSpan != nil {
+				logger.Warn("Multiple root spans found in single trace",
+					zap.String("existingRootSpanID", rootSpan.span.SpanID().String()),
+					zap.String("newRootSpanID", span.SpanID().String()))
+				// We'll keep the earliest span as root
+				if span.StartTimestamp() < rootSpan.span.StartTimestamp() {
+					orphanedSpans = append(orphanedSpans, rootSpan)
+					rootSpan = node
+				} else {
+					orphanedSpans = append(orphanedSpans, node)
+				}
+			} else {
+				rootSpan = node
+			}
+		}
+	}
+
+	if len(orphanedSpans) > 0 {
+		logger.Warn("orphaned spans found", zap.Int("orphanedSpans", len(orphanedSpans)))
+	}
+
+	// If no root span was found, use the earliest span as root
+	if rootSpan == nil && len(spans) > 0 {
+		earliestSpan := spanMap[spans[0].SpanID()]
+		earliestTime := spans[0].StartTimestamp()
+
+		for _, node := range spanMap {
+			if node.span.StartTimestamp() < earliestTime {
+				earliestTime = node.span.StartTimestamp()
+				earliestSpan = node
+			}
+		}
+
+		rootSpan = earliestSpan
+		logger.Warn("No root span found in trace, using earliest span as root",
+			zap.String("selectedRootSpanID", rootSpan.span.SpanID().String()))
+	}
+
+	for _, node := range spanMap {
+		if node == rootSpan {
+			continue
+		}
+
+		parentID := node.span.ParentSpanID()
+		if parent, exists := spanMap[parentID]; exists {
+			parent.children = append(parent.children, node)
+		}
+	}
+
+	return rootSpan
+}

processor/coralogixprocessor/internal/transactions/span_tree_test.go

@@ -0,0 +1,82 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package transactions
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+func TestBuildSpanTreeSingleSpan(t *testing.T) {
+	logger := zap.NewNop()
+	span := createSpan("span1", pcommon.SpanID([8]byte{1}), pcommon.SpanID([8]byte{}), 100)
+	spans := []ptrace.Span{span}
+
+	root := buildSpanTree(spans, logger)
+
+	assert.NotNil(t, root)
+	assert.Equal(t, span.SpanID(), root.span.SpanID())
+	assert.Empty(t, root.children)
+}
+
+func TestBuildSpanTreeParentChild(t *testing.T) {
+	logger := zap.NewNop()
+	parentSpan := createSpan("parent", pcommon.SpanID([8]byte{1}), pcommon.SpanID([8]byte{}), 100)
+	childSpan := createSpan("child", pcommon.SpanID([8]byte{2}), parentSpan.SpanID(), 200)
+	spans := []ptrace.Span{childSpan, parentSpan}
+
+	root := buildSpanTree(spans, logger)
+
+	assert.NotNil(t, root)
+	assert.Equal(t, parentSpan.SpanID(), root.span.SpanID())
+	assert.Len(t, root.children, 1)
+	assert.Equal(t, childSpan.SpanID(), root.children[0].span.SpanID())
+}
+
+func TestBuildSpanTreeMultipleRoots(t *testing.T) {
+	logger := zap.NewNop()
+	root1 := createSpan("root1", pcommon.SpanID([8]byte{1}), pcommon.SpanID([8]byte{}), 100)
+	root2 := createSpan("root2", pcommon.SpanID([8]byte{2}), pcommon.SpanID([8]byte{}), 50)
+	spans := []ptrace.Span{root1, root2}
+
+	root := buildSpanTree(spans, logger)
+
+	assert.NotNil(t, root)
+	assert.Equal(t, root2.SpanID(), root.span.SpanID())
+}
+
+func TestBuildSpanTreeNoRoot(t *testing.T) {
+	logger := zap.NewNop()
+	span1 := createSpan("span1", pcommon.SpanID([8]byte{1}), pcommon.SpanID([8]byte{3}), 100)
+	span2 := createSpan("span2", pcommon.SpanID([8]byte{2}), pcommon.SpanID([8]byte{3}), 50)
+	spans := []ptrace.Span{span1, span2}
+
+	root := buildSpanTree(spans, logger)
+
+	assert.NotNil(t, root)
+	assert.Equal(t, span2.SpanID(), root.span.SpanID())
+}
+
+func TestBuildSpanTreeEmpty(t *testing.T) {
+	logger := zap.NewNop()
+	spans := []ptrace.Span{}
+
+	root := buildSpanTree(spans, logger)
+
+	assert.Nil(t, root)
+}
+
+func createSpan(name string, spanID pcommon.SpanID, parentSpanID pcommon.SpanID, startTime int64) ptrace.Span {
+	span := ptrace.NewSpan()
+	span.SetName(name)
+	span.SetSpanID(spanID)
+	span.SetParentSpanID(parentSpanID)
+	span.SetStartTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, startTime)))
+	return span
+}