graph-builder: instrument and expose graph metrics

graph-builder/Cargo.toml

@@ -4,14 +4,21 @@ version = "0.1.0"
 authors = ["Alex Crawford <crawford@redhat.com>"]
 
 [dependencies]
+actix = "^0.7.6"
 actix-web = "^0.7.8"
 cincinnati = { path = "../cincinnati" }
 commons = { path = "../commons" }
+dkregistry = { git = "https://github.com/camallo/dkregistry-rs.git", rev = "a10c1aba444ce5c07e113c81160eb67c9711e327" }
 env_logger = "^0.6.0"
-itertools = "^0.7.8"
 failure = "^0.1.1"
 flate2 = "^1.0.1"
+futures = "0.1"
+itertools = "^0.7.8"
+lazy_static = "^1.2.0"
 log = "^0.4.3"
+prometheus = "^0.4.2"
+quay = { path = "../quay" }
+regex = "^1.1.0"
 reqwest = "^0.9.0"
 semver = { version = "^0.9.0", features = [ "serde" ] }
 serde = "^1.0.70"
@@ -20,12 +27,7 @@ serde_json = "^1.0.22"
 structopt = "^0.2.10"
 tar = "^0.4.16"
 tokio = "0.1"
-dkregistry = { git = "https://github.com/camallo/dkregistry-rs.git", rev = "a10c1aba444ce5c07e113c81160eb67c9711e327" }
-futures = "0.1"
-quay = { path = "../quay" }
 url = "^1.7.2"
-regex = "^1.1.0"
-
 
 [features]
 test-net = []

graph-builder/src/graph.rs

@@ -17,13 +17,44 @@ use cincinnati::{plugins, AbstractRelease, Graph, Release, CONTENT_TYPE};
 use commons::GraphError;
 use config;
 use failure::Error;
+use prometheus::{Counter, IntGauge};
 use registry::{self, Registry};
 use serde_json;
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, RwLock};
 use std::thread;
 
