link transaction

elastic · Jan 31, 2024 · 16d1fb6 · 16d1fb6
1 parent eeb92c1
commit 16d1fb6
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 3 deletions.
diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go
@@ -51,6 +51,7 @@ type Bulk interface {
 	Search(ctx context.Context, index string, body []byte, opts ...Opt) (*es.ResultT, error)
 	HasTracer() bool
 	StartTransaction(name, transactionType string) *apm.Transaction
+	StartTransactionOptions(name, transactionType string, opts apm.TransactionOptions) *apm.Transaction
 
 	// Multi Operation API's run in the bulk engine
 	MCreate(ctx context.Context, ops []MultiOp, opts ...Opt) ([]BulkIndexerResponseItem, error)

diff --git a/internal/pkg/bulk/opBulk.go b/internal/pkg/bulk/opBulk.go
@@ -300,3 +300,7 @@ func (b *Bulker) HasTracer() bool {
 func (b *Bulker) StartTransaction(name, transactionType string) *apm.Transaction {
 	return b.tracer.StartTransaction(name, transactionType)
 }
+
+func (b *Bulker) StartTransactionOptions(name, transactionType string, opts apm.TransactionOptions) *apm.Transaction {
+	return b.tracer.StartTransactionOptions(name, transactionType, opts)
+}
diff --git a/internal/pkg/policy/monitor.go b/internal/pkg/policy/monitor.go
@@ -165,7 +165,7 @@ LOOP:
 
 			m.dispatchPending(iCtx)
 			endTrans(trans)
-		case hits := <-s.Output():
+		case hits := <-s.Output(): // TODO would be nice to attach transaction IDs to hits, but would likely need a bigger refactor.
 			if m.bulker.HasTracer() {
 				trans = m.bulker.StartTransaction("output policies", "policy_monitor")
 				iCtx = apm.ContextWithTransaction(ctx, trans)
@@ -282,8 +282,12 @@ func (m *monitorT) loadPolicies(ctx context.Context) error {
 	span, ctx := apm.StartSpan(ctx, "Load policies", "load")
 	defer span.End()
 
-	if m.bulker.HasTracer() { // TODO link bulker transaction with the policy_monitor one
-		trans := m.bulker.StartTransaction("Load policies", "bulker")
+	if m.bulker.HasTracer() {
+		tctx := span.TraceContext()
+		trans := m.bulker.StartTransactionOptions("Load policies", "bulker", apm.TransactionOptions{Links: []apm.SpanLink{{
+			Trace: tctx.Trace,
+			Span:  tctx.Span,
+		}}})
 		ctx = apm.ContextWithTransaction(ctx, trans)
 		defer trans.End()
 	}
@@ -305,6 +309,9 @@ func (m *monitorT) loadPolicies(ctx context.Context) error {
 }
 
 func (m *monitorT) processPolicies(ctx context.Context, policies []model.Policy) error {
+	span, ctx := apm.StartSpan(ctx, "process policies", "process")
+	defer span.End()
+
 	if len(policies) == 0 {
 		return nil
 	}

diff --git a/internal/pkg/testing/bulk.go b/internal/pkg/testing/bulk.go
@@ -144,4 +144,8 @@ func (m *MockBulk) StartTransaction(name, transactionType string) *apm.Transacti
 	return nil
 }
 
+func (m *MockBulk) StartTransactionOptions(name, transactionType string, opts apm.TransactionOptions) *apm.Transaction {
+	return nil
+}
+
 var _ bulk.Bulk = (*MockBulk)(nil)