21907973557

Committed 11 Feb 2026 01:58PM UTC coverage: 78.095%. First build

Build # 21907973557

Build Type

Pull #1424

github

Committed by

web-flow

Commit Message

Merge 536cab794 into 91a9cd6fa

Pull Request Pull Request #1424: feat: hmac client token acquisition

Run Details

180 of 233 new or added lines in 9 files covered. (77.25%)

10143 of 12988 relevant lines covered (78.1%)

5653.0 hits per line

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

25.81

/crates/oss/unleash-edge-http-client/src/instance_data.rs

use reqwest::{StatusCode, Url};
use std::sync::Arc;
use tokio::sync::RwLock;

use crate::{ClientMetaInformation, UnleashClient};
use tracing::{debug, warn};
use unleash_edge_cli::{CliArgs, EdgeMode};
use unleash_edge_types::BackgroundTask;
use unleash_edge_types::errors::EdgeError;
use unleash_edge_types::metrics::instance_data::EdgeInstanceData;

#[derive(Debug, Clone)]
pub struct InstanceDataSender {
    pub unleash_client: Arc<UnleashClient>,
    pub token: String,
    pub base_path: String,
}

#[derive(Debug, Clone)]
pub enum InstanceDataSending {
    SendNothing,
    SendInstanceData(InstanceDataSender),
}

impl InstanceDataSending {
    pub fn from_args(
        args: CliArgs,
        client_meta_information: &ClientMetaInformation,
        http_client: reqwest::Client,
    ) -> Result<Self, EdgeError> {
        match args.mode {
            EdgeMode::Edge(edge_args) => edge_args
                .tokens
                .first()
                .map(|token| {
                    let unleash_client = Url::parse(&edge_args.upstream_url.clone())
                        .map(|url| {
                            UnleashClient::from_url_with_backing_client(
                                url,
                                args.auth_headers
                                    .upstream_auth_header
                                    .clone()
                                    .unwrap_or("Authorization".to_string()),
                                http_client,
                                client_meta_information.clone(),
                            )
                        })
                        .map(|c| {
                            c.with_custom_client_headers(edge_args.custom_client_headers.clone())
                        })
                        .map(Arc::new)
                        .map_err(|_| EdgeError::InvalidServerUrl(edge_args.upstream_url.clone()))
                        .expect("Could not construct UnleashClient");
                    let instance_data_sender = InstanceDataSender {
                        unleash_client,
                        token: token.token.clone(),
                        base_path: args.http.base_path.clone(),
                    };
                    InstanceDataSending::SendInstanceData(instance_data_sender)
                })
                .map(Ok)
                .unwrap_or(Ok(InstanceDataSending::SendNothing)),
            _ => Ok(InstanceDataSending::SendNothing),
        }
    }
}

#[derive(Debug)]
enum InstanceDataSendError {
    Backoff(String),
    Unexpected(String),
}

async fn send_instance_data(
    instance_data_sender: &Arc<InstanceDataSending>,
    our_instance_data: &Arc<EdgeInstanceData>,
    downstream_instance_data: &Arc<RwLock<Vec<EdgeInstanceData>>>,
) -> Result<(), InstanceDataSendError> {
    match instance_data_sender.as_ref() {
        InstanceDataSending::SendNothing => {
            debug!("No instance data sender found. Doing nothing.");
            Ok(())
        }
        InstanceDataSending::SendInstanceData(instance_data_sender) => {
            let observed_data = our_instance_data.observe(
                downstream_instance_data.read().await.clone(),
                &instance_data_sender.base_path,
            );
            let status = instance_data_sender
                .unleash_client
                .post_edge_observability_data(observed_data, &instance_data_sender.token)
                .await;

            if let Err(e) = status {
                match e {
                    EdgeError::EdgeMetricsRequestError(status, _) => {
                        match status {
                            StatusCode::NOT_FOUND => {
                                downstream_instance_data.write().await.clear();
                                our_instance_data.clear_time_windowed_metrics();
                                Err(InstanceDataSendError::Backoff("Our upstream is not running a version that supports edge metrics.".into()))
                            }
                            StatusCode::FORBIDDEN => {
                                downstream_instance_data.write().await.clear();
                                our_instance_data.clear_time_windowed_metrics();
                                Err(InstanceDataSendError::Backoff("Upstream edge metrics said our token wasn't allowed to post data".into()))
                            }
                            _ => Err(InstanceDataSendError::Unexpected(format!(
                                "Failed to post instance data due to unknown error {e:?}"
                            ))),
                        }
                    }
                    _ => Err(InstanceDataSendError::Unexpected(format!(
                        "Failed to post instance data due to unknown error {e:?}"
                    ))),
                }
            } else {
                downstream_instance_data.write().await.clear();
                our_instance_data.clear_time_windowed_metrics();
                Ok(())
            }
        }
    }
}

