Move record-progress processing to separate thread

Cargo.toml

@@ -19,6 +19,7 @@ chrono = { version = "0.4", features = ["serde"] }
 chrono-humanize = "0.1.1"
 crates-index = "0.16.2"
 crossbeam-utils = "0.5"
+crossbeam-channel = "0.5"
 csv = "1.0.2"
 dotenv = "0.13"
 failure = "0.1.3"

src/agent/api.rs

@@ -36,6 +36,7 @@ impl ResponseExt for ::reqwest::blocking::Response {
         match self.status() {
             StatusCode::NOT_FOUND => return Err(AgentApiError::InvalidEndpoint.into()),
             StatusCode::BAD_GATEWAY
+            | StatusCode::TOO_MANY_REQUESTS
             | StatusCode::SERVICE_UNAVAILABLE
             | StatusCode::GATEWAY_TIMEOUT => {
                 return Err(AgentApiError::ServerUnavailable.into());
@@ -50,6 +51,7 @@ impl ResponseExt for ::reqwest::blocking::Response {
             .with_context(|_| format!("failed to parse API response (status code {})", status,))?;
         match result {
             ApiResponse::Success { result } => Ok(result),
+            ApiResponse::SlowDown => Err(AgentApiError::ServerUnavailable.into()),
             ApiResponse::InternalError { error } => {
                 Err(AgentApiError::InternalServerError(error).into())
             }

src/server/api_types.rs

@@ -20,6 +20,7 @@ pub struct AgentConfig {
 #[serde(tag = "status", rename_all = "kebab-case")]
 pub enum ApiResponse<T> {
     Success { result: T },
+    SlowDown,
     InternalError { error: String },
     Unauthorized,
     NotFound,
@@ -41,8 +42,9 @@ impl ApiResponse<()> {
 
 impl<T> ApiResponse<T> {
     fn status_code(&self) -> StatusCode {
-        match *self {
+        match self {
             ApiResponse::Success { .. } => StatusCode::OK,
+            ApiResponse::SlowDown => StatusCode::TOO_MANY_REQUESTS,
             ApiResponse::InternalError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             ApiResponse::Unauthorized => StatusCode::UNAUTHORIZED,
             ApiResponse::NotFound => StatusCode::NOT_FOUND,

src/server/metrics.rs

@@ -5,8 +5,8 @@ use crate::server::agents::Agent;
 use chrono::{DateTime, Utc};
 use prometheus::proto::{Metric, MetricFamily};
 use prometheus::{
-    HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, __register_counter_vec, __register_gauge,
-    __register_gauge_vec,
+    HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, __register_counter_vec,
+    __register_gauge, __register_gauge_vec, opts, register_counter, register_int_counter,
 };
 
 const JOBS_METRIC: &str = "crater_completed_jobs_total";
@@ -18,6 +18,7 @@ const ENDPOINT_TIME: &str = "crater_endpoint_time_seconds";
 #[derive(Clone)]
 pub struct Metrics {
     crater_completed_jobs_total: IntCounterVec,
+    pub crater_bounced_record_progress: IntCounter,
     crater_agent_failure: IntCounterVec,
     crater_work_status: IntGaugeVec,
     crater_last_crates_update: IntGauge,
@@ -29,6 +30,10 @@ impl Metrics {
         let jobs_opts = prometheus::opts!(JOBS_METRIC, "total completed jobs");
         let crater_completed_jobs_total =
             prometheus::register_int_counter_vec!(jobs_opts, &["agent", "experiment"])?;
+        let crater_bounced_record_progress = prometheus::register_int_counter!(
+            "crater_bounced_record_progress",
+            "hits with full record progress queue"
+        )?;
         let failure_opts = prometheus::opts!(AGENT_FAILED, "total completed jobs");
         let crater_agent_failure =
             prometheus::register_int_counter_vec!(failure_opts, &["agent", "experiment"])?;
@@ -47,6 +52,7 @@ impl Metrics {
 
         Ok(Metrics {
             crater_completed_jobs_total,
+            crater_bounced_record_progress,
             crater_agent_failure,
             crater_work_status,
             crater_last_crates_update,

src/server/mod.rs

@@ -44,6 +44,7 @@ pub struct Data {
     pub agents: Agents,
     pub db: Database,
     pub reports_worker: reports::ReportsWorker,
+    pub record_progress_worker: routes::agent::RecordProgressThread,
     pub acl: ACL,
     pub metrics: Metrics,
 }
@@ -80,6 +81,10 @@ pub fn run(config: Config, bind: SocketAddr) -> Fallible<()> {
     info!("initialized metrics...");
 
     let data = Data {
+        record_progress_worker: routes::agent::RecordProgressThread::new(
+            db.clone(),
+            metrics.clone(),
+        ),
         config,
         tokens,
         agents,
@@ -100,7 +105,9 @@ pub fn run(config: Config, bind: SocketAddr) -> Fallible<()> {
     let data = Arc::new(data);
     let github_data = github_data.map(Arc::new);
 
+    let record_progress_worker = data.record_progress_worker.clone();
     let routes = warp::any()
+        .and(warp::any().map(move || record_progress_worker.clone().start_request()))
         .and(
             warp::any()
                 .and(
@@ -118,13 +125,15 @@ pub fn run(config: Config, bind: SocketAddr) -> Fallible<()> {
                 .or(routes::ui::routes(data))
                 .unify(),
         )
-        .map(|mut resp: Response<Body>| {
-            resp.headers_mut().insert(
-                http::header::SERVER,
-                HeaderValue::from_static(&SERVER_HEADER),
-            );
-            resp
-        });
+        .map(
+            |_guard: routes::agent::RequestGuard, mut resp: Response<Body>| {
+                resp.headers_mut().insert(
+                    http::header::SERVER,
+                    HeaderValue::from_static(&SERVER_HEADER),
+                );
+                resp
+            },
+        );
 
     warp::serve(routes).run(bind);