opsqueue 0.30.3

lightweight batch processing queue for heavy loads
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179

//! Define common metrics exposed via a Prometheus endpoint
//!
//! This allows inspection of how well the queue is performing under production load.
//!
//! Note that we explicitly have a separate endpoint to check the queue health,
//! which is more fine-grained than Prometheus' way to check whether a service is 'up'.
use axum_prometheus::{
    metrics::{describe_counter, describe_gauge, describe_histogram, gauge, Unit},
    metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle},
    utils::SECONDS_DURATION_BUCKETS,
    GenericMetricLayer, PrometheusMetricLayer, AXUM_HTTP_REQUESTS_DURATION_SECONDS,
};
use tokio_util::sync::CancellationToken;

use crate::db::DBPools;

pub const SUBMISSIONS_TOTAL_COUNTER: &str = "submissions_total_count";
pub const SUBMISSIONS_COMPLETED_COUNTER: &str = "submissions_completed_count";
pub const SUBMISSIONS_FAILED_COUNTER: &str = "submissions_failed_count";
pub const SUBMISSIONS_DURATION_COMPLETE_HISTOGRAM: &str = "submissions_complete_duration_seconds";
pub const SUBMISSIONS_DURATION_FAIL_HISTOGRAM: &str = "submissions_fail_duration_seconds";

pub const CHUNKS_COMPLETED_COUNTER: &str = "chunks_completed_count";
pub const CHUNKS_FAILED_COUNTER: &str = "chunks_failed_count";
pub const CHUNKS_RETRIED_COUNTER: &str = "chunks_retried_count";
pub const CHUNKS_SKIPPED_COUNTER: &str = "chunks_skipped_count";
pub const CHUNKS_TOTAL_COUNTER: &str = "chunks_total_count";
pub const CHUNKS_BACKLOG_GAUGE: &str = "chunks_in_backlog_count";

pub const CHUNKS_DURATION_COMPLETED_HISTOGRAM: &str = "chunks_duration_completed_seconds";
pub const CHUNKS_DURATION_FAILED_HISTOGRAM: &str = "chunks_duration_failure_seconds";

pub const RESERVER_RESERVATIONS_SUCCEEDED_COUNTER: &str = "reserver_reservations_succeeded_count";
pub const RESERVER_RESERVATIONS_FAILED_COUNTER: &str = "reserver_reservations_failed_count";
pub const RESERVER_CHUNKS_RESERVED_GAUGE: &str = "reserver_chunks_reserved_count";

pub const CONSUMERS_CONNECTED_GAUGE: &str = "consumers_connected_count";
pub const CONSUMER_FETCH_AND_RESERVE_CHUNKS_HISTOGRAM: &str =
    "consumer_fetch_and_reserve_chunks_duration_seconds";
pub const CONSUMER_COMPLETE_CHUNK_DURATION: &str = "consumer_complete_chunk_duration_seconds";
pub const CONSUMER_FAIL_CHUNK_DURATION: &str = "consumer_fail_chunk_duration_seconds";

pub const OPERATIONS_BACKLOG_GAUGE: &str = "operations_in_backlog_count";

pub fn describe_metrics() {
    describe_counter!(SUBMISSIONS_TOTAL_COUNTER, Unit::Count, "Total count of submissions (in backlog + completed + failed), i.e. total that ever entered the system");
    describe_counter!(
        SUBMISSIONS_COMPLETED_COUNTER,
        Unit::Count,
        "Number of submissions completed successfully"
    );
    describe_counter!(
        SUBMISSIONS_FAILED_COUNTER,
        Unit::Count,
        "Number of submissions failed permanently"
    );
    describe_histogram!(SUBMISSIONS_DURATION_COMPLETE_HISTOGRAM, Unit::Seconds, "Time between a submission entering the system and its final chunk being completed. Does not count failed submissions.");

    describe_counter!(
        CHUNKS_COMPLETED_COUNTER,
        Unit::Count,
        "Number of chunks completed"
    );
    describe_counter!(
        CHUNKS_FAILED_COUNTER,
        Unit::Count,
        "Number of chunks failed permanently (retries exhausted). Does not include skipped chunks"
    );
    describe_counter!(
        CHUNKS_RETRIED_COUNTER,
        Unit::Count,
        "Number of chunks that failed temporarily and will be retried"
    );
    describe_counter!(
        CHUNKS_SKIPPED_COUNTER,
        Unit::Count,
        "Number of chunks skipped (because another chunk in the submission failed)"
    );
    describe_counter!(CHUNKS_TOTAL_COUNTER, Unit::Count, "Total count of chunks (in backlog + completed + failed), i.e. total that ever entered the system");
    // We could calculate the backlog size from TOTAL - COMPLETED - FAILED - SKIPPED
    // but since it will commonly be used for checking whether we should autoscale,
    // it's much nicer to measure/expose it directly
    describe_gauge!(CHUNKS_BACKLOG_GAUGE, Unit::Count, "Number of chunks in the backlog. Note that this is a _gauge_ reflecting the accurate state of the DB");
    describe_histogram!(CHUNKS_DURATION_COMPLETED_HISTOGRAM, Unit::Seconds, "How long it took from the moment a chunk was reserved until 'complete_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data)");
    describe_histogram!(CHUNKS_DURATION_FAILED_HISTOGRAM, Unit::Seconds, "How long it took from the moment a chunk was reserved until 'fail_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data). Includes chunks that are retried.");

    describe_gauge!(RESERVER_CHUNKS_RESERVED_GAUGE, Unit::Count, "Number of chunks currently reserved by the reserver, i.e. being worked on by the consumers");

    describe_gauge!(
        CONSUMERS_CONNECTED_GAUGE,
        Unit::Count,
        "Number of healthy websocket connections between the system and consumers"
    );
    describe_histogram!(CONSUMER_FETCH_AND_RESERVE_CHUNKS_HISTOGRAM, Unit::Seconds, "Time spent by Opsqueue (SQLite + reserver) to reserve `limit` chunks for a consumer using strategy `strategy`");
    describe_histogram!(
        CONSUMER_COMPLETE_CHUNK_DURATION,
        Unit::Seconds,
        "Time spent by Opsqueue to mark a given chunk as completed"
    );
    describe_histogram!(
        CONSUMER_FAIL_CHUNK_DURATION,
        Unit::Seconds,
        "Time spent by Opsqueue to mark a given chunk as failed"
    );

    describe_gauge!(
        OPERATIONS_BACKLOG_GAUGE,
        Unit::Count,
        "Slight overestimation of the number of operations in the backlog (chunk_size * chunk_count for each submission). For simplicity, we always consider the last chunk full. This is a gauge reflecting the rough current DB state."
    );
}

pub type PrometheusConfig = (
    GenericMetricLayer<'static, PrometheusHandle, axum_prometheus::Handle>,
    PrometheusHandle,
);

#[must_use]
pub fn setup_prometheus() -> (
    GenericMetricLayer<'static, PrometheusHandle, axum_prometheus::Handle>,
    PrometheusHandle,
) {
    // PrometheusMetricLayer::pair()
    let metric_layer = PrometheusMetricLayer::new();
    // This is the default if you use `PrometheusMetricLayer::pair`.
    let metric_handle = PrometheusBuilder::new()
        .set_buckets_for_metric(
            Matcher::Full(AXUM_HTTP_REQUESTS_DURATION_SECONDS.to_string()),
            SECONDS_DURATION_BUCKETS,
        )
        .expect("Building Prometheus failed")
        .install_recorder()
        .expect("Installing global Prometheus recorder failed");

    describe_metrics();

    (metric_layer, metric_handle)
}

/// Returns the number of seconds contained by this TimeDelta as f64, with nanosecond precision.
///
/// Adapted from <https://doc.rust-lang.org/std/time/struct.Duration.html#method.as_secs_f64>
pub fn time_delta_as_f64(td: chrono::TimeDelta) -> f64 {
    const NANOS_PER_SEC: f64 = 1_000_000_000.0;
    (td.num_seconds() as f64) + (td.subsec_nanos() as f64) / NANOS_PER_SEC
}

/// Calculates the backlog-size metrics used for autoscaling.
pub async fn calculate_scaling_metrics(db_pool: &DBPools) -> anyhow::Result<()> {
    let mut conn = db_pool.reader_conn().await?;
    let chunks_backlog_count: u64 = crate::common::chunk::db::count_chunks(&mut conn)
        .await?
        .into();
    gauge!(CHUNKS_BACKLOG_GAUGE).set(chunks_backlog_count as f64);
    let ops_backlog_count: f64 =
        crate::common::chunk::db::count_ops_in_backlog_estimate(&mut conn).await?;
    gauge!(OPERATIONS_BACKLOG_GAUGE).set(ops_backlog_count);

    Ok(())
}

/// Calculate the backlog-size metrics used for autoscaling periodically;
/// currently every 5 seconds.
pub async fn periodically_calculate_scaling_metrics(
    db_pool: &DBPools,
    cancellation_token: &CancellationToken,
) {
    const METRICS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
    loop {
        tokio::select! {
            _ = cancellation_token.cancelled() => break,
            _ = tokio::time::sleep(METRICS_INTERVAL) => {
                if let Err(e) = calculate_scaling_metrics(db_pool).await {
                    tracing::error!("Error calculating scaling metrics: {}", e);
                }
            }
        }
    }
}