fix(utilization_metric): run a separate task for utilization to ensure it is regularly published #22070

esensar · 2024-12-21T12:51:38Z

Summary

This adds a separate task that runs periodically to emit utilization metrics and collect messages from components that need their utilization metrics calculated. This ensures that utilization metric is published even when no events are running through a component.

Change Type

Bug fix
New feature
Non-functional (chore, refactoring, docs)
Performance

Is this a breaking change?

Yes
No

How did you test this PR?

Ran vector with internal metrics and observer that utilization was updated every ~5 secs, instead of only when events are running.

Does this PR include user facing changes?

Yes. Please add a changelog fragment based on our guidelines.
No. A maintainer will apply the "no-changelog" label to this PR.

Checklist

Please read our Vector contributor resources.
If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
run dd-rust-license-tool write to regenerate the license inventory and commit the changes (if any). More details here.

References

Closes: Prometheus stats "stuck" on last value seen for transforms using aggregations (vector_utilization) #20216

…e it is regularly published This adds a separate task that runs periodically to emit utilization metrics and collect messages from components that need their utilization metrics calculated. This ensures that utilization metric is published even when no events are running through a component. Fixes: vectordotdev#20216

esensar · 2024-12-21T12:52:32Z

I have left this as a draft, since I am not sure how to handle shutdown (which shutdown signal to use) and how to name the task (or maybe run it in a completely different way, to not mix it up with components).

Also, gauge is passed into the timer instead of using the macro inside the timer to ensure that correct labels are inherited from the tracing context.

esensar · 2025-01-09T16:24:31Z

@pront
Any suggestion for running this separate task? It is currently started as following:

running_topology.utilization_task =
    // TODO: how to name this custom task?
    Some(tokio::spawn(Task::new("".into(), "", async move {
        utilization_emitter
            .run_utilization(ShutdownSignal::noop())
            .await;
        // TODO: new task output type for this? Or handle this task in a completely
        // different way
        Ok(TaskOutput::Healthcheck)
    })));

I am not sure how to pass the shutdown signal to it (and if I should do it at all, it made sense to me, but I might have misunderstood some part of the topology). Also, I currently create a task with empty name, but maybe it would make more sense to run it in a different way compared to other tasks?

pront · 2025-01-09T20:50:55Z

Hi @esensar,

This is a complex so I checked out this PR to do some testing;

config:

api:
  enabled: true

sources:
  internal_metrics_1:
    type: internal_metrics

transforms:
  filter_utilization:
    type: filter
    inputs: ["internal_metrics_1"]
    condition: .name == "utilization"

sinks:
  console:
    inputs: ["filter_utilization"]
    type: console
    encoding:
      codec: json
      json:
        pretty: true

Sample output:

/Users/pavlos.rontidis/.cargo/bin/cargo run --color=always --profile dev -- --config /Users/pavlos.rontidis/CLionProjects/vector/pront/configs/internal_metrics.yaml
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.70s
     Running `target/debug/vector --config /Users/pavlos.rontidis/CLionProjects/vector/pront/configs/internal_metrics.yaml`
2025-01-09T20:46:27.736727Z  INFO vector::app: Log level is enabled. level="info"
2025-01-09T20:46:27.741218Z  INFO vector::app: Loading configs. paths=["/Users/pavlos.rontidis/CLionProjects/vector/pront/configs/internal_metrics.yaml"]
2025-01-09T20:46:27.766384Z  INFO vector::topology::running: Running healthchecks.
2025-01-09T20:46:27.767489Z  INFO vector::topology::builder: Healthcheck passed.
2025-01-09T20:46:27.769222Z  INFO vector: Vector has started. debug="true" version="0.44.0" arch="aarch64" revision=""
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "console",
    "component_kind": "sink",
    "component_type": "console",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:27.770905Z",
  "kind": "absolute",
  "gauge": {
    "value": 1.0
  }
}
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "filter_utilization",
    "component_kind": "transform",
    "component_type": "filter",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:27.770905Z",
  "kind": "absolute",
  "gauge": {
    "value": 1.0
  }
}
2025-01-09T20:46:27.777873Z  INFO vector::internal_events::api: API server running. address=127.0.0.1:8686 playground=http://127.0.0.1:8686/playground graphql=http://127.0.0.1:8686/graphql
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "filter_utilization",
    "component_kind": "transform",
    "component_type": "filter",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:37.771882Z",
  "kind": "absolute",
  "gauge": {
    "value": 0.010011816446046937
  }
}
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "console",
    "component_kind": "sink",
    "component_type": "console",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:37.771882Z",
  "kind": "absolute",
  "gauge": {
    "value": 0.01004418815411736
  }
}
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "filter_utilization",
    "component_kind": "transform",
    "component_type": "filter",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:47.771505Z",
  "kind": "absolute",
  "gauge": {
    "value": 0.0001184493997704478
  }
}
{
  "name": "utilization",
  "namespace": "vector",
  "tags": {
    "component_id": "console",
    "component_kind": "sink",
    "component_type": "console",
    "host": "COMP-LPF0JYPP2Q"
  },
  "timestamp": "2025-01-09T20:46:47.771505Z",
  "kind": "absolute",
  "gauge": {
    "value": 0.00010693227629135064
  }
}
...

