feat: add dependency processor using Apache Beam #6560
Conversation
Signed-off-by: yunmaoQu <[email protected]>
yunmaoQu
force-pushed
the
add-dependency-processor
branch
from
af5f794 to
60fb334
Compare
plugin/storage/memory/memory.go
Outdated
| perTenant: make(map[string]*Tenant), | ||
| defaultConfig: cfg, | ||
| perTenant: make(map[string]*Tenant), | ||
| useNewDependencies: false, // 添加初始化 |
There was a problem hiding this comment.
please use English in comments
| // Copyright (c) 2025 The Jaeger Authors. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package dependencyprocessor |
There was a problem hiding this comment.
if this is placed under /processors/ is needs to follow the OTEL Processor framework patterns, like having a factory, etc.
@yurishkuro I have fixed it |
Signed-off-by: yunmaoQu <[email protected]>
| type Config struct { | ||
| AggregationInterval time.Duration `yaml:"aggregation_interval"` | ||
| InactivityTimeout time.Duration `yaml:"inactivity_timeout"` | ||
| Store *memory.Store `yaml:"-"` |
There was a problem hiding this comment.
Storage is not part of the config. The processor itself can retrieve storage from storage extension - see adaptive sampling processor for example.
| ) | ||
|
|
||
| type Config struct { | ||
| AggregationInterval time.Duration `yaml:"aggregation_interval"` |
| require.NoError(t, err) | ||
|
|
||
| // Wait for the processor to process the trace | ||
| time.Sleep(cfg.AggregationInterval + 100*time.Millisecond) |
There was a problem hiding this comment.
We don't use sleep in tests, use assert.Eventually
|
|
||
| func DefaultConfig() Config { | ||
| return Config{ | ||
| AggregationInterval: 5 * time.Second, // Default dependency aggregation interval: 5 seconds |
There was a problem hiding this comment.
This is way too frequent. We used to flush every 15min at Uber, since the topology does not change that often, but I can see a need for more frequent flushing if we want to capture movement of metrics.
| AggregationInterval: 5 * time.Second, // Default dependency aggregation interval: 5 seconds | |
| AggregationInterval: 10 * time.Minute, |
The comment is redundant.
There was a problem hiding this comment.
It would be nice to flush every minute to support investigating metrics movement, but it will create too much data. We may have to consider integrating with Prometheus instead, and only focus on capturing topology here, which definitely does not need 1min frequency.
| } | ||
|
|
||
| // createDefaultConfig returns the default configuration for the dependency processor. | ||
| func createDefaultConfig() component.Config { |
There was a problem hiding this comment.
Why is it repeated vs config? Please keep in one place.
| "github.com/jaegertracing/jaeger/storage_v2/v1adapter" | ||
| ) | ||
|
|
||
| type dependencyProcessor struct { |
There was a problem hiding this comment.
This is not using Apache beam.
I think that's not a bad thing in the initial phase. When running with all in one, beam is not needed (unless we want to support checkpoints). But in multi-collector deployment this solution won't work. We can address it in the following steps.
There was a problem hiding this comment.
This is not using Apache beam.
I think that's not a bad thing in the initial phase. When running with all in one, beam is not needed (unless we want to support checkpoints). But in multi-collector deployment this solution won't work. We can address it in the following steps.
Except this ,I update all based on your review.
| func TestDependencyProcessorEndToEnd(t *testing.T) { | ||
| // Create a mock receiver, processor, and exporter | ||
| factory := NewFactory() | ||
| cfg := factory.CreateDefaultConfig().(*Config) |
There was a problem hiding this comment.
Should use very short aggregation interval here
Signed-off-by: yunmaoQu <[email protected]>
|
@yurishkuro Except this ,I update all based on your review. |
Which problem is this PR solving?
Resolves #5911
Description of the changes
How was this change tested?
Checklist
jaeger:make lint testjaeger-ui:npm run lintandnpm run test