feat: add dependency processor using Apache Beam #6560

yunmaoQu · 2025-01-17T17:37:56Z

Which problem is this PR solving?

Resolves #5911

Description of the changes

add dependency processor using Apache Beam

How was this change tested?

e2e tests

Checklist

I have read https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
I have signed all commits
I have added unit tests for the new functionality
I have run lint and test steps successfully
- for jaeger: make lint test
- for jaeger-ui: npm run lint and npm run test

Signed-off-by: yunmaoQu <[email protected]>

plugin/storage/memory/memory.go

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

yurishkuro

where is it hooked up to anything?
what would be the e2e testing for this component?

yunmaoQu · 2025-01-18T02:52:23Z

where is it hooked up to anything?

what would be the e2e testing for this component?

@yurishkuro I have fixed it

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro

@mahadzaryab1 interesting direction here

cmd/jaeger/internal/processors/dependencyprocessor/config.go

cmd/jaeger/internal/processors/dependencyprocessor/e2e_test.go

cmd/jaeger/internal/processors/dependencyprocessor/config.go

cmd/jaeger/internal/processors/dependencyprocessor/factory.go

cmd/jaeger/internal/processors/dependencyprocessor/processor.go

cmd/jaeger/internal/processors/dependencyprocessor/e2e_test.go

Signed-off-by: yunmaoQu <[email protected]>

yunmaoQu · 2025-01-20T07:49:26Z

@yurishkuro Except this ,I update all based on your review.

yurishkuro · 2025-01-20T16:39:17Z

cmd/jaeger/internal/processors/dependencyprocessor/processor.go

+	config           *Config
+	aggregator       *dependencyAggregator // Define the aggregator below.
+	telset           component.TelemetrySettings
+	dependencyWriter *memory.Store


as I mentioned, you cannot have concrete store dependency here. The processor needs to work with any storage supported by Jaeger, as long as they implement WriteDependencies.

Example:

jaeger/cmd/jaeger/internal/extension/remotesampling/extension.go

Line 187 in 04b1ea4

f, err := jaegerstorage.GetStorageFactory(storageName, host)

yurishkuro · 2025-01-20T16:40:27Z

cmd/jaeger/internal/processors/dependencyprocessor/processor.go