+lazy_static! {
+    static ref GRAPH_FINAL_RELEASES: IntGauge = register_int_gauge!(
+        "cincinnati_gb_graph_final_releases",
+        "Number of releases in the final graph, after processing"
+    )
+    .unwrap();
+    static ref GRAPH_UPSTREAM_RAW_RELEASES: IntGauge = register_int_gauge!(
+        "cincinnati_gb_graph_upstream_raw_releases",
+        "Number of releases fetched from upstream, before processing"
+    )
+    .unwrap();
+    static ref UPSTREAM_SCRAPES: Counter = register_counter!(
+        "cincinnati_gb_graph_upstream_scrapes_total",
+        "Total number of upstream scrapes"
+    )
+    .unwrap();
+    static ref UPSTREAM_ERRORS: Counter = register_counter!(
+        "cincinnati_gb_graph_upstream_errors_total",
+        "Total number of upstream scraping errors"
+    )
+    .unwrap();
+    static ref V1_GRAPH_INCOMING_REQS: Counter = register_counter!(
+        "cincinnati_gb_v1_graph_incoming_requests_total",
+        "Total number of incoming HTTP client request to /v1/graph"
+    )
+    .unwrap();
+}
+
 pub fn index(req: HttpRequest<State>) -> Result<HttpResponse, GraphError> {
+    V1_GRAPH_INCOMING_REQS.inc();
+
     // Check that the client can accept JSON media type.
     commons::ensure_content_type(req.headers(), CONTENT_TYPE)?;
 
@@ -113,14 +144,17 @@ pub fn run<'a>(opts: &'a config::Options, state: &State) -> ! {
 
         debug!("graph update triggered");
 
-        let releases = match registry::fetch_releases(
+        let scrape = registry::fetch_releases(
             &registry,
             &opts.repository,
             username.as_ref().map(String::as_ref),
             password.as_ref().map(String::as_ref),
             &mut cache,
             &opts.quay_manifestref_key,
-        ) {
+        );
+        UPSTREAM_SCRAPES.inc();
+
+        let releases = match scrape {
             Ok(releases) => {
                 if releases.is_empty() {
                     warn!(
@@ -132,11 +166,13 @@ pub fn run<'a>(opts: &'a config::Options, state: &State) -> ! {
                 releases
             }
             Err(err) => {
+                UPSTREAM_ERRORS.inc();
                 err.iter_chain()
                     .for_each(|cause| error!("failed to fetch all release metadata: {}", cause));
                 vec![]
             }
         };
+        GRAPH_UPSTREAM_RAW_RELEASES.set(releases.len() as i64);
 
         let graph = match create_graph(releases) {
             Ok(graph) => graph,
@@ -164,10 +200,9 @@ pub fn run<'a>(opts: &'a config::Options, state: &State) -> ! {
         match serde_json::to_string(&graph) {
             Ok(json) => {
                 *state.json.write().expect("json lock has been poisoned") = json;
-                debug!(
-                    "graph update completed, {} valid releases",
-                    graph.releases_count()
-                );
+                let nodes_count = graph.releases_count();
+                GRAPH_FINAL_RELEASES.set(nodes_count as i64);
+                debug!("graph update completed, {} valid releases", nodes_count);
             }
             Err(err) => error!("Failed to serialize graph: {}", err),
         };

graph-builder/src/lib.rs

@@ -1,29 +1,34 @@
+extern crate actix_web;
 extern crate cincinnati;
 extern crate commons;
 extern crate dkregistry;
 extern crate env_logger;
+#[macro_use]
+extern crate failure;
 extern crate flate2;
 extern crate futures;
 extern crate itertools;
+#[macro_use]
+extern crate lazy_static;
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate prometheus;
+extern crate quay;
+extern crate regex;
 extern crate reqwest;
 extern crate semver;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
-extern crate actix_web;
 extern crate serde_json;
-extern crate tar;
-extern crate tokio;
-#[macro_use]
-extern crate failure;
-#[macro_use]
-extern crate log;
 #[macro_use]
 extern crate structopt;
-extern crate quay;
-extern crate regex;
+extern crate tar;
+extern crate tokio;
 
 pub mod config;
 pub mod graph;
+pub mod metrics;
 pub mod registry;
 pub mod release;

graph-builder/src/main.rs

@@ -12,21 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+extern crate actix;
 extern crate actix_web;
 extern crate failure;
 extern crate graph_builder;
 extern crate log;
 extern crate structopt;
 
-use graph_builder::{config, graph};
+use graph_builder::{config, graph, metrics};
 
 use actix_web::{http::Method, middleware::Logger, server, App};
 use failure::Error;
 use log::LevelFilter;
-use std::thread;
+use std::{net, thread};
 use structopt::StructOpt;
 
 fn main() -> Result<(), Error> {
+    let sys = actix::System::new("graph-builder");
+
     let opts = config::Options::from_args();
 
     env_logger::Builder::from_default_env()
@@ -50,6 +53,20 @@ fn main() -> Result<(), Error> {
         thread::spawn(move || graph::run(&opts, &state));
     }
 
+    // TODO(lucab): make these configurable.
+    let status_address: net::IpAddr = net::Ipv4Addr::UNSPECIFIED.into();
+    let status_port = 9080;
+
+    // Status service.
+    server::new(|| {
+        App::new()
+            .middleware(Logger::default())
+            .route("/metrics", Method::GET, metrics::serve)
+    })
+    .bind((status_address, status_port))?
+    .start();
+
+    // Main service.
     server::new(move || {
         let app_prefix = app_prefix.clone();
         let state = state.clone();
@@ -59,6 +76,8 @@ fn main() -> Result<(), Error> {
             .route("/v1/graph", Method::GET, graph::index)
     })
     .bind(addr)?
-    .run();
+    .start();
+
+    sys.run();
     Ok(())
 }

graph-builder/src/metrics.rs

@@ -0,0 +1,21 @@
+//! Metrics service.
+
+use actix_web::{HttpRequest, HttpResponse};
+use futures::future;
+use futures::prelude::*;
+use prometheus;
+
+/// Serve metrics requests (Prometheus textual format).
+pub fn serve(_req: HttpRequest<()>) -> Box<Future<Item = HttpResponse, Error = failure::Error>> {
+    use prometheus::Encoder;
+
+    let resp = future::ok(prometheus::gather())
+        .and_then(|metrics| {
+            let tenc = prometheus::TextEncoder::new();
+            let mut buf = vec![];
+            tenc.encode(&metrics, &mut buf).and(Ok(buf))
+        })
+        .from_err()
+        .map(|content| HttpResponse::Ok().body(content));
+    Box::new(resp)
+}