Move compute_ctl structs used in HTTP API and spec file to separate c…

…rate.

This is in preparation of using compute_ctl to launch postgres nodes
in the neon_local control plane. And seems like a good idea to
separate the public interfaces anyway.

One non-mechanical change here is that we now use a RwLock rather than
atomics to protect the ComputeNode::metrics field. We were not using
atomics for performance but for convenience here, and an RwLock is now
more convenient.

Cargo.toml

@@ -132,6 +132,7 @@ tokio-tar = { git = "https://github.com/neondatabase/tokio-tar.git", rev="404df6
 heapless = { default-features=false, features=[], git = "https://github.com/japaric/heapless.git", rev = "644653bf3b831c6bb4963be2de24804acf5e5001" } # upstream release pending
 
 ## Local libraries
+compute_api = { version = "0.1", path = "./libs/compute_api/" }
 consumption_metrics = { version = "0.1", path = "./libs/consumption_metrics/" }
 metrics = { version = "0.1", path = "./libs/metrics/" }
 pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" }

compute_tools/Cargo.toml

@@ -27,4 +27,5 @@ tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 url.workspace = true
 
+compute_api.workspace = true
 workspace_hack.workspace = true

compute_tools/src/bin/compute_ctl.rs

@@ -41,15 +41,17 @@ use anyhow::{Context, Result};
 use chrono::Utc;
 use clap::Arg;
 use tracing::{error, info};
+use url::Url;
+
+use compute_api::models::{ComputeMetrics, ComputeState, ComputeStatus};
+use compute_api::spec::ComputeSpec;
 
-use compute_tools::compute::{ComputeMetrics, ComputeNode, ComputeState, ComputeStatus};
+use compute_tools::compute::ComputeNode;
 use compute_tools::http::api::launch_http_server;
 use compute_tools::logger::*;
 use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::pg_helpers::*;
-use compute_tools::spec::*;
-use url::Url;
 
 fn main() -> Result<()> {
     init_tracing_and_logging(DEFAULT_LOG_LEVEL)?;
@@ -145,8 +147,10 @@ fn main() -> Result<()> {
         .find("neon.timeline_id")
         .expect("tenant id should be provided");
 
+    let now = Utc::now();
+
     let compute_state = ComputeNode {
-        start_time: Utc::now(),
+        start_time: now,
         connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
         pgdata: pgdata.to_string(),
         pgbin: pgbin.to_string(),
@@ -155,8 +159,12 @@ fn main() -> Result<()> {
         timeline,
         pageserver_connstr,
         storage_auth_token,
-        metrics: ComputeMetrics::default(),
-        state: RwLock::new(ComputeState::new()),
+        metrics: RwLock::new(ComputeMetrics::default()),
+        state: RwLock::new(ComputeState {
+            status: ComputeStatus::Init,
+            last_active: now,
+            error: None,
+        }),
     };
     let compute = Arc::new(compute_state);
 

compute_tools/src/compute.rs

@@ -19,16 +19,17 @@ use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::RwLock;
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use postgres::{Client, NoTls};
-use serde::{Serialize, Serializer};
 use tokio_postgres;
 use tracing::{info, instrument, warn};
 
+use compute_api::models::{ComputeMetrics, ComputeState, ComputeStatus};
+use compute_api::spec::ComputeSpec;
+
 use crate::checker::create_writability_check_data;
 use crate::config;
 use crate::pg_helpers::*;
@@ -46,62 +47,13 @@ pub struct ComputeNode {
     pub timeline: String,
     pub pageserver_connstr: String,
     pub storage_auth_token: Option<String>,
-    pub metrics: ComputeMetrics,
+    pub metrics: RwLock<ComputeMetrics>,
     /// Volatile part of the `ComputeNode` so should be used under `RwLock`
     /// to allow HTTP API server to serve status requests, while configuration
     /// is in progress.
     pub state: RwLock<ComputeState>,
 }
 
-fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
-where
-    S: Serializer,
-{
-    x.to_rfc3339().serialize(s)
-}
-
-#[derive(Serialize)]
-#[serde(rename_all = "snake_case")]
-pub struct ComputeState {
-    pub status: ComputeStatus,
-    /// Timestamp of the last Postgres activity
-    #[serde(serialize_with = "rfc3339_serialize")]
-    pub last_active: DateTime<Utc>,
-    pub error: Option<String>,
-}
-
-impl ComputeState {
-    pub fn new() -> Self {
-        Self {
-            status: ComputeStatus::Init,
-            last_active: Utc::now(),
-            error: None,
-        }
-    }
-}
-
-impl Default for ComputeState {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-#[derive(Serialize, Clone, Copy, PartialEq, Eq)]
-#[serde(rename_all = "snake_case")]
-pub enum ComputeStatus {
-    Init,
-    Running,
-    Failed,
-}
-
-#[derive(Default, Serialize)]
-pub struct ComputeMetrics {
-    pub sync_safekeepers_ms: AtomicU64,
-    pub basebackup_ms: AtomicU64,
-    pub config_ms: AtomicU64,
-    pub total_startup_ms: AtomicU64,
-}
-
 impl ComputeNode {
     pub fn set_status(&self, status: ComputeStatus) {
         self.state.write().unwrap().status = status;
@@ -155,15 +107,11 @@ impl ComputeNode {
         ar.set_ignore_zeros(true);
         ar.unpack(&self.pgdata)?;
 
-        self.metrics.basebackup_ms.store(
-            Utc::now()
-                .signed_duration_since(start_time)
-                .to_std()
-                .unwrap()
-                .as_millis() as u64,
-            Ordering::Relaxed,
-        );
-
+        self.metrics.write().unwrap().basebackup_ms = Utc::now()
+            .signed_duration_since(start_time)
+            .to_std()
+            .unwrap()
+            .as_millis() as u64;
         Ok(())
     }
 
@@ -201,14 +149,11 @@ impl ComputeNode {
             );
         }
 
-        self.metrics.sync_safekeepers_ms.store(
-            Utc::now()
-                .signed_duration_since(start_time)
-                .to_std()
-                .unwrap()
-                .as_millis() as u64,
-            Ordering::Relaxed,
-        );
+        self.metrics.write().unwrap().sync_safekeepers_ms = Utc::now()
+            .signed_duration_since(start_time)
+            .to_std()
+            .unwrap()
+            .as_millis() as u64;
 
         let lsn = String::from(String::from_utf8(sync_output.stdout)?.trim());
 
@@ -340,23 +285,19 @@ impl ComputeNode {
         self.apply_config()?;
 
         let startup_end_time = Utc::now();
-        self.metrics.config_ms.store(
-            startup_end_time
+        {
+            let mut metrics = self.metrics.write().unwrap();
+            metrics.config_ms = startup_end_time
                 .signed_duration_since(start_time)
                 .to_std()
                 .unwrap()
-                .as_millis() as u64,
-            Ordering::Relaxed,
-        );
-        self.metrics.total_startup_ms.store(
-            startup_end_time
+                .as_millis() as u64;
+            metrics.total_startup_ms = startup_end_time
                 .signed_duration_since(self.start_time)
                 .to_std()
                 .unwrap()
-                .as_millis() as u64,
-            Ordering::Relaxed,
-        );
-
+                .as_millis() as u64;
+        }
         self.set_status(ComputeStatus::Running);
 
         Ok(pg)

compute_tools/src/config.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 use anyhow::Result;
 
 use crate::pg_helpers::PgOptionsSerialize;
-use crate::spec::ComputeSpec;
+use compute_api::spec::ComputeSpec;
 
 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.

compute_tools/src/http/api.rs

@@ -31,7 +31,8 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
         // future use for Prometheus metrics format.
         (&Method::GET, "/metrics.json") => {
             info!("serving /metrics.json GET request");
-            Response::new(Body::from(serde_json::to_string(&compute.metrics).unwrap()))
+            let metrics = compute.metrics.read().unwrap();
+            Response::new(Body::from(serde_json::to_string(&*metrics).unwrap()))
         }
 
         // Collect Postgres current usage insights