superstac-engine 0.1.0

Runtime for superstac: orchestrates federated STAC search, health monitoring, and capability introspection.
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

use parking_lot::Mutex;
use reqwest::Client;
use std::{collections::HashMap, sync::Arc};
use tokio::{task::JoinHandle, time::interval};

use crate::types::SharedStorage;
use superstac_core::{errors::SuperSTACError, models::catalog::Catalog};

pub struct HealthManager {
    tasks: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
    client: Client,
}

impl HealthManager {
    pub fn new(client: Client) -> Self {
        Self {
            tasks: Arc::new(Mutex::new(HashMap::new())),
            client,
        }
    }

    pub async fn start(
        &self,
        storage: SharedStorage,
        catalogs: Vec<Catalog>,
    ) -> Result<(), SuperSTACError> {
        tracing::debug!(catalogs = catalogs.len(), "starting health monitor");

        for catalog in catalogs {
            let healthy = self.check_once(Arc::clone(&storage), &catalog).await;

            if healthy {
                // TODO, check settings to decide whether to spawn monitor or not
                self.spawn_monitor(Arc::clone(&storage), catalog);
            }
        }

        Ok(())
    }

    pub async fn stop_all(&self) {
        let mut tasks = self.tasks.lock();

        for (_, handle) in tasks.drain() {
            handle.abort();
        }
    }

    pub async fn stop(&self, id: &str) {
        if let Some(handle) = self.tasks.lock().remove(id) {
            handle.abort();
        }
    }

    pub async fn check_once(&self, storage: SharedStorage, catalog: &Catalog) -> bool {
        let endpoint = &catalog.health_status.endpoint;

        let range = catalog.settings.healthy_status_code_range;

        let healthy = match self.client.get(endpoint).send().await {
            Ok(resp) => {
                let status = resp.status().as_u16();

                range.0 <= status && status <= range.1
            }

            Err(_) => false,
        };

        tracing::debug!(catalog = %catalog.id, healthy, "health check");

        {
            let mut storage = storage.lock();

            if let Err(e) = storage.update_health(&catalog.id, healthy) {
                tracing::warn!(catalog = %catalog.id, error = %e, "failed to persist health update");
            }
        }

        healthy
    }

    pub fn spawn_monitor(&self, storage: SharedStorage, catalog: Catalog) {
        let id = catalog.id.clone();

        if self.tasks.lock().contains_key(&id) {
            return;
        }

        let endpoint = catalog.health_status.endpoint.clone();

        let range = catalog.settings.healthy_status_code_range;

        let duration = catalog.settings.health_check_strategy.as_duration();

        let client = self.client.clone();

        let task_id = id.clone();

        let handle = tokio::spawn(async move {
            let mut ticker = interval(duration);

            loop {
                ticker.tick().await;

                let healthy = match client.get(&endpoint).send().await {
                    Ok(resp) => {
                        let status = resp.status().as_u16();

                        range.0 <= status && status <= range.1
                    }

                    Err(_) => false,
                };

                tracing::debug!(catalog = %task_id, healthy, "health tick");

                {
                    let mut storage = storage.lock();

                    if let Err(e) = storage.update_health(&task_id, healthy) {
                        tracing::warn!(
                            catalog = %task_id,
                            error = %e,
                            "failed to persist health update"
                        );
                    }
                }
            }
        });

        tracing::debug!(
            catalog = %id,
            interval_secs = duration.as_secs(),
            "spawned health monitor"
        );

        self.tasks.lock().insert(id, handle);
    }
}