+func (tp *dependencyProcessor) Shutdown(ctx context.Context) error {
+	close(tp.closeChan)
+	if tp.aggregator != nil {
+		if err := tp.aggregator.Close(); err != nil {


if aggregator has a Close() function why does it need to be passed closeChan?

cmd/jaeger/internal/processors/dependencyprocessor/config.go

cmd/jaeger/internal/processors/dependencyprocessor/e2e_test.go

yurishkuro · 2025-01-20T16:45:50Z

cmd/jaeger/internal/processors/dependencyprocessor/config.go

+	// is considered complete and ready for dependency aggregation.
+	// Default trace completion timeout: 2 seconds of inactivity
+	InactivityTimeout time.Duration `yaml:"inactivity_timeout"`
+}


please add Validate method and use valid: notations in the field tags.

this is not what I meant - these validations can all be done automatically via "github.com/asaskevich/govalidator" & govalidator.ValidateStruct(cfg)

e.g. cmd/jaeger/internal/extension/jaegerquery/config.go

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

cmd/jaeger/internal/processors/dependencyprocessor/factory.go

cmd/jaeger/internal/processors/dependencyprocessor/processor.go

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

cmd/jaeger/internal/processors/dependencyprocessor/e2e_test.go

Signed-off-by: yunmaoQu <[email protected]>

yunmaoQu · 2025-01-30T06:46:38Z

@yurishkuro I have fixed it

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro · 2025-01-31T18:26:57Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+func (agg *dependencyAggregator) Start(closeChan chan struct{}) {
+	agg.closeChan = closeChan


Suggested change

func (agg *dependencyAggregator) Start(closeChan chan struct{}) {

agg.closeChan = closeChan

func (agg *dependencyAggregator) Start() {

yurishkuro · 2025-01-31T18:36:57Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+		eventTime:   time.Now(),
+	}
+	select {
+	case agg.inputChan <- event:


What is the motivation for having this done in the background instead of in the caller goroutine? Are the operations on Beam pipeline threadsafe or is this the reason for separation?

The motivation for processing spans in the background (via a separate goroutine) rather than in the caller goroutine is primarily related to performance optimization, decoupling of concerns, and ensuring thread safety when interacting with the Apache Beam pipeline

yurishkuro · 2025-01-31T18:38:00Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+		config:           &cfg,
+		telset:           telset,
+		dependencyWriter: dependencyWriter,
+		inputChan:        make(chan spanEvent, 1000),


what's the motivation for making this a bound queue?

The motivation for making the inputChan a bounded queue (a buffered channel with a fixed size, e.g., 1000) is primarily to manage backpressure, control resource usage, and ensure system stability in high-throughput scenarios

The motivation for making the inputChan a bounded queue (a buffered channel with a fixed size, e.g., 1000) is primarily to manage backpressure, control resource usage, and ensure system stability in high-throughput scenarios

When a channel is unbounded, it cannot be written to unless there there is a reader waiting to consume it, so it provides a natural back pressure as the caller goroutine will be blocked and hold the remote caller. And it does not allow the queue to grow and accumulate unprocessed data while making it look like the processing was immediately successful.

@yurishkuro Ok,i will fix it . Could the part of code is ready to be merged ?

yunmaoQu · 2025-02-04T09:54:35Z

Hi @mahadzaryab1 , would you mind taking a look at my PR when you have a moment? I'd appreciate your feedback. Thanks!

yunmaoQu · 2025-02-04T09:56:33Z

@yurishkuro I will fix the part you mentioned . Could the part of code is ready to be merged ?

yunmaoQu · 2025-02-04T10:00:08Z

Hey @mahadzaryab1 , I know you're familiar with this part of the code. Could you give my PR a look and share your thoughts? Thank you!

mahadzaryab1 · 2025-02-05T04:05:13Z

Hey @mahadzaryab1 , I know you're familiar with this part of the code. Could you give my PR a look and share your thoughts? Thank you!

yeah absolutely - i'll take a look soon

mahadzaryab1 · 2025-02-08T16:04:55Z

@yunmaoQu It looks like the CI is failing - are you able to fix the tests before I take a look?

yunmaoQu · 2025-02-11T08:25:00Z

@yunmaoQu It looks like the CI is failing - are you able to fix the tests before I take a look?

@mahadzaryab1 I have some unchanged files with check annotations.Could you help me with that .I think the part I focus is Done.

Signed-off-by: yunmaoQu <[email protected]>

…pendency-processor

yunmaoQu · 2025-02-17T14:55:16Z

@mahadzaryab1 I have some unchanged files with check annotations.Could you help me with that .I think the part I focus is Done.

yurishkuro · 2025-02-17T15:05:40Z

@yunmaoQu the CI build is all red, please make sure it all builds cleanly.

yurishkuro · 2025-02-17T15:07:07Z

internal/storage/v1/api/spanstore/interface.go

 // Writer writes spans to storage.
 type Writer interface {
 	WriteSpan(ctx context.Context, span *model.Span) error
+	WriteDependencies(ctx context.Context, ts time.Time, dependencies []model.DependencyLink) error


this should be in a separate interface, spanstore.Writer is not required to support dependencies.

also it should be in the v2 storage, no reason to update v1.

github-actions · 2025-07-28T02:16:52Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. You may re-open it if you need more time.

graphite-app · 2025-07-28T02:20:22Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+	timestamped := beam.ParDo(s, func(event spanEvent) beam.WindowValue {
+		return beam.WindowValue{
+			Timestamp: event.eventTime,
+			Windows:   beam.window.IntervalWindow{Start: event.eventTime, End: event.eventTime.Add(agg.config.InactivityTimeout)},


The code references beam.window.IntervalWindow, but this appears to be an incorrect import path. The correct approach would be to import the window package separately and use window.IntervalWindow. Please update the import statement to include:

"github.com/apache/beam/sdks/v2/go/pkg/beam/window"

Then modify the code to use:

Windows: window.IntervalWindow{Start: event.eventTime, End: event.eventTime.Add(agg.config.InactivityTimeout)},

This will ensure proper access to the Beam windowing functionality.

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

graphite-app · 2025-07-28T02:20:23Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+
+	"github.com/jaegertracing/jaeger/model"
+	"github.com/jaegertracing/jaeger/storage/spanstore"
+)


The imports section is missing required packages for the Apache Beam window and stats functionality. Please add the following imports:

"github.com/apache/beam/sdks/v2/go/pkg/beam/window" "github.com/apache/beam/sdks/v2/go/pkg/beam/stats"

These packages are referenced in the code (e.g., beam.window.IntervalWindow and beam.stats.GroupByKey) but not imported, which will cause compilation errors.

Suggested change

import (

"context"

"sync"

"time"

"github.com/apache/beam/sdks/v2/go/pkg/beam"

"go.opentelemetry.io/collector/component"

"go.opentelemetry.io/collector/pdata/pcommon"

"go.opentelemetry.io/collector/pdata/ptrace"

"go.uber.org/zap"

"github.com/jaegertracing/jaeger/model"

"github.com/jaegertracing/jaeger/storage/spanstore"

)

import (

"context"

"sync"

"time"

"github.com/apache/beam/sdks/v2/go/pkg/beam"

"github.com/apache/beam/sdks/v2/go/pkg/beam/stats"

"github.com/apache/beam/sdks/v2/go/pkg/beam/window"

"go.opentelemetry.io/collector/component"

"go.opentelemetry.io/collector/pdata/pcommon"

"go.opentelemetry.io/collector/pdata/ptrace"

"go.uber.org/zap"

"github.com/jaegertracing/jaeger/model"

"github.com/jaegertracing/jaeger/storage/spanstore"

)

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

graphite-app · 2025-07-28T02:20:24Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+	dependencies := beam.ParDo(s, func(key pcommon.TraceID, iter func(*spanEvent) bool) []*model.DependencyLink {
+		spanMap := make(map[pcommon.SpanID]spanEvent)
+		var event *spanEvent
+
+		// Build span map
+		for iter(event) {
+			spanMap[event.span.SpanID()] = *event
+		}


There appears to be an issue with the iterator usage in this code. The pointer variable event is declared but never initialized before being passed to the iterator function. Since iter(event) expects to populate the pointer, it should be initialized first.

Consider changing:

var event *spanEvent // Build span map for iter(event) { spanMap[event.span.SpanID()] = *event }

To:

event := new(spanEvent) // Build span map for iter(event) { spanMap[event.span.SpanID()] = *event }

This ensures the iterator has a valid pointer to populate during each iteration.

Suggested change

dependencies := beam.ParDo(s, func(key pcommon.TraceID, iter func(*spanEvent) bool) []*model.DependencyLink {

spanMap := make(map[pcommon.SpanID]spanEvent)

var event *spanEvent

// Build span map

for iter(event) {

spanMap[event.span.SpanID()] = *event

}

dependencies := beam.ParDo(s, func(key pcommon.TraceID, iter func(*spanEvent) bool) []*model.DependencyLink {

spanMap := make(map[pcommon.SpanID]spanEvent)

event := new(spanEvent)

// Build span map

for iter(event) {

spanMap[event.span.SpanID()] = *event

}

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

graphite-app · 2025-07-28T02:20:25Z

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go

+func depMapToSlice(deps map[string]*model.DependencyLink) []*model.DependencyLink {
+	result := make([]*model.DependencyLink, 0, len(deps))
+	for _, dep := range deps {
+		result = append(result, dep)
+	}
+	return result


There's a type mismatch between the return value of depMapToSlice() and what the WriteDependencies() interface method expects. The function currently returns []*model.DependencyLink (slice of pointers), but the interface method expects []model.DependencyLink (slice of values). This will cause compilation errors when trying to pass the result to the writer.

Consider modifying the function to return the correct type:

func depMapToSlice(deps map[string]*model.DependencyLink) []model.DependencyLink { result := make([]model.DependencyLink, 0, len(deps)) for _, dep := range deps { result = append(result, *dep) // Dereference the pointer } return result }

Suggested change

func depMapToSlice(deps map[string]*model.DependencyLink) []*model.DependencyLink {

result := make([]*model.DependencyLink, 0, len(deps))

for _, dep := range deps {

result = append(result, dep)

}

return result

func depMapToSlice(deps map[string]*model.DependencyLink) []model.DependencyLink {

result := make([]model.DependencyLink, 0, len(deps))

for _, dep := range deps {

result = append(result, *dep)

}

return result

}

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

github-actions · 2025-10-06T02:07:36Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. You may re-open it if you need more time.

yunmaoQu requested a review from a team as a code owner January 17, 2025 17:37

yunmaoQu requested a review from joe-elliott January 17, 2025 17:37

feat: add dependency processor using Apache Beam

60fb334

Signed-off-by: yunmaoQu <[email protected]>

yunmaoQu force-pushed the add-dependency-processor branch from af5f794 to 60fb334 Compare January 17, 2025 17:42

yurishkuro reviewed Jan 17, 2025

View reviewed changes

plugin/storage/memory/memory.go Outdated Show resolved Hide resolved

yurishkuro reviewed Jan 17, 2025

View reviewed changes

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go Show resolved Hide resolved

yurishkuro reviewed Jan 17, 2025

View reviewed changes

fix: address review comments

5ccf68d

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro reviewed Jan 19, 2025

View reviewed changes

update

05dc78f

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro reviewed Jan 20, 2025

View reviewed changes

fix

3c7ef35

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro reviewed Jan 31, 2025

View reviewed changes

cmd/jaeger/internal/processors/dependencyprocessor/aggregator.go Outdated Show resolved Hide resolved

fix

de25c36

Signed-off-by: yunmaoQu <[email protected]>

yurishkuro reviewed Jan 31, 2025

View reviewed changes

yurishkuro assigned mahadzaryab1 Jan 31, 2025

yunmaoQu added 2 commits February 13, 2025 07:37

fix e2e tests

fc01ef9

Signed-off-by: yunmaoQu <[email protected]>

Merge branch 'main' of https://github.com/yunmaoQu/jaeger into add-de…

c7145d1

…pendency-processor

yurishkuro reviewed Feb 17, 2025

View reviewed changes

github-actions bot added the stale The issue/PR has become stale and may be auto-closed label Jul 28, 2025

graphite-app bot reviewed Jul 28, 2025

View reviewed changes

github-actions bot removed the stale The issue/PR has become stale and may be auto-closed label Aug 4, 2025

github-actions bot added the stale The issue/PR has become stale and may be auto-closed label Oct 6, 2025

		func (agg *dependencyAggregator) Start(closeChan chan struct{}) {
		agg.closeChan = closeChan

	func (agg *dependencyAggregator) Start(closeChan chan struct{}) {
	agg.closeChan = closeChan
	func (agg *dependencyAggregator) Start() {

feat: add dependency processor using Apache Beam #6560

Are you sure you want to change the base?

feat: add dependency processor using Apache Beam #6560

Uh oh!

Conversation

yunmaoQu commented Jan 17, 2025

Which problem is this PR solving?

Description of the changes

How was this change tested?

Checklist

Uh oh!

Uh oh!

Uh oh!

yurishkuro left a comment

Choose a reason for hiding this comment

Uh oh!

yunmaoQu commented Jan 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

yurishkuro left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yunmaoQu commented Jan 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yunmaoQu commented Jan 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yunmaoQu Feb 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yunmaoQu commented Feb 4, 2025

Uh oh!

yunmaoQu commented Feb 4, 2025

Uh oh!

yunmaoQu commented Feb 4, 2025

Uh oh!

mahadzaryab1 commented Feb 5, 2025

Uh oh!

yunmaoQu commented Jan 18, 2025 •

edited

Loading

yunmaoQu commented Jan 20, 2025 •

edited

Loading

yunmaoQu commented Jan 30, 2025 •

edited

Loading

yunmaoQu Feb 2, 2025 •

edited

Loading

yunmaoQu commented Feb 11, 2025 •

edited

Loading