fix: cleanup metrics and observability setup and communication with the TUI

.github/workflows/nightly-tool-integrations.yml

@@ -101,7 +101,7 @@ jobs:
       AMARU_LISTEN_ADDRESS: 127.0.0.1:3000
       AMARU_PEER_ADDRESS: preprod-node.play.dev.cardano.org:3001
       AMARU_WITH_OPEN_TELEMETRY: ${{ matrix.tool.metrics || 'false' }}
-      OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318"
+      OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4317"
       TOOL_TEST_TIMEOUT: ${{ matrix.tool.timeout || 120 }}
     strategy:
       fail-fast: false

.github/workflows/template-ledger-epoch-snapshots.yml

@@ -57,7 +57,6 @@ jobs:
       TRACE_COMPARE_LOG: trace-compare.log
       AMARU_WITH_OPEN_TELEMETRY: "true"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4317"
-      OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: "http://localhost:4318/v1/metrics"
       OTEL_METRIC_EXPORT_INTERVAL: 1000
       RUST_BACKTRACE: 1
 

CHANGELOG.md

@@ -35,7 +35,7 @@
   ```
 -->
 
 ## v10.11.20260806 _[unreleased; planned for 2026-08-06]_
 
 ### Added
 
@@ -53,6 +53,7 @@
 - **amaru-ledger**: keep only slim stake summaries in runtime memory, and rebuild the full account-heavy stake distribution from snapshots when computing rewards.
 - **amaru-ledger**: compute rewards and stake distributions asynchronously to prevent blocking the main roll forward loop from times to times.
 - **amaru**: bootstrap snapshots now are retrieved directly from R2 (no embedded manifests) and compressed with zstandard. ([#1012][])
+- **amaru**: metrics are now (also) exported through gRPC on `:4317` by default instead of `:4318` over HTTP.
 
 ### Removed
 

Cargo.toml

@@ -78,6 +78,7 @@ once_cell = "1.21.4"
 opentelemetry = "0.32.0"
 opentelemetry-appender-tracing = "0.32.0"
 opentelemetry-otlp = { version = "0.32.0", features = [ "grpc-tonic", "http-proto", "logs", "reqwest-blocking-client", ] }
+opentelemetry-proto = { version = "0.32.0", default-features = false, features = [ "gen-tonic", "trace", "logs", "metrics" ] }
 opentelemetry-semantic-conventions = { version = "0.32.1", features = ["semconv_experimental"] }
 opentelemetry_sdk = { version = "0.32.1", features = ["logs"] }
 ouroboros = "0.18.5"
@@ -111,6 +112,7 @@ thiserror = "2.0.19"
 tokio = { version = "1.53.0", features = ["sync"] }
 tokio-util = "0.7.18"
 toml = "1.1.3"
+tonic = "0.14.6"
 tracing = { version = "0.1.40", features = ["valuable"] }
 tracing-opentelemetry = "0.33.0"
 tracing-subscriber = { version = "0.3.23", features = [ "env-filter", "std", "json", ] }

crates/amaru-kernel/src/cardano/era_history.rs

@@ -377,32 +377,6 @@ pub enum EraHistoryFileError {
     JsonParseError(#[from] serde_json::Error),
 }
 
-/// Load an `EraHistory` from a JSON file.
-///
-/// # Arguments
-///
-/// * `path` - Path to the JSON file containing era history data
-///
-/// # Returns
-///
-/// Returns a Result containing the `EraHistory` if successful, or an `EraHistoryFileError` if the file
-/// cannot be read or parsed.
-///
-/// # Example
-///
-/// ```no_run
-/// use amaru_kernel::load_era_history_from_file;
-/// use std::path::Path;
-///
-/// let era_history = load_era_history_from_file(Path::new("era_history.json")).unwrap();
-/// ```
-pub fn load_era_history_from_file(path: &Path) -> Result<EraHistory, EraHistoryFileError> {
-    let file = File::open(path).map_err(EraHistoryFileError::FileOpenError)?;
-    let reader = BufReader::new(file);
-
-    serde_json::from_reader(reader).map_err(EraHistoryFileError::JsonParseError)
-}
-
 impl<C> cbor::Encode<C> for EraHistory {
     fn encode<W: cbor::encode::Write>(
         &self,
@@ -451,6 +425,32 @@ pub struct EpochEraBounds {
 // horizon is the end of the epoch containing the end of the current era's safe zone relative to
 // the current tip. Returns number of milliseconds elapsed since the system start time.
 impl EraHistory {
+    /// Load an `EraHistory` from a JSON file.
+    ///
+    /// # Arguments
+    ///
+    /// * `path` - Path to the JSON file containing era history data
+    ///
+    /// # Returns
+    ///
+    /// Returns a Result containing the `EraHistory` if successful, or an `EraHistoryFileError` if the file
+    /// cannot be read or parsed.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use amaru_kernel::EraHistory;
+    /// use std::path::Path;
+    ///
+    /// let era_history = EraHistory::load(Path::new("era_history.json")).unwrap();
+    /// ```
+    pub fn load(path: &Path) -> Result<Self, EraHistoryFileError> {
+        let file = File::open(path).map_err(EraHistoryFileError::FileOpenError)?;
+        let reader = BufReader::new(file);
+
+        serde_json::from_reader(reader).map_err(EraHistoryFileError::JsonParseError)
+    }
+
     pub fn new(eras: &[EraSummary], stability_window: Slot) -> EraHistory {
         #[expect(clippy::panic)]
         if eras.is_empty() {
@@ -751,7 +751,7 @@ mod tests {
     use super::*;
     use crate::{
         Epoch, MAINNET_ERA_HISTORY, PREPROD_ERA_HISTORY, PREVIEW_ERA_HISTORY, Slot, any_era_params, any_network_name,
-        from_cbor_no_leftovers_with, load_era_history_from_file, to_cbor,
+        from_cbor_no_leftovers_with, to_cbor,
     };
 
     prop_compose! {
@@ -1268,7 +1268,7 @@ mod tests {
         file.write_all(json_data.as_bytes()).expect("Failed to write JSON data to file");
 
         let loaded_era_history =
-            load_era_history_from_file(temp_file_path.as_path()).expect("Failed to load EraHistory from file");
+            EraHistory::load(temp_file_path.as_path()).expect("Failed to load EraHistory from file");
 
         assert_eq!(*original_era_history, loaded_era_history, "Era histories don't match");
 
@@ -1289,7 +1289,7 @@ mod tests {
     fn test_era_history_file_open_error() {
         let non_existent_path = Path::new("non_existent_file.json");
 
-        let result = load_era_history_from_file(non_existent_path);
+        let result = EraHistory::load(non_existent_path);
 
         match result {
             Err(EraHistoryFileError::FileOpenError(_)) => {
@@ -1310,7 +1310,7 @@ mod tests {
 
         file.write_all(invalid_json.as_bytes()).expect("Failed to write invalid JSON data to file");
 
-        let result = load_era_history_from_file(temp_file_path.as_path());
+        let result = EraHistory::load(temp_file_path.as_path());
 
         match result {
             Err(EraHistoryFileError::JsonParseError(_)) => {

crates/amaru-kernel/src/lib.rs

@@ -96,8 +96,7 @@ pub use cardano::{
     epoch::Epoch,
     era_bound::EraBound,
     era_history::{
-        EraHistory, EraHistoryError, EraHistoryFileError, MAINNET_ERA_HISTORY, PREPROD_ERA_HISTORY,
-        PREVIEW_ERA_HISTORY, load_era_history_from_file,
+        EraHistory, EraHistoryError, EraHistoryFileError, MAINNET_ERA_HISTORY, PREPROD_ERA_HISTORY, PREVIEW_ERA_HISTORY,
     },
     era_name::{EraName, EraNameError},
     era_params::EraParams,

crates/amaru-kernel/src/utils.rs

@@ -18,6 +18,7 @@ pub mod duration;
 #[cfg(all(any(test, feature = "test-utils"), not(target_family = "wasm"), not(target_arch = "riscv32")))]
 pub mod memory;
 pub mod path;
+pub mod process;
 pub mod serde;
 pub mod string;
 #[cfg(any(test, feature = "test-utils"))]

crates/amaru-kernel/src/utils/process.rs

@@ -0,0 +1,111 @@
+// Copyright 2026 PRAGMA
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![cfg(unix)]
+
+use std::process::Command;
+
+pub fn sample_process_memory(pid: u32) -> Option<u64> {
+    let output = if cfg!(target_os = "macos") {
+        Command::new("top").args(["-l", "1", "-pid", &pid.to_string(), "-stats", "pid,mem"]).output().ok()?
+    } else {
+        Command::new("top").args(["-b", "-n", "1", "-p", &pid.to_string()]).output().ok()?
+    };
+
+    if !output.status.success() {
+        return None;
+    }
+
+    parse_top_mem(&String::from_utf8_lossy(&output.stdout), pid)
+}
+
+fn parse_top_mem(output: &str, pid: u32) -> Option<u64> {
+    output.lines().rev().find_map(|line| {
+        let mut fields = line.split_whitespace();
+        if fields.next()?.parse::<u32>().ok()? != pid {
+            return None;
+        }
+
+        if cfg!(target_os = "linux") {
+            for _ in 0..4 {
+                fields.next()?;
+            }
+        }
+
+        let multiplier = if cfg!(target_os = "linux") { 1024 } else { 1 };
+
+        parse_value_with_unit(fields.next()?, multiplier)
+    })
+}
+
+fn parse_value_with_unit(value: &str, plain_multiplier: u64) -> Option<u64> {
+    let value = value.trim_end_matches('+');
+    let suffix = value.chars().last()?;
+
+    let multiplier = match suffix {
+        'K' | 'k' => 1_024f64,
+        'M' | 'm' => 1_024f64 * 1_024f64,
+        'G' | 'g' => 1_024f64 * 1_024f64 * 1_024f64,
+        'T' | 't' => 1_024f64 * 1_024f64 * 1_024f64 * 1_024f64,
+        'P' | 'p' => 1_024f64 * 1_024f64 * 1_024f64 * 1_024f64 * 1_024f64,
+        '0'..='9' => {
+            let amount = value.parse::<u64>().ok()?;
+            return Some(amount.saturating_mul(plain_multiplier));
+        }
+        _ => return None,
+    };
+
+    let amount = value[..value.len() - 1].parse::<f64>().ok()?;
+
+    Some((amount * multiplier).round() as u64)
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use test_case::test_case;
+
+    use super::{parse_top_mem, parse_value_with_unit};
+
+    #[test_case("1201K", 1 => Some(1_229_824))]
+    #[test_case("1.5M", 1 => Some(1_572_864))]
+    #[test_case("2.0G", 1 => Some(2_147_483_648))]
+    #[test_case("42", 1 => Some(42))]
+    #[test_case("42", 1024 => Some(43_008))]
+    #[test_case("1.5g", 1 => Some(1_610_612_736))]
+    #[test_case("150M+", 1 => Some(157_286_400))]
+    fn parses_top_memory_suffixes(value: &str, plain_multiplier: u64) -> Option<u64> {
+        parse_value_with_unit(value, plain_multiplier)
+    }
+
+    #[test]
+    fn parse_top_process_for_memory() {
+        let output = if cfg!(target_os = "linux") {
+            [
+                "top - 12:00:00 up 1 day,  1 user,  load average: 0.00, 0.00, 0.00",
+                "Tasks:   1 total,   1 running,   0 sleeping,   0 stopped,   0 zombie",
+                "%Cpu(s):  0.0 us,  0.0 sy,  0.0 ni,100.0 id,  0.0 wa,  0.0 hi,  0.0 si,  0.0 st ",
+                "MiB Mem :   1024.0 total,    256.0 free,    512.0 used,    256.0 buff/cache",
+                "",
+                "    PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND",
+                "  73194 user      20   0 1234567 654321  12345 S   0.0   0.1   0:00.01 amaru",
+                "",
+            ]
+            .join("\n")
+        } else {
+            ["Processes: 1 total", "PID    MEM", "73194  654321K", ""].join("\n")
+        };
+
+        assert_eq!(parse_top_mem(&output, 73_194), Some(670_024_704));
+    }
+}

crates/amaru-ledger/src/store/columns/proposals.rs

@@ -52,13 +52,13 @@ impl<'a, C> cbor::decode::Decode<'a, C> for Row {
 
 #[cfg(any(test, feature = "test-utils"))]
 pub mod tests {
-    use amaru_kernel::{any_proposal, any_proposal_pointer, prop_cbor_roundtrip};
+    use amaru_kernel::{any_proposal, any_proposal_pointer};
     use proptest::{prelude::*, prop_compose};
 
     use super::*;
 
     #[cfg(not(target_os = "windows"))]
-    prop_cbor_roundtrip!(prop_cbor_roundtrip_row, Row, any_row(u64::MAX));
+    amaru_kernel::prop_cbor_roundtrip!(prop_cbor_roundtrip_row, Row, any_row(u64::MAX));
 
     prop_compose! {
         pub fn any_row(max_slot: u64)(

crates/amaru-metrics/src/consensus.rs

@@ -17,6 +17,8 @@ use std::sync::OnceLock;
 
 #[cfg(not(target_arch = "wasm32"))]
 use opentelemetry::KeyValue;
+#[cfg(not(target_arch = "wasm32"))]
+use opentelemetry::metrics::Meter as OpenTelemetryMeter;
 
 #[cfg(not(target_arch = "wasm32"))]
 use crate::{Counter, Histogram};
@@ -62,6 +64,10 @@ impl MetricRecorder for ConsensusMetrics {
         static FORK_SWITCH_DURATION: OnceLock<Histogram<u64>> = OnceLock::new();
         static FORK_SWITCH_TOTAL: OnceLock<Counter<u64>> = OnceLock::new();
 
+        let Some(meter) = meter.get() else {
+            return;
+        };
+
         match self {
             ConsensusMetrics::HeaderLifecycle {
                 outcome,
@@ -133,7 +139,7 @@ impl MetricRecorder for ConsensusMetrics {
 /// Record a duration to its histogram, if present, without touching any counter.
 #[cfg(not(target_arch = "wasm32"))]
 fn record_optional_duration(
-    meter: &Meter,
+    meter: &OpenTelemetryMeter,
     duration: &'static OnceLock<Histogram<u64>>,
     duration_name: &'static str,
     duration_description: &'static str,
@@ -153,7 +159,7 @@ fn record_optional_duration(
 #[cfg(not(target_arch = "wasm32"))]
 #[allow(clippy::too_many_arguments)]
 fn record_duration(
-    meter: &Meter,
+    meter: &OpenTelemetryMeter,
     duration: &'static OnceLock<Histogram<u64>>,
     total: &'static OnceLock<Counter<u64>>,
     duration_name: &'static str,

crates/amaru-metrics/src/ledger.rs

@@ -63,6 +63,17 @@ impl MetricRecorder for LedgerMetrics {
 #[cfg(not(target_arch = "wasm32"))]
 impl MetricRecorder for LedgerMetrics {
     fn record_to_meter(&self, meter: &Meter) {
+        crate::protocol::TipBlockMetrics {
+            hash: self.block_header_hash.clone(),
+            parent_hash: self.parent_block_header_hash.clone(),
+            issuer_verification_key_hash: self.issuer_verification_key_hash.clone(),
+        }
+        .record_to_meter(meter);
+
+        let Some(meter) = meter.get() else {
+            return;
+        };
+
         static BLOCK_HEIGHT: OnceLock<Gauge<u64>> = OnceLock::new();
         static SLOT_NUM: OnceLock<Gauge<u64>> = OnceLock::new();
         static SLOT_IN_EPOCH: OnceLock<Gauge<u64>> = OnceLock::new();
@@ -128,13 +139,6 @@ impl MetricRecorder for LedgerMetrics {
         density.record(self.density, &[]);
         current_kes_period.record(self.current_kes_period, &[]);
         remaining_kes_periods.record(self.remaining_kes_periods, &[]);
-
-        crate::protocol::TipBlockMetrics {
-            hash: self.block_header_hash.clone(),
-            parent_hash: self.parent_block_header_hash.clone(),
-            issuer_verification_key_hash: self.issuer_verification_key_hash.clone(),
-        }
-        .record_to_meter(meter);
     }
 }