Repository: vectordotdev/vector (language: rust)

Issue:
vector top componentname pattern*
### A note for the community

<!-- Please keep this note for the community -->
* Please vote on this issue by adding a 👍 [reaction](https://blog.github.com/2016-03-10-add-reactions-to-pull-requests-issues-and-comments/) to the original issue to help the community and maintainers prioritize this request
* If you are interested in working on this issue or have submitted a pull request, please leave a comment
<!-- Thank you for keeping this note for the community -->


### Use Cases

allow limiting monitored components for "vector top" by specifying a list of component names and/or component name partterns

### Attempted Solutions

_No response_

### Proposal

_No response_

### References

_No response_

### Version

_No response_


Discussion / hints:

Fix patch:
diff --git a/changelog.d/14282_vector_top_filter_components.feature.md b/changelog.d/14282_vector_top_filter_components.feature.md
new file mode 100644
index 0000000000000..38311e4603f8e
--- /dev/null
+++ b/changelog.d/14282_vector_top_filter_components.feature.md
@@ -0,0 +1,10 @@
+`vector top` now supports filtering out components by their component ID using glob patterns with a new `--components` option.
+This is very similar to `vector tap` `--outputs-of` and `--inputs-of` options. This can be useful
+in cases where you have a lot of components and they don't fit in the terminal (as scrolling is not supported yet in `vector top`).
+By default, all components are shown with a glob pattern of `*`.
+
+The glob pattern semantics can be found in the [`glob` crate documentation](https://docs.rs/glob/latest/glob/).
+
+Example usage: `vector top --components "demo*"` will only show the components that match the glob pattern `demo*`.
+
+authors: jorgehermo9
diff --git a/src/top/cmd.rs b/src/top/cmd.rs
index 148c46957855c..12a3b7b45caf7 100644
--- a/src/top/cmd.rs
+++ b/src/top/cmd.rs
@@ -98,7 +98,7 @@ async fn subscription(
         // Initialize state. On future reconnects, we re-initialize state in
         // order to accurately capture added, removed, and edited
         // components.
-        let state = match metrics::init_components(&client).await {
+        let state = match metrics::init_components(&client, &opts.components).await {
             Ok(state) => state,
             Err(_) => {
                 tokio::time::sleep(Duration::from_millis(RECONNECT_DELAY)).await;
@@ -116,7 +116,12 @@ async fn subscription(
         };
 
         // Subscribe to updated metrics
-        let finished = metrics::subscribe(subscription_client, tx.clone(), opts.interval as i64);
+        let finished = metrics::subscribe(
+            subscription_client,
+            tx.clone(),
+            opts.interval as i64,
+            opts.components.clone(),
+        );
 
         _ = tx
             .send(EventType::ConnectionUpdated(ConnectionStatus::Connected(
diff --git a/src/top/metrics.rs b/src/top/metrics.rs
index 67a3d76a753a5..057974790ad11 100644
--- a/src/top/metrics.rs
+++ b/src/top/metrics.rs
@@ -3,6 +3,7 @@ use std::{
     sync::Arc,
 };
 
+use glob::Pattern;
 use tokio::task::JoinHandle;
 use tokio_stream::StreamExt;
 use vector_lib::api_client::{
@@ -13,8 +14,22 @@ use vector_lib::api_client::{
 use super::state::{self, OutputMetrics};
 use crate::{config::ComponentKey, top::state::SentEventsMetric};
 
+fn component_matches_patterns(component_id: &str, components_patterns: &[Pattern]) -> bool {
+    if components_patterns.is_empty() {
+        return true;
+    }
+
+    components_patterns
+        .iter()
+        .any(|pattern| pattern.matches(component_id))
+}
+
 /// Components that have been added
-async fn component_added(client: Arc<SubscriptionClient>, tx: state::EventTx) {
+async fn component_added(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_added();
     };
@@ -22,7 +37,11 @@ async fn component_added(client: Arc<SubscriptionClient>, tx: state::EventTx) {
     while let Some(Some(res)) = stream.next().await {
         if let Some(d) = res.data {
             let c = d.component_added;
-            let key = ComponentKey::from(c.component_id);
+            let component_id = c.component_id;
+            if !component_matches_patterns(&component_id, &components_patterns) {
+                continue;
+            }
+            let key = ComponentKey::from(component_id);
             _ = tx
                 .send(state::EventType::ComponentAdded(state::ComponentRow {
                     key,
@@ -48,7 +67,12 @@ async fn component_added(client: Arc<SubscriptionClient>, tx: state::EventTx) {
 
 /// Allocated bytes per component
 #[cfg(feature = "allocation-tracing")]
-async fn allocated_bytes(client: Arc<SubscriptionClient>, tx: state::EventTx, interval: i64) {
+async fn allocated_bytes(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_allocated_bytes_subscription(interval);
     };
@@ -59,9 +83,12 @@ async fn allocated_bytes(client: Arc<SubscriptionClient>, tx: state::EventTx, in
             _ = tx
                 .send(state::EventType::AllocatedBytes(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| {
                             (
-                                ComponentKey::from(c.component_id.as_str()),
+                                ComponentKey::from(c.component_id),
                                 c.metric.allocated_bytes as i64,
                             )
                         })
@@ -72,7 +99,11 @@ async fn allocated_bytes(client: Arc<SubscriptionClient>, tx: state::EventTx, in
     }
 }
 /// Components that have been removed
-async fn component_removed(client: Arc<SubscriptionClient>, tx: state::EventTx) {
+async fn component_removed(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_removed();
     };
@@ -80,13 +111,22 @@ async fn component_removed(client: Arc<SubscriptionClient>, tx: state::EventTx)
     while let Some(Some(res)) = stream.next().await {
         if let Some(d) = res.data {
             let c = d.component_removed;
-            let id = ComponentKey::from(c.component_id.as_str());
+            let component_id = c.component_id;
+            if !component_matches_patterns(&component_id, &components_patterns) {
+                continue;
+            }
+            let id = ComponentKey::from(component_id);
             _ = tx.send(state::EventType::ComponentRemoved(id)).await;
         }
     }
 }
 
-async fn received_bytes_totals(client: Arc<SubscriptionClient>, tx: state::EventTx, interval: i64) {
+async fn received_bytes_totals(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_received_bytes_totals_subscription(interval);
     };
@@ -94,12 +134,16 @@ async fn received_bytes_totals(client: Arc<SubscriptionClient>, tx: state::Event
     while let Some(Some(res)) = stream.next().await {
         if let Some(d) = res.data {
             let c = d.component_received_bytes_totals;
+
             _ = tx
                 .send(state::EventType::ReceivedBytesTotals(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| {
                             (
-                                ComponentKey::from(c.component_id.as_str()),
+                                ComponentKey::from(c.component_id),
                                 c.metric.received_bytes_total as i64,
                             )
                         })
@@ -114,6 +158,7 @@ async fn received_bytes_throughputs(
     client: Arc<SubscriptionClient>,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
 ) {
     tokio::pin! {
         let stream = client.component_received_bytes_throughputs_subscription(interval);
@@ -126,7 +171,10 @@ async fn received_bytes_throughputs(
                 .send(state::EventType::ReceivedBytesThroughputs(
                     interval,
                     c.into_iter()
-                        .map(|c| (ComponentKey::from(c.component_id.as_str()), c.throughput))
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
+                        .map(|c| (ComponentKey::from(c.component_id), c.throughput))
                         .collect(),
                 ))
                 .await;
@@ -138,6 +186,7 @@ async fn received_events_totals(
     client: Arc<SubscriptionClient>,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
 ) {
     tokio::pin! {
         let stream = client.component_received_events_totals_subscription(interval);
@@ -149,9 +198,12 @@ async fn received_events_totals(
             _ = tx
                 .send(state::EventType::ReceivedEventsTotals(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| {
                             (
-                                ComponentKey::from(c.component_id.as_str()),
+                                ComponentKey::from(c.component_id),
                                 c.metric.received_events_total as i64,
                             )
                         })
@@ -166,6 +218,7 @@ async fn received_events_throughputs(
     client: Arc<SubscriptionClient>,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
 ) {
     tokio::pin! {
         let stream = client.component_received_events_throughputs_subscription(interval);
@@ -178,7 +231,10 @@ async fn received_events_throughputs(
                 .send(state::EventType::ReceivedEventsThroughputs(
                     interval,
                     c.into_iter()
-                        .map(|c| (ComponentKey::from(c.component_id.as_str()), c.throughput))
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
+                        .map(|c| (ComponentKey::from(c.component_id), c.throughput))
                         .collect(),
                 ))
                 .await;
@@ -186,7 +242,12 @@ async fn received_events_throughputs(
     }
 }
 
-async fn sent_bytes_totals(client: Arc<SubscriptionClient>, tx: state::EventTx, interval: i64) {
+async fn sent_bytes_totals(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_sent_bytes_totals_subscription(interval);
     };
@@ -197,9 +258,12 @@ async fn sent_bytes_totals(client: Arc<SubscriptionClient>, tx: state::EventTx,
             _ = tx
                 .send(state::EventType::SentBytesTotals(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| {
                             (
-                                ComponentKey::from(c.component_id.as_str()),
+                                ComponentKey::from(c.component_id),
                                 c.metric.sent_bytes_total as i64,
                             )
                         })
@@ -214,6 +278,7 @@ async fn sent_bytes_throughputs(
     client: Arc<SubscriptionClient>,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
 ) {
     tokio::pin! {
         let stream = client.component_sent_bytes_throughputs_subscription(interval);
@@ -226,7 +291,10 @@ async fn sent_bytes_throughputs(
                 .send(state::EventType::SentBytesThroughputs(
                     interval,
                     c.into_iter()
-                        .map(|c| (ComponentKey::from(c.component_id.as_str()), c.throughput))
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
+                        .map(|c| (ComponentKey::from(c.component_id), c.throughput))
                         .collect(),
                 ))
                 .await;
@@ -234,7 +302,12 @@ async fn sent_bytes_throughputs(
     }
 }
 
-async fn sent_events_totals(client: Arc<SubscriptionClient>, tx: state::EventTx, interval: i64) {
+async fn sent_events_totals(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_sent_events_totals_subscription(interval);
     };
@@ -245,6 +318,9 @@ async fn sent_events_totals(client: Arc<SubscriptionClient>, tx: state::EventTx,
             _ = tx
                 .send(state::EventType::SentEventsTotals(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| SentEventsMetric {
                             key: ComponentKey::from(c.component_id.as_str()),
                             total: c.metric.sent_events_total as i64,
@@ -261,6 +337,7 @@ async fn sent_events_throughputs(
     client: Arc<SubscriptionClient>,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
 ) {
     tokio::pin! {
         let stream = client.component_sent_events_throughputs_subscription(interval);
@@ -273,6 +350,9 @@ async fn sent_events_throughputs(
                 .send(state::EventType::SentEventsThroughputs(
                     interval,
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| SentEventsMetric {
                             key: ComponentKey::from(c.component_id.as_str()),
                             total: c.throughput,
@@ -285,7 +365,12 @@ async fn sent_events_throughputs(
     }
 }
 
-async fn errors_totals(client: Arc<SubscriptionClient>, tx: state::EventTx, interval: i64) {
+async fn errors_totals(
+    client: Arc<SubscriptionClient>,
+    tx: state::EventTx,
+    interval: i64,
+    components_patterns: Arc<Vec<Pattern>>,
+) {
     tokio::pin! {
         let stream = client.component_errors_totals_subscription(interval);
     };
@@ -296,6 +381,9 @@ async fn errors_totals(client: Arc<SubscriptionClient>, tx: state::EventTx, inte
             _ = tx
                 .send(state::EventType::ErrorsTotals(
                     c.into_iter()
+                        .filter(|c| {
+                            component_matches_patterns(&c.component_id, &components_patterns)
+                        })
                         .map(|c| {
                             (
                                 ComponentKey::from(c.component_id.as_str()),
@@ -315,57 +403,91 @@ pub fn subscribe(
     client: SubscriptionClient,
     tx: state::EventTx,
     interval: i64,
+    components_patterns: Vec<Pattern>,
 ) -> Vec<JoinHandle<()>> {
     let client = Arc::new(client);
-
+    let components_patterns = Arc::new(components_patterns);
     vec![
-        tokio::spawn(component_added(Arc::clone(&client), tx.clone())),
-        tokio::spawn(component_removed(Arc::clone(&client), tx.clone())),
+        tokio::spawn(component_added(
+            Arc::clone(&client),
+            tx.clone(),
+            Arc::clone(&components_patterns),
+        )),
+        tokio::spawn(component_removed(
+            Arc::clone(&client),
+            tx.clone(),
+            Arc::clone(&components_patterns),
+        )),
         tokio::spawn(received_bytes_totals(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         tokio::spawn(received_bytes_throughputs(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         tokio::spawn(received_events_totals(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         tokio::spawn(received_events_throughputs(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
+        )),
+        tokio::spawn(sent_bytes_totals(
+            Arc::clone(&client),
+            tx.clone(),
+            interval,
+            Arc::clone(&components_patterns),
         )),
-        tokio::spawn(sent_bytes_totals(Arc::clone(&client), tx.clone(), interval)),
         tokio::spawn(sent_bytes_throughputs(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         tokio::spawn(sent_events_totals(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         tokio::spawn(sent_events_throughputs(
             Arc::clone(&client),
             tx.clone(),
             interval,
+            Arc::clone(&components_patterns),
         )),
         #[cfg(feature = "allocation-tracing")]
-        tokio::spawn(allocated_bytes(Arc::clone(&client), tx.clone(), interval)),
-        tokio::spawn(errors_totals(Arc::clone(&client), tx, interval)),
+        tokio::spawn(allocated_bytes(
+            Arc::clone(&client),
+            tx.clone(),
+            interval,
+            Arc::clone(&components_patterns),
+        )),
+        tokio::spawn(errors_totals(
+            Arc::clone(&client),
+            tx,
+            interval,
+            Arc::clone(&components_patterns),
+        )),
     ]
 }
 
 /// Retrieve the initial components/metrics for first paint. Further updating the metrics
 /// will be handled by subscriptions.
-pub async fn init_components(client: &Client) -> Result<state::State, ()> {
+pub async fn init_components(
+    client: &Client,
+    components_patterns: &[Pattern],
+) -> Result<state::State, ()> {
     // Execute a query to get the latest components, and aggregate metrics for each resource.
     // Since we don't know currently have a mechanism for scrolling/paging through results,
     // we're using an artificially high page size to capture all likely component configurations.
@@ -378,10 +500,11 @@ pub async fn init_components(client: &Client) -> Result<state::State, ()> {
         .components
         .edges
         .into_iter()
-        .flat_map(|edge| {
+        .filter(|edge| component_matches_patterns(&edge.node.component_id, components_patterns))
+        .map(|edge| {
             let d = edge.node;
             let key = ComponentKey::from(d.component_id);
-            Some((
+            (
                 key.clone(),
                 state::ComponentRow {
                     key,
@@ -405,7 +528,7 @@ pub async fn init_components(client: &Client) -> Result<state::State, ()> {
                     allocated_bytes: 0,
                     errors: 0,
                 },
-            ))
+            )
         })
         .collect::<BTreeMap<_, _>>();
 
diff --git a/src/top/mod.rs b/src/top/mod.rs
index 790d7c6025dc0..2e92398cfabc0 100644
--- a/src/top/mod.rs
+++ b/src/top/mod.rs
@@ -9,6 +9,7 @@ use clap::Parser;
 pub use cmd::cmd;
 pub use cmd::top;
 pub use dashboard::is_tty;
+use glob::Pattern;
 use url::Url;
 
 use crate::config::api::default_graphql_url;
@@ -34,6 +35,10 @@ pub struct Opts {
     /// By default, top will attempt to reconnect if the connection drops.
     #[arg(short, long)]
     no_reconnect: bool,
+
+    /// Components IDs to observe (comma-separated; accepts glob patterns)
+    #[arg(default_value = "*", value_delimiter(','), short = 'c', long)]
+    components: Vec<Pattern>,
 }
 
 impl Opts {
diff --git a/website/cue/reference/cli.cue b/website/cue/reference/cli.cue
index f191a17427820..763188264df78 100644
--- a/website/cue/reference/cli.cue
+++ b/website/cue/reference/cli.cue
@@ -400,15 +400,21 @@ cli: {
 			}
 
 			options: {
-				"refresh-interval": {
+				"components": {
+					_short:      "c"
+					description: "Components IDs to observe (comma-separated; accepts glob patterns)"
+					type:        "list"
+					default:     "*"
+				}
+				"interval": {
 					_short:      "i"
-					description: "How often the screen refreshes (in milliseconds)"
+					description: "Interval to sample metrics at, in milliseconds"
 					type:        "integer"
-					default:     500
+					default:     1000
 				}
 				"url": {
 					_short:      "u"
-					description: "The URL for the GraphQL endpoint of the running Vector instance"
+					description: "Vector GraphQL API server endpoint"
 					type:        "string"
 				}
 			}


Test patch: