From 27f6d2cf034b4229b700aa0c0ad614222486a416 Mon Sep 17 00:00:00 2001
From: Orne Brocaar <info@brocaar.com>
Date: Wed, 5 Mar 2025 14:10:40 +0000
Subject: [PATCH] Implement full FUOTA flow + UI components.
---
api/proto/api/fuota.proto | 43 +-
api/rust/proto/chirpstack/api/fuota.proto | 43 +-
.../up.sql | 6 +-
.../up.sql | 6 +-
chirpstack/src/api/device_profile.rs | 9 +-
chirpstack/src/api/fuota.rs | 99 +++-
chirpstack/src/applayer/clocksync.rs | 2 +-
chirpstack/src/applayer/fragmentation.rs | 96 +++
chirpstack/src/applayer/fuota/flow.rs | 548 ++++++++++++++++++
chirpstack/src/applayer/fuota/mod.rs | 9 +
chirpstack/src/applayer/fuota/scheduler.rs | 45 ++
chirpstack/src/applayer/mod.rs | 13 +-
chirpstack/src/applayer/multicastsetup.rs | 139 +++++
chirpstack/src/cmd/root.rs | 3 +-
chirpstack/src/storage/fields/fuota.rs | 3 +
chirpstack/src/storage/fields/mod.rs | 5 +-
chirpstack/src/storage/fuota.rs | 128 +++-
chirpstack/src/storage/schema_postgres.rs | 6 +-
chirpstack/src/storage/schema_sqlite.rs | 6 +-
ui/src/stores/FuotaStore.ts | 37 +-
.../device-profiles/CreateDeviceProfile.tsx | 6 +-
.../device-profiles/DeviceProfileForm.tsx | 15 +-
ui/src/views/devices/ListDevices.tsx | 2 +-
.../views/fuota/FuotaDeploymentDashboard.tsx | 128 ++++
ui/src/views/fuota/FuotaDeploymentDevices.tsx | 91 ++-
ui/src/views/fuota/FuotaDeploymentForm.tsx | 22 +-
.../views/fuota/FuotaDeploymentGateways.tsx | 4 +-
ui/src/views/fuota/FuotaDeploymentLayout.tsx | 10 +-
ui/src/views/gateways/ListGateways.tsx | 6 +-
ui/src/views/helpers.ts | 22 +
30 files changed, 1451 insertions(+), 101 deletions(-)
create mode 100644 chirpstack/src/applayer/fragmentation.rs
create mode 100644 chirpstack/src/applayer/fuota/flow.rs
create mode 100644 chirpstack/src/applayer/fuota/mod.rs
create mode 100644 chirpstack/src/applayer/fuota/scheduler.rs
create mode 100644 chirpstack/src/applayer/multicastsetup.rs
create mode 100644 ui/src/views/fuota/FuotaDeploymentDashboard.tsx
diff --git a/api/proto/api/fuota.proto b/api/proto/api/fuota.proto
index ead9b854..4cfd546e 100644
--- a/api/proto/api/fuota.proto
+++ b/api/proto/api/fuota.proto
@@ -57,6 +57,9 @@ service FuotaService {
// Remove the given Gateway IDs from the FUOTA deployment.
rpc RemoveGateways(RemoveGatewaysFromFuotaDeploymentRequest) returns (google.protobuf.Empty) {}
// GetLogs returns the logs for the FUOTA deployment.
+
+ // List jobs for the given FUOTA deployment.
+ rpc ListJobs(ListFuotaDeploymentJobsRequest) returns (ListFuotaDeploymentJobsResponse) {}
}
enum RequestFragmentationSessionStatus {
@@ -185,8 +188,8 @@ message FuotaDeploymentDeviceListItem {
// Created at timestamp.
google.protobuf.Timestamp created_at = 3;
- // Updated at timestamp.
- google.protobuf.Timestamp updated_at = 4;
+ // Completed at timestamp.
+ google.protobuf.Timestamp completed_at = 4;
// McGroupSetup completed at timestamp.
google.protobuf.Timestamp mc_group_setup_completed_at = 5;
@@ -199,6 +202,9 @@ message FuotaDeploymentDeviceListItem {
// FragStatus completed at timestamp.
google.protobuf.Timestamp frag_status_completed_at = 8;
+
+ // Error message.
+ string error_msg = 9;
}
message FuotaDeploymentGatewayListItem {
@@ -354,3 +360,36 @@ message ListFuotaDeploymentGatewaysResponse {
// Result-set.
repeated FuotaDeploymentGatewayListItem result = 2;
}
+
+message ListFuotaDeploymentJobsRequest {
+ // FUOTA Deployment ID.
+ string fuota_deployment_id = 1;
+}
+
+message ListFuotaDeploymentJobsResponse {
+ // Jobs.
+ repeated FuotaDeploymentJob jobs = 1;
+}
+
+message FuotaDeploymentJob {
+ // Job identifier.
+ string job = 1;
+
+ // Created at.
+ google.protobuf.Timestamp created_at = 2;
+
+ // Completed at.
+ google.protobuf.Timestamp completed_at = 3;
+
+ // Max. retry count.
+ uint32 max_retry_count = 4;
+
+ // Attempt count.
+ uint32 attempt_count = 5;
+
+ // Scheduler run after.
+ google.protobuf.Timestamp scheduler_run_after = 6;
+
+ // Error message.
+ string error_msg = 7;
+}
diff --git a/api/rust/proto/chirpstack/api/fuota.proto b/api/rust/proto/chirpstack/api/fuota.proto
index ead9b854..4cfd546e 100644
--- a/api/rust/proto/chirpstack/api/fuota.proto
+++ b/api/rust/proto/chirpstack/api/fuota.proto
@@ -57,6 +57,9 @@ service FuotaService {
// Remove the given Gateway IDs from the FUOTA deployment.
rpc RemoveGateways(RemoveGatewaysFromFuotaDeploymentRequest) returns (google.protobuf.Empty) {}
// GetLogs returns the logs for the FUOTA deployment.
+
+ // List jobs for the given FUOTA deployment.
+ rpc ListJobs(ListFuotaDeploymentJobsRequest) returns (ListFuotaDeploymentJobsResponse) {}
}
enum RequestFragmentationSessionStatus {
@@ -185,8 +188,8 @@ message FuotaDeploymentDeviceListItem {
// Created at timestamp.
google.protobuf.Timestamp created_at = 3;
- // Updated at timestamp.
- google.protobuf.Timestamp updated_at = 4;
+ // Completed at timestamp.
+ google.protobuf.Timestamp completed_at = 4;
// McGroupSetup completed at timestamp.
google.protobuf.Timestamp mc_group_setup_completed_at = 5;
@@ -199,6 +202,9 @@ message FuotaDeploymentDeviceListItem {
// FragStatus completed at timestamp.
google.protobuf.Timestamp frag_status_completed_at = 8;
+
+ // Error message.
+ string error_msg = 9;
}
message FuotaDeploymentGatewayListItem {
@@ -354,3 +360,36 @@ message ListFuotaDeploymentGatewaysResponse {
// Result-set.
repeated FuotaDeploymentGatewayListItem result = 2;
}
+
+message ListFuotaDeploymentJobsRequest {
+ // FUOTA Deployment ID.
+ string fuota_deployment_id = 1;
+}
+
+message ListFuotaDeploymentJobsResponse {
+ // Jobs.
+ repeated FuotaDeploymentJob jobs = 1;
+}
+
+message FuotaDeploymentJob {
+ // Job identifier.
+ string job = 1;
+
+ // Created at.
+ google.protobuf.Timestamp created_at = 2;
+
+ // Completed at.
+ google.protobuf.Timestamp completed_at = 3;
+
+ // Max. retry count.
+ uint32 max_retry_count = 4;
+
+ // Attempt count.
+ uint32 attempt_count = 5;
+
+ // Scheduler run after.
+ google.protobuf.Timestamp scheduler_run_after = 6;
+
+ // Error message.
+ string error_msg = 7;
+}
diff --git a/chirpstack/migrations_postgres/2025-01-21-093745_add_fuota_support/up.sql b/chirpstack/migrations_postgres/2025-01-21-093745_add_fuota_support/up.sql
index b84d85fd..7e9164ea 100644
--- a/chirpstack/migrations_postgres/2025-01-21-093745_add_fuota_support/up.sql
+++ b/chirpstack/migrations_postgres/2025-01-21-093745_add_fuota_support/up.sql
@@ -30,12 +30,12 @@ create table fuota_deployment_device (
fuota_deployment_id uuid not null references fuota_deployment on delete cascade,
dev_eui bytea not null references device on delete cascade,
created_at timestamp with time zone not null,
- updated_at timestamp with time zone not null,
+ completed_at timestamp with time zone null,
mc_group_setup_completed_at timestamp with time zone null,
mc_session_completed_at timestamp with time zone null,
frag_session_setup_completed_at timestamp with time zone null,
frag_status_completed_at timestamp with time zone null,
- return_msg text not null,
+ error_msg text not null,
primary key (fuota_deployment_id, dev_eui)
);
@@ -56,7 +56,7 @@ create table fuota_deployment_job (
max_retry_count smallint not null,
attempt_count smallint not null,
scheduler_run_after timestamp with time zone not null,
- return_msg text not null,
+ error_msg text not null,
primary key (fuota_deployment_id, job)
);
diff --git a/chirpstack/migrations_sqlite/2025-01-27-100007_add_fuota_support/up.sql b/chirpstack/migrations_sqlite/2025-01-27-100007_add_fuota_support/up.sql
index 2abc38b6..0b1ae080 100644
--- a/chirpstack/migrations_sqlite/2025-01-27-100007_add_fuota_support/up.sql
+++ b/chirpstack/migrations_sqlite/2025-01-27-100007_add_fuota_support/up.sql
@@ -30,12 +30,12 @@ create table fuota_deployment_device (
fuota_deployment_id text not null references fuota_deployment on delete cascade,
dev_eui blob not null references device on delete cascade,
created_at datetime not null,
- updated_at datetime not null,
+ completed_at datetime null,
mc_group_setup_completed_at datetime null,
mc_session_completed_at datetime null,
frag_session_setup_completed_at datetime null,
frag_status_completed_at datetime null,
- return_msg text not null,
+ error_msg text not null,
primary key (fuota_deployment_id, dev_eui)
);
@@ -56,7 +56,7 @@ create table fuota_deployment_job (
max_retry_count smallint not null,
attempt_count smallint not null,
scheduler_run_after datetime not null,
- return_msg text not null,
+ error_msg text not null,
primary key (fuota_deployment_id, job)
);
diff --git a/chirpstack/src/api/device_profile.rs b/chirpstack/src/api/device_profile.rs
index 3cd1718c..220783c2 100644
--- a/chirpstack/src/api/device_profile.rs
+++ b/chirpstack/src/api/device_profile.rs
@@ -599,7 +599,14 @@ pub mod test {
mac_version: common::MacVersion::Lorawan103.into(),
reg_params_revision: common::RegParamsRevision::A.into(),
adr_algorithm_id: "default".into(),
- app_layer_params: Some(api::AppLayerParams::default()),
+ app_layer_params: Some(api::AppLayerParams {
+ ts003_version: api::Ts003Version::Ts003NotImplemented.into(),
+ ts003_f_port: 202,
+ ts004_version: api::Ts004Version::Ts004NotImplemented.into(),
+ ts004_f_port: 201,
+ ts005_version: api::Ts005Version::Ts005NotImplemented.into(),
+ ts005_f_port: 200,
+ }),
..Default::default()
}),
get_resp.get_ref().device_profile
diff --git a/chirpstack/src/api/fuota.rs b/chirpstack/src/api/fuota.rs
index 461f1879..7925266c 100644
--- a/chirpstack/src/api/fuota.rs
+++ b/chirpstack/src/api/fuota.rs
@@ -192,6 +192,13 @@ impl FuotaService for Fuota {
)
.await?;
+ let d = fuota::get_deployment(id).await.map_err(|e| e.status())?;
+ if d.started_at.is_some() {
+ return Err(Status::failed_precondition(
+ "FUOTA deployment has already started",
+ ));
+ }
+
let mut dp = fuota::FuotaDeployment {
id: id.into(),
name: req_dp.name.clone(),
@@ -361,6 +368,13 @@ impl FuotaService for Fuota {
)
.await?;
+ let d = fuota::get_deployment(dp_id).await.map_err(|e| e.status())?;
+ if d.started_at.is_some() {
+ return Err(Status::failed_precondition(
+ "FUOTA deployment has already started",
+ ));
+ }
+
let mut dev_euis = Vec::with_capacity(req.dev_euis.len());
for dev_eui in &req.dev_euis {
dev_euis.push(EUI64::from_str(dev_eui).map_err(|e| e.status())?);
@@ -438,7 +452,10 @@ impl FuotaService for Fuota {
fuota_deployment_id: d.fuota_deployment_id.to_string(),
dev_eui: d.dev_eui.to_string(),
created_at: Some(helpers::datetime_to_prost_timestamp(&d.created_at)),
- updated_at: Some(helpers::datetime_to_prost_timestamp(&d.updated_at)),
+ completed_at: d
+ .completed_at
+ .as_ref()
+ .map(|ts| helpers::datetime_to_prost_timestamp(ts)),
mc_group_setup_completed_at: d
.mc_group_setup_completed_at
.as_ref()
@@ -455,6 +472,7 @@ impl FuotaService for Fuota {
.frag_status_completed_at
.as_ref()
.map(|ts| helpers::datetime_to_prost_timestamp(ts)),
+ error_msg: d.error_msg.clone(),
})
.collect(),
});
@@ -480,6 +498,13 @@ impl FuotaService for Fuota {
)
.await?;
+ let d = fuota::get_deployment(dp_id).await.map_err(|e| e.status())?;
+ if d.started_at.is_some() {
+ return Err(Status::failed_precondition(
+ "FUOTA deployment has already started",
+ ));
+ }
+
let mut gateway_ids = Vec::with_capacity(req.gateway_ids.len());
for gateway_id in &req.gateway_ids {
gateway_ids.push(EUI64::from_str(gateway_id).map_err(|e| e.status())?);
@@ -566,6 +591,48 @@ impl FuotaService for Fuota {
);
Ok(resp)
}
+
+ async fn list_jobs(
+ &self,
+ request: Request<api::ListFuotaDeploymentJobsRequest>,
+ ) -> Result<Response<api::ListFuotaDeploymentJobsResponse>, Status> {
+ let req = request.get_ref();
+ let dp_id = Uuid::from_str(&req.fuota_deployment_id).map_err(|e| e.status())?;
+
+ self.validator
+ .validate(
+ request.extensions(),
+ validator::ValidateFuotaDeploymentAccess::new(validator::Flag::Read, dp_id),
+ )
+ .await?;
+
+ let jobs = fuota::list_jobs(dp_id).await.map_err(|e| e.status())?;
+
+ let mut resp = Response::new(api::ListFuotaDeploymentJobsResponse {
+ jobs: jobs
+ .iter()
+ .map(|j| api::FuotaDeploymentJob {
+ job: j.job.to_string(),
+ created_at: Some(helpers::datetime_to_prost_timestamp(&j.created_at)),
+ completed_at: j
+ .completed_at
+ .as_ref()
+ .map(|ts| helpers::datetime_to_prost_timestamp(ts)),
+ max_retry_count: j.max_retry_count as u32,
+ attempt_count: j.attempt_count as u32,
+ scheduler_run_after: Some(helpers::datetime_to_prost_timestamp(
+ &j.scheduler_run_after,
+ )),
+ error_msg: j.error_msg.clone(),
+ })
+ .collect(),
+ });
+ resp.metadata_mut().insert(
+ "x-log-fuota_deployment_id",
+ req.fuota_deployment_id.parse().unwrap(),
+ );
+ Ok(resp)
+ }
}
#[cfg(test)]
@@ -723,21 +790,6 @@ mod test {
assert_eq!(1, list_resp.result.len());
assert_eq!(create_resp.id, list_resp.result[0].id);
- // start deployment
- let start_req = get_request(
- &u.id,
- api::StartFuotaDeploymentRequest {
- id: create_resp.id.clone(),
- },
- );
- service.start_deployment(start_req).await.unwrap();
- let jobs = fuota::list_jobs(Uuid::from_str(&create_resp.id).unwrap())
- .await
- .unwrap();
- assert_eq!(1, jobs.len());
- assert_eq!(create_resp.id, jobs[0].fuota_deployment_id.to_string());
- assert_eq!(fields::FuotaJob::CreateMcGroup, jobs[0].job);
-
// add device
let add_dev_req = get_request(
&u.id,
@@ -835,6 +887,21 @@ mod test {
assert_eq!(0, list_gws_resp.total_count);
assert_eq!(0, list_gws_resp.result.len());
+ // start deployment
+ let start_req = get_request(
+ &u.id,
+ api::StartFuotaDeploymentRequest {
+ id: create_resp.id.clone(),
+ },
+ );
+ service.start_deployment(start_req).await.unwrap();
+ let jobs = fuota::list_jobs(Uuid::from_str(&create_resp.id).unwrap())
+ .await
+ .unwrap();
+ assert_eq!(1, jobs.len());
+ assert_eq!(create_resp.id, jobs[0].fuota_deployment_id.to_string());
+ assert_eq!(fields::FuotaJob::CreateMcGroup, jobs[0].job);
+
// delete deployment
let delete_req = get_request(
&u.id,
diff --git a/chirpstack/src/applayer/clocksync.rs b/chirpstack/src/applayer/clocksync.rs
index 9bdbd3d3..cc400a80 100644
--- a/chirpstack/src/applayer/clocksync.rs
+++ b/chirpstack/src/applayer/clocksync.rs
@@ -201,7 +201,7 @@ mod test {
name: "test-dp".into(),
tenant_id: t.id,
app_layer_params: fields::AppLayerParams {
- ts003_version: Some(fields::Ts003Version::V100),
+ ts003_version: Some(Ts003Version::V100),
..Default::default()
},
..Default::default()
diff --git a/chirpstack/src/applayer/fragmentation.rs b/chirpstack/src/applayer/fragmentation.rs
new file mode 100644
index 00000000..92af2263
--- /dev/null
+++ b/chirpstack/src/applayer/fragmentation.rs
@@ -0,0 +1,96 @@
+use anyhow::Result;
+use chrono::Utc;
+use tracing::{info, warn};
+
+use crate::storage::fields::device_profile::Ts004Version;
+use crate::storage::{device, device_profile, fuota};
+use lrwn::applayer::fragmentation;
+
+pub async fn handle_uplink(
+ dev: &device::Device,
+ dp: &device_profile::DeviceProfile,
+ data: &[u8],
+) -> Result<()> {
+ let version = dp
+ .app_layer_params
+ .ts004_version
+ .ok_or_else(|| anyhow!("Device does not support TS004"))?;
+
+ match version {
+ Ts004Version::V100 => handle_uplink_v100(dev, data).await,
+ }
+}
+
+async fn handle_uplink_v100(dev: &device::Device, data: &[u8]) -> Result<()> {
+ let pl = fragmentation::v1::Payload::from_slice(true, data)?;
+
+ match pl {
+ fragmentation::v1::Payload::FragSessionSetupAns(pl) => {
+ handle_v1_frag_session_setup_ans(dev, pl).await?
+ }
+ fragmentation::v1::Payload::FragSessionStatusAns(pl) => {
+ handle_v1_frag_session_status_ans(dev, pl).await?
+ }
+ _ => {}
+ }
+
+ Ok(())
+}
+
+async fn handle_v1_frag_session_setup_ans(
+ dev: &device::Device,
+ pl: fragmentation::v1::FragSessionSetupAnsPayload,
+) -> Result<()> {
+ info!("Handling FragSessionSetupAns");
+
+ let mut fuota_dev = fuota::get_latest_device_by_dev_eui(dev.dev_eui).await?;
+
+ if pl.encoding_unsupported
+ | pl.not_enough_memory
+ | pl.frag_session_index_not_supported
+ | pl.wrong_descriptor
+ {
+ warn!(
+ frag_index = pl.frag_index,
+ encoding_unsupported = pl.encoding_unsupported,
+ not_enough_memory = pl.not_enough_memory,
+ frag_session_index_not_supported = pl.frag_session_index_not_supported,
+ wrong_descriptor = pl.wrong_descriptor,
+ "FragSessionAns contains errors"
+ );
+ fuota_dev.error_msg = format!("Error: FragSessionAns response encoding_unsupported={}, not_enough_memory={}, frag_session_index_not_supported={}, wrong_descriptor={}", pl.encoding_unsupported, pl.not_enough_memory, pl.frag_session_index_not_supported, pl.wrong_descriptor);
+ } else {
+ fuota_dev.frag_session_setup_completed_at = Some(Utc::now());
+ }
+
+ let _ = fuota::update_device(fuota_dev).await?;
+
+ Ok(())
+}
+
+async fn handle_v1_frag_session_status_ans(
+ dev: &device::Device,
+ pl: fragmentation::v1::FragSessionStatusAnsPayload,
+) -> Result<()> {
+ info!("Handling FragSessionStatusAnsPayload");
+
+ let mut fuota_dev = fuota::get_latest_device_by_dev_eui(dev.dev_eui).await?;
+
+ if pl.missing_frag != 0 || pl.status.not_enough_matrix_memory {
+ warn!(
+ frag_index = pl.received_and_index.frag_index,
+ nb_frag_received = pl.received_and_index.nb_frag_received,
+ missing_frag = pl.missing_frag,
+ not_enough_matrix_memory = pl.status.not_enough_matrix_memory,
+ "FragSessionStatusAns contains errors"
+ );
+
+ fuota_dev.error_msg = format!("Error: FragSessionStatusAns response nb_frag_received={}, missing_frag={}, not_enough_matrix_memory={}", pl.received_and_index.nb_frag_received, pl.missing_frag, pl.status.not_enough_matrix_memory);
+ } else {
+ fuota_dev.frag_status_completed_at = Some(Utc::now());
+ }
+
+ let _ = fuota::update_device(fuota_dev).await?;
+
+ Ok(())
+}
diff --git a/chirpstack/src/applayer/fuota/flow.rs b/chirpstack/src/applayer/fuota/flow.rs
new file mode 100644
index 00000000..8a18fd16
--- /dev/null
+++ b/chirpstack/src/applayer/fuota/flow.rs
@@ -0,0 +1,548 @@
+use std::time::Duration;
+
+use anyhow::Result;
+use chrono::{DateTime, TimeDelta, Utc};
+use tracing::info;
+
+use lrwn::applayer::{fragmentation, multicastsetup};
+use lrwn::region::MacVersion;
+
+use crate::config;
+use crate::downlink;
+use crate::gpstime::ToGpsTime;
+use crate::storage::fields::{FuotaJob, RequestFragmentationSessionStatus};
+use crate::storage::{device_keys, device_profile, device_queue, fuota, multicast};
+
+pub struct Flow {
+ scheduler_interval: Duration,
+ job: fuota::FuotaDeploymentJob,
+ fuota_deployment: fuota::FuotaDeployment,
+ device_profile: device_profile::DeviceProfile,
+}
+
+impl Flow {
+ pub async fn handle_job(job: fuota::FuotaDeploymentJob) -> Result<()> {
+ let conf = config::get();
+
+ let fuota_deployment = fuota::get_deployment(job.fuota_deployment_id.into()).await?;
+ let device_profile =
+ device_profile::get(&fuota_deployment.device_profile_id.into()).await?;
+
+ let mut flow = Flow {
+ job,
+ fuota_deployment,
+ device_profile,
+ scheduler_interval: conf.network.scheduler.interval,
+ };
+ flow.dispatch().await
+ }
+
+ async fn dispatch(&mut self) -> Result<()> {
+ let resp = match self.job.job {
+ FuotaJob::CreateMcGroup => self.create_mc_group().await,
+ FuotaJob::AddDevsToMcGroup => self.add_devices_to_multicast_group().await,
+ FuotaJob::AddGwsToMcGroup => self.add_gateways_to_multicast_group().await,
+ FuotaJob::McGroupSetup => self.multicast_group_setup().await,
+ FuotaJob::FragSessionSetup => self.fragmentation_session_setup().await,
+ FuotaJob::McSession => self.multicast_session_setup().await,
+ FuotaJob::Enqueue => self.enqueue().await,
+ FuotaJob::FragStatus => self.fragmentation_status().await,
+ FuotaJob::Complete => self.complete().await,
+ };
+
+ match resp {
+ Ok(Some((next_job, scheduler_run_after))) => {
+ if self.job.job == next_job {
+ // Re-run the same job in the future.
+ let mut job = self.job.clone();
+ job.scheduler_run_after = scheduler_run_after;
+ let _ = fuota::update_job(job).await?;
+ } else {
+ // Update the current job (to increment the attempt count).
+ let job = self.job.clone();
+ let _ = fuota::update_job(job).await?;
+
+ // Create the next job (which automatically sets the current job to completed).
+ let _ = fuota::create_job(fuota::FuotaDeploymentJob {
+ fuota_deployment_id: self.job.fuota_deployment_id,
+ job: next_job,
+ max_retry_count: match next_job {
+ FuotaJob::McGroupSetup
+ | FuotaJob::FragSessionSetup
+ | FuotaJob::McSession => self.fuota_deployment.unicast_max_retry_count,
+ _ => 0,
+ },
+ scheduler_run_after,
+ ..Default::default()
+ })
+ .await?;
+ }
+ }
+ Ok(None) => {
+ // No further jobs to execute, set the current job to completed.
+ let mut job = self.job.clone();
+ job.completed_at = Some(Utc::now());
+ let _ = fuota::update_job(job).await?;
+ }
+ Err(e) => {
+ // Re-run the same job in the future.
+ let mut job = self.job.clone();
+ job.scheduler_run_after = Utc::now() + self.scheduler_interval;
+ job.error_msg = format!("Error: {}", e);
+ let _ = fuota::update_job(job).await?;
+ return Err(e);
+ }
+ }
+
+ Ok(())
+ }
+
+ async fn create_mc_group(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // If this job fails, then there is no need to execute the others.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(None);
+ }
+
+ info!("Creating multicast-group for FUOTA deployment");
+ self.job.attempt_count += 1;
+
+ // Get McAppSKey + McNwkSKey.
+ let mc_app_s_key = multicastsetup::v1::get_mc_app_s_key(
+ self.fuota_deployment.multicast_key,
+ self.fuota_deployment.multicast_addr,
+ )?;
+ let mc_nwk_s_key = multicastsetup::v1::get_mc_net_s_key(
+ self.fuota_deployment.multicast_key,
+ self.fuota_deployment.multicast_addr,
+ )?;
+
+ let _ = multicast::create(multicast::MulticastGroup {
+ id: self.fuota_deployment.id,
+ application_id: self.fuota_deployment.application_id,
+ name: format!("fuota-{}", self.fuota_deployment.id),
+ region: self.device_profile.region,
+ mc_addr: self.fuota_deployment.multicast_addr,
+ mc_nwk_s_key,
+ mc_app_s_key,
+ f_cnt: 0,
+ group_type: self.fuota_deployment.multicast_group_type.clone(),
+ frequency: self.fuota_deployment.multicast_frequency,
+ dr: self.fuota_deployment.multicast_dr,
+ class_b_ping_slot_nb_k: self.fuota_deployment.multicast_class_b_ping_slot_nb_k,
+ class_c_scheduling_type: self.fuota_deployment.multicast_class_c_scheduling_type,
+ ..Default::default()
+ })
+ .await?;
+
+ Ok(Some((FuotaJob::AddDevsToMcGroup, Utc::now())))
+ }
+
+ async fn add_devices_to_multicast_group(
+ &mut self,
+ ) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // If this job fails, then there is no need to execute the others.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(None);
+ }
+
+ info!("Adding devices to multicast-group");
+ self.job.attempt_count += 1;
+
+ let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?;
+ for fuota_d in fuota_devices {
+ multicast::add_device(&fuota_d.fuota_deployment_id, &fuota_d.dev_eui).await?;
+ }
+
+ Ok(Some((FuotaJob::AddGwsToMcGroup, Utc::now())))
+ }
+
+ async fn add_gateways_to_multicast_group(
+ &mut self,
+ ) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // If this job fails, then there is no need to execute the others.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(None);
+ }
+
+ info!("Adding gateways to multicast-group (if any)");
+ self.job.attempt_count += 1;
+
+ let fuota_gws = fuota::get_gateways(self.job.fuota_deployment_id.into(), -1, 0).await?;
+ for fuota_gw in fuota_gws {
+ multicast::add_gateway(&fuota_gw.fuota_deployment_id, &fuota_gw.gateway_id).await?;
+ }
+
+ Ok(Some((FuotaJob::McGroupSetup, Utc::now())))
+ }
+
+ async fn multicast_group_setup(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(Some((FuotaJob::FragSessionSetup, Utc::now())));
+ }
+
+ info!("Sending McGroupSetupReq commands to devices");
+ self.job.attempt_count += 1;
+
+ let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?;
+
+ // Filter on devices that have not completed the McGroupSetup.
+ let fuota_devices: Vec<fuota::FuotaDeploymentDevice> = fuota_devices
+ .into_iter()
+ .filter(|d| d.mc_group_setup_completed_at.is_none())
+ .collect();
+
+ for fuota_dev in &fuota_devices {
+ let dev_keys = device_keys::get(&fuota_dev.dev_eui).await?;
+ let mc_root_key = match self.device_profile.mac_version {
+ MacVersion::LORAWAN_1_0_0
+ | MacVersion::LORAWAN_1_0_1
+ | MacVersion::LORAWAN_1_0_2
+ | MacVersion::LORAWAN_1_0_3
+ | MacVersion::LORAWAN_1_0_4 => {
+ multicastsetup::v1::get_mc_root_key_for_gen_app_key(dev_keys.gen_app_key)?
+ }
+ MacVersion::LORAWAN_1_1_0 | MacVersion::Latest => {
+ multicastsetup::v1::get_mc_root_key_for_app_key(dev_keys.app_key)?
+ }
+ };
+ let mc_ke_key = multicastsetup::v1::get_mc_ke_key(mc_root_key)?;
+ let mc_key_encrypted =
+ multicastsetup::v1::encrypt_mc_key(mc_ke_key, self.fuota_deployment.multicast_key);
+
+ let pl = multicastsetup::v1::Payload::McGroupSetupReq(
+ multicastsetup::v1::McGroupSetupReqPayload {
+ mc_group_id_header: multicastsetup::v1::McGroupSetupReqPayloadMcGroupIdHeader {
+ mc_group_id: 0,
+ },
+ mc_addr: self.fuota_deployment.multicast_addr,
+ mc_key_encrypted,
+ min_mc_f_count: 0,
+ max_mc_f_count: u32::MAX,
+ },
+ );
+
+ device_queue::enqueue_item(device_queue::DeviceQueueItem {
+ dev_eui: fuota_dev.dev_eui,
+ f_port: self.device_profile.app_layer_params.ts005_f_port.into(),
+ data: pl.to_vec()?,
+ ..Default::default()
+ })
+ .await?;
+ }
+
+ if !fuota_devices.is_empty() {
+ // There are devices pending setup, we need to re-run this job.
+ let scheduler_run_after =
+ Utc::now() + TimeDelta::seconds(self.device_profile.uplink_interval as i64);
+ Ok(Some((FuotaJob::McGroupSetup, scheduler_run_after)))
+ } else {
+ // All devices have completed the setup, move on to next job.
+ Ok(Some((FuotaJob::FragSessionSetup, Utc::now())))
+ }
+ }
+
+ async fn fragmentation_session_setup(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(Some((FuotaJob::McSession, Utc::now())));
+ }
+
+ info!("Set timeout error to devices that did not respond to McGroupSetupReq");
+ fuota::set_device_timeout_error(self.fuota_deployment.id.into(), true, false, false, false)
+ .await?;
+
+ info!("Sending FragSessionSetupReq commands to devices");
+ self.job.attempt_count += 1;
+
+ let fragment_size = self.fuota_deployment.fragmentation_fragment_size as usize;
+ let fragments =
+ (self.fuota_deployment.payload.len() as f32 / fragment_size as f32).ceil() as usize;
+ let padding =
+ (fragment_size - (self.fuota_deployment.payload.len() % fragment_size)) % fragment_size;
+
+ let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?;
+
+ // Filter on devices that have completed the previous step, but not yet the FragSessionSetup.
+ let fuota_devices: Vec<fuota::FuotaDeploymentDevice> = fuota_devices
+ .into_iter()
+ .filter(|d| {
+ d.mc_group_setup_completed_at.is_some()
+ && d.frag_session_setup_completed_at.is_none()
+ })
+ .collect();
+
+ for fuota_dev in &fuota_devices {
+ let pl = fragmentation::v1::Payload::FragSessionSetupReq(
+ fragmentation::v1::FragSessionSetupReqPayload {
+ frag_session: fragmentation::v1::FragSessionSetuReqPayloadFragSession {
+ mc_group_bit_mask: [true, false, false, false],
+ frag_index: 0,
+ },
+ nb_frag: fragments as u16,
+ frag_size: fragment_size as u8,
+ padding: padding as u8,
+ control: fragmentation::v1::FragSessionSetuReqPayloadControl {
+ block_ack_delay: 0,
+ fragmentation_matrix: 0,
+ },
+ descriptor: [0, 0, 0, 0],
+ },
+ );
+
+ device_queue::enqueue_item(device_queue::DeviceQueueItem {
+ dev_eui: fuota_dev.dev_eui,
+ f_port: self.device_profile.app_layer_params.ts004_f_port.into(),
+ data: pl.to_vec()?,
+ ..Default::default()
+ })
+ .await?;
+ }
+
+ if !fuota_devices.is_empty() {
+ // There are devices pending setup, we need to re-run this job.
+ let scheduler_run_after =
+ Utc::now() + TimeDelta::seconds(self.device_profile.uplink_interval as i64);
+ Ok(Some((FuotaJob::FragSessionSetup, scheduler_run_after)))
+ } else {
+ // All devices have completed the setup, move on to next job.
+ Ok(Some((FuotaJob::McSession, Utc::now())))
+ }
+ }
+
+ async fn multicast_session_setup(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(Some((FuotaJob::Enqueue, Utc::now())));
+ }
+
+ info!("Set timeout error to devices that did not respond to FragSessionSetupReq");
+ fuota::set_device_timeout_error(self.fuota_deployment.id.into(), false, false, true, false)
+ .await?;
+
+ info!("Sending McClassB/McClassCSessionReq commands to devices");
+ self.job.attempt_count += 1;
+
+ let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?;
+
+ // Filter on devices that have completed the previous step, but not yet the McSession.
+ let fuota_devices: Vec<fuota::FuotaDeploymentDevice> = fuota_devices
+ .into_iter()
+ .filter(|d| {
+ d.frag_session_setup_completed_at.is_some() && d.mc_session_completed_at.is_none()
+ })
+ .collect();
+
+ for fuota_dev in &fuota_devices {
+ // We want to start the session (retry_count + 1) x the uplink_interval.
+ // Note that retry_count=0 means only one attempt.
+ let session_start = (Utc::now()
+ + TimeDelta::seconds(
+ (self.job.max_retry_count as i64 + 1)
+ * self.device_profile.uplink_interval as i64,
+ ))
+ .to_gps_time()
+ .num_seconds()
+ % (1 << 32);
+
+ let pl = match self.fuota_deployment.multicast_group_type.as_ref() {
+ "B" => multicastsetup::v1::Payload::McClassBSessionReq(
+ multicastsetup::v1::McClassBSessionReqPayload {
+ mc_group_id_header:
+ multicastsetup::v1::McClassBSessionReqPayloadMcGroupIdHeader {
+ mc_group_id: 0,
+ },
+ session_time: (session_start - (session_start % 128)) as u32,
+ time_out_periodicity:
+ multicastsetup::v1::McClassBSessionReqPayloadTimeOutPeriodicity {
+ time_out: self.fuota_deployment.multicast_timeout as u8,
+ periodicity: self.fuota_deployment.multicast_class_b_ping_slot_nb_k
+ as u8,
+ },
+ dl_frequ: self.fuota_deployment.multicast_frequency as u32,
+ dr: self.fuota_deployment.multicast_dr as u8,
+ },
+ ),
+ "C" => multicastsetup::v1::Payload::McClassCSessionReq(
+ multicastsetup::v1::McClassCSessionReqPayload {
+ mc_group_id_header:
+ multicastsetup::v1::McClassCSessionReqPayloadMcGroupIdHeader {
+ mc_group_id: 0,
+ },
+ session_time: session_start as u32,
+ session_time_out:
+ multicastsetup::v1::McClassCSessionReqPayloadSessionTimeOut {
+ time_out: self.fuota_deployment.multicast_timeout as u8,
+ },
+ dl_frequ: self.fuota_deployment.multicast_frequency as u32,
+ dr: self.fuota_deployment.multicast_dr as u8,
+ },
+ ),
+ _ => {
+ return Err(anyhow!(
+ "Unsupported group-type: {}",
+ self.fuota_deployment.multicast_group_type
+ ))
+ }
+ };
+
+ device_queue::enqueue_item(device_queue::DeviceQueueItem {
+ dev_eui: fuota_dev.dev_eui,
+ f_port: self.device_profile.app_layer_params.ts005_f_port.into(),
+ data: pl.to_vec()?,
+ ..Default::default()
+ })
+ .await?;
+ }
+
+ // In this case we need to exactly try the max. attempts, because this is what the
+ // session-start time calculation is based on. If we continue with enqueueing too
+ // early, the multicast-session hasn't started yet.
+ let scheduler_run_after =
+ Utc::now() + TimeDelta::seconds(self.device_profile.uplink_interval as i64);
+ Ok(Some((FuotaJob::McSession, scheduler_run_after)))
+ }
+
+ async fn enqueue(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(Some((FuotaJob::FragStatus, Utc::now())));
+ }
+
+ info!("Set timeout error to devices that did not respond to McSessionReq");
+ fuota::set_device_timeout_error(self.fuota_deployment.id.into(), false, true, false, false)
+ .await?;
+
+ info!("Enqueueing fragmented payload to multicast group");
+ self.job.attempt_count += 1;
+
+ let payload_length = self.fuota_deployment.payload.len();
+ let fragment_size = self.fuota_deployment.fragmentation_fragment_size as usize;
+ let padding = (fragment_size - (payload_length % fragment_size)) % fragment_size;
+
+ let fragments = (payload_length as f32 / fragment_size as f32).ceil() as usize;
+ let redundancy = (fragments as f32
+ * self.fuota_deployment.fragmentation_redundancy_percentage as f32
+ / 100.0)
+ .ceil() as usize;
+
+ let mut payload = self.fuota_deployment.payload.clone();
+ payload.extend_from_slice(&vec![0; padding]);
+
+ let encoded_fragments = fragmentation::v1::encode(&payload, fragment_size, redundancy)?;
+
+ for (i, frag) in encoded_fragments.iter().enumerate() {
+ let pl =
+ fragmentation::v1::Payload::DataFragment(fragmentation::v1::DataFragmentPayload {
+ index_and_n: fragmentation::v1::DataFragmentPayloadIndexAndN {
+ frag_index: 0,
+ n: (i + 1) as u16,
+ },
+ data: frag.clone(),
+ });
+
+ let _ = downlink::multicast::enqueue(multicast::MulticastGroupQueueItem {
+ multicast_group_id: self.fuota_deployment.id,
+ f_port: self.device_profile.app_layer_params.ts004_f_port as i16,
+ data: pl.to_vec()?,
+ ..Default::default()
+ })
+ .await?;
+ }
+
+ match self.fuota_deployment.request_fragmentation_session_status {
+ RequestFragmentationSessionStatus::NoRequest => {
+ Ok(Some((FuotaJob::Complete, Utc::now())))
+ }
+ RequestFragmentationSessionStatus::AfterFragEnqueue => {
+ Ok(Some((FuotaJob::FragStatus, Utc::now())))
+ }
+ RequestFragmentationSessionStatus::AfterSessTimeout => {
+ let timeout = match self.fuota_deployment.multicast_group_type.as_ref() {
+ "B" => Duration::from_secs(
+ 128 * (1 << self.fuota_deployment.multicast_timeout as u64),
+ ),
+ "C" => Duration::from_secs(1 << self.fuota_deployment.multicast_timeout as u64),
+ _ => return Err(anyhow!("Invalid multicast-group type")),
+ };
+ Ok(Some((FuotaJob::FragStatus, Utc::now() + timeout)))
+ }
+ }
+ }
+
+ async fn fragmentation_status(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(Some((FuotaJob::Complete, Utc::now())));
+ }
+
+ info!("Enqueue FragSessionStatusReq");
+ self.job.attempt_count += 1;
+
+ let fuota_devices = fuota::get_devices(self.job.fuota_deployment_id.into(), -1, 0).await?;
+
+ // Filter on devices that have completed the multicast-session setup but
+ // not yet responded to the FragSessionStatusReq.
+ let fuota_devices: Vec<fuota::FuotaDeploymentDevice> = fuota_devices
+ .into_iter()
+ .filter(|d| d.mc_session_completed_at.is_some() && d.frag_status_completed_at.is_none())
+ .collect();
+
+ for fuota_dev in &fuota_devices {
+ let pl = fragmentation::v1::Payload::FragSessionStatusReq(
+ fragmentation::v1::FragSessionStatusReqPayload {
+ participants: true,
+ frag_index: 0,
+ },
+ );
+
+ device_queue::enqueue_item(device_queue::DeviceQueueItem {
+ dev_eui: fuota_dev.dev_eui,
+ f_port: self.device_profile.app_layer_params.ts004_f_port.into(),
+ data: pl.to_vec()?,
+ ..Default::default()
+ })
+ .await?;
+ }
+
+ if !fuota_devices.is_empty() {
+ // There are devices pending setup, we need to re-run this job.
+ let scheduler_run_after =
+ Utc::now() + TimeDelta::seconds(self.device_profile.uplink_interval as i64);
+ Ok(Some((FuotaJob::FragStatus, scheduler_run_after)))
+ } else {
+ Ok(Some((FuotaJob::Complete, Utc::now())))
+ }
+ }
+
+ async fn complete(&mut self) -> Result<Option<(FuotaJob, DateTime<Utc>)>> {
+ // Proceed with next step after reaching the max attempts.
+ if self.job.attempt_count > self.job.max_retry_count {
+ return Ok(None);
+ }
+
+ info!("Complete FUOTA deployment");
+ self.job.attempt_count += 1;
+ self.fuota_deployment.completed_at = Some(Utc::now());
+
+ if self.fuota_deployment.request_fragmentation_session_status
+ == RequestFragmentationSessionStatus::NoRequest
+ {
+ fuota::set_device_completed(self.fuota_deployment.id.into(), true, true, true, false)
+ .await?;
+ } else {
+ fuota::set_device_completed(self.fuota_deployment.id.into(), true, true, true, true)
+ .await?;
+ fuota::set_device_timeout_error(
+ self.fuota_deployment.id.into(),
+ false,
+ false,
+ false,
+ true,
+ )
+ .await?;
+ }
+
+ Ok(None)
+ }
+}
diff --git a/chirpstack/src/applayer/fuota/mod.rs b/chirpstack/src/applayer/fuota/mod.rs
new file mode 100644
index 00000000..842d2922
--- /dev/null
+++ b/chirpstack/src/applayer/fuota/mod.rs
@@ -0,0 +1,9 @@
+use tracing::info;
+
+pub mod flow;
+pub mod scheduler;
+
+pub async fn setup() {
+ info!("Setting up FUOTA scheduler loop");
+ tokio::spawn(scheduler::scheduler_loop());
+}
diff --git a/chirpstack/src/applayer/fuota/scheduler.rs b/chirpstack/src/applayer/fuota/scheduler.rs
new file mode 100644
index 00000000..f684792e
--- /dev/null
+++ b/chirpstack/src/applayer/fuota/scheduler.rs
@@ -0,0 +1,45 @@
+use anyhow::Result;
+use tokio::time::sleep;
+use tracing::{error, span, trace, Instrument, Level};
+
+use crate::applayer::fuota::flow;
+use crate::config;
+use crate::storage::fuota;
+
+pub async fn scheduler_loop() {
+ let conf = config::get();
+
+ loop {
+ trace!("Starting fuota scheduler_loop run");
+ if let Err(err) = schedule_batch(conf.network.scheduler.batch_size).await {
+ error!(error = %err, "Scheduling FUOTA batch error");
+ } else {
+ trace!("schedule_batch completed without error");
+ }
+ sleep(conf.network.scheduler.interval).await;
+ }
+}
+
+async fn schedule_batch(size: usize) -> Result<()> {
+ trace!("Get schedulable fuota jobs");
+ let jobs = fuota::get_schedulable_jobs(size).await?;
+ trace!(job_count = jobs.len(), "Got this number of fuota jobs");
+
+ let mut handles = vec![];
+
+ for job in jobs {
+ // Spawn the batch as async tasks.
+ let handle = tokio::spawn(async move {
+ let span = span!(Level::INFO, "job", fuota_deployment_id = %job.fuota_deployment_id, job = %job.job);
+
+ if let Err(e) = flow::Flow::handle_job(job).instrument(span).await {
+ error!(error = %e, "Handle FUOTA job error");
+ }
+ });
+ handles.push(handle);
+ }
+
+ futures::future::join_all(handles).await;
+
+ Ok(())
+}
diff --git a/chirpstack/src/applayer/mod.rs b/chirpstack/src/applayer/mod.rs
index ef27a556..f15e3fc7 100644
--- a/chirpstack/src/applayer/mod.rs
+++ b/chirpstack/src/applayer/mod.rs
@@ -5,6 +5,9 @@ use crate::storage::{device, device_profile};
use chirpstack_api::gw;
pub mod clocksync;
+pub mod fragmentation;
+pub mod fuota;
+pub mod multicastsetup;
pub async fn handle_uplink(
dev: &device::Device,
@@ -31,9 +34,15 @@ async fn _handle_uplink(
.instrument(span)
.await
} else if dp.app_layer_params.ts004_f_port == f_port {
- unimplemented!()
+ let span = span!(Level::INFO, "ts004");
+ fragmentation::handle_uplink(dev, dp, data)
+ .instrument(span)
+ .await
} else if dp.app_layer_params.ts005_f_port == f_port {
- unimplemented!()
+ let span = span!(Level::INFO, "ts005");
+ multicastsetup::handle_uplink(dev, dp, data)
+ .instrument(span)
+ .await
} else {
return Err(anyhow!("Unexpected f_port {}", f_port));
}
diff --git a/chirpstack/src/applayer/multicastsetup.rs b/chirpstack/src/applayer/multicastsetup.rs
new file mode 100644
index 00000000..c070273e
--- /dev/null
+++ b/chirpstack/src/applayer/multicastsetup.rs
@@ -0,0 +1,139 @@
+use anyhow::Result;
+use chrono::Utc;
+use tracing::{info, warn};
+
+use crate::storage::fields::device_profile::Ts005Version;
+use crate::storage::{device, device_profile, fuota};
+use lrwn::applayer::multicastsetup;
+
+pub async fn handle_uplink(
+ dev: &device::Device,
+ dp: &device_profile::DeviceProfile,
+ data: &[u8],
+) -> Result<()> {
+ let version = dp
+ .app_layer_params
+ .ts005_version
+ .ok_or_else(|| anyhow!("Device does not support TS005"))?;
+
+ match version {
+ Ts005Version::V100 => handle_uplink_v100(dev, data).await,
+ }
+}
+
+async fn handle_uplink_v100(dev: &device::Device, data: &[u8]) -> Result<()> {
+ let pl = multicastsetup::v1::Payload::from_slice(true, data)?;
+
+ match pl {
+ multicastsetup::v1::Payload::McGroupSetupAns(pl) => {
+ handle_v1_mc_group_setup_ans(dev, pl).await?
+ }
+ multicastsetup::v1::Payload::McClassBSessionAns(pl) => {
+ handle_v1_mc_class_b_session_ans(dev, pl).await?
+ }
+ multicastsetup::v1::Payload::McClassCSessionAns(pl) => {
+ handle_v1_mc_class_c_session_ans(dev, pl).await?
+ }
+ _ => {}
+ }
+
+ Ok(())
+}
+
+async fn handle_v1_mc_group_setup_ans(
+ dev: &device::Device,
+ pl: multicastsetup::v1::McGroupSetupAnsPayload,
+) -> Result<()> {
+ info!("Handling McGroupSetupAns");
+
+ let mut fuota_dev = fuota::get_latest_device_by_dev_eui(dev.dev_eui).await?;
+
+ if pl.mc_group_id_header.id_error {
+ warn!(
+ mc_group_id = pl.mc_group_id_header.mc_group_id,
+ id_error = true,
+ "McGroupSetupAns contains errors"
+ );
+ fuota_dev.error_msg = "Error: McGroupSetupAns response id_error=true".into();
+ } else {
+ fuota_dev.mc_group_setup_completed_at = Some(Utc::now());
+ }
+
+ let _ = fuota::update_device(fuota_dev).await?;
+
+ Ok(())
+}
+
+async fn handle_v1_mc_class_b_session_ans(
+ dev: &device::Device,
+ pl: multicastsetup::v1::McClassBSessionAnsPayload,
+) -> Result<()> {
+ info!("Handling McClassBSessionAns");
+
+ let mut fuota_dev = fuota::get_latest_device_by_dev_eui(dev.dev_eui).await?;
+
+ if pl.status_and_mc_group_id.dr_error
+ | pl.status_and_mc_group_id.freq_error
+ | pl.status_and_mc_group_id.mc_group_undefined
+ {
+ warn!(
+ dr_error = pl.status_and_mc_group_id.dr_error,
+ freq_error = pl.status_and_mc_group_id.freq_error,
+ mc_group_undefined = pl.status_and_mc_group_id.mc_group_undefined,
+ "McClassBSessionAns contains errors"
+ );
+
+ fuota_dev.error_msg= format!("Error: McClassBSessionAns response dr_error: {}, freq_error: {}, mc_group_undefined: {}",
+ pl.status_and_mc_group_id.dr_error,
+ pl.status_and_mc_group_id.freq_error,
+ pl.status_and_mc_group_id.mc_group_undefined,
+ );
+ } else {
+ info!(
+ time_to_start = pl.time_to_start.unwrap_or_default(),
+ "McClassBSessionAns OK"
+ );
+ fuota_dev.mc_session_completed_at = Some(Utc::now());
+ }
+
+ let _ = fuota::update_device(fuota_dev).await?;
+
+ Ok(())
+}
+
+async fn handle_v1_mc_class_c_session_ans(
+ dev: &device::Device,
+ pl: multicastsetup::v1::McClassCSessionAnsPayload,
+) -> Result<()> {
+ info!("Handling McClassCSessionAns");
+
+ let mut fuota_dev = fuota::get_latest_device_by_dev_eui(dev.dev_eui).await?;
+
+ if pl.status_and_mc_group_id.dr_error
+ | pl.status_and_mc_group_id.freq_error
+ | pl.status_and_mc_group_id.mc_group_undefined
+ {
+ warn!(
+ dr_error = pl.status_and_mc_group_id.dr_error,
+ freq_error = pl.status_and_mc_group_id.freq_error,
+ mc_group_undefined = pl.status_and_mc_group_id.mc_group_undefined,
+ "McClassCSessionAns contains errors"
+ );
+
+ fuota_dev.error_msg = format!("Error: McClassCSessionAns response dr_error: {}, freq_error: {}, mc_group_undefined: {}",
+ pl.status_and_mc_group_id.dr_error,
+ pl.status_and_mc_group_id.freq_error,
+ pl.status_and_mc_group_id.mc_group_undefined,
+ );
+ } else {
+ info!(
+ time_to_start = pl.time_to_start.unwrap_or_default(),
+ "McClassCSessionAns OK"
+ );
+ fuota_dev.mc_session_completed_at = Some(Utc::now());
+ }
+
+ let _ = fuota::update_device(fuota_dev).await?;
+
+ Ok(())
+}
diff --git a/chirpstack/src/cmd/root.rs b/chirpstack/src/cmd/root.rs
index c4dc3d0c..e44bca17 100644
--- a/chirpstack/src/cmd/root.rs
+++ b/chirpstack/src/cmd/root.rs
@@ -5,7 +5,7 @@ use signal_hook_tokio::Signals;
use tracing::{info, warn};
use crate::gateway;
-use crate::{adr, api, backend, downlink, integration, region, storage};
+use crate::{adr, api, applayer::fuota, backend, downlink, integration, region, storage};
pub async fn run() -> Result<()> {
info!(
@@ -21,6 +21,7 @@ pub async fn run() -> Result<()> {
integration::setup().await?;
gateway::backend::setup().await?;
downlink::setup().await;
+ fuota::setup().await;
api::setup().await?;
let mut signals = Signals::new([SIGINT, SIGTERM]).unwrap();
diff --git a/chirpstack/src/storage/fields/fuota.rs b/chirpstack/src/storage/fields/fuota.rs
index 56c62691..d1cbe83c 100644
--- a/chirpstack/src/storage/fields/fuota.rs
+++ b/chirpstack/src/storage/fields/fuota.rs
@@ -91,6 +91,7 @@ pub enum FuotaJob {
FragSessionSetup,
Enqueue,
FragStatus,
+ Complete,
}
impl fmt::Display for FuotaJob {
@@ -110,6 +111,7 @@ impl From<&FuotaJob> for String {
FuotaJob::FragSessionSetup => "FRAG_SESSION_SETUP",
FuotaJob::Enqueue => "ENQUEUE",
FuotaJob::FragStatus => "FRAG_STATUS",
+ FuotaJob::Complete => "COMPLETE",
}
.to_string()
}
@@ -128,6 +130,7 @@ impl TryFrom<&str> for FuotaJob {
"FRAG_SESSION_SETUP" => Self::FragSessionSetup,
"ENQUEUE" => Self::Enqueue,
"FRAG_STATUS" => Self::FragStatus,
+ "COMPLETE" => Self::Complete,
_ => return Err(anyhow!("Invalid FuotaJob value: {}", value)),
})
}
diff --git a/chirpstack/src/storage/fields/mod.rs b/chirpstack/src/storage/fields/mod.rs
index 531915fa..77ff8a85 100644
--- a/chirpstack/src/storage/fields/mod.rs
+++ b/chirpstack/src/storage/fields/mod.rs
@@ -10,10 +10,7 @@ mod uuid;
pub use big_decimal::BigDecimal;
pub use dev_nonces::DevNonces;
-pub use device_profile::{
- AbpParams, AppLayerParams, ClassBParams, ClassCParams, RelayParams, Ts003Version, Ts004Version,
- Ts005Version,
-};
+pub use device_profile::{AbpParams, AppLayerParams, ClassBParams, ClassCParams, RelayParams};
pub use device_session::DeviceSession;
pub use fuota::{FuotaJob, RequestFragmentationSessionStatus};
pub use key_value::KeyValue;
diff --git a/chirpstack/src/storage/fuota.rs b/chirpstack/src/storage/fuota.rs
index 18966631..042c33fb 100644
--- a/chirpstack/src/storage/fuota.rs
+++ b/chirpstack/src/storage/fuota.rs
@@ -96,12 +96,12 @@ pub struct FuotaDeploymentDevice {
pub fuota_deployment_id: fields::Uuid,
pub dev_eui: EUI64,
pub created_at: DateTime<Utc>,
- pub updated_at: DateTime<Utc>,
+ pub completed_at: Option<DateTime<Utc>>,
pub mc_group_setup_completed_at: Option<DateTime<Utc>>,
pub mc_session_completed_at: Option<DateTime<Utc>>,
pub frag_session_setup_completed_at: Option<DateTime<Utc>>,
pub frag_status_completed_at: Option<DateTime<Utc>>,
- pub return_msg: String,
+ pub error_msg: String,
}
impl Default for FuotaDeploymentDevice {
@@ -112,12 +112,12 @@ impl Default for FuotaDeploymentDevice {
fuota_deployment_id: Uuid::nil().into(),
dev_eui: EUI64::default(),
created_at: now,
- updated_at: now,
+ completed_at: None,
mc_group_setup_completed_at: None,
mc_session_completed_at: None,
frag_session_setup_completed_at: None,
frag_status_completed_at: None,
- return_msg: "".into(),
+ error_msg: "".into(),
}
}
}
@@ -150,7 +150,7 @@ pub struct FuotaDeploymentJob {
pub max_retry_count: i16,
pub attempt_count: i16,
pub scheduler_run_after: DateTime<Utc>,
- pub return_msg: String,
+ pub error_msg: String,
}
impl Default for FuotaDeploymentJob {
@@ -165,7 +165,7 @@ impl Default for FuotaDeploymentJob {
max_retry_count: 0,
attempt_count: 0,
scheduler_run_after: now,
- return_msg: "".into(),
+ error_msg: "".into(),
}
}
}
@@ -351,12 +351,10 @@ pub async fn get_devices(
.map_err(|e| Error::from_diesel(e, "".into()))
}
-pub async fn get_device(
- fuota_deployment_id: Uuid,
- dev_eui: EUI64,
-) -> Result<FuotaDeploymentDevice, Error> {
+pub async fn get_latest_device_by_dev_eui(dev_eui: EUI64) -> Result<FuotaDeploymentDevice, Error> {
fuota_deployment_device::dsl::fuota_deployment_device
- .find((&fields::Uuid::from(fuota_deployment_id), &dev_eui))
+ .filter(fuota_deployment_device::dsl::dev_eui.eq(&dev_eui))
+ .order_by(fuota_deployment_device::created_at.desc())
.first(&mut get_async_db_conn().await?)
.await
.map_err(|e| Error::from_diesel(e, dev_eui.to_string()))
@@ -368,13 +366,13 @@ pub async fn update_device(d: FuotaDeploymentDevice) -> Result<FuotaDeploymentDe
.find((&d.fuota_deployment_id, &d.dev_eui)),
)
.set((
- fuota_deployment_device::updated_at.eq(Utc::now()),
+ fuota_deployment_device::completed_at.eq(&d.completed_at),
fuota_deployment_device::mc_group_setup_completed_at.eq(&d.mc_group_setup_completed_at),
fuota_deployment_device::mc_session_completed_at.eq(&d.mc_session_completed_at),
fuota_deployment_device::frag_session_setup_completed_at
.eq(&d.frag_session_setup_completed_at),
fuota_deployment_device::frag_status_completed_at.eq(&d.frag_status_completed_at),
- fuota_deployment_device::return_msg.eq(&d.return_msg),
+ fuota_deployment_device::error_msg.eq(&d.error_msg),
))
.get_result(&mut get_async_db_conn().await?)
.await
@@ -412,6 +410,91 @@ pub async fn get_device_count(fuota_deployment_id: Uuid) -> Result<i64, Error> {
.map_err(|e| Error::from_diesel(e, "".into()))
}
+pub async fn set_device_timeout_error(
+ fuota_deployment_id: Uuid,
+ mc_group_setup_timeout: bool,
+ mc_session_timeout: bool,
+ frag_session_setup_timeout: bool,
+ frag_status_timeout: bool,
+) -> Result<()> {
+ let fuota_deployment_id = fields::Uuid::from(fuota_deployment_id);
+
+ let mut error_msg = String::new();
+ if mc_group_setup_timeout {
+ error_msg = "McGroupSetupReq timeout.".into();
+ }
+ if mc_session_timeout {
+ error_msg = "McSessionReq timeout".into();
+ }
+ if frag_session_setup_timeout {
+ error_msg = "FragSessionSetupReq timeout.".into();
+ }
+ if frag_status_timeout {
+ error_msg = "FragStatusReq timeout.".into();
+ }
+
+ let mut q = diesel::update(fuota_deployment_device::table)
+ .set(fuota_deployment_device::dsl::error_msg.eq(&error_msg))
+ .filter(fuota_deployment_device::dsl::fuota_deployment_id.eq(&fuota_deployment_id))
+ .filter(fuota_deployment_device::dsl::error_msg.is_not_null())
+ .into_boxed();
+
+ if mc_group_setup_timeout {
+ q = q.filter(fuota_deployment_device::dsl::mc_group_setup_completed_at.is_null());
+ }
+
+ if mc_session_timeout {
+ q = q.filter(fuota_deployment_device::dsl::mc_session_completed_at.is_null());
+ }
+
+ if frag_session_setup_timeout {
+ q = q.filter(fuota_deployment_device::dsl::frag_session_setup_completed_at.is_null());
+ }
+
+ if frag_status_timeout {
+ q = q.filter(fuota_deployment_device::dsl::frag_status_completed_at.is_null());
+ }
+
+ q.execute(&mut get_async_db_conn().await?).await?;
+
+ Ok(())
+}
+
+pub async fn set_device_completed(
+ fuota_deployment_id: Uuid,
+ mc_group_setup_completed: bool,
+ mc_session_completed: bool,
+ frag_session_setup_completed: bool,
+ frag_status_completed: bool,
+) -> Result<()> {
+ let fuota_deployment_id = fields::Uuid::from(fuota_deployment_id);
+
+ let mut q = diesel::update(fuota_deployment_device::table)
+ .set(fuota_deployment_device::dsl::completed_at.eq(Some(Utc::now())))
+ .filter(fuota_deployment_device::dsl::fuota_deployment_id.eq(&fuota_deployment_id))
+ .into_boxed();
+
+ if mc_group_setup_completed {
+ q = q.filter(fuota_deployment_device::dsl::mc_group_setup_completed_at.is_not_null());
+ }
+
+ if mc_session_completed {
+ q = q.filter(fuota_deployment_device::dsl::mc_session_completed_at.is_not_null());
+ }
+
+ if frag_session_setup_completed {
+ q = q.filter(fuota_deployment_device::dsl::frag_session_setup_completed_at.is_not_null());
+ }
+
+ if frag_status_completed {
+ q = q.filter(fuota_deployment_device::dsl::frag_status_completed_at.is_not_null());
+ }
+
+ q.execute(&mut get_async_db_conn().await?).await?;
+
+ Ok(())
+}
+
pub async fn add_gateways(fuota_deployment_id: Uuid, gateway_ids: Vec<EUI64>) -> Result<(), Error> {
let mut errors = Vec::new();
@@ -439,7 +522,7 @@ pub async fn add_gateways(fuota_deployment_id: Uuid, gateway_ids: Vec<EUI64>) ->
let res = diesel::insert_into(fuota_deployment_gateway::table)
.values(&FuotaDeploymentGateway {
fuota_deployment_id: fuota_deployment_id.into(),
- gateway_id: gateway_id,
+ gateway_id,
..Default::default()
})
.execute(&mut get_async_db_conn().await?)
@@ -554,7 +637,7 @@ pub async fn update_job(j: FuotaDeploymentJob) -> Result<FuotaDeploymentJob, Err
fuota_deployment_job::completed_at.eq(&j.completed_at),
fuota_deployment_job::attempt_count.eq(&j.attempt_count),
fuota_deployment_job::scheduler_run_after.eq(&j.scheduler_run_after),
- fuota_deployment_job::return_msg.eq(&j.return_msg),
+ fuota_deployment_job::error_msg.eq(&j.error_msg),
))
.get_result(&mut get_async_db_conn().await?)
.await
@@ -650,7 +733,11 @@ pub async fn get_max_fragment_size(d: &FuotaDeployment) -> Result<usize> {
.n
- 3;
- Ok(max_pl_size)
+ Ok(if max_pl_size > d.payload.len() {
+ d.payload.len()
+ } else {
+ max_pl_size
+ })
}
pub fn get_multicast_timeout(d: &FuotaDeployment) -> Result<usize> {
@@ -878,10 +965,10 @@ mod test {
assert_eq!(d.id, devices[0].fuota_deployment_id);
// get device
- let mut fuota_d = get_device(d.id.into(), dev.dev_eui).await.unwrap();
- fuota_d.return_msg = "Error: kaboom".into();
- let fuota_d = update_device(fuota_d).await.unwrap();
- assert_eq!("Error: kaboom", fuota_d.return_msg);
+ let mut devices = get_devices(d.id.into(), 1, 0).await.unwrap();
+ devices[0].error_msg = "Error: kaboom".into();
+ let fuota_d = update_device(devices[0].clone()).await.unwrap();
+ assert_eq!("Error: kaboom", fuota_d.error_msg);
// remove devices
remove_devices(d.id.into(), vec![dev.dev_eui])
@@ -1095,6 +1182,7 @@ mod test {
device_profile_id: dp.id,
name: "test-fuota-deployment".into(),
multicast_dr: 5,
+ payload: vec![0; 1000],
..Default::default()
})
.await
diff --git a/chirpstack/src/storage/schema_postgres.rs b/chirpstack/src/storage/schema_postgres.rs
index 6cb64b6e..36df751c 100644
--- a/chirpstack/src/storage/schema_postgres.rs
+++ b/chirpstack/src/storage/schema_postgres.rs
@@ -221,12 +221,12 @@ diesel::table! {
fuota_deployment_id -> Uuid,
dev_eui -> Bytea,
created_at -> Timestamptz,
- updated_at -> Timestamptz,
+ completed_at -> Nullable<Timestamptz>,
mc_group_setup_completed_at -> Nullable<Timestamptz>,
mc_session_completed_at -> Nullable<Timestamptz>,
frag_session_setup_completed_at -> Nullable<Timestamptz>,
frag_status_completed_at -> Nullable<Timestamptz>,
- return_msg -> Text,
+ error_msg -> Text,
}
}
@@ -248,7 +248,7 @@ diesel::table! {
max_retry_count -> Int2,
attempt_count -> Int2,
scheduler_run_after -> Timestamptz,
- return_msg -> Text,
+ error_msg -> Text,
}
}
diff --git a/chirpstack/src/storage/schema_sqlite.rs b/chirpstack/src/storage/schema_sqlite.rs
index 6df4a2f0..72265e0b 100644
--- a/chirpstack/src/storage/schema_sqlite.rs
+++ b/chirpstack/src/storage/schema_sqlite.rs
@@ -197,12 +197,12 @@ diesel::table! {
fuota_deployment_id -> Text,
dev_eui -> Binary,
created_at -> TimestamptzSqlite,
- updated_at -> TimestamptzSqlite,
+ completed_at -> Nullable<TimestamptzSqlite>,
mc_group_setup_completed_at -> Nullable<TimestamptzSqlite>,
mc_session_completed_at -> Nullable<TimestamptzSqlite>,
frag_session_setup_completed_at -> Nullable<TimestamptzSqlite>,
frag_status_completed_at -> Nullable<TimestamptzSqlite>,
- return_msg -> Text,
+ error_msg -> Text,
}
}
@@ -223,7 +223,7 @@ diesel::table! {
max_retry_count -> SmallInt,
attempt_count -> SmallInt,
scheduler_run_after -> TimestamptzSqlite,
- return_msg -> Text,
+ error_msg -> Text,
}
}
diff --git a/ui/src/stores/FuotaStore.ts b/ui/src/stores/FuotaStore.ts
index dc3d77b4..6f8bff12 100644
--- a/ui/src/stores/FuotaStore.ts
+++ b/ui/src/stores/FuotaStore.ts
@@ -21,6 +21,8 @@ import type {
ListFuotaDeploymentGatewaysRequest,
ListFuotaDeploymentGatewaysResponse,
StartFuotaDeploymentRequest,
+ ListFuotaDeploymentJobsRequest,
+ ListFuotaDeploymentJobsResponse,
} from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
import SessionStore from "./SessionStore";
@@ -40,7 +42,7 @@ class FuotaStore extends EventEmitter {
callbackFunc: (resp: CreateFuotaDeploymentResponse) => void,
) => {
this.client.createDeployment(req, SessionStore.getMetadata(), (err, resp) => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -56,7 +58,7 @@ class FuotaStore extends EventEmitter {
getDeployment = (req: GetFuotaDeploymentRequest, callbackFunc: (resp: GetFuotaDeploymentResponse) => void) => {
this.client.getDeployment(req, SessionStore.getMetadata(), (err, resp) => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -67,7 +69,7 @@ class FuotaStore extends EventEmitter {
updateDeployment = (req: UpdateFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.updateDeployment(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -85,7 +87,7 @@ class FuotaStore extends EventEmitter {
deleteDeployment = (req: DeleteFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.deleteDeployment(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -101,7 +103,7 @@ class FuotaStore extends EventEmitter {
startDeployment = (req: StartFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.startDeployment(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -119,7 +121,7 @@ class FuotaStore extends EventEmitter {
listDeployments = (req: ListFuotaDeploymentsRequest, callbackFunc: (resp: ListFuotaDeploymentsResponse) => void) => {
this.client.listDeployments(req, SessionStore.getMetadata(), (err, resp) => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -130,7 +132,7 @@ class FuotaStore extends EventEmitter {
addDevices = (req: AddDevicesToFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.addDevices(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -146,7 +148,7 @@ class FuotaStore extends EventEmitter {
removeDevices = (req: RemoveDevicesFromFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.removeDevices(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -165,7 +167,7 @@ class FuotaStore extends EventEmitter {
callbackFunc: (resp: ListFuotaDeploymentDevicesResponse) => void,
) => {
this.client.listDevices(req, SessionStore.getMetadata(), (err, resp) => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -176,7 +178,7 @@ class FuotaStore extends EventEmitter {
addGateways = (req: AddGatewaysToFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.addGateways(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -192,7 +194,7 @@ class FuotaStore extends EventEmitter {
removeGateways = (req: RemoveGatewaysFromFuotaDeploymentRequest, callbackFunc: () => void) => {
this.client.removeGateways(req, SessionStore.getMetadata(), err => {
- if (err != null) {
+ if (err !== null) {
HandleError(err);
return;
}
@@ -211,7 +213,18 @@ class FuotaStore extends EventEmitter {
callbackFunc: (resp: ListFuotaDeploymentGatewaysResponse) => void,
) => {
this.client.listGateways(req, SessionStore.getMetadata(), (err, resp) => {
- if (err != null) {
+ if (err !== null) {
+ HandleError(err);
+ return;
+ }
+
+ callbackFunc(resp);
+ });
+ };
+
+ listJobs = (req: ListFuotaDeploymentJobsRequest, callbackFunc: (resp: ListFuotaDeploymentJobsResponse) => void) => {
+ this.client.listJobs(req, SessionStore.getMetadata(), (err, resp) => {
+ if (err !== null) {
HandleError(err);
return;
}
diff --git a/ui/src/views/device-profiles/CreateDeviceProfile.tsx b/ui/src/views/device-profiles/CreateDeviceProfile.tsx
index 9bdd47a0..562edba4 100644
--- a/ui/src/views/device-profiles/CreateDeviceProfile.tsx
+++ b/ui/src/views/device-profiles/CreateDeviceProfile.tsx
@@ -5,7 +5,11 @@ import { PageHeader } from "@ant-design/pro-layout";
import { MacVersion, RegParamsRevision } from "@chirpstack/chirpstack-api-grpc-web/common/common_pb";
import type { CreateDeviceProfileResponse } from "@chirpstack/chirpstack-api-grpc-web/api/device_profile_pb";
-import { DeviceProfile, CreateDeviceProfileRequest, AppLayerParams } from "@chirpstack/chirpstack-api-grpc-web/api/device_profile_pb";
+import {
+ DeviceProfile,
+ CreateDeviceProfileRequest,
+ AppLayerParams,
+} from "@chirpstack/chirpstack-api-grpc-web/api/device_profile_pb";
import type { Tenant } from "@chirpstack/chirpstack-api-grpc-web/api/tenant_pb";
diff --git a/ui/src/views/device-profiles/DeviceProfileForm.tsx b/ui/src/views/device-profiles/DeviceProfileForm.tsx
index 9fb9f294..59c589ae 100644
--- a/ui/src/views/device-profiles/DeviceProfileForm.tsx
+++ b/ui/src/views/device-profiles/DeviceProfileForm.tsx
@@ -1081,10 +1081,7 @@ function DeviceProfileForm(props: IProps) {
</Form.Item>
</Col>
<Col span={12}>
- <Form.Item
- label="Clock sync fPort (TS003)"
- name={["appLayerParams", "ts003FPort"]}
- >
+ <Form.Item label="Clock sync fPort (TS003)" name={["appLayerParams", "ts003FPort"]}>
<InputNumber min={0} max={255} disabled={props.disabled} />
</Form.Item>
</Col>
@@ -1103,10 +1100,7 @@ function DeviceProfileForm(props: IProps) {
</Form.Item>
</Col>
<Col span={12}>
- <Form.Item
- label="Fragmented data block transport fPort (TS004)"
- name={["appLayerParams", "ts004FPort"]}
- >
+ <Form.Item label="Fragmented data block transport fPort (TS004)" name={["appLayerParams", "ts004FPort"]}>
<InputNumber min={0} max={255} disabled={props.disabled} />
</Form.Item>
</Col>
@@ -1125,10 +1119,7 @@ function DeviceProfileForm(props: IProps) {
</Form.Item>
</Col>
<Col span={12}>
- <Form.Item
- label="Remote multicast setup fPort (TS005)"
- name={["appLayerParams", "ts005FPort"]}
- >
+ <Form.Item label="Remote multicast setup fPort (TS005)" name={["appLayerParams", "ts005FPort"]}>
<InputNumber min={0} max={255} disabled={props.disabled} />
</Form.Item>
</Col>
diff --git a/ui/src/views/devices/ListDevices.tsx b/ui/src/views/devices/ListDevices.tsx
index 585b3fd1..5f0b7541 100644
--- a/ui/src/views/devices/ListDevices.tsx
+++ b/ui/src/views/devices/ListDevices.tsx
@@ -36,7 +36,7 @@ import {
} from "@chirpstack/chirpstack-api-grpc-web/api/multicast_group_pb";
import type { ListRelaysResponse, RelayListItem } from "@chirpstack/chirpstack-api-grpc-web/api/relay_pb";
import { ListRelaysRequest, AddRelayDeviceRequest } from "@chirpstack/chirpstack-api-grpc-web/api/relay_pb";
-import {
+import type {
ListFuotaDeploymentsResponse,
FuotaDeploymentListItem,
} from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
diff --git a/ui/src/views/fuota/FuotaDeploymentDashboard.tsx b/ui/src/views/fuota/FuotaDeploymentDashboard.tsx
new file mode 100644
index 00000000..5039adb1
--- /dev/null
+++ b/ui/src/views/fuota/FuotaDeploymentDashboard.tsx
@@ -0,0 +1,128 @@
+import { useState, useEffect } from "react";
+
+import { Spin, Button, Space, Timeline, Row, Col, TimelineProps, Card, Tag, Popover, Table } from "antd";
+import { LoadingOutlined, ReloadOutlined } from "@ant-design/icons";
+import type { ColumnsType } from "antd/es/table";
+import { format } from "date-fns";
+
+import { format_dt, format_dt_from_secs } from "../helpers";
+
+import { ListFuotaDeploymentJobsRequest } from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
+
+import {
+ GetFuotaDeploymentResponse,
+ ListFuotaDeploymentJobsResponse,
+ FuotaDeploymentJob,
+} from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
+
+import FuotaStore from "../../stores/FuotaStore";
+
+interface IProps {
+ getFuotaDeploymentResponse: GetFuotaDeploymentResponse;
+}
+
+function FuotaDeploymentDashboard(props: IProps) {
+ const [fuotaJobs, setFuotaJobs] = useState<FuotaDeploymentJob.AsObject[]>([]);
+
+ useEffect(() => {
+ getFuotaJobs();
+
+ const interval = setInterval(() => {
+ if (!props.getFuotaDeploymentResponse.getCompletedAt()) {
+ getFuotaJobs();
+ }
+ }, 10000);
+
+ return () => clearInterval(interval);
+ }, [props.getFuotaDeploymentResponse]);
+
+ const jobs: Record<string, string> = {
+ CREATE_MC_GROUP: "Create multicast group",
+ ADD_DEVS_TO_MC_GROUP: "Add devices to multicast group",
+ ADD_GWS_TO_MC_GROUP: "Add gateways to multicast group",
+ MC_GROUP_SETUP: "Multicast group setup",
+ FRAG_SESSION_SETUP: "Fragmentation session setup",
+ MC_SESSION: "Multicast session setup",
+ ENQUEUE: "Enqueue fragments",
+ FRAG_STATUS: "Request fragmentation status",
+ COMPLETE: "Complete deployment",
+ };
+
+ const columns: ColumnsType<FuotaDeploymentJob.AsObject> = [
+ {
+ title: "Status",
+ key: "status",
+ width: 100,
+ render: (_text, record) => {
+ if (record.errorMsg !== "") {
+ return (
+ <Popover content={record.errorMsg} placement="right">
+ <Tag color="red">error</Tag>
+ </Popover>
+ );
+ } else if (!record.completedAt) {
+ return <Spin indicator={<LoadingOutlined spin />} size="small" />;
+ } else {
+ return <Tag color="green">ok</Tag>;
+ }
+ },
+ },
+ {
+ title: "Job",
+ dataIndex: "job",
+ key: "job",
+ render: text => jobs[text],
+ width: 250,
+ },
+ {
+ title: "Created at",
+ dataIndex: "createdAt",
+ key: "createdAt",
+ render: (_text, record) => format_dt_from_secs(record.createdAt?.seconds),
+ width: 250,
+ },
+ {
+ title: "Completed at",
+ dataIndex: "completedAt",
+ key: "completedAt",
+ render: (_text, record) => format_dt_from_secs(record.completedAt?.seconds),
+ width: 250,
+ },
+ {
+ title: "Attempt count",
+ dataIndex: "attemptCount",
+ key: "attemptCount",
+ width: 150,
+ },
+ {
+ title: "Max. retry",
+ dataIndex: "maxRetryCount",
+ key: "maxRetryCount",
+ width: 150,
+ },
+ ];
+
+ const getFuotaJobs = () => {
+ const req = new ListFuotaDeploymentJobsRequest();
+ req.setFuotaDeploymentId(props.getFuotaDeploymentResponse.getDeployment()!.getId());
+ FuotaStore.listJobs(req, (resp: ListFuotaDeploymentJobsResponse) => {
+ const obj = resp.toObject();
+ setFuotaJobs(obj.jobsList);
+ });
+ };
+
+ let loadingProps = undefined;
+ if (props.getFuotaDeploymentResponse.getStartedAt() && fuotaJobs.length === 0) {
+ loadingProps = {
+ delay: 300,
+ };
+ }
+
+ return (
+ <Space style={{ width: "100%" }} direction="vertical">
+ <Table loading={loadingProps} dataSource={fuotaJobs} columns={columns} pagination={false} />
+ </Space>
+ );
+}
+
+export default FuotaDeploymentDashboard;
diff --git a/ui/src/views/fuota/FuotaDeploymentDevices.tsx b/ui/src/views/fuota/FuotaDeploymentDevices.tsx
index a7e95977..246e8bbb 100644
--- a/ui/src/views/fuota/FuotaDeploymentDevices.tsx
+++ b/ui/src/views/fuota/FuotaDeploymentDevices.tsx
@@ -1,14 +1,16 @@
-import { useState } from "react";
+import { useState, useEffect } from "react";
-import { Space, Button, Popconfirm } from "antd";
+import { Tag, Space, Button, Popconfirm, Spin, Typography, Popover } from "antd";
+import { LoadingOutlined, ZoomInOutlined } from "@ant-design/icons";
import type { ColumnsType } from "antd/es/table";
+import { format } from "date-fns";
import {
ListFuotaDeploymentDevicesRequest,
RemoveDevicesFromFuotaDeploymentRequest,
} from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
import type {
- FuotaDeployment,
+ GetFuotaDeploymentResponse,
ListFuotaDeploymentDevicesResponse,
FuotaDeploymentDeviceListItem,
} from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
@@ -18,19 +20,94 @@ import DataTable from "../../components/DataTable";
import FuotaStore from "../../stores/FuotaStore";
interface IProps {
- fuotaDeployment: FuotaDeployment;
+ getFuotaDeploymentResponse: GetFuotaDeploymentResponse;
}
function FuotaDeploymentDevices(props: IProps) {
const [selectedRowIds, setSelectedRowIds] = useState<string[]>([]);
const [refreshKey, setRefreshKey] = useState<number>(0);
+ useEffect(() => {
+ const interval = setInterval(() => {
+ if (!props.getFuotaDeploymentResponse.getCompletedAt()) {
+ setRefreshKey(refreshKey + 1);
+ }
+ }, 10000);
+
+ return () => clearInterval(interval);
+ }, [props.getFuotaDeploymentResponse, refreshKey]);
+
const columns: ColumnsType<FuotaDeploymentDeviceListItem.AsObject> = [
+ {
+ title: "Status",
+ key: "status",
+ width: 100,
+ render: (_text, record) => {
+ if (record.errorMsg !== "") {
+ return (
+ <Popover content={record.errorMsg} placement="right">
+ <Tag color="red">error</Tag>
+ </Popover>
+ );
+ } else if (record.completedAt) {
+ return <Tag color="green">ok</Tag>;
+ } else if (props.getFuotaDeploymentResponse.getStartedAt()) {
+ return <Spin indicator={<LoadingOutlined spin />} size="small" />;
+ } else {
+ return "";
+ }
+ },
+ },
{
title: "DevEUI",
dataIndex: "devEui",
key: "devEui",
- // width: 250,
+ width: 250,
+ render: text => <Typography.Text code>{text}</Typography.Text>,
+ },
+ {
+ title: "Mc. group setup completed at",
+ key: "mcGroupSetupCompletedAt",
+ render: (_text, record) => {
+ if (record.mcGroupSetupCompletedAt !== undefined) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(record.mcGroupSetupCompletedAt.seconds);
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ }
+ },
+ },
+ {
+ title: "Frag. session setup completed at",
+ key: "fragSessionSetupCompletedAt",
+ render: (_text, record) => {
+ if (record.fragSessionSetupCompletedAt !== undefined) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(record.fragSessionSetupCompletedAt.seconds);
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ }
+ },
+ },
+ {
+ title: "Mc. session completed at",
+ key: "mcSessionCompletedAt",
+ render: (_text, record) => {
+ if (record.mcSessionCompletedAt !== undefined) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(record.mcSessionCompletedAt.seconds);
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ }
+ },
+ },
+ {
+ title: "Frag. status completed at",
+ key: "fragStatusCompletedAt",
+ render: (_text, record) => {
+ if (record.fragStatusCompletedAt !== undefined) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(record.fragStatusCompletedAt.seconds);
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ }
+ },
},
];
@@ -47,7 +124,7 @@ function FuotaDeploymentDevices(props: IProps) {
callbackFunc: GetPageCallbackFunc,
) => {
const req = new ListFuotaDeploymentDevicesRequest();
- req.setFuotaDeploymentId(props.fuotaDeployment.getId());
+ req.setFuotaDeploymentId(props.getFuotaDeploymentResponse.getDeployment()!.getId());
req.setLimit(limit);
req.setOffset(offset);
@@ -59,7 +136,7 @@ function FuotaDeploymentDevices(props: IProps) {
const removeDevices = () => {
const req = new RemoveDevicesFromFuotaDeploymentRequest();
- req.setFuotaDeploymentId(props.fuotaDeployment.getId());
+ req.setFuotaDeploymentId(props.getFuotaDeploymentResponse.getDeployment()!.getId());
req.setDevEuisList(selectedRowIds);
FuotaStore.removeDevices(req, () => {
diff --git a/ui/src/views/fuota/FuotaDeploymentForm.tsx b/ui/src/views/fuota/FuotaDeploymentForm.tsx
index 2036eec4..00fd85f2 100644
--- a/ui/src/views/fuota/FuotaDeploymentForm.tsx
+++ b/ui/src/views/fuota/FuotaDeploymentForm.tsx
@@ -4,7 +4,7 @@ import { Form, Input, InputNumber, Select, Row, Col, Button, Upload, UploadFile,
import { UploadOutlined } from "@ant-design/icons";
import type { Tenant } from "@chirpstack/chirpstack-api-grpc-web/api/tenant_pb";
-import { FuotaDeployment } from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
+import { FuotaDeployment, RequestFragmentationSessionStatus } from "@chirpstack/chirpstack-api-grpc-web/api/fuota_pb";
import type {
ListDeviceProfilesResponse,
GetDeviceProfileResponse,
@@ -72,6 +72,7 @@ function FuotaDeploymentForm(props: IProps) {
d.setMulticastDr(v.multicastDr);
d.setMulticastFrequency(v.multicastFrequency);
d.setFragmentationRedundancyPercentage(v.fragmentationRedundancyPercentage);
+ d.setRequestFragmentationSessionStatus(v.requestFragmentationSessionStatus);
d.setCalculateMulticastTimeout(v.calculateMulticastTimeout);
d.setMulticastTimeout(v.multicastTimeout);
d.setCalculateFragmentationFragmentSize(v.calculateFragmentationFragmentSize);
@@ -244,6 +245,25 @@ function FuotaDeploymentForm(props: IProps) {
</Form.Item>
</Col>
</Row>
+ <Row gutter={24}>
+ <Col span={8}>
+ <Form.Item
+ label="Fragmentation status request"
+ name="requestFragmentationSessionStatus"
+ tooltip="After fragment enqueue is recommended for Class-A devices, after session timeout is recommended for Class-B / Class-C devices."
+ >
+ <Select disabled={props.disabled}>
+ <Select.Option value={RequestFragmentationSessionStatus.NO_REQUEST}>Do not request</Select.Option>
+ <Select.Option value={RequestFragmentationSessionStatus.AFTER_FRAGMENT_ENQUEUE}>
+ After fragment enqueue
+ </Select.Option>
+ <Select.Option value={RequestFragmentationSessionStatus.AFTER_SESSION_TIMEOUT}>
+ After session timeout
+ </Select.Option>
+ </Select>
+ </Form.Item>
+ </Col>
+ </Row>
<Row gutter={24}>
<Col span={6}>
<Form.Item
diff --git a/ui/src/views/fuota/FuotaDeploymentGateways.tsx b/ui/src/views/fuota/FuotaDeploymentGateways.tsx
index ae0b343f..aa0b7391 100644
--- a/ui/src/views/fuota/FuotaDeploymentGateways.tsx
+++ b/ui/src/views/fuota/FuotaDeploymentGateways.tsx
@@ -1,6 +1,6 @@
import { useState } from "react";
-import { Space, Button, Popconfirm } from "antd";
+import { Space, Button, Popconfirm, Typography } from "antd";
import type { ColumnsType } from "antd/es/table";
import {
@@ -30,7 +30,7 @@ function FuotaDeploymentGateways(props: IProps) {
title: "Gateway ID",
dataIndex: "gatewayId",
key: "gatewayId",
- // width: 250,
+ render: text => <Typography.Text code>{text}</Typography.Text>,
},
];
diff --git a/ui/src/views/fuota/FuotaDeploymentLayout.tsx b/ui/src/views/fuota/FuotaDeploymentLayout.tsx
index e7e57d4e..8f04ad8e 100644
--- a/ui/src/views/fuota/FuotaDeploymentLayout.tsx
+++ b/ui/src/views/fuota/FuotaDeploymentLayout.tsx
@@ -21,6 +21,7 @@ import { useTitle } from "../helpers";
import EditFuotaDeployment from "./EditFuotaDeployment";
import FuotaDeploymentDevices from "./FuotaDeploymentDevices";
import FuotaDeploymentGateways from "./FuotaDeploymentGateways";
+import FuotaDeploymentDashboard from "./FuotaDeploymentDashboard";
interface IProps {
tenant: Tenant;
@@ -202,6 +203,10 @@ function FuotaDeploymentLayout(props: IProps) {
</Menu.Item>
</Menu>
<Routes>
+ <Route
+ path="/"
+ element={<FuotaDeploymentDashboard getFuotaDeploymentResponse={getFuotaDeploymentResponse} />}
+ />
<Route
path="/edit"
element={
@@ -213,7 +218,10 @@ function FuotaDeploymentLayout(props: IProps) {
/>
}
/>
- <Route path="/devices" element={<FuotaDeploymentDevices fuotaDeployment={d} />} />
+ <Route
+ path="/devices"
+ element={<FuotaDeploymentDevices getFuotaDeploymentResponse={getFuotaDeploymentResponse} />}
+ />
<Route path="/gateways" element={<FuotaDeploymentGateways fuotaDeployment={d} />} />
</Routes>
</Card>
diff --git a/ui/src/views/gateways/ListGateways.tsx b/ui/src/views/gateways/ListGateways.tsx
index e2e82d25..4ae75ed8 100644
--- a/ui/src/views/gateways/ListGateways.tsx
+++ b/ui/src/views/gateways/ListGateways.tsx
@@ -142,7 +142,7 @@ function ListGateways(props: IProps) {
req.setTenantId(props.tenant.getId());
let mgGroups: MulticastGroup[] = [];
- let fuotaDeployments: FuotaDeployment[] = [];
+ let fDeployments: FuotaDeployment[] = [];
ApplicationStore.list(req, (resp: ListApplicationsResponse) => {
for (const app of resp.getResultList()) {
@@ -169,7 +169,7 @@ function ListGateways(props: IProps) {
fuotaReq.setLimit(999);
fuotaReq.setApplicationId(app.getId());
FuotaStore.listDeployments(fuotaReq, (resp: ListFuotaDeploymentsResponse) => {
- fuotaDeployments.push({
+ fDeployments.push({
title: app.getName(),
value: "",
disabled: true,
@@ -181,7 +181,7 @@ function ListGateways(props: IProps) {
// The above can also be done using setFuotaDeployments and a callback
// function, but this introduces a race-condition when executed twice.
- setFuotaDeployments(fuotaDeployments);
+ setFuotaDeployments(fDeployments);
});
}
});
diff --git a/ui/src/views/helpers.ts b/ui/src/views/helpers.ts
index 81387716..f60375b9 100644
--- a/ui/src/views/helpers.ts
+++ b/ui/src/views/helpers.ts
@@ -1,4 +1,6 @@
import { notification } from "antd";
+import { format } from "date-fns";
+import * as google_protobuf_timestamp_pb from "google-protobuf/google/protobuf/timestamp_pb";
import { MacVersion, RegParamsRevision } from "@chirpstack/chirpstack-api-grpc-web/common/common_pb";
import { useRef, useEffect } from "react";
@@ -87,3 +89,23 @@ export function useTitle(...v: unknown[]) {
};
}, [documentDefined, v]);
}
+
+export function format_dt(dt?: google_protobuf_timestamp_pb.Timestamp): string {
+ if (dt) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(dt.getSeconds());
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ } else {
+ return "";
+ }
+}
+
+export function format_dt_from_secs(secs?: number): string {
+ if (secs) {
+ const ts = new Date(0);
+ ts.setUTCSeconds(secs);
+ return format(ts, "yyyy-MM-dd HH:mm:ss");
+ } else {
+ return "";
+ }
+}