fixup

.gitignore

@@ -17,6 +17,7 @@ production_database_voice_clone_report_download.sh
 secrets.toml
 secrets.yaml
 twitch_secrets.toml
+.DS_Store
 
 # Old env secrets
 .*-secrets

Cargo.lock

@@ -3074,6 +3074,7 @@ dependencies = [
  "my-workspace-hack",
  "mysql_queries",
  "once_cell",
+ "opentelemetry",
  "r2d2_redis",
  "rand 0.8.5",
  "redis_common",
@@ -4161,6 +4162,22 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "opentelemetry"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e32339a5dc40459130b3bd269e9892439f55b33e772d2a9d402a789baaf4e8a"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "indexmap",
+ "js-sys",
+ "once_cell",
+ "pin-project-lite",
+ "thiserror",
+ "urlencoding",
+]
+
 [[package]]
 name = "ordered-multimap"
 version = "0.4.3"

crates/service/job/inference_job/Cargo.toml

@@ -80,3 +80,4 @@ my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" }
 time = "0.3.30"
 mime = "0.3.17"
 rand = { version = "0.8.5", features = [] }
+opentelemetry = {version = "0.21.0", features = ["metrics"]}

crates/service/job/inference_job/src/job/job_loop/main_loop.rs

@@ -9,13 +9,17 @@ use jobs_common::noop_logger::NoOpLogger;
 use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::{AvailableInferenceJob, list_available_generic_inference_jobs, ListAvailableGenericInferenceJobArgs};
 use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_completely_failed::mark_generic_inference_job_completely_failed;
 use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_failure::mark_generic_inference_job_failure;
+use opentelemetry::metrics::AsyncInstrument;
 
+use crate::OTEL_METER_NAME;
 use crate::job::job_loop::clear_full_filesystem::clear_full_filesystem;
 use crate::job::job_loop::process_single_job::process_single_job;
 use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
 use crate::job::job_loop::process_single_job_success_case::ProcessSingleJobSuccessCase;
 use crate::job_dependencies::JobDependencies;
 
+use opentelemetry::{global as otel, KeyValue as OtelAttribute, metrics::Unit};
+
 // Job runner timeouts (guards MySQL)
 const START_TIMEOUT_MILLIS : u64 = 500;
 const INCREASE_TIMEOUT_MILLIS : u64 = 1000;
@@ -118,7 +122,12 @@ pub async fn main_loop(job_dependencies: JobDependencies) {
 
 async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<AvailableInferenceJob>) -> AnyhowResult<()> {
   let job_count = jobs.len();
+  let meter = otel::meter(OTEL_METER_NAME);
+  // TODO: total_duration as histogram for min/max/avg/percentiles
+  let job_duration_metric = meter.i64_observable_gauge("job_duration").with_unit(Unit::new("ms")).init();
+
   for (i, job) in jobs.into_iter().enumerate() {
+    let start_time = Instant::now();
     let result = process_single_job(job_dependencies, &job).await;
 
     match result {
@@ -145,6 +154,15 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<Availab
         let _r = handle_error(&job_dependencies, &job, err).await?;
       }
     }
+    let job_duration = Instant::now().duration_since(start_time);;
+    job_duration_metric.observe(job_duration.as_millis() as i64,
+      &[
+        // TODO: need job_ prefix?
+        OtelAttribute::new("job_id", format!("{:?}", job.id)),
+        OtelAttribute::new("job_inference_category", job.inference_category.to_str()),
+        OtelAttribute::new("job_inference_token", job.inference_job_token.as_str().to_owned()),
+      ]
+    )
   }
 
   Ok(())

crates/service/job/inference_job/src/main.rs

@@ -22,6 +22,7 @@ use std::path::PathBuf;
 use std::time::Duration;
 
 use log::{info, warn};
+use opentelemetry::metrics;
 use r2d2_redis::r2d2;
 use r2d2_redis::RedisConnectionManager;
 use sqlx::mysql::MySqlPoolOptions;
@@ -72,6 +73,8 @@ const ENV_PUBLIC_BUCKET_NAME : &str = "PUBLIC_BUCKET_NAME";
 // HTTP sidecar
 const ENV_TTS_INFERENCE_SIDECAR_HOSTNAME: &str = "TTS_INFERENCE_SIDECAR_HOSTNAME";
 
+const OTEL_METER_NAME: &str = "inference-job";
+
 //#[tokio::main]
 #[actix_web::main]
 async fn main() -> AnyhowResult<()> {