IntenseWebs/LibreQoS
3
0 Fork 0
mirror of https://github.com/LibreQoE/LibreQoS.git synced 2026-10-24 22:07:40 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix: Bound node manager wait paths

Browse Source
This commit is contained in:
Herbert Wolverson committed 2026-04-30 15:18:49 -05:00
1 parent 1c8bba363d
commit 7e8fe7bd7d
19 files changed
+673 -363
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/rust/lqos_bus/src/bus/client.rs
+43
View File
@@ -2,9 +2,11 @@
// SPDX-License-Identifier: AGPL-3.0-or-later WITH LicenseRef-LibreQoS-Exception
use crate::{BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse, BusSession, bus::BusClientError};
use std::time::Duration;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::UnixStream,
time::timeout,
};
use tracing::error;
@@ -50,6 +52,16 @@ impl LibreqosBusClient {
})
}
/// Creates a new `LibreqosBusClient`, failing if connection setup exceeds
/// `request_timeout`.
///
/// Side effects: opens a Unix socket connection to the local `lqosd` bus.
pub async fn new_with_timeout(request_timeout: Duration) -> Result<Self, BusClientError> {
timeout(request_timeout, Self::new())
.await
.map_err(|_| BusClientError::TimedOut)?
}
/// Sends a request to the bus and waits for a response.
///
/// ## Arguments
@@ -91,6 +103,21 @@ impl LibreqosBusClient {
}
Ok(response.responses)
}
/// Sends a request to the bus and waits for a bounded response.
///
/// Side effects: writes the request to the connected Unix socket. If this
/// call times out, discard the client because the stream may later receive
/// the timed-out reply.
pub async fn request_with_timeout(
&mut self,
requests: Vec<BusRequest>,
request_timeout: Duration,
) -> Result<Vec<BusResponse>, BusClientError> {
timeout(request_timeout, self.request(requests))
.await
.map_err(|_| BusClientError::TimedOut)?
}
}
/// Convenient wrapper for accessing the bus, for a single request-response cycle. This
@@ -108,3 +135,19 @@ pub async fn bus_request(requests: Vec<BusRequest>) -> Result<Vec<BusResponse>,
};
client.request(requests).await
}
/// Convenient wrapper for a single bus request with an end-to-end timeout.
///
/// Side effects: opens a Unix socket connection, sends the request, and closes
/// the connection when the client is dropped.
pub async fn bus_request_with_timeout(
requests: Vec<BusRequest>,
request_timeout: Duration,
) -> Result<Vec<BusResponse>, BusClientError> {
timeout(request_timeout, async move {
let mut client = LibreqosBusClient::new().await?;
client.request(requests).await
})
.await
.map_err(|_| BusClientError::TimedOut)?
}
src/rust/lqos_bus/src/bus/mod.rs
+4 -1
View File
@@ -9,7 +9,7 @@ mod request;
pub mod response;
mod session;
mod unix_socket_server;
pub use client::{LibreqosBusClient, bus_request};
pub use client::{LibreqosBusClient, bus_request, bus_request_with_timeout};
pub use queue_data::*;
pub use reply::BusReply;
pub use request::{
@@ -55,6 +55,9 @@ pub enum BusClientError {
/// Reading the reply from the socket failed.
#[error("Cannot read from socket")]
StreamReadError,
/// The socket operation exceeded the caller's deadline.
#[error("Timed out waiting for bus response")]
TimedOut,
/// The socket connection is no longer usable.
#[error("Stream is no longer connected")]
StreamNotConnected,
src/rust/lqos_bus/src/bus/unix_socket_server.rs
+47 -5
View File
@@ -5,21 +5,63 @@ use crate::{
BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse,
bus::client::{MAGIC_NUMBER, MAGIC_RESPONSE},
};
use std::{ffi::CString, fs::remove_file};
use std::{ffi::CString, fs::remove_file, time::Duration};
use thiserror::Error;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::UnixListener,
task::spawn_blocking,
time::timeout,
};
use tracing::{debug, error, info, warn};
use super::BUS_SOCKET_DIRECTORY;
use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame};
const BUS_HANDLER_TIMEOUT: Duration = Duration::from_secs(30);
fn dropped_reply_response_count(reply: &BusReply) -> usize {
reply.responses.len()
}
fn timeout_reply(request_count: usize) -> BusReply {
BusReply {
responses: (0..request_count)
.map(|_| BusResponse::Fail("Bus request handler timed out".to_string()))
.collect(),
}
}
async fn handle_requests_with_deadline(
handle_bus_requests: fn(&[BusRequest], &mut Vec<BusResponse>),
requests: Vec<BusRequest>,
) -> BusReply {
let request_count = requests.len();
let handler = spawn_blocking(move || {
let mut response = BusReply {
responses: Vec::with_capacity(request_count),
};
handle_bus_requests(&requests, &mut response.responses);
response
});
match timeout(BUS_HANDLER_TIMEOUT, handler).await {
Ok(Ok(response)) => response,
Ok(Err(err)) => {
warn!("Bus request handler task failed: {err}");
BusReply {
responses: (0..request_count)
.map(|_| BusResponse::Fail("Bus request handler failed".to_string()))
.collect(),
}
}
Err(_) => {
warn!("Bus request handler timed out after {BUS_HANDLER_TIMEOUT:?}");
timeout_reply(request_count)
}
}
}
/// Implements a Tokio-friendly server using Unix Sockets and the bus protocol.
/// Requests are handled and then forwarded to the handler.
pub struct UnixSocketServer {}
@@ -113,8 +155,7 @@ impl UnixSocketServer {
ret = bus_rx.recv() => {
// We received a channel-based message
if let Some((reply_channel, msg)) = ret {
let mut response = BusReply { responses: Vec::with_capacity(8) };
handle_bus_requests(&[msg], &mut response.responses);
let response = handle_requests_with_deadline(handle_bus_requests, vec![msg]).await;
if let Err(reply) = reply_channel.send(response) {
warn!(
dropped_response_count = dropped_reply_response_count(&reply),
@@ -182,8 +223,9 @@ impl UnixSocketServer {
debug!("Received request: {:?}", request);
// Handle the request and build the response
let mut response = BusReply { responses: Vec::with_capacity(8) };
handle_bus_requests(&request.requests, &mut response.responses);
let response =
handle_requests_with_deadline(handle_bus_requests, request.requests)
.await;
// Encode the response
let Ok(encoded_response) = encode_reply_cbor(&response) else {
src/rust/lqos_bus/src/lib.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub use bus::{
BUS_SOCKET_PATH, BakeryCapacityReportInterface, BlackboardSystem, BusClientError, BusReply,
BusRequest, BusResponse, BusSession, CakeDiffTinTransit, CakeDiffTransit, CakeTransit,
LibreqosBusClient, QueueStoreTransit, SchedulerProgressReport, TopFlowType, UnixSocketServer,
UrgentSeverity, UrgentSource, bus_request,
UrgentSeverity, UrgentSource, bus_request, bus_request_with_timeout,
};
pub use tc_handle::TcHandle;
src/rust/lqos_network_devices/src/actor.rs
+76 -48
View File
@@ -5,18 +5,18 @@ use crate::dynamic_store::{load_dynamic_circuits_from_disk, persist_dynamic_circ
use crate::state;
use crate::{DaemonHooks, ShapedDevicesCatalog, load_network_json, load_shaped_devices};
use anyhow::{Context, Result, anyhow};
use crossbeam_channel::{Receiver, Sender};
use crossbeam_channel::{Receiver, RecvTimeoutError, SendTimeoutError, Sender, bounded};
use ip_network_table::IpNetworkTable;
use once_cell::sync::OnceCell;
use std::collections::HashSet;
use std::net::IpAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::oneshot;
use tracing::{debug, error, warn};
const RELOAD_RETRY_DELAY_MS: u64 = 500;
const RELOAD_ATTEMPTS: usize = 2;
const ACTOR_COMMAND_TIMEOUT: Duration = Duration::from_secs(10);
const DYNAMIC_CIRCUITS_MAINTENANCE_TICK_SECONDS: u64 = 60;
const MAX_UNKNOWN_IP_PROMOTIONS_PER_OBSERVATION: usize = 32;
@@ -44,41 +44,67 @@ fn sender() -> Result<Sender<NetworkDevicesCommand>> {
.ok_or_else(|| anyhow!("lqos_network_devices runtime actor is not running"))
}
fn send_actor_command(command: NetworkDevicesCommand, operation: &str) -> Result<()> {
sender()?
.send_timeout(command, ACTOR_COMMAND_TIMEOUT)
.map_err(|err| match err {
SendTimeoutError::Timeout(_) => {
anyhow!("{operation} timed out while queueing lqos_network_devices actor command")
}
SendTimeoutError::Disconnected(_) => {
anyhow!("{operation} failed because lqos_network_devices actor stopped")
}
})
}
fn receive_actor_reply<T>(reply_rx: Receiver<Result<T>>, operation: &str) -> Result<T> {
match reply_rx.recv_timeout(ACTOR_COMMAND_TIMEOUT) {
Ok(result) => result,
Err(RecvTimeoutError::Timeout) => {
Err(anyhow!("{operation} timed out waiting for actor reply"))
}
Err(RecvTimeoutError::Disconnected) => Err(anyhow!("{operation} reply channel closed")),
}
}
pub(crate) fn request_reload_shaped_devices(reason: &str) -> Result<()> {
let (reply_tx, reply_rx) = oneshot::channel();
sender()?.send(NetworkDevicesCommand::ReloadShapedDevices {
reason: reason.to_string(),
reply: reply_tx,
})?;
reply_rx
.blocking_recv()
.map_err(|_| anyhow!("ShapedDevices reload reply channel closed"))?
let (reply_tx, reply_rx) = bounded(1);
send_actor_command(
NetworkDevicesCommand::ReloadShapedDevices {
reason: reason.to_string(),
reply: reply_tx,
},
"ShapedDevices reload",
)?;
receive_actor_reply(reply_rx, "ShapedDevices reload")
}
pub(crate) fn request_reload_network_json(reason: &str) -> Result<()> {
let (reply_tx, reply_rx) = oneshot::channel();
sender()?.send(NetworkDevicesCommand::ReloadNetworkJson {
reason: reason.to_string(),
reply: reply_tx,
})?;
reply_rx
.blocking_recv()
.map_err(|_| anyhow!("NetworkJson reload reply channel closed"))?
let (reply_tx, reply_rx) = bounded(1);
send_actor_command(
NetworkDevicesCommand::ReloadNetworkJson {
reason: reason.to_string(),
reply: reply_tx,
},
"NetworkJson reload",
)?;
receive_actor_reply(reply_rx, "NetworkJson reload")
}
pub(crate) fn apply_shaped_devices_snapshot(
reason: &str,
shaped: lqos_config::ConfigShapedDevices,
) -> Result<()> {
let (reply_tx, reply_rx) = oneshot::channel();
sender()?.send(NetworkDevicesCommand::ApplyShapedDevicesSnapshot {
reason: reason.to_string(),
shaped: Box::new(shaped),
reply: reply_tx,
})?;
reply_rx
.blocking_recv()
.map_err(|_| anyhow!("Apply shaped devices reply channel closed"))?
let (reply_tx, reply_rx) = bounded(1);
send_actor_command(
NetworkDevicesCommand::ApplyShapedDevicesSnapshot {
reason: reason.to_string(),
shaped: Box::new(shaped),
reply: reply_tx,
},
"Apply shaped devices",
)?;
receive_actor_reply(reply_rx, "Apply shaped devices")
}
pub(crate) fn report_observations(observations: &[CircuitObservation]) {
@@ -95,51 +121,53 @@ pub(crate) fn report_observations(observations: &[CircuitObservation]) {
}
pub(crate) fn upsert_dynamic_circuit(shaped_device: lqos_config::ShapedDevice) -> Result<()> {
let (reply_tx, reply_rx) = oneshot::channel();
sender()?.send(NetworkDevicesCommand::UpsertDynamicCircuit {
shaped_device: Box::new(shaped_device),
reply: reply_tx,
})?;
reply_rx
.blocking_recv()
.map_err(|_| anyhow!("Upsert dynamic circuit reply channel closed"))?
let (reply_tx, reply_rx) = bounded(1);
send_actor_command(
NetworkDevicesCommand::UpsertDynamicCircuit {
shaped_device: Box::new(shaped_device),
reply: reply_tx,
},
"Upsert dynamic circuit",
)?;
receive_actor_reply(reply_rx, "Upsert dynamic circuit")
}
pub(crate) fn remove_dynamic_circuit(circuit_id: &str) -> Result<bool> {
let (reply_tx, reply_rx) = oneshot::channel();
sender()?.send(NetworkDevicesCommand::RemoveDynamicCircuit {
circuit_id: circuit_id.to_string(),
reply: reply_tx,
})?;
reply_rx
.blocking_recv()
.map_err(|_| anyhow!("Remove dynamic circuit reply channel closed"))?
let (reply_tx, reply_rx) = bounded(1);
send_actor_command(
NetworkDevicesCommand::RemoveDynamicCircuit {
circuit_id: circuit_id.to_string(),
reply: reply_tx,
},
"Remove dynamic circuit",
)?;
receive_actor_reply(reply_rx, "Remove dynamic circuit")
}
enum NetworkDevicesCommand {
ReloadShapedDevices {
reason: String,
reply: oneshot::Sender<Result<()>>,
reply: Sender<Result<()>>,
},
ReloadNetworkJson {
reason: String,
reply: oneshot::Sender<Result<()>>,
reply: Sender<Result<()>>,
},
ApplyShapedDevicesSnapshot {
reason: String,
shaped: Box<lqos_config::ConfigShapedDevices>,
reply: oneshot::Sender<Result<()>>,
reply: Sender<Result<()>>,
},
ReportObservations {
observations: Vec<CircuitObservation>,
},
UpsertDynamicCircuit {
shaped_device: Box<lqos_config::ShapedDevice>,
reply: oneshot::Sender<Result<()>>,
reply: Sender<Result<()>>,
},
RemoveDynamicCircuit {
circuit_id: String,
reply: oneshot::Sender<Result<bool>>,
reply: Sender<Result<bool>>,
},
}
src/rust/lqos_probe/src/lib.rs
+113 -16
View File
@@ -15,8 +15,14 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use surge_ping::{Client, Config as PingConfig, ICMP, IcmpPacket, PingIdentifier, PingSequence};
use thiserror::Error;
use tokio::sync::{Semaphore, mpsc, oneshot};
use tokio::time::timeout;
use tracing::warn;
const PROBE_DNS_RESOLUTION_TIMEOUT: Duration = Duration::from_secs(2);
const PROBE_WORKER_GRACE: Duration = Duration::from_secs(1);
const PROBE_MANAGER_REPLY_GRACE: Duration = Duration::from_secs(5);
const PROBE_BATCH_CONCURRENCY_ESTIMATE: usize = 256;
/// Logical consumer class for a probe request.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize, Allocative)]
pub enum ProbeClass {
@@ -135,18 +141,23 @@ impl ProbeClient {
if requests.is_empty() {
return Ok(Vec::new());
}
let wait_timeout = probe_batch_wait_timeout(&requests);
let (reply_tx, reply_rx) = oneshot::channel();
self.tx
.send(ProbeCommand::ProbeBatch {
requests,
max_age,
reply: reply_tx,
})
.await
.map_err(|_| ProbeClientError::ManagerUnavailable)?;
reply_rx
.await
.map_err(|_| ProbeClientError::ManagerUnavailable)
timeout(wait_timeout, async {
self.tx
.send(ProbeCommand::ProbeBatch {
requests,
max_age,
reply: reply_tx,
})
.await
.map_err(|_| ProbeClientError::ManagerUnavailable)?;
reply_rx
.await
.map_err(|_| ProbeClientError::ManagerUnavailable)
})
.await
.map_err(|_| ProbeClientError::TimedOut)?
}
/// Probes a batch of targets for reachability.
@@ -192,6 +203,9 @@ pub enum ProbeClientError {
/// The manager returned no observations for the request.
#[error("shared probe manager returned no observations")]
EmptyResponse,
/// The manager did not answer before the caller's deadline.
#[error("shared probe manager timed out")]
TimedOut,
}
/// Background shared probe manager.
@@ -213,6 +227,27 @@ impl ProbeManager {
}
}
fn probe_worker_timeout(request_timeout: Duration) -> Duration {
PROBE_DNS_RESOLUTION_TIMEOUT
.saturating_add(request_timeout)
.saturating_add(PROBE_WORKER_GRACE)
}
fn probe_batch_wait_timeout(requests: &[ProbeRequest]) -> Duration {
let worker_timeout = requests
.iter()
.map(|request| probe_worker_timeout(request.timeout))
.max()
.unwrap_or(PROBE_MANAGER_REPLY_GRACE);
let wave_count = requests
.len()
.max(1)
.div_ceil(PROBE_BATCH_CONCURRENCY_ESTIMATE) as u32;
worker_timeout
.saturating_mul(wave_count)
.saturating_add(PROBE_MANAGER_REPLY_GRACE)
}
#[derive(Clone, Debug)]
struct ProbeCacheEntry {
normalized_target: String,
@@ -313,7 +348,23 @@ impl ProbeManagerState {
let semaphore = self.semaphore.clone();
join_set.spawn(async move {
let _permit = semaphore.acquire_owned().await.ok();
let observation = execute_probe(&request, &key.normalized_target).await;
let observation = match timeout(
probe_worker_timeout(request.timeout),
execute_probe(&request, &key.normalized_target),
)
.await
{
Ok(observation) => observation,
Err(_) => ProbeCacheEntry {
normalized_target: key.normalized_target.clone(),
resolved_ip: None,
kind: request.kind,
observed_at_unix_ms: now_unix_ms(),
reachable: false,
rtt_ms: None,
error: Some("probe worker timed out".to_string()),
},
};
(key, observation)
});
}
@@ -529,9 +580,15 @@ async fn resolve_target(normalized_target: &str) -> Result<(IpAddr, String), Str
return Ok((ip, ip.to_string()));
}
let mut addrs = tokio::net::lookup_host((normalized_target, 80))
.await
.map_err(|err| format!("unable to resolve target '{normalized_target}': {err}"))?;
let lookup = timeout(
PROBE_DNS_RESOLUTION_TIMEOUT,
tokio::net::lookup_host((normalized_target, 80)),
)
.await
.map_err(|_| format!("timed out resolving target '{normalized_target}'"))?;
let mut addrs =
lookup.map_err(|err| format!("unable to resolve target '{normalized_target}': {err}"))?;
let Some(addr) = addrs.next() else {
return Err(format!("unable to resolve target '{normalized_target}'"));
};
@@ -540,7 +597,11 @@ async fn resolve_target(normalized_target: &str) -> Result<(IpAddr, String), Str
#[cfg(test)]
mod tests {
use super::{ProbeKind, ProbeRequest, duration_to_millis, normalize_target};
use super::{
PROBE_DNS_RESOLUTION_TIMEOUT, PROBE_MANAGER_REPLY_GRACE, PROBE_WORKER_GRACE, ProbeKind,
ProbeRequest, duration_to_millis, normalize_target, probe_batch_wait_timeout,
probe_worker_timeout,
};
use std::time::Duration;
#[test]
@@ -576,4 +637,40 @@ mod tests {
assert_eq!(reachability.kind, ProbeKind::Reachability);
assert_eq!(rtt.kind, ProbeKind::RoundTripTime);
}
#[test]
fn probe_deadlines_include_dns_probe_and_reply_grace() {
let request = ProbeRequest::reachability(
"example.com",
super::ProbeClass::UiMonitor,
Duration::from_millis(750),
);
assert_eq!(
probe_worker_timeout(request.timeout),
PROBE_DNS_RESOLUTION_TIMEOUT + Duration::from_millis(750) + PROBE_WORKER_GRACE
);
assert_eq!(
probe_batch_wait_timeout(&[request]),
PROBE_DNS_RESOLUTION_TIMEOUT
+ Duration::from_millis(750)
+ PROBE_WORKER_GRACE
+ PROBE_MANAGER_REPLY_GRACE
);
}
#[test]
fn probe_batch_deadline_scales_for_multiple_probe_waves() {
let requests = vec![
ProbeRequest::reachability(
"example.com",
super::ProbeClass::UiMonitor,
Duration::from_secs(1),
);
257
];
assert_eq!(
probe_batch_wait_timeout(&requests),
probe_worker_timeout(Duration::from_secs(1)) * 2 + PROBE_MANAGER_REPLY_GRACE
);
}
}
src/rust/lqos_topology/src/runtime.rs
+9 -5
View File
@@ -1,5 +1,5 @@
use anyhow::{Context, Result};
use lqos_bus::{BusRequest, BusResponse, bus_request};
use lqos_bus::{BusRequest, BusResponse, bus_request_with_timeout};
use lqos_config::{
TopologyAttachmentEndpointStatus, TopologyAttachmentHealthEntry,
TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus,
@@ -21,6 +21,7 @@ use crate::{
};
const TOPOLOGY_PROBE_MAX_AGE_MS: u64 = 250;
const TOPOLOGY_PROBE_BUS_TIMEOUT: Duration = Duration::from_secs(8);
fn now_unix() -> Option<u64> {
SystemTime::now()
@@ -120,10 +121,13 @@ async fn probe_specs(
return Ok(HashMap::new());
}
let responses = bus_request(vec![BusRequest::ProbeBatch {
requests: probe_requests,
max_age_ms: TOPOLOGY_PROBE_MAX_AGE_MS,
}])
let responses = bus_request_with_timeout(
vec![BusRequest::ProbeBatch {
requests: probe_requests,
max_age_ms: TOPOLOGY_PROBE_MAX_AGE_MS,
}],
TOPOLOGY_PROBE_BUS_TIMEOUT,
)
.await
.map_err(|err| anyhow::anyhow!("unable to query shared probe manager: {err}"))?;
let Some(response) = responses.into_iter().next() else {
src/rust/lqosd/src/node_manager/local_api/config.rs
+9 -5
View File
@@ -7,7 +7,7 @@ use axum::body::Bytes;
use axum::http::StatusCode;
use axum::http::header;
use default_net::get_interfaces;
use lqos_bus::{BusRequest, BusResponse, bus_request};
use lqos_bus::{BusRequest, BusResponse, bus_request_with_timeout};
use lqos_config::{
Config, ConfigShapedDevices, NetworkJson, ShapedDevice, UserRole, WebUser, WebUsers,
};
@@ -17,13 +17,14 @@ use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::io::{Cursor, ErrorKind, Write};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
const COBRAND_FILE_NAME: &str = "cobrand.png";
const COBRAND_DISPLAY_HEIGHT_PX: u64 = 48;
const COBRAND_MAX_DISPLAY_WIDTH_PX: u64 = 176;
const COBRAND_MAX_DECODE_BYTES: usize = 64 * 1024 * 1024;
const PNG_SIGNATURE: &[u8; 8] = b"\x89PNG\r\n\x1a\n";
const CONFIG_BUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum CobrandUploadValidationError {
@@ -802,9 +803,12 @@ pub async fn update_lqosd_config_data(
let existing =
lqos_config::load_config().map_err(|_| "Unable to load the current config".to_string())?;
apply_secret_updates(existing.as_ref(), &mut config, &clear_secrets);
let mut responses = bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(config))])
.await
.map_err(|err| format!("Unable to update config: {err}"))?;
let mut responses = bus_request_with_timeout(
vec![BusRequest::UpdateLqosdConfig(Box::new(config))],
CONFIG_BUS_REQUEST_TIMEOUT,
)
.await
.map_err(|err| format!("Unable to update config: {err}"))?;
match responses.pop() {
Some(BusResponse::Ack) => Ok(()),
src/rust/lqosd/src/node_manager/local_api/lts.rs
+27 -7
View File
@@ -7,7 +7,7 @@ use crate::node_manager::local_api::circuit_count;
use axum::http::StatusCode;
pub use last_24_hours::*;
use lqos_bus::LtsCapabilitiesSummary;
use lqos_bus::{BusRequest, bus_request};
use lqos_bus::{BusRequest, bus_request_with_timeout};
use lqos_config::load_config;
use serde::{Deserialize, Serialize};
pub use shaper_status::ShaperStatus;
@@ -18,6 +18,8 @@ use tracing::{debug, info, warn};
use uuid::Uuid;
const SIGNUP_POLL_INTERVAL: Duration = Duration::from_secs(10);
const CONFIG_BUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const SERVICE_RESTART_TIMEOUT: Duration = Duration::from_secs(10);
const INSIGHT_FREE_TRIAL_STATUS_CODE: i32 = 2;
const INSIGHT_TRIAL_DAYS_HINT: i32 = 30;
@@ -132,9 +134,15 @@ async fn apply_insight_license(
.clone();
cfg.long_term_stats.gather_stats = true;
cfg.long_term_stats.license_key = Some(license_key.clone());
bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(cfg))])
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
bus_request_with_timeout(
vec![BusRequest::UpdateLqosdConfig(Box::new(cfg))],
CONFIG_BUS_REQUEST_TIMEOUT,
)
.await
.map_err(|err| {
warn!("Unable to update config with Insight license: {err}");
StatusCode::GATEWAY_TIMEOUT
})?;
crate::lts2_sys::capabilities::set_signup_bootstrap_active(false);
crate::lts2_sys::capabilities::clear_bootstrap_suppression();
crate::lts2_sys::capabilities::wake_control_channel();
@@ -147,9 +155,21 @@ async fn apply_insight_license(
info!("LQOSD configuration updated with new Insight license key.");
if restart_lqos_api {
let _ = Command::new("/bin/systemctl")
.args(["restart", "lqos_api"])
.output();
let restart = tokio::time::timeout(
SERVICE_RESTART_TIMEOUT,
tokio::task::spawn_blocking(|| {
Command::new("/bin/systemctl")
.args(["restart", "lqos_api"])
.output()
}),
)
.await;
match restart {
Ok(Ok(Ok(_))) => {}
Ok(Ok(Err(err))) => warn!("Unable to restart lqos_api: {err}"),
Ok(Err(err)) => warn!("lqos_api restart task failed: {err}"),
Err(_) => warn!("Timed out restarting lqos_api"),
}
}
Ok(())
src/rust/lqosd/src/node_manager/local_api/lts/last_24_hours.rs
+117 -124
View File
@@ -1,8 +1,13 @@
use crate::node_manager::shaper_queries_actor::ShaperQueryCommand;
use axum::http::StatusCode;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tokio::time::timeout;
use tracing::warn;
const SHAPER_QUERY_RESPONSE_TIMEOUT: Duration = Duration::from_secs(35);
const SHAPER_QUERY_SEND_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Serialize, Deserialize, Copy, Clone, Debug)]
pub struct ThroughputData {
time: i64, // Unix timestamp
@@ -136,20 +141,54 @@ pub struct RecentMedians {
pub last_week: (i64, i64),
}
async fn wait_for_shaper_response<T>(
context: &'static str,
rx: tokio::sync::oneshot::Receiver<Vec<T>>,
) -> Result<Vec<T>, StatusCode> {
match timeout(SHAPER_QUERY_RESPONSE_TIMEOUT, rx).await {
Ok(Ok(response)) => Ok(response),
Ok(Err(err)) => {
warn!("Error getting {context}: {err:?}");
Err(StatusCode::INTERNAL_SERVER_ERROR)
}
Err(_) => {
warn!("Timed out waiting for {context}");
Err(StatusCode::GATEWAY_TIMEOUT)
}
}
}
async fn send_shaper_query(
context: &'static str,
shaper_query: &tokio::sync::mpsc::Sender<ShaperQueryCommand>,
command: ShaperQueryCommand,
) -> Result<(), StatusCode> {
match timeout(SHAPER_QUERY_SEND_TIMEOUT, shaper_query.send(command)).await {
Ok(Ok(())) => Ok(()),
Ok(Err(_)) => {
warn!("Error sending {context} query");
Err(StatusCode::INTERNAL_SERVER_ERROR)
}
Err(_) => {
warn!("Timed out queueing {context} query");
Err(StatusCode::GATEWAY_TIMEOUT)
}
}
}
pub async fn throughput_period_data(
shaper_query: tokio::sync::mpsc::Sender<ShaperQueryCommand>,
seconds: i32,
) -> Result<Vec<ThroughputData>, StatusCode> {
super::insight_gate().await?;
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperThroughput { seconds, reply: tx })
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting total throughput: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"throughput",
&shaper_query,
ShaperQueryCommand::ShaperThroughput { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("total throughput", rx).await?;
Ok(throughput)
}
@@ -159,17 +198,13 @@ pub async fn packets_period_data(
) -> Result<Vec<FullPacketData>, StatusCode> {
super::insight_gate().await?;
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperPackets { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending packets period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting packets period: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"packets period",
&shaper_query,
ShaperQueryCommand::ShaperPackets { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("packets period", rx).await?;
Ok(throughput)
}
@@ -179,17 +214,13 @@ pub async fn percent_shaped_period_data(
) -> Result<Vec<PercentShapedWeb>, StatusCode> {
super::insight_gate().await?;
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperPercent { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending percent shaped period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting percent shaped: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"percent shaped period",
&shaper_query,
ShaperQueryCommand::ShaperPercent { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("percent shaped", rx).await?;
Ok(throughput)
}
@@ -199,17 +230,13 @@ pub async fn percent_flows_period_data(
) -> Result<Vec<FlowCountViewWeb>, StatusCode> {
super::insight_gate().await?;
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperFlows { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"flows period",
&shaper_query,
ShaperQueryCommand::ShaperFlows { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("flows", rx).await?;
Ok(throughput)
}
@@ -218,19 +245,14 @@ pub async fn rtt_histo_period_data(
seconds: i32,
) -> Result<Vec<ShaperRttHistogramEntry>, StatusCode> {
super::insight_gate().await?;
tracing::error!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperRttHistogram { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"RTT histogram period",
&shaper_query,
ShaperQueryCommand::ShaperRttHistogram { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("RTT histogram", rx).await?;
Ok(throughput)
}
@@ -239,19 +261,14 @@ pub async fn top10_downloaders_period_data(
seconds: i32,
) -> Result<Vec<Top10Circuit>, StatusCode> {
super::insight_gate().await?;
tracing::error!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperTopDownloaders { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"top downloaders period",
&shaper_query,
ShaperQueryCommand::ShaperTopDownloaders { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("top downloaders", rx).await?;
Ok(throughput)
}
@@ -260,19 +277,14 @@ pub async fn worst10_rtt_period_data(
seconds: i32,
) -> Result<Vec<Worst10RttCircuit>, StatusCode> {
super::insight_gate().await?;
tracing::error!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperWorstRtt { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"worst RTT period",
&shaper_query,
ShaperQueryCommand::ShaperWorstRtt { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("worst RTT", rx).await?;
Ok(throughput)
}
@@ -281,19 +293,14 @@ pub async fn worst10_rxmit_period_data(
seconds: i32,
) -> Result<Vec<Worst10RxmitCircuit>, StatusCode> {
super::insight_gate().await?;
tracing::error!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperWorstRxmit { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"worst retransmits period",
&shaper_query,
ShaperQueryCommand::ShaperWorstRxmit { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("worst retransmits", rx).await?;
Ok(throughput)
}
@@ -302,19 +309,14 @@ pub async fn top10_flows_period_data(
seconds: i32,
) -> Result<Vec<AsnFlowSizeWeb>, StatusCode> {
super::insight_gate().await?;
tracing::error!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperTopFlows { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"top flows period",
&shaper_query,
ShaperQueryCommand::ShaperTopFlows { seconds, reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("top flows", rx).await?;
Ok(throughput)
}
@@ -322,19 +324,14 @@ pub async fn recent_medians_data(
shaper_query: tokio::sync::mpsc::Sender<ShaperQueryCommand>,
) -> Result<Vec<RecentMedians>, StatusCode> {
super::insight_gate().await?;
tracing::debug!("rtt_histo_period");
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::ShaperRecentMedian { reply: tx })
.await
.map_err(|_| {
warn!("Error sending flows period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let throughput = rx.await.map_err(|e| {
warn!("Error getting flows: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"recent median",
&shaper_query,
ShaperQueryCommand::ShaperRecentMedian { reply: tx },
)
.await?;
let throughput = wait_for_shaper_response("recent medians", rx).await?;
Ok(throughput)
}
@@ -344,16 +341,12 @@ pub async fn cake_period_data(
) -> Result<Vec<CakeData>, StatusCode> {
super::insight_gate().await?;
let (tx, rx) = tokio::sync::oneshot::channel();
shaper_query
.send(ShaperQueryCommand::CakeTotals { seconds, reply: tx })
.await
.map_err(|_| {
warn!("Error sending cake stats period");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let response = rx.await.map_err(|e| {
warn!("Error getting cake stats: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
send_shaper_query(
"cake stats period",
&shaper_query,
ShaperQueryCommand::CakeTotals { seconds, reply: tx },
)
.await?;
let response = wait_for_shaper_response("cake stats", rx).await?;
Ok(response)
}
src/rust/lqosd/src/node_manager/shaper_queries_actor/queries.rs
+26 -11
View File
@@ -9,8 +9,11 @@ use crate::node_manager::shaper_queries_actor::ShaperQueryCommand;
use crate::node_manager::shaper_queries_actor::caches::Caches;
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::timeout;
use tracing::{info, warn};
const HISTORY_QUERY_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(30);
#[derive(Clone)]
struct HistoryClient {
control_tx: tokio::sync::mpsc::Sender<ControlChannelCommand>,
@@ -26,21 +29,30 @@ impl HistoryClient {
request: RemoteInsightRequest,
) -> Result<HistoryQueryResultPayload, ()> {
let (tx, rx) = oneshot::channel();
if self
.control_tx
.send(ControlChannelCommand::FetchHistory {
let send_result = timeout(
HISTORY_QUERY_ATTEMPT_TIMEOUT,
self.control_tx.send(ControlChannelCommand::FetchHistory {
request,
responder: tx,
})
.await
.is_err()
{
return Err(());
}),
)
.await;
match send_result {
Ok(Ok(())) => {}
Ok(Err(_)) => return Err(()),
Err(_) => {
warn!("Timed out queueing Insight history request");
return Err(());
}
}
match rx.await {
Ok(result) => result,
Err(_) => Err(()),
match timeout(HISTORY_QUERY_ATTEMPT_TIMEOUT, rx).await {
Ok(Ok(result)) => result,
Ok(Err(_)) => Err(()),
Err(_) => {
warn!("Timed out waiting for Insight history response");
Err(())
}
}
}
}
@@ -89,6 +101,9 @@ macro_rules! shaper_query {
break;
}
Err(()) => {
if my_reply.is_closed() {
break;
}
attempts += 1;
if attempts >= 3 {
warn!("History request exhausted retries");
src/rust/lqosd/src/node_manager/ws.rs
+68 -49
View File
@@ -7,7 +7,9 @@
use std::collections::{BTreeMap, HashSet};
use std::sync::Arc;
use std::time::Duration;
use crate::lts2_sys::control_channel::ControlChannelCommand;
use crate::node_manager::auth::{LoginResult, login_from_token};
use crate::node_manager::local_api::{
circuit, circuit_count, config, cpu_affinity, dashboard_themes, device_counts, directories,
@@ -55,6 +57,21 @@ mod ticker;
const WS_VERSION: &str = include_str!("../../../../VERSION_STRING");
const HANDSHAKE_TIMEOUT_SECS: u64 = 10;
const CONTROL_CHANNEL_SEND_TIMEOUT: Duration = Duration::from_secs(5);
async fn send_control_command(
control_tx: &tokio::sync::mpsc::Sender<ControlChannelCommand>,
command: ControlChannelCommand,
) -> bool {
match tokio::time::timeout(CONTROL_CHANNEL_SEND_TIMEOUT, control_tx.send(command)).await {
Ok(Ok(())) => true,
Ok(Err(_)) => false,
Err(_) => {
warn!("Timed out queueing websocket control-channel command");
false
}
}
}
/// Provides an Axum router for the websocket system. Exposes a single /ws route that supports
/// pubsub subscriptions and private commands.
@@ -742,14 +759,13 @@ async fn receive_channel_message(
_ => {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
let control_tx = request_state.private_state.control_tx();
if control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketList {
responder: reply_tx,
},
)
.await
.is_err()
if !send_control_command(
&control_tx,
ControlChannelCommand::SupportTicketList {
responder: reply_tx,
},
)
.await
{
if send_ws_response(
&tx,
@@ -766,8 +782,11 @@ async fn receive_channel_message(
tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await;
match result {
Ok(Ok(Ok(tickets))) => {
if send_ws_response(&tx, WsResponse::SupportTicketListResult { tickets })
.await
if send_ws_response(
&tx,
WsResponse::SupportTicketListResult { tickets },
)
.await
{
return true;
}
@@ -816,15 +835,14 @@ async fn receive_channel_message(
_ => {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
let control_tx = request_state.private_state.control_tx();
if control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketGet {
ticket_id,
responder: reply_tx,
},
)
.await
.is_err()
if !send_control_command(
&control_tx,
ControlChannelCommand::SupportTicketGet {
ticket_id,
responder: reply_tx,
},
)
.await
{
if send_ws_response(
&tx,
@@ -899,18 +917,17 @@ async fn receive_channel_message(
} else {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
let control_tx = request_state.private_state.control_tx();
if control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketCreate {
subject,
priority,
body,
commentor,
responder: reply_tx,
},
)
.await
.is_err()
if !send_control_command(
&control_tx,
ControlChannelCommand::SupportTicketCreate {
subject,
priority,
body,
commentor,
responder: reply_tx,
},
)
.await
{
if send_ws_response(
&tx,
@@ -923,12 +940,15 @@ async fn receive_channel_message(
return true;
}
} else {
let result = tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx)
.await;
let result =
tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await;
match result {
Ok(Ok(Ok(ticket))) => {
if send_ws_response(&tx, WsResponse::SupportTicketCreateResult { ticket })
.await
if send_ws_response(
&tx,
WsResponse::SupportTicketCreateResult { ticket },
)
.await
{
return true;
}
@@ -988,18 +1008,17 @@ async fn receive_channel_message(
.unwrap_or(0);
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
let control_tx = request_state.private_state.control_tx();
if control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketAddComment {
ticket_id,
commentor,
body,
date,
responder: reply_tx,
},
)
.await
.is_err()
if !send_control_command(
&control_tx,
ControlChannelCommand::SupportTicketAddComment {
ticket_id,
commentor,
body,
date,
responder: reply_tx,
},
)
.await
{
if send_ws_response(
&tx,
@@ -1012,8 +1031,8 @@ async fn receive_channel_message(
return true;
}
} else {
let result = tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx)
.await;
let result =
tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await;
match result {
Ok(Ok(Ok(()))) => {
if send_ws_response(
src/rust/lqosd/src/node_manager/ws/single_user_channels.rs
+61 -32
View File
@@ -6,6 +6,7 @@ pub(crate) mod flows_by_circuit;
mod ping_monitor;
mod tree_attached_circuits;
use crate::lts2_sys::control_channel::ControlChannelCommand;
use crate::node_manager::ws::messages::{PrivateRequest, WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::cake_watcher::cake_watcher;
use crate::node_manager::ws::single_user_channels::circuit::circuit_watcher;
@@ -13,17 +14,52 @@ use crate::node_manager::ws::single_user_channels::circuit_metrics::watch_circui
use crate::node_manager::ws::single_user_channels::ping_monitor::ping_monitor;
use crate::node_manager::ws::single_user_channels::tree_attached_circuits::watch_tree_attached_circuits;
use lqos_probe::ProbeClient;
use std::{sync::Arc, time::Duration};
use tokio::spawn;
use tokio::sync::mpsc::Sender;
use tracing::info;
use tokio::sync::mpsc::{Sender, error::TrySendError};
use tokio::time::timeout;
use tracing::{debug, info};
const CONTROL_CHANNEL_SEND_TIMEOUT: Duration = Duration::from_secs(5);
/// Sends a private watcher payload without waiting behind a slow websocket.
pub(super) fn try_send_private_payload(
tx: &Sender<Arc<Vec<u8>>>,
payload: Arc<Vec<u8>>,
channel_name: &'static str,
) -> bool {
match tx.try_send(payload) {
Ok(()) => true,
Err(TrySendError::Full(_)) => {
debug!("{channel_name} outbound queue full; stopping watcher");
false
}
Err(TrySendError::Closed(_)) => {
debug!("{channel_name} channel closed");
false
}
}
}
async fn send_control_command(
control_tx: &tokio::sync::mpsc::Sender<ControlChannelCommand>,
command: ControlChannelCommand,
context: &'static str,
) {
match timeout(CONTROL_CHANNEL_SEND_TIMEOUT, control_tx.send(command)).await {
Ok(Ok(())) => {}
Ok(Err(_)) => debug!("{context} control channel is closed"),
Err(_) => debug!("{context} timed out queueing control channel command"),
}
}
pub struct PrivateState {
tx: Sender<std::sync::Arc<Vec<u8>>>,
tx: Sender<Arc<Vec<u8>>>,
bus_tx: Sender<(
tokio::sync::oneshot::Sender<lqos_bus::BusReply>,
lqos_bus::BusRequest,
)>,
control_tx: tokio::sync::mpsc::Sender<crate::lts2_sys::control_channel::ControlChannelCommand>,
control_tx: tokio::sync::mpsc::Sender<ControlChannelCommand>,
probe_client: ProbeClient,
browser_language: Option<String>,
chatbot_request: Option<u64>,
@@ -41,9 +77,7 @@ impl PrivateState {
tokio::sync::oneshot::Sender<lqos_bus::BusReply>,
lqos_bus::BusRequest,
)>,
control_tx: tokio::sync::mpsc::Sender<
crate::lts2_sys::control_channel::ControlChannelCommand,
>,
control_tx: tokio::sync::mpsc::Sender<ControlChannelCommand>,
probe_client: ProbeClient,
browser_language: Option<String>,
) -> Self {
@@ -62,9 +96,7 @@ impl PrivateState {
}
}
pub fn control_tx(
&self,
) -> tokio::sync::mpsc::Sender<crate::lts2_sys::control_channel::ControlChannelCommand> {
pub fn control_tx(&self) -> tokio::sync::mpsc::Sender<ControlChannelCommand> {
self.control_tx.clone()
}
@@ -205,7 +237,7 @@ impl PrivateState {
message: message.to_string(),
};
if let Ok(payload) = encode_ws_message(&response) {
let _ = self.tx.send(payload).await;
let _ = try_send_private_payload(&self.tx, payload, "Chatbot");
}
return;
}
@@ -220,7 +252,7 @@ impl PrivateState {
let text = String::from_utf8_lossy(&chunk).to_string();
let response = WsResponse::ChatbotChunk { text };
if let Ok(payload) = encode_ws_message(&response) {
if to_client.send(payload).await.is_err() {
if !try_send_private_payload(&to_client, payload, "Chatbot") {
break;
}
} else {
@@ -229,17 +261,17 @@ impl PrivateState {
}
});
let _ = self
.control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::StartChat {
request_id,
browser_ts_ms,
browser_language: self.browser_language.clone(),
stream: stream_tx,
},
)
.await;
send_control_command(
&self.control_tx,
ControlChannelCommand::StartChat {
request_id,
browser_ts_ms,
browser_language: self.browser_language.clone(),
stream: stream_tx,
},
"StartChat",
)
.await;
info!(
"[chatbot] starting session request_id={} browser_ts_ms={:?}",
request_id, browser_ts_ms
@@ -250,15 +282,12 @@ impl PrivateState {
let Some(request_id) = self.chatbot_request else {
return;
};
let _ = self
.control_tx
.send(
crate::lts2_sys::control_channel::ControlChannelCommand::ChatSend {
request_id,
text,
},
)
.await;
send_control_command(
&self.control_tx,
ControlChannelCommand::ChatSend { request_id, text },
"ChatSend",
)
.await;
}
}
src/rust/lqosd/src/node_manager/ws/single_user_channels/cake_watcher.rs
+5 -5
View File
@@ -1,4 +1,5 @@
use crate::node_manager::ws::messages::{WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::try_send_private_payload;
use lqos_queue_tracker::{add_watched_queue, get_raw_circuit_data, still_watching};
pub(super) async fn cake_watcher(
@@ -16,11 +17,10 @@ pub(super) async fn cake_watcher(
if let lqos_bus::BusResponse::RawQueueData(Some(msg)) = get_raw_circuit_data(&circuit_id) {
let response = WsResponse::CakeWatcher { data: *msg };
if let Ok(payload) = encode_ws_message(&response) {
let send_result = tx.send(payload).await;
if send_result.is_err() {
break;
}
if let Ok(payload) = encode_ws_message(&response)
&& !try_send_private_payload(&tx, payload, "CakeWatcher")
{
break;
}
}
}
src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs
+2 -1
View File
@@ -1,5 +1,6 @@
use crate::node_manager::local_api::circuit_activity::{CircuitSummaryData, circuit_flow_counts};
use crate::node_manager::ws::messages::{CircuitDevicesResult, WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::try_send_private_payload;
use crate::node_manager::ws::ticker::all_circuits;
use crate::rtt_exclusions;
use crate::throughput_tracker::{circuit_current_qoo, circuit_current_rtt_p50_nanos};
@@ -121,7 +122,7 @@ pub(super) async fn circuit_watcher(
};
if let Ok(payload) = encode_ws_message(&response) {
if tx.send(payload).await.is_err() {
if !try_send_private_payload(&tx, payload, "CircuitWatcher") {
info!("CircuitWatcher channel is gone");
break;
}
src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit_metrics.rs
+2 -3
View File
@@ -2,9 +2,9 @@ use crate::node_manager::local_api::circuit_live::{
CircuitLiveMetrics, CircuitMetricsQuery, circuit_live_metrics,
};
use crate::node_manager::ws::messages::{WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::try_send_private_payload;
use std::time::Duration;
use tokio::time::MissedTickBehavior;
use tracing::debug;
pub(super) async fn watch_circuit_metrics(
query: CircuitMetricsQuery,
@@ -33,8 +33,7 @@ pub(super) async fn watch_circuit_metrics(
if let Some(response) = response {
match encode_ws_message(&response) {
Ok(payload) => {
if tx.send(payload).await.is_err() {
debug!("CircuitMetrics watcher channel closed");
if !try_send_private_payload(&tx, payload, "CircuitMetrics") {
break;
}
}
src/rust/lqosd/src/node_manager/ws/single_user_channels/ping_monitor.rs
+20 -19
View File
@@ -1,8 +1,9 @@
use crate::node_manager::ws::messages::{PingState, WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::try_send_private_payload;
use lqos_probe::{ProbeClass, ProbeClient};
use std::time::Duration;
use tokio::time::MissedTickBehavior;
use tracing::{debug, info};
use tracing::debug;
const UI_MONITOR_PROBE_MAX_AGE: Duration = Duration::from_millis(250);
@@ -32,13 +33,15 @@ pub(super) async fn ping_monitor(
}
};
for ((ip, label), observation) in ip_addresses.iter().zip(observations) {
'observations: for ((ip, label), observation) in ip_addresses.iter().zip(observations) {
if observation.reachable {
let ping_time = observation
.rtt_ms
.map(|rtt_ms| Duration::from_secs_f64(rtt_ms / 1000.0))
.unwrap_or_else(|| Duration::from_secs(0));
send_alive(tx.clone(), ip.clone(), ping_time, label.clone()).await;
if !send_alive(&tx, ip.clone(), ping_time, label.clone()) {
break 'observations;
}
} else {
if let Some(error) = observation.error.as_deref() {
debug!(
@@ -46,7 +49,9 @@ pub(super) async fn ping_monitor(
observation.normalized_target, error
);
}
send_timeout(tx.clone(), ip.clone()).await;
if !send_timeout(&tx, ip.clone()) {
break 'observations;
}
}
}
@@ -55,7 +60,7 @@ pub(super) async fn ping_monitor(
result: PingState::ChannelTest,
};
if let Ok(payload) = encode_ws_message(&channel_test) {
if tx.send(payload).await.is_err() {
if !try_send_private_payload(&tx, payload, "PingMonitor") {
debug!("Channel is gone");
break;
}
@@ -65,24 +70,22 @@ pub(super) async fn ping_monitor(
}
}
async fn send_timeout(tx: tokio::sync::mpsc::Sender<std::sync::Arc<Vec<u8>>>, ip: String) {
fn send_timeout(tx: &tokio::sync::mpsc::Sender<std::sync::Arc<Vec<u8>>>, ip: String) -> bool {
let result = WsResponse::PingMonitor {
ip,
result: PingState::NoResponse,
};
if let Ok(payload) = encode_ws_message(&result)
&& tx.send(payload).await.is_err()
{
info!("Channel is gone");
}
encode_ws_message(&result)
.map(|payload| try_send_private_payload(tx, payload, "PingMonitor"))
.unwrap_or(false)
}
async fn send_alive(
tx: tokio::sync::mpsc::Sender<std::sync::Arc<Vec<u8>>>,
fn send_alive(
tx: &tokio::sync::mpsc::Sender<std::sync::Arc<Vec<u8>>>,
ip: String,
ping_time: Duration,
label: String,
) {
) -> bool {
let result = WsResponse::PingMonitor {
ip,
result: PingState::Ping {
@@ -90,9 +93,7 @@ async fn send_alive(
label,
},
};
if let Ok(payload) = encode_ws_message(&result)
&& tx.send(payload).await.is_err()
{
info!("Channel is gone");
}
encode_ws_message(&result)
.map(|payload| try_send_private_payload(tx, payload, "PingMonitor"))
.unwrap_or(false)
}
src/rust/lqosd/src/node_manager/ws/single_user_channels/tree_attached_circuits.rs
+2 -3
View File
@@ -2,9 +2,9 @@ use crate::node_manager::local_api::tree_attached_circuits::{
TreeAttachedCircuitsPage, TreeAttachedCircuitsQuery, tree_attached_circuits,
};
use crate::node_manager::ws::messages::{WsResponse, encode_ws_message};
use crate::node_manager::ws::single_user_channels::try_send_private_payload;
use std::time::Duration;
use tokio::time::MissedTickBehavior;
use tracing::debug;
pub(super) async fn watch_tree_attached_circuits(
query: TreeAttachedCircuitsQuery,
@@ -29,8 +29,7 @@ pub(super) async fn watch_tree_attached_circuits(
if let Some(response) = response {
match encode_ws_message(&response) {
Ok(payload) => {
if tx.send(payload).await.is_err() {
debug!("TreeAttachedCircuits watcher channel closed");
if !try_send_private_payload(&tx, payload, "TreeAttachedCircuits") {
break;
}
}
src/rust/lqosd/src/node_manager/ws/ticker/network_tree.rs
+41 -28
View File
@@ -3,7 +3,42 @@ use crate::node_manager::ws::publish_subscribe::PubSub;
use crate::node_manager::ws::published_channels::PublishedChannels;
use lqos_bus::{BusReply, BusRequest, BusResponse, Circuit};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc::Sender;
use tokio::time::timeout;
const INTERNAL_BUS_TIMEOUT: Duration = Duration::from_secs(5);
async fn request_internal_bus(
context: &'static str,
bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
request: BusRequest,
) -> Option<BusReply> {
let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
match timeout(INTERNAL_BUS_TIMEOUT, bus_tx.send((tx, request))).await {
Ok(Ok(())) => {}
Ok(Err(err)) => {
tracing::warn!("{context}: failed to send request to bus: {err:?}");
return None;
}
Err(_) => {
tracing::warn!("{context}: timed out queueing request to bus");
return None;
}
}
match timeout(INTERNAL_BUS_TIMEOUT, rx).await {
Ok(Ok(replies)) => Some(replies),
Ok(Err(err)) => {
tracing::warn!("{context}: failed to receive response from bus: {err:?}");
None
}
Err(_) => {
tracing::warn!("{context}: timed out waiting for bus response");
None
}
}
}
pub async fn network_tree(
channels: Arc<PubSub>,
@@ -16,21 +51,10 @@ pub async fn network_tree(
return;
}
let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
let request = BusRequest::GetFullNetworkMap;
if let Err(e) = bus_tx.send((tx, request)).await {
tracing::warn!("NetworkTree: failed to send request to bus: {:?}", e);
let Some(replies) =
request_internal_bus("NetworkTree", bus_tx, BusRequest::GetFullNetworkMap).await
else {
return;
}
let replies = match rx.await {
Ok(r) => r,
Err(e) => {
tracing::warn!(
"NetworkTree: failed to receive throughput from bus: {:?}",
e
);
return;
}
};
for reply in replies.responses.into_iter() {
if let BusResponse::NetworkMap(nodes) = reply {
@@ -43,21 +67,10 @@ pub async fn network_tree(
pub async fn all_circuits(
bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
) -> Vec<Circuit> {
let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
let request = BusRequest::GetAllCircuits;
if let Err(e) = bus_tx.send((tx, request)).await {
tracing::warn!("AllCircuits: failed to send request to bus: {:?}", e);
let Some(replies) =
request_internal_bus("AllCircuits", bus_tx, BusRequest::GetAllCircuits).await
else {
return Vec::new();
}
let replies = match rx.await {
Ok(r) => r,
Err(e) => {
tracing::warn!(
"AllCircuits: failed to receive throughput from bus: {:?}",
e
);
return Vec::new();
}
};
for reply in replies.responses.into_iter() {
if let BusResponse::CircuitData(circuits) = reply {