Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

enhancement(observability): Events processed metrics for Topology #4472

Closed
wants to merge 7 commits into from
Closed

enhancement(observability): Events processed metrics for Topology #4472

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0a5c6bc
source events
leebenson Oct 9, 2020
c0b6fcb
all topology
leebenson Oct 9, 2020
caf4fd3
less matchy
leebenson Oct 9, 2020
99227f4
clippy love
leebenson Oct 9, 2020
65fcfaa
Merge branch 'master' into leebenson/source-events-processed
leebenson Oct 9, 2020
05bd94c
Merge branch 'master' into leebenson/source-events-processed
leebenson Oct 9, 2020
87636d7
Update schema.json
leebenson Oct 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions graphql/schema.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,18 @@
},
"isDeprecated": false,
"deprecationReason": null
},
{
"name": "eventsProcessed",
"description": "Metric indicating events processed for the current sink",
"args": [],
"type": {
"kind": "OBJECT",
"name": "EventsProcessed",
"ofType": null
},
"isDeprecated": false,
"deprecationReason": null
}
],
"inputFields": null,
Expand Down Expand Up @@ -503,6 +515,18 @@
},
"isDeprecated": false,
"deprecationReason": null
},
{
"name": "eventsProcessed",
"description": "Metric indicating events processed for the current source",
"args": [],
"type": {
"kind": "OBJECT",
"name": "EventsProcessed",
"ofType": null
},
"isDeprecated": false,
"deprecationReason": null
}
],
"inputFields": null,
Expand Down Expand Up @@ -741,6 +765,18 @@
},
"isDeprecated": false,
"deprecationReason": null
},
{
"name": "eventsProcessed",
"description": null,
"args": [],
"type": {
"kind": "OBJECT",
"name": "EventsProcessed",
"ofType": null
},
"isDeprecated": false,
"deprecationReason": null
}
],
"inputFields": null,
Expand Down Expand Up @@ -832,6 +868,18 @@
},
"isDeprecated": false,
"deprecationReason": null
},
{
"name": "eventsProcessed",
"description": "Metric indicating events processed for the current transform",
"args": [],
"type": {
"kind": "OBJECT",
"name": "EventsProcessed",
"ofType": null
},
"isDeprecated": false,
"deprecationReason": null
}
],
"inputFields": null,
Expand Down
30 changes: 29 additions & 1 deletion src/api/schema/metrics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,19 @@
use crate::event::{Event, Metric, MetricValue};
use crate::metrics::{capture_metrics, get_controller};
use crate::metrics::{capture_metrics, get_controller, Controller};
use async_graphql::{validators::IntRange, Interface, Object, Subscription};
use async_stream::stream;
use chrono::{DateTime, Utc};
use lazy_static::lazy_static;
use std::sync::{Arc, Mutex};
use tokio::stream::{Stream, StreamExt};
use tokio::time::Duration;

lazy_static! {
static ref GLOBAL_CONTROLLER: Arc<Mutex<&'static Controller>> = Arc::new(Mutex::new(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this imply that all our metrics emit code will be interacting with blocking mutexes?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, this is actually old code. It's been removed in #4431. I initially thought it would be necessary to lock the controller for reads, since GraphQL resolvers are async and could be pulling from stats simultaneously. But the Controller is already capable of parallel reads.

get_controller().expect("Metrics system not initialized. Please report.")
));
}

pub struct Uptime(Metric);

#[Object]
Expand Down Expand Up @@ -153,3 +161,23 @@ fn get_metrics(interval: i32) -> impl Stream<Item = Metric> {
}
}
}

/// Get the events processed by topology component name
pub fn topology_events_processed(topology_name: String) -> Option<EventsProcessed> {
let key = String::from("component_name");

capture_metrics(
*GLOBAL_CONTROLLER
.lock()
.expect("Couldn't acquire metrics lock. Please report."),
)
.find(|ev| match ev {
Event::Metric(m)
if m.name.as_str().eq("events_processed") && m.tag_matches(&key, &topology_name) =>
{
true
}
_ => false,
})
.map(|ev| EventsProcessed(ev.into_metric()))
}
21 changes: 20 additions & 1 deletion src/api/schema/topology.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use super::metrics;
use crate::config::{Config, DataType};
use async_graphql::{Enum, Interface, Object};
use lazy_static::lazy_static;
Expand Down Expand Up @@ -62,6 +63,11 @@ impl Source {
_ => None,
})
}

/// Metric indicating events processed for the current source
async fn events_processed(&self) -> Option<metrics::EventsProcessed> {
metrics::topology_events_processed(self.0.name.clone())
}
}

#[derive(Clone)]
Expand Down Expand Up @@ -102,6 +108,11 @@ impl Transform {
_ => None,
})
}

/// Metric indicating events processed for the current transform
async fn events_processed(&self) -> Option<metrics::EventsProcessed> {
metrics::topology_events_processed(self.0.name.clone())
}
}

#[derive(Clone)]
Expand Down Expand Up @@ -143,9 +154,17 @@ impl Sink {
})
.collect()
}

/// Metric indicating events processed for the current sink
async fn events_processed(&self) -> Option<metrics::EventsProcessed> {
metrics::topology_events_processed(self.0.name.clone())
}
}

#[Interface(field(name = "name", type = "String"))]
#[Interface(
field(name = "name", type = "String"),
field(name = "events_processed", type = "Option<metrics::EventsProcessed>")
)]
#[derive(Clone)]
pub enum Topology {
Source(Source),
Expand Down
8 changes: 8 additions & 0 deletions src/event/metric.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,14 @@ impl Metric {
value,
}
}

/// Returns `true` if `name` tag is present, and matches the provided `value`
pub fn tag_matches(&self, name: &str, value: &str) -> bool {
self.tags
.as_ref()
.filter(|t| t.get(name).filter(|v| *v == value).is_some())
.is_some()
}
}

impl Display for Metric {
Expand Down