pub fn create_once_off_send_instance_data(
    instance_data_sender: Arc<InstanceDataSending>,
    our_instance_data: Arc<EdgeInstanceData>,
    downstream_instance_data: Arc<RwLock<Vec<EdgeInstanceData>>>,
) -> BackgroundTask {
    let instance_data_sender = instance_data_sender.clone();
    let our_instance_data = our_instance_data.clone();
    let downstream_instance_data = downstream_instance_data.clone();

    Box::pin(async move {
        let result = send_instance_data(
            &instance_data_sender,
            &our_instance_data,
            &downstream_instance_data,
        )
        .await;

        if let Err(err) = result {
            warn!("Failed to send last set of instance data during graceful exit: {err:?}");
        }
    })
}

pub fn create_send_instance_data_task(
    instance_data_sender: Arc<InstanceDataSending>,
    our_instance_data: Arc<EdgeInstanceData>,
    downstream_instance_data: Arc<RwLock<Vec<EdgeInstanceData>>>,
) -> BackgroundTask {
    let mut errors = 0;
    let delay = std::time::Duration::from_secs(60);
    Box::pin(async move {
        loop {
            tokio::time::sleep(
                std::time::Duration::from_secs(60) + delay * std::cmp::min(errors, 10),
            )
            .await;

            let result = send_instance_data(
                &instance_data_sender,
                &our_instance_data,
                &downstream_instance_data,
            )
            .await;
            match result {
                Ok(_) => {
                    debug!("Successfully posted observability metrics.");
                    errors = 0;
                    downstream_instance_data.write().await.clear();
                    our_instance_data.clear_time_windowed_metrics();
                }
                Err(err) => match err {
                    InstanceDataSendError::Backoff(message) => {
                        warn!(message);
                        errors += 1;
                    }
                    InstanceDataSendError::Unexpected(message) => {
                        warn!(message);
                    }
                },
            }
        }
    })
}

