varpulis-cluster 0.10.0

Distributed execution cluster for Varpulis streaming analytics
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

//! Pipeline migration types and status tracking.

use std::time::Instant;

use serde::{Deserialize, Serialize};
use varpulis_runtime::persistence::EngineCheckpoint;

use crate::worker::WorkerId;

/// Status of a pipeline migration.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MigrationStatus {
    /// Extracting state from source worker
    Checkpointing,
    /// Deploying pipeline to target worker
    Deploying,
    /// Restoring state on target worker
    Restoring,
    /// Updating routing to point to target
    Switching,
    /// Removing pipeline from source worker
    CleaningUp,
    /// Migration completed successfully
    Completed,
    /// Migration failed
    Failed(String),
}

impl std::fmt::Display for MigrationStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Checkpointing => write!(f, "checkpointing"),
            Self::Deploying => write!(f, "deploying"),
            Self::Restoring => write!(f, "restoring"),
            Self::Switching => write!(f, "switching"),
            Self::CleaningUp => write!(f, "cleaning_up"),
            Self::Completed => write!(f, "completed"),
            Self::Failed(msg) => write!(f, "failed: {}", msg),
        }
    }
}

/// Reason for initiating a migration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MigrationReason {
    /// Source worker died
    Failover,
    /// Load rebalancing
    Rebalance,
    /// Source worker is draining
    Drain,
    /// Operator-initiated
    Manual,
    /// Connector lost connectivity
    ConnectorFailure,
}

impl std::fmt::Display for MigrationReason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Failover => write!(f, "failover"),
            Self::Rebalance => write!(f, "rebalance"),
            Self::Drain => write!(f, "drain"),
            Self::Manual => write!(f, "manual"),
            Self::ConnectorFailure => write!(f, "connector_failure"),
        }
    }
}

/// Tracks an in-progress pipeline migration.
#[derive(Debug, Clone)]
pub struct MigrationTask {
    pub id: String,
    pub pipeline_name: String,
    pub group_id: String,
    pub source_worker: WorkerId,
    pub target_worker: WorkerId,
    pub status: MigrationStatus,
    pub started_at: Instant,
    pub checkpoint: Option<EngineCheckpoint>,
    pub reason: MigrationReason,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_migration_status_display() {
        assert_eq!(MigrationStatus::Checkpointing.to_string(), "checkpointing");
        assert_eq!(MigrationStatus::Deploying.to_string(), "deploying");
        assert_eq!(MigrationStatus::Restoring.to_string(), "restoring");
        assert_eq!(MigrationStatus::Switching.to_string(), "switching");
        assert_eq!(MigrationStatus::CleaningUp.to_string(), "cleaning_up");
        assert_eq!(MigrationStatus::Completed.to_string(), "completed");
        assert_eq!(
            MigrationStatus::Failed("timeout".into()).to_string(),
            "failed: timeout"
        );
    }

    #[test]
    fn test_migration_reason_display() {
        assert_eq!(MigrationReason::Failover.to_string(), "failover");
        assert_eq!(MigrationReason::Rebalance.to_string(), "rebalance");
        assert_eq!(MigrationReason::Drain.to_string(), "drain");
        assert_eq!(MigrationReason::Manual.to_string(), "manual");
        assert_eq!(
            MigrationReason::ConnectorFailure.to_string(),
            "connector_failure"
        );
    }

    #[test]
    fn test_migration_status_serde() {
        for status in [
            MigrationStatus::Checkpointing,
            MigrationStatus::Deploying,
            MigrationStatus::Restoring,
            MigrationStatus::Switching,
            MigrationStatus::CleaningUp,
            MigrationStatus::Completed,
            MigrationStatus::Failed("err".into()),
        ] {
            let json = serde_json::to_string(&status).unwrap();
            let parsed: MigrationStatus = serde_json::from_str(&json).unwrap();
            assert_eq!(parsed, status);
        }
    }

    #[test]
    fn test_migration_reason_serde() {
        for reason in [
            MigrationReason::Failover,
            MigrationReason::Rebalance,
            MigrationReason::Drain,
            MigrationReason::Manual,
            MigrationReason::ConnectorFailure,
        ] {
            let json = serde_json::to_string(&reason).unwrap();
            let _parsed: MigrationReason = serde_json::from_str(&json).unwrap();
        }
    }
}