Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions clients/openframe-client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ use crate::services::{
InitialKeyService, LastKnownGoodService, UpdateCleanupService, UpdateHandlerService,
UpdateStateService,
};
use crate::services::{MachineIdService, MACHINE_ID_HEADER};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServerConfig {
Expand Down Expand Up @@ -206,8 +207,21 @@ impl Client {
let config_service = AgentConfigurationService::new(directory_manager.clone())
.context("Failed to initialize device configuration service")?;

let machine_id_service = MachineIdService::new(&directory_manager);
let machine_id = machine_id_service
.get_or_create()
.context("Failed to get or create machine ID")?;

let mut default_headers = reqwest::header::HeaderMap::new();
default_headers.insert(
MACHINE_ID_HEADER,
reqwest::header::HeaderValue::from_str(&machine_id)
.context("Invalid machine ID for header")?,
);

let http_client = reqwest::Client::builder()
.timeout(Duration::from_secs(HTTP_CLIENT_TIMEOUT_SECS))
.default_headers(default_headers.clone())
// disable TLS verification for dev mode only
.danger_accept_invalid_certs(initial_configuration_service.is_local_mode()?)
.no_proxy()
Expand All @@ -218,6 +232,7 @@ impl Client {

let download_client = reqwest::Client::builder()
.timeout(Duration::from_secs(DOWNLOAD_CLIENT_TIMEOUT_SECS))
.default_headers(default_headers)
.danger_accept_invalid_certs(initial_configuration_service.is_local_mode()?)
.no_proxy()
.pool_max_idle_per_host(0)
Expand Down Expand Up @@ -293,6 +308,7 @@ impl Client {
auth_service.clone(),
tls_config_provider,
deactivation_service.clone(),
machine_id_service.clone(),
);

// Initialize tool agent file client
Expand Down
18 changes: 14 additions & 4 deletions clients/openframe-client/src/logging/nats_streaming.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use crate::platform::DirectoryManager;
use crate::services::device_data_fetcher::DeviceDataFetcher;
use crate::services::{
AgentConfigurationService, InitialConfigurationService, InstalledToolsService,
MachineIdService, MACHINE_ID_HEADER,
};

use super::log_parser::LogBatchMessage;
Expand All @@ -21,22 +22,27 @@ const RECONNECT_DELAY_SECS: u64 = 5;
const INITIAL_KEY_CHECK_INTERVAL_SECS: u64 = 10;
const SOURCE_DISCOVERY_INTERVAL_SECS: u64 = 30;
const NATS_SUBJECT: &str = "agents.logs";
const NATS_HEADER_MACHINE_ID: &str = "openframe-client";

pub struct NatsLogConnection {
jetstream: Option<jetstream::Context>,
server_host: String,
tenant_domain: String,
initial_key: String,
machine_id: String,
}

impl NatsLogConnection {
pub fn new(server_host: String, tenant_domain: String, initial_key: String) -> Self {
pub fn new(
server_host: String,
tenant_domain: String,
initial_key: String,
machine_id: String,
) -> Self {
Self {
jetstream: None,
server_host,
tenant_domain,
initial_key,
machine_id,
}
}

Expand All @@ -51,7 +57,7 @@ impl NatsLogConnection {
let client = async_nats::ConnectOptions::new()
.custom_header("x-tenant-domain", &self.tenant_domain)
.custom_header("x-initial-key", &self.initial_key)
.custom_header("x-machine-id", NATS_HEADER_MACHINE_ID)
.custom_header(MACHINE_ID_HEADER, &self.machine_id)
.retry_on_initial_connect()
.reconnect_delay_callback(|attempt| {
let delay = Duration::from_secs(RECONNECT_DELAY_SECS);
Expand Down Expand Up @@ -119,6 +125,7 @@ pub struct LogStreamingRunManager {
agent_config_service: AgentConfigurationService,
installed_tools_service: InstalledToolsService,
directory_manager: DirectoryManager,
machine_id: String,
}

impl LogStreamingRunManager {
Expand All @@ -138,6 +145,7 @@ impl LogStreamingRunManager {

let log_file_path = directory_manager.logs_dir().join("openframe.log");
let offset_file_path = directory_manager.secured_dir().join("log_stream_offset");
let machine_id = MachineIdService::new(directory_manager).get_or_create()?;

Ok(Self {
server_host,
Expand All @@ -149,6 +157,7 @@ impl LogStreamingRunManager {
agent_config_service: agent_config_service.clone(),
installed_tools_service: installed_tools_service.clone(),
directory_manager: directory_manager.clone(),
machine_id,
})
}

Expand All @@ -160,6 +169,7 @@ impl LogStreamingRunManager {
self.server_host.clone(),
self.tenant_domain.clone(),
initial_key,
self.machine_id.clone(),
);

loop {
Expand Down
73 changes: 73 additions & 0 deletions clients/openframe-client/src/services/machine_id_service.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
use anyhow::{Context, Result};
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use tracing::{debug, info};
use uuid::Uuid;

use crate::platform::DirectoryManager;

pub const MACHINE_ID_HEADER: &str = "x-machine-id";

// Locally generated machine identity, distinct from the server-assigned machine_id in agent config.
// Persisted in the shared app-support dir so integrated tool agents (mesh, fleet) can read it.
#[derive(Clone)]
pub struct MachineIdService {
file_path: PathBuf,
cached_id: Arc<std::sync::RwLock<Option<String>>>,
}

impl MachineIdService {
pub fn new(directory_manager: &DirectoryManager) -> Self {
Self {
file_path: directory_manager.app_support_dir().join("machine_id"),
cached_id: Arc::new(std::sync::RwLock::new(None)),
}
}

pub fn get_or_create(&self) -> Result<String> {
if let Some(id) = self.cached_id.read().unwrap().clone() {
return Ok(id);
}

if let Ok(id) = self.read() {
debug!("Using existing machine ID: {}", id);
*self.cached_id.write().unwrap() = Some(id.clone());
return Ok(id);
}

let id = Uuid::new_v4().to_string();
self.write(&id)?;
info!("Generated new machine ID: {}", id);
*self.cached_id.write().unwrap() = Some(id.clone());
Ok(id)
Comment on lines +28 to +43

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

Create the persisted machine ID atomically.

Line 33 ignores every read error. An empty, unreadable, or transiently unavailable file then causes Lines 39-42 to generate a new ID.

Concurrent callers can also both observe a missing file and write different IDs. The RwLock does not protect the full read-create-write sequence across service instances or processes. HTTP, NATS control, and NATS logs can then use different x-machine-id values.

Generate only when the file is NotFound. Serialize initialization with an interprocess lock or atomic create-and-re-read. Write through a temporary file and rename it atomically.

Based on learnings, implement this behavior in the upstream tenant repository first, then retain a 1:1 port here.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@clients/openframe-client/src/services/machine_id_service.rs` around lines 28
- 43, Update MachineIdService::get_or_create to propagate read errors other than
NotFound instead of generating a replacement ID. For a missing file, serialize
initialization across processes using an interprocess lock or atomic
create-and-re-read, write the generated ID through a temporary file, and
atomically rename it into place; then cache and return the persisted winner so
concurrent service instances share one machine ID. Implement the same behavior
in the upstream tenant repository first, then port it 1:1 here.

Source: Learnings

}

// Cached id only; empty string until get_or_create has run.
pub fn get(&self) -> String {
self.cached_id.read().unwrap().clone().unwrap_or_default()
}

fn read(&self) -> Result<String> {
let content = fs::read_to_string(&self.file_path)
.with_context(|| format!("Failed to read {}", self.file_path.display()))?;

let id = content.trim();
if id.is_empty() {
anyhow::bail!("Machine ID file is empty");
}
Ok(id.to_string())
}

fn write(&self, id: &str) -> Result<()> {
if let Some(parent) = self.file_path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("Failed to create {}", parent.display()))?;
}

fs::write(&self.file_path, id)
.with_context(|| format!("Failed to write {}", self.file_path.display()))?;

Ok(())
}
}
2 changes: 2 additions & 0 deletions clients/openframe-client/src/services/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ pub mod last_known_good_service;
pub mod local_tls_config_provider;
pub mod machine_heartbeat_publisher;
pub mod machine_heartbeat_run_manager;
pub mod machine_id_service;
pub mod mesh_self_heal_service;
pub mod nats_connection_manager;
pub mod nats_message_publisher;
Expand Down Expand Up @@ -54,6 +55,7 @@ pub use last_known_good_service::LastKnownGoodService;
pub use local_tls_config_provider::LocalTlsConfigProvider;
pub use machine_heartbeat_publisher::MachineHeartbeatPublisher;
pub use machine_heartbeat_run_manager::MachineHeartbeatRunManager;
pub use machine_id_service::{MachineIdService, MACHINE_ID_HEADER};
pub use nats_connection_manager::NatsConnectionManager;
pub use nats_message_publisher::NatsMessagePublisher;
pub use openframe_client_info_service::OpenFrameClientInfoService;
Expand Down
11 changes: 9 additions & 2 deletions clients/openframe-client/src/services/nats_connection_manager.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
use crate::services::agent_configuration_service::AgentConfigurationService;
use crate::services::deactivation_service::DeactivationService;
use crate::services::local_tls_config_provider::LocalTlsConfigProvider;
use crate::services::{AgentAuthService, InitialConfigurationService};
use crate::services::{
AgentAuthService, InitialConfigurationService, MachineIdService, MACHINE_ID_HEADER,
};
use anyhow::{Context, Result};
use async_nats::{Client, Event};
use log::error;
Expand All @@ -25,6 +27,7 @@ pub struct NatsConnectionManager {
initial_configuration_service: InitialConfigurationService,
auth_service: AgentAuthService,
deactivation: Arc<DeactivationService>,
machine_id_service: MachineIdService,
}

impl NatsConnectionManager {
Expand All @@ -38,6 +41,7 @@ impl NatsConnectionManager {
auth_service: AgentAuthService,
tls_config_provider: LocalTlsConfigProvider,
deactivation: Arc<DeactivationService>,
machine_id_service: MachineIdService,
) -> Self {
let (reconnect_tx, _) = broadcast::channel(16);
Self {
Expand All @@ -49,6 +53,7 @@ impl NatsConnectionManager {
initial_configuration_service,
auth_service,
deactivation,
machine_id_service,
}
}

Expand All @@ -57,7 +62,9 @@ impl NatsConnectionManager {
}

pub async fn connect(&self) -> Result<()> {
// Server-assigned machine_id names the NATS connection; the local one goes in the header
let machine_id = self.config_service.get_machine_id()?;
let local_machine_id = self.machine_id_service.get();

info!(
hostname = %self.nats_server_url,
Expand Down Expand Up @@ -129,7 +136,7 @@ impl NatsConnectionManager {
.await
}
})
.custom_header("X-MACHINE-ID", &machine_id);
.custom_header(MACHINE_ID_HEADER, &local_machine_id);

// Only add TLS config in development mode
if self.initial_configuration_service.is_local_mode()? {
Expand Down
Loading