ExternalVendorCode/chirpstack
1
0
Fork 0
mirror of https://github.com/chirpstack/chirpstack.git synced 2025-04-13 22:23:07 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Improve log output.

Browse Source 
This improves the log output on the following points:

In case when .context(...) is used, the error printed in the logs would
only show the outer error. This has been improved to show the complete
chain of errors.

In many cases error! and warn! logs were used for logging related to
user input. This is very inconvenient when using log output for
monitoring as it can trigger alerts based on end-user mistakes. While
refactoring the logs, the following criteria has been used:

* error = Messages that need attention and that are within the domain
  of the ChirpStack administrator.
* warn = Messages that should stand out of the other log messages, but
  that might be end-user (or end-device) related. Depending on if you
  have control of the full chain or not, you might or might not want to
  be alerted based on these messages.

Tracing spans have been improved to make it easier to correlate between
log messages and events.
This commit is contained in:
Orne Brocaar 2023-10-24 11:20:57 +01:00
parent dc5be942d4
commit 3d4455b968
36 changed files with 254 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
chirpstack/src/api/auth/validator.rs
View File

@ -1,3 +1,4 @@
use anyhow::Result;
use async_trait::async_trait;
use diesel::dsl;
use diesel::prelude::*;
@ -10,6 +11,7 @@ use lrwn::EUI64;
use super::error::Error;
use crate::api::auth::AuthID;
use crate::helpers::errors::PrintFullError;
use crate::storage::get_db_conn;
use crate::storage::schema::{
api_key, application, device, device_profile, gateway, multicast_group, tenant_user, user,
@ -67,7 +69,7 @@ pub trait Validator {
}
Err(e) => {
error!(
error = %e,
error = %e.full(),
"Validator function error"
);
Err(Status::internal(""))

39
chirpstack/src/api/backend/mod.rs
View File

@ -8,12 +8,13 @@ use anyhow::Result;
use redis::streams::StreamReadReply;
use tokio::sync::oneshot;
use tokio::task;
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, span, warn, Instrument, Level};
use uuid::Uuid;
use warp::{http::StatusCode, Filter, Reply};
use crate::backend::{joinserver, keywrap, roaming};
use crate::downlink::data_fns;
use crate::helpers::errors::PrintFullError;
use crate::storage::{
device_session, error::Error as StorageError, get_redis_conn, passive_roaming, redis_key,
};
@ -26,7 +27,7 @@ use lrwn::{AES128Key, NetID, EUI64};
pub async fn setup() -> Result<()> {
let conf = config::get();
if conf.backend_interfaces.bind.is_empty() {
warn!("Backend interfaces API is disabled");
info!("Backend interfaces API interface is disabled");
return Ok(());
}
@ -69,6 +70,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
body.advance(cnt);
}
// TODO: should this be improved?
debug!("JSON: {}", String::from_utf8(b.clone()).unwrap_or_default());
let bp: BasePayload = match serde_json::from_slice(&b) {
@ -79,7 +81,12 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
}
};
info!(sender_id = %hex::encode(&bp.sender_id), transaction_id = %bp.transaction_id, message_type = ?bp.message_type, "Request received");
let span = span!(Level::INFO, "request", sender_id = %hex::encode(&bp.sender_id), receiver_id = %hex::encode(&bp.receiver_id), message_type = ?bp.message_type, transaction_id = bp.transaction_id);
_handle_request(bp, b).instrument(span).await
}
pub async fn _handle_request(bp: BasePayload, b: Vec<u8>) -> http::Response<hyper::Body> {
info!("Request received");
let sender_client = {
if bp.sender_id.len() == 8 {
@ -87,7 +94,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
let sender_id = match EUI64::from_slice(&bp.sender_id) {
Ok(v) => v,
Err(e) => {
error!(error = %e, "Error decoding SenderID as EUI64");
warn!(error = %e.full(), "Error decoding SenderID as EUI64");
let msg = format!("Error decoding SenderID: {}", e);
let pl = bp.to_base_payload_result(backend::ResultCode::MalformedRequest, &msg);
return warp::reply::json(&pl).into_response();
@ -97,7 +104,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
match joinserver::get(&sender_id) {
Ok(v) => v,
Err(_) => {
error!(sender_id = %sender_id, "Unknown SenderID");
warn!("Unknown SenderID");
let msg = format!("Unknown SenderID: {}", sender_id);
let pl = bp.to_base_payload_result(backend::ResultCode::UnknownSender, &msg);
return warp::reply::json(&pl).into_response();
@ -108,7 +115,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
let sender_id = match NetID::from_slice(&bp.sender_id) {
Ok(v) => v,
Err(e) => {
error!(error = %e, "Error decoding SenderID as NetID");
warn!(error = %e.full(), "Error decoding SenderID as NetID");
let msg = format!("Error decoding SenderID: {}", e);
let pl = bp.to_base_payload_result(backend::ResultCode::MalformedRequest, &msg);
return warp::reply::json(&pl).into_response();
@ -118,7 +125,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
match roaming::get(&sender_id) {
Ok(v) => v,
Err(_) => {
error!(sender_id = %sender_id, "Unknown SenderID");
warn!("Unknown SenderID");
let msg = format!("Unknown SenderID: {}", sender_id);
let pl = bp.to_base_payload_result(backend::ResultCode::UnknownSender, &msg);
return warp::reply::json(&pl).into_response();
@ -126,7 +133,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
}
} else {
// Unknown size
error!(sender_id = ?bp.sender_id, "Invalid SenderID length");
warn!("Invalid SenderID length");
let pl = bp.to_base_payload_result(
backend::ResultCode::MalformedRequest,
"Invalid SenderID length",
@ -139,7 +146,7 @@ pub async fn handle_request(mut body: impl warp::Buf) -> http::Response<hyper::B
if bp.is_answer() {
tokio::spawn(async move {
if let Err(e) = handle_async_ans(&bp, &b).await {
error!(error = %e, "Handle async answer error");
error!(error = %e.full(), "Handle async answer error");
}
});
return warp::reply::with_status("", StatusCode::OK).into_response();
@ -197,7 +204,7 @@ async fn handle_pr_start_req(
};
if let Err(e) = sender_client.pr_start_ans(backend::Role::FNS, &ans).await {
error!(error = %e, "Send async PRStartAns error");
error!(error = %e.full(), "Send async PRStartAns error");
}
});
@ -345,7 +352,7 @@ async fn handle_pr_stop_req(
};
if let Err(e) = sender_client.pr_stop_ans(backend::Role::SNS, &ans).await {
error!(error = %e, "Send async PRStopAns error");
error!(error = %e.full(), "Send async PRStopAns error");
}
});
@ -373,7 +380,7 @@ async fn _handle_pr_stop_req(b: &[u8]) -> Result<backend::PRStopAnsPayload> {
for sess_id in sess_ids {
if let Err(e) = passive_roaming::delete(sess_id).await {
error!(error = %e, "Delete passive-roaming device-session error");
error!(error = %e.full(), "Delete passive-roaming device-session error");
}
}
@ -415,7 +422,7 @@ async fn handle_xmit_data_req(
};
if let Err(e) = sender_client.xmit_data_ans(sender_role, &ans).await {
error!(error = %e, "Send async XmitDataAns error");
error!(error = %e.full(), "Send async XmitDataAns error");
}
});
@ -534,7 +541,11 @@ pub async fn get_async_receiver(
}
}
_ => {
error!(key = %k, "Unexpected key in async stream");
error!(
transaction_id = transaction_id,
key = %key,
"Unexpected key in async stream"
);
}
}
}

5
chirpstack/src/api/mod.rs
View File

@ -39,6 +39,7 @@ use chirpstack_api::api::user_service_server::UserServiceServer;
use super::config;
use crate::api::auth::validator;
use crate::helpers::errors::PrintFullError;
use crate::monitoring::prometheus;
use crate::requestlog;
@ -420,8 +421,8 @@ where
};
task::spawn(async move {
if let Err(err) = requestlog::log_request(&req_log).await {
error!("Log request error, error: {}", err);
if let Err(e) = requestlog::log_request(&req_log).await {
error!(error = %e.full(), "Log request error");
}
});
}

5
chirpstack/src/api/oidc.rs
View File

@ -18,6 +18,7 @@ use tracing::{error, trace};
use warp::{Rejection, Reply};
use crate::config;
use crate::helpers::errors::PrintFullError;
use crate::storage::{get_redis_conn, redis_key};
pub type User = UserInfoClaims<EmptyAdditionalClaims, CoreGenderClaim>;
@ -32,7 +33,7 @@ pub async fn login_handler() -> Result<impl Reply, Rejection> {
let client = match get_client().await {
Ok(v) => v,
Err(e) => {
error!(error = %e, "Get OIDC client error");
error!(error = %e.full(), "Get OIDC client error");
return Ok(warp::reply::with_status(
"Internal error",
warp::http::StatusCode::INTERNAL_SERVER_ERROR,
@ -52,7 +53,7 @@ pub async fn login_handler() -> Result<impl Reply, Rejection> {
.url();
if let Err(e) = store_nonce(&csrf_state, &nonce).await {
error!(error = %e, "Store nonce error");
error!(error = %e.full(), "Store nonce error");
return Ok(warp::reply::with_status(
"Internal error",
warp::http::StatusCode::INTERNAL_SERVER_ERROR,

7
chirpstack/src/backend/joinserver.rs
View File

@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use anyhow::Result;
use tracing::{info, span, Level};
use tracing::info;
use crate::config;
use backend::{Client, ClientConfig};
@ -17,10 +17,7 @@ pub fn setup() -> Result<()> {
let conf = config::get();
for js in &conf.join_server.servers {
let span = span!(Level::INFO, "setup", join_eui = %js.join_eui);
let _guard = span.enter();
info!("Configuring Join Server");
info!(join_eui = %js.join_eui, "Configuring Join Server");
let c = Client::new(ClientConfig {
sender_id: conf.network.net_id.to_vec(),

4
chirpstack/src/downlink/classb.rs
View File

@ -3,7 +3,7 @@ use aes::cipher::{BlockEncrypt, KeyInit};
use aes::{Aes128, Block};
use anyhow::Result;
use chrono::Duration;
use tracing::info;
use tracing::debug;
use lrwn::DevAddr;
@ -63,7 +63,7 @@ pub fn get_next_ping_slot_after(
+ (*SLOT_LEN * ((ping_offset + n * ping_period) as i32));
if ping_slot_ts > after_gps_epoch_ts {
info!(
debug!(
dev_addr = %dev_addr,
beacon_start_time_s = beacon_start_ts.num_seconds(),
after_beacon_start_time_ms = (ping_slot_ts - beacon_start_ts).num_milliseconds(),

30
chirpstack/src/downlink/data.rs
View File

@ -5,7 +5,7 @@ use std::time::Duration;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use rand::Rng;
use tracing::{error, span, trace, warn, Instrument, Level};
use tracing::{span, trace, warn, Instrument, Level};
use crate::api::backend::get_async_receiver;
use crate::api::helpers::{FromProto, ToProto};
@ -65,9 +65,11 @@ impl Data {
must_ack: bool,
mac_commands: Vec<lrwn::MACCommandSet>,
) -> Result<()> {
let span = span!(Level::TRACE, "data_down", downlink_id = %ufs.uplink_set_id);
let downlink_id: u32 = rand::thread_rng().gen();
let span = span!(Level::INFO, "data_down", downlink_id = downlink_id);
Data::_handle_response(
downlink_id,
ufs,
dev_gw_rx_info,
tenant,
@ -97,9 +99,11 @@ impl Data {
must_ack: bool,
mac_commands: Vec<lrwn::MACCommandSet>,
) -> Result<()> {
let span = span!(Level::TRACE, "data_down", downlink_id = %ufs.uplink_set_id);
let downlink_id: u32 = rand::thread_rng().gen();
let span = span!(Level::INFO, "data_down", downlink_id = downlink_id);
Data::_handle_response_relayed(
downlink_id,
relay_ctx,
ufs,
dev_gw_rx_info,
@ -117,15 +121,18 @@ impl Data {
}
pub async fn handle_schedule_next_queue_item(device: device::Device) -> Result<()> {
let span = span!(Level::TRACE, "schedule", dev_eui = %device.dev_eui);
let downlink_id: u32 = rand::thread_rng().gen();
let span =
span!(Level::INFO, "schedule", dev_eui = %device.dev_eui, downlink_id = downlink_id);
Data::_handle_schedule_next_queue_item(device)
Data::_handle_schedule_next_queue_item(downlink_id, device)
.instrument(span)
.await
}
#[allow(clippy::too_many_arguments)]
async fn _handle_response(
downlink_id: u32,
ufs: UplinkFrameSet,
dev_gw_rx_info: internal::DeviceGatewayRxInfo,
tenant: tenant::Tenant,
@ -160,7 +167,7 @@ impl Data {
device_gateway_rx_info: Some(dev_gw_rx_info),
downlink_gateway: None,
downlink_frame: gw::DownlinkFrame {
downlink_id: rand::thread_rng().gen(),
downlink_id,
..Default::default()
},
downlink_frame_items: Vec::new(),
@ -194,6 +201,7 @@ impl Data {
#[allow(clippy::too_many_arguments)]
async fn _handle_response_relayed(
downlink_id: u32,
relay_ctx: RelayContext,
ufs: UplinkFrameSet,
dev_gw_rx_info: internal::DeviceGatewayRxInfo,
@ -229,7 +237,7 @@ impl Data {
device_gateway_rx_info: Some(dev_gw_rx_info),
downlink_gateway: None,
downlink_frame: gw::DownlinkFrame {
downlink_id: rand::thread_rng().gen(),
downlink_id,
..Default::default()
},
downlink_frame_items: Vec::new(),
@ -253,7 +261,7 @@ impl Data {
Ok(())
}
async fn _handle_schedule_next_queue_item(dev: device::Device) -> Result<()> {
async fn _handle_schedule_next_queue_item(downlink_id: u32, dev: device::Device) -> Result<()> {
trace!("Handle schedule next-queue item flow");
let dp = device_profile::get(&dev.device_profile_id).await?;
@ -280,7 +288,7 @@ impl Data {
device_gateway_rx_info: Some(dev_gw),
downlink_gateway: None,
downlink_frame: gw::DownlinkFrame {
downlink_id: rand::thread_rng().gen(),
downlink_id,
..Default::default()
},
downlink_frame_items: vec![],
@ -1485,7 +1493,7 @@ impl Data {
// available slot).
if !found {
if free_slots.is_empty() {
error!(relay_dev_eui = %self.device.dev_eui, "Relay does not have any free UpdateUplinkListReq slots");
warn!(relay_dev_eui = %self.device.dev_eui, "Relay does not have any free UpdateUplinkListReq slots");
continue;
}
@ -1867,7 +1875,7 @@ impl Data {
// available slot).
if !found {
if free_slots.is_empty() {
error!(relay_dev_eui = %self.device.dev_eui, "Relay does have have any free FilterListReq slots");
warn!(relay_dev_eui = %self.device.dev_eui, "Relay does have have any free FilterListReq slots");
continue;
}

6
chirpstack/src/downlink/data_fns.rs
View File

@ -21,7 +21,11 @@ impl Data {
pl: backend::XmitDataReqPayload,
dl_meta: backend::DLMetaData,
) -> Result<()> {
let span = span!(Level::INFO, "xmit_data_req_pr");
let span = span!(
Level::INFO,
"xmit_data_req_pr",
transaction_id = pl.base.transaction_id
);
Data::_handle(pl, dl_meta).instrument(span).await
}

2
chirpstack/src/downlink/multicast.rs
View File

@ -30,7 +30,7 @@ pub struct Multicast {
impl Multicast {
pub async fn handle_schedule_queue_item(qi: multicast::MulticastGroupQueueItem) -> Result<()> {
let span = span!(Level::TRACE, "schedule", multicast_group_id = %qi.multicast_group_id, gateway_id = %qi.gateway_id);
let span = span!(Level::INFO, "multicast", multicast_group_id = %qi.multicast_group_id, gateway_id = %qi.gateway_id);
Multicast::_handle_schedule_queue_item(qi)
.instrument(span)

8
chirpstack/src/downlink/roaming.rs
View File

@ -25,9 +25,11 @@ pub struct PassiveRoamingDownlink {
impl PassiveRoamingDownlink {
pub async fn handle(ufs: UplinkFrameSet, phy: Vec<u8>, dl_meta: DLMetaData) -> Result<()> {
let span = span!(Level::TRACE, "passive_roaming");
let fut = PassiveRoamingDownlink::_handle(ufs, phy, dl_meta);
fut.instrument(span).await
let span = span!(Level::INFO, "passive_roaming");
PassiveRoamingDownlink::_handle(ufs, phy, dl_meta)
.instrument(span)
.await
}
async fn _handle(ufs: UplinkFrameSet, phy: Vec<u8>, dl_meta: DLMetaData) -> Result<()> {

3
chirpstack/src/downlink/scheduler.rs
View File

@ -5,6 +5,7 @@ use tracing::{error, trace};
use super::data;
use super::multicast as mcast;
use crate::config;
use crate::helpers::errors::PrintFullError;
use crate::storage::{device, multicast};
pub async fn class_b_c_scheduler_loop() {
@ -79,7 +80,7 @@ pub async fn schedule_multicast_group_queue_batch(size: usize) -> Result<()> {
for qi in items {
let handle = tokio::spawn(async move {
if let Err(e) = mcast::Multicast::handle_schedule_queue_item(qi).await {
error!(error = %e, "Schedule multicast-group queue item failed");
error!(error = %e.full(), "Schedule multicast-group queue item failed");
}
});
handles.push(handle);

2
chirpstack/src/downlink/tx_ack.rs
View File

@ -39,7 +39,7 @@ pub struct TxAck {
impl TxAck {
pub async fn handle(tx_ack: gw::DownlinkTxAck) {
let span = span!(Level::TRACE, "tx_ack", downlink_id = tx_ack.downlink_id);
let span = span!(Level::INFO, "tx_ack", downlink_id = tx_ack.downlink_id);
if let Err(e) = TxAck::_handle(tx_ack).instrument(span).await {
error!(error = %e, "Handling tx ack error");
}

3
chirpstack/src/eventlog.rs
View File

@ -10,6 +10,7 @@ use tokio::task;
use tracing::{debug, error, trace};
use crate::config;
use crate::helpers::errors::PrintFullError;
use crate::storage::{get_redis_conn, redis_key};
use chirpstack_api::{api, integration};
@ -298,7 +299,7 @@ pub async fn get_event_logs(
return Err(e);
}
error!(key = %k, error = %e, "Parsing frame-log error");
error!(key = %k, error = %e.full(), "Parsing frame-log error");
}
}
}

11
chirpstack/src/framelog.rs
View File

@ -14,6 +14,7 @@ use tracing::{debug, error, trace, warn};
use lrwn::EUI64;
use crate::config;
use crate::helpers::errors::PrintFullError;
use crate::storage::{get_redis_conn, redis_key};
use chirpstack_api::api;
@ -279,12 +280,12 @@ pub async fn get_frame_logs(
let mut phy = lrwn::PhyPayload::from_slice(&pl.phy_payload)?;
if pl.plaintext_f_opts {
if let Err(e) = phy.decode_f_opts_to_mac_commands() {
warn!(error = %e, "Decode f_opts to mac-commands error");
warn!(error = %e.full(), "Decode f_opts to mac-commands error");
}
}
if pl.plaintext_frm_payload {
if let Err(e) = phy.decode_frm_payload() {
warn!(error = %e, "Decode frm_payload error");
warn!(error = %e.full(), "Decode frm_payload error");
}
}
@ -317,12 +318,12 @@ pub async fn get_frame_logs(
let mut phy = lrwn::PhyPayload::from_slice(&pl.phy_payload)?;
if pl.plaintext_f_opts {
if let Err(e) = phy.decode_f_opts_to_mac_commands() {
warn!(error = %e, "Decode f_opts to mac-commands error");
warn!(error = %e.full(), "Decode f_opts to mac-commands error");
}
}
if pl.plaintext_frm_payload {
if let Err(e) = phy.decode_frm_payload() {
warn!(error = %e, "Decode frm_payload error");
warn!(error = %e.full(), "Decode frm_payload error");
}
}
@ -363,7 +364,7 @@ pub async fn get_frame_logs(
return Err(e);
}
error!(key = %k, error = %e, "Parsing frame-log error");
error!(key = %k, error = %e.full(), "Parsing frame-log error");
}
}
}

3
chirpstack/src/gateway/backend/mod.rs
View File

@ -3,7 +3,7 @@ use std::collections::HashMap;
use anyhow::{Context, Result};
use async_trait::async_trait;
use tokio::sync::RwLock;
use tracing::{info, warn};
use tracing::info;
use crate::config;
@ -31,7 +31,6 @@ pub async fn setup() -> Result<()> {
info!("Setting up gateway backends for the different regions");
for region in &conf.regions {
if !conf.network.enabled_regions.contains(&region.id) {
warn!("Config exists, but region is not enabled. To enable it, add '{}' to 'network.enabled_regions'", region.id);
continue;
}

36
chirpstack/src/gateway/backend/mqtt.rs
View File

@ -138,7 +138,7 @@ impl<'a> MqttBackend<'a> {
info!(region_config_id = %ctx.region_config_id, "Connected to MQTT broker");
if let Err(e) = subscribe_tx.try_send(()) {
error!(region_config_id = %ctx.region_config_id, error = %e, "Send to subscribe channel error");
error!(region_id = %ctx.region_config_id, error = %e, "Send to subscribe channel error");
}
});
client.set_connection_lost_callback(|client| {
@ -148,7 +148,7 @@ impl<'a> MqttBackend<'a> {
.downcast_ref::<MqttContext>()
.unwrap();
info!(region_config_id = %ctx.region_config_id, "MQTT connection to broker lost");
info!(region_id = %ctx.region_config_id, "MQTT connection to broker lost");
});
// connection options
@ -164,6 +164,7 @@ impl<'a> MqttBackend<'a> {
}
if !conf.ca_cert.is_empty() || !conf.tls_cert.is_empty() || !conf.tls_key.is_empty() {
info!(
region_id = %region_config_id,
ca_cert = conf.ca_cert.as_str(),
tls_cert = conf.tls_cert.as_str(),
tls_key = conf.tls_key.as_str(),
@ -205,7 +206,7 @@ impl<'a> MqttBackend<'a> {
};
// connect
info!(region_config_id = %region_config_id, server_uri = %conf.server, clean_session = conf.clean_session, client_id = %client_id, "Connecting to MQTT broker");
info!(region_id = %region_config_id, server_uri = %conf.server, clean_session = conf.clean_session, client_id = %client_id, "Connecting to MQTT broker");
b.client
.connect(conn_opts)
.await
@ -217,7 +218,7 @@ impl<'a> MqttBackend<'a> {
let v4_migrate = conf.v4_migrate;
async move {
info!("Starting MQTT consumer loop");
info!(region_id = %region_config_id, "Starting MQTT consumer loop");
while let Some(msg_opt) = stream.next().await {
if let Some(msg) = msg_opt {
message_callback(v4_migrate, &region_config_id, region_common_name, msg)
@ -245,9 +246,9 @@ impl<'a> MqttBackend<'a> {
async move {
while subscribe_rx.recv().await.is_some() {
info!(region_config_id = %region_config_id, event_topic = %event_topic, "Subscribing to gateway event topic");
info!(region_id = %region_config_id, event_topic = %event_topic, "Subscribing to gateway event topic");
if let Err(e) = client.subscribe(&event_topic, qos).await {
error!(region_config_id = %region_config_id, event_topic = %event_topic, error = %e, "MQTT subscribe error");
error!(region_id = %region_config_id, event_topic = %event_topic, error = %e, "MQTT subscribe error");
}
}
}
@ -271,6 +272,13 @@ impl<'a> MqttBackend<'a> {
#[async_trait]
impl GatewayBackend for MqttBackend<'_> {
async fn send_downlink(&self, df: &chirpstack_api::gw::DownlinkFrame) -> Result<()> {
let ctx = self
.client
.user_data()
.unwrap()
.downcast_ref::<MqttContext>()
.unwrap();
COMMAND_COUNTER
.get_or_create(&CommandLabels {
command: "down".to_string(),
@ -289,7 +297,7 @@ impl GatewayBackend for MqttBackend<'_> {
false => df.encode_to_vec(),
};
info!(gateway_id = %df.gateway_id, topic = %topic, json = json, "Sending downlink frame");
info!(region_id = %ctx.region_config_id, gateway_id = %df.gateway_id, topic = %topic, json = json, "Sending downlink frame");
let msg = mqtt::Message::new(topic, b, self.qos as i32);
self.client.publish(msg).await?;
trace!("Message sent");
@ -301,6 +309,13 @@ impl GatewayBackend for MqttBackend<'_> {
&self,
gw_conf: &chirpstack_api::gw::GatewayConfiguration,
) -> Result<()> {
let ctx = self
.client
.user_data()
.unwrap()
.downcast_ref::<MqttContext>()
.unwrap();
COMMAND_COUNTER
.get_or_create(&CommandLabels {
command: "config".to_string(),
@ -313,7 +328,7 @@ impl GatewayBackend for MqttBackend<'_> {
false => gw_conf.encode_to_vec(),
};
info!(gateway_id = %gw_conf.gateway_id, topic = %topic, json = json, "Sending gateway configuration");
info!(region_id = %ctx.region_config_id, gateway_id = %gw_conf.gateway_id, topic = %topic, json = json, "Sending gateway configuration");
let msg = mqtt::Message::new(topic, b, self.qos as i32);
self.client.publish(msg).await?;
trace!("Message sent");
@ -340,7 +355,7 @@ async fn message_callback(
let err = || -> Result<()> {
if locked? {
trace!(
region_config_id = region_config_id,
region_id = region_config_id,
topic = topic,
qos = qos,
"Message is already handled by different instance"
@ -351,7 +366,7 @@ async fn message_callback(
let json = payload_is_json(b);
info!(
region_config_id = region_config_id,
region_id = region_config_id,
topic = topic,
qos = qos,
json = json,
@ -437,6 +452,7 @@ async fn message_callback(
if err.is_some() {
error!(
region_id = %region_config_id,
topic = topic,
qos = qos,
"Processing gateway event error: {}",

27
chirpstack/src/helpers/errors.rs Normal file
View File

@ -0,0 +1,27 @@
pub trait PrintFullError {
fn full(&self) -> String;
}
impl PrintFullError for anyhow::Error {
fn full(&self) -> String {
format!("{:#}", self)
}
}
impl PrintFullError for crate::storage::error::Error {
fn full(&self) -> String {
format!("{:#}", self)
}
}
impl PrintFullError for crate::api::auth::error::Error {
fn full(&self) -> String {
format!("{:#}", self)
}
}
impl PrintFullError for lrwn::Error {
fn full(&self) -> String {
format!("{:#}", self)
}
}

1
chirpstack/src/helpers/mod.rs Normal file
View File

@ -0,0 +1 @@
pub mod errors;

7
chirpstack/src/integration/http.rs
View File

@ -6,7 +6,7 @@ use async_trait::async_trait;
use prost::Message;
use reqwest::header::{HeaderMap, HeaderName, CONTENT_TYPE};
use reqwest::Client;
use tracing::{error, info, trace};
use tracing::{info, trace, warn};
use super::Integration as IntegrationTrait;
use crate::storage::application::HttpConfiguration;
@ -59,15 +59,16 @@ impl Integration {
.send()
.await;
// We log the errors as warn as these endpoints are user-defined.
match res {
Ok(res) => match res.error_for_status() {
Ok(_) => {}
Err(e) => {
error!(event = %event, url = %url, error = %e, "Posting event failed");
warn!(event = %event, url = %url, error = %e, "Posting event failed");
}
},
Err(e) => {
error!(event = %event, url = %url, error = %e, "Posting event failed");
warn!(event = %event, url = %url, error = %e, "Posting event failed");
}
}
}

4
chirpstack/src/integration/ifttt.rs
View File

@ -6,7 +6,7 @@ use async_trait::async_trait;
use reqwest::header::{HeaderMap, CONTENT_TYPE};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use tracing::{error, info, trace};
use tracing::{info, trace, warn};
use super::Integration as IntegrationTrait;
use crate::codec;
@ -72,7 +72,7 @@ impl Integration {
match res.error_for_status() {
Ok(_) => Ok(()),
Err(e) => {
error!(event = %event, error = %e, "Sending event to IFTTT failed");
warn!(event = %event, error = %e, "Sending event to IFTTT failed");
Err(anyhow::Error::new(e))
}
}

9
chirpstack/src/integration/loracloud/mod.rs
View File

@ -10,6 +10,7 @@ use uuid::Uuid;
use super::Integration as IntegrationTrait;
use crate::gpstime::ToGpsTime;
use crate::helpers::errors::PrintFullError;
use crate::integration::{integration_event, location_event};
use crate::storage::application::LoraCloudConfiguration;
use crate::storage::device_queue;
@ -561,7 +562,7 @@ impl Integration {
match resp {
Ok(v) => Ok(Some(v)),
Err(e) => {
warn!(error = %e, "GNSS geolocation failed");
warn!(error = %e.full(), "GNSS geolocation failed");
Ok(None)
}
}
@ -643,7 +644,7 @@ impl Integration {
match resp {
Ok(v) => Ok(Some(v)),
Err(e) => {
warn!(error = %e, "Wifi geolocation failed");
warn!(error = %e.full(), "Wifi geolocation failed");
Ok(None)
}
}
@ -666,7 +667,7 @@ impl Integration {
match resp {
Ok(v) => Ok(Some(v)),
Err(e) => {
warn!(error = %e, "TDOA geolocation failed");
warn!(error = %e.full(), "TDOA geolocation failed");
Ok(None)
}
}
@ -689,7 +690,7 @@ impl Integration {
match resp {
Ok(v) => Ok(Some(v)),
Err(e) => {
warn!(error = %e, "RSSI geolocation failed");
warn!(error = %e.full(), "RSSI geolocation failed");
Ok(None)
}
}

21
chirpstack/src/integration/mod.rs
View File

@ -5,9 +5,10 @@ use anyhow::{Context, Result};
use async_trait::async_trait;
use futures::future::join_all;
use tokio::sync::RwLock;
use tracing::{error, info};
use tracing::{info, warn};
use uuid::Uuid;
use crate::helpers::errors::PrintFullError;
use crate::storage::{application, device, device_profile, device_queue};
use crate::{codec, config};
use chirpstack_api::integration;
@ -197,7 +198,7 @@ pub async fn uplink_event(
async move {
if let Err(err) = _uplink_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Uplink event error");
warn!(application_id = %application_id, error = %err.full(), "Uplink event error");
}
}
});
@ -239,7 +240,7 @@ pub async fn join_event(
async move {
if let Err(err) = _join_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Join event error");
warn!(application_id = %application_id, error = %err.full(), "Join event error");
}
}
});
@ -281,7 +282,7 @@ pub async fn ack_event(
async move {
if let Err(err) = _ack_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Ack event error");
warn!(application_id = %application_id, error = %err.full(), "Ack event error");
}
}
});
@ -323,7 +324,7 @@ pub async fn txack_event(
async move {
if let Err(err) = _txack_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Txack event error");
warn!(application_id = %application_id, error = %err.full(), "Txack event error");
}
}
});
@ -365,7 +366,7 @@ pub async fn log_event(
async move {
if let Err(err) = _log_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Log event error");
warn!(application_id = %application_id, error = %err.full(), "Log event error");
}
}
});
@ -407,7 +408,7 @@ pub async fn status_event(
async move {
if let Err(err) = _status_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Status event error");
warn!(application_id = %application_id, error = %err.full(), "Status event error");
}
}
});
@ -449,7 +450,7 @@ pub async fn location_event(
async move {
if let Err(err) = _location_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Location event error");
warn!(application_id = %application_id, error = %err.full(), "Location event error");
}
}
});
@ -491,7 +492,7 @@ pub async fn integration_event(
async move {
if let Err(err) = _integration_event(application_id, &vars, &pl).await {
error!(application_id = %application_id, error = %err, "Location event error");
warn!(application_id = %application_id, error = %err.full(), "Location event error");
}
}
});
@ -571,6 +572,6 @@ async fn handle_down_command(application_id: String, pl: integration::DownlinkCo
.err();
if err.is_some() {
error!(dev_eui = %pl.dev_eui, error = %err.as_ref().unwrap(), "Handling downlink command error");
warn!(dev_eui = %pl.dev_eui, error = %err.as_ref().unwrap().full(), "Handling downlink command error");
}
}

4
chirpstack/src/maccommand/device_time.rs
View File

@ -2,7 +2,7 @@ use std::time::Duration;
use anyhow::Result;
use chrono::{DateTime, Utc};
use tracing::{error, info};
use tracing::{info, warn};
use crate::gpstime::ToGpsTime;
use crate::storage::device;
@ -27,7 +27,7 @@ pub fn handle(
time_since_gps_epoch: match gps_time.to_std() {
Ok(v) => v,
Err(e) => {
error!(error = %e, "To GPS time error");
warn!(error = %e, "To GPS time error");
Duration::from_secs(0)
}
},

4
chirpstack/src/maccommand/end_device_conf.rs
View File

@ -1,5 +1,5 @@
use anyhow::Result;
use tracing::{error, info};
use tracing::{info, warn};
use crate::storage::device;
use chirpstack_api::internal;
@ -56,7 +56,7 @@ pub fn handle(
relay.second_channel_freq = req_pl.second_ch_freq;
}
} else {
error!(
warn!(
dev_eui = %dev.dev_eui,
second_ch_freq_ack = ans_pl.second_ch_freq_ack,
second_ch_dr_ack = ans_pl.second_ch_dr_ack,

4
chirpstack/src/maccommand/filter_list.rs
View File

@ -1,5 +1,5 @@
use anyhow::Result;
use tracing::{error, info};
use tracing::{info, warn};
use crate::storage::device;
use chirpstack_api::internal;
@ -53,7 +53,7 @@ pub fn handle(
}
}
} else {
error!(
warn!(
dev_eui = %dev.dev_eui,
filter_list_action_ack = ans_pl.filter_list_action_ack,
filter_list_len_ack = ans_pl.filter_list_len_ack,

7
chirpstack/src/maccommand/mod.rs
View File

@ -1,9 +1,10 @@
use std::collections::HashMap;
use anyhow::Result;
use tracing::error;
use tracing::{error, warn};
use crate::config;
use crate::helpers::errors::PrintFullError;
use crate::storage::{application, device, device_profile, mac_command, tenant};
use crate::uplink::UplinkFrameSet;
use chirpstack_api::internal;
@ -112,7 +113,7 @@ pub async fn handle_uplink<'a>(
{
Ok(v) => v,
Err(e) => {
error!(dev_eui = %dev_eui, cid = %cid, error = %e, "Handle mac-command error");
warn!(dev_eui = %dev_eui, cid = %cid, error = %e.full(), "Handle mac-command error");
continue;
}
};
@ -168,7 +169,7 @@ async fn handle(
ctrl_uplink_list::handle(dev, ds, block, pending_block).await
}
_ => {
error!(cid = %cid, "Unexpected CID");
warn!(cid = %cid, "Unexpected CID");
// Return OK, we don't want to break out of the uplink handling.
Ok(None)
}

12
chirpstack/src/maccommand/new_channel.rs
View File

@ -1,7 +1,7 @@
use std::collections::HashMap;
use anyhow::Result;
use tracing::{error, info};
use tracing::{info, warn};
use crate::storage::device;
use chirpstack_api::internal;
@ -123,7 +123,15 @@ pub fn handle(
.or_insert(0);
*count += 1;
error!(dev_eui = %dev.dev_eui, freq = req_pl.freq, channel = req_pl.ch_index, min_dr = req_pl.min_dr, max_dr = req_pl.max_dr, dr_range_ok = ans_pl.dr_range_ok, channel_freq_ok = ans_pl.channel_freq_ok, "NewChannelReq not acknowledged");
warn!(
dev_eui = %dev.dev_eui,
freq = req_pl.freq,
channel = req_pl.ch_index,
min_dr = req_pl.min_dr,
max_dr = req_pl.max_dr,
dr_range_ok = ans_pl.dr_range_ok,
channel_freq_ok = ans_pl.channel_freq_ok,
"NewChannelReq not acknowledged");
}
}

4
chirpstack/src/maccommand/relay_conf.rs
View File

@ -1,5 +1,5 @@
use anyhow::Result;
use tracing::{error, info};
use tracing::{info, warn};
use crate::storage::device;
use chirpstack_api::internal;
@ -55,7 +55,7 @@ pub fn handle(
req_pl.channel_settings_relay.second_ch_ack_offset as u32;
}
} else {
error!(
warn!(
dev_eui = %dev.dev_eui,
second_ch_ack_offset_ack = ans_pl.second_ch_ack_offset_ack,
second_ch_dr_ack = ans_pl.second_ch_dr_ack,

1
chirpstack/src/main.rs
View File

@ -34,6 +34,7 @@ mod eventlog;
mod framelog;
mod gateway;
mod gpstime;
mod helpers;
mod integration;
mod maccommand;
mod metalog;

3
chirpstack/src/region.rs
View File

@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use anyhow::{Context, Result};
use tracing::{info, span, trace, warn, Level};
use tracing::{info, span, trace, Level};
use crate::config;
use lrwn::region;
@ -23,7 +23,6 @@ pub fn setup() -> Result<()> {
let _guard = span.enter();
if !conf.network.enabled_regions.contains(&r.id) {
warn!("Config exists, but region is not enabled. To enable it, add '{}' to 'network.enabled_regions'", r.id);
continue;
}

9
chirpstack/src/storage/device_session.rs
View File

@ -4,12 +4,13 @@ use std::io::Cursor;
use anyhow::{Context, Result};
use prost::Message;
use tokio::task;
use tracing::{error, info, trace, warn};
use tracing::{error, info, trace};
use super::error::Error;
use super::{get_redis_conn, redis_key};
use crate::api::helpers::FromProto;
use crate::config;
use crate::helpers::errors::PrintFullError;
use chirpstack_api::internal;
use lrwn::{AES128Key, DevAddr, Payload, PhyPayload, EUI64};
@ -191,7 +192,7 @@ pub async fn get_for_phypayload_and_incr_f_cnt_up(
if let Some(relay) = &ds.relay {
if !relayed && relay.ed_relay_only {
warn!(
info!(
dev_eui = hex::encode(ds.dev_eui),
"Only communication through relay is allowed"
);
@ -357,10 +358,10 @@ async fn get_for_dev_addr(dev_addr: DevAddr) -> Result<Vec<internal::DeviceSessi
Err(e) => {
if let Error::NotFound(_) = e {
if let Err(e) = remove_dev_eui_from_dev_addr_set(dev_addr, dev_eui).await {
error!(dev_addr = %dev_addr, dev_eui = %dev_eui, error = %e, "Remove DevEUI from DevAddr->DevEUI set error");
error!(dev_addr = %dev_addr, dev_eui = %dev_eui, error = %e.full(), "Remove DevEUI from DevAddr->DevEUI set error");
}
} else {
error!(dev_addr = %dev_addr, dev_eui = %dev_eui, error = %e, "Get device-session for DevEUI error");
error!(dev_addr = %dev_addr, dev_eui = %dev_eui, error = %e.full(), "Get device-session for DevEUI error");
}
continue;
}

29
chirpstack/src/uplink/data.rs
View File

@ -12,6 +12,7 @@ use super::{
};
use crate::api::helpers::ToProto;
use crate::backend::roaming;
use crate::helpers::errors::PrintFullError;
use crate::storage::error::Error as StorageError;
use crate::storage::{
application,
@ -51,7 +52,7 @@ pub struct Data {
impl Data {
pub async fn handle(ufs: UplinkFrameSet) {
let span = span!(Level::INFO, "data_up");
let span = span!(Level::INFO, "data_up", dev_eui = tracing::field::Empty);
if let Err(e) = Data::_handle(ufs).instrument(span).await {
match e.downcast_ref::<Error>() {
@ -59,7 +60,7 @@ impl Data {
// nothing to do
}
Some(_) | None => {
error!(error = %e, "Handle uplink error");
error!(error = %e.full(), "Handle uplink error");
}
}
}
@ -81,7 +82,7 @@ impl Data {
// nothing to do
}
Some(_) | None => {
error!(error = %e, "Handle relayed uplink error");
error!(error = %e.full(), "Handle relayed uplink error");
}
}
}
@ -110,7 +111,13 @@ impl Data {
ctx.handle_passive_roaming_device().await?;
ctx.get_device_session().await?;
ctx.get_device().await?;
// Add dev_eui to span
let span = tracing::Span::current();
span.record("dev_eui", ctx.device.as_ref().unwrap().dev_eui.to_string());
ctx.get_device_profile().await?;
ctx.get_application().await?;
ctx.get_tenant().await?;
@ -262,11 +269,11 @@ impl Data {
},
Err(e) => match e {
StorageError::NotFound(s) => {
warn!(dev_addr = %s, "No device-session exists for dev_addr");
info!(dev_addr = %s, "No device-session exists for dev_addr");
return Err(Error::Abort);
}
StorageError::InvalidMIC => {
warn!(dev_addr = %dev_addr, "None of the device-sessions for dev_addr resulted in valid MIC");
info!(dev_addr = %dev_addr, "None of the device-sessions for dev_addr resulted in valid MIC");
// Log uplink for null DevEUI.
let mut ufl: api::UplinkFrameLog = (&self.uplink_frame_set).try_into()?;
@ -330,11 +337,11 @@ impl Data {
},
Err(e) => match e {
StorageError::NotFound(s) => {
warn!(dev_addr = %s, "No device-session exists for dev_addr");
info!(dev_addr = %s, "No device-session exists for dev_addr");
return Err(Error::Abort);
}
StorageError::InvalidMIC => {
warn!(dev_addr = %dev_addr, "None of the device-sessions for dev_addr resulted in valid MIC");
info!(dev_addr = %dev_addr, "None of the device-sessions for dev_addr resulted in valid MIC");
return Err(Error::Abort);
}
_ => {
@ -600,13 +607,13 @@ impl Data {
if ds.mac_version().to_string().starts_with("1.0") {
if let Err(e) = self.phy_payload.decode_f_opts_to_mac_commands() {
// This avoids failing in case of a corrupted mac-command in the frm_payload.
warn!(error = %e, "Decoding f_opts mac-commands failed");
warn!(error = %e.full(), "Decoding f_opts mac-commands failed");
}
} else {
let nwk_s_enc_key = AES128Key::from_slice(&ds.nwk_s_enc_key)?;
if let Err(e) = self.phy_payload.decrypt_f_opts(&nwk_s_enc_key) {
// This avoids failing in case of a corrupted mac-command in the frm_payload.
warn!(error = %e, "Decrypting f_opts mac-commands failed");
warn!(error = %e.full(), "Decrypting f_opts mac-commands failed");
}
}
@ -627,7 +634,7 @@ impl Data {
let nwk_s_enc_key = AES128Key::from_slice(&ds.nwk_s_enc_key)?;
if let Err(e) = self.phy_payload.decrypt_frm_payload(&nwk_s_enc_key) {
// This avoids failing in case of a corrupted mac-command in the frm_payload.
warn!(error = %e, "Decrypting frm_payload failed");
warn!(error = %e.full(), "Decrypting frm_payload failed");
}
} else if !self._is_end_to_end_encrypted() {
if let Some(app_s_key) = &ds.app_s_key {
@ -1136,7 +1143,7 @@ impl Data {
let qi = match device_queue::get_pending_for_dev_eui(&dev.dev_eui).await {
Ok(v) => v,
Err(e) => {
warn!(dev_eui = %dev.dev_eui, error = %e, "Get pending queue-item error");
warn!(dev_eui = %dev.dev_eui, error = %e.full(), "Get pending queue-item error");
return Ok(());
}
};

7
chirpstack/src/uplink/data_fns.rs
View File

@ -6,6 +6,7 @@ use uuid::Uuid;
use super::{error::Error, filter_rx_info_by_public_only, UplinkFrameSet};
use crate::api::backend::get_async_receiver;
use crate::backend::{keywrap, roaming};
use crate::helpers::errors::PrintFullError;
use crate::storage::passive_roaming;
use crate::uplink::helpers;
use chirpstack_api::internal;
@ -26,7 +27,7 @@ impl Data {
// nothing to do
}
Some(_) | None => {
error!(error = %e, "Handle passive-roaming uplink error");
error!(error = %e.full(), "Handle passive-roaming uplink error");
}
}
}
@ -85,7 +86,7 @@ impl Data {
let ds = match self.start_pr_session(net_id).await {
Ok(v) => v,
Err(e) => {
error!(net_id = %net_id, error = %e, "Start passive-roaming error");
error!(net_id = %net_id, error = %e.full(), "Start passive-roaming error");
continue;
}
};
@ -136,7 +137,7 @@ impl Data {
.xmit_data_req(backend::Role::SNS, &mut req, async_receiver)
.await
{
error!(net_id = %net_id, error = %e, "XmitDataReq failed");
error!(net_id = %net_id, error = %e.full(), "XmitDataReq failed");
}
}

17
chirpstack/src/uplink/join.rs
View File

@ -19,6 +19,7 @@ use super::{
use crate::api::{backend::get_async_receiver, helpers::ToProto};
use crate::backend::{joinserver, keywrap, roaming};
use crate::helpers::errors::PrintFullError;
use crate::storage::device_session;
use crate::storage::{
application,
@ -57,7 +58,7 @@ pub struct JoinRequest {
impl JoinRequest {
pub async fn handle(ufs: UplinkFrameSet) {
let span = span!(Level::INFO, "join_request");
let span = span!(Level::INFO, "join_request", dev_eui = tracing::field::Empty);
if let Err(e) = JoinRequest::_handle(ufs).instrument(span).await {
match e.downcast_ref::<Error>() {
@ -65,7 +66,7 @@ impl JoinRequest {
// nothing to do
}
Some(_) | None => {
error!(error = %e, "Handle join-request error");
error!(error = %e.full(), "Handle join-request error");
}
}
}
@ -83,7 +84,7 @@ impl JoinRequest {
// nothing to do
}
Some(_) | None => {
error!(error = %e, "Handle relayed join-request error");
error!(error = %e.full(), "Handle relayed join-request error");
}
}
}
@ -113,6 +114,14 @@ impl JoinRequest {
};
ctx.get_join_request_payload()?;
// Add resolved DevEUI to the span
let span = tracing::Span::current();
span.record(
"dev_eui",
ctx.join_request.as_ref().unwrap().dev_eui.to_string(),
);
ctx.get_device_or_try_pr_roaming().await?;
ctx.get_device_keys_or_js_client().await?; // used to validate MIC + if we need external JS
ctx.get_application().await?;
@ -397,7 +406,7 @@ impl JoinRequest {
// In case the relay context is not set and relay_ed_relay_only is set, abort.
if self.relay_context.is_none() && self.device_profile.as_ref().unwrap().relay_ed_relay_only
{
warn!(dev_eui = %self.device.as_ref().unwrap().dev_eui, "Only communication through relay is allowed");
info!(dev_eui = %self.device.as_ref().unwrap().dev_eui, "Only communication through relay is allowed");
return Err(Error::Abort);
}
Ok(())

39
chirpstack/src/uplink/mod.rs
View File

@ -13,6 +13,7 @@ use uuid::Uuid;
use crate::config;
use crate::framelog;
use crate::helpers::errors::PrintFullError;
use crate::storage::{
device, device_profile, error::Error as StorageError, gateway, get_redis_conn, redis_key,
};
@ -116,7 +117,7 @@ pub struct RoamingMetaData {
pub async fn deduplicate_uplink(event: gw::UplinkFrame) {
if let Err(e) = _deduplicate_uplink(event).await {
error!(error = %e, "Deduplication error");
error!(error = %e.full(), "Deduplication error");
}
}
@ -189,7 +190,8 @@ async fn deduplicate_put(key: &str, ttl: Duration, event: &gw::UplinkFrame) -> R
.arg(&key)
.arg(ttl.as_millis() as usize)
.ignore()
.query(&mut c)?;
.query(&mut c)
.context("Deduplication put")?;
Ok(())
}
@ -209,7 +211,8 @@ async fn deduplicate_locked(key: &str, ttl: Duration) -> Result<bool> {
.arg("PX")
.arg(ttl.as_millis() as usize)
.arg("NX")
.query(&mut *c)?;
.query(&mut *c)
.context("Deduplication locked")?;
Ok(!set)
}
@ -222,7 +225,10 @@ async fn deduplicate_collect(key: &str) -> Result<gw::UplinkFrameSet> {
let key = key.to_string();
move || -> Result<gw::UplinkFrameSet> {
let mut c = get_redis_conn()?;
let items_b: Vec<Vec<u8>> = redis::cmd("SMEMBERS").arg(&key).query(&mut *c)?;
let items_b: Vec<Vec<u8>> = redis::cmd("SMEMBERS")
.arg(&key)
.query(&mut *c)
.context("Deduplication collect")?;
if items_b.is_empty() {
return Err(anyhow!("Zero items in collect set"));
@ -233,7 +239,8 @@ async fn deduplicate_collect(key: &str) -> Result<gw::UplinkFrameSet> {
};
for b in items_b {
let event = gw::UplinkFrame::decode(&mut Cursor::new(b))?;
let event =
gw::UplinkFrame::decode(&mut Cursor::new(b)).context("Decode UplinkFrame")?;
if event.tx_info.is_none() {
warn!("tx_info of uplink event is empty, skipping");
@ -298,18 +305,20 @@ pub async fn handle_uplink(deduplication_id: Uuid, uplink: gw::UplinkFrameSet) -
)?;
info!(
m_type = uplink.phy_payload.mhdr.m_type.to_string().as_str(),
m_type = %uplink.phy_payload.mhdr.m_type,
"Uplink received"
);
debug!("Updating gateway meta-data for uplink frame-set");
update_gateway_metadata(&mut uplink).await?;
update_gateway_metadata(&mut uplink)
.await
.context("Update gateway meta-data")?;
debug!("Logging uplink frame to Redis Stream");
let ufl: api::UplinkFrameLog = (&uplink).try_into()?;
framelog::log_uplink_for_gateways(&ufl)
.await
.context("log_uplink_for_gateways error")?;
.context("Log uplink for gateways")?;
match uplink.phy_payload.mhdr.m_type {
MType::JoinRequest => join::JoinRequest::handle(uplink).await,
@ -328,7 +337,7 @@ pub async fn handle_uplink(deduplication_id: Uuid, uplink: gw::UplinkFrameSet) -
async fn update_gateway_metadata(ufs: &mut UplinkFrameSet) -> Result<()> {
let conf = config::get();
for rx_info in &mut ufs.rx_info_set {
let gw_id = EUI64::from_str(&rx_info.gateway_id)?;
let gw_id = EUI64::from_str(&rx_info.gateway_id).context("Gateway ID")?;
let gw_meta = match gateway::get_meta(&gw_id).await {
Ok(v) => v,
Err(e) => {
@ -341,8 +350,8 @@ async fn update_gateway_metadata(ufs: &mut UplinkFrameSet) -> Result<()> {
}
error!(
gateway_id = gw_id.to_string().as_str(),
error = format!("{}", e).as_str(),
gateway_id = %gw_id,
error = %e.full(),
"Getting gateway meta-data failed"
);
continue;
@ -375,7 +384,7 @@ fn filter_rx_info_by_tenant_id(tenant_id: Uuid, uplink: &mut UplinkFrameSet) ->
let mut rx_info_set: Vec<gw::UplinkRxInfo> = Vec::new();
for rx_info in &uplink.rx_info_set {
let gateway_id = EUI64::from_str(&rx_info.gateway_id)?;
let gateway_id = EUI64::from_str(&rx_info.gateway_id).context("Gateway ID")?;
let region_config_id = rx_info
.metadata
.get("region_config_id")
@ -412,7 +421,7 @@ fn filter_rx_info_by_public_only(uplink: &mut UplinkFrameSet) -> Result<()> {
let mut rx_info_set: Vec<gw::UplinkRxInfo> = Vec::new();
for rx_info in &uplink.rx_info_set {
let gateway_id = EUI64::from_str(&rx_info.gateway_id)?;
let gateway_id = EUI64::from_str(&rx_info.gateway_id).context("Gateway ID")?;
if !(*uplink
.gateway_private_up_map
.get(&gateway_id)
@ -424,7 +433,7 @@ fn filter_rx_info_by_public_only(uplink: &mut UplinkFrameSet) -> Result<()> {
uplink.rx_info_set = rx_info_set;
if uplink.rx_info_set.is_empty() {
return Err(anyhow!("rx_info_set has no items"));
return Err(anyhow!("rx_info_set is empty"));
}
Ok(())
@ -446,7 +455,7 @@ fn filter_rx_info_by_region_config_id(
uplink.rx_info_set = rx_info_set;
if uplink.rx_info_set.is_empty() {
return Err(anyhow!("rx_info_set has no items"));
return Err(anyhow!("rx_info_set is empty"));
}
Ok(())

18
chirpstack/src/uplink/stats.rs
View File

@ -5,9 +5,10 @@ use std::str::FromStr;
use anyhow::{Context, Result};
use chrono::{DateTime, Local};
use tracing::{error, info, span, trace, Instrument, Level};
use tracing::{error, info, span, trace, warn, Instrument, Level};
use crate::gateway::backend as gateway_backend;
use crate::helpers::errors::PrintFullError;
use crate::storage::{error::Error, gateway, metrics};
use crate::{config, region};
use chirpstack_api::{common, gw};
@ -22,13 +23,13 @@ pub struct Stats {
impl Stats {
pub async fn handle(s: gw::GatewayStats) {
let gateway_id = match if !s.gateway_id.is_empty() {
EUI64::from_str(&s.gateway_id)
EUI64::from_str(&s.gateway_id).context("Gateway ID")
} else {
EUI64::from_slice(&s.gateway_id_legacy)
EUI64::from_slice(&s.gateway_id_legacy).context("Legacy Gateway ID")
} {
Ok(v) => v,
Err(e) => {
error!(error = %e, "Decode stats gateway_id error");
warn!(error = %e.full(), "Decode stats gateway_id error");
return;
}
};
@ -43,11 +44,11 @@ impl Stats {
// the database.
let conf = config::get();
if !conf.gateway.allow_unknown_gateways {
error!(error = %e, "Handle gateway stats error");
error!(error = %e.full(), "Handle gateway stats error");
}
}
Some(_) | None => {
error!(error = %e, "Handle gateway stats error");
error!(error = %e.full(), "Handle gateway stats error");
}
}
}
@ -80,7 +81,7 @@ impl Stats {
&self.stats.metadata,
)
.await
.context("Update gateway state")?,
.context("Update gateway state and location")?,
);
} else {
self.gateway = Some(
@ -155,7 +156,8 @@ impl Stats {
&format!("gw:{}", self.gateway.as_ref().unwrap().gateway_id),
&m,
)
.await?;
.await
.context("Save gateway stats")?;
Ok(())
}