pub struct TreeWorker { /* private fields */ }Expand description
Worker that distributes command execution across a cluster topology.
TreeWorker uses PropagationTreeRouter to determine routing through
gateway nodes, and PropagationChannel to communicate with gateways.
Nodes directly reachable (same hop as root) are executed via direct
workers (SSH or local).
Mirrors ClusterShell’s TreeWorker class.
Implementations§
Source§impl TreeWorker
impl TreeWorker
Sourcepub fn new(nodes: NodeSet, command: String, config: TreeWorkerConfig) -> Self
pub fn new(nodes: NodeSet, command: String, config: TreeWorkerConfig) -> Self
Create a new TreeWorker for command execution.
§Arguments
nodes- Target nodes to execute on.command- Command to execute on each node.config- Tree worker configuration.
Sourcepub fn new_copy(
nodes: NodeSet,
config: TreeWorkerConfig,
) -> Result<Self, WorkerError>
pub fn new_copy( nodes: NodeSet, config: TreeWorkerConfig, ) -> Result<Self, WorkerError>
Create a new TreeWorker for a copy or reverse-copy (rcopy) operation.
config.source must be set (mirrors the upstream ValueError
raised when neither command nor source is provided).
§Errors
Returns WorkerError::General if config.source is None.
Sourcepub fn set_router(&mut self, router: PropagationTreeRouter)
pub fn set_router(&mut self, router: PropagationTreeRouter)
Set the topology router for this worker.
Must be called before start(). The router determines how commands
are distributed across gateway nodes.
Sourcepub fn set_topology(
&mut self,
root: &str,
topology: &TopologyTree,
) -> Result<(), WorkerError>
pub fn set_topology( &mut self, root: &str, topology: &TopologyTree, ) -> Result<(), WorkerError>
Sourcepub fn state(&self) -> WorkerState
pub fn state(&self) -> WorkerState
Get the current state.
Sourcepub fn events(&self) -> &[TreeWorkerEvent]
pub fn events(&self) -> &[TreeWorkerEvent]
Get collected events from remote nodes.
Sourcepub fn drain_events(&mut self) -> Vec<TreeWorkerEvent>
pub fn drain_events(&mut self) -> Vec<TreeWorkerEvent>
Drain collected events.
Sourcepub fn num_timeout(&self) -> usize
pub fn num_timeout(&self) -> usize
Get the number of timed out nodes.
Sourcepub fn timeout_nodes(&self) -> &HashSet<String>
pub fn timeout_nodes(&self) -> &HashSet<String>
Get timed out nodes.
Sourcepub fn has_timeout(&self) -> bool
pub fn has_timeout(&self) -> bool
Whether any node has timed out.
Sourcepub fn is_done(&self) -> bool
pub fn is_done(&self) -> bool
Check if the worker is done (all targets completed or timed out).
Sourcepub fn start(&mut self) -> Result<Vec<(NodeSet, NodeSet)>, WorkerError>
pub fn start(&mut self) -> Result<Vec<(NodeSet, NodeSet)>, WorkerError>
Start the tree worker — distribute work across gateways.
This sets up propagation channels for gateway nodes and prepares direct execution for leaf nodes.
Sourcepub fn start_gateway(
&mut self,
gateway: &str,
topology_bytes: &[u8],
) -> Result<(), WorkerError>
pub fn start_gateway( &mut self, gateway: &str, topology_bytes: &[u8], ) -> Result<(), WorkerError>
Start a gateway’s propagation channel with topology data.
§Arguments
gateway- The gateway node name.topology_bytes- Serialized topology data to send to gateway.
Sourcepub fn sync_gateway_pickups(&mut self, gateway: &str)
pub fn sync_gateway_pickups(&mut self, gateway: &str)
Emit pickups for gateway targets whose control message has actually been sent on the wire.
Mirrors PropagationChannel._send_ctl firing _emit_pickup
after channel.send (upstream ClusterShell #594): once channel
setup is complete, a queued control message has left the local
process and its targets can no longer be rerouted. Call this
after the channel processes the configuration ACK (and after
TreeWorker::start_gateway); it is a no-op while the control
message is still queued behind an un-ACKed configuration.
Sourcepub fn emit_pickup(&mut self, node: &str)
pub fn emit_pickup(&mut self, node: &str)
Fire ev_pickup for a node, or queue it until ev_start fires.
Mirrors upstream TreeWorker._emit_pickup (#594): direct child
workers route pickups through MetaWorkerEventHandler.ev_pickup;
gateway-routed targets route them through the propagation channel
once the control message leaves the local process (see
TreeWorker::sync_gateway_pickups). Emission is skipped once
the worker has been aborted.
Sourcepub fn notify_child_start(&mut self)
pub fn notify_child_start(&mut self)
Notify the meta worker that a direct child worker has started.
MetaWorkerEventHandler.ev_start equivalent: bumps the started
count and re-evaluates initialization.
Sourcepub fn check_ini(&mut self)
pub fn check_ini(&mut self)
Fire ev_start once all direct child workers have started, then
flush pickup events buffered during launch.
Mirrors upstream TreeWorker._check_ini (#594): this part runs
once. If the worker is aborted on (or before) ev_start, the
flush stops at the first aborted check and no pickups are
emitted.
Sourcepub fn mark_unreachable(&mut self, node: &str)
pub fn mark_unreachable(&mut self, node: &str)
Mark a node as unreachable in the router, so later dispatches and
relaunches avoid it (upstream PropagationTreeRouter .mark_unreachable pass-through used before _relaunch).
Sourcepub fn relaunch(
&mut self,
previous_gateway: &str,
) -> Result<Vec<(NodeSet, NodeSet)>, WorkerError>
pub fn relaunch( &mut self, previous_gateway: &str, ) -> Result<Vec<(NodeSet, NodeSet)>, WorkerError>
Redistribute and relaunch targets that were running on a failed
gateway (upstream TreeWorker._relaunch).
Emits a reroute routing event BEFORE re-dispatching, so a
rerouted target’s ev_pickup can only fire afterwards — and,
because pickup requires the control message to have been sent,
exactly once (#594). Upstream only reroutes targets whose control
message never left the failed channel; callers should do the same
(a target already picked up on the failed gateway must not be
relaunched).
Returns the new (gateway, targets) distribution; entries with gateway == targets are directly reachable and must be handled by the caller like any direct target.
Sourcepub fn process_result(
&mut self,
gateway: &str,
result: PropagationResult,
) -> Result<bool, WorkerError>
pub fn process_result( &mut self, gateway: &str, result: PropagationResult, ) -> Result<bool, WorkerError>
Process a propagation result from a gateway.
Call this when the gateway channel receives results. Returns true if the worker is now complete.
Sourcepub fn record_direct_close(&mut self, node: &str, rc: i32)
pub fn record_direct_close(&mut self, node: &str, rc: i32)
Record completion of a direct (non-gateway) node.
Called when a direct SSH/local worker reports node completion.
Sourcepub fn record_direct_timeout(&mut self, node: &str)
pub fn record_direct_timeout(&mut self, node: &str)
Record a direct node timeout.
Sourcepub fn write_remote(&mut self, buf: &[u8])
pub fn write_remote(&mut self, buf: &[u8])
Write data to all active gateway channels.
Like upstream TreeWorker._write_remote, this never fires
ev_pickup: write traffic on an already-picked-up node must not
double-fire (#594).
Sourcepub fn set_write_eof_remote(&mut self)
pub fn set_write_eof_remote(&mut self)
Send write EOF to all active gateway channels.
Like upstream TreeWorker._set_write_eof_remote, this never
fires ev_pickup (#594).
Sourcepub fn abort(&mut self)
pub fn abort(&mut self)
Abort the tree worker.
Marks all remaining gateway targets as failed with EX_PROTOCOL (76).
Also stops pending and future pickup events (#594): buffered
pickups are dropped and TreeWorker::emit_pickup becomes a no-op.
Sourcepub fn is_aborted(&self) -> bool
pub fn is_aborted(&self) -> bool
Whether the worker has been aborted (upstream _aborted).
Sourcepub fn is_initialized(&self) -> bool
pub fn is_initialized(&self) -> bool
Whether ev_start has fired (upstream _initialized).
Sourcepub fn pending_pickup_count(&self) -> usize
pub fn pending_pickup_count(&self) -> usize
Number of pickup events currently buffered before ev_start.
Sourcepub fn config(&self) -> &TreeWorkerConfig
pub fn config(&self) -> &TreeWorkerConfig
Get the worker configuration.
Sourcepub fn stderr(&self) -> bool
pub fn stderr(&self) -> bool
Whether stderr is captured separately from stdout.
Always true for copy/rcopy workers: internal tar/scp errors are kept off stdout (upstream ClusterShell #622).
Sourcepub fn report_gateway_stderr(
&mut self,
gateway: &str,
nodes: &NodeSet,
buf: &[u8],
)
pub fn report_gateway_stderr( &mut self, gateway: &str, nodes: &NodeSet, buf: &[u8], )
Report a gateway-originated error buffer as stderr lines
attributed to nodes, as seen from gateway.
Mirrors PropagationChannel._report_stderr delivering to this
meta worker (upstream ClusterShell 90d3195): a trailing newline
is appended first so an empty buffer still yields one (empty)
line — splitlines() would drop it entirely (#249).
Sourcepub fn report_gateway_error(&mut self, gateway: &str, reason: &str)
pub fn report_gateway_error(&mut self, gateway: &str, reason: &str)
Report an error message received from a gateway before channel setup completed (e.g. the gateway rejected the configuration payload).
Mirrors upstream 90d3195: such errors used to be logged and
dropped; they are now reported as stderr of the gateway node
itself, so they reach ev_read/ev_error on this worker
instead of vanishing.
Sourcepub fn process_channel_error(
&mut self,
gateway: &str,
err: &MessageProcessingError,
)
pub fn process_channel_error( &mut self, gateway: &str, err: &MessageProcessingError, )
Handle a MessageProcessingError raised while parsing traffic
from gateway (e.g. a malformed or unknown gateway message).
Mirrors the upstream initiator-side Channel._do_read path
(commit a027c3e): the error is fed back as a stderr message from
the gateway node instead of crashing the worker. The worker
state is left untouched; closing the failed channel is the
driver’s responsibility.
Sourcepub fn on_remote_node_msgline(
&mut self,
node: &str,
msg: &[u8],
sname: &str,
gateway: &str,
)
pub fn on_remote_node_msgline( &mut self, node: &str, msg: &[u8], sname: &str, gateway: &str, )
Process a message line received from a remote node through a
gateway (upstream TreeWorker._on_remote_node_msgline).
For rcopy workers, stdout carries the base64-encoded tar stream
and is accumulated for later extraction instead of being reported
(partial base64 decoding requires a multiple of 4 characters;
trailing chars are kept for the next message). Everything else is
recorded as a stdout/stderr event.
Sourcepub fn finalize_rcopy_node(
&mut self,
node: &str,
gateway: &str,
) -> Result<usize, WorkerError>
pub fn finalize_rcopy_node( &mut self, node: &str, gateway: &str, ) -> Result<usize, WorkerError>
Finalize rcopy for a node whose remote side has closed: extract the accumulated tar stream into the destination directory.
Mirrors the reverse-copy branch of upstream
TreeWorker._on_remote_node_close. Extraction uses
TarExtractFilter::FullyTrusted semantics (PEP 706, upstream
d2fcd16). Any failure — corrupt archive, I/O error — is reported
as a stderr event attributed to the node carrying the gateway’s
message (upstream 973870b), never panicked or silently dropped.
Returns the number of extracted bytes on success; errors are reported through events, so this only fails on worker misuse (missing destination).
Sourcepub fn active_gateway_count(&self) -> usize
pub fn active_gateway_count(&self) -> usize
Get the number of active gateways.
Sourcepub fn active_gateways(&self) -> Vec<&str>
pub fn active_gateways(&self) -> Vec<&str>
Get active gateway names.