operon 0.7.0

A workflow engine for parallel, incremental scheduling of DAG-defined multiplex tasks.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use async_trait::async_trait;
use uuid::Uuid;

use crate::meta_storage::{MetaBackend, MetaClientApi, MetaConnApi};
use crate::scheduler::SchedulerError;
use crate::scheduler::context::SchedulerContext;
use crate::scheduler::states::running::RunningState;
use crate::scheduler::states::{NextState, SchedulerTransition, TransitionState};
use crate::schema::{RunFootprint, RunState};
use crate::service::OperonService;
use crate::storage::OperonStorage;

pub(super) struct StartTransition<Svc, Sto, MSto>
where
    Svc: OperonService,
    Sto: OperonStorage,
    MSto: MetaBackend,
{
    ctx: SchedulerContext<Svc, Sto, MSto>,
    channel_size: usize,
    run_id: Uuid,
    clean: bool,
}

impl<Svc, Sto, MSto> StartTransition<Svc, Sto, MSto>
where
    Svc: OperonService,
    Sto: OperonStorage,
    MSto: MetaBackend,
{
    pub(super) fn new(
        ctx: SchedulerContext<Svc, Sto, MSto>,
        channel_size: usize,
        run_id: Uuid,
        clean: bool,
    ) -> Self {
        Self {
            ctx,
            channel_size,
            run_id,
            clean,
        }
    }

    pub(super) fn state(
        ctx: SchedulerContext<Svc, Sto, MSto>,
        channel_size: usize,
        run_id: Uuid,
        clean: bool,
    ) -> TransitionState<SchedulerError<Svc::Error, Sto::Error, MSto::Error>> {
        TransitionState::new(Self::new(ctx, channel_size, run_id, clean))
    }

    fn into_running(self, execution_id: Uuid) -> RunningState<Svc, Sto, MSto> {
        RunningState::new(
            self.ctx,
            self.channel_size,
            self.run_id,
            execution_id,
            self.clean,
        )
    }
}

#[async_trait]
impl<Svc, Sto, MSto> SchedulerTransition for StartTransition<Svc, Sto, MSto>
where
    Svc: OperonService,
    Sto: OperonStorage,
    MSto: MetaBackend,
{
    type Error = SchedulerError<Svc::Error, Sto::Error, MSto::Error>;

    fn warn_msg(&self) -> Option<&'static str> {
        None
    }

    async fn execute(self) -> Result<NextState<Self::Error>, Self::Error> {
        let footprint = RunFootprint::new(self.run_id, RunState::Running);
        let execution_id = Uuid::new_v4();

        let conn = self.ctx.meta_storage.scheduler_conn().await?;

        conn.as_client().upsert_run(&footprint).await?;
        conn.as_client()
            .put_execution(footprint.metadata.run_id, execution_id)
            .await?;

        Ok(NextState::next(self.into_running(execution_id)))
    }
}