1	use reqwest::{StatusCode, Url};
2	use std::sync::Arc;
3	use tokio::sync::RwLock;
4
5	use crate::{ClientMetaInformation, UnleashClient};
6	use tracing::{debug, warn};
7	use unleash_edge_cli::{CliArgs, EdgeMode};
8	use unleash_edge_types::BackgroundTask;
9	use unleash_edge_types::errors::EdgeError;
10	use unleash_edge_types::metrics::instance_data::EdgeInstanceData;
11
12	#[derive(Debug, Clone)]
13	pub struct InstanceDataSender {
14	pub unleash_client: Arc<UnleashClient>,
15	pub token: String,
16	pub base_path: String,
17	}
18
19	#[derive(Debug, Clone)]
20	pub enum InstanceDataSending {
21	SendNothing,
22	SendInstanceData(InstanceDataSender),
23	}
24
25	impl InstanceDataSending {
26	pub fn from_args(	1✔
27	args: CliArgs,	1✔
28	client_meta_information: &ClientMetaInformation,	1✔
29	http_client: reqwest::Client,	1✔
30	) -> Result<Self, EdgeError> {	1✔
31	match args.mode {	1✔
32	EdgeMode::Edge(edge_args) => edge_args	1✔
33	.tokens	1✔
34	.first()	1✔
35	.map(\|token\| {	1✔
36	let unleash_client = Url::parse(&edge_args.upstream_url.clone())	×
37	.map(\|url\| {	×
38	UnleashClient::from_url_with_backing_client(	×
39	url,	×
40	args.auth_headers	×
41	.upstream_auth_header	×
42	.clone()	×
43	.unwrap_or("Authorization".to_string()),	×
44	http_client,	×
45	client_meta_information.clone(),	×
46	)
47	})	×
48	.map(\|c\| {	×
49	c.with_custom_client_headers(edge_args.custom_client_headers.clone())	×
50	})	×
51	.map(Arc::new)	×
52	.map_err(\|_\| EdgeError::InvalidServerUrl(edge_args.upstream_url.clone()))	×
53	.expect("Could not construct UnleashClient");	×
54	let instance_data_sender = InstanceDataSender {	×
55	unleash_client,	×
NEW 56	token: token.token.clone(),	×
57	base_path: args.http.base_path.clone(),	×
58	};	×
59	InstanceDataSending::SendInstanceData(instance_data_sender)	×
60	})	×
61	.map(Ok)	1✔
62	.unwrap_or(Ok(InstanceDataSending::SendNothing)),	1✔
63	_ => Ok(InstanceDataSending::SendNothing),	×
64	}
65	}	1✔
66	}
67
68	#[derive(Debug)]
69	enum InstanceDataSendError {
70	Backoff(String),
71	Unexpected(String),
72	}
73
74	async fn send_instance_data(	×
75	instance_data_sender: &Arc<InstanceDataSending>,	×
76	our_instance_data: &Arc<EdgeInstanceData>,	×
77	downstream_instance_data: &Arc<RwLock<Vec<EdgeInstanceData>>>,	×
78	) -> Result<(), InstanceDataSendError> {	×
79	match instance_data_sender.as_ref() {	×
80	InstanceDataSending::SendNothing => {
81	debug!("No instance data sender found. Doing nothing.");	×
82	Ok(())	×
83	}
84	InstanceDataSending::SendInstanceData(instance_data_sender) => {	×
85	let observed_data = our_instance_data.observe(	×
86	downstream_instance_data.read().await.clone(),	×
87	&instance_data_sender.base_path,	×
88	);
89	let status = instance_data_sender	×
90	.unleash_client	×
91	.post_edge_observability_data(observed_data, &instance_data_sender.token)	×
92	.await;	×
93
94	if let Err(e) = status {	×
95	match e {	×
96	EdgeError::EdgeMetricsRequestError(status, _) => {	×
97	match status {	×
98	StatusCode::NOT_FOUND => {
99	downstream_instance_data.write().await.clear();	×
100	our_instance_data.clear_time_windowed_metrics();	×
101	Err(InstanceDataSendError::Backoff("Our upstream is not running a version that supports edge metrics.".into()))	×
102	}
103	StatusCode::FORBIDDEN => {
104	downstream_instance_data.write().await.clear();	×
105	our_instance_data.clear_time_windowed_metrics();	×
106	Err(InstanceDataSendError::Backoff("Upstream edge metrics said our token wasn't allowed to post data".into()))	×
107	}
108	_ => Err(InstanceDataSendError::Unexpected(format!(	×
109	"Failed to post instance data due to unknown error {e:?}"	×
110	))),	×
111	}
112	}
113	_ => Err(InstanceDataSendError::Unexpected(format!(	×
114	"Failed to post instance data due to unknown error {e:?}"	×
115	))),	×
116	}
117	} else {
118	downstream_instance_data.write().await.clear();	×
119	our_instance_data.clear_time_windowed_metrics();	×
120	Ok(())	×
121	}
122	}
123	}
124	}	×
125
126	pub fn create_once_off_send_instance_data(	1✔
127	instance_data_sender: Arc<InstanceDataSending>,	1✔
128	our_instance_data: Arc<EdgeInstanceData>,	1✔
129	downstream_instance_data: Arc<RwLock<Vec<EdgeInstanceData>>>,	1✔
130	) -> BackgroundTask {	1✔
131	let instance_data_sender = instance_data_sender.clone();	1✔
132	let our_instance_data = our_instance_data.clone();	1✔
133	let downstream_instance_data = downstream_instance_data.clone();	1✔
134
135	Box::pin(async move {	1✔
136	let result = send_instance_data(	×
137	&instance_data_sender,	×
138	&our_instance_data,	×
139	&downstream_instance_data,	×
140	)	×
141	.await;	×
142
143	if let Err(err) = result {	×
144	warn!("Failed to send last set of instance data during graceful exit: {err:?}");	×
145	}	×
146	})	×
147	}	1✔
148
149	pub fn create_send_instance_data_task(	1✔
150	instance_data_sender: Arc<InstanceDataSending>,	1✔
151	our_instance_data: Arc<EdgeInstanceData>,	1✔
152	downstream_instance_data: Arc<RwLock<Vec<EdgeInstanceData>>>,	1✔
153	) -> BackgroundTask {	1✔
154	let mut errors = 0;	1✔
155	let delay = std::time::Duration::from_secs(60);	1✔
156	Box::pin(async move {	1✔
157	loop {
158	tokio::time::sleep(	×
159	std::time::Duration::from_secs(60) + delay * std::cmp::min(errors, 10),	×
160	)	×
161	.await;	×
162
163	let result = send_instance_data(	×
164	&instance_data_sender,	×
165	&our_instance_data,	×
166	&downstream_instance_data,	×
167	)	×
168	.await;	×
169	match result {	×
170	Ok(_) => {
171	debug!("Successfully posted observability metrics.");	×
172	errors = 0;	×
173	downstream_instance_data.write().await.clear();	×
174	our_instance_data.clear_time_windowed_metrics();	×
175	}
176	Err(err) => match err {	×
177	InstanceDataSendError::Backoff(message) => {	×
178	warn!(message);	×
179	errors += 1;	×
180	}
181	InstanceDataSendError::Unexpected(message) => {	×
182	warn!(message);	×
183	}
184	},
185	}
186	}
187	})
188	}	1✔

Unleash / unleash-edge / 21907973557

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