Leaving this here as context. Will followup with more questions.

src/topology/builder.rs

pront · 2025-01-09T20:52:39Z

src/topology/running.rs

@@ -1053,6 +1055,17 @@ impl RunningTopology {
        running_topology.connect_diff(&diff, &mut pieces).await;
        running_topology.spawn_diff(&diff, pieces);

+        running_topology.utilization_task =


(Still trying to parse the details here)
Do we join this handle at any point?

I don't think we do. I forgot to add that it seems. Should it be joined in stop? I can see that other tasks are joined there.

Yes, that sounds right.

pront · 2025-01-09T20:53:47Z

src/topology/running.rs

@@ -1053,6 +1055,17 @@ impl RunningTopology {
        running_topology.connect_diff(&diff, &mut pieces).await;
        running_topology.spawn_diff(&diff, pieces);

+        running_topology.utilization_task =
+            // TODO: how to name this custom task?
+            Some(tokio::spawn(Task::new("".into(), "", async move {


A possible name utilization_heartbeat. But here I have a more basic question, do we expect this to repeat every 5 seconds (the hardcoded value)? In my test it seems like it was way more frequent.

Yes, it is expected every 5 seconds. Not sure what went wrong there, in my testing it was repeated every 5 seconds (even though utilization was printed from the sink every second, it was only updated by this component every 5).

pront · 2025-01-09T21:20:20Z

cc @lukesteensen (just in case you are interested in this one)

Co-authored-by: Pavlos Rontidis <[email protected]>

github-actions bot added the domain: topology Anything related to Vector's topology code label Dec 21, 2024

Add changelog entry

59b7d19

pront self-assigned this Jan 2, 2025

pront reviewed Jan 9, 2025

View reviewed changes

esensar and others added 3 commits January 10, 2025 13:49

Remove unnecessary clone when building utilization task

416e1cf

Co-authored-by: Pavlos Rontidis <[email protected]>

Name utilization task utilization_heartbeat

bb5d7ca

Join utilization_task when stopping topology

d7d8694

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(utilization_metric): run a separate task for utilization to ensure it is regularly published #22070

fix(utilization_metric): run a separate task for utilization to ensure it is regularly published #22070

esensar commented Dec 21, 2024

esensar commented Dec 21, 2024 •

edited

Loading

esensar commented Jan 9, 2025

pront commented Jan 9, 2025

pront Jan 9, 2025

esensar Jan 10, 2025

pront Jan 10, 2025 •

edited

Loading

pront Jan 9, 2025

esensar Jan 10, 2025

pront commented Jan 9, 2025

fix(utilization_metric): run a separate task for utilization to ensure it is regularly published #22070

Are you sure you want to change the base?

fix(utilization_metric): run a separate task for utilization to ensure it is regularly published #22070

Conversation

esensar commented Dec 21, 2024

Summary

Change Type

Is this a breaking change?

How did you test this PR?

Does this PR include user facing changes?

Checklist

References

esensar commented Dec 21, 2024 • edited Loading

esensar commented Jan 9, 2025

pront commented Jan 9, 2025

pront Jan 9, 2025

Choose a reason for hiding this comment

esensar Jan 10, 2025

Choose a reason for hiding this comment

pront Jan 10, 2025 • edited Loading

Choose a reason for hiding this comment

pront Jan 9, 2025

Choose a reason for hiding this comment

esensar Jan 10, 2025

Choose a reason for hiding this comment

pront commented Jan 9, 2025

esensar commented Dec 21, 2024 •

edited

Loading

pront Jan 10, 2025 •

edited

Loading