tatara_engine/domain/
scheduler.rs

1use anyhow::Result;
2use std::sync::Arc;
3use std::time::Duration;
4use tracing::{debug, info, warn};
5
6use crate::client::executor::Executor;
7use crate::domain::evaluation::Evaluator;
8use crate::domain::store_adapter::ClusterStoreAdapter;
9
10/// Runs the scheduling loop: evaluates pending jobs and dispatches allocations.
11///
12/// The scheduler only proposes allocations when this node is the Raft leader.
13/// This prevents duplicate allocations across a multi-node cluster.
14pub struct Scheduler {
15    evaluator: Evaluator,
16    executor: Arc<Executor>,
17    store: Arc<ClusterStoreAdapter>,
18    eval_interval: Duration,
19}
20
21impl Scheduler {
22    pub fn new(
23        store: Arc<ClusterStoreAdapter>,
24        executor: Arc<Executor>,
25        eval_interval_secs: u64,
26    ) -> Self {
27        Self {
28            evaluator: Evaluator::new(store.clone()),
29            executor,
30            store,
31            eval_interval: Duration::from_secs(eval_interval_secs),
32        }
33    }
34
35    /// Run the scheduler loop until cancelled.
36    pub async fn run(&self) -> Result<()> {
37        info!("Scheduler started (interval: {:?})", self.eval_interval);
38        let mut interval = tokio::time::interval(self.eval_interval);
39
40        loop {
41            interval.tick().await;
42
43            // Leader-affinity: only the Raft leader schedules new allocations.
44            if !self.store.is_leader().await {
45                debug!("Not leader, skipping scheduling tick");
46                continue;
47            }
48
49            match self.evaluator.evaluate().await {
50                Ok(allocations) => {
51                    for alloc in allocations {
52                        info!(
53                            alloc_id = %alloc.id,
54                            job_id = %alloc.job_id,
55                            group = %alloc.group_name,
56                            node = %alloc.node_id,
57                            "Created allocation"
58                        );
59                        if let Err(e) = self.executor.start_allocation(alloc).await {
60                            warn!(error = %e, "Failed to start allocation");
61                        }
62                    }
63                }
64                Err(e) => {
65                    warn!(error = %e, "Evaluation cycle failed");
66                }
67            }
68        }
69    }
70}
tatara_engine/domain/scheduler.rs

tatara_engine/domain/
scheduler.rs
