DCU compute metrics

examples/manifest.yaml

@@ -92,6 +92,9 @@ data:
       - job_name: proxy
         static_configs:
           - targets: ["proxy:9187"]
+      - job_name: operator
+        static_configs:
+          - targets: ["operator:9187"]
 kind: ConfigMap
 metadata:
   name: prometheus-vol
@@ -310,6 +313,12 @@ spec:
           env:
             - name: ADDR
               value: "0.0.0.0:9187"
+            - name: METRICS_DELAY
+              value: "60"
+            - name: PROMETHEUS_URL
+              value: "http://prometheus/api/v1"
+            - name: DCU_PER_PACKAGE
+              value: "preview=5,preprod=5,mainnet=5"
 ---
 apiVersion: v1
 kind: Service
@@ -324,7 +333,7 @@ spec:
   type: ClusterIP
   ports:
     - name: operator
-      port: 80
+      port: 9187
       targetPort: 9187
       protocol: TCP
 ---

examples/setup

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-~/go/bin/kind create cluster
+kind create cluster
 
 echo "Building operator CRD"
 cargo run --bin=crdgen > env-crd.yaml --manifest-path ../operator/Cargo.toml
@@ -14,9 +14,9 @@ echo "Building operator image"
 docker build -t operator:1.0 -f ../docker/dockerfile.operator ../
 
 echo "Loading proxy image"
-~/go/bin/kind load docker-image proxy:1.0
+kind load docker-image proxy:1.0
 
 echo "Loading operator image"
-~/go/bin/kind load docker-image operator:1.0
+kind load docker-image operator:1.0
 
 kubectl apply -f manifest.yaml

operator/Cargo.toml

@@ -7,7 +7,6 @@ default-run = "controller"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-actix-web = "4.4.1"
 argon2 = "0.5.3"
 bech32 = "0.11.0"
 dotenv = "0.15.0"
@@ -24,6 +23,12 @@ thiserror = "1.0.52"
 tokio = { version = "1.35.1", features = ["macros", "rt-multi-thread"] }
 tracing = "0.1.40"
 tracing-subscriber = "0.3.18"
+http-body-util = "0.1.0"
+hyper = { version = "1.1.0", features = ["full"] }
+hyper-util = { version = "0.1.3", features = ["full"] }
+reqwest = { version = "0.11.23", features = ["json"] }
+regex = "1.10.2"
+chrono = "0.4.31"
 
 [[bin]]
 name="controller"

operator/src/config.rs

@@ -1,5 +1,5 @@
 use lazy_static::lazy_static;
-use std::env;
+use std::{collections::HashMap, env, time::Duration};
 
 lazy_static! {
     static ref CONTROLLER_CONFIG: Config = Config::from_env();
@@ -14,6 +14,9 @@ pub struct Config {
     pub dns_zone: String,
     pub extension_name: String,
     pub api_key_salt: String,
+    pub metrics_delay: Duration,
+    pub prometheus_url: String,
+    pub dcu_per_package: HashMap<String, f64>,
 }
 
 impl Config {
@@ -22,6 +25,25 @@ impl Config {
             dns_zone: env::var("DNS_ZONE").unwrap_or("demeter.run".into()),
             extension_name: env::var("EXTENSION_NAME").unwrap_or("node-m1".into()),
             api_key_salt: env::var("API_KEY_SALT").unwrap_or("cardano-node-salt".into()),
+            metrics_delay: Duration::from_secs(
+                std::env::var("METRICS_DELAY")
+                    .expect("METRICS_DELAY must be set")
+                    .parse::<u64>()
+                    .expect("METRICS_DELAY must be a number"),
+            ),
+            prometheus_url: env::var("PROMETHEUS_URL").expect("PROMETHEUS_URL must be set"),
+            dcu_per_package: env::var("DCU_PER_PACKAGE")
+                .expect("DCU_PER_PACKAGE must be set")
+                .split(',')
+                .map(|pair| {
+                    let parts: Vec<&str> = pair.split('=').collect();
+                    let dcu = parts[1]
+                        .parse::<f64>()
+                        .expect("DCU_PER_PACKAGE must be NETWORK=NUMBER");
+
+                    (parts[0].into(), dcu)
+                })
+                .collect(),
         }
     }
 }

operator/src/lib.rs

@@ -18,6 +18,12 @@ pub enum Error {
 
     #[error("Bech32 Error: {0}")]
     Bech32Error(String),
+
+    #[error("Http Request error: {0}")]
+    HttpError(String),
+
+    #[error("Config Error: {0}")]
+    ConfigError(String),
 }
 impl Error {
     pub fn metric_label(&self) -> String {
@@ -50,7 +56,7 @@ impl From<bech32::primitives::hrp::Error> for Error {
     }
 }
 
-#[derive(Clone, Default)]
+#[derive(Clone)]
 pub struct State {
     registry: Registry,
     pub metrics: Metrics,
@@ -66,6 +72,11 @@ impl State {
         self.registry.gather()
     }
 }
+impl Default for State {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
 pub enum Network {
@@ -89,8 +100,8 @@ impl Display for Network {
     }
 }
 
-pub use kube;
 pub use k8s_openapi;
+pub use kube;
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 

operator/src/main.rs

@@ -1,27 +1,9 @@
-use actix_web::{
-    get, middleware, web::Data, App, HttpRequest, HttpResponse, HttpServer, Responder,
-};
 use dotenv::dotenv;
-use prometheus::{Encoder, TextEncoder};
 use std::{io, sync::Arc};
-use tracing::{info, Level};
+use tracing::Level;
 
 use operator::{controller, metrics as metrics_collector, State};
 
-#[get("/metrics")]
-async fn metrics(c: Data<Arc<State>>, _req: HttpRequest) -> impl Responder {
-    let metrics = c.metrics_collected();
-    let encoder = TextEncoder::new();
-    let mut buffer = vec![];
-    encoder.encode(&metrics, &mut buffer).unwrap();
-    HttpResponse::Ok().body(buffer)
-}
-
-#[get("/health")]
-async fn health(_: HttpRequest) -> impl Responder {
-    HttpResponse::Ok().json("healthy")
-}
-
 #[tokio::main]
 async fn main() -> io::Result<()> {
     dotenv().ok();
@@ -30,22 +12,10 @@ async fn main() -> io::Result<()> {
 
     let state = Arc::new(State::default());
 
-    let controller = controller::run(state.clone());
-    let metrics_collector = metrics_collector::run_metrics_collector(state.clone());
-
-    let addr = std::env::var("ADDR").unwrap_or("0.0.0.0:8080".into());
-
-    let server = HttpServer::new(move || {
-        App::new()
-            .app_data(Data::new(state.clone()))
-            .wrap(middleware::Logger::default())
-            .service(health)
-            .service(metrics)
-    })
-    .bind(&addr)?;
-    info!({ addr }, "metrics server running");
+    metrics_collector::run_metrics_collector(state.clone());
+    metrics_collector::run_metrics_server(state.clone());
 
-    tokio::join!(server.run(), controller, metrics_collector).0?;
+    controller::run(state.clone()).await;
 
     Ok(())
 }