Skip to content

Commit 7580885

Browse files
authored
Merge pull request #4338 from rina23q/feat/agent-service-commands
feat: expands workflow engine to support service commands
2 parents c53fcb4 + 5536b80 commit 7580885

45 files changed

Lines changed: 2457 additions & 213 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Cargo.lock‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/common/tedge_config/src/tedge_toml/tedge_config.rs‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ use std::net::Ipv4Addr;
6767
use std::num::NonZeroU16;
6868
use std::path::PathBuf;
6969
use std::sync::Arc;
70+
use tedge_api::file_transfer_url::EntityStoreUrls;
7071
use tedge_api::file_transfer_url::FileTransferUrls;
7172
use tedge_api::file_transfer_url::Protocol;
7273
use tedge_api::mqtt_topics::EntityTopicId;
@@ -1917,13 +1918,23 @@ impl TEdgeConfigReaderHttp {
19171918
/// Builds the URLs at which the File Transfer Service HTTP server exposes files,
19181919
/// as seen by a client connecting to `http.client.host:http.client.port`.
19191920
pub fn file_transfer_urls(&self) -> FileTransferUrls {
1920-
let authority: Arc<str> = format!("{}:{}", self.client.host, self.client.port).into();
1921+
let (authority, protocol) = self.client_authority();
1922+
FileTransferUrls::new(authority, protocol)
1923+
}
1924+
1925+
pub fn entity_store_urls(&self) -> EntityStoreUrls {
1926+
let (authority, protocol) = self.client_authority();
1927+
EntityStoreUrls::new(authority, protocol)
1928+
}
1929+
1930+
fn client_authority(&self) -> (Arc<str>, Protocol) {
1931+
let authority = format!("{}:{}", self.client.host, self.client.port).into();
19211932
let protocol = if self.is_secure() {
19221933
Protocol::Https
19231934
} else {
19241935
Protocol::Http
19251936
};
1926-
FileTransferUrls::new(authority, protocol)
1937+
(authority, protocol)
19271938
}
19281939
}
19291940

‎crates/core/tedge/src/supervisor.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ use tedge_config::TEdgeConfig;
1010
use tedge_mapper::MapperName;
1111
use tedge_supervisor::RuntimeFactory;
1212
use tedge_supervisor::Supervisor;
13+
use tedge_supervisor::SupervisorMode;
1314
use tedge_supervisor::Unit;
1415
use tedge_supervisor::UnitKind;
1516

@@ -86,7 +87,7 @@ pub async fn run(opt: RunAllOpt) -> anyhow::Result<()> {
8687
let mapper = mapper.clone();
8788
async move {
8889
let config = TEdgeConfig::load(&config_dir).await?;
89-
tedge_mapper::build(mapper, config).await
90+
tedge_mapper::build(mapper, config, SupervisorMode::MultiUnit).await
9091
}
9192
.boxed()
9293
});

‎crates/core/tedge_agent/Cargo.toml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ tedge_config_manager = { workspace = true }
3838
tedge_downloader_ext = { workspace = true }
3939
tedge_file_system_ext = { workspace = true }
4040
tedge_health_ext = { workspace = true }
41+
tedge_http_ext = { workspace = true }
4142
tedge_log_manager = { workspace = true }
4243
tedge_mqtt_ext = { workspace = true }
4344
tedge_script_ext = { workspace = true }
@@ -60,10 +61,12 @@ proptest = { workspace = true }
6061
rcgen = { workspace = true }
6162
ron = { workspace = true }
6263
tedge_actors = { workspace = true, features = ["test-helpers"] }
64+
tedge_http_ext = { workspace = true, features = ["test_helpers"] }
6365
tedge_mqtt_ext = { workspace = true, features = ["test-helpers"] }
6466
tedge_test_utils = { workspace = true }
6567
tempfile = { workspace = true }
6668
test-case = { workspace = true }
69+
tokio = { workspace = true, features = ["test-util"] }
6770
tower = { workspace = true }
6871

6972
[lints]

‎crates/core/tedge_agent/src/agent.rs‎

Lines changed: 78 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@ use crate::entity_manager::server::EntityStoreServer;
55
use crate::entity_manager::server::EntityStoreServerConfig;
66
use crate::http_server::actor::HttpServerBuilder;
77
use crate::http_server::actor::HttpServerConfig;
8+
use crate::operation_workflows::install_service_workflows;
9+
use crate::operation_workflows::EntityStoreClient;
810
use crate::operation_workflows::OperationConfig;
911
use crate::operation_workflows::WorkflowActorBuilder;
1012
use crate::restart_manager::builder::RestartManagerBuilder;
@@ -35,6 +37,7 @@ use tedge_actors::Sequential;
3537
use tedge_actors::ServerActorBuilder;
3638
use tedge_actors::ServerConfig;
3739
use tedge_api::entity_store::EntityRegistrationMessage;
40+
use tedge_api::file_transfer_url::EntityStoreUrls;
3841
use tedge_api::file_transfer_url::FileTransferUrls;
3942
use tedge_api::mqtt_topics::DeviceTopicId;
4043
use tedge_api::mqtt_topics::EntityTopicId;
@@ -43,6 +46,8 @@ use tedge_api::mqtt_topics::OperationType;
4346
use tedge_api::mqtt_topics::Service;
4447
use tedge_api::mqtt_topics::ServiceTopicId;
4548
use tedge_api::path::DataDir;
49+
use tedge_api::service_command::ActionCapabilities;
50+
use tedge_api::service_command::TEDGE_SERVICE_ACTIONS;
4651
use tedge_api::EntityStore;
4752
use tedge_config::tedge_toml::TEdgeConfigReaderService;
4853
use tedge_config_ext::ConfigPublisherBuilder;
@@ -52,6 +57,7 @@ use tedge_config_manager::ConfigManagerOptions;
5257
use tedge_downloader_ext::DownloaderActor;
5358
use tedge_file_system_ext::FsWatchActorBuilder;
5459
use tedge_health_ext::HealthMonitorBuilder;
60+
use tedge_http_ext::HttpActor;
5561
use tedge_log_manager::LogManagerBuilder;
5662
use tedge_log_manager::LogManagerConfig;
5763
use tedge_log_manager::LogManagerOptions;
@@ -85,6 +91,8 @@ pub(crate) struct AgentConfig {
8591
pub service_topic_id: ServiceTopicId,
8692
pub mqtt_topic_root: Arc<str>,
8793
pub file_transfer_urls: FileTransferUrls,
94+
pub entity_store_urls: EntityStoreUrls,
95+
pub http_client_tls_config: rustls::ClientConfig,
8896
pub service: TEdgeConfigReaderService,
8997
pub identity: Option<Identity>,
9098
pub cloud_root_certs: CloudHttpConfig,
@@ -139,6 +147,8 @@ impl AgentConfig {
139147
.with_session_name(mqtt_session_name);
140148

141149
let file_transfer_urls = tedge_config.http.file_transfer_urls();
150+
let entity_store_urls = tedge_config.http.entity_store_urls();
151+
let http_client_tls_config = tedge_config.http.client_tls_config()?;
142152

143153
// HTTP config
144154
let data_dir = tedge_config.data_root();
@@ -221,6 +231,8 @@ impl AgentConfig {
221231
mqtt_device_topic_id,
222232
service_topic_id,
223233
file_transfer_urls,
234+
entity_store_urls,
235+
http_client_tls_config,
224236
identity,
225237
cloud_root_certs,
226238
is_sudo_enabled,
@@ -281,6 +293,8 @@ impl Agent {
281293
ShellExecuteBuilder::try_new(&self.config.operations_dir).await?;
282294
}
283295

296+
install_service_workflows(&self.config.operations_dir).await?;
297+
284298
// Inotify actor
285299
let mut fs_watch_actor_builder = FsWatchActorBuilder::new();
286300

@@ -301,6 +315,67 @@ impl Agent {
301315
let mut uploader_actor_builder =
302316
UploaderActor::new(self.config.identity, self.config.cloud_root_certs).builder();
303317

318+
let device_topic_id = self.config.mqtt_device_topic_id.clone();
319+
let mqtt_schema = MqttSchema::with_root(self.config.mqtt_topic_root.to_string());
320+
321+
// TODO: replace with a call to entity store when we stop assuming default MQTT schema
322+
let is_main_device = device_topic_id == EntityTopicId::default_main_device();
323+
let entity_store = if is_main_device {
324+
info!("Running as a main device, starting File Transfer Service");
325+
326+
let state_dir = agent_state_dir(&self.config.state_dir, &self.config.config_dir);
327+
let state_dir = state_dir.path();
328+
let clean_start = self.config.entity_store_clean_start;
329+
let telemetry_cache_size = 0; // Agent need not cache any data messages, the mapper would
330+
331+
let main_device = EntityRegistrationMessage::main_device(None);
332+
let entity_store = EntityStore::with_main_device(
333+
mqtt_schema.clone(),
334+
main_device,
335+
telemetry_cache_size,
336+
state_dir,
337+
clean_start,
338+
)?;
339+
let entity_store_server_config =
340+
EntityStoreServerConfig::new(mqtt_schema.clone(), self.config.entity_auto_register);
341+
let entity_store_server = EntityStoreServer::new(
342+
entity_store_server_config,
343+
entity_store,
344+
&mut mqtt_actor_builder,
345+
);
346+
let mut entity_store_actor_builder =
347+
ServerActorBuilder::new(entity_store_server, &ServerConfig::default(), Sequential);
348+
mqtt_actor_builder.connect_mapped_sink(
349+
entity_manager::server::subscriptions(&mqtt_schema),
350+
&entity_store_actor_builder,
351+
|message| {
352+
Some(RequestEnvelope {
353+
request: EntityStoreRequest::MqttMessage(message),
354+
reply_to: Box::new(NullSender),
355+
})
356+
},
357+
);
358+
359+
let entity_store = EntityStoreClient::local(&mut entity_store_actor_builder);
360+
361+
let file_transfer_server_builder = HttpServerBuilder::try_bind(
362+
self.config.http_config,
363+
&mut entity_store_actor_builder,
364+
)
365+
.await?;
366+
367+
runtime.spawn(file_transfer_server_builder).await?;
368+
runtime.spawn(entity_store_actor_builder).await?;
369+
entity_store
370+
} else {
371+
info!("Running as a child device: File Transfer Service disabled");
372+
let mut http_actor = HttpActor::new(self.config.http_client_tls_config).builder();
373+
let entity_store =
374+
EntityStoreClient::remote(self.config.entity_store_urls, &mut http_actor);
375+
runtime.spawn(http_actor).await?;
376+
entity_store
377+
};
378+
304379
// Software update actor
305380
let mut software_update_builder = SoftwareManagerBuilder::new(self.config.sw_update_config);
306381

@@ -312,19 +387,18 @@ impl Agent {
312387
&mut fs_watch_actor_builder,
313388
&mut downloader_actor_builder,
314389
&mut uploader_actor_builder,
390+
entity_store,
315391
);
316392
workflow_actor_builder.register_builtin_operation(&mut restart_actor_builder);
317393
workflow_actor_builder.register_builtin_operation(&mut software_update_builder);
318394

319395
// Health actor
320396
// TODO: take a user-configurable service topic id
321-
let device_topic_id = self.config.mqtt_device_topic_id.clone();
322397
let service_topic_id = self.config.service_topic_id.clone();
323398
let service = Service {
324399
service_topic_id: service_topic_id.clone(),
325400
device_topic_id: DeviceTopicId::new(device_topic_id.clone()),
326401
};
327-
let mqtt_schema = MqttSchema::with_root(self.config.mqtt_topic_root.to_string());
328402

329403
let twin_manager_config = TwinManagerConfig::new(
330404
self.config.config_dir.root().to_path_buf(),
@@ -347,7 +421,8 @@ impl Agent {
347421
&mut mqtt_actor_builder,
348422
&mqtt_schema,
349423
&self.config.service,
350-
);
424+
)
425+
.with_action_capabilities(ActionCapabilities::declaring(TEDGE_SERVICE_ACTIONS));
351426

352427
// Instantiate config manager actor if either config_snapshot or config_update operation is enabled
353428
let config_actor_builder: Option<ConfigManagerBuilder> =
@@ -417,56 +492,6 @@ impl Agent {
417492
None
418493
};
419494

420-
// TODO: replace with a call to entity store when we stop assuming default MQTT schema
421-
let is_main_device = device_topic_id == EntityTopicId::default_main_device();
422-
if is_main_device {
423-
info!("Running as a main device, starting File Transfer Service");
424-
425-
let state_dir = agent_state_dir(&self.config.state_dir, &self.config.config_dir);
426-
let state_dir = state_dir.path();
427-
let clean_start = self.config.entity_store_clean_start;
428-
let telemetry_cache_size = 0; // Agent need not cache any data messages, the mapper would
429-
430-
let main_device = EntityRegistrationMessage::main_device(None);
431-
let entity_store = EntityStore::with_main_device(
432-
mqtt_schema.clone(),
433-
main_device,
434-
telemetry_cache_size,
435-
state_dir,
436-
clean_start,
437-
)?;
438-
let entity_store_server_config =
439-
EntityStoreServerConfig::new(mqtt_schema.clone(), self.config.entity_auto_register);
440-
let entity_store_server = EntityStoreServer::new(
441-
entity_store_server_config,
442-
entity_store,
443-
&mut mqtt_actor_builder,
444-
);
445-
let mut entity_store_actor_builder =
446-
ServerActorBuilder::new(entity_store_server, &ServerConfig::default(), Sequential);
447-
mqtt_actor_builder.connect_mapped_sink(
448-
entity_manager::server::subscriptions(&mqtt_schema),
449-
&entity_store_actor_builder,
450-
|message| {
451-
Some(RequestEnvelope {
452-
request: EntityStoreRequest::MqttMessage(message),
453-
reply_to: Box::new(NullSender),
454-
})
455-
},
456-
);
457-
458-
let file_transfer_server_builder = HttpServerBuilder::try_bind(
459-
self.config.http_config,
460-
&mut entity_store_actor_builder,
461-
)
462-
.await?;
463-
464-
runtime.spawn(file_transfer_server_builder).await?;
465-
runtime.spawn(entity_store_actor_builder).await?;
466-
} else {
467-
info!("Running as a child device: File Transfer Service disabled");
468-
}
469-
470495
// Spawn all
471496
runtime.spawn(mqtt_actor_builder).await?;
472497
runtime.spawn(fs_watch_actor_builder).await?;

0 commit comments

Comments
 (0)