5973865399

Committed 25 Aug 2023 09:11AM UTC coverage: 75.4%. First build

Build # 5973865399

Build Type

Pull #1918

github

Committed by

Jesse-Bakker

Commit Message

Make file watcher non-blocking

Pull Request Pull Request #1918: Make file watcher non-blocking

Run Details

20 of 20 new or added lines in 1 file covered. (100.0%)

46983 of 62312 relevant lines covered (75.4%)

47954.47 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/dozer-cli/src/live/progress.rs

use std::{collections::HashMap, ops::Deref, sync::atomic::Ordering, time::Duration};

use dozer_types::grpc_types::live::{ConnectResponse, Metric, ProgressResponse};
use prometheus_parse::Value;
use tokio::time::interval;

use crate::shutdown::ShutdownReceiver;

use super::LiveError;

const PROGRESS_POLL_FREQUENCY: u64 = 100;
const METRICS_ENDPOINT: &str = "http://localhost:9000/metrics";
pub async fn progress_stream(
    tx: tokio::sync::broadcast::Sender<ConnectResponse>,
    shutdown_receiver: ShutdownReceiver,
) -> Result<(), LiveError> {
    let mut retry_interval = interval(Duration::from_millis(PROGRESS_POLL_FREQUENCY));

    let mut progress: HashMap<String, Metric> = HashMap::new();

    loop {
        if !shutdown_receiver.get_running_flag().load(Ordering::Relaxed) {
            return Ok(());
        }
        let text = reqwest::get(METRICS_ENDPOINT)
            .await?
            .error_for_status()?
            .text()
            .await?;
        let lines = text.lines().map(|line| Ok(line.to_string()));

        if let Ok(metrics) = prometheus_parse::Scrape::parse(lines) {
            for sample in metrics.samples {
                if let Value::Counter(count) = sample.value {
                    progress.insert(
                        sample.metric,
                        Metric {
                            value: count as u32,
                            labels: sample.labels.deref().clone(),
                            ts: sample.timestamp.timestamp_millis() as u32,
                        },
                    );
                }
            }

            if tx
                .send(ConnectResponse {
                    live: None,
                    progress: Some(ProgressResponse {
                        progress: progress.clone(),
                    }),
                    build: None,
                })
                .is_err()
            {
                // If the receiver is dropped, we're done here.
                return Ok(());
            }
        }

        retry_interval.tick().await;
    }
}

1	use std::{collections::HashMap, ops::Deref, sync::atomic::Ordering, time::Duration};
2
3	use dozer_types::grpc_types::live::{ConnectResponse, Metric, ProgressResponse};
4	use prometheus_parse::Value;
5	use tokio::time::interval;
6
7	use crate::shutdown::ShutdownReceiver;
8
9	use super::LiveError;
10
11	const PROGRESS_POLL_FREQUENCY: u64 = 100;
12	const METRICS_ENDPOINT: &str = "http://localhost:9000/metrics";
13	pub async fn progress_stream(	×
14	tx: tokio::sync::broadcast::Sender<ConnectResponse>,	×
15	shutdown_receiver: ShutdownReceiver,	×
16	) -> Result<(), LiveError> {	×
17	let mut retry_interval = interval(Duration::from_millis(PROGRESS_POLL_FREQUENCY));	×
18		×
19	let mut progress: HashMap<String, Metric> = HashMap::new();	×
20
21	loop {
22	if !shutdown_receiver.get_running_flag().load(Ordering::Relaxed) {	×
23	return Ok(());	×
24	}	×
25	let text = reqwest::get(METRICS_ENDPOINT)	×
26	.await?	×
27	.error_for_status()?	×
28	.text()	×
29	.await?;	×
30	let lines = text.lines().map(\|line\| Ok(line.to_string()));	×
31
32	if let Ok(metrics) = prometheus_parse::Scrape::parse(lines) {	×
33	for sample in metrics.samples {	×
34	if let Value::Counter(count) = sample.value {	×
35	progress.insert(	×
36	sample.metric,	×
37	Metric {	×
38	value: count as u32,	×
39	labels: sample.labels.deref().clone(),	×
40	ts: sample.timestamp.timestamp_millis() as u32,	×
41	},	×
42	);	×
43	}	×
44	}
45
46	if tx	×
47	.send(ConnectResponse {	×
48	live: None,	×
49	progress: Some(ProgressResponse {	×
50	progress: progress.clone(),	×
51	}),	×
52	build: None,	×
53	})	×
54	.is_err()	×
55	{
56	// If the receiver is dropped, we're done here.
57	return Ok(());	×
58	}	×
59	}	×
60
61	retry_interval.tick().await;	×
62	}
63	}	×

getdozer / dozer / 5973865399

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous