enhancement(api): Support tapping component inputs with vector tap (#11293)

* Add inputs as part of tap's watched resources

* Wire up inputs_of/outputs_of pattern options

Requires a change in the API to rename the patterns argument to
outputsPatterns and support an inputsPatterns argument.

* Implement basic input tapping

* Fix notifications for matching input patterns

* Fix existing tests given new types

* Add new tests for tapping inputs

* Minor clean up

* Make inputsPatterns optional in subscription

* Fix clippy errors

* Update cli cue

* Add 0.21 version guide entry for api change

* component features

Signed-off-by: Lee Benson <[email protected]>

Co-authored-by: Lee Benson <[email protected]>

Author: wbew

lib/vector-api-client/graphql/schema.json

@@ -5566,7 +5566,7 @@
               "description": "A stream of events emitted from matched component ID patterns",
               "args": [
                 {
-                  "name": "patterns",
+                  "name": "outputsPatterns",
                   "description": null,
                   "type": {
                     "kind": "NON_NULL",
@@ -5587,6 +5587,24 @@
                   },
                   "defaultValue": null
                 },
+                {
+                  "name": "inputsPatterns",
+                  "description": null,
+                  "type": {
+                    "kind": "LIST",
+                    "name": null,
+                    "ofType": {
+                      "kind": "NON_NULL",
+                      "name": null,
+                      "ofType": {
+                        "kind": "SCALAR",
+                        "name": "String",
+                        "ofType": null
+                      }
+                    }
+                  },
+                  "defaultValue": null
+                },
                 {
                   "name": "interval",
                   "description": null,

lib/vector-api-client/graphql/subscriptions/output_events_by_component_id_patterns.graphql

@@ -1,6 +1,6 @@
 subscription OutputEventsByComponentIdPatternsSubscription(
-    $patterns: [String!]!, $limit: Int!, $interval: Int!, $encoding: EventEncodingType!){
-    outputEventsByComponentIdPatterns(patterns: $patterns, limit: $limit, interval: $interval) {
+    $outputsPatterns: [String!]!, $inputsPatterns: [String!], $limit: Int!, $interval: Int!, $encoding: EventEncodingType!){
+    outputEventsByComponentIdPatterns(outputsPatterns: $outputsPatterns, inputsPatterns: $inputsPatterns, limit: $limit, interval: $interval) {
         __typename
         ... on Log {
             componentId

lib/vector-api-client/src/gql/tap.rs

@@ -69,7 +69,8 @@ pub trait TapSubscriptionExt {
     /// Executes an output events subscription.
     fn output_events_by_component_id_patterns_subscription(
         &self,
-        component_patterns: Vec<String>,
+        outputs_patterns: Vec<String>,
+        inputs_patterns: Vec<String>,
         encoding: TapEncodingFormat,
         limit: i64,
         interval: i64,
@@ -80,14 +81,16 @@ impl TapSubscriptionExt for crate::SubscriptionClient {
     /// Executes an output events subscription.
     fn output_events_by_component_id_patterns_subscription(
         &self,
-        patterns: Vec<String>,
+        outputs_patterns: Vec<String>,
+        inputs_patterns: Vec<String>,
         encoding: TapEncodingFormat,
         limit: i64,
         interval: i64,
     ) -> BoxedSubscription<OutputEventsByComponentIdPatternsSubscription> {
         let request_body = OutputEventsByComponentIdPatternsSubscription::build_query(
             output_events_by_component_id_patterns_subscription::Variables {
-                patterns,
+                outputs_patterns,
+                inputs_patterns: Some(inputs_patterns),
                 limit,
                 interval,
                 encoding: encoding.into(),

src/api/schema/events/mod.rs

@@ -5,6 +5,8 @@ pub mod notification;
 pub mod output;
 pub mod trace;
 
+use std::collections::HashSet;
+
 use async_graphql::{Context, Subscription};
 use encoding::EventEncodingType;
 use futures::Stream;
@@ -16,6 +18,32 @@ use tokio_stream::wrappers::ReceiverStream;
 
 use crate::{api::tap::TapController, topology::WatchRx};
 
+/// Patterns (glob) used by tap to match against components and access events
+/// flowing into (for_inputs) or out of (for_outputs) specified components
+#[derive(Debug)]
+pub struct TapPatterns {
+    pub for_outputs: HashSet<String>,
+    pub for_inputs: HashSet<String>,
+}
+
+impl TapPatterns {
+    pub const fn new(for_outputs: HashSet<String>, for_inputs: HashSet<String>) -> Self {
+        Self {
+            for_outputs,
+            for_inputs,
+        }
+    }
+
+    /// Get all user-specified patterns
+    pub fn all_patterns(&self) -> HashSet<String> {
+        self.for_outputs
+            .iter()
+            .cloned()
+            .chain(self.for_inputs.iter().cloned())
+            .collect()
+    }
+}
+
 #[derive(Debug, Default)]
 pub struct EventsSubscription;
 
@@ -25,12 +53,17 @@ impl EventsSubscription {
     pub async fn output_events_by_component_id_patterns<'a>(
         &'a self,
         ctx: &'a Context<'a>,
-        patterns: Vec<String>,
+        outputs_patterns: Vec<String>,
+        inputs_patterns: Option<Vec<String>>,
         #[graphql(default = 500)] interval: u32,
         #[graphql(default = 100, validator(minimum = 1, maximum = 10_000))] limit: u32,
     ) -> impl Stream<Item = Vec<OutputEventsPayload>> + 'a {
         let watch_rx = ctx.data_unchecked::<WatchRx>().clone();
 
+        let patterns = TapPatterns {
+            for_outputs: outputs_patterns.into_iter().collect(),
+            for_inputs: inputs_patterns.unwrap_or_default().into_iter().collect(),
+        };
         // Client input is confined to `u32` to provide sensible bounds.
         create_events_stream(watch_rx, patterns, interval as u64, limit as usize)
     }
@@ -41,7 +74,7 @@ impl EventsSubscription {
 /// all matching events; filtering should be done at the caller level.
 pub(crate) fn create_events_stream(
     watch_rx: WatchRx,
-    component_id_patterns: Vec<String>,
+    patterns: TapPatterns,
     interval: u64,
     limit: usize,
 ) -> impl Stream<Item = Vec<OutputEventsPayload>> {
@@ -57,7 +90,7 @@ pub(crate) fn create_events_stream(
     tokio::spawn(async move {
         // Create a tap controller. When this drops out of scope, clean up will be performed on the
         // event handlers and topology observation that the tap controller provides.
-        let _tap_controller = TapController::new(watch_rx, tap_tx, &component_id_patterns);
+        let _tap_controller = TapController::new(watch_rx, tap_tx, patterns);
 
         // A tick interval to represent when to 'cut' the results back to the client.
         let mut interval = time::interval(time::Duration::from_millis(interval));