michel/clusterflux-public
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Update public backend API surface

Browse source 
Private source commit: ba3f7ce2b6d9
This commit is contained in:
Clusterflux Release 2026-07-26 17:22:33 +02:00
parent 784463622c
commit 26fdcb9d84
34 changed files with 5473 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

16
crates/clusterflux-coordinator/src/lib.rs
View file

@ -585,6 +585,22 @@ impl Coordinator {
.count()
}
pub fn tenant_count(&self) -> usize {
self.durable.tenants.len()
}
pub fn user_count(&self) -> usize {
self.durable.users.len()
}
pub fn project_count(&self) -> usize {
self.durable.projects.len()
}
pub fn node_identity_count(&self) -> usize {
self.durable.node_identities.len()
}
pub fn node_identity(
&self,
tenant: &TenantId,

186
crates/clusterflux-coordinator/src/service.rs
View file

@ -7,9 +7,10 @@ use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};
use clusterflux_core::{
Actor, AgentId, ArtifactRegistry, CapabilityReportError, CredentialKind, Digest, LimitError,
NativeQuicTransport, NodeDescriptor, NodeId, PanelState, Placement, ProcessId, ProjectId,
RateLimit, TenantId, TransportError, UserId,
Actor, AgentId, ApiError, ApiErrorCategory, ApiErrorCode, ArtifactRegistry,
CapabilityReportError, CredentialKind, Digest, DownloadError, LimitError, NativeQuicTransport,
NodeDescriptor, NodeId, PanelError, PanelState, Placement, ProcessId, ProjectId, RateLimit,
TenantId, TransportError, UserId,
};
use thiserror::Error;
@ -34,6 +35,7 @@ mod quota;
mod relay;
mod routing;
mod signed_nodes;
mod summaries;
mod tcp;
mod wire_protocol;
use authorization::authorize_authenticated_user_operation;
@ -43,12 +45,14 @@ use keys::{
ProcessControlKey, TaskAssignmentKey, TaskControlKey, TaskRestartKey,
};
pub use protocol::{
ArtifactTransferAssignment, AuthenticatedCoordinatorRequest, CoordinatorRequest,
CoordinatorResponse, DebugAcknowledgementState, DebugAuditEvent,
DebugParticipantAcknowledgement, SourcePreparationDisposition, SourcePreparationStatus,
TaskAssignment, TaskAttemptSnapshot, TaskAttemptState, TaskCancellationTarget,
TaskCompletionEvent, TaskExecutor, TaskFailureResolution, TaskReplacementBundle,
TaskTerminalState, VirtualProcessStatus, WorkflowActor,
ArtifactAvailability, ArtifactRetentionState, ArtifactSummary, ArtifactTransferAssignment,
AuthenticatedCoordinatorRequest, CoordinatorRequest, CoordinatorResponse,
DebugAcknowledgementState, DebugAuditEvent, DebugEpochSummary, DebugParticipantAcknowledgement,
NodeSummary, ProcessActivityState, ProcessFinalResult, ProcessLifecycleState, ProcessSummary,
RecentLogEntry, SourcePreparationDisposition, SourcePreparationStatus, TaskAssignment,
TaskAttemptSnapshot, TaskAttemptState, TaskCancellationTarget, TaskCompletionEvent,
TaskExecutor, TaskFailureResolution, TaskLogStream, TaskReplacementBundle, TaskTerminalState,
VirtualProcessStatus, WorkflowActor,
};
pub use quota::CoordinatorQuotaConfiguration;
pub use relay::{
@ -69,9 +73,15 @@ const MAX_RESTART_CHECKPOINTS_PER_PROCESS: usize = 128;
const MAX_TASK_EVENTS_TOTAL: usize = 8_192;
const MAX_DEBUG_AUDIT_EVENTS_TOTAL: usize = 8_192;
const MAX_RESTART_CHECKPOINTS_TOTAL: usize = 4_096;
const MAX_TASK_ATTEMPT_HISTORIES: usize = 4_096;
const MAX_TASK_ATTEMPT_HISTORIES: usize = 1_000_000;
const MAX_IN_FLIGHT_TASKS_PER_PROCESS: usize = 256;
const MAX_NODE_REPORTED_OBJECTS_PER_KIND: usize = 1_024;
const MAX_RECENT_LOG_ENTRIES_PER_PROCESS: usize = 256;
const MAX_RECENT_LOG_ENTRIES_PER_PROJECT: usize = 1_024;
const MAX_RECENT_LOG_BYTES_PER_PROJECT: usize = 512 * 1024;
const MAX_RECENT_LOG_CHUNK_BYTES: usize = 16 * 1024;
const MAX_RECENT_PROCESS_SUMMARIES_PER_PROJECT: usize = 32;
const MAX_RECENT_PROCESS_SUMMARIES_TOTAL: usize = 8_192;
const DEFAULT_NODE_STALE_AFTER_SECONDS: u64 = 30;
fn bounded_ttl(requested: u64, maximum: u64) -> u64 {
requested.clamp(1, maximum)
@ -111,6 +121,24 @@ pub struct CoordinatorAdmission {
pub max_artifact_download_ttl_seconds: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CoordinatorOperationalMetrics {
pub tenants: usize,
pub users: usize,
pub projects: usize,
pub enrolled_nodes: usize,
pub reported_nodes: usize,
pub live_nodes: usize,
pub active_processes: usize,
pub active_coordinator_mains: usize,
pub max_active_coordinator_mains: usize,
pub active_tasks: usize,
pub queued_tasks: usize,
pub artifacts: usize,
pub retained_download_links: usize,
pub relay: ArtifactRelayUsage,
}
impl Default for CoordinatorAdmission {
fn default() -> Self {
Self {
@ -151,6 +179,98 @@ pub enum CoordinatorServiceError {
Durable(String),
}
impl CoordinatorServiceError {
pub fn api_error(&self, request_id: impl Into<String>) -> ApiError {
let request_id = request_id.into();
let message = self.to_string();
let (code, category, retryable) = match self {
Self::Io(_) => (
ApiErrorCode::TemporaryCapacity,
ApiErrorCategory::Availability,
true,
),
Self::Json(_)
| Self::CapabilityReport(_)
| Self::InvalidArtifactPath(_)
| Self::InvalidTaskLogTail(_) => (
ApiErrorCode::ValidationError,
ApiErrorCategory::Validation,
false,
),
Self::Protocol(_) | Self::Coordinator(CoordinatorError::Unauthorized(_)) => {
return ApiError::from_message(request_id, message);
}
Self::Coordinator(CoordinatorError::UnknownNode) => {
(ApiErrorCode::NotFound, ApiErrorCategory::State, false)
}
Self::Coordinator(CoordinatorError::Enrollment(_)) => (
ApiErrorCode::Unauthenticated,
ApiErrorCategory::Authentication,
false,
),
Self::Coordinator(CoordinatorError::StaleProcessEpoch { .. }) => {
(ApiErrorCode::Conflict, ApiErrorCategory::State, true)
}
Self::Download(DownloadError::NotFound) => {
(ApiErrorCode::NotFound, ApiErrorCategory::State, false)
}
Self::Download(DownloadError::Unavailable)
| Self::Download(DownloadError::DirectConnectivityUnavailable(_)) => (
ApiErrorCode::ArtifactUnavailable,
ApiErrorCategory::Availability,
true,
),
Self::Download(DownloadError::LimitExceeded { .. }) => (
ApiErrorCode::ArtifactLimitExceeded,
ApiErrorCategory::Resource,
false,
),
Self::Download(DownloadError::Unauthorized(_))
| Self::Download(DownloadError::InvalidToken)
| Self::Download(DownloadError::Expired)
| Self::Download(DownloadError::Revoked) => (
ApiErrorCode::Forbidden,
ApiErrorCategory::Authorization,
false,
),
Self::Download(DownloadError::Usage(_)) | Self::Resource(_) => (
ApiErrorCode::QuotaExceeded,
ApiErrorCategory::Resource,
true,
),
Self::Scheduler(_) => (
ApiErrorCode::NoCapableNode,
ApiErrorCategory::Availability,
true,
),
Self::Transport(_) => (
ApiErrorCode::NodeOffline,
ApiErrorCategory::Availability,
true,
),
Self::Panel(PanelError::RateLimited) => (
ApiErrorCode::QuotaExceeded,
ApiErrorCategory::Resource,
true,
),
Self::Panel(PanelError::UnknownWidget(_)) => {
(ApiErrorCode::NotFound, ApiErrorCategory::State, false)
}
Self::Panel(_) => (
ApiErrorCode::Forbidden,
ApiErrorCategory::Authorization,
false,
),
Self::Durable(_) => (
ApiErrorCode::InternalError,
ApiErrorCategory::Internal,
true,
),
};
ApiError::new(code, category, message, retryable, request_id)
}
}
pub struct CoordinatorService {
coordinator: Coordinator,
store: RuntimeDurableStore,
@ -161,6 +281,22 @@ pub struct CoordinatorService {
enrollment_grants: BTreeMap<EnrollmentGrantKey, clusterflux_core::EnrollmentGrant>,
task_events: VecDeque<TaskCompletionEvent>,
process_scope_history: VecDeque<ProcessControlKey>,
process_summaries: BTreeMap<ProcessControlKey, summaries::StoredProcessSummary>,
process_summary_order: VecDeque<ProcessControlKey>,
next_process_summary_order: u64,
recent_logs: BTreeMap<(TenantId, ProjectId), VecDeque<RecentLogEntry>>,
recent_log_dropped_through: BTreeMap<ProcessControlKey, u64>,
recent_log_offsets: BTreeMap<
(
TenantId,
ProjectId,
ProcessId,
clusterflux_core::TaskInstanceId,
String,
),
u64,
>,
next_recent_log_sequence: u64,
debug_audit_events: VecDeque<DebugAuditEvent>,
debug_epochs: BTreeMap<ProcessControlKey, u64>,
debug_epoch_runtime: BTreeMap<ProcessControlKey, debug::DebugEpochRuntime>,
@ -215,6 +351,29 @@ impl CoordinatorService {
self.artifact_relay.usage()
}
pub fn operational_metrics(&self) -> CoordinatorOperationalMetrics {
CoordinatorOperationalMetrics {
tenants: self.coordinator.tenant_count(),
users: self.coordinator.user_count(),
projects: self.coordinator.project_count(),
enrolled_nodes: self.coordinator.node_identity_count(),
reported_nodes: self.node_descriptors.len(),
live_nodes: self
.node_descriptors
.keys()
.filter(|scope| self.node_is_live(scope))
.count(),
active_processes: self.coordinator.active_process_count(),
active_coordinator_mains: self.main_runtime.active_main_count(),
max_active_coordinator_mains: self.main_runtime.max_active_mains(),
active_tasks: self.active_tasks.len(),
queued_tasks: self.pending_task_launches.len(),
artifacts: self.artifact_registry.artifact_count(),
retained_download_links: self.artifact_registry.retained_download_link_count(),
relay: self.artifact_relay.usage(),
}
}
fn commit_artifact_relay(
&mut self,
candidate: relay::ArtifactRelayLedger,
@ -404,6 +563,13 @@ impl CoordinatorService {
enrollment_grants: BTreeMap::new(),
task_events: VecDeque::new(),
process_scope_history: VecDeque::new(),
process_summaries: BTreeMap::new(),
process_summary_order: VecDeque::new(),
next_process_summary_order: 1,
recent_logs: BTreeMap::new(),
recent_log_dropped_through: BTreeMap::new(),
recent_log_offsets: BTreeMap::new(),
next_recent_log_sequence: 1,
debug_audit_events: VecDeque::new(),
debug_epochs: BTreeMap::new(),
debug_epoch_runtime: BTreeMap::new(),

17
crates/clusterflux-coordinator/src/service/authorization.rs
View file

@ -17,6 +17,7 @@ pub(super) enum PublicUserOperation {
RevokeAgentPublicKey,
CreateNodeEnrollmentGrant,
ListNodeDescriptors,
ListNodeSummaries,
RevokeNodeCredential,
StartProcess,
ScheduleTask,
@ -24,6 +25,7 @@ pub(super) enum PublicUserOperation {
CancelProcess,
AbortProcess,
ListProcesses,
ListProcessSummaries,
QuotaStatus,
RestartTask,
ResolveTaskFailure,
@ -35,7 +37,10 @@ pub(super) enum PublicUserOperation {
InspectDebugEpoch,
ListTaskEvents,
ListTaskSnapshots,
ListRecentLogs,
JoinTask,
ListArtifacts,
GetArtifact,
CreateArtifactDownloadLink,
OpenArtifactDownloadStream,
RevokeArtifactDownloadLink,
@ -56,6 +61,7 @@ impl PublicUserOperation {
Self::RevokeAgentPublicKey => "revoke_agent_public_key",
Self::CreateNodeEnrollmentGrant => "create_node_enrollment_grant",
Self::ListNodeDescriptors => "list_node_descriptors",
Self::ListNodeSummaries => "list_node_summaries",
Self::RevokeNodeCredential => "revoke_node_credential",
Self::StartProcess => "start_process",
Self::ScheduleTask => "schedule_task",
@ -63,6 +69,7 @@ impl PublicUserOperation {
Self::CancelProcess => "cancel_process",
Self::AbortProcess => "abort_process",
Self::ListProcesses => "list_processes",
Self::ListProcessSummaries => "list_process_summaries",
Self::QuotaStatus => "quota_status",
Self::RestartTask => "restart_task",
Self::ResolveTaskFailure => "resolve_task_failure",
@ -74,7 +81,10 @@ impl PublicUserOperation {
Self::InspectDebugEpoch => "inspect_debug_epoch",
Self::ListTaskEvents => "list_task_events",
Self::ListTaskSnapshots => "list_task_snapshots",
Self::ListRecentLogs => "list_recent_logs",
Self::JoinTask => "join_task",
Self::ListArtifacts => "list_artifacts",
Self::GetArtifact => "get_artifact",
Self::CreateArtifactDownloadLink => "create_artifact_download_link",
Self::OpenArtifactDownloadStream => "open_artifact_download_stream",
Self::RevokeArtifactDownloadLink => "revoke_artifact_download_link",
@ -105,6 +115,7 @@ impl From<&AuthenticatedCoordinatorRequest> for PublicUserOperation {
Self::CreateNodeEnrollmentGrant
}
AuthenticatedCoordinatorRequest::ListNodeDescriptors => Self::ListNodeDescriptors,
AuthenticatedCoordinatorRequest::ListNodeSummaries { .. } => Self::ListNodeSummaries,
AuthenticatedCoordinatorRequest::RevokeNodeCredential { .. } => {
Self::RevokeNodeCredential
}
@ -114,6 +125,9 @@ impl From<&AuthenticatedCoordinatorRequest> for PublicUserOperation {
AuthenticatedCoordinatorRequest::CancelProcess { .. } => Self::CancelProcess,
AuthenticatedCoordinatorRequest::AbortProcess { .. } => Self::AbortProcess,
AuthenticatedCoordinatorRequest::ListProcesses => Self::ListProcesses,
AuthenticatedCoordinatorRequest::ListProcessSummaries { .. } => {
Self::ListProcessSummaries
}
AuthenticatedCoordinatorRequest::QuotaStatus => Self::QuotaStatus,
AuthenticatedCoordinatorRequest::RestartTask { .. } => Self::RestartTask,
AuthenticatedCoordinatorRequest::ResolveTaskFailure { .. } => Self::ResolveTaskFailure,
@ -129,7 +143,10 @@ impl From<&AuthenticatedCoordinatorRequest> for PublicUserOperation {
AuthenticatedCoordinatorRequest::InspectDebugEpoch { .. } => Self::InspectDebugEpoch,
AuthenticatedCoordinatorRequest::ListTaskEvents { .. } => Self::ListTaskEvents,
AuthenticatedCoordinatorRequest::ListTaskSnapshots { .. } => Self::ListTaskSnapshots,
AuthenticatedCoordinatorRequest::ListRecentLogs { .. } => Self::ListRecentLogs,
AuthenticatedCoordinatorRequest::JoinTask { .. } => Self::JoinTask,
AuthenticatedCoordinatorRequest::ListArtifacts { .. } => Self::ListArtifacts,
AuthenticatedCoordinatorRequest::GetArtifact { .. } => Self::GetArtifact,
AuthenticatedCoordinatorRequest::CreateArtifactDownloadLink { .. } => {
Self::CreateArtifactDownloadLink
}

301
crates/clusterflux-coordinator/src/service/logs.rs
View file

@ -9,7 +9,10 @@ use super::keys::{process_control_key, task_control_key, task_restart_key};
use super::protocol::TaskAttemptState;
use super::{
artifact_id_from_path, CoordinatorResponse, CoordinatorService, CoordinatorServiceError,
TaskCompletionEvent, TaskTerminalState, MAX_TASK_LOG_TAIL_BYTES,
RecentLogEntry, TaskCompletionEvent, TaskLogStream, TaskTerminalState,
MAX_RECENT_LOG_BYTES_PER_PROJECT, MAX_RECENT_LOG_CHUNK_BYTES,
MAX_RECENT_LOG_ENTRIES_PER_PROCESS, MAX_RECENT_LOG_ENTRIES_PER_PROJECT,
MAX_TASK_LOG_TAIL_BYTES,
};
impl CoordinatorService {
@ -42,6 +45,34 @@ impl CoordinatorService {
.can_charge_log_bytes(&tenant, &project, reported_bytes, now_epoch_seconds)?;
self.quota
.charge_log_bytes(&tenant, &project, reported_bytes, now_epoch_seconds)?;
let stdout_key =
recent_log_offset_key(&tenant, &project, &process, &task, &TaskLogStream::Stdout);
let stderr_key =
recent_log_offset_key(&tenant, &project, &process, &task, &TaskLogStream::Stderr);
if !stdout_tail.is_empty() && !self.recent_log_offsets.contains_key(&stdout_key) {
self.record_recent_log(
tenant.clone(),
project.clone(),
process.clone(),
task.clone(),
TaskLogStream::Stdout,
stdout_tail.clone(),
stdout_truncated,
now_epoch_seconds,
);
}
if !stderr_tail.is_empty() && !self.recent_log_offsets.contains_key(&stderr_key) {
self.record_recent_log(
tenant.clone(),
project.clone(),
process.clone(),
task.clone(),
TaskLogStream::Stderr,
stderr_tail.clone(),
stderr_truncated,
now_epoch_seconds,
);
}
Ok(CoordinatorResponse::TaskLogRecorded {
process,
task,
@ -61,6 +92,98 @@ impl CoordinatorService {
})
}
#[allow(clippy::too_many_arguments)]
pub(super) fn handle_report_task_log_chunk(
&mut self,
tenant: String,
project: String,
process: String,
node: String,
task: String,
stream: TaskLogStream,
offset: u64,
source_bytes: u64,
text: String,
truncated: bool,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
if text.len() > MAX_RECENT_LOG_CHUNK_BYTES {
return Err(CoordinatorServiceError::InvalidTaskLogTail(format!(
"live log chunk is {} bytes; max is {MAX_RECENT_LOG_CHUNK_BYTES}",
text.len()
)));
}
if source_bytes == 0 && !text.is_empty() && !truncated {
return Err(CoordinatorServiceError::Protocol(
"live log chunk source_bytes must describe non-empty text".to_owned(),
));
}
if source_bytes > (MAX_RECENT_LOG_CHUNK_BYTES as u64).saturating_mul(4) {
return Err(CoordinatorServiceError::Protocol(
"live log chunk source_bytes exceeds the bounded chunk allowance".to_owned(),
));
}
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let process = ProcessId::new(process);
let node = NodeId::new(node);
let task = TaskInstanceId::new(task);
self.authorize_node_for_process_or_termination(&node, &tenant, &project, &process)?;
let key = recent_log_offset_key(&tenant, &project, &process, &task, &stream);
let expected = self.recent_log_offsets.get(&key).copied().unwrap_or(0);
let end = offset.checked_add(source_bytes).ok_or_else(|| {
CoordinatorServiceError::Protocol(
"live log chunk offset exceeds the supported range".to_owned(),
)
})?;
if end <= expected {
return Ok(CoordinatorResponse::TaskLogChunkRecorded {
process,
task,
sequence: None,
next_offset: expected,
});
}
let now_epoch_seconds = self.current_epoch_seconds()?;
if offset > expected {
self.record_recent_log(
tenant.clone(),
project.clone(),
process.clone(),
task.clone(),
stream.clone(),
format!("[log output lost: {} bytes]", offset - expected),
true,
now_epoch_seconds,
);
} else if offset < expected {
return Ok(CoordinatorResponse::TaskLogChunkRecorded {
process,
task,
sequence: None,
next_offset: expected,
});
}
let sequence = (!text.is_empty()).then(|| {
self.record_recent_log(
tenant.clone(),
project.clone(),
process.clone(),
task.clone(),
stream,
text,
truncated,
now_epoch_seconds,
)
});
self.recent_log_offsets.insert(key, end);
Ok(CoordinatorResponse::TaskLogChunkRecorded {
process,
task,
sequence,
next_offset: end,
})
}
pub(super) fn handle_report_vfs_metadata(
&mut self,
tenant: String,
@ -221,6 +344,12 @@ impl CoordinatorService {
self.task_aborts.remove(&task_key);
self.debug_commands.remove(&task_key);
self.active_tasks.remove(&task_key);
self.clear_recent_log_offsets_for_task(
&event.tenant,
&event.project,
&event.process,
&event.task,
);
self.task_assignments.retain(|_, assignments| {
assignments.retain(|assignment| {
assignment.tenant != event.tenant
@ -320,7 +449,7 @@ impl CoordinatorService {
Ok(CoordinatorResponse::TaskSnapshots { snapshots })
}
fn authorize_task_event_process_scope(
pub(super) fn authorize_task_event_process_scope(
&self,
tenant: &TenantId,
project: &ProjectId,
@ -360,6 +489,47 @@ impl CoordinatorService {
Ok(())
}
pub(super) fn handle_list_recent_logs(
&mut self,
tenant: String,
project: String,
actor_user: String,
process: String,
task: Option<String>,
after_sequence: Option<u64>,
limit: u32,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let _actor = UserId::new(actor_user);
let process = ProcessId::new(process);
let task = task.map(TaskInstanceId::new);
self.authorize_task_event_process_scope(&tenant, &project, &process)?;
let after_sequence = after_sequence.unwrap_or(0);
let retained = self.recent_logs.get(&(tenant.clone(), project.clone()));
let history_truncated = self
.recent_log_dropped_through
.get(&process_control_key(&tenant, &project, &process))
.is_some_and(|dropped_through| *dropped_through > after_sequence);
let entries = retained
.into_iter()
.flatten()
.filter(|entry| {
entry.process == process
&& entry.sequence > after_sequence
&& task.as_ref().is_none_or(|task| &entry.task == task)
})
.take(limit as usize)
.cloned()
.collect::<Vec<_>>();
let next_sequence = entries.last().map(|entry| entry.sequence);
Ok(CoordinatorResponse::RecentLogs {
entries,
next_sequence,
history_truncated,
})
}
pub(super) fn handle_join_task(
&mut self,
tenant: String,
@ -515,6 +685,94 @@ impl CoordinatorService {
self.task_events.push_back(event);
}
#[allow(clippy::too_many_arguments)]
fn record_recent_log(
&mut self,
tenant: TenantId,
project: ProjectId,
process: ProcessId,
task: TaskInstanceId,
stream: TaskLogStream,
mut text: String,
mut truncated: bool,
server_timestamp_epoch_seconds: u64,
) -> u64 {
if text.len() > MAX_RECENT_LOG_CHUNK_BYTES {
let mut boundary = MAX_RECENT_LOG_CHUNK_BYTES;
while !text.is_char_boundary(boundary) {
boundary -= 1;
}
text.truncate(boundary);
truncated = true;
}
let sequence = self.next_recent_log_sequence;
self.next_recent_log_sequence = self.next_recent_log_sequence.saturating_add(1);
let logs = self
.recent_logs
.entry((tenant.clone(), project.clone()))
.or_default();
let mut dropped = Vec::new();
while logs.iter().filter(|entry| entry.process == process).count()
>= MAX_RECENT_LOG_ENTRIES_PER_PROCESS
{
let Some(index) = logs.iter().position(|entry| entry.process == process) else {
break;
};
if let Some(entry) = logs.remove(index) {
dropped.push(entry);
}
}
while logs.len() >= MAX_RECENT_LOG_ENTRIES_PER_PROJECT
|| logs
.iter()
.map(|entry| entry.text.len())
.sum::<usize>()
.saturating_add(text.len())
> MAX_RECENT_LOG_BYTES_PER_PROJECT
{
match logs.pop_front() {
Some(entry) => dropped.push(entry),
None => break,
}
}
logs.push_back(RecentLogEntry {
sequence,
process,
task,
stream,
text,
server_timestamp_epoch_seconds,
truncated,
});
for entry in dropped {
let key = process_control_key(&tenant, &project, &entry.process);
self.recent_log_dropped_through
.entry(key)
.and_modify(|dropped_through| {
*dropped_through = (*dropped_through).max(entry.sequence);
})
.or_insert(entry.sequence);
}
sequence
}
pub(super) fn clear_recent_log_offsets_for_task(
&mut self,
tenant: &TenantId,
project: &ProjectId,
process: &ProcessId,
task: &TaskInstanceId,
) {
self.recent_log_offsets.retain(
|(entry_tenant, entry_project, entry_process, entry_task, _), _| {
entry_tenant != tenant
|| entry_project != project
|| entry_process != process
|| entry_task != task
},
);
}
fn finish_task_attempt(&mut self, event: &mut TaskCompletionEvent) -> bool {
let key = task_restart_key(&event.tenant, &event.project, &event.process, &event.task);
let Some(attempt) = self
@ -619,6 +877,25 @@ impl CoordinatorService {
{
return Ok(false);
}
let final_result = if cancellation_completed {
super::ProcessFinalResult::Cancelled
} else if self.task_events.iter().any(|event| {
&event.tenant == tenant
&& &event.project == project
&& &event.process == process
&& event.terminal_state == TaskTerminalState::Failed
}) {
super::ProcessFinalResult::Failed
} else {
super::ProcessFinalResult::Completed
};
self.record_process_terminal(
tenant,
project,
process,
final_result,
self.current_epoch_seconds()?,
);
self.coordinator.abort_process(tenant, project, process)?;
self.clear_debug_state_for_process(tenant, project, process);
self.clear_operator_panel_state(tenant, project, process);
@ -736,6 +1013,26 @@ fn checked_reported_log_bytes(
})
}
fn recent_log_offset_key(
tenant: &TenantId,
project: &ProjectId,
process: &ProcessId,
task: &TaskInstanceId,
stream: &TaskLogStream,
) -> (TenantId, ProjectId, ProcessId, TaskInstanceId, String) {
(
tenant.clone(),
project.clone(),
process.clone(),
task.clone(),
match stream {
TaskLogStream::Stdout => "stdout",
TaskLogStream::Stderr => "stderr",
}
.to_owned(),
)
}
fn validate_task_log_tail(kind: &str, value: &str) -> Result<(), CoordinatorServiceError> {
if value.len() > MAX_TASK_LOG_TAIL_BYTES {
return Err(CoordinatorServiceError::InvalidTaskLogTail(format!(

22
crates/clusterflux-coordinator/src/service/main_runtime.rs
View file

@ -109,6 +109,14 @@ impl Default for CoordinatorMainRuntime {
}
impl CoordinatorMainRuntime {
pub(super) fn active_main_count(&self) -> usize {
self.controls.len()
}
pub(super) fn max_active_mains(&self) -> usize {
self.max_active_mains
}
pub(super) fn configure(
&mut self,
configuration: super::CoordinatorMainRuntimeConfiguration,
@ -729,6 +737,13 @@ impl CoordinatorService {
&process,
"coordinator main launch failed admission or validation",
);
self.record_process_terminal(
&tenant,
&project,
&process,
super::ProcessFinalResult::Failed,
self.liveness_now_epoch_seconds(),
);
let _ = self.coordinator.abort_process(&tenant, &project, &process);
}
result
@ -992,6 +1007,13 @@ impl CoordinatorService {
}
}
self.process_aborts.insert(process_key.clone());
self.record_process_terminal(
&scope.tenant,
&scope.project,
&scope.process,
super::ProcessFinalResult::Failed,
self.liveness_now_epoch_seconds(),
);
let _ = self
.coordinator
.abort_process(&scope.tenant, &scope.project, &scope.process);

19
crates/clusterflux-coordinator/src/service/processes.rs
View file

@ -392,11 +392,12 @@ impl CoordinatorService {
event.tenant != tenant || event.project != project || event.process != process
});
let active = self.coordinator.start_process_for_launch_attempt(
tenant,
project,
tenant.clone(),
project.clone(),
process.clone(),
launch_attempt.map(clusterflux_core::LaunchAttemptId::new),
);
self.record_process_started(&tenant, &project, &process, now_epoch_seconds);
Ok(CoordinatorResponse::ProcessStarted {
process,
launch_attempt: active
@ -513,6 +514,13 @@ impl CoordinatorService {
}
let process_key = process_control_key(&tenant, &project, &process);
if cancelled_tasks.is_empty() && !self.main_runtime.controls.contains_key(&process_key) {
self.record_process_terminal(
&tenant,
&project,
&process,
super::ProcessFinalResult::Cancelled,
self.current_epoch_seconds()?,
);
self.coordinator
.abort_process(&tenant, &project, &process)?;
self.clear_operator_panel_state(&tenant, &project, &process);
@ -598,6 +606,13 @@ impl CoordinatorService {
}
}
self.record_process_terminal(
&tenant,
&project,
&process,
super::ProcessFinalResult::Cancelled,
self.current_epoch_seconds()?,
);
if let Some(launch_attempt) = launch_attempt.as_ref() {
self.coordinator.abort_process_for_launch_attempt(
&tenant,

214
crates/clusterflux-coordinator/src/service/protocol.rs
View file

@ -150,6 +150,15 @@ pub enum CoordinatorRequest {
project: String,
actor_user: String,
},
ListNodeSummaries {
tenant: String,
project: String,
actor_user: String,
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
RevokeNodeCredential {
tenant: String,
project: String,
@ -303,6 +312,15 @@ pub enum CoordinatorRequest {
project: String,
actor_user: String,
},
ListProcessSummaries {
tenant: String,
project: String,
actor_user: String,
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
QuotaStatus {
tenant: String,
project: String,
@ -429,6 +447,19 @@ pub enum CoordinatorRequest {
stderr_truncated: bool,
backpressured: bool,
},
ReportTaskLogChunk {
tenant: String,
project: String,
process: String,
node: String,
task: String,
stream: TaskLogStream,
offset: u64,
source_bytes: u64,
text: String,
#[serde(default)]
truncated: bool,
},
ReportVfsMetadata {
tenant: String,
project: String,
@ -478,6 +509,18 @@ pub enum CoordinatorRequest {
actor_user: String,
process: String,
},
ListRecentLogs {
tenant: String,
project: String,
actor_user: String,
process: String,
#[serde(default)]
task: Option<String>,
#[serde(default)]
after_sequence: Option<u64>,
#[serde(default = "default_log_page_limit")]
limit: u32,
},
JoinTask {
tenant: String,
project: String,
@ -510,6 +553,23 @@ pub enum CoordinatorRequest {
#[serde(default = "default_download_ttl_seconds")]
ttl_seconds: u64,
},
ListArtifacts {
tenant: String,
project: String,
actor_user: String,
#[serde(default)]
process: Option<String>,
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
GetArtifact {
tenant: String,
project: String,
actor_user: String,
artifact: String,
},
OpenArtifactDownloadStream {
tenant: String,
project: String,
@ -666,6 +726,18 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))
}
CoordinatorRequest::ListNodeSummaries {
tenant,
project,
actor_user,
cursor,
limit,
} => {
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))?;
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
CoordinatorRequest::ExchangeNodeEnrollmentGrant {
tenant,
project,
@ -901,6 +973,14 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
task,
..
}
| CoordinatorRequest::ReportTaskLogChunk {
tenant,
project,
process,
node,
task,
..
}
| CoordinatorRequest::ReportVfsMetadata {
tenant,
project,
@ -979,6 +1059,23 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
validate_user(actor_user, &format!("{path}.actor_user"))?;
validate_process(process, &format!("{path}.process"))
}
CoordinatorRequest::ListRecentLogs {
tenant,
project,
actor_user,
process,
task,
limit,
..
} => {
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))?;
validate_process(process, &format!("{path}.process"))?;
if let Some(task) = task {
validate_task_instance(task, &format!("{path}.task"))?;
}
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
CoordinatorRequest::AbortProcess {
tenant,
project,
@ -1007,6 +1104,18 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))
}
CoordinatorRequest::ListProcessSummaries {
tenant,
project,
actor_user,
cursor,
limit,
} => {
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))?;
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 100)
}
CoordinatorRequest::RestartTask {
tenant,
project,
@ -1080,6 +1189,20 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
validate_process(process, &format!("{path}.process"))?;
validate_external_token(widget_id, &format!("{path}.widget_id"), 256)
}
CoordinatorRequest::ListArtifacts {
tenant,
project,
actor_user,
process,
cursor,
limit,
} => {
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))?;
validate_optional_process(process.as_deref(), &format!("{path}.process"))?;
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
CoordinatorRequest::CreateArtifactDownloadLink {
tenant,
project,
@ -1100,6 +1223,12 @@ fn validate_coordinator_request(request: &CoordinatorRequest, path: &str) -> Res
actor_user,
artifact,
..
}
| CoordinatorRequest::GetArtifact {
tenant,
project,
actor_user,
artifact,
} => {
validate_tenant_project(tenant, project, path)?;
validate_user(actor_user, &format!("{path}.actor_user"))?;
@ -1133,6 +1262,14 @@ fn validate_authenticated_request(
| AuthenticatedCoordinatorRequest::ListNodeDescriptors
| AuthenticatedCoordinatorRequest::ListProcesses
| AuthenticatedCoordinatorRequest::QuotaStatus => Ok(()),
AuthenticatedCoordinatorRequest::ListNodeSummaries { cursor, limit } => {
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
AuthenticatedCoordinatorRequest::ListProcessSummaries { cursor, limit } => {
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 100)
}
AuthenticatedCoordinatorRequest::CreateProject { project, .. }
| AuthenticatedCoordinatorRequest::SelectProject { project } => {
validate_project(project, &format!("{path}.project"))
@ -1188,6 +1325,18 @@ fn validate_authenticated_request(
| AuthenticatedCoordinatorRequest::ListTaskSnapshots { process } => {
validate_process(process, &format!("{path}.process"))
}
AuthenticatedCoordinatorRequest::ListRecentLogs {
process,
task,
limit,
..
} => {
validate_process(process, &format!("{path}.process"))?;
if let Some(task) = task {
validate_task_instance(task, &format!("{path}.task"))?;
}
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
AuthenticatedCoordinatorRequest::RestartTask { process, task, .. }
| AuthenticatedCoordinatorRequest::ResolveTaskFailure { process, task, .. }
| AuthenticatedCoordinatorRequest::JoinTask { process, task } => {
@ -1205,9 +1354,19 @@ fn validate_authenticated_request(
AuthenticatedCoordinatorRequest::ListTaskEvents { process } => {
validate_optional_process(process.as_deref(), &format!("{path}.process"))
}
AuthenticatedCoordinatorRequest::ListArtifacts {
process,
cursor,
limit,
} => {
validate_optional_process(process.as_deref(), &format!("{path}.process"))?;
validate_optional_cursor(cursor.as_deref(), &format!("{path}.cursor"))?;
validate_page_limit(*limit, &format!("{path}.limit"), 200)
}
AuthenticatedCoordinatorRequest::CreateArtifactDownloadLink { artifact, .. }
| AuthenticatedCoordinatorRequest::OpenArtifactDownloadStream { artifact, .. }
| AuthenticatedCoordinatorRequest::RevokeArtifactDownloadLink { artifact, .. } => {
| AuthenticatedCoordinatorRequest::RevokeArtifactDownloadLink { artifact, .. }
| AuthenticatedCoordinatorRequest::GetArtifact { artifact } => {
validate_artifact(artifact, &format!("{path}.artifact"))
}
AuthenticatedCoordinatorRequest::ExportArtifactToNode {
@ -1362,6 +1521,19 @@ fn validate_optional_process(value: Option<&str>, path: &str) -> Result<(), Stri
value.map_or(Ok(()), |value| validate_process(value, path))
}
fn validate_optional_cursor(value: Option<&str>, path: &str) -> Result<(), String> {
value.map_or(Ok(()), |value| validate_external_token(value, path, 256))
}
fn validate_page_limit(value: u32, path: &str, maximum: u32) -> Result<(), String> {
if value == 0 || value > maximum {
return Err(format!(
"malformed pagination limit {path}: expected 1 through {maximum}, received {value}"
));
}
Ok(())
}
fn validate_optional_launch_attempt(value: Option<&str>, path: &str) -> Result<(), String> {
value.map_or(Ok(()), |value| validate_launch_attempt(value, path))
}
@ -1686,6 +1858,12 @@ pub enum AuthenticatedCoordinatorRequest {
ttl_seconds: u64,
},
ListNodeDescriptors,
ListNodeSummaries {
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
RevokeNodeCredential {
node: String,
},
@ -1722,6 +1900,12 @@ pub enum AuthenticatedCoordinatorRequest {
launch_attempt: Option<String>,
},
ListProcesses,
ListProcessSummaries {
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
QuotaStatus,
RestartTask {
process: String,
@ -1766,6 +1950,15 @@ pub enum AuthenticatedCoordinatorRequest {
ListTaskSnapshots {
process: String,
},
ListRecentLogs {
process: String,
#[serde(default)]
task: Option<String>,
#[serde(default)]
after_sequence: Option<u64>,
#[serde(default = "default_log_page_limit")]
limit: u32,
},
JoinTask {
process: String,
task: String,
@ -1776,6 +1969,17 @@ pub enum AuthenticatedCoordinatorRequest {
#[serde(default = "default_download_ttl_seconds")]
ttl_seconds: u64,
},
ListArtifacts {
#[serde(default)]
process: Option<String>,
#[serde(default)]
cursor: Option<String>,
#[serde(default = "default_page_limit")]
limit: u32,
},
GetArtifact {
artifact: String,
},
OpenArtifactDownloadStream {
artifact: String,
max_bytes: u64,
@ -1798,6 +2002,14 @@ fn default_download_ttl_seconds() -> u64 {
900
}
fn default_page_limit() -> u32 {
50
}
fn default_log_page_limit() -> u32 {
100
}
fn default_node_enrollment_ttl_seconds() -> u64 {
900
}

163
crates/clusterflux-coordinator/src/service/protocol/responses.rs
View file

@ -183,6 +183,121 @@ pub struct VirtualProcessStatus {
pub coordinator_epoch: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NodeSummary {
pub id: NodeId,
pub display_name: String,
pub online: bool,
pub stale: bool,
pub last_seen_epoch_seconds: Option<u64>,
pub capabilities: NodeCapabilities,
pub direct_connectivity: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProcessLifecycleState {
Active,
RecentTerminal,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProcessActivityState {
Running,
WaitingForNode,
WaitingForTask,
AwaitingAction,
DebugEpochPartial,
Cancelling,
Completed,
Failed,
Cancelled,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProcessFinalResult {
Completed,
Failed,
Cancelled,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DebugEpochSummary {
pub epoch: u64,
pub command: String,
pub fully_frozen: bool,
pub partially_frozen: bool,
pub fully_resumed: bool,
pub failed: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProcessSummary {
pub process: ProcessId,
pub lifecycle: ProcessLifecycleState,
pub activity: ProcessActivityState,
pub main_wait_state: Option<String>,
pub started_at_epoch_seconds: u64,
pub ended_at_epoch_seconds: Option<u64>,
pub final_result: Option<ProcessFinalResult>,
pub connected_nodes: Vec<NodeId>,
pub current_debug_epoch: Option<DebugEpochSummary>,
pub order_cursor: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ArtifactAvailability {
Available,
NodeOffline,
Unavailable,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ArtifactRetentionState {
NodeRetained,
ExplicitStorage,
Lost,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactSummary {
pub id: ArtifactId,
pub display_path: String,
pub display_name: String,
pub process: ProcessId,
pub producer_task: TaskInstanceId,
pub safe_node: Option<NodeId>,
pub digest: Digest,
pub size_bytes: u64,
pub availability: ArtifactAvailability,
pub downloadable_now: bool,
pub retention_state: ArtifactRetentionState,
pub explicit_storage: bool,
pub order_cursor: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TaskLogStream {
Stdout,
Stderr,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RecentLogEntry {
pub sequence: u64,
pub process: ProcessId,
pub task: TaskInstanceId,
pub stream: TaskLogStream,
pub text: String,
pub server_timestamp_epoch_seconds: u64,
pub truncated: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum SourcePreparationDisposition {
Pending { reason: String },
@ -282,6 +397,11 @@ pub enum CoordinatorResponse {
descriptors: Vec<NodeDescriptor>,
actor: UserId,
},
NodeSummaries {
nodes: Vec<NodeSummary>,
next_cursor: Option<String>,
actor: UserId,
},
NodeCredentialRevoked {
node: NodeId,
tenant: TenantId,
@ -373,6 +493,11 @@ pub enum CoordinatorResponse {
processes: Vec<VirtualProcessStatus>,
actor: UserId,
},
ProcessSummaries {
processes: Vec<ProcessSummary>,
next_cursor: Option<String>,
actor: UserId,
},
QuotaStatus {
tenant: TenantId,
project: ProjectId,
@ -483,6 +608,17 @@ pub enum CoordinatorResponse {
stderr_tail: String,
backpressured: bool,
},
TaskLogChunkRecorded {
process: ProcessId,
task: TaskInstanceId,
sequence: Option<u64>,
next_offset: u64,
},
RecentLogs {
entries: Vec<RecentLogEntry>,
next_sequence: Option<u64>,
history_truncated: bool,
},
VfsMetadataRecorded {
process: ProcessId,
task: TaskInstanceId,
@ -519,6 +655,13 @@ pub enum CoordinatorResponse {
ArtifactDownloadLink {
link: DownloadLink,
},
Artifacts {
artifacts: Vec<ArtifactSummary>,
next_cursor: Option<String>,
},
Artifact {
artifact: ArtifactSummary,
},
ArtifactDownloadLinkRevoked {
link: DownloadLink,
},
@ -543,6 +686,24 @@ pub enum CoordinatorResponse {
artifact_size_bytes: u64,
},
Error {
message: String,
#[serde(flatten)]
error: clusterflux_core::ApiError,
},
}
impl CoordinatorResponse {
pub fn error(request_id: impl Into<String>, message: impl Into<String>) -> Self {
Self::Error {
error: clusterflux_core::ApiError::from_message(request_id, message),
}
}
pub fn service_error(
request_id: impl Into<String>,
error: &crate::service::CoordinatorServiceError,
) -> Self {
Self::Error {
error: error.api_error(request_id),
}
}
}

154
crates/clusterflux-coordinator/src/service/relay.rs
View file

@ -65,6 +65,16 @@ pub struct ArtifactRelayUsage {
pub egress_bytes: u64,
pub abandoned_or_failed_bytes: u64,
pub reserved_bytes: u64,
pub lifetime_ingress_bytes: u64,
pub lifetime_egress_bytes: u64,
pub lifetime_abandoned_or_failed_bytes: u64,
pub completed_transfers: u64,
pub failed_transfers: u64,
pub cancelled_transfers: u64,
pub expired_transfers: u64,
pub tracked_scopes: usize,
pub max_active_global: usize,
pub max_tracked_scopes: usize,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
@ -77,6 +87,20 @@ pub struct ArtifactRelayDurableState {
pub ingress_used: u64,
pub egress_used: u64,
pub abandoned_or_failed_used: u64,
#[serde(default)]
pub lifetime_ingress_bytes: u64,
#[serde(default)]
pub lifetime_egress_bytes: u64,
#[serde(default)]
pub lifetime_abandoned_or_failed_bytes: u64,
#[serde(default)]
pub completed_transfers: u64,
#[serde(default)]
pub failed_transfers: u64,
#[serde(default)]
pub cancelled_transfers: u64,
#[serde(default)]
pub expired_transfers: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@ -145,6 +169,13 @@ pub(super) struct ArtifactRelayLedger {
ingress_used: u64,
egress_used: u64,
abandoned_or_failed_used: u64,
lifetime_ingress_bytes: u64,
lifetime_egress_bytes: u64,
lifetime_abandoned_or_failed_bytes: u64,
completed_transfers: u64,
failed_transfers: u64,
cancelled_transfers: u64,
expired_transfers: u64,
}
impl Default for ArtifactRelayLedger {
@ -165,6 +196,13 @@ impl ArtifactRelayLedger {
ingress_used: 0,
egress_used: 0,
abandoned_or_failed_used: 0,
lifetime_ingress_bytes: 0,
lifetime_egress_bytes: 0,
lifetime_abandoned_or_failed_bytes: 0,
completed_transfers: 0,
failed_transfers: 0,
cancelled_transfers: 0,
expired_transfers: 0,
}
}
@ -218,6 +256,13 @@ impl ArtifactRelayLedger {
ingress_used: state.ingress_used,
egress_used: state.egress_used,
abandoned_or_failed_used: state.abandoned_or_failed_used,
lifetime_ingress_bytes: state.lifetime_ingress_bytes,
lifetime_egress_bytes: state.lifetime_egress_bytes,
lifetime_abandoned_or_failed_bytes: state.lifetime_abandoned_or_failed_bytes,
completed_transfers: state.completed_transfers,
failed_transfers: state.failed_transfers,
cancelled_transfers: state.cancelled_transfers,
expired_transfers: state.expired_transfers,
}
}
@ -260,6 +305,13 @@ impl ArtifactRelayLedger {
ingress_used: self.ingress_used,
egress_used: self.egress_used,
abandoned_or_failed_used: self.abandoned_or_failed_used,
lifetime_ingress_bytes: self.lifetime_ingress_bytes,
lifetime_egress_bytes: self.lifetime_egress_bytes,
lifetime_abandoned_or_failed_bytes: self.lifetime_abandoned_or_failed_bytes,
completed_transfers: self.completed_transfers,
failed_transfers: self.failed_transfers,
cancelled_transfers: self.cancelled_transfers,
expired_transfers: self.expired_transfers,
}
}
@ -491,9 +543,11 @@ impl ArtifactRelayLedger {
if ingress {
reservation.ingress_bytes = reservation.ingress_bytes.saturating_add(bytes);
self.ingress_used = self.ingress_used.saturating_add(bytes);
self.lifetime_ingress_bytes = self.lifetime_ingress_bytes.saturating_add(bytes);
} else {
reservation.egress_bytes = reservation.egress_bytes.saturating_add(bytes);
self.egress_used = self.egress_used.saturating_add(bytes);
self.lifetime_egress_bytes = self.lifetime_egress_bytes.saturating_add(bytes);
}
let project_key = (
reservation.scope.tenant.clone(),
@ -554,10 +608,29 @@ impl ArtifactRelayLedger {
pub(super) fn finish(&mut self, key: &str, reason: RelayFinishReason) {
if let Some(reservation) = self.reservations.remove(key) {
if reason != RelayFinishReason::Completed {
let abandoned_or_failed_bytes = reservation
.ingress_bytes
.saturating_add(reservation.egress_bytes);
self.abandoned_or_failed_used = self
.abandoned_or_failed_used
.saturating_add(reservation.ingress_bytes)
.saturating_add(reservation.egress_bytes);
.saturating_add(abandoned_or_failed_bytes);
self.lifetime_abandoned_or_failed_bytes = self
.lifetime_abandoned_or_failed_bytes
.saturating_add(abandoned_or_failed_bytes);
}
match reason {
RelayFinishReason::Completed => {
self.completed_transfers = self.completed_transfers.saturating_add(1);
}
RelayFinishReason::Failed => {
self.failed_transfers = self.failed_transfers.saturating_add(1);
}
RelayFinishReason::Cancelled => {
self.cancelled_transfers = self.cancelled_transfers.saturating_add(1);
}
RelayFinishReason::Expired => {
self.expired_transfers = self.expired_transfers.saturating_add(1);
}
}
}
}
@ -580,6 +653,18 @@ impl ArtifactRelayLedger {
ingress_bytes: self.ingress_used,
egress_bytes: self.egress_used,
abandoned_or_failed_bytes: self.abandoned_or_failed_used,
lifetime_ingress_bytes: self.lifetime_ingress_bytes,
lifetime_egress_bytes: self.lifetime_egress_bytes,
lifetime_abandoned_or_failed_bytes: self.lifetime_abandoned_or_failed_bytes,
completed_transfers: self.completed_transfers,
failed_transfers: self.failed_transfers,
cancelled_transfers: self.cancelled_transfers,
expired_transfers: self.expired_transfers,
tracked_scopes: self.project_used.len()
+ self.tenant_used.len()
+ self.account_used.len(),
max_active_global: self.configuration.max_active_global,
max_tracked_scopes: self.configuration.max_tracked_scopes,
reserved_bytes: self
.reservations
.values()
@ -588,3 +673,68 @@ impl ArtifactRelayLedger {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn lifetime_metrics_survive_period_rollover_and_durable_round_trip() {
let mut ledger = ArtifactRelayLedger::new(ArtifactRelayConfiguration::unlimited());
ledger
.reserve(
"transfer".to_owned(),
TenantId::from("tenant"),
ProjectId::from("project"),
UserId::from("user"),
16,
16,
100,
1,
)
.unwrap();
ledger.charge_ingress("transfer", 24, 1).unwrap();
ledger.charge_egress("transfer", 24, 1).unwrap();
ledger.finish("transfer", RelayFinishReason::Completed);
let usage = ledger.usage();
assert_eq!(usage.lifetime_ingress_bytes, 24);
assert_eq!(usage.lifetime_egress_bytes, 24);
assert_eq!(usage.completed_transfers, 1);
ledger.prepare_period(u64::MAX);
let usage = ledger.usage();
assert_eq!(usage.ingress_bytes, 0);
assert_eq!(usage.egress_bytes, 0);
assert_eq!(usage.lifetime_ingress_bytes, 24);
assert_eq!(usage.lifetime_egress_bytes, 24);
let restored = ArtifactRelayLedger::from_durable(
ArtifactRelayConfiguration::unlimited(),
ledger.durable_state(),
);
let usage = restored.usage();
assert_eq!(usage.lifetime_ingress_bytes, 24);
assert_eq!(usage.lifetime_egress_bytes, 24);
assert_eq!(usage.completed_transfers, 1);
}
#[test]
fn older_durable_relay_state_defaults_new_metrics() {
let state: ArtifactRelayDurableState = serde_json::from_value(serde_json::json!({
"reservations": {},
"period": 1,
"project_used": [],
"tenant_used": [],
"account_used": [],
"ingress_used": 3,
"egress_used": 4,
"abandoned_or_failed_used": 2
}))
.unwrap();
assert_eq!(state.lifetime_ingress_bytes, 0);
assert_eq!(state.lifetime_egress_bytes, 0);
assert_eq!(state.completed_transfers, 0);
}
}

96
crates/clusterflux-coordinator/src/service/routing.rs
View file

@ -298,6 +298,13 @@ impl CoordinatorService {
project,
actor_user,
} => self.handle_list_node_descriptors(tenant, project, actor_user),
CoordinatorRequest::ListNodeSummaries {
tenant,
project,
actor_user,
cursor,
limit,
} => self.handle_list_node_summaries(tenant, project, actor_user, cursor, limit),
CoordinatorRequest::RevokeNodeCredential {
tenant,
project,
@ -421,6 +428,13 @@ impl CoordinatorService {
project,
actor_user,
} => self.handle_list_processes(tenant, project, actor_user),
CoordinatorRequest::ListProcessSummaries {
tenant,
project,
actor_user,
cursor,
limit,
} => self.handle_list_process_summaries(tenant, project, actor_user, cursor, limit),
CoordinatorRequest::QuotaStatus {
tenant,
project,
@ -466,7 +480,8 @@ impl CoordinatorService {
CoordinatorRequest::PollDebugCommand { .. }
| CoordinatorRequest::ReportDebugState { .. }
| CoordinatorRequest::ReportDebugProbeHit { .. } => self.reject_unsigned_node_request(),
CoordinatorRequest::ReportTaskLog { .. } => self.reject_unsigned_node_request(),
CoordinatorRequest::ReportTaskLog { .. }
| CoordinatorRequest::ReportTaskLogChunk { .. } => self.reject_unsigned_node_request(),
CoordinatorRequest::ReportVfsMetadata { .. } => self.reject_unsigned_node_request(),
CoordinatorRequest::TaskCompleted { .. } => self.reject_unsigned_node_request(),
CoordinatorRequest::ListTaskEvents {
@ -481,6 +496,23 @@ impl CoordinatorService {
actor_user,
process,
} => self.handle_list_task_snapshots(tenant, project, actor_user, process),
CoordinatorRequest::ListRecentLogs {
tenant,
project,
actor_user,
process,
task,
after_sequence,
limit,
} => self.handle_list_recent_logs(
tenant,
project,
actor_user,
process,
task,
after_sequence,
limit,
),
CoordinatorRequest::JoinTask {
tenant,
project,
@ -527,6 +559,20 @@ impl CoordinatorService {
max_bytes,
ttl_seconds,
),
CoordinatorRequest::ListArtifacts {
tenant,
project,
actor_user,
process,
cursor,
limit,
} => self.handle_list_artifacts(tenant, project, actor_user, process, cursor, limit),
CoordinatorRequest::GetArtifact {
tenant,
project,
actor_user,
artifact,
} => self.handle_get_artifact(tenant, project, actor_user, artifact),
CoordinatorRequest::OpenArtifactDownloadStream {
tenant,
project,
@ -696,6 +742,14 @@ impl CoordinatorService {
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
),
AuthenticatedCoordinatorRequest::ListNodeSummaries { cursor, limit } => self
.handle_list_node_summaries(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
cursor,
limit,
),
AuthenticatedCoordinatorRequest::RevokeNodeCredential { node } => self
.handle_revoke_node_credential(
context.tenant.as_str().to_owned(),
@ -747,6 +801,14 @@ impl CoordinatorService {
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
),
AuthenticatedCoordinatorRequest::ListProcessSummaries { cursor, limit } => self
.handle_list_process_summaries(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
cursor,
limit,
),
AuthenticatedCoordinatorRequest::QuotaStatus => self.handle_quota_status(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
@ -802,6 +864,20 @@ impl CoordinatorService {
actor.as_str().to_owned(),
process,
),
AuthenticatedCoordinatorRequest::ListRecentLogs {
process,
task,
after_sequence,
limit,
} => self.handle_list_recent_logs(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
process,
task,
after_sequence,
limit,
),
AuthenticatedCoordinatorRequest::JoinTask { process, task } => self.handle_join_task(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
@ -821,6 +897,24 @@ impl CoordinatorService {
max_bytes,
ttl_seconds,
),
AuthenticatedCoordinatorRequest::ListArtifacts {
process,
cursor,
limit,
} => self.handle_list_artifacts(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
process,
cursor,
limit,
),
AuthenticatedCoordinatorRequest::GetArtifact { artifact } => self.handle_get_artifact(
context.tenant.as_str().to_owned(),
context.project.as_str().to_owned(),
actor.as_str().to_owned(),
artifact,
),
AuthenticatedCoordinatorRequest::OpenArtifactDownloadStream {
artifact,
max_bytes,

30
crates/clusterflux-coordinator/src/service/signed_nodes.rs
View file

@ -253,6 +253,29 @@ impl CoordinatorService {
stderr_truncated,
backpressured,
),
CoordinatorRequest::ReportTaskLogChunk {
tenant,
project,
process,
node,
task,
stream,
offset,
source_bytes,
text,
truncated,
} => self.handle_report_task_log_chunk(
tenant,
project,
process,
node,
task,
stream,
offset,
source_bytes,
text,
truncated,
),
CoordinatorRequest::ReportVfsMetadata {
tenant,
project,
@ -346,6 +369,7 @@ fn signed_node_request_kind(
CoordinatorRequest::ReportDebugState { .. } => Ok("report_debug_state"),
CoordinatorRequest::ReportDebugProbeHit { .. } => Ok("report_debug_probe_hit"),
CoordinatorRequest::ReportTaskLog { .. } => Ok("report_task_log"),
CoordinatorRequest::ReportTaskLogChunk { .. } => Ok("report_task_log_chunk"),
CoordinatorRequest::ReportVfsMetadata { .. } => Ok("report_vfs_metadata"),
CoordinatorRequest::TaskCompleted { .. } => Ok("task_completed"),
_ => Err(CoordinatorError::Unauthorized(
@ -441,6 +465,12 @@ fn signed_node_request_scope(
node,
..
}
| CoordinatorRequest::ReportTaskLogChunk {
tenant,
project,
node,
..
}
| CoordinatorRequest::ReportVfsMetadata {
tenant,
project,

500
crates/clusterflux-coordinator/src/service/summaries.rs Normal file
View file

@ -0,0 +1,500 @@
use std::collections::BTreeSet;
use std::time::Instant;
use clusterflux_core::{
ArtifactId, ArtifactMetadata, NodeId, ProcessId, ProjectId, TenantId, UserId,
};
use super::keys::{process_control_key, ProcessControlKey};
use super::{
ArtifactAvailability, ArtifactRetentionState, ArtifactSummary, CoordinatorResponse,
CoordinatorService, CoordinatorServiceError, DebugAcknowledgementState, DebugEpochSummary,
NodeSummary, ProcessActivityState, ProcessFinalResult, ProcessLifecycleState, ProcessSummary,
TaskAttemptState,
};
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct StoredProcessSummary {
pub(super) started_at_epoch_seconds: u64,
pub(super) ended_at_epoch_seconds: Option<u64>,
pub(super) final_result: Option<ProcessFinalResult>,
pub(super) connected_nodes: Vec<NodeId>,
pub(super) order: u64,
}
impl CoordinatorService {
pub(super) fn handle_list_node_summaries(
&mut self,
tenant: String,
project: String,
actor_user: String,
cursor: Option<String>,
limit: u32,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let actor = UserId::new(actor_user);
let cursor = cursor.as_deref();
let mut nodes = self
.node_descriptors
.iter()
.filter(|(scope, descriptor)| {
scope.tenant == tenant
&& scope.project == project
&& cursor.is_none_or(|cursor| descriptor.id.as_str() > cursor)
})
.map(|(scope, descriptor)| {
let online = self.node_is_live(scope);
NodeSummary {
id: descriptor.id.clone(),
display_name: descriptor.id.as_str().to_owned(),
online,
stale: !online,
last_seen_epoch_seconds: self.node_last_seen_epoch_seconds.get(scope).copied(),
capabilities: descriptor.capabilities.clone(),
direct_connectivity: descriptor.direct_connectivity,
}
})
.collect::<Vec<_>>();
nodes.sort_by(|left, right| left.id.cmp(&right.id));
let has_more = nodes.len() > limit as usize;
nodes.truncate(limit as usize);
let next_cursor = has_more
.then(|| nodes.last().map(|node| node.id.as_str().to_owned()))
.flatten();
Ok(CoordinatorResponse::NodeSummaries {
nodes,
next_cursor,
actor,
})
}
pub(super) fn record_process_started(
&mut self,
tenant: &TenantId,
project: &ProjectId,
process: &ProcessId,
now_epoch_seconds: u64,
) {
let key = process_control_key(tenant, project, process);
if let Some(logs) = self.recent_logs.get_mut(&(tenant.clone(), project.clone())) {
logs.retain(|entry| &entry.process != process);
if logs.is_empty() {
self.recent_logs.remove(&(tenant.clone(), project.clone()));
}
}
self.recent_log_dropped_through.remove(&key);
self.recent_log_offsets
.retain(|(entry_tenant, entry_project, entry_process, _, _), _| {
entry_tenant != tenant || entry_project != project || entry_process != process
});
self.process_summary_order
.retain(|retained| retained != &key);
self.evict_process_summaries_for_project(tenant, project);
self.evict_process_summaries_total();
let order = self.next_process_summary_order;
self.next_process_summary_order = self.next_process_summary_order.saturating_add(1);
self.process_summaries.insert(
key.clone(),
StoredProcessSummary {
started_at_epoch_seconds: now_epoch_seconds,
ended_at_epoch_seconds: None,
final_result: None,
connected_nodes: Vec::new(),
order,
},
);
self.process_summary_order.push_back(key);
}
pub(super) fn record_process_terminal(
&mut self,
tenant: &TenantId,
project: &ProjectId,
process: &ProcessId,
final_result: ProcessFinalResult,
now_epoch_seconds: u64,
) {
let key = process_control_key(tenant, project, process);
let connected_nodes = self
.coordinator
.active_process(tenant, project, process)
.map(|active| active.connected_nodes.iter().cloned().collect())
.unwrap_or_default();
let order = self.next_process_summary_order;
let entry = self
.process_summaries
.entry(key.clone())
.or_insert_with(|| {
self.next_process_summary_order = self.next_process_summary_order.saturating_add(1);
self.process_summary_order.push_back(key.clone());
StoredProcessSummary {
started_at_epoch_seconds: now_epoch_seconds,
ended_at_epoch_seconds: None,
final_result: None,
connected_nodes: Vec::new(),
order,
}
});
entry.ended_at_epoch_seconds = Some(now_epoch_seconds);
entry.final_result = Some(final_result);
entry.connected_nodes = connected_nodes;
}
pub(super) fn handle_list_process_summaries(
&mut self,
tenant: String,
project: String,
actor_user: String,
cursor: Option<String>,
limit: u32,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let actor = UserId::new(actor_user);
let cursor = parse_order_cursor(cursor.as_deref(), "process")?;
let mut stored = self
.process_summaries
.iter()
.filter(|((entry_tenant, entry_project, _), summary)| {
entry_tenant == &tenant
&& entry_project == &project
&& cursor.is_none_or(|cursor| summary.order < cursor)
})
.map(|(key, summary)| (key.clone(), summary.clone()))
.collect::<Vec<_>>();
stored.sort_by(|(_, left), (_, right)| right.order.cmp(&left.order));
let has_more = stored.len() > limit as usize;
stored.truncate(limit as usize);
let processes = stored
.into_iter()
.map(|(key, stored)| self.process_summary_from_stored(&key, stored))
.collect::<Vec<_>>();
let next_cursor = has_more
.then(|| processes.last().map(|process| process.order_cursor.clone()))
.flatten();
Ok(CoordinatorResponse::ProcessSummaries {
processes,
next_cursor,
actor,
})
}
fn process_summary_from_stored(
&self,
key: &ProcessControlKey,
stored: StoredProcessSummary,
) -> ProcessSummary {
let (tenant, project, process) = key;
let active = self.coordinator.active_process(tenant, project, process);
let process_key = process_control_key(tenant, project, process);
let main_wait_state = active.and_then(|_| {
if self.pending_task_launches.iter().any(|pending| {
&pending.tenant == tenant
&& &pending.project == project
&& &pending.process == process
}) {
Some("waiting_for_node".to_owned())
} else if self
.main_runtime
.is_waiting_for_task(tenant, project, process)
{
Some("waiting_for_task".to_owned())
} else {
None
}
});
let current_debug_epoch = self.debug_epoch_summary(&process_key);
let awaiting_action = self.task_attempts.iter().any(
|((attempt_tenant, attempt_project, attempt_process, _), attempts)| {
attempt_tenant == tenant
&& attempt_project == project
&& attempt_process == process
&& attempts.iter().any(|attempt| {
attempt.current && attempt.state == TaskAttemptState::FailedAwaitingAction
})
},
);
let activity = if let Some(result) = &stored.final_result {
match result {
ProcessFinalResult::Completed => ProcessActivityState::Completed,
ProcessFinalResult::Failed => ProcessActivityState::Failed,
ProcessFinalResult::Cancelled => ProcessActivityState::Cancelled,
}
} else if self.process_cancellations.contains(&process_key) {
ProcessActivityState::Cancelling
} else if current_debug_epoch
.as_ref()
.is_some_and(|epoch| epoch.partially_frozen)
{
ProcessActivityState::DebugEpochPartial
} else if awaiting_action {
ProcessActivityState::AwaitingAction
} else {
match main_wait_state.as_deref() {
Some("waiting_for_node") => ProcessActivityState::WaitingForNode,
Some("waiting_for_task") => ProcessActivityState::WaitingForTask,
_ => ProcessActivityState::Running,
}
};
let connected_nodes = active
.map(|active| active.connected_nodes.iter().cloned().collect())
.unwrap_or(stored.connected_nodes);
ProcessSummary {
process: process.clone(),
lifecycle: if active.is_some() {
ProcessLifecycleState::Active
} else {
ProcessLifecycleState::RecentTerminal
},
activity,
main_wait_state,
started_at_epoch_seconds: stored.started_at_epoch_seconds,
ended_at_epoch_seconds: stored.ended_at_epoch_seconds,
final_result: stored.final_result,
connected_nodes,
current_debug_epoch,
order_cursor: format!("process:{}", stored.order),
}
}
fn debug_epoch_summary(&self, key: &ProcessControlKey) -> Option<DebugEpochSummary> {
let runtime = self.debug_epoch_runtime.get(key)?;
let acknowledgements = runtime.acknowledgements.values().collect::<Vec<_>>();
let all_acknowledged = !runtime.expected.is_empty()
&& runtime
.expected
.iter()
.all(|participant| runtime.acknowledgements.contains_key(participant));
let fully_frozen = runtime.command == "freeze"
&& all_acknowledged
&& acknowledgements
.iter()
.all(|ack| ack.state == DebugAcknowledgementState::Frozen);
let freeze_deadline_elapsed =
runtime.command == "freeze" && Instant::now() >= runtime.deadline;
let frozen_count = acknowledgements
.iter()
.filter(|ack| ack.state == DebugAcknowledgementState::Frozen)
.count();
let partially_frozen = freeze_deadline_elapsed && frozen_count > 0 && !fully_frozen;
let fully_resumed = runtime.command == "resume"
&& all_acknowledged
&& acknowledgements
.iter()
.all(|ack| ack.state == DebugAcknowledgementState::Running);
let failed = acknowledgements
.iter()
.any(|ack| ack.state == DebugAcknowledgementState::Failed)
|| (freeze_deadline_elapsed
&& runtime
.expected
.iter()
.any(|participant| !runtime.acknowledgements.contains_key(participant)));
Some(DebugEpochSummary {
epoch: runtime.epoch,
command: runtime.command.clone(),
fully_frozen,
partially_frozen,
fully_resumed,
failed,
})
}
pub(super) fn handle_list_artifacts(
&mut self,
tenant: String,
project: String,
actor_user: String,
process: Option<String>,
cursor: Option<String>,
limit: u32,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let _actor = UserId::new(actor_user);
let process = process.map(ProcessId::new);
if let Some(process) = &process {
self.authorize_task_event_process_scope(&tenant, &project, process)?;
}
let cursor = parse_order_cursor(cursor.as_deref(), "artifact")?;
let mut metadata = self
.artifact_registry
.metadata_for_project(&tenant, &project)
.filter(|metadata| {
process
.as_ref()
.is_none_or(|process| &metadata.process == process)
&& cursor.is_none_or(|cursor| metadata.flushed_epoch < cursor)
})
.cloned()
.collect::<Vec<_>>();
metadata.sort_by(|left, right| right.flushed_epoch.cmp(&left.flushed_epoch));
let has_more = metadata.len() > limit as usize;
metadata.truncate(limit as usize);
let artifacts = metadata
.into_iter()
.map(|metadata| self.artifact_summary(metadata))
.collect::<Vec<_>>();
let next_cursor = has_more
.then(|| {
artifacts
.last()
.map(|artifact| artifact.order_cursor.clone())
})
.flatten();
Ok(CoordinatorResponse::Artifacts {
artifacts,
next_cursor,
})
}
pub(super) fn handle_get_artifact(
&mut self,
tenant: String,
project: String,
actor_user: String,
artifact: String,
) -> Result<CoordinatorResponse, CoordinatorServiceError> {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let _actor = UserId::new(actor_user);
let artifact = ArtifactId::new(artifact);
let metadata = self
.artifact_registry
.metadata(&tenant, &project, &artifact)
.cloned()
.ok_or(clusterflux_core::DownloadError::NotFound)?;
Ok(CoordinatorResponse::Artifact {
artifact: self.artifact_summary(metadata),
})
}
fn artifact_summary(&self, metadata: ArtifactMetadata) -> ArtifactSummary {
let live_retaining_nodes = metadata
.retaining_nodes
.iter()
.filter(|node| {
self.node_is_live(&crate::NodeScopeKey::from_refs(
&metadata.tenant,
&metadata.project,
node,
))
})
.cloned()
.collect::<BTreeSet<_>>();
let safe_node = live_retaining_nodes
.iter()
.next()
.cloned()
.or_else(|| metadata.retaining_nodes.iter().next().cloned());
let explicit_storage = !metadata.explicit_locations.is_empty();
let downloadable_now = !live_retaining_nodes.is_empty() || explicit_storage;
let availability = if downloadable_now {
ArtifactAvailability::Available
} else if !metadata.retaining_nodes.is_empty() {
ArtifactAvailability::NodeOffline
} else {
ArtifactAvailability::Unavailable
};
let retention_state = if explicit_storage {
ArtifactRetentionState::ExplicitStorage
} else if !metadata.retaining_nodes.is_empty() {
ArtifactRetentionState::NodeRetained
} else {
ArtifactRetentionState::Lost
};
let display_suffix = metadata.id.as_str().replace(':', "/");
let display_path = format!("/vfs/artifacts/{display_suffix}");
let display_name = display_suffix
.rsplit('/')
.next()
.unwrap_or(metadata.id.as_str())
.to_owned();
ArtifactSummary {
id: metadata.id,
display_path,
display_name,
process: metadata.process,
producer_task: metadata.producer_task,
safe_node,
digest: metadata.digest,
size_bytes: metadata.size,
availability,
downloadable_now,
retention_state,
explicit_storage,
order_cursor: format!("artifact:{}", metadata.flushed_epoch),
}
}
fn evict_process_summaries_for_project(&mut self, tenant: &TenantId, project: &ProjectId) {
while self
.process_summaries
.keys()
.filter(|(entry_tenant, entry_project, _)| {
entry_tenant == tenant && entry_project == project
})
.count()
>= super::MAX_RECENT_PROCESS_SUMMARIES_PER_PROJECT
{
let candidate = self.process_summary_order.iter().find(|key| {
&key.0 == tenant
&& &key.1 == project
&& self
.process_summaries
.get(*key)
.is_some_and(|summary| summary.final_result.is_some())
});
let Some(candidate) = candidate.cloned() else {
break;
};
self.process_summaries.remove(&candidate);
self.recent_log_dropped_through.remove(&candidate);
self.process_summary_order
.retain(|retained| retained != &candidate);
}
}
fn evict_process_summaries_total(&mut self) {
while self.process_summaries.len() >= super::MAX_RECENT_PROCESS_SUMMARIES_TOTAL {
let candidate = self.process_summary_order.iter().find(|key| {
self.process_summaries
.get(*key)
.is_some_and(|summary| summary.final_result.is_some())
});
let Some(candidate) = candidate.cloned() else {
break;
};
self.process_summaries.remove(&candidate);
self.recent_log_dropped_through.remove(&candidate);
self.process_summary_order
.retain(|retained| retained != &candidate);
}
}
}
fn parse_order_cursor(
cursor: Option<&str>,
expected_kind: &str,
) -> Result<Option<u64>, CoordinatorServiceError> {
cursor
.map(|cursor| {
let (kind, order) = cursor.split_once(':').ok_or_else(|| {
CoordinatorServiceError::Protocol(format!(
"invalid {expected_kind} pagination cursor"
))
})?;
if kind != expected_kind {
return Err(CoordinatorServiceError::Protocol(format!(
"invalid {expected_kind} pagination cursor"
)));
}
order.parse::<u64>().map_err(|_| {
CoordinatorServiceError::Protocol(format!(
"invalid {expected_kind} pagination cursor"
))
})
})
.transpose()
}

75
crates/clusterflux-coordinator/src/service/tcp.rs
View file

@ -79,17 +79,15 @@ impl CoordinatorService {
continue;
}
let response = match decode_wire_request(&line) {
Ok(request) => match authorize_client_request(&request, authority_mode)
.and_then(|()| self.handle_request(request))
{
Ok(response) => response,
Err(err) => CoordinatorResponse::Error {
message: err.to_string(),
},
},
Err(err) => CoordinatorResponse::Error {
message: err.to_string(),
},
Ok((request_id, request)) => {
match authorize_client_request(&request, authority_mode)
.and_then(|()| self.handle_request(request))
{
Ok(response) => response,
Err(err) => CoordinatorResponse::service_error(request_id, &err),
}
}
Err(err) => CoordinatorResponse::service_error(wire_request_id_hint(&line), &err),
};
serde_json::to_writer(&mut writer, &response)?;
writer.write_all(b"\n")?;
@ -114,25 +112,19 @@ fn handle_shared_stream(
continue;
}
let response = match decode_wire_request(&line) {
Ok(request) => match authorize_client_request(&request, authority_mode) {
Ok((request_id, request)) => match authorize_client_request(&request, authority_mode) {
Ok(()) => match service.lock() {
Ok(mut service) => match service.handle_request(request) {
Ok(response) => response,
Err(err) => CoordinatorResponse::Error {
message: err.to_string(),
},
},
Err(_) => CoordinatorResponse::Error {
message: "coordinator service lock poisoned".to_owned(),
Err(err) => CoordinatorResponse::service_error(request_id, &err),
},
Err(_) => {
CoordinatorResponse::error(request_id, "coordinator service lock poisoned")
}
},
Err(err) => CoordinatorResponse::Error {
message: err.to_string(),
},
},
Err(err) => CoordinatorResponse::Error {
message: err.to_string(),
Err(err) => CoordinatorResponse::service_error(request_id, &err),
},
Err(err) => CoordinatorResponse::service_error(wire_request_id_hint(&line), &err),
};
serde_json::to_writer(&mut writer, &response)?;
writer.write_all(b"\n")?;
@ -146,12 +138,27 @@ pub fn bind_listener(addr: &str) -> Result<(TcpListener, SocketAddr), Coordinato
Ok((listener, addr))
}
fn decode_wire_request(line: &str) -> Result<CoordinatorRequest, CoordinatorServiceError> {
fn decode_wire_request(
line: &str,
) -> Result<(String, CoordinatorRequest), CoordinatorServiceError> {
serde_json::from_str::<super::CoordinatorWireRequest>(line)?
.into_request()
.into_parts()
.map_err(CoordinatorServiceError::Protocol)
}
fn wire_request_id_hint(line: &str) -> String {
serde_json::from_str::<serde_json::Value>(line)
.ok()
.and_then(|value| {
value
.get("request_id")
.and_then(|value| value.as_str())
.map(str::to_owned)
})
.filter(|request_id| clusterflux_core::RequestId::try_new(request_id.clone()).is_ok())
.unwrap_or_else(|| "unavailable".to_owned())
}
fn authorize_client_request(
request: &CoordinatorRequest,
authority_mode: ClientAuthorityMode,
@ -180,10 +187,11 @@ fn authorize_client_request(
}
| CoordinatorRequest::AdminStatus { .. }
| CoordinatorRequest::SuspendTenant { .. } => Ok(()),
_ => Err(CoordinatorServiceError::Protocol(
_ => Err(crate::CoordinatorError::Unauthorized(
"strict Core Client authority requires an authenticated CLI session, signed Agent, signed Node, enrollment grant exchange, or admin credential; request-body identity fields are not authority"
.to_owned(),
)),
)
.into()),
}
}
@ -253,16 +261,19 @@ mod transport_boundary_tests {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let CoordinatorResponse::Error { message } =
let CoordinatorResponse::Error { error } =
serde_json::from_str::<CoordinatorResponse>(&line).unwrap()
else {
panic!("malformed identifier request unexpectedly succeeded");
};
assert!(
message.contains("malformed external identifier")
&& message.contains("request.request.process"),
"unexpected malformed identifier response: {message}"
error.message.contains("malformed external identifier")
&& error.message.contains("request.request.process"),
"unexpected malformed identifier response: {}",
error.message
);
assert_eq!(error.request_id, format!("malformed-{index}"));
assert_eq!(error.code, clusterflux_core::ApiErrorCode::ValidationError);
let valid = coordinator_wire_request(
format!("healthy-{index}"),

714
crates/clusterflux-coordinator/src/service/tests.rs
View file

@ -829,6 +829,7 @@ fn signed_node_request_auto(request: CoordinatorRequest) -> CoordinatorRequest {
| CoordinatorRequest::ReportDebugState { node, .. }
| CoordinatorRequest::ReportDebugProbeHit { node, .. }
| CoordinatorRequest::ReportTaskLog { node, .. }
| CoordinatorRequest::ReportTaskLogChunk { node, .. }
| CoordinatorRequest::ReportVfsMetadata { node, .. }
| CoordinatorRequest::TaskCompleted { node, .. } => node.clone(),
CoordinatorRequest::RequestRendezvous { source, .. } => source.node.to_string(),
@ -874,6 +875,9 @@ fn signed_node_request_auto_with_private_key(
(node.clone(), "report_debug_probe_hit")
}
CoordinatorRequest::ReportTaskLog { node, .. } => (node.clone(), "report_task_log"),
CoordinatorRequest::ReportTaskLogChunk { node, .. } => {
(node.clone(), "report_task_log_chunk")
}
CoordinatorRequest::ReportVfsMetadata { node, .. } => (node.clone(), "report_vfs_metadata"),
CoordinatorRequest::TaskCompleted { node, .. } => (node.clone(), "task_completed"),
_ => panic!("test helper only signs node-originated requests"),
@ -7850,12 +7854,16 @@ fn strict_service_stream_rejects_body_authority_and_accepts_cli_session() {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let CoordinatorResponse::Error { message } =
let CoordinatorResponse::Error { error } =
serde_json::from_str::<CoordinatorResponse>(&line).unwrap()
else {
panic!("expected strict body-authority denial");
};
assert!(message.contains("request-body identity fields are not authority"));
assert!(error
.message
.contains("request-body identity fields are not authority"));
assert_eq!(error.request_id, "strict-stream-forged");
assert_eq!(error.code, clusterflux_core::ApiErrorCode::Forbidden);
write_coordinator_wire_request(
&mut stream,
@ -7911,10 +7919,14 @@ fn service_stream_rejects_invalid_versioned_envelope_metadata() {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let response = serde_json::from_str::<CoordinatorResponse>(&line).unwrap();
let CoordinatorResponse::Error { message } = response else {
let CoordinatorResponse::Error { error } = response else {
panic!("expected invalid wire envelope response");
};
assert!(message.contains("operation attach_node does not match payload operation ping"));
assert!(error
.message
.contains("operation attach_node does not match payload operation ping"));
assert_eq!(error.request_id, "bad-operation");
assert_eq!(error.code, clusterflux_core::ApiErrorCode::ValidationError);
stream.shutdown(std::net::Shutdown::Both).unwrap();
server.join().unwrap();
@ -8011,3 +8023,697 @@ fn coordinator_generates_and_bounds_node_enrollment_grants() {
assert_ne!(first_grant, second_grant);
assert_eq!(expires_at_epoch_seconds, 100 + 15 * 60);
}
#[test]
fn web_process_summaries_are_scoped_paginated_and_retain_authoritative_terminal_state() {
let mut service = CoordinatorService::new(7);
for (tenant, project, user, secret) in [
("tenant-a", "project-a", "user-a", "session-a"),
("tenant-b", "project-b", "user-b", "session-b"),
] {
service
.issue_cli_session(
TenantId::from(tenant),
ProjectId::from(project),
UserId::from(user),
secret,
None,
)
.unwrap();
}
service.set_server_time(100);
service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "process-one".to_owned(),
restart: false,
},
})
.unwrap();
let CoordinatorResponse::ProcessSummaries {
processes,
next_cursor,
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: None,
limit: 1,
},
})
.unwrap()
else {
panic!("expected process summaries");
};
assert_eq!(processes.len(), 1);
assert_eq!(processes[0].process, ProcessId::from("process-one"));
assert_eq!(processes[0].lifecycle, ProcessLifecycleState::Active);
assert_eq!(processes[0].activity, ProcessActivityState::Running);
assert_eq!(processes[0].started_at_epoch_seconds, 100);
assert!(next_cursor.is_none());
let CoordinatorResponse::ProcessSummaries { processes, .. } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-b".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: None,
limit: 10,
},
})
.unwrap()
else {
panic!("expected scoped process summaries");
};
assert!(processes.is_empty());
service.set_server_time(120);
service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::AbortProcess {
process: "process-one".to_owned(),
launch_attempt: None,
},
})
.unwrap();
let CoordinatorResponse::ProcessSummaries { processes, .. } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: None,
limit: 10,
},
})
.unwrap()
else {
panic!("expected terminal process summary");
};
assert_eq!(
processes[0].lifecycle,
ProcessLifecycleState::RecentTerminal
);
assert_eq!(processes[0].activity, ProcessActivityState::Cancelled);
assert_eq!(
processes[0].final_result,
Some(ProcessFinalResult::Cancelled)
);
assert_eq!(processes[0].ended_at_epoch_seconds, Some(120));
service.set_server_time(130);
service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: "process-two".to_owned(),
restart: false,
},
})
.unwrap();
let CoordinatorResponse::ProcessSummaries {
processes,
next_cursor: Some(cursor),
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: None,
limit: 1,
},
})
.unwrap()
else {
panic!("expected first process summary page");
};
assert_eq!(processes[0].process, ProcessId::from("process-two"));
let CoordinatorResponse::ProcessSummaries {
processes,
next_cursor: None,
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: Some(cursor),
limit: 1,
},
})
.unwrap()
else {
panic!("expected final process summary page");
};
assert_eq!(processes[0].process, ProcessId::from("process-one"));
let oversized = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListProcessSummaries {
cursor: None,
limit: 101,
},
})
.unwrap_err();
assert!(oversized.to_string().contains("limit"));
assert!(oversized.to_string().contains("100"));
}
#[test]
fn web_node_summaries_are_scoped_paginated_and_hard_bounded() {
let mut service = CoordinatorService::new(7);
service
.issue_cli_session(
TenantId::from("tenant"),
ProjectId::from("project"),
UserId::from("user"),
"session",
None,
)
.unwrap();
for node in ["node-a", "node-b", "node-c"] {
enroll_test_node(
&mut service,
"tenant",
"project",
node,
&test_node_public_key(node),
);
service
.handle_signed_node_request_auto(CoordinatorRequest::ReportNodeCapabilities {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
node: node.to_owned(),
capabilities: linux_capabilities(),
cached_environment_digests: Vec::new(),
dependency_cache_digests: Vec::new(),
source_snapshots: Vec::new(),
artifact_locations: Vec::new(),
direct_connectivity: true,
online: true,
})
.unwrap();
}
let CoordinatorResponse::NodeSummaries {
nodes,
next_cursor: Some(cursor),
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session".to_owned(),
request: AuthenticatedCoordinatorRequest::ListNodeSummaries {
cursor: None,
limit: 2,
},
})
.unwrap()
else {
panic!("expected first node-summary page");
};
assert_eq!(
nodes
.iter()
.map(|node| node.id.as_str())
.collect::<Vec<_>>(),
["node-a", "node-b"]
);
let CoordinatorResponse::NodeSummaries {
nodes,
next_cursor: None,
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session".to_owned(),
request: AuthenticatedCoordinatorRequest::ListNodeSummaries {
cursor: Some(cursor),
limit: 2,
},
})
.unwrap()
else {
panic!("expected final node-summary page");
};
assert_eq!(nodes.len(), 1);
assert_eq!(nodes[0].id, NodeId::from("node-c"));
let oversized = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session".to_owned(),
request: AuthenticatedCoordinatorRequest::ListNodeSummaries {
cursor: None,
limit: 201,
},
})
.unwrap_err();
assert!(oversized.to_string().contains("limit"));
assert!(oversized.to_string().contains("200"));
}
#[test]
fn web_artifact_queries_are_scoped_paginated_and_track_retention_availability() {
let mut service = CoordinatorService::new(7);
for (tenant, project, user, secret, node) in [
("tenant-a", "project-a", "user-a", "session-a", "node-a"),
("tenant-b", "project-b", "user-b", "session-b", "node-b"),
] {
service
.issue_cli_session(
TenantId::from(tenant),
ProjectId::from(project),
UserId::from(user),
secret,
None,
)
.unwrap();
enroll_test_node(
&mut service,
tenant,
project,
node,
&test_node_public_key(node),
);
}
service.set_server_time(100);
for (tenant, project, node) in [
("tenant-a", "project-a", "node-a"),
("tenant-b", "project-b", "node-b"),
] {
service
.handle_signed_node_request_auto(CoordinatorRequest::ReportNodeCapabilities {
tenant: tenant.to_owned(),
project: project.to_owned(),
node: node.to_owned(),
capabilities: linux_capabilities(),
cached_environment_digests: Vec::new(),
dependency_cache_digests: Vec::new(),
source_snapshots: Vec::new(),
artifact_locations: vec!["shared-artifact".to_owned()],
direct_connectivity: true,
online: false,
})
.unwrap();
}
let CoordinatorResponse::NodeSummaries { nodes, .. } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListNodeSummaries {
cursor: None,
limit: 200,
},
})
.unwrap()
else {
panic!("expected node summaries");
};
assert_eq!(nodes.len(), 1);
assert_eq!(nodes[0].id, NodeId::from("node-a"));
assert!(nodes[0].online);
assert!(!nodes[0].stale);
assert_eq!(nodes[0].last_seen_epoch_seconds, Some(100));
assert_eq!(nodes[0].capabilities.os, Os::Linux);
for (tenant, project, node, digest) in [
("tenant-a", "project-a", "node-a", "tenant-a-bytes"),
("tenant-b", "project-b", "node-b", "tenant-b-bytes"),
] {
service.artifact_registry.flush_metadata(ArtifactFlush {
id: ArtifactId::from("shared-artifact"),
tenant: TenantId::from(tenant),
project: ProjectId::from(project),
process: ProcessId::from("process-one"),
producer_task: TaskInstanceId::from("task-one"),
retaining_node: NodeId::from(node),
digest: Digest::sha256(digest),
size: digest.len() as u64,
});
}
service.artifact_registry.flush_metadata(ArtifactFlush {
id: ArtifactId::from("second-artifact"),
tenant: TenantId::from("tenant-a"),
project: ProjectId::from("project-a"),
process: ProcessId::from("process-two"),
producer_task: TaskInstanceId::from("task-two"),
retaining_node: NodeId::from("node-a"),
digest: Digest::sha256("second"),
size: 6,
});
let CoordinatorResponse::Artifacts {
artifacts,
next_cursor: Some(cursor),
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListArtifacts {
process: None,
cursor: None,
limit: 1,
},
})
.unwrap()
else {
panic!("expected first artifact page");
};
assert_eq!(artifacts.len(), 1);
assert_eq!(artifacts[0].id, ArtifactId::from("second-artifact"));
assert_eq!(artifacts[0].availability, ArtifactAvailability::Available);
let CoordinatorResponse::Artifacts {
artifacts,
next_cursor: None,
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListArtifacts {
process: None,
cursor: Some(cursor),
limit: 1,
},
})
.unwrap()
else {
panic!("expected final artifact page");
};
assert_eq!(artifacts[0].id, ArtifactId::from("shared-artifact"));
assert_eq!(artifacts[0].digest, Digest::sha256("tenant-a-bytes"));
let cross_tenant = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::GetArtifact {
artifact: "tenant-b-only".to_owned(),
},
})
.unwrap_err();
assert!(cross_tenant.to_string().contains("does not exist"));
let CoordinatorResponse::Artifact { artifact } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-b".to_owned(),
request: AuthenticatedCoordinatorRequest::GetArtifact {
artifact: "shared-artifact".to_owned(),
},
})
.unwrap()
else {
panic!("expected tenant-b artifact");
};
assert_eq!(artifact.digest, Digest::sha256("tenant-b-bytes"));
service.set_server_time(131);
let CoordinatorResponse::NodeSummaries { nodes, .. } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListNodeSummaries {
cursor: None,
limit: 200,
},
})
.unwrap()
else {
panic!("expected stale node summary");
};
assert!(!nodes[0].online);
assert!(nodes[0].stale);
assert_eq!(nodes[0].last_seen_epoch_seconds, Some(100));
let CoordinatorResponse::Artifact { artifact } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::GetArtifact {
artifact: "shared-artifact".to_owned(),
},
})
.unwrap()
else {
panic!("expected offline artifact metadata");
};
assert_eq!(artifact.availability, ArtifactAvailability::NodeOffline);
assert!(!artifact.downloadable_now);
service
.artifact_registry
.sync_to_explicit_store(
&TenantId::from("tenant-a"),
&ProjectId::from("project-a"),
&ArtifactId::from("shared-artifact"),
"store://tenant-a/shared-artifact",
)
.unwrap();
let CoordinatorResponse::Artifact { artifact } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::GetArtifact {
artifact: "shared-artifact".to_owned(),
},
})
.unwrap()
else {
panic!("expected explicitly retained artifact metadata");
};
assert_eq!(artifact.availability, ArtifactAvailability::Available);
assert_eq!(
artifact.retention_state,
ArtifactRetentionState::ExplicitStorage
);
assert!(artifact.downloadable_now);
}
#[test]
fn web_recent_logs_are_signed_scoped_cursor_safe_and_memory_bounded() {
let mut service = CoordinatorService::new(7);
for (tenant, project, user, secret, node, process) in [
(
"tenant-a",
"project-a",
"user-a",
"session-a",
"node-a",
"process-shared",
),
(
"tenant-b",
"project-b",
"user-b",
"session-b",
"node-b",
"process-b",
),
] {
service
.issue_cli_session(
TenantId::from(tenant),
ProjectId::from(project),
UserId::from(user),
secret,
None,
)
.unwrap();
enroll_test_node(
&mut service,
tenant,
project,
node,
&test_node_public_key(node),
);
service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: secret.to_owned(),
request: AuthenticatedCoordinatorRequest::StartProcess {
launch_attempt: None,
process: process.to_owned(),
restart: false,
},
})
.unwrap();
service
.handle_signed_node_request_auto(CoordinatorRequest::ReconnectNode {
tenant: tenant.to_owned(),
project: project.to_owned(),
node: node.to_owned(),
process: process.to_owned(),
epoch: 7,
})
.unwrap();
}
service.set_server_time(100);
let first = service
.handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLogChunk {
tenant: "tenant-a".to_owned(),
project: "project-a".to_owned(),
process: "process-shared".to_owned(),
node: "node-a".to_owned(),
task: "task-one".to_owned(),
stream: TaskLogStream::Stdout,
offset: 0,
source_bytes: 5,
text: "hello".to_owned(),
truncated: false,
})
.unwrap();
let CoordinatorResponse::TaskLogChunkRecorded {
sequence: Some(first_sequence),
next_offset: 5,
..
} = first
else {
panic!("expected first live log sequence");
};
let retry = service
.handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLogChunk {
tenant: "tenant-a".to_owned(),
project: "project-a".to_owned(),
process: "process-shared".to_owned(),
node: "node-a".to_owned(),
task: "task-one".to_owned(),
stream: TaskLogStream::Stdout,
offset: 0,
source_bytes: 5,
text: "hello".to_owned(),
truncated: false,
})
.unwrap();
assert!(matches!(
retry,
CoordinatorResponse::TaskLogChunkRecorded { sequence: None, .. }
));
service
.handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLogChunk {
tenant: "tenant-a".to_owned(),
project: "project-a".to_owned(),
process: "process-shared".to_owned(),
node: "node-a".to_owned(),
task: "task-one".to_owned(),
stream: TaskLogStream::Stdout,
offset: 8,
source_bytes: 2,
text: "ok".to_owned(),
truncated: false,
})
.unwrap();
let CoordinatorResponse::RecentLogs {
entries,
next_sequence: Some(cursor),
history_truncated: false,
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListRecentLogs {
process: "process-shared".to_owned(),
task: None,
after_sequence: None,
limit: 2,
},
})
.unwrap()
else {
panic!("expected first recent-log page");
};
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].sequence, first_sequence);
assert_eq!(entries[0].text, "hello");
assert!(entries[1].text.contains("3 bytes"));
assert!(entries[1].truncated);
let CoordinatorResponse::RecentLogs { entries, .. } = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListRecentLogs {
process: "process-shared".to_owned(),
task: None,
after_sequence: Some(cursor),
limit: 2,
},
})
.unwrap()
else {
panic!("expected second recent-log page");
};
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].text, "ok");
let cross_tenant = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-b".to_owned(),
request: AuthenticatedCoordinatorRequest::ListRecentLogs {
process: "process-shared".to_owned(),
task: None,
after_sequence: None,
limit: 10,
},
})
.unwrap_err();
assert!(cross_tenant.to_string().contains("outside"));
let oversized = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListRecentLogs {
process: "process-shared".to_owned(),
task: None,
after_sequence: None,
limit: 201,
},
})
.unwrap_err();
assert!(oversized.to_string().contains("limit"));
assert!(oversized.to_string().contains("200"));
service
.handle_report_task_log_chunk(
"tenant-b".to_owned(),
"project-b".to_owned(),
"process-b".to_owned(),
"node-b".to_owned(),
"task-b".to_owned(),
TaskLogStream::Stderr,
0,
1,
"b".to_owned(),
false,
)
.unwrap();
for offset in 10..310 {
service
.handle_report_task_log_chunk(
"tenant-a".to_owned(),
"project-a".to_owned(),
"process-shared".to_owned(),
"node-a".to_owned(),
"task-one".to_owned(),
TaskLogStream::Stdout,
offset,
1,
"x".to_owned(),
false,
)
.unwrap();
}
let tenant_a_logs =
&service.recent_logs[&(TenantId::from("tenant-a"), ProjectId::from("project-a"))];
assert!(tenant_a_logs.len() <= MAX_RECENT_LOG_ENTRIES_PER_PROCESS);
let tenant_b_logs =
&service.recent_logs[&(TenantId::from("tenant-b"), ProjectId::from("project-b"))];
assert_eq!(tenant_b_logs.len(), 1);
assert_eq!(tenant_b_logs[0].text, "b");
let CoordinatorResponse::RecentLogs {
entries,
history_truncated,
..
} = service
.handle_request(CoordinatorRequest::Authenticated {
session_secret: "session-a".to_owned(),
request: AuthenticatedCoordinatorRequest::ListRecentLogs {
process: "process-shared".to_owned(),
task: None,
after_sequence: None,
limit: 200,
},
})
.unwrap()
else {
panic!("expected bounded recent-log response");
};
assert_eq!(entries.len(), 200);
assert!(history_truncated);
}

12
crates/clusterflux-coordinator/src/service/wire_protocol.rs
View file

@ -12,8 +12,12 @@ pub enum CoordinatorWireRequest {
impl CoordinatorWireRequest {
pub fn into_request(self) -> Result<CoordinatorRequest, String> {
self.into_parts().map(|(_, request)| request)
}
pub fn into_parts(self) -> Result<(String, CoordinatorRequest), String> {
match self {
Self::Envelope(envelope) => envelope.into_request(),
Self::Envelope(envelope) => envelope.into_parts(),
}
}
}
@ -32,6 +36,10 @@ pub struct CoordinatorRequestEnvelope {
impl CoordinatorRequestEnvelope {
pub fn into_request(self) -> Result<CoordinatorRequest, String> {
self.into_parts().map(|(_, request)| request)
}
pub fn into_parts(self) -> Result<(String, CoordinatorRequest), String> {
if self.envelope_type != COORDINATOR_WIRE_REQUEST_TYPE {
return Err(format!(
"unsupported coordinator wire request type {}; expected {}",
@ -54,6 +62,6 @@ impl CoordinatorRequestEnvelope {
self.operation, payload_operation
));
}
Ok(self.payload)
Ok((self.request_id, self.payload))
}
}