Introduce dynamic routing connector #932
Conversation
rogercoll
left a comment
rogercoll
left a comment
There was a problem hiding this comment.
Could you add this component to the versions.yaml file? 🙏 (created this issue to add a CI check)
vigneshshanmugam
left a comment
vigneshshanmugam
left a comment
There was a problem hiding this comment.
Did a quick pass, Approach looks good to me. However I felt if there is any other way we can make the configuration simpler. I ll do another extensive review in a followup.
There was a problem hiding this comment.
We should emit some metrics on the selected pipeline with the provided metadata keys as attributes, but can come as a followup.
There was a problem hiding this comment.
Yeah, a follow-up is planned.
axw
left a comment
axw
left a comment
There was a problem hiding this comment.
Looks good in general, left some suggestions
| return c.router.Shutdown(ctx) | ||
| } | ||
|
|
||
| func (c *metricsConnector) Capabilities() consumer.Capabilities { |
There was a problem hiding this comment.
Ques - Is this required? Shouldn't it default to false anyway?
There was a problem hiding this comment.
There is no default, we need to implement connector.Metrics interface and this is required.
|
|
||
| ```yaml | ||
| connectors: | ||
| dynamicrouting: |
There was a problem hiding this comment.
After sleeping over it, I think we can still improve the configuration to a better model, currently it leaks metadatakeys across several keys which is a bit confusing. Taking a bit of inspiration from our ratelimiter resolvers, I m thinking something like
connectors:
dynamicrouting:
# Metadata configuration
routing_keys:
partition_by: ["x-tenant-id"] # serves the same purpose as primary
measure_by: ["x-forwarded-for", "user-agent"] # replaces the metadata_keys
routing_class:
low_cardinality:
max_cardinality: 10
pipelines: ["traces/small_batch"]
medium_cardinality:
max_cardinality: 100
pipelines: ["traces/medium_batch"]
high_cardinality:
max_cardinality: .inf
pipelines: ["traces/large_batch"]
default_class: ["traces/default"]
evaluation_interval: 30s
overrides: #similar to what we do with ratelimiter
- matches:
x-tenant-id: ["premium"]
pipelines: ["traces/premium"] Thoughts? We dont need to introduce overrides, just thought it would be valuable to have it in our sleeves.
There was a problem hiding this comment.
I don't like overrides - atleast for now but I like others. Will make the changes. For overrides, lets follow-up if we see the need.
There was a problem hiding this comment.
routing_class: low_cardinality:
I am a bit more inclined towards keeping this as a slice instead of a map, and I am not sure about the routing_class name - it is more of a bucket which is chosen based on calculated cardinality. WDYT about keeping this bit the same?
| for _, mk := range vs { | ||
| if _, err := pkb.WriteString(mk); err != nil { | ||
| r.logger.Error( | ||
| "unexpected failure on concatenating primary metadata keys", |
There was a problem hiding this comment.
keys -> values, would be good to log the values to make it easy for debugging?
| } | ||
| } | ||
|
|
||
| func newTestMetrics(resourceIDs, scopeIDs, metricIDs, sumDPIDs string) pmetric.Metrics { |
There was a problem hiding this comment.
Ques - dont we have any testutil in upstream that can create a dummy Metric, Log, trace record?
axw
left a comment
axw
left a comment
There was a problem hiding this comment.
LGTM. I like @vigneshshanmugam's suggestion for the config to be in terms of partition keys and counting keys - seems clearer to me. I'm good with it either way though.
4 participants
Dynamic routing connector allows selecting pipelines based on the configured set of metadata keys. For example: route to different batching pipelines based on observed connection count.
Related to: https://github.com/elastic/hosted-otel-collector/issues/1349