Skip to content

Commit

Permalink
Tweak logs and attempt to avoid races around removing nodes (#504)
Browse files Browse the repository at this point in the history 
* Tweak logs and attempt to avoid races around removing nodes

* wrapping_add in assign_id
  • Loading branch information
@jsdw
jsdw committed Oct 10, 2022
1 parent 41c93a8 commit e7d15d0
Show file tree
Hide file tree
Showing 5 changed files with 53 additions and 52 deletions.
9 changes: 7 additions & 2 deletions backend/common/src/assign_id.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ where

pub fn assign_id(&mut self, details: Details) -> Id {
let this_id = self.current_id;
self.current_id += 1;
// It's very unlikely we'll ever overflow the ID limit, but in case we do,
// a wrapping_add will almost certainly be fine:
self.current_id = self.current_id.wrapping_add(1);
self.mapping.insert(this_id, details);
this_id.into()
}
Expand All @@ -73,7 +75,10 @@ where
}

pub fn clear(&mut self) {
*self = AssignId::new();
// Leave the `current_id` as-is. Why? To avoid reusing IDs and risking
// race conditions where old messages can accidentally screw with new nodes
// that have been assigned the same ID.
self.mapping = BiMap::new();
}

pub fn iter(&self) -> impl Iterator<Item = (Id, &Details)> {
Expand Down
17 changes: 8 additions & 9 deletions backend/telemetry_core/src/aggregator/inner_loop.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ impl InnerLoop {
}

if let Err(e) = metered_tx.send(msg) {
log::error!("Cannot send message into aggregator: {}", e);
log::error!("Cannot send message into aggregator: {e}");
break;
}
}
Expand Down Expand Up @@ -386,10 +386,11 @@ impl InnerLoop {
let node_id = match self.node_ids.remove_by_right(&(shard_conn_id, local_id)) {
Some((node_id, _)) => node_id,
None => {
log::error!(
"Cannot find ID for node with shard/connectionId of {:?}/{:?}",
shard_conn_id,
local_id
// It's possible that some race between removing and disconnecting shards might lead to
// more than one remove message for the same node. This isn't really a problem, but we
// hope it won't happen so make a note if it does:
log::debug!(
"Remove: Cannot find ID for node with shard/connectionId of {shard_conn_id:?}/{local_id:?}"
);
return;
}
Expand All @@ -401,9 +402,7 @@ impl InnerLoop {
Some(id) => *id,
None => {
log::error!(
"Cannot find ID for node with shard/connectionId of {:?}/{:?}",
shard_conn_id,
local_id
"Update: Cannot find ID for node with shard/connectionId of {shard_conn_id:?}/{local_id:?}"
);
return;
}
Expand Down Expand Up @@ -606,7 +605,7 @@ impl InnerLoop {
let removed_details = match self.node_state.remove_node(node_id) {
Some(remove_details) => remove_details,
None => {
log::error!("Could not find node {:?}", node_id);
log::error!("Could not find node {node_id:?}");
return;
}
};
Expand Down
34 changes: 9 additions & 25 deletions backend/telemetry_core/src/main.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -251,21 +251,15 @@ where
break;
}
if let Err(e) = msg_info {
log::error!(
"Shutting down websocket connection: Failed to receive data: {}",
e
);
log::error!("Shutting down websocket connection: Failed to receive data: {e}");
break;
}

let msg: internal_messages::FromShardAggregator =
match bincode::options().deserialize(&bytes) {
Ok(msg) => msg,
Err(e) => {
log::error!(
"Failed to deserialize message from shard; booting it: {}",
e
);
log::error!("Failed to deserialize message from shard; booting it: {e}");
break;
}
};
Expand All @@ -292,7 +286,7 @@ where
};

if let Err(e) = tx_to_aggregator.send(aggregator_msg).await {
log::error!("Failed to send message to aggregator; closing shard: {}", e);
log::error!("Failed to send message to aggregator; closing shard: {e}");
break;
}
}
Expand Down Expand Up @@ -325,13 +319,10 @@ where
.expect("message to shard should serialize");

if let Err(e) = ws_send.send_binary(bytes).await {
log::error!("Failed to send message to aggregator; closing shard: {}", e)
log::error!("Failed to send message to aggregator; closing shard: {e}")
}
if let Err(e) = ws_send.flush().await {
log::error!(
"Failed to flush message to aggregator; closing shard: {}",
e
)
log::error!("Failed to flush message to aggregator; closing shard: {e}")
}
}

Expand Down Expand Up @@ -374,7 +365,7 @@ where
channel: tx_to_feed_conn,
};
if let Err(e) = tx_to_aggregator.send(init_msg).await {
log::error!("Error sending message to aggregator: {}", e);
log::error!("Error sending message to aggregator: {e}");
return (tx_to_aggregator, ws_send);
}

Expand All @@ -399,10 +390,7 @@ where
break;
}
if let Err(e) = msg_info {
log::error!(
"Shutting down websocket connection: Failed to receive data: {}",
e
);
log::error!("Shutting down websocket connection: Failed to receive data: {e}");
break;
}

Expand All @@ -416,16 +404,12 @@ where
let cmd = match FromFeedWebsocket::from_str(&text) {
Ok(cmd) => cmd,
Err(e) => {
log::warn!(
"Ignoring invalid command '{}' from the frontend: {}",
text,
e
);
log::warn!("Ignoring invalid command '{text}' from the frontend: {e}");
continue;
}
};
if let Err(e) = tx_to_aggregator.send(cmd).await {
log::error!("Failed to send message to aggregator; closing feed: {}", e);
log::error!("Failed to send message to aggregator; closing feed: {e}");
break;
}
}
Expand Down
22 changes: 16 additions & 6 deletions backend/telemetry_shard/src/aggregator.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -261,9 +261,14 @@ impl Aggregator {
// Remove references to this single node:
to_local_id.remove_by_id(local_id);
muted.remove(&local_id);
let _ = tx_to_telemetry_core
.send_async(FromShardAggregator::RemoveNode { local_id })
.await;

// If we're not connected to the core, don't buffer up remove messages. The core will remove
// all nodes associated with this shard anyway, so the remove message would be redundant.
if connected_to_telemetry_core {
let _ = tx_to_telemetry_core
.send_async(FromShardAggregator::RemoveNode { local_id })
.await;
}
}
ToAggregator::FromWebsocket(disconnected_conn_id, FromWebsocket::Disconnected) => {
// Find all of the local IDs corresponding to the disconnected connection ID and
Expand All @@ -280,9 +285,14 @@ impl Aggregator {
for local_id in local_ids_disconnected {
to_local_id.remove_by_id(local_id);
muted.remove(&local_id);
let _ = tx_to_telemetry_core
.send_async(FromShardAggregator::RemoveNode { local_id })
.await;

// If we're not connected to the core, don't buffer up remove messages. The core will remove
// all nodes associated with this shard anyway, so the remove message would be redundant.
if connected_to_telemetry_core {
let _ = tx_to_telemetry_core
.send_async(FromShardAggregator::RemoveNode { local_id })
.await;
}
}
}
ToAggregator::FromTelemetryCore(FromTelemetryCore::Mute {
Expand Down
23 changes: 13 additions & 10 deletions backend/telemetry_shard/src/main.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ where
close_connection: close_connection_tx.clone(),
};
if let Err(e) = tx_to_aggregator.send(init_msg).await {
log::error!("Error sending message to aggregator: {}", e);
log::error!("Shutting down websocket connection from {real_addr:?}: Error sending message to aggregator: {e}");
return (tx_to_aggregator, ws_send);
}

Expand All @@ -254,7 +254,7 @@ where
// The close channel has fired, so end the loop. `ws_recv.receive_data` is
// *not* cancel safe, but since we're closing the connection we don't care.
_ = close_connection_rx.recv_async() => {
log::info!("connection to {:?} being closed", real_addr);
log::info!("connection to {real_addr:?} being closed");
break
},
// Receive data and relay it on to our main select loop below.
Expand All @@ -263,7 +263,7 @@ where
break;
}
if let Err(e) = msg_info {
log::error!("Shutting down websocket connection: Failed to receive data: {}", e);
log::error!("Shutting down websocket connection from {real_addr:?}: Failed to receive data: {e}");
break;
}
if ws_tx_atomic.unbounded_send(bytes).is_err() {
Expand Down Expand Up @@ -292,14 +292,14 @@ where
.collect();

for &message_id in &stale_ids {
log::info!("Removing stale node with message ID {} from {:?}", message_id, real_addr);
log::info!("Removing stale node with message ID {message_id} from {real_addr:?}");
allowed_message_ids.remove(&message_id);
let _ = tx_to_aggregator.send(FromWebsocket::Remove { message_id } ).await;
}

if !stale_ids.is_empty() && allowed_message_ids.is_empty() {
// End the entire connection if no recent messages came in for any ID.
log::info!("Closing stale connection from {:?}", real_addr);
log::info!("Closing stale connection from {real_addr:?}");
break;
}
},
Expand All @@ -316,7 +316,7 @@ where
let this_bytes_per_second = rolling_total_bytes.total() / 10;
if this_bytes_per_second > bytes_per_second {
block_list.block_addr(real_addr, "Too much traffic");
log::error!("Shutting down websocket connection: Too much traffic ({}bps averaged over last 10s)", this_bytes_per_second);
log::error!("Shutting down websocket connection: Too much traffic ({this_bytes_per_second}bps averaged over last 10s)");
break;
}

Expand All @@ -327,7 +327,7 @@ where
Err(e) => {
let bytes: &[u8] = bytes.get(..512).unwrap_or_else(|| &bytes);
let msg_start = std::str::from_utf8(bytes).unwrap_or_else(|_| "INVALID UTF8");
log::warn!("Failed to parse node message ({}): {}", msg_start, e);
log::warn!("Failed to parse node message ({msg_start}): {e}");
continue;
},
#[cfg(not(debug))]
Expand All @@ -347,15 +347,15 @@ where
if let node_message::Payload::SystemConnected(info) = payload {
// Too many nodes seen on this connection? Ignore this one.
if allowed_message_ids.len() >= max_nodes_per_connection {
log::info!("Ignoring new node from {:?} (we've hit the max of {} nodes per connection)", real_addr, max_nodes_per_connection);
log::info!("Ignoring new node with ID {message_id} from {real_addr:?} (we've hit the max of {max_nodes_per_connection} nodes per connection)");
continue;
}

// Note of the message ID, allowing telemetry for it.
allowed_message_ids.insert(message_id, Instant::now());

// Tell the aggregator loop about the new node.
log::info!("Adding node with message ID {} from {:?}", message_id, real_addr);
log::info!("Adding node with message ID {message_id} from {real_addr:?}");
let _ = tx_to_aggregator.send(FromWebsocket::Add {
message_id,
ip: real_addr,
Expand All @@ -369,9 +369,12 @@ where
if let Some(last_seen) = allowed_message_ids.get_mut(&message_id) {
*last_seen = Instant::now();
if let Err(e) = tx_to_aggregator.send(FromWebsocket::Update { message_id, payload } ).await {
log::error!("Failed to send node message to aggregator: {}", e);
log::error!("Failed to send node message to aggregator: {e}");
continue;
}
} else {
log::info!("Ignoring message with ID {message_id} from {real_addr:?} (we've hit the max of {max_nodes_per_connection} nodes per connection)");
continue;
}
}
}
Expand Down

0 comments on commit e7d15d0

Please sign in to comment.