From 99da9c6b876f51766a436867d9c687a3c6882f3b Mon Sep 17 00:00:00 2001 From: Jason Hall Date: Sat, 2 May 2026 19:38:04 -0400 Subject: [PATCH] metrics: ship Cloud Monitoring time series; refactor gcp_auth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements monitoring-plan.md. - src/gcp_auth.rs: shared TokenProvider. Mints a multi-scope (logging.write + monitoring.write) JWT once and caches the bearer; cloud_log + metrics threads share it. Also moves http_post, parse_signing_key, b64url, now_unix_secs, unix_to_rfc3339, device_mac out of cloud_log.rs. - src/cloud_log.rs: cloud_log::run takes Arc instead of owning the JWT/token state. LogQueue::stats() exposes depth + lifetime drop count for the metrics path. - src/metrics.rs: periodic snapshot → POST /v3/projects/{id}/timeSeries. GAUGE INT64 metrics under custom.googleapis.com/esp32/: free_heap, free_heap_internal, min_free_heap, largest_free_block, stack_hwm (label task=main|ota|cloud_log|metrics), wifi_rssi, wifi_channel, cpu_freq_mhz, uptime_secs, nvs_used/free_entries, cloud_log_queue_depth, cloud_log_dropped_total. Auto-creates MetricDescriptors on first POST. Cadence via NVS metric_intvl (gcp namespace), default 300s, 0 disables. - Each spawned thread publishes its TaskHandle_t into a module-level AtomicUsize so metrics can read uxTaskGetStackHighWaterMark per task without holding handles for them. (Xtensa StackType_t is uint8_t, so the API returns bytes — no word-size multiplication.) Plus: - sdkconfig: enable CONFIG_MBEDTLS_DYNAMIC_BUFFER (+ free_config_data, free_ca_cert) so three concurrent TLS clients (cloud_log + metrics + OTA) don't blow our heap budget. Recovers ~15-25 KB per session. - Makefile: `make flash` and `make run` now pass --partition-table. Without it, espflash 4.x writes a *default* (factory-only) partition table over our OTA layout, silently bricking OTA on the device. - README + provisioning.toml.example: gcloud command for roles/monitoring.metricWriter; document metrics_interval_secs. - tools/provision: write metric_intvl NVS u32 (default 300). Co-Authored-By: Claude Opus 4.7 (1M context) --- Makefile | 8 +- README.md | 37 +++- provisioning.toml.example | 4 +- sdkconfig.defaults.in | 10 ++ src/cloud_log.rs | 321 +++++++-------------------------- src/gcp_auth.rs | 269 ++++++++++++++++++++++++++++ src/main.rs | 63 +++++-- src/metrics.rs | 347 ++++++++++++++++++++++++++++++++++++ src/ota.rs | 1 + tools/provision/src/main.rs | 14 ++ 10 files changed, 793 insertions(+), 281 deletions(-) create mode 100644 src/gcp_auth.rs create mode 100644 src/metrics.rs diff --git a/Makefile b/Makefile index 5dd7e3f..3268236 100644 --- a/Makefile +++ b/Makefile @@ -102,8 +102,12 @@ ensure-python-shim: ln -sf "$$PY" $(PYTHON_SHIM)/python3 && \ ln -sf "$$PY" $(PYTHON_SHIM)/python +# Pass --partition-table on every flash, even app-only. Without it, +# espflash 4.x writes a *default* (factory-only) partition table over +# our OTA layout — silently bricking OTA on the device. Re-passing +# partitions.csv is a no-op when it already matches what's on flash. flash: build - espflash flash --port $(PORT) $(BIN) + espflash flash --port $(PORT) --partition-table partitions.csv $(BIN) # Full reflash: erase entire flash, then write bootloader + partition table # + app. Use this when the partition table changes, or as the recovery @@ -124,7 +128,7 @@ monitor: espflash monitor --port $(PORT) $(if $(MAKE_TERMOUT),,--non-interactive) run: build - espflash flash --port $(PORT) --monitor $(BIN) + espflash flash --port $(PORT) --partition-table partitions.csv --monitor $(BIN) clean: cargo clean diff --git a/README.md b/README.md index 21da0d5..c715fcb 100644 --- a/README.md +++ b/README.md @@ -74,12 +74,18 @@ PAT setup) and a real cosign OIDC flow the first time per ~10min window — a browser pops to authenticate. CI does this automatically via the GitHub Actions workflow's ambient OIDC token. -## Optional: GCP Cloud Logging +## Optional: GCP Cloud Logging + Monitoring The firmware can ship structured `tracing` events (both app and OTA) to -Google Cloud Logging. Cloud logging is **opt-in per device** — without -the `[gcp]` block in `provisioning.toml` the device boots normally and -just emits to serial. Full design in [`logs-plan.md`](logs-plan.md). +Google Cloud Logging *and* periodic chip-health metrics (heap, stack, +wifi, cpu, uptime, …) to Cloud Monitoring. Both are **opt-in per +device** — without the `[gcp]` block in `provisioning.toml` the device +boots normally and just emits to serial. Full design in +[`logs-plan.md`](logs-plan.md) and [`monitoring-plan.md`](monitoring-plan.md). + +One service account and one key cover both APIs (the JWT requests +both scopes; one cached access token is shared by the cloud_log and +metrics threads). One-time GCP setup, using `gcloud`: @@ -93,14 +99,18 @@ gcloud iam service-accounts create $SA_NAME \ --display-name="ESP32 device logger" \ --project=$PROJECT_ID -# Grant only logging.logWriter — least privilege. The device can write -# log entries; it cannot read, delete, or do anything else. +# Grant only logging.logWriter + monitoring.metricWriter — least +# privilege. The device can write log entries and metric points; +# it cannot read, delete, or do anything else. gcloud projects add-iam-policy-binding $PROJECT_ID \ --member="serviceAccount:$SA_EMAIL" \ --role="roles/logging.logWriter" +gcloud projects add-iam-policy-binding $PROJECT_ID \ + --member="serviceAccount:$SA_EMAIL" \ + --role="roles/monitoring.metricWriter" # Create + download a JSON key. Keep this file safe — anyone with it -# can write logs as this SA. +# can write logs + metrics as this SA. gcloud iam service-accounts keys create gcp-sa-key.json \ --iam-account=$SA_EMAIL \ --project=$PROJECT_ID @@ -125,6 +135,19 @@ Cloud Logging UI (e.g. `jsonPayload.module = "esp32_blinky::ota"` or `= "esp32_blinky::sig"`). +Metrics land under `custom.googleapis.com/esp32/` (e.g. +`free_heap`, `wifi_rssi`, `stack_hwm` with a `task` label). Snapshot +cadence is `metrics_interval_secs` in the `[gcp]` block (default 300s, +0 disables — cloud_log keeps running). Inspect with: + +```bash +gcloud monitoring time-series list \ + --filter='metric.type=starts_with("custom.googleapis.com/esp32/")' \ + --interval-end-time=$(date -u +%Y-%m-%dT%H:%M:%SZ) \ + --interval-start-time=$(date -u -v-30M +%Y-%m-%dT%H:%M:%SZ) \ + --project=$PROJECT_ID +``` + **Threat model**: the SA private key sits in NVS unencrypted. Anyone with physical access to the chip can extract it. Mitigation is strict scoping (only `roles/logging.logWriter`, only on a diff --git a/provisioning.toml.example b/provisioning.toml.example index 0ec81f0..93a4184 100644 --- a/provisioning.toml.example +++ b/provisioning.toml.example @@ -30,7 +30,8 @@ issuer = "https://token.actions.githubusercontent.com" # Optional. If absent, the device boots with serial-only logging and # never tries to talk to GCP. Uncomment + fill in to enable shipping -# tracing events (app + OTA) to Google Cloud Logging. +# tracing events (app + OTA) to Google Cloud Logging and chip-health +# metrics (heap, stack, wifi, cpu, uptime, …) to Cloud Monitoring. # # [gcp] # project_id = "my-logs-project" @@ -38,3 +39,4 @@ issuer = "https://token.actions.githubusercontent.com" # sa_key_id = "abc123..." # the `private_key_id` field from the SA JSON key # sa_key_pem = "gcp-sa-key.pem" # path to the PKCS#8 RSA private key PEM # min_severity = "info" # trace / debug / info / warn / error +# metrics_interval_secs = 300 # 0 = metrics off (logs still ship); default 300 diff --git a/sdkconfig.defaults.in b/sdkconfig.defaults.in index 9f36f92..50b00d1 100644 --- a/sdkconfig.defaults.in +++ b/sdkconfig.defaults.in @@ -7,6 +7,16 @@ CONFIG_PTHREAD_TASK_STACK_SIZE_DEFAULT=16384 CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=y CONFIG_MBEDTLS_CERTIFICATE_BUNDLE_DEFAULT_FULL=y +# Allow mbedtls to grow its TLS I/O buffers dynamically per-session and +# free them between handshakes, instead of pinning the default 16+16 KB +# in/out buffers per concurrent session. Recovers ~15-25 KB per active +# session — needed because we run three concurrent TLS clients +# (cloud_log, metrics, OTA) and three full-size buffer pairs blow past +# our heap budget (`min_free_heap` watermark hit ~9 KB). +CONFIG_MBEDTLS_DYNAMIC_BUFFER=y +CONFIG_MBEDTLS_DYNAMIC_FREE_CONFIG_DATA=y +CONFIG_MBEDTLS_DYNAMIC_FREE_CA_CERT=y + # OTA: custom partition table with two app slots, plus app-rollback support # so a bad OTA can be reverted by the bootloader on next boot. CONFIG_PARTITION_TABLE_CUSTOM=y diff --git a/src/cloud_log.rs b/src/cloud_log.rs index d74d2fa..844d19b 100644 --- a/src/cloud_log.rs +++ b/src/cloud_log.rs @@ -1,28 +1,19 @@ //! Ship structured logs from the device to Google Cloud Logging. //! -//! Implementation in stages: -//! - This commit: NVS-loaded config, tracing layer that captures events -//! into a bounded ring buffer, background sender task that *would* -//! POST batches but currently just writes them to serial as a stub. -//! - Next commit: real POST via NTP-synced timestamps + service-account -//! JWT auth + the Cloud Logging REST API. +//! - `CloudLogLayer` is a `tracing_subscriber::Layer` that captures +//! events into a bounded ring buffer. +//! - `run()` is a background sender thread that drains the buffer and +//! POSTs batches to `logging.googleapis.com/v2/entries:write` using +//! a Bearer token from `gcp_auth::TokenProvider` (shared with +//! `metrics`). //! //! Cloud logging is opt-in per device. If the `gcp` NVS namespace is //! missing required keys (`project_id`, `sa_email`, `sa_key_id`, //! `sa_key_pem`), the firmware boots normally with serial-only logs. -use anyhow::{anyhow, bail, Context, Result}; -use base64::Engine; -use embedded_svc::http::client::Client; -use embedded_svc::io::Write as _; -use esp_idf_svc::http::client::{Configuration as HttpConfig, EspHttpConnection, FollowRedirectsPolicy}; -use esp_idf_svc::http::Method; +use anyhow::{anyhow, Result}; use esp_idf_svc::nvs::{EspDefaultNvsPartition, EspNvs, NvsDefault}; -use rsa::pkcs1v15::SigningKey; -use rsa::pkcs8::DecodePrivateKey; -use rsa::signature::{SignatureEncoding, Signer}; -use rsa::RsaPrivateKey; -use serde::{Deserialize, Serialize}; +use serde::Serialize; use std::collections::VecDeque; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -30,12 +21,17 @@ use tracing::{Event, Level, Subscriber}; use tracing_subscriber::layer::Context as LayerContext; use tracing_subscriber::Layer; +use crate::gcp_auth::{device_mac, http_post, now_unix_secs, unix_to_rfc3339, TokenProvider}; + const NVS_GCP_NS: &str = "gcp"; const NVS_PROJECT_ID: &str = "project_id"; const NVS_SA_EMAIL: &str = "sa_email"; const NVS_SA_KEY_ID: &str = "sa_key_id"; const NVS_SA_KEY_PEM: &str = "sa_key_pem"; const NVS_MIN_SEVERITY: &str = "min_severity"; +// 15-char NVS key limit forces the abbreviation; the toml field stays +// readable as `metrics_interval_secs`. +const NVS_METRICS_INTERVAL_SECS: &str = "metric_intvl"; /// Capacity of the in-RAM log queue. When the queue is full, oldest /// entries are dropped and counted; the count surfaces as @@ -44,7 +40,11 @@ pub const QUEUE_CAPACITY: usize = 256; const FLUSH_INTERVAL: Duration = Duration::from_secs(5); const BATCH_MAX_ENTRIES: usize = 50; -/// GCP-side config + opt-in flag, read from NVS at boot. +/// Default snapshot cadence if `metrics_interval` is absent in NVS. +pub const DEFAULT_METRICS_INTERVAL_SECS: u32 = 300; + +/// GCP-side config + opt-in flag, read from NVS at boot. Shared by +/// `cloud_log` (logs) and `metrics` (Cloud Monitoring time series). #[derive(Clone)] pub struct GcpConfig { pub project_id: String, @@ -52,6 +52,8 @@ pub struct GcpConfig { pub sa_key_id: String, pub sa_key_pem: Vec, pub min_severity: Level, + /// 0 = metrics disabled (cloud_log still runs); >0 = sample period. + pub metrics_interval_secs: u32, } impl GcpConfig { @@ -86,6 +88,11 @@ impl GcpConfig { sa_key_id: k, sa_key_pem: pem, min_severity: read_severity(&nvs), + metrics_interval_secs: nvs + .get_u32(NVS_METRICS_INTERVAL_SECS) + .ok() + .flatten() + .unwrap_or(DEFAULT_METRICS_INTERVAL_SECS), })), _ => Ok(None), } @@ -153,6 +160,8 @@ struct QueueInner { deque: VecDeque, capacity: usize, pending_dropped: u32, + /// Lifetime drop count (for the metrics path; never reset). + dropped_total: u64, } impl LogQueue { @@ -162,6 +171,7 @@ impl LogQueue { deque: VecDeque::with_capacity(capacity), capacity, pending_dropped: 0, + dropped_total: 0, })), } } @@ -180,6 +190,7 @@ impl LogQueue { if g.deque.len() == g.capacity { g.deque.pop_front(); g.pending_dropped = g.pending_dropped.saturating_add(1); + g.dropped_total = g.dropped_total.saturating_add(1); } g.deque.push_back(entry); } @@ -193,6 +204,14 @@ impl LogQueue { let n = g.deque.len().min(max); g.deque.drain(..n).collect() } + + /// Current queue depth + lifetime drop count, for the metrics path. + pub fn stats(&self) -> (usize, u64) { + match self.inner.lock() { + Ok(g) => (g.deque.len(), g.dropped_total), + Err(_) => (0, 0), + } + } } /// `tracing_subscriber::Layer` that captures events into a `LogQueue`. @@ -209,11 +228,15 @@ impl CloudLogLayer { impl Layer for CloudLogLayer { fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { - // Skip events emitted by this module itself, otherwise the - // sender's "post failed" / "token refreshed" tracing calls - // get pushed onto the queue it's draining, creating a tight - // feedback loop. Cloud_log's own messages stay serial-only. - if event.metadata().target() == module_path!() { + // Skip events emitted by the modules that drive the cloud + // pipeline themselves — otherwise their tracing calls land on + // the queue they're draining (or that the metrics POST path + // uses) and create a tight feedback loop. + let target = event.metadata().target(); + if target == module_path!() + || target == "esp32_blinky::gcp_auth" + || target == "esp32_blinky::metrics" + { return; } let level = *event.metadata().level(); @@ -228,7 +251,7 @@ impl Layer for CloudLogLayer { let entry = LogEntry { timestamp_unix_secs: now_unix_secs(), severity: level, - target: event.metadata().target().to_string(), + target: target.to_string(), message: visitor.message, fields: visitor.fields, dropped_before: 0, @@ -237,20 +260,6 @@ impl Layer for CloudLogLayer { } } -/// Wall-clock time in UNIX seconds, or None if the system clock is -/// still at the ESP-IDF default epoch (1970). NTP sync hasn't happened -/// yet → return None and let Cloud Logging assign server-side time. -fn now_unix_secs() -> Option { - use std::time::{SystemTime, UNIX_EPOCH}; - let secs = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs(); - // Anything before 2020-01-01 means NTP hasn't synced yet. - if secs < 1_577_836_800 { - None - } else { - Some(secs) - } -} - #[derive(Default)] struct FieldCapture { message: String, @@ -306,13 +315,10 @@ impl FieldCapture { } /// Sender thread main loop. Drains the queue every `FLUSH_INTERVAL`, -/// mints/refreshes a service-account access token as needed, and -/// POSTs batches to Cloud Logging. -/// -/// Uses tracing internally — the `module_path!()` filter in -/// `CloudLogLayer::on_event` keeps cloud_log's own messages out of -/// the queue (no feedback loop). -pub fn run(cfg: GcpConfig, queue: LogQueue) -> ! { +/// pulls the bearer from the shared `TokenProvider`, and POSTs batches +/// to Cloud Logging. +pub fn run(cfg: GcpConfig, auth: Arc, queue: LogQueue) -> ! { + crate::metrics::publish_self(&crate::metrics::handles::CLOUD_LOG); tracing::info!( project = %cfg.project_id, sa = %cfg.sa_email, @@ -320,27 +326,10 @@ pub fn run(cfg: GcpConfig, queue: LogQueue) -> ! { "cloud_log: sender starting", ); - // Parse the SA private key once at startup; if it's malformed - // we can't do anything useful and log loudly forever. - let signing_key = match parse_signing_key(&cfg.sa_key_pem) { - Ok(k) => k, - Err(e) => { - loop { - tracing::error!( - error = %format!("{:#}", e), - "cloud_log: SA key parse failed; sender disabled", - ); - std::thread::sleep(Duration::from_secs(300)); - } - } - }; - let mac = device_mac(); let log_name = format!("projects/{}/logs/esp32-firmware", cfg.project_id); - let mut token: Option = None; let mut consecutive_failures: u32 = 0; - loop { let sleep_for = if consecutive_failures > 0 { // Exponential backoff capped at 5 min for cloud-logging @@ -358,29 +347,20 @@ pub fn run(cfg: GcpConfig, queue: LogQueue) -> ! { continue; } - if token.as_ref().map_or(true, |t| t.expired_or_close()) { - match mint_access_token(&cfg, &signing_key) { - Ok(t) => { - tracing::debug!( - expires_in_secs = t.expires_at_unix.saturating_sub(now_unix_secs().unwrap_or(0)), - "cloud_log: minted new access token", - ); - token = Some(t); - } - Err(e) => { - consecutive_failures = consecutive_failures.saturating_add(1); - tracing::warn!( - failures = consecutive_failures, - error = %format!("{:#}", e), - "cloud_log: token mint failed", - ); - continue; - } + let bearer = match auth.get_or_refresh() { + Ok(b) => b, + Err(e) => { + consecutive_failures = consecutive_failures.saturating_add(1); + tracing::warn!( + failures = consecutive_failures, + error = %format!("{:#}", e), + "cloud_log: token mint failed", + ); + continue; } - } + }; - let bearer = &token.as_ref().unwrap().token; - match post_batch(&log_name, &cfg.project_id, &mac, bearer, &batch) { + match post_batch(&log_name, &cfg.project_id, &mac, &bearer, &batch) { Ok(()) => { tracing::debug!( entries = batch.len(), @@ -405,101 +385,6 @@ pub fn run(cfg: GcpConfig, queue: LogQueue) -> ! { } } -struct CachedToken { - token: String, - expires_at_unix: u64, -} - -impl CachedToken { - fn expired_or_close(&self) -> bool { - // Refresh 5 minutes before expiry to avoid races. - match now_unix_secs() { - Some(now) => now + 300 >= self.expires_at_unix, - None => true, - } - } -} - -fn parse_signing_key(pem_bytes: &[u8]) -> Result> { - // Trim trailing whitespace — jq -r adds a final newline on top of - // the PEM's own trailing newline, and pem-rfc7468 then tries to - // parse a second (empty) block and errors at "pre-encapsulation - // boundary". Tolerate both forms. - let pem_str = std::str::from_utf8(pem_bytes) - .context("SA key PEM is not UTF-8")? - .trim(); - let key = RsaPrivateKey::from_pkcs8_pem(pem_str).context("parse SA PKCS#8 private key")?; - Ok(SigningKey::::new(key)) -} - -#[derive(Serialize)] -struct JwtHeader<'a> { - alg: &'static str, - typ: &'static str, - kid: &'a str, -} - -#[derive(Serialize)] -struct JwtClaims<'a> { - iss: &'a str, - scope: &'static str, - aud: &'static str, - iat: u64, - exp: u64, -} - -#[derive(Deserialize)] -struct TokenResp { - access_token: String, - expires_in: u64, -} - -fn mint_access_token( - cfg: &GcpConfig, - signing_key: &SigningKey, -) -> Result { - let now = now_unix_secs().ok_or_else(|| anyhow!("NTP not synced; cannot mint JWT"))?; - - let header = JwtHeader { - alg: "RS256", - typ: "JWT", - kid: &cfg.sa_key_id, - }; - let claims = JwtClaims { - iss: &cfg.sa_email, - scope: "https://www.googleapis.com/auth/logging.write", - aud: "https://oauth2.googleapis.com/token", - iat: now, - exp: now + 3600, - }; - - let header_b64 = b64url(&serde_json::to_vec(&header)?); - let claims_b64 = b64url(&serde_json::to_vec(&claims)?); - let signing_input = format!("{}.{}", header_b64, claims_b64); - let sig = signing_key.sign(signing_input.as_bytes()); - let sig_b64 = b64url(sig.to_bytes().as_ref()); - let jwt = format!("{}.{}", signing_input, sig_b64); - - let body = format!( - "grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer&assertion={}", - jwt - ); - let resp_bytes = http_post( - "https://oauth2.googleapis.com/token", - "application/x-www-form-urlencoded", - body.as_bytes(), - None, - ) - .context("POST oauth2/token")?; - let resp: TokenResp = serde_json::from_slice(&resp_bytes) - .context("parse token response JSON")?; - - Ok(CachedToken { - token: resp.access_token, - expires_at_unix: now + resp.expires_in, - }) -} - #[derive(Serialize)] struct WriteEntriesRequest<'a> { #[serde(rename = "logName")] @@ -547,7 +432,6 @@ fn post_batch( timestamp: e.timestamp_unix_secs.and_then(unix_to_rfc3339), }) .collect(); - let _ = (project_id, mac); // used inline in MonitoredResource below let req = WriteEntriesRequest { log_name, @@ -603,82 +487,3 @@ fn build_payload(e: &LogEntry) -> serde_json::Value { } serde_json::Value::Object(map) } - -fn unix_to_rfc3339(secs: u64) -> Option { - use time::format_description::well_known::Rfc3339; - use time::OffsetDateTime; - OffsetDateTime::from_unix_timestamp(secs as i64) - .ok() - .and_then(|dt| dt.format(&Rfc3339).ok()) -} - -fn b64url(input: &[u8]) -> String { - base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(input) -} - -/// Single helper for HTTPS POST. Returns the response body bytes. -/// Errors on any non-2xx status with the body included for diagnosis. -fn http_post( - url: &str, - content_type: &str, - body: &[u8], - bearer: Option<&str>, -) -> Result> { - let conn = EspHttpConnection::new(&HttpConfig { - crt_bundle_attach: Some(esp_idf_svc::sys::esp_crt_bundle_attach), - follow_redirects_policy: FollowRedirectsPolicy::FollowAll, - timeout: Some(Duration::from_secs(30)), - buffer_size: Some(2048), - buffer_size_tx: Some(4096), - ..Default::default() - })?; - let mut client = Client::wrap(conn); - let body_len = body.len().to_string(); - let mut headers: Vec<(&str, &str)> = vec![ - ("content-type", content_type), - ("content-length", body_len.as_str()), - ("accept", "application/json"), - ]; - if let Some(b) = bearer { - headers.push(("authorization", b)); - } - let mut req = client.request(Method::Post, url, &headers)?; - req.write_all(body).context("write request body")?; - req.flush().ok(); - let mut resp = req.submit()?; - let status = resp.status(); - let mut buf = Vec::with_capacity(1024); - let mut chunk = [0u8; 1024]; - loop { - let n = resp.read(&mut chunk).context("read response chunk")?; - if n == 0 { - break; - } - buf.extend_from_slice(&chunk[..n]); - } - if !(200..300).contains(&status) { - bail!( - "POST {} -> HTTP {} body={}", - url, - status, - String::from_utf8_lossy(&buf) - ); - } - Ok(buf) -} - -/// MAC address as a `aabbccddeeff` hex string. Used as the `node_id` -/// label on Cloud Logging entries to identify which device the log -/// came from. -fn device_mac() -> String { - let mut mac = [0u8; 8]; - unsafe { - esp_idf_svc::sys::esp_efuse_mac_get_default(mac.as_mut_ptr()); - } - let mut s = String::with_capacity(12); - for b in &mac[..6] { - use std::fmt::Write as _; - let _ = write!(s, "{:02x}", b); - } - s -} diff --git a/src/gcp_auth.rs b/src/gcp_auth.rs new file mode 100644 index 0000000..356db50 --- /dev/null +++ b/src/gcp_auth.rs @@ -0,0 +1,269 @@ +//! Google Cloud auth shared between cloud_log + metrics. +//! +//! One service account, one private key, one cached access token, +//! used by both the logging POST path (`cloud_log`) and the +//! monitoring POST path (`metrics`). The JWT requests both +//! `logging.write` and `monitoring.write` scopes so a single +//! exchange covers both APIs. +//! +//! Also exposes the HTTPS / time / base64url helpers both modules +//! need so we don't duplicate them. + +use anyhow::{anyhow, bail, Context, Result}; +use base64::Engine; +use embedded_svc::http::client::Client; +use embedded_svc::io::Write as _; +use esp_idf_svc::http::client::{ + Configuration as HttpConfig, EspHttpConnection, FollowRedirectsPolicy, +}; +use esp_idf_svc::http::Method; +use rsa::pkcs1v15::SigningKey; +use rsa::pkcs8::DecodePrivateKey; +use rsa::signature::{SignatureEncoding, Signer}; +use rsa::RsaPrivateKey; +use serde::{Deserialize, Serialize}; +use std::sync::Mutex; +use std::time::Duration; + +use crate::cloud_log::GcpConfig; + +/// A bearer token cached until ~5 min before expiry. +struct CachedToken { + token: String, + expires_at_unix: u64, +} + +impl CachedToken { + fn fresh(&self) -> bool { + match now_unix_secs() { + // Refresh 5 minutes before expiry to avoid races. + Some(now) => now + 300 < self.expires_at_unix, + None => false, + } + } +} + +/// Mints + caches a single OAuth2 access token shareable across +/// cloud_log and metrics. Multi-scope so one token covers both APIs. +pub struct TokenProvider { + cfg: GcpConfig, + signing_key: SigningKey, + cache: Mutex>, +} + +impl TokenProvider { + pub fn new(cfg: GcpConfig) -> Result { + let signing_key = parse_signing_key(&cfg.sa_key_pem)?; + Ok(Self { + cfg, + signing_key, + cache: Mutex::new(None), + }) + } + + /// Returns a Bearer token. Mints a new one (RSA sign + HTTPS POST + /// to oauth2.googleapis.com) only when the cached one is missing + /// or near expiry. Holding the lock through mint serialises + /// concurrent callers, but mint runs ~hourly so the contention + /// window is fine. + pub fn get_or_refresh(&self) -> Result { + let mut g = self + .cache + .lock() + .map_err(|_| anyhow!("token cache mutex poisoned"))?; + if let Some(t) = g.as_ref() { + if t.fresh() { + return Ok(t.token.clone()); + } + } + let new = mint_access_token(&self.cfg, &self.signing_key) + .context("mint GCP access token")?; + let bearer = new.token.clone(); + let exp_in = new + .expires_at_unix + .saturating_sub(now_unix_secs().unwrap_or(0)); + tracing::debug!( + expires_in_secs = exp_in, + "gcp_auth: minted new access token", + ); + *g = Some(new); + Ok(bearer) + } +} + +fn parse_signing_key(pem_bytes: &[u8]) -> Result> { + // Trim trailing whitespace — jq -r adds a final newline on top of + // the PEM's own trailing newline, and pem-rfc7468 then tries to + // parse a second (empty) block and errors at "pre-encapsulation + // boundary". Tolerate both forms. + let pem_str = std::str::from_utf8(pem_bytes) + .context("SA key PEM is not UTF-8")? + .trim(); + let key = + RsaPrivateKey::from_pkcs8_pem(pem_str).context("parse SA PKCS#8 private key")?; + Ok(SigningKey::::new(key)) +} + +#[derive(Serialize)] +struct JwtHeader<'a> { + alg: &'static str, + typ: &'static str, + kid: &'a str, +} + +#[derive(Serialize)] +struct JwtClaims<'a> { + iss: &'a str, + scope: &'static str, + aud: &'static str, + iat: u64, + exp: u64, +} + +#[derive(Deserialize)] +struct TokenResp { + access_token: String, + expires_in: u64, +} + +/// Multi-scope so the resulting access token works against both +/// `logging.googleapis.com` and `monitoring.googleapis.com`. +const SCOPES: &str = + "https://www.googleapis.com/auth/logging.write https://www.googleapis.com/auth/monitoring.write"; + +fn mint_access_token( + cfg: &GcpConfig, + signing_key: &SigningKey, +) -> Result { + let now = now_unix_secs().ok_or_else(|| anyhow!("NTP not synced; cannot mint JWT"))?; + + let header = JwtHeader { + alg: "RS256", + typ: "JWT", + kid: &cfg.sa_key_id, + }; + let claims = JwtClaims { + iss: &cfg.sa_email, + scope: SCOPES, + aud: "https://oauth2.googleapis.com/token", + iat: now, + exp: now + 3600, + }; + + let header_b64 = b64url(&serde_json::to_vec(&header)?); + let claims_b64 = b64url(&serde_json::to_vec(&claims)?); + let signing_input = format!("{}.{}", header_b64, claims_b64); + let sig = signing_key.sign(signing_input.as_bytes()); + let sig_b64 = b64url(sig.to_bytes().as_ref()); + let jwt = format!("{}.{}", signing_input, sig_b64); + + let body = format!( + "grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer&assertion={}", + jwt + ); + let resp_bytes = http_post( + "https://oauth2.googleapis.com/token", + "application/x-www-form-urlencoded", + body.as_bytes(), + None, + ) + .context("POST oauth2/token")?; + let resp: TokenResp = + serde_json::from_slice(&resp_bytes).context("parse token response JSON")?; + + Ok(CachedToken { + token: resp.access_token, + expires_at_unix: now + resp.expires_in, + }) +} + +/// Wall-clock time in UNIX seconds, or None if the system clock is +/// still at the ESP-IDF default epoch (anything before 2020-01-01 +/// counts as not-yet-synced). +pub fn now_unix_secs() -> Option { + use std::time::{SystemTime, UNIX_EPOCH}; + let secs = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs(); + if secs < 1_577_836_800 { + None + } else { + Some(secs) + } +} + +pub fn unix_to_rfc3339(secs: u64) -> Option { + use time::format_description::well_known::Rfc3339; + use time::OffsetDateTime; + OffsetDateTime::from_unix_timestamp(secs as i64) + .ok() + .and_then(|dt| dt.format(&Rfc3339).ok()) +} + +pub fn b64url(input: &[u8]) -> String { + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(input) +} + +/// HTTPS POST helper. Returns the response body bytes. Errors on any +/// non-2xx with the body included for diagnosis. +pub fn http_post( + url: &str, + content_type: &str, + body: &[u8], + bearer: Option<&str>, +) -> Result> { + let conn = EspHttpConnection::new(&HttpConfig { + crt_bundle_attach: Some(esp_idf_svc::sys::esp_crt_bundle_attach), + follow_redirects_policy: FollowRedirectsPolicy::FollowAll, + timeout: Some(Duration::from_secs(30)), + buffer_size: Some(2048), + buffer_size_tx: Some(4096), + ..Default::default() + })?; + let mut client = Client::wrap(conn); + let body_len = body.len().to_string(); + let mut headers: Vec<(&str, &str)> = vec![ + ("content-type", content_type), + ("content-length", body_len.as_str()), + ("accept", "application/json"), + ]; + if let Some(b) = bearer { + headers.push(("authorization", b)); + } + let mut req = client.request(Method::Post, url, &headers)?; + req.write_all(body).context("write request body")?; + req.flush().ok(); + let mut resp = req.submit()?; + let status = resp.status(); + let mut buf = Vec::with_capacity(1024); + let mut chunk = [0u8; 1024]; + loop { + let n = resp.read(&mut chunk).context("read response chunk")?; + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + if !(200..300).contains(&status) { + bail!( + "POST {} -> HTTP {} body={}", + url, + status, + String::from_utf8_lossy(&buf) + ); + } + Ok(buf) +} + +/// MAC address as a `aabbccddeeff` hex string. Used as the `node_id` +/// label on Cloud Logging entries + Cloud Monitoring resource labels. +pub fn device_mac() -> String { + let mut mac = [0u8; 8]; + unsafe { + esp_idf_svc::sys::esp_efuse_mac_get_default(mac.as_mut_ptr()); + } + let mut s = String::with_capacity(12); + for b in &mac[..6] { + use std::fmt::Write as _; + let _ = write!(s, "{:02x}", b); + } + s +} diff --git a/src/main.rs b/src/main.rs index 3959ff5..7a16fec 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use anyhow::{anyhow, Result}; +use embedded_svc::http::client::Client; use esp_idf_svc::eventloop::EspSystemEventLoop; use esp_idf_svc::hal::peripherals::Peripherals; -use embedded_svc::http::client::Client; use esp_idf_svc::http::client::{Configuration as HttpConfig, EspHttpConnection}; use esp_idf_svc::http::Method; use esp_idf_svc::nvs::{EspDefaultNvsPartition, EspNvs}; @@ -12,6 +12,8 @@ use std::ffi::CStr; use std::time::Duration; mod cloud_log; +mod gcp_auth; +mod metrics; mod ota; mod sig; mod trust; @@ -29,6 +31,7 @@ const FW_VERSION: &str = env!("GIT_SHA"); fn main() -> Result<()> { esp_idf_svc::sys::link_patches(); esp_idf_svc::log::EspLogger::initialize_default(); + metrics::publish_self(&metrics::handles::MAIN); // Take NVS as early as possible — needed to decide whether to // install the cloud-log tracing subscriber, *before* any tracing @@ -69,9 +72,7 @@ fn main() -> Result<()> { }; let trust = match trust::TrustConfig::load(nvs.clone())? { Some(t) => t, - None => block_unprovisioned( - "trust/identities or trust/fulcio_{root,inter} missing in NVS", - ), + None => block_unprovisioned("trust/identities or trust/fulcio_{root,inter} missing in NVS"), }; tracing::info!( identities = trust.identities.len(), @@ -148,11 +149,44 @@ fn main() -> Result<()> { .expect("spawn ota thread"); if let (Some(cfg), Some(queue)) = (gcp, log_queue) { - std::thread::Builder::new() - // RSA signing + HTTPS POST. 32 KB matches the OTA loop budget. - .stack_size(32 * 1024) - .spawn(move || cloud_log::run(cfg, queue)) - .expect("spawn cloud_log thread"); + // Build the shared TokenProvider once — both cloud_log and + // metrics get an Arc clone and share the cached access token + // (multi-scope JWT covers both APIs). Parsing the SA key at + // construction time means a malformed key fails fast here + // rather than once per minute in the sender thread. + let auth = match gcp_auth::TokenProvider::new(cfg.clone()) { + Ok(a) => Some(std::sync::Arc::new(a)), + Err(e) => { + tracing::error!( + error = %format!("{:#}", e), + "gcp_auth: TokenProvider init failed; cloud_log + metrics disabled", + ); + None + } + }; + + if let Some(auth) = auth { + let cl_cfg = cfg.clone(); + let cl_auth = auth.clone(); + let cl_queue = queue.clone(); + std::thread::Builder::new() + // RSA signing + HTTPS POST. 32 KB matches the OTA loop budget. + .stack_size(32 * 1024) + .spawn(move || cloud_log::run(cl_cfg, cl_auth, cl_queue)) + .expect("spawn cloud_log thread"); + + if cfg.metrics_interval_secs > 0 { + let m_cfg = cfg; + let m_auth = auth; + let m_queue = queue; + std::thread::Builder::new() + // No crypto on hot path (token cached via auth). + // Mainly HTTPS POST + serde_json. 16 KB sufficient. + .stack_size(16 * 1024) + .spawn(move || metrics::run(m_cfg, m_auth, m_queue)) + .expect("spawn metrics thread"); + } + } } tracing::info!("main: idling, OTA loop running in background"); @@ -186,10 +220,14 @@ fn connect_wifi(wifi: &mut BlockingWifi>, ssid: &str, pass: &st }; wifi.set_configuration(&WifiConfig::Client(ClientConfiguration { - ssid: ssid.try_into().map_err(|_| anyhow!("SSID too long (max 32 bytes)"))?, + ssid: ssid + .try_into() + .map_err(|_| anyhow!("SSID too long (max 32 bytes)"))?, bssid: None, auth_method, - password: pass.try_into().map_err(|_| anyhow!("password too long (max 64 bytes)"))?, + password: pass + .try_into() + .map_err(|_| anyhow!("password too long (max 64 bytes)"))?, channel: None, ..Default::default() }))?; @@ -232,7 +270,6 @@ fn block_unprovisioned(reason: &str) -> ! { } } - fn fetch(url: &str) -> Result<()> { let conn = EspHttpConnection::new(&HttpConfig { crt_bundle_attach: Some(esp_idf_svc::sys::esp_crt_bundle_attach), @@ -255,7 +292,7 @@ fn fetch(url: &str) -> Result<()> { total += n; body.extend_from_slice(&buf[..n]); } - tracing::info!( + tracing::debug!( url = url, status = status, bytes = total, diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 0000000..89f8b32 --- /dev/null +++ b/src/metrics.rs @@ -0,0 +1,347 @@ +//! Periodic device-health snapshots → Cloud Monitoring time series. +//! +//! Wakes every `metrics_interval_secs`, collects a chip-wide snapshot +//! (heap, stack hwm per task, wifi RSSI/channel, cpu freq, uptime, +//! NVS stats, log queue depth + drops), and POSTs one +//! `CreateTimeSeries` request to +//! `https://monitoring.googleapis.com/v3/projects//timeSeries`. +//! +//! Cloud Monitoring auto-creates a `MetricDescriptor` on first write +//! per metric type, so there's no separate provisioning step. All +//! metrics are GAUGE INT64 in v1. + +use anyhow::{Context, Result}; +use serde::Serialize; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use crate::cloud_log::{GcpConfig, LogQueue}; +use crate::gcp_auth::{device_mac, http_post, unix_to_rfc3339, TokenProvider}; + +const METRIC_PREFIX: &str = "custom.googleapis.com/esp32"; + +/// FreeRTOS task handles published by each spawned thread so the +/// metrics loop can call `uxTaskGetStackHighWaterMark` per task. +/// Stored as `usize` (not `*mut c_void`) so we can use atomics; 0 +/// means "not yet published, omit this field from the snapshot". +pub mod handles { + use std::sync::atomic::AtomicUsize; + pub static MAIN: AtomicUsize = AtomicUsize::new(0); + pub static OTA: AtomicUsize = AtomicUsize::new(0); + pub static CLOUD_LOG: AtomicUsize = AtomicUsize::new(0); + pub static METRICS: AtomicUsize = AtomicUsize::new(0); +} + +/// Each thread calls this once at start of its run loop. +pub fn publish_self(slot: &AtomicUsize) { + let h = unsafe { esp_idf_svc::sys::xTaskGetCurrentTaskHandle() } as usize; + slot.store(h, Ordering::Relaxed); +} + +fn read_handle(slot: &AtomicUsize) -> Option { + let v = slot.load(Ordering::Relaxed); + if v == 0 { + None + } else { + Some(v as esp_idf_svc::sys::TaskHandle_t) + } +} + +pub fn run(cfg: GcpConfig, auth: Arc, queue: LogQueue) -> ! { + publish_self(&handles::METRICS); + + tracing::info!( + project = %cfg.project_id, + interval_secs = cfg.metrics_interval_secs, + "metrics: sender starting", + ); + + let mac = device_mac(); + let url = format!( + "https://monitoring.googleapis.com/v3/projects/{}/timeSeries", + cfg.project_id + ); + let interval = Duration::from_secs(cfg.metrics_interval_secs.max(1) as u64); + let mut consecutive_failures: u32 = 0; + + loop { + // Backoff on failure, otherwise sleep the configured interval. + let sleep_for = if consecutive_failures > 0 { + let exp = consecutive_failures.min(4); + interval.saturating_mul(1 << exp).min(Duration::from_secs(3600)) + } else { + interval + }; + std::thread::sleep(sleep_for); + + let snapshot = collect(&queue); + let bearer = match auth.get_or_refresh() { + Ok(b) => b, + Err(e) => { + consecutive_failures = consecutive_failures.saturating_add(1); + tracing::warn!( + failures = consecutive_failures, + error = %format!("{:#}", e), + "metrics: token mint failed", + ); + continue; + } + }; + + match post_time_series(&url, &cfg.project_id, &mac, &bearer, &snapshot) { + Ok(()) => { + tracing::debug!( + series = snapshot.series_count(), + "metrics: posted", + ); + consecutive_failures = 0; + } + Err(e) => { + consecutive_failures = consecutive_failures.saturating_add(1); + tracing::warn!( + failures = consecutive_failures, + error = %format!("{:#}", e), + "metrics: post failed", + ); + } + } + } +} + +#[derive(Default)] +struct Snapshot { + // Memory + free_heap: Option, + free_heap_internal: Option, + min_free_heap: Option, + largest_free_block: Option, + // Per-task stack high-water-mark (bytes remaining), task name → value + stack_hwm: Vec<(&'static str, i64)>, + // Wifi + wifi_rssi: Option, + wifi_channel: Option, + // CPU + cpu_freq_mhz: Option, + // Boot / uptime + uptime_secs: Option, + // NVS + nvs_used_entries: Option, + nvs_free_entries: Option, + // Cloud_log queue + cloud_log_queue_depth: Option, + cloud_log_dropped_total: Option, +} + +impl Snapshot { + fn series_count(&self) -> usize { + let scalars = [ + self.free_heap.is_some(), + self.free_heap_internal.is_some(), + self.min_free_heap.is_some(), + self.largest_free_block.is_some(), + self.wifi_rssi.is_some(), + self.wifi_channel.is_some(), + self.cpu_freq_mhz.is_some(), + self.uptime_secs.is_some(), + self.nvs_used_entries.is_some(), + self.nvs_free_entries.is_some(), + self.cloud_log_queue_depth.is_some(), + self.cloud_log_dropped_total.is_some(), + ]; + scalars.iter().filter(|b| **b).count() + self.stack_hwm.len() + } +} + +fn collect(queue: &LogQueue) -> Snapshot { + let mut s = Snapshot::default(); + + // Memory + unsafe { + s.free_heap = Some(esp_idf_svc::sys::esp_get_free_heap_size() as i64); + s.min_free_heap = + Some(esp_idf_svc::sys::esp_get_minimum_free_heap_size() as i64); + let largest = esp_idf_svc::sys::heap_caps_get_largest_free_block( + esp_idf_svc::sys::MALLOC_CAP_DEFAULT, + ); + s.largest_free_block = Some(largest as i64); + let internal = esp_idf_svc::sys::heap_caps_get_free_size( + esp_idf_svc::sys::MALLOC_CAP_INTERNAL, + ); + s.free_heap_internal = Some(internal as i64); + } + + // Stack high-water-mark per published task. ESP-IDF's + // `uxTaskGetStackHighWaterMark` returns the value in StackType_t + // units; on Xtensa StackType_t is uint8_t, so the return is bytes + // remaining at low-water. (Generic FreeRTOS docs say "words"; the + // ESP-IDF port differs.) + for (name, slot) in [ + ("main", &handles::MAIN), + ("ota", &handles::OTA), + ("cloud_log", &handles::CLOUD_LOG), + ("metrics", &handles::METRICS), + ] { + if let Some(h) = read_handle(slot) { + let bytes = unsafe { esp_idf_svc::sys::uxTaskGetStackHighWaterMark(h) }; + s.stack_hwm.push((name, bytes as i64)); + } + } + + // Wifi (rssi + channel, only meaningful when associated) + let mut ap_info: esp_idf_svc::sys::wifi_ap_record_t = unsafe { core::mem::zeroed() }; + let err = unsafe { esp_idf_svc::sys::esp_wifi_sta_get_ap_info(&mut ap_info) }; + if err == esp_idf_svc::sys::ESP_OK { + s.wifi_rssi = Some(ap_info.rssi as i64); + s.wifi_channel = Some(ap_info.primary as i64); + } + + // CPU clock. The runtime accessor `esp_clk_cpu_freq` isn't exposed + // through esp-idf-svc's bindings, so report the build-time + // configured default. Accurate while CONFIG_PM_ENABLE=n (our + // setup); when we adopt esp_pm_* this needs to use a runtime API. + s.cpu_freq_mhz = Some(esp_idf_svc::sys::CONFIG_ESP_DEFAULT_CPU_FREQ_MHZ as i64); + + // Uptime (esp_timer_get_time returns microseconds since boot) + unsafe { + let micros = esp_idf_svc::sys::esp_timer_get_time(); + s.uptime_secs = Some(micros / 1_000_000); + } + + // NVS stats (default partition; partition_name = NULL) + let mut nvs_stats: esp_idf_svc::sys::nvs_stats_t = unsafe { core::mem::zeroed() }; + let err = unsafe { + esp_idf_svc::sys::nvs_get_stats(core::ptr::null(), &mut nvs_stats) + }; + if err == esp_idf_svc::sys::ESP_OK { + s.nvs_used_entries = Some(nvs_stats.used_entries as i64); + s.nvs_free_entries = Some(nvs_stats.free_entries as i64); + } + + // Log queue stats + let (depth, dropped) = queue.stats(); + s.cloud_log_queue_depth = Some(depth as i64); + s.cloud_log_dropped_total = Some(dropped as i64); + + s +} + +#[derive(Serialize)] +struct CreateTimeSeriesRequest { + #[serde(rename = "timeSeries")] + time_series: Vec, +} + +#[derive(Serialize)] +struct TimeSeries { + metric: Metric, + resource: MonitoredResource, + #[serde(rename = "metricKind")] + metric_kind: &'static str, + #[serde(rename = "valueType")] + value_type: &'static str, + points: Vec, +} + +#[derive(Serialize)] +struct Metric { + #[serde(rename = "type")] + type_: String, + #[serde(skip_serializing_if = "serde_json::Map::is_empty")] + labels: serde_json::Map, +} + +#[derive(Serialize, Clone)] +struct MonitoredResource { + #[serde(rename = "type")] + type_: &'static str, + labels: serde_json::Map, +} + +#[derive(Serialize)] +struct Point { + interval: Interval, + value: PointValue, +} + +#[derive(Serialize)] +struct Interval { + #[serde(rename = "endTime")] + end_time: String, +} + +#[derive(Serialize)] +struct PointValue { + #[serde(rename = "int64Value")] + int64_value: String, +} + +fn post_time_series( + url: &str, + project_id: &str, + mac: &str, + bearer: &str, + snapshot: &Snapshot, +) -> Result<()> { + let now_secs = crate::gcp_auth::now_unix_secs() + .ok_or_else(|| anyhow::anyhow!("NTP not synced; cannot stamp metric points"))?; + let end_time = unix_to_rfc3339(now_secs) + .ok_or_else(|| anyhow::anyhow!("RFC3339 format failed"))?; + + let resource = MonitoredResource { + type_: "generic_node", + labels: { + let mut m = serde_json::Map::new(); + m.insert("project_id".into(), project_id.into()); + m.insert("location".into(), "global".into()); + m.insert("namespace".into(), "esp32".into()); + m.insert("node_id".into(), mac.into()); + m + }, + }; + + let mut series = Vec::with_capacity(snapshot.series_count()); + let mut push = |name: &str, labels: serde_json::Map, value: i64| { + series.push(TimeSeries { + metric: Metric { + type_: format!("{}/{}", METRIC_PREFIX, name), + labels, + }, + resource: resource.clone(), + metric_kind: "GAUGE", + value_type: "INT64", + points: vec![Point { + interval: Interval { + end_time: end_time.clone(), + }, + value: PointValue { + int64_value: value.to_string(), + }, + }], + }); + }; + + let no_labels = serde_json::Map::new(); + if let Some(v) = snapshot.free_heap { push("free_heap", no_labels.clone(), v); } + if let Some(v) = snapshot.free_heap_internal { push("free_heap_internal", no_labels.clone(), v); } + if let Some(v) = snapshot.min_free_heap { push("min_free_heap", no_labels.clone(), v); } + if let Some(v) = snapshot.largest_free_block { push("largest_free_block", no_labels.clone(), v); } + if let Some(v) = snapshot.wifi_rssi { push("wifi_rssi", no_labels.clone(), v); } + if let Some(v) = snapshot.wifi_channel { push("wifi_channel", no_labels.clone(), v); } + if let Some(v) = snapshot.cpu_freq_mhz { push("cpu_freq_mhz", no_labels.clone(), v); } + if let Some(v) = snapshot.uptime_secs { push("uptime_secs", no_labels.clone(), v); } + if let Some(v) = snapshot.nvs_used_entries { push("nvs_used_entries", no_labels.clone(), v); } + if let Some(v) = snapshot.nvs_free_entries { push("nvs_free_entries", no_labels.clone(), v); } + if let Some(v) = snapshot.cloud_log_queue_depth { push("cloud_log_queue_depth", no_labels.clone(), v); } + if let Some(v) = snapshot.cloud_log_dropped_total { push("cloud_log_dropped_total", no_labels.clone(), v); } + for (task, value) in &snapshot.stack_hwm { + let mut labels = serde_json::Map::new(); + labels.insert("task".into(), serde_json::Value::String((*task).into())); + push("stack_hwm", labels, *value); + } + + let req = CreateTimeSeriesRequest { time_series: series }; + let body = serde_json::to_vec(&req).context("serialize CreateTimeSeries body")?; + let auth = format!("Bearer {}", bearer); + http_post(url, "application/json", &body, Some(&auth)).map(|_| ()) +} diff --git a/src/ota.rs b/src/ota.rs index e92bdcb..54b12db 100644 --- a/src/ota.rs +++ b/src/ota.rs @@ -91,6 +91,7 @@ pub fn run( fw_version: &str, trust: crate::trust::TrustConfig, ) -> ! { + crate::metrics::publish_self(&crate::metrics::handles::OTA); let mut nvs = match EspNvs::new(nvs_partition, NVS_NAMESPACE, true) { Ok(n) => n, Err(e) => { diff --git a/tools/provision/src/main.rs b/tools/provision/src/main.rs index 45f67fc..eab144f 100644 --- a/tools/provision/src/main.rs +++ b/tools/provision/src/main.rs @@ -91,12 +91,20 @@ struct GcpConfig { /// Logging (everything still goes to serial). #[serde(default = "default_severity")] min_severity: String, + /// Seconds between Cloud Monitoring metric snapshots. Default 300 + /// (5 min). 0 = metrics disabled (cloud_log still runs). + #[serde(default = "default_metrics_interval")] + metrics_interval_secs: u32, } fn default_severity() -> String { "info".to_string() } +fn default_metrics_interval() -> u32 { + 300 +} + fn severity_to_u8(s: &str) -> Result { match s.to_ascii_lowercase().as_str() { "trace" => Ok(0), @@ -303,6 +311,12 @@ fn write_csv( ])?; let severity = severity_to_u8(&gcp.min_severity)?; wtr.write_record(&["min_severity", "data", "u8", &severity.to_string()])?; + wtr.write_record(&[ + "metric_intvl", + "data", + "u32", + &gcp.metrics_interval_secs.to_string(), + ])?; } wtr.flush()?;