Skip to content

Commit 901b469

Browse files
andrasbacsaiandrasbacsai
committed
feat(coold): add command and response dispatch logging
1 parent cb33efa commit 901b469

4 files changed

Lines changed: 158 additions & 14 deletions

File tree

‎coold/src/grpc/handlers.rs‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,9 @@ pub async fn handle(
2727
corrosion: &CorrosionClient,
2828
tx: mpsc::Sender<ClientMsg>,
2929
) {
30+
let command_type = command_type(&command);
31+
info!(%request_id, %command_type, "coold command received");
32+
3033
match command {
3134
server_msg::Command::Ping(_) => {
3235
debug!(%request_id, "ping command reached handler after fast-path; ignoring");
@@ -408,11 +411,71 @@ fn error_body(error: anyhow::Error) -> response::Body {
408411

409412
async fn send_response(tx: &mpsc::Sender<ClientMsg>, response: Response) {
410413
let request_id = response.request_id.clone();
414+
let response_type = response_body_type(response.body.as_ref());
411415
let msg = ClientMsg {
412416
payload: Some(client_msg::Payload::Response(response)),
413417
};
418+
419+
info!(%request_id, %response_type, "coold response enqueueing");
414420
if let Err(e) = tx.send(msg).await {
415-
warn!(%request_id, error = %e, "failed to enqueue response");
421+
warn!(%request_id, %response_type, error = %e, "failed to enqueue response");
422+
} else {
423+
info!(%request_id, %response_type, "coold response enqueued");
424+
}
425+
}
426+
427+
fn command_type(command: &server_msg::Command) -> &'static str {
428+
match command {
429+
server_msg::Command::Ping(_) => "ping",
430+
server_msg::Command::ImagesPull(_) => "images.pull",
431+
server_msg::Command::ImagesList(_) => "images.list",
432+
server_msg::Command::ImagesDelete(_) => "images.delete",
433+
server_msg::Command::ContainersCreate(_) => "containers.create",
434+
server_msg::Command::ContainersStart(_) => "containers.start",
435+
server_msg::Command::ContainersStop(_) => "containers.stop",
436+
server_msg::Command::ContainersRestart(_) => "containers.restart",
437+
server_msg::Command::ContainersDelete(_) => "containers.delete",
438+
server_msg::Command::ContainersInspect(_) => "containers.inspect",
439+
server_msg::Command::ContainersList(_) => "containers.list",
440+
server_msg::Command::ContainersLogs(_) => "containers.logs",
441+
server_msg::Command::ContainersExec(_) => "containers.exec",
442+
server_msg::Command::ContainersHealthcheckRun(_) => "containers.healthcheck.run",
443+
server_msg::Command::IngressApply(_) => "ingress.apply",
444+
server_msg::Command::IngressStop(_) => "ingress.stop",
445+
server_msg::Command::FirewallAllow(_) => "firewall.allow",
446+
server_msg::Command::FirewallRevoke(_) => "firewall.revoke",
447+
server_msg::Command::FirewallList(_) => "firewall.list",
448+
server_msg::Command::FirewallReconcile(_) => "firewall.reconcile",
449+
server_msg::Command::CooldLogs(_) => "coold.logs",
450+
server_msg::Command::CorrosionTables(_) => "corrosion.tables",
451+
}
452+
}
453+
454+
fn response_body_type(body: Option<&response::Body>) -> &'static str {
455+
match body {
456+
Some(response::Body::ImagesPull(_)) => "images.pull",
457+
Some(response::Body::ImagesList(_)) => "images.list",
458+
Some(response::Body::ImagesDelete(_)) => "images.delete",
459+
Some(response::Body::ContainersCreate(_)) => "containers.create",
460+
Some(response::Body::ContainersStart(_)) => "containers.start",
461+
Some(response::Body::ContainersStop(_)) => "containers.stop",
462+
Some(response::Body::ContainersRestart(_)) => "containers.restart",
463+
Some(response::Body::ContainersDelete(_)) => "containers.delete",
464+
Some(response::Body::ContainersInspect(_)) => "containers.inspect",
465+
Some(response::Body::ContainersList(_)) => "containers.list",
466+
Some(response::Body::ContainersLogs(_)) => "containers.logs",
467+
Some(response::Body::ContainersExec(_)) => "containers.exec",
468+
Some(response::Body::ContainersHealthcheckRun(_)) => "containers.healthcheck.run",
469+
Some(response::Body::IngressApply(_)) => "ingress.apply",
470+
Some(response::Body::IngressStop(_)) => "ingress.stop",
471+
Some(response::Body::FirewallAllow(_)) => "firewall.allow",
472+
Some(response::Body::FirewallRevoke(_)) => "firewall.revoke",
473+
Some(response::Body::FirewallList(_)) => "firewall.list",
474+
Some(response::Body::FirewallReconcile(_)) => "firewall.reconcile",
475+
Some(response::Body::CooldLogs(_)) => "coold.logs",
476+
Some(response::Body::CorrosionTables(_)) => "corrosion.tables",
477+
Some(response::Body::Error(_)) => "error",
478+
None => "none",
416479
}
417480
}
418481

‎flux/src/envelope.rs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,34 @@ pub enum CommandPayload {
151151
},
152152
}
153153

154+
impl CommandPayload {
155+
pub fn kind(&self) -> &'static str {
156+
match self {
157+
Self::ImagesPull { .. } => "images.pull",
158+
Self::ImagesList => "images.list",
159+
Self::ImagesDelete { .. } => "images.delete",
160+
Self::ContainersCreate { .. } => "containers.create",
161+
Self::ContainersStart { .. } => "containers.start",
162+
Self::ContainersStop { .. } => "containers.stop",
163+
Self::ContainersRestart { .. } => "containers.restart",
164+
Self::ContainersDelete { .. } => "containers.delete",
165+
Self::ContainersInspect { .. } => "containers.inspect",
166+
Self::ContainersList => "containers.list",
167+
Self::ContainersLogs { .. } => "containers.logs",
168+
Self::ContainersExec { .. } => "containers.exec",
169+
Self::ContainersHealthcheckRun { .. } => "containers.healthcheck.run",
170+
Self::ApplyIngress { .. } => "ingress.apply",
171+
Self::StopIngress { .. } => "ingress.stop",
172+
Self::FirewallAllow { .. } => "firewall.allow",
173+
Self::FirewallRevoke { .. } => "firewall.revoke",
174+
Self::FirewallList { .. } => "firewall.list",
175+
Self::FirewallReconcile => "firewall.reconcile",
176+
Self::CooldLogs { .. } => "coold.logs",
177+
Self::CorrosionTables { .. } => "corrosion.tables",
178+
}
179+
}
180+
}
181+
154182
#[derive(Debug, Deserialize)]
155183
pub struct PortMapping {
156184
pub host_ip: String,

‎flux/src/main.rs‎

Lines changed: 41 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ mod grpc_server {
8585

8686
use coolify_proto::agent::v1::{
8787
agent_server::{Agent, AgentServer},
88-
client_msg, ClientMsg, ServerMsg,
88+
client_msg, response, ClientMsg, ServerMsg,
8989
};
9090

9191
use crate::{
@@ -319,7 +319,7 @@ mod grpc_server {
319319
Ok(ClientMsg {
320320
payload: Some(client_msg::Payload::Response(resp)),
321321
}) => {
322-
deliver_response(&pending, resp);
322+
deliver_response(&pending, &host_id_clone, resp);
323323
}
324324
Ok(ClientMsg {
325325
payload: Some(client_msg::Payload::ResourceStatusUpdate(update)),
@@ -384,12 +384,19 @@ mod grpc_server {
384384

385385
/// Map a coold gRPC `Response` to `ResponseData` and hand it to the
386386
/// pending entry. Unknown request_id → drop.
387-
fn deliver_response(pending: &Pending, resp: coolify_proto::agent::v1::Response) {
387+
fn deliver_response(
388+
pending: &Pending,
389+
host_id: &str,
390+
resp: coolify_proto::agent::v1::Response,
391+
) {
388392
let request_id = resp.request_id.clone();
393+
let response_type = response_body_type(resp.body.as_ref());
394+
info!(%request_id, %host_id, %response_type, "coold response received from stream");
395+
389396
let kind = match pending.get(&request_id) {
390397
Some(e) => e.kind,
391398
None => {
392-
warn!(%request_id, "response for unknown request_id; dropping");
399+
warn!(%request_id, %host_id, %response_type, "response for unknown request_id; dropping");
393400
return;
394401
}
395402
};
@@ -399,12 +406,41 @@ mod grpc_server {
399406
body: resp.body,
400407
};
401408
let Some(body) = ResponseBody::try_from_proto(resp) else {
402-
warn!(%request_id, ?kind, "unsupported response body; dropping");
409+
warn!(%request_id, %host_id, %response_type, ?kind, "unsupported response body; dropping");
403410
return;
404411
};
405412
let data = ResponseData::Coold(body);
406413

407414
pending.deliver(&request_id, data);
415+
info!(%request_id, %host_id, %response_type, ?kind, "coold response delivered to pending dispatch");
416+
}
417+
418+
fn response_body_type(body: Option<&response::Body>) -> &'static str {
419+
match body {
420+
Some(response::Body::ImagesPull(_)) => "images.pull",
421+
Some(response::Body::ImagesList(_)) => "images.list",
422+
Some(response::Body::ImagesDelete(_)) => "images.delete",
423+
Some(response::Body::ContainersCreate(_)) => "containers.create",
424+
Some(response::Body::ContainersStart(_)) => "containers.start",
425+
Some(response::Body::ContainersStop(_)) => "containers.stop",
426+
Some(response::Body::ContainersRestart(_)) => "containers.restart",
427+
Some(response::Body::ContainersDelete(_)) => "containers.delete",
428+
Some(response::Body::ContainersInspect(_)) => "containers.inspect",
429+
Some(response::Body::ContainersList(_)) => "containers.list",
430+
Some(response::Body::ContainersLogs(_)) => "containers.logs",
431+
Some(response::Body::ContainersExec(_)) => "containers.exec",
432+
Some(response::Body::ContainersHealthcheckRun(_)) => "containers.healthcheck.run",
433+
Some(response::Body::IngressApply(_)) => "ingress.apply",
434+
Some(response::Body::IngressStop(_)) => "ingress.stop",
435+
Some(response::Body::FirewallAllow(_)) => "firewall.allow",
436+
Some(response::Body::FirewallRevoke(_)) => "firewall.revoke",
437+
Some(response::Body::FirewallList(_)) => "firewall.list",
438+
Some(response::Body::FirewallReconcile(_)) => "firewall.reconcile",
439+
Some(response::Body::CooldLogs(_)) => "coold.logs",
440+
Some(response::Body::CorrosionTables(_)) => "corrosion.tables",
441+
Some(response::Body::Error(_)) => "error",
442+
None => "none",
443+
}
408444
}
409445

410446
pub(super) fn coold_status_update(

‎flux/src/unix_bridge.rs‎

Lines changed: 25 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,9 @@ async fn stream_inventory(State(st): State<AppState>) -> impl IntoResponse {
143143
async fn coold_dispatch(State(st): State<AppState>, Json(env): Json<DispatchEnvelope>) -> Response {
144144
let request_id = env.request_id.clone();
145145
let host_id = env.host_id.clone();
146+
let command_type = env.command.kind();
147+
148+
info!(%request_id, %host_id, %command_type, "coold dispatch received");
146149

147150
match route_coold(&st.streams, env) {
148151
RouteOutcome::SendCoold {
@@ -177,10 +180,13 @@ async fn coold_dispatch(State(st): State<AppState>, Json(env): Json<DispatchEnve
177180
};
178181
if tx.send(msg).await.is_err() {
179182
st.pending.remove(&request_id);
183+
warn!(%request_id, host_id = %target, %command_type, "coold dispatch stream send failed");
180184
return coold_err(&request_id, 503, "host stream send failed");
181185
}
182186

183-
await_coold(&request_id, rx, st.dispatch_timeout).await
187+
info!(%request_id, host_id = %target, %command_type, "coold dispatch forwarded to stream");
188+
189+
await_coold(&request_id, &target, command_type, rx, st.dispatch_timeout).await
184190
}
185191
RouteOutcome::PushError { code, message } => {
186192
warn!(%request_id, %host_id, %code, %message, "coold dispatch rejected");
@@ -191,19 +197,30 @@ async fn coold_dispatch(State(st): State<AppState>, Json(env): Json<DispatchEnve
191197

192198
async fn await_coold(
193199
request_id: &str,
200+
host_id: &str,
201+
command_type: &str,
194202
rx: oneshot::Receiver<ResponseData>,
195203
timeout: Duration,
196204
) -> Response {
197205
match tokio::time::timeout(timeout, rx).await {
198-
Ok(Ok(ResponseData::Coold(body))) => Json(ResponseEnvelope {
199-
request_id: request_id.to_owned(),
200-
body,
201-
})
202-
.into_response(),
206+
Ok(Ok(ResponseData::Coold(body))) => {
207+
info!(%request_id, %host_id, %command_type, "coold dispatch response returned to caller");
208+
Json(ResponseEnvelope {
209+
request_id: request_id.to_owned(),
210+
body,
211+
})
212+
.into_response()
213+
}
203214
// Sink dropped without a value (sweeper evicted on timeout, or the
204215
// entry was removed by a send-failure path above).
205-
Ok(Err(_)) => coold_err(request_id, 504, "dispatch timeout"),
206-
Err(_) => coold_err(request_id, 504, "dispatch timeout"),
216+
Ok(Err(_)) => {
217+
warn!(%request_id, %host_id, %command_type, "coold dispatch response sink dropped");
218+
coold_err(request_id, 504, "dispatch timeout")
219+
}
220+
Err(_) => {
221+
warn!(%request_id, %host_id, %command_type, "coold dispatch await timed out");
222+
coold_err(request_id, 504, "dispatch timeout")
223+
}
207224
}
208225
}
209226

0 commit comments

Comments
 (0)