Implement Observation

This commit uploads observations in parallel. It appears
our API only supports uploading a single observation
at a time, but we collect a possible vector observations,
one for each group, or potentially more depending on
how many were created during the obseravtion period.

This commit sends a request to the backend to upload
a new observation for each observation, but it does
it in parallel for performance.

Author: RobbieMcKinstry

Cargo.lock

@@ -3,17 +3,20 @@ use std::sync::Arc;
 
 use super::{
     BoxedIngress, BoxedMonitor, BoxedPlatform, IngressBuilder, MonitorBuilder, PlatformBuilder,
+    StatusCode,
 };
 use crate::Cli;
 use crate::fs::UserCreds;
+use crate::{artifacts::LambdaZip, fs::Session, metrics::ResponseStatusCode};
+use miette::miette;
 use miette::{IntoDiagnostic, Result, bail};
 use multitool_sdk::apis::{Api, ApiClient, configuration::Configuration};
-use multitool_sdk::models::{ApplicationDetails, LoginRequest, LoginSuccess, WorkspaceSummary};
-use uuid::Uuid;
-
-use crate::{
-    artifacts::LambdaZip, fs::Session, metrics::ResponseStatusCode, stats::CategoricalObservation,
+use multitool_sdk::models::{
+    ApplicationDetails, ApplicationGroup, CreateResponseCodeMetricsRequest,
+    CreateResponseCodeMetricsSuccess, LoginRequest, LoginSuccess, WorkspaceSummary,
 };
+use tokio::task::JoinSet;
+use uuid::Uuid;
 
 pub(crate) use deploy_meta::*;
 
@@ -159,8 +162,47 @@ impl BackendClient {
     }
 
     /// Upload a batch of observations to the backend.
-    pub async fn upload_observations(&self, data: Vec<()>) -> Result<()> {
-        todo!();
+    pub async fn upload_observations(
+        &self,
+        meta: &DeploymentMetadata,
+        data: Vec<StatusCode>,
+    ) -> Result<()> {
+        let mut req_waiter = JoinSet::new();
+
+        for item in data {
+            let group = match item.group() {
+                crate::stats::Group::Control => ApplicationGroup::Baseline,
+                crate::stats::Group::Experimental => ApplicationGroup::Canary,
+            };
+            let req_body = CreateResponseCodeMetricsRequest {
+                app_class: group,
+                status_2xx_count: item.get_count(&ResponseStatusCode::_2XX) as i32,
+                status_4xx_count: item.get_count(&ResponseStatusCode::_4XX) as i32,
+                status_5xx_count: item.get_count(&ResponseStatusCode::_5XX) as i32,
+            };
+            let workspace_id = meta.workspace_id().to_string();
+            let application_id = meta.application_id().to_string();
+            let deployment_id = *meta.deployment_id();
+            let cloned_client = self.clone();
+            req_waiter.spawn_local(async move {
+                cloned_client
+                    .client
+                    .response_code_metrics_api()
+                    .create_response_code_metrics(
+                        &workspace_id,
+                        &application_id,
+                        deployment_id,
+                        req_body,
+                    )
+                    .await
+            });
+        }
+        let results = req_waiter.join_all().await;
+        let result: std::result::Result<Vec<CreateResponseCodeMetricsSuccess>, _> =
+            results.into_iter().collect();
+        result
+            .map(|_| ())
+            .map_err(|err| miette!("Error uploading observation: {err}"))
     }
 
     /// Return information about the workspace given its name.

src/adapters/backend/mod.rs

@@ -40,6 +40,14 @@ impl<const N: usize, Cat: Categorical<N>> CategoricalObservation<N, Cat> {
     pub fn increment_by(&mut self, category: &Cat, count: u32) {
         self.histogram.increment_by(category, count);
     }
+
+    pub fn group(&self) -> Group {
+        self.group
+    }
+
+    pub fn get_count(&self, cat: &Cat) -> u32 {
+        self.histogram.get_count(cat)
+    }
 }
 
 impl<const N: usize, Cat: Categorical<N> + fmt::Debug> fmt::Debug

src/stats/observation.rs

@@ -1,4 +1,10 @@
-use crate::adapters::{BackendClient, DeploymentMetadata, StateId};
+use async_trait::async_trait;
+
+use crate::{
+    Shutdownable,
+    adapters::{BackendClient, DeploymentMetadata, StateId},
+    subsystems::ShutdownResult,
+};
 
 pub(super) struct LockManagementSubsystem {
     /// We use this client to refresh locks.
@@ -9,3 +15,11 @@ pub(super) struct LockManagementSubsystem {
     /// This is the state that this manager is locking.
     state_id: StateId,
 }
+
+#[async_trait]
+impl Shutdownable for LockManagementSubsystem {
+    async fn shutdown(&mut self) -> ShutdownResult {
+        // Release any of the locks we've taken.
+        todo!()
+    }
+}

src/subsystems/relay/lock_mgmt.rs

@@ -7,7 +7,7 @@ use tokio::{select, sync::mpsc::Receiver, time::Duration};
 use tokio_graceful_shutdown::{IntoSubsystem, SubsystemBuilder, SubsystemHandle};
 
 use crate::{
-    adapters::{BackendClient, BoxedIngress, BoxedPlatform, DeploymentMetadata},
+    adapters::{BackendClient, BoxedIngress, BoxedPlatform, DeploymentMetadata, StatusCode},
     stats::Observation,
 };
 
@@ -80,7 +80,7 @@ impl<T: Observation + Send + 'static> RelaySubsystem<T> {
 }
 
 #[async_trait]
-impl<T: Observation + Send + Sync> IntoSubsystem<Report> for RelaySubsystem<T> {
+impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
     async fn run(mut self, subsys: SubsystemHandle) -> Result<()> {
         // Kick off a task to poll the backend for new states.
         let mut poller = self.new_poller();
@@ -111,8 +111,7 @@ impl<T: Observation + Send + Sync> IntoSubsystem<Report> for RelaySubsystem<T> {
                 //   When a new observation arrives, we send it to the backend.
                 elem = observations.recv() => {
                     if let Some(batch) = elem {
-                        // self.backend.upload_observations(batch).await?;
-                        self.backend.upload_observations(vec![]).await?;
+                        self.backend.upload_observations(&self.meta, batch).await?;
                     } else {
                         // The stream has been closed, so we should shutdown.
                         subsys.request_shutdown();