Compare commits
No commits in common. "main" and "release-501cb6d672f9" have entirely different histories.
main
...
release-50
12 changed files with 166 additions and 430 deletions
|
|
@ -109,21 +109,6 @@ impl ClusterfluxClient {
|
|||
}
|
||||
}
|
||||
|
||||
pub async fn cancel_browser_login(
|
||||
&self,
|
||||
transaction_id: impl Into<String>,
|
||||
) -> Result<(), ClientError> {
|
||||
match self
|
||||
.send_login(LoginRequest::CancelWebBrowserLogin {
|
||||
transaction_id: transaction_id.into(),
|
||||
})
|
||||
.await?
|
||||
{
|
||||
WireResponse::WebBrowserLoginCancelled {} => Ok(()),
|
||||
_ => Err(unexpected_response("web_browser_login_cancelled")),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn account_status(&self) -> Result<AccountStatus, ClientError> {
|
||||
match self
|
||||
.send_authenticated(AuthenticatedRequest::AuthStatus)
|
||||
|
|
|
|||
|
|
@ -122,9 +122,6 @@ pub(crate) enum AuthenticatedRequest {
|
|||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
pub(crate) enum LoginRequest {
|
||||
BeginWebBrowserLogin {},
|
||||
CancelWebBrowserLogin {
|
||||
transaction_id: String,
|
||||
},
|
||||
ExchangeWebLoginHandoff {
|
||||
transaction_id: String,
|
||||
handoff_code: String,
|
||||
|
|
@ -300,7 +297,6 @@ pub(crate) enum WireResponse {
|
|||
authorization_url: String,
|
||||
expires_at_epoch_seconds: u64,
|
||||
},
|
||||
WebBrowserLoginCancelled {},
|
||||
WebBrowserSession {
|
||||
session: WireBrowserSession,
|
||||
},
|
||||
|
|
@ -338,7 +334,6 @@ mod tests {
|
|||
"abort_process",
|
||||
"auth_status",
|
||||
"begin_web_browser_login",
|
||||
"cancel_web_browser_login",
|
||||
"cancel_process",
|
||||
"create_artifact_download_link",
|
||||
"create_debug_epoch",
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use clusterflux_client::{
|
|||
ClusterfluxClient, ControlTransport, MockTransport, ProjectId, SessionCredential, TenantId,
|
||||
UserId, CLIENT_API_VERSION,
|
||||
};
|
||||
use clusterflux_control::{CONTROL_API_PATH, LOGIN_API_PATH};
|
||||
use clusterflux_control::CONTROL_API_PATH;
|
||||
use clusterflux_coordinator::service::CoordinatorService;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
|
|
@ -64,32 +64,6 @@ async fn structured_errors_retain_machine_fields_and_originating_request_id() {
|
|||
assert!(!error.retryable);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn browser_login_cancellation_uses_the_login_boundary_and_exact_transaction() {
|
||||
let transport = MockTransport::from_json_responses([
|
||||
json!({ "type": "web_browser_login_cancelled" }).to_string(),
|
||||
]);
|
||||
let client = ClusterfluxClient::with_transport(transport.clone());
|
||||
|
||||
client
|
||||
.cancel_browser_login("login-transaction")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let requests = transport.requests();
|
||||
assert_eq!(requests.len(), 1);
|
||||
assert_eq!(requests[0].api_path, LOGIN_API_PATH);
|
||||
let envelope: Value = serde_json::from_slice(&requests[0].body).unwrap();
|
||||
assert_eq!(envelope["operation"], "cancel_web_browser_login");
|
||||
assert_eq!(
|
||||
envelope["payload"],
|
||||
json!({
|
||||
"type": "cancel_web_browser_login",
|
||||
"transaction_id": "login-transaction"
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_mismatched_error_request_id_is_rejected_as_a_protocol_error() {
|
||||
let transport = MockTransport::from_json_responses([json!({
|
||||
|
|
|
|||
|
|
@ -5,12 +5,6 @@
|
|||
"request": { "type": "begin_web_browser_login" },
|
||||
"response": { "type": "web_browser_login_started", "transaction_id": "login-transaction", "authorization_url": "https://auth.clusterflux.lesstuff.com/authorize?state=opaque", "expires_at_epoch_seconds": 1000 }
|
||||
},
|
||||
{
|
||||
"operation": "cancel_web_browser_login",
|
||||
"boundary": "login",
|
||||
"request": { "type": "cancel_web_browser_login", "transaction_id": "login-transaction" },
|
||||
"response": { "type": "web_browser_login_cancelled" }
|
||||
},
|
||||
{
|
||||
"operation": "exchange_web_login_handoff",
|
||||
"boundary": "login",
|
||||
|
|
|
|||
|
|
@ -304,13 +304,6 @@ pub struct CoordinatorService {
|
|||
clusterflux_core::TaskInstanceId,
|
||||
String,
|
||||
)>,
|
||||
recent_log_quota_truncated_streams: BTreeSet<(
|
||||
TenantId,
|
||||
ProjectId,
|
||||
ProcessId,
|
||||
clusterflux_core::TaskInstanceId,
|
||||
String,
|
||||
)>,
|
||||
next_recent_log_sequence: u64,
|
||||
debug_audit_events: VecDeque<DebugAuditEvent>,
|
||||
debug_epochs: BTreeMap<ProcessControlKey, u64>,
|
||||
|
|
@ -586,7 +579,6 @@ impl CoordinatorService {
|
|||
recent_log_dropped_through: BTreeMap::new(),
|
||||
recent_log_accounted_bytes: BTreeMap::new(),
|
||||
recent_log_truncated_streams: BTreeSet::new(),
|
||||
recent_log_quota_truncated_streams: BTreeSet::new(),
|
||||
next_recent_log_sequence: 1,
|
||||
debug_audit_events: VecDeque::new(),
|
||||
debug_epochs: BTreeMap::new(),
|
||||
|
|
|
|||
|
|
@ -40,7 +40,19 @@ impl CoordinatorService {
|
|||
validate_task_log_tail("stdout_tail", &stdout_tail)?;
|
||||
validate_task_log_tail("stderr_tail", &stderr_tail)?;
|
||||
let now_epoch_seconds = self.current_epoch_seconds()?;
|
||||
let stdout_retained = self.accept_final_log_stream(
|
||||
let unaccounted_bytes = self.unaccounted_final_log_bytes(
|
||||
&tenant,
|
||||
&project,
|
||||
&process,
|
||||
&task,
|
||||
stdout_bytes,
|
||||
stderr_bytes,
|
||||
)?;
|
||||
self.quota
|
||||
.can_charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?;
|
||||
self.quota
|
||||
.charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?;
|
||||
self.reconcile_final_log_stream(
|
||||
&tenant,
|
||||
&project,
|
||||
&process,
|
||||
|
|
@ -50,8 +62,8 @@ impl CoordinatorService {
|
|||
&stdout_tail,
|
||||
stdout_truncated,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
let stderr_retained = self.accept_final_log_stream(
|
||||
);
|
||||
self.reconcile_final_log_stream(
|
||||
&tenant,
|
||||
&project,
|
||||
&process,
|
||||
|
|
@ -61,22 +73,18 @@ impl CoordinatorService {
|
|||
&stderr_tail,
|
||||
stderr_truncated,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
);
|
||||
Ok(CoordinatorResponse::TaskLogRecorded {
|
||||
process,
|
||||
task,
|
||||
stdout_bytes,
|
||||
stderr_bytes,
|
||||
stdout_tail: if !stdout_retained {
|
||||
"[log output truncated at project log quota]".to_owned()
|
||||
} else if stdout_truncated {
|
||||
stdout_tail: if stdout_truncated {
|
||||
format!("{stdout_tail}\n... truncated")
|
||||
} else {
|
||||
stdout_tail
|
||||
},
|
||||
stderr_tail: if !stderr_retained {
|
||||
"[log output truncated at project log quota]".to_owned()
|
||||
} else if stderr_truncated {
|
||||
stderr_tail: if stderr_truncated {
|
||||
format!("{stderr_tail}\n... truncated")
|
||||
} else {
|
||||
stderr_tail
|
||||
|
|
@ -142,41 +150,11 @@ impl CoordinatorService {
|
|||
});
|
||||
}
|
||||
let now_epoch_seconds = self.current_epoch_seconds()?;
|
||||
if self.recent_log_quota_truncated_streams.contains(&key) {
|
||||
if end > expected {
|
||||
self.recent_log_accounted_bytes.insert(key, end);
|
||||
}
|
||||
return Ok(CoordinatorResponse::TaskLogChunkRecorded {
|
||||
process,
|
||||
task,
|
||||
sequence: None,
|
||||
next_offset: end.max(expected),
|
||||
});
|
||||
}
|
||||
let newly_accounted = end.saturating_sub(expected);
|
||||
if self
|
||||
.quota
|
||||
.charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)
|
||||
.is_err()
|
||||
{
|
||||
if end > expected {
|
||||
self.recent_log_accounted_bytes.insert(key.clone(), end);
|
||||
}
|
||||
let sequence = self.mark_log_quota_truncated(
|
||||
&tenant,
|
||||
&project,
|
||||
&process,
|
||||
&task,
|
||||
&stream,
|
||||
now_epoch_seconds,
|
||||
);
|
||||
return Ok(CoordinatorResponse::TaskLogChunkRecorded {
|
||||
process,
|
||||
task,
|
||||
sequence,
|
||||
next_offset: end.max(expected),
|
||||
});
|
||||
}
|
||||
self.quota
|
||||
.can_charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
|
||||
self.quota
|
||||
.charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
|
||||
if offset > expected {
|
||||
self.record_recent_log(
|
||||
tenant.clone(),
|
||||
|
|
@ -356,7 +334,27 @@ impl CoordinatorService {
|
|||
result,
|
||||
};
|
||||
let now_epoch_seconds = self.current_epoch_seconds()?;
|
||||
let stdout_retained = self.accept_final_log_stream(
|
||||
let unaccounted_bytes = self.unaccounted_final_log_bytes(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
&event.process,
|
||||
&event.task,
|
||||
event.stdout_bytes,
|
||||
event.stderr_bytes,
|
||||
)?;
|
||||
self.quota.can_charge_log_bytes(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
unaccounted_bytes,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
self.quota.charge_log_bytes(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
unaccounted_bytes,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
self.reconcile_final_log_stream(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
&event.process,
|
||||
|
|
@ -366,8 +364,8 @@ impl CoordinatorService {
|
|||
&event.stdout_tail,
|
||||
event.stdout_truncated,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
let stderr_retained = self.accept_final_log_stream(
|
||||
);
|
||||
self.reconcile_final_log_stream(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
&event.process,
|
||||
|
|
@ -377,15 +375,7 @@ impl CoordinatorService {
|
|||
&event.stderr_tail,
|
||||
event.stderr_truncated,
|
||||
now_epoch_seconds,
|
||||
)?;
|
||||
if !stdout_retained {
|
||||
event.stdout_tail = "[log output truncated at project log quota]".to_owned();
|
||||
event.stdout_truncated = true;
|
||||
}
|
||||
if !stderr_retained {
|
||||
event.stderr_tail = "[log output truncated at project log quota]".to_owned();
|
||||
event.stderr_truncated = true;
|
||||
}
|
||||
);
|
||||
let task_key = task_control_key(
|
||||
&event.tenant,
|
||||
&event.project,
|
||||
|
|
@ -843,94 +833,48 @@ impl CoordinatorService {
|
|||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn accept_final_log_stream(
|
||||
&mut self,
|
||||
fn unaccounted_final_log_bytes(
|
||||
&self,
|
||||
tenant: &TenantId,
|
||||
project: &ProjectId,
|
||||
process: &ProcessId,
|
||||
task: &TaskInstanceId,
|
||||
stream: TaskLogStream,
|
||||
total_source_bytes: u64,
|
||||
final_tail: &str,
|
||||
source_truncated: bool,
|
||||
now_epoch_seconds: u64,
|
||||
) -> Result<bool, CoordinatorServiceError> {
|
||||
let key = recent_log_offset_key(tenant, project, process, task, &stream);
|
||||
let accounted = self
|
||||
stdout_bytes: u64,
|
||||
stderr_bytes: u64,
|
||||
) -> Result<u64, CoordinatorServiceError> {
|
||||
let stdout_accounted = self
|
||||
.recent_log_accounted_bytes
|
||||
.get(&key)
|
||||
.get(&recent_log_offset_key(
|
||||
tenant,
|
||||
project,
|
||||
process,
|
||||
task,
|
||||
&TaskLogStream::Stdout,
|
||||
))
|
||||
.copied()
|
||||
.unwrap_or(0);
|
||||
let remaining = total_source_bytes.checked_sub(accounted).ok_or_else(|| {
|
||||
let stream_name = match stream {
|
||||
TaskLogStream::Stdout => "stdout",
|
||||
TaskLogStream::Stderr => "stderr",
|
||||
};
|
||||
let stderr_accounted = self
|
||||
.recent_log_accounted_bytes
|
||||
.get(&recent_log_offset_key(
|
||||
tenant,
|
||||
project,
|
||||
process,
|
||||
task,
|
||||
&TaskLogStream::Stderr,
|
||||
))
|
||||
.copied()
|
||||
.unwrap_or(0);
|
||||
let stdout_remaining = stdout_bytes.checked_sub(stdout_accounted).ok_or_else(|| {
|
||||
CoordinatorServiceError::Protocol(format!(
|
||||
"final {stream_name} byte count {total_source_bytes} is below the {accounted} live bytes already accounted"
|
||||
"final stdout byte count {stdout_bytes} is below the {stdout_accounted} live bytes already accounted"
|
||||
))
|
||||
})?;
|
||||
if self.recent_log_quota_truncated_streams.contains(&key) {
|
||||
self.recent_log_accounted_bytes
|
||||
.insert(key, total_source_bytes);
|
||||
return Ok(false);
|
||||
}
|
||||
if self
|
||||
.quota
|
||||
.charge_log_bytes(tenant, project, remaining, now_epoch_seconds)
|
||||
.is_err()
|
||||
{
|
||||
self.recent_log_accounted_bytes
|
||||
.insert(key, total_source_bytes);
|
||||
self.mark_log_quota_truncated(
|
||||
tenant,
|
||||
project,
|
||||
process,
|
||||
task,
|
||||
&stream,
|
||||
now_epoch_seconds,
|
||||
);
|
||||
return Ok(false);
|
||||
}
|
||||
self.reconcile_final_log_stream(
|
||||
tenant,
|
||||
project,
|
||||
process,
|
||||
task,
|
||||
stream,
|
||||
total_source_bytes,
|
||||
final_tail,
|
||||
source_truncated,
|
||||
now_epoch_seconds,
|
||||
);
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn mark_log_quota_truncated(
|
||||
&mut self,
|
||||
tenant: &TenantId,
|
||||
project: &ProjectId,
|
||||
process: &ProcessId,
|
||||
task: &TaskInstanceId,
|
||||
stream: &TaskLogStream,
|
||||
now_epoch_seconds: u64,
|
||||
) -> Option<u64> {
|
||||
let key = recent_log_offset_key(tenant, project, process, task, stream);
|
||||
if !self.recent_log_quota_truncated_streams.insert(key.clone()) {
|
||||
return None;
|
||||
}
|
||||
self.recent_log_truncated_streams.insert(key);
|
||||
Some(self.record_recent_log(
|
||||
tenant.clone(),
|
||||
project.clone(),
|
||||
process.clone(),
|
||||
task.clone(),
|
||||
stream.clone(),
|
||||
"[log output truncated at project log quota]".to_owned(),
|
||||
true,
|
||||
now_epoch_seconds,
|
||||
let stderr_remaining = stderr_bytes.checked_sub(stderr_accounted).ok_or_else(|| {
|
||||
CoordinatorServiceError::Protocol(format!(
|
||||
"final stderr byte count {stderr_bytes} is below the {stderr_accounted} live bytes already accounted"
|
||||
))
|
||||
})?;
|
||||
checked_reported_log_bytes(stdout_remaining, stderr_remaining)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
|
|
@ -1076,14 +1020,6 @@ impl CoordinatorService {
|
|||
|| entry_task != task
|
||||
},
|
||||
);
|
||||
self.recent_log_quota_truncated_streams.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 {
|
||||
|
|
@ -1324,6 +1260,17 @@ impl CoordinatorService {
|
|||
}
|
||||
}
|
||||
|
||||
fn checked_reported_log_bytes(
|
||||
stdout_bytes: u64,
|
||||
stderr_bytes: u64,
|
||||
) -> Result<u64, CoordinatorServiceError> {
|
||||
stdout_bytes.checked_add(stderr_bytes).ok_or_else(|| {
|
||||
CoordinatorServiceError::Protocol(
|
||||
"reported task log byte counts exceed the supported range".to_owned(),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn recent_log_offset_key(
|
||||
tenant: &TenantId,
|
||||
project: &ProjectId,
|
||||
|
|
@ -1358,11 +1305,11 @@ fn bounded_log_tail(mut value: String, truncated: &mut bool) -> String {
|
|||
if value.len() <= MAX_TASK_LOG_TAIL_BYTES {
|
||||
return value;
|
||||
}
|
||||
let mut boundary = value.len() - MAX_TASK_LOG_TAIL_BYTES;
|
||||
while boundary < value.len() && !value.is_char_boundary(boundary) {
|
||||
boundary += 1;
|
||||
let mut boundary = MAX_TASK_LOG_TAIL_BYTES;
|
||||
while !value.is_char_boundary(boundary) {
|
||||
boundary -= 1;
|
||||
}
|
||||
value.drain(..boundary);
|
||||
value.truncate(boundary);
|
||||
*truncated = true;
|
||||
value
|
||||
}
|
||||
|
|
|
|||
|
|
@ -260,6 +260,22 @@ impl CoordinatorQuota {
|
|||
self.charge(tenant, project, LimitKind::ApiCall, 1, now_epoch_seconds)
|
||||
}
|
||||
|
||||
pub(super) fn can_charge_log_bytes(
|
||||
&self,
|
||||
tenant: &TenantId,
|
||||
project: &ProjectId,
|
||||
bytes: u64,
|
||||
now_epoch_seconds: u64,
|
||||
) -> Result<(), LimitError> {
|
||||
self.can_charge(
|
||||
tenant,
|
||||
project,
|
||||
LimitKind::LogBytes,
|
||||
bytes,
|
||||
now_epoch_seconds,
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) fn charge_log_bytes(
|
||||
&mut self,
|
||||
tenant: &TenantId,
|
||||
|
|
|
|||
|
|
@ -98,11 +98,6 @@ impl CoordinatorService {
|
|||
entry_tenant != tenant || entry_project != project || entry_process != process
|
||||
},
|
||||
);
|
||||
self.recent_log_quota_truncated_streams.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);
|
||||
|
|
@ -530,10 +525,6 @@ impl CoordinatorService {
|
|||
.retain(|(tenant, project, process, _, _)| {
|
||||
tenant != &key.0 || project != &key.1 || process != &key.2
|
||||
});
|
||||
self.recent_log_quota_truncated_streams
|
||||
.retain(|(tenant, project, process, _, _)| {
|
||||
tenant != &key.0 || project != &key.1 || process != &key.2
|
||||
});
|
||||
self.process_summary_order
|
||||
.retain(|retained| retained != key);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2435,7 +2435,7 @@ fn authenticated_api_calls_are_metered_per_tenant_and_project_before_dispatch()
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports() {
|
||||
fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
|
||||
let mut limits = ResourceLimits::unlimited();
|
||||
limits.limits.insert(LimitKind::LogBytes, 4);
|
||||
let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap();
|
||||
|
|
@ -2494,7 +2494,7 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports()
|
|||
backpressured: false,
|
||||
})
|
||||
.unwrap();
|
||||
let truncated = service
|
||||
let denied = service
|
||||
.handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLog {
|
||||
tenant: "tenant".to_owned(),
|
||||
project: "project".to_owned(),
|
||||
|
|
@ -2509,22 +2509,8 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports()
|
|||
stderr_truncated: false,
|
||||
backpressured: true,
|
||||
})
|
||||
.unwrap();
|
||||
let CoordinatorResponse::TaskLogRecorded {
|
||||
stdout_tail,
|
||||
stdout_bytes: 4,
|
||||
..
|
||||
} = truncated
|
||||
else {
|
||||
panic!("expected a successful truncated task-log report");
|
||||
};
|
||||
assert_eq!(stdout_tail, "[log output truncated at project log quota]");
|
||||
assert!(service
|
||||
.recent_logs
|
||||
.get(&(TenantId::from("tenant"), ProjectId::from("project")))
|
||||
.unwrap()
|
||||
.iter()
|
||||
.any(|entry| entry.text.contains("project log quota") && entry.truncated));
|
||||
.unwrap_err();
|
||||
assert!(denied.to_string().contains("LogBytes"));
|
||||
assert_eq!(
|
||||
service
|
||||
.quota
|
||||
|
|
@ -2533,140 +2519,6 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports()
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn log_quota_exhaustion_cannot_strand_task_completion_or_artifact_publication() {
|
||||
let mut limits = ResourceLimits::unlimited();
|
||||
limits.limits.insert(LimitKind::LogBytes, 4);
|
||||
let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap();
|
||||
let mut service = CoordinatorService::new_with_admin_token_database_url_and_quota(
|
||||
7,
|
||||
"test-admin-token",
|
||||
None,
|
||||
quota,
|
||||
)
|
||||
.unwrap();
|
||||
service.set_server_time(30);
|
||||
service
|
||||
.handle_request(CoordinatorRequest::AttachNode {
|
||||
tenant: "tenant".to_owned(),
|
||||
project: "project".to_owned(),
|
||||
node: "node".to_owned(),
|
||||
public_key: test_node_public_key("node"),
|
||||
})
|
||||
.unwrap();
|
||||
service
|
||||
.handle_request(CoordinatorRequest::StartProcess {
|
||||
launch_attempt: None,
|
||||
tenant: "tenant".to_owned(),
|
||||
project: "project".to_owned(),
|
||||
actor_user: None,
|
||||
actor_agent: None,
|
||||
agent_public_key_fingerprint: None,
|
||||
agent_signature: 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();
|
||||
register_test_task_assignment(
|
||||
&mut service,
|
||||
"tenant",
|
||||
"project",
|
||||
"process",
|
||||
"node",
|
||||
"compile",
|
||||
"compile-one",
|
||||
7,
|
||||
);
|
||||
service.record_task_completion_event(TaskCompletionEvent {
|
||||
tenant: TenantId::from("tenant"),
|
||||
project: ProjectId::from("project"),
|
||||
process: ProcessId::from("process"),
|
||||
node: NodeId::from("coordinator-main"),
|
||||
executor: TaskExecutor::CoordinatorMain,
|
||||
task_definition: TaskDefinitionId::from("build"),
|
||||
task: TaskInstanceId::from("main"),
|
||||
attempt_id: None,
|
||||
placement: None,
|
||||
terminal_state: TaskTerminalState::Completed,
|
||||
status_code: Some(0),
|
||||
stdout_bytes: 0,
|
||||
stderr_bytes: 0,
|
||||
stdout_tail: String::new(),
|
||||
stderr_tail: String::new(),
|
||||
stdout_truncated: false,
|
||||
stderr_truncated: false,
|
||||
artifact_path: None,
|
||||
artifact_digest: None,
|
||||
artifact_size_bytes: None,
|
||||
result: None,
|
||||
});
|
||||
|
||||
service
|
||||
.handle_signed_node_request_auto(CoordinatorRequest::TaskCompleted {
|
||||
tenant: "tenant".to_owned(),
|
||||
project: "project".to_owned(),
|
||||
process: "process".to_owned(),
|
||||
node: "node".to_owned(),
|
||||
task: "compile-one".to_owned(),
|
||||
terminal_state: Some(TaskTerminalState::Completed),
|
||||
status_code: Some(0),
|
||||
stdout_bytes: 64,
|
||||
stderr_bytes: 0,
|
||||
stdout_tail: "the-real-final-tail".to_owned(),
|
||||
stderr_tail: String::new(),
|
||||
stdout_truncated: true,
|
||||
stderr_truncated: false,
|
||||
artifact_path: Some("/vfs/artifacts/result.bin".to_owned()),
|
||||
artifact_digest: Some(Digest::sha256("artifact bytes")),
|
||||
artifact_size_bytes: Some(14),
|
||||
result: None,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let event = service
|
||||
.task_events
|
||||
.iter()
|
||||
.find(|event| event.task == TaskInstanceId::from("compile-one"))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
event.stdout_tail,
|
||||
"[log output truncated at project log quota]"
|
||||
);
|
||||
assert!(event.stdout_truncated);
|
||||
assert!(!service
|
||||
.active_tasks
|
||||
.iter()
|
||||
.any(|key| key.4 == TaskInstanceId::from("compile-one")));
|
||||
assert!(service
|
||||
.coordinator
|
||||
.active_process(
|
||||
&TenantId::from("tenant"),
|
||||
&ProjectId::from("project"),
|
||||
&ProcessId::from("process"),
|
||||
)
|
||||
.is_none());
|
||||
assert!(matches!(
|
||||
service
|
||||
.handle_request(CoordinatorRequest::GetArtifact {
|
||||
tenant: "tenant".to_owned(),
|
||||
project: "project".to_owned(),
|
||||
actor_user: "user".to_owned(),
|
||||
artifact: "result.bin".to_owned(),
|
||||
})
|
||||
.unwrap(),
|
||||
CoordinatorResponse::Artifact { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn service_attaches_node_starts_process_and_records_scoped_task_event() {
|
||||
let mut service = CoordinatorService::new(7);
|
||||
|
|
|
|||
|
|
@ -9,26 +9,6 @@ struct LiveLogChunk {
|
|||
truncated: bool,
|
||||
}
|
||||
|
||||
fn append_bounded_tail(tail: &mut Vec<u8>, bytes: &[u8], maximum: usize) {
|
||||
if maximum == 0 {
|
||||
tail.clear();
|
||||
return;
|
||||
}
|
||||
if bytes.len() >= maximum {
|
||||
tail.clear();
|
||||
tail.extend_from_slice(&bytes[bytes.len() - maximum..]);
|
||||
return;
|
||||
}
|
||||
let overflow = tail
|
||||
.len()
|
||||
.saturating_add(bytes.len())
|
||||
.saturating_sub(maximum);
|
||||
if overflow > 0 {
|
||||
tail.drain(..overflow);
|
||||
}
|
||||
tail.extend_from_slice(bytes);
|
||||
}
|
||||
|
||||
fn redact_safe_live_log_prefix(
|
||||
pending: &[u8],
|
||||
configured_secrets: &[String],
|
||||
|
|
@ -299,9 +279,9 @@ impl CoordinatorControlledProcessRunner {
|
|||
let mut captured = Vec::new();
|
||||
let mut buffer = [0_u8; 16 * 1024];
|
||||
let stream_base = source_bytes_total.load(Ordering::Relaxed);
|
||||
let mut source_bytes_read = 0_u64;
|
||||
let mut pending_offset = stream_base;
|
||||
let mut pending = Vec::new();
|
||||
let mut discarded = false;
|
||||
loop {
|
||||
let count = reader
|
||||
.read(&mut buffer)
|
||||
|
|
@ -314,9 +294,11 @@ impl CoordinatorControlledProcessRunner {
|
|||
Ordering::Relaxed,
|
||||
|current| Some(current.saturating_add(count as u64)),
|
||||
);
|
||||
source_bytes_read = source_bytes_read.saturating_add(count as u64);
|
||||
append_bounded_tail(&mut captured, &buffer[..count], maximum);
|
||||
pending.extend_from_slice(&buffer[..count]);
|
||||
let remaining = maximum.saturating_sub(captured.len());
|
||||
let retained = count.min(remaining);
|
||||
if retained > 0 {
|
||||
captured.extend_from_slice(&buffer[..retained]);
|
||||
pending.extend_from_slice(&buffer[..retained]);
|
||||
if let Some((consumed, text)) =
|
||||
redact_safe_live_log_prefix(&pending, &configured_secrets, false)
|
||||
{
|
||||
|
|
@ -331,6 +313,10 @@ impl CoordinatorControlledProcessRunner {
|
|||
pending_offset = pending_offset.saturating_add(consumed as u64);
|
||||
}
|
||||
}
|
||||
if retained < count {
|
||||
discarded = true;
|
||||
}
|
||||
}
|
||||
if let Some((consumed, text)) =
|
||||
redact_safe_live_log_prefix(&pending, &configured_secrets, true)
|
||||
{
|
||||
|
|
@ -342,10 +328,10 @@ impl CoordinatorControlledProcessRunner {
|
|||
truncated: false,
|
||||
});
|
||||
}
|
||||
if source_bytes_read > maximum as u64 {
|
||||
if discarded {
|
||||
let _ = sender.try_send(LiveLogChunk {
|
||||
stream,
|
||||
offset: stream_base.saturating_add(source_bytes_read),
|
||||
offset: stream_base.saturating_add(maximum as u64),
|
||||
source_bytes: 0,
|
||||
bytes: b"[log output truncated at node capture limit]".to_vec(),
|
||||
truncated: true,
|
||||
|
|
@ -630,11 +616,11 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn bounded_capture_retains_the_real_tail_and_complete_source_byte_count() {
|
||||
fn bounded_capture_retains_the_complete_source_byte_count() {
|
||||
let source_bytes = Arc::new(AtomicU64::new(5));
|
||||
let (sender, receiver) = mpsc::sync_channel(8);
|
||||
let reader = CoordinatorControlledProcessRunner::drain_bounded(
|
||||
Cursor::new(b"0123456789abcdefghijklmnopqrstuv".to_vec()),
|
||||
Cursor::new(vec![b'x'; 32]),
|
||||
8,
|
||||
"stdout",
|
||||
sender,
|
||||
|
|
@ -644,14 +630,14 @@ mod tests {
|
|||
let captured = reader.join().unwrap().unwrap();
|
||||
let chunks = receiver.into_iter().collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(captured, b"opqrstuv");
|
||||
assert_eq!(captured, vec![b'x'; 8]);
|
||||
assert_eq!(source_bytes.load(Ordering::Relaxed), 37);
|
||||
assert_eq!(
|
||||
chunks.iter().map(|chunk| chunk.source_bytes).sum::<u64>(),
|
||||
32
|
||||
8
|
||||
);
|
||||
assert_eq!(chunks[0].offset, 5);
|
||||
assert_eq!(chunks.last().unwrap().offset, 37);
|
||||
assert_eq!(chunks.last().unwrap().offset, 13);
|
||||
assert!(chunks.iter().any(|chunk| chunk.truncated));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use clusterflux_core::{
|
||||
CommandInvocation, Digest, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, VfsOverlay,
|
||||
VfsPath,
|
||||
CommandInvocation, Digest, LogBuffer, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject,
|
||||
VfsOverlay, VfsPath,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
|
|
@ -113,20 +113,24 @@ impl LocalCommandExecutor {
|
|||
}
|
||||
|
||||
pub(super) fn capture_command_logs(
|
||||
_task: &TaskInstanceId,
|
||||
task: &TaskInstanceId,
|
||||
stdout: &[u8],
|
||||
stderr: &[u8],
|
||||
max_log_bytes: usize,
|
||||
) -> CapturedCommandLogs {
|
||||
let stdout_truncated = stdout.len() > max_log_bytes;
|
||||
let stderr_truncated = stderr.len() > max_log_bytes;
|
||||
let stdout_start = stdout.len().saturating_sub(max_log_bytes);
|
||||
let stderr_start = stderr.len().saturating_sub(max_log_bytes);
|
||||
let mut logs = LogBuffer::new(max_log_bytes);
|
||||
logs.push(task.clone(), stdout);
|
||||
logs.push(task.clone(), stderr);
|
||||
let records = logs.records();
|
||||
let stdout_record = &records[0];
|
||||
let stderr_record = &records[1];
|
||||
debug_assert_eq!(&stdout_record.task, task);
|
||||
debug_assert_eq!(&stderr_record.task, task);
|
||||
CapturedCommandLogs {
|
||||
stdout: String::from_utf8_lossy(&stdout[stdout_start..]).into_owned(),
|
||||
stderr: String::from_utf8_lossy(&stderr[stderr_start..]).into_owned(),
|
||||
stdout_truncated,
|
||||
stderr_truncated,
|
||||
backpressured: stdout_truncated || stderr_truncated,
|
||||
stdout: String::from_utf8_lossy(&stdout_record.bytes).into_owned(),
|
||||
stderr: String::from_utf8_lossy(&stderr_record.bytes).into_owned(),
|
||||
stdout_truncated: stdout_record.truncated,
|
||||
stderr_truncated: stderr_record.truncated,
|
||||
backpressured: logs.backpressured(),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -741,7 +741,7 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn linux_backend_retains_final_log_tail_without_truncating_staged_artifact_bytes() {
|
||||
fn linux_backend_caps_logs_without_truncating_staged_artifact_bytes() {
|
||||
let invocation = CommandInvocation {
|
||||
program: "cargo".to_owned(),
|
||||
args: vec!["build".to_owned()],
|
||||
|
|
@ -774,7 +774,7 @@ mod tests {
|
|||
.unwrap();
|
||||
|
||||
assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux"));
|
||||
assert_eq!(output.stdout, "cdef");
|
||||
assert_eq!(output.stdout, "abcd");
|
||||
assert!(output.stdout_truncated);
|
||||
assert!(output.log_backpressured);
|
||||
assert_eq!(output.staged_artifact.as_ref().unwrap().size, 6);
|
||||
|
|
@ -992,7 +992,7 @@ mod tests {
|
|||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn local_command_executor_retains_each_stream_tail_and_reports_backpressure() {
|
||||
fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() {
|
||||
let executor = LocalCommandExecutor {
|
||||
node: clusterflux_core::NodeId::from("node"),
|
||||
hosted_control_plane: false,
|
||||
|
|
@ -1021,10 +1021,10 @@ mod tests {
|
|||
.unwrap();
|
||||
|
||||
assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux"));
|
||||
assert_eq!(output.stdout, "cdef");
|
||||
assert_eq!(output.stderr, "err");
|
||||
assert_eq!(output.stdout, "abcd");
|
||||
assert_eq!(output.stderr, "");
|
||||
assert!(output.stdout_truncated);
|
||||
assert!(!output.stderr_truncated);
|
||||
assert!(output.stderr_truncated);
|
||||
assert!(output.log_backpressured);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue