refactor: move BeeMsg handlers to their own files

Author: rustybee42

mgmtd/src/bee_msg.rs

@@ -18,12 +18,33 @@ use sqlite_check::sql;
 use std::collections::HashMap;
 use std::fmt::Display;
 
-mod buddy_group;
-mod misc;
-mod node;
-mod quota;
-mod storage_pool;
-mod target;
+mod common;
+
+mod ack;
+mod authenticate_channel;
+mod change_target_consistency_states;
+mod get_mirror_buddy_groups;
+mod get_node_capacity_pools;
+mod get_nodes;
+mod get_states_and_buddy_groups;
+mod get_storage_pools;
+mod get_target_mappings;
+mod get_target_states;
+mod heartbeat;
+mod heartbeat_request;
+mod map_targets;
+mod map_targets_resp;
+mod peer_info;
+mod refresh_capacity_pools;
+mod register_node;
+mod register_target;
+mod remove_node;
+mod remove_node_resp;
+mod request_exceeded_quota;
+mod set_channel_direct;
+mod set_mirror_buddy_groups_resp;
+mod set_storage_target_info;
+mod set_target_consistency_states;
 
 /// Msg request handler for requests where no response is expected.
 /// To handle a message, implement this and add it to the dispatch list with `=> _`.

mgmtd/src/bee_msg/ack.rs

@@ -0,0 +1,9 @@
+use super::*;
+use shared::bee_msg::misc::*;
+
+impl HandleNoResponse for Ack {
+    async fn handle(self, _ctx: &Context, req: &mut impl Request) -> Result<()> {
+        log::debug!("Ignoring Ack from {:?}: Id: {:?}", req.addr(), self.ack_id);
+        Ok(())
+    }
+}

mgmtd/src/bee_msg/authenticate_channel.rs

@@ -0,0 +1,24 @@
+use super::*;
+use shared::bee_msg::misc::*;
+
+impl HandleNoResponse for AuthenticateChannel {
+    async fn handle(self, ctx: &Context, req: &mut impl Request) -> Result<()> {
+        if let Some(ref secret) = ctx.info.auth_secret {
+            if secret == &self.auth_secret {
+                req.authenticate_connection();
+            } else {
+                log::error!(
+                    "Peer {:?} tried to authenticate stream with wrong secret",
+                    req.addr()
+                );
+            }
+        } else {
+            log::debug!(
+                "Peer {:?} tried to authenticate stream, but authentication is not required",
+                req.addr()
+            );
+        }
+
+        Ok(())
+    }
+}

mgmtd/src/bee_msg/change_target_consistency_states.rs

@@ -0,0 +1,63 @@
+use super::*;
+use common::update_last_contact_times;
+use shared::bee_msg::target::*;
+
+impl HandleWithResponse for ChangeTargetConsistencyStates {
+    type Response = ChangeTargetConsistencyStatesResp;
+
+    fn error_response() -> Self::Response {
+        ChangeTargetConsistencyStatesResp {
+            result: OpsErr::INTERNAL,
+        }
+    }
+
+    async fn handle(self, ctx: &Context, __req: &mut impl Request) -> Result<Self::Response> {
+        fail_on_pre_shutdown(ctx)?;
+
+        // self.old_states is currently completely ignored. If something reports a non-GOOD state, I
+        // see no apparent reason to that the old state matches before setting. We have the
+        // authority, whatever nodes think their old state was doesn't matter.
+
+        let changed = ctx
+            .db
+            .write_tx(move |tx| {
+                let node_type = self.node_type.try_into()?;
+
+                // Check given target Ids exist
+                db::target::validate_ids(tx, &self.target_ids, node_type)?;
+
+                // Old management updates contact time while handling this message (comes usually in
+                // every 30 seconds), so we do it as well
+                update_last_contact_times(tx, &self.target_ids, node_type)?;
+
+                let affected = db::target::update_consistency_states(
+                    tx,
+                    self.target_ids
+                        .into_iter()
+                        .zip(self.new_states.iter().copied()),
+                    node_type,
+                )?;
+
+                Ok(affected > 0)
+            })
+            .await?;
+
+        log::debug!(
+            "Updated target consistency states for {:?} nodes",
+            self.node_type
+        );
+
+        if changed {
+            notify_nodes(
+                ctx,
+                &[NodeType::Meta, NodeType::Storage, NodeType::Client],
+                &RefreshTargetStates { ack_id: "".into() },
+            )
+            .await;
+        }
+
+        Ok(ChangeTargetConsistencyStatesResp {
+            result: OpsErr::SUCCESS,
+        })
+    }
+}

mgmtd/src/bee_msg/common.rs

@@ -0,0 +1,260 @@
+use super::*;
+use crate::db::node_nic::ReplaceNic;
+use db::misc::MetaRoot;
+use rusqlite::Transaction;
+use shared::bee_msg::node::*;
+use shared::bee_msg::target::*;
+use shared::types::{NodeId, TargetId};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Processes incoming node information. Registers new nodes if config allows it
+pub(super) async fn update_node(msg: RegisterNode, ctx: &Context) -> Result<NodeId> {
+    let nics = msg.nics.clone();
+    let requested_node_id = msg.node_id;
+    let info = ctx.info;
+
+    let licensed_machines = match ctx.license.get_num_machines() {
+        Ok(n) => n,
+        Err(err) => {
+            log::debug!(
+                "Could not obtain number of licensed machines, defaulting to unlimited: {err:#}"
+            );
+            u32::MAX
+        }
+    };
+
+    let (node, meta_root, is_new) = ctx
+        .db
+        .write_tx_no_sync(move |tx| {
+            let node = if msg.node_id == 0 {
+                // No node ID given => new node
+                None
+            } else {
+                // If Some is returned, the node with node_id already exists
+                try_resolve_num_id(tx, EntityType::Node, msg.node_type, msg.node_id)?
+            };
+
+            let machine_uuid = if matches!(msg.node_type, NodeType::Meta | NodeType::Storage)
+                && !msg.machine_uuid.is_empty()
+            {
+                Some(std::str::from_utf8(&msg.machine_uuid)?)
+            } else {
+                None
+            };
+
+            if let Some(machine_uuid) = machine_uuid {
+                if db::node::count_machines(tx, machine_uuid, node.as_ref().map(|n| n.uid))?
+                    >= licensed_machines
+                {
+                    bail!("Licensed machine limit reached. Node registration denied.");
+                }
+            }
+
+            let (node, is_new) = if let Some(node) = node {
+                // Existing node, update data
+                db::node::update(tx, node.uid, msg.port, machine_uuid)?;
+
+                (node, false)
+            } else {
+                // New node, do additional checks and insert data
+
+                // Check node registration is allowed. This should ignore registering client
+                // nodes.
+                if msg.node_type != NodeType::Client && info.user_config.registration_disable {
+                    bail!("Registration of new nodes is not allowed");
+                }
+
+                let new_alias = if msg.node_type == NodeType::Client {
+                    // In versions prior to 8.0 the string node ID generated by the client
+                    // started with a number which is not allowed by the new alias schema.
+                    // As part of BeeGFS 8 the nodeID generated for each client mount was
+                    // updated to no longer start with a number, thus it is unlikely this
+                    // would happen unless BeeGFS 8 was mounted by a BeeGFS 7 client.
+
+                    let new_alias = String::from_utf8(msg.node_alias)
+                        .ok()
+                        .and_then(|s| Alias::try_from(s).ok());
+
+                    if new_alias.is_none() {
+                        log::warn!(
+                            "Unable to use alias requested by client (possibly the\
+client version < 8.0)"
+                        );
+                    }
+                    new_alias
+                } else {
+                    None
+                };
+
+                // Insert new node entry
+                let node = db::node::insert(tx, msg.node_id, new_alias, msg.node_type, msg.port)?;
+
+                // if this is a meta node, auto-add a corresponding meta target after the node.
+                if msg.node_type == NodeType::Meta {
+                    // Convert the NodeID to a TargetID. Due to the difference in bitsize, meta
+                    // node IDs are not allowed to be bigger than u16
+                    let Ok(target_id) = TargetId::try_from(node.num_id()) else {
+                        bail!(
+                            "{} is not a valid numeric meta node id\
+(must be between 1 and 65535)",
+                            node.num_id()
+                        );
+                    };
+
+                    db::target::insert_meta(tx, target_id, None)?;
+                }
+
+                (node, true)
+            };
+
+            // Update the corresponding nic lists
+            db::node_nic::replace(
+                tx,
+                node.uid,
+                msg.nics.iter().map(|e| ReplaceNic {
+                    nic_type: e.nic_type,
+                    addr: &e.addr,
+                    name: String::from_utf8_lossy(&e.name),
+                }),
+            )?;
+
+            let meta_root = match node.node_type() {
+                // In case this is a meta node, the requester expects info about the meta
+                // root
+                NodeType::Meta => db::misc::get_meta_root(tx)?,
+                _ => MetaRoot::Unknown,
+            };
+
+            Ok((node, meta_root, is_new))
+        })
+        .await?;
+
+    ctx.conn.replace_node_addrs(
+        node.uid,
+        nics.clone()
+            .into_iter()
+            .map(|e| SocketAddr::new(e.addr, msg.port))
+            .collect::<Arc<_>>(),
+    );
+
+    if is_new {
+        log::info!("Registered new node {node} (Requested Numeric Id: {requested_node_id})",);
+    } else {
+        log::debug!("Updated node {node} node",);
+    }
+
+    let node_num_id = node.num_id();
+
+    // notify all nodes
+    notify_nodes(
+        ctx,
+        match node.node_type() {
+            NodeType::Meta => &[NodeType::Meta, NodeType::Client],
+            NodeType::Storage => &[NodeType::Meta, NodeType::Storage, NodeType::Client],
+            NodeType::Client => &[NodeType::Meta],
+            _ => &[],
+        },
+        &Heartbeat {
+            instance_version: 0,
+            nic_list_version: 0,
+            node_type: node.node_type(),
+            node_alias: String::from(node.alias).into_bytes(),
+            ack_id: "".into(),
+            node_num_id,
+            root_num_id: match meta_root {
+                MetaRoot::Unknown => 0,
+                MetaRoot::Normal(node_id, _) => node_id,
+                MetaRoot::Mirrored(group_id) => group_id.into(),
+            },
+            is_root_mirrored: match meta_root {
+                MetaRoot::Unknown | MetaRoot::Normal(_, _) => 0,
+                MetaRoot::Mirrored(_) => 1,
+            },
+            port: msg.port,
+            port_tcp_unused: msg.port,
+            nic_list: nics,
+            machine_uuid: vec![], // No need for the other nodes to know machine UUIDs
+        },
+    )
+    .await;
+
+    Ok(node_num_id)
+}
+
+pub(super) fn get_targets_with_states(
+    tx: &Transaction,
+    pre_shutdown: bool,
+    node_type: NodeTypeServer,
+    node_offline_timeout: Duration,
+) -> Result<Vec<(TargetId, TargetConsistencyState, TargetReachabilityState)>> {
+    let targets = tx.query_map_collect(
+        sql!(
+            "SELECT t.target_id, t.consistency,
+                (UNIXEPOCH('now') - UNIXEPOCH(t.last_update)), gp.p_target_id, gs.s_target_id
+            FROM targets AS t
+            INNER JOIN nodes AS n USING(node_type, node_id)
+            LEFT JOIN buddy_groups AS gp ON gp.p_target_id = t.target_id AND gp.node_type = t.node_type
+            LEFT JOIN buddy_groups AS gs ON gs.s_target_id = t.target_id AND gs.node_type = t.node_type
+            WHERE t.node_type = ?1"
+        ),
+        [node_type.sql_variant()],
+        |row| {
+            let is_primary = row.get::<_, Option<TargetId>>(3)?.is_some();
+            let is_secondary = row.get::<_, Option<TargetId>>(4)?.is_some();
+
+            Ok((
+                row.get(0)?,
+                TargetConsistencyState::from_row(row, 1)?,
+                if !pre_shutdown || is_secondary {
+                    let age = Duration::from_secs(row.get(2)?);
+
+                    // We never want to report a primary node of a buddy group as offline since this
+                    // is considered invalid. Instead we just report ProbablyOffline and wait for the switchover.
+                    if !is_primary && age > node_offline_timeout {
+                        TargetReachabilityState::Offline
+                    } else if age > node_offline_timeout / 2 {
+                        TargetReachabilityState::ProbablyOffline
+                    } else {
+                        TargetReachabilityState::Online
+                    }
+                } else {
+                    TargetReachabilityState::ProbablyOffline
+                },
+            ))
+        },
+    )?;
+
+    Ok(targets)
+}
+
+/// Updates the `last_contact` time for all the nodes belonging to the passed targets and the
+/// targets `last_update` times themselves.
+pub(super) fn update_last_contact_times(
+    tx: &Transaction,
+    target_ids: &[TargetId],
+    node_type: NodeTypeServer,
+) -> Result<()> {
+    let target_ids_param = sqlite::rarray_param(target_ids.iter().copied());
+
+    tx.execute_cached(
+        sql!(
+            "UPDATE nodes AS n SET last_contact = DATETIME('now')
+            WHERE n.node_uid IN (
+            SELECT DISTINCT node_uid FROM targets_ext
+            WHERE target_id IN rarray(?1) AND node_type = ?2)"
+        ),
+        rusqlite::params![&target_ids_param, node_type.sql_variant()],
+    )?;
+
+    tx.execute_cached(
+        sql!(
+            "UPDATE targets SET last_update = DATETIME('now')
+            WHERE target_id IN rarray(?1) AND node_type = ?2"
+        ),
+        rusqlite::params![&target_ids_param, node_type.sql_variant()],
+    )?;
+
+    Ok(())
+}