Merge pull request #137 from chrivers/chrivers/service-instancing

Rework `svc` ("service") crate, to support templated services.

These are analogous to systemd template service, in which a service name can contain `@` to indicate it is a template.

In this way, instead of manually registering `foo-1`, `foo-2`, etc,
we can register `foo@`, and then start instances `1`, `2`, etc.

This makes service management cleaner and simpler, and provides a way to recognize groups
of related services.

Author: chrivers

crates/bifrost-api/src/service.rs

@@ -1,16 +1,18 @@
 use std::collections::BTreeMap;
 
 use serde::{Deserialize, Serialize};
-use svc::traits::ServiceState;
 use uuid::Uuid;
 
+use svc::serviceid::ServiceName;
+use svc::traits::ServiceState;
+
 use crate::Client;
 use crate::error::BifrostResult;
 
 #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
 pub struct Service {
     pub id: Uuid,
-    pub name: String,
+    pub name: ServiceName,
     pub state: ServiceState,
 }
 
@@ -24,12 +26,12 @@ impl Client {
         self.get("service").await
     }
 
-    pub async fn service_stop(&self, id: Uuid) -> BifrostResult<()> {
+    pub async fn service_stop(&self, id: Uuid) -> BifrostResult<Uuid> {
         self.put(&format!("service/{id}"), ServiceState::Stopped)
             .await
     }
 
-    pub async fn service_start(&self, id: Uuid) -> BifrostResult<()> {
+    pub async fn service_start(&self, id: Uuid) -> BifrostResult<Uuid> {
         self.put(&format!("service/{id}"), ServiceState::Running)
             .await
     }

crates/svc/src/error.rs

@@ -3,7 +3,7 @@ use std::error::Error;
 use thiserror::Error;
 
 use crate::manager::{ServiceEvent, SvmRequest};
-use crate::serviceid::ServiceId;
+use crate::serviceid::{ServiceId, ServiceName};
 use crate::traits::ServiceState;
 
 #[derive(Error, Debug)]
@@ -43,13 +43,22 @@ pub enum SvcError {
     ServiceNotFound(ServiceId),
 
     #[error("Service {0} already exists")]
-    ServiceAlreadyExists(String),
+    ServiceAlreadyExists(ServiceName),
 
     #[error("All services stopped")]
     Shutdown,
 
     #[error("Service has failed")]
     ServiceFailed,
+
+    #[error("Templated service generation failed")]
+    ServiceGeneration(Box<dyn Error + Send>),
+}
+
+impl SvcError {
+    pub fn generation(err: impl Error + Send + 'static) -> Self {
+        Self::ServiceGeneration(Box::new(err))
+    }
 }
 
 #[derive(Error, Debug)]

crates/svc/src/lib.rs

@@ -10,3 +10,5 @@ pub mod manager;
 pub mod rpc;
 #[cfg(feature = "manager")]
 pub mod runservice;
+#[cfg(feature = "manager")]
+pub mod template;

crates/svc/src/manager.rs

@@ -15,13 +15,14 @@ use uuid::Uuid;
 use crate::error::{RunSvcError, SvcError, SvcResult};
 use crate::rpc::RpcRequest;
 use crate::runservice::StandardService;
-use crate::serviceid::{IntoServiceId, ServiceId};
+use crate::serviceid::{IntoServiceId, ServiceId, ServiceName};
+use crate::template::ServiceTemplate;
 use crate::traits::{Service, ServiceRunner, ServiceState};
 
 #[derive(Debug)]
 pub struct ServiceInstance {
     tx: watch::Sender<ServiceState>,
-    name: String,
+    name: ServiceName,
     state: ServiceState,
     abort_handle: AbortHandle,
 }
@@ -60,13 +61,14 @@ impl ServiceEvent {
 
 /// A request to a [`ServiceManager`]
 pub enum SvmRequest {
-    Stop(RpcRequest<ServiceId, SvcResult<()>>),
-    Start(RpcRequest<ServiceId, SvcResult<()>>),
+    Stop(RpcRequest<ServiceId, SvcResult<Uuid>>),
+    Start(RpcRequest<ServiceId, SvcResult<Uuid>>),
     Status(RpcRequest<ServiceId, SvcResult<ServiceState>>),
-    List(RpcRequest<(), Vec<(Uuid, String)>>),
+    List(RpcRequest<(), Vec<(Uuid, ServiceName)>>),
     Resolve(RpcRequest<ServiceId, SvcResult<Uuid>>),
-    LookupName(RpcRequest<ServiceId, SvcResult<String>>),
+    LookupName(RpcRequest<ServiceId, SvcResult<ServiceName>>),
     Register(RpcRequest<(String, ServiceFunc), SvcResult<Uuid>>),
+    RegisterTemplate(RpcRequest<(String, Box<dyn ServiceTemplate>), SvcResult<()>>),
     Subscribe(RpcRequest<mpsc::UnboundedSender<ServiceEvent>, SvcResult<Uuid>>),
     Shutdown(RpcRequest<(), ()>),
 }
@@ -128,19 +130,29 @@ impl SvmClient {
         .await?
     }
 
-    pub async fn start(&mut self, id: impl IntoServiceId) -> SvcResult<()> {
+    pub async fn register_template(
+        &mut self,
+        name: impl AsRef<str>,
+        generator: impl ServiceTemplate + 'static,
+    ) -> SvcResult<()> {
+        let name = name.as_ref().to_string();
+        self.rpc(SvmRequest::RegisterTemplate, (name, Box::new(generator)))
+            .await?
+    }
+
+    pub async fn start(&mut self, id: impl IntoServiceId) -> SvcResult<Uuid> {
         self.rpc(SvmRequest::Start, id.service_id()).await?
     }
 
-    pub async fn stop(&mut self, id: impl IntoServiceId) -> SvcResult<()> {
+    pub async fn stop(&mut self, id: impl IntoServiceId) -> SvcResult<Uuid> {
         self.rpc(SvmRequest::Stop, id.service_id()).await?
     }
 
     pub async fn resolve(&mut self, id: impl IntoServiceId) -> SvcResult<Uuid> {
         self.rpc(SvmRequest::Resolve, id.service_id()).await?
     }
 
-    pub async fn lookup_name(&mut self, id: impl IntoServiceId) -> SvcResult<String> {
+    pub async fn lookup_name(&mut self, id: impl IntoServiceId) -> SvcResult<ServiceName> {
         self.rpc(SvmRequest::LookupName, id.service_id()).await?
     }
 
@@ -197,7 +209,7 @@ impl SvmClient {
         self.rpc(SvmRequest::Status, id.service_id()).await?
     }
 
-    pub async fn list(&mut self) -> SvcResult<Vec<(Uuid, String)>> {
+    pub async fn list(&mut self) -> SvcResult<Vec<(Uuid, ServiceName)>> {
         self.rpc(SvmRequest::List, ()).await
     }
 
@@ -214,6 +226,10 @@ impl Debug for SvmRequest {
             Self::Status(arg0) => f.debug_tuple("Status").field(arg0).finish(),
             Self::List(arg0) => f.debug_tuple("List").field(arg0).finish(),
             Self::Register(_arg0) => f.debug_tuple("Register").field(&"<service>").finish(),
+            Self::RegisterTemplate(_arg0) => f
+                .debug_tuple("RegisterTemplate")
+                .field(&"<service>")
+                .finish(),
             Self::Resolve(arg0) => f.debug_tuple("Resolve").field(arg0).finish(),
             Self::LookupName(arg0) => f.debug_tuple("ResolveName").field(arg0).finish(),
             Self::Subscribe(_arg0) => f.debug_tuple("Subscribe").finish(),
@@ -229,8 +245,9 @@ pub struct ServiceManager {
     service_tx: mpsc::UnboundedSender<ServiceEvent>,
     subscribers: BTreeMap<Uuid, mpsc::UnboundedSender<ServiceEvent>>,
     svcs: BTreeMap<Uuid, ServiceInstance>,
-    names: BTreeMap<String, Uuid>,
+    names: BTreeMap<ServiceName, Uuid>,
     tasks: JoinSet<Result<(), RunSvcError>>,
+    templates: BTreeMap<String, Box<dyn ServiceTemplate>>,
     shutdown: bool,
 }
 
@@ -254,6 +271,7 @@ impl ServiceManager {
             svcs: BTreeMap::new(),
             names: BTreeMap::new(),
             tasks: JoinSet::new(),
+            templates: BTreeMap::new(),
             shutdown: false,
         }
     }
@@ -284,8 +302,7 @@ impl ServiceManager {
         self.control_tx.clone()
     }
 
-    fn register(&mut self, name: &str, svc: ServiceFunc) -> SvcResult<Uuid> {
-        let name = name.to_string();
+    fn register(&mut self, name: ServiceName, svc: ServiceFunc) -> SvcResult<Uuid> {
         if self.names.contains_key(&name) {
             return Err(SvcError::ServiceAlreadyExists(name));
         }
@@ -297,7 +314,7 @@ impl ServiceManager {
 
         let rec = ServiceInstance {
             tx,
-            name: name.to_string(),
+            name: name.clone(),
             state: ServiceState::Registered,
             abort_handle,
         };
@@ -317,7 +334,7 @@ impl ServiceManager {
         match &id {
             ServiceId::Name(name) => self
                 .names
-                .get(name.as_str())
+                .get(name)
                 .ok_or_else(|| SvcError::ServiceNotFound(id))
                 .copied(),
             ServiceId::Id(uuid) => {
@@ -351,23 +368,49 @@ impl ServiceManager {
         Ok(&self.svcs[&id])
     }
 
-    fn start(&self, id: impl IntoServiceId) -> SvcResult<()> {
-        self.get(&id).and_then(|svc| {
+    fn start(&mut self, id: impl IntoServiceId) -> SvcResult<Uuid> {
+        let id = id.service_id();
+
+        // if the service is known, attempt to start it
+        if let Ok(svc) = self.get(&id) {
             log::debug!("Starting service: {id} {}", &svc.name);
-            Ok(svc.tx.send(ServiceState::Running)?)
-        })
+            svc.tx.send(ServiceState::Running)?;
+            return self.resolve(&id);
+        }
+
+        // ..else, check if it's a named instance
+        let ServiceId::Name(svc_name) = &id else {
+            return Err(SvcError::ServiceNotFound(id));
+        };
+
+        let Some(inst) = svc_name.instance() else {
+            return Err(SvcError::ServiceNotFound(id));
+        };
+
+        let Some(tmpl) = &self.templates.get(svc_name.name()) else {
+            return Err(SvcError::ServiceNotFound(id));
+        };
+
+        let inner = tmpl.generate(inst.to_string())?;
+        let svc = StandardService::new(svc_name.name(), inner);
+
+        let uuid = self.register(svc_name.clone(), svc.boxed())?;
+
+        Ok(uuid)
     }
 
-    fn stop(&self, id: impl IntoServiceId) -> SvcResult<()> {
+    fn stop(&self, id: impl IntoServiceId) -> SvcResult<Uuid> {
         let id = self.resolve(id)?;
 
         if self.svcs[&id].state == ServiceState::Stopped {
-            return Ok(());
+            return Ok(id);
         }
 
         log::debug!("Stopping service: {id} {}", self.svcs[&id].name);
         self.get(id)
-            .and_then(|svc| Ok(svc.tx.send(ServiceState::Stopped)?))
+            .and_then(|svc| Ok(svc.tx.send(ServiceState::Stopped)?))?;
+
+        Ok(id)
     }
 
     fn notify_subscribers(&mut self, event: ServiceEvent) {
@@ -412,12 +455,19 @@ impl ServiceManager {
                 let mut res = vec![];
 
                 for (name, id) in &self.names {
-                    res.push((*id, name.to_string()));
+                    res.push((*id, name.clone()));
                 }
                 res
             }),
 
-            SvmRequest::Register(rpc) => rpc.respond(|(name, svc)| self.register(&name, svc)),
+            SvmRequest::Register(rpc) => {
+                rpc.respond(|(name, svc)| self.register(ServiceName::from(name), svc));
+            }
+
+            SvmRequest::RegisterTemplate(rpc) => rpc.respond(|(name, tmpl)| {
+                self.templates.insert(name, tmpl);
+                Ok(())
+            }),
 
             SvmRequest::Resolve(rpc) => rpc.respond(|id| self.resolve(&id)),
 

crates/svc/src/policy.rs

@@ -4,12 +4,14 @@ use std::time::Duration;
 #[cfg(feature = "manager")]
 use tokio::time::sleep;
 
+#[derive(Debug, Clone, Copy)]
 pub enum Retry {
     No,
     Limit(u32),
     Forever,
 }
 
+#[derive(Debug, Clone, Copy)]
 pub struct Policy {
     pub retry: Retry,
     pub delay: Option<Duration>,

crates/svc/src/runservice.rs

@@ -5,9 +5,9 @@ use tokio::time::sleep;
 use uuid::Uuid;
 
 use crate::error::RunSvcError;
-use crate::manager::ServiceEvent;
+use crate::manager::{ServiceEvent, ServiceFunc};
 use crate::policy::{Policy, Retry};
-use crate::traits::{Service, ServiceRunner, ServiceState};
+use crate::traits::{Service, ServiceRunner, ServiceState, StopResult};
 
 #[allow(clippy::struct_field_names)]
 struct State {
@@ -101,6 +101,12 @@ impl<S: Service> StandardService<S> {
     }
 }
 
+impl<S: Service + 'static> StandardService<S> {
+    pub fn boxed(self) -> ServiceFunc {
+        Box::new(|a, b, c| self.run(a, b, c))
+    }
+}
+
 #[allow(clippy::too_many_lines)]
 #[async_trait]
 impl<S: Service> ServiceRunner for StandardService<S> {
@@ -192,23 +198,27 @@ impl<S: Service> ServiceRunner for StandardService<S> {
                             }
                         },
                         _ = rx.changed() => if *rx.borrow() == ServiceState::Stopped {
-                            if S::SIGNAL_STOP {
-                                log::trace!(target:target, "Service state change requested (graceful)");
-                                svc.signal_stop().await.map_err(|e| RunSvcError::ServiceError(Box::new(e)))?;
-                                tokio::select! {
-                                    res = svc.run() => {
-                                        log::trace!(target:target, "Service finished running within timeout: {res:?}");
-                                    },
-                                    () = sleep(Duration::from_secs(1)) => {
-                                        log::warn!("timeout");
+                            log::trace!(target:target, "Stopping service");
+                            let stop = svc.signal_stop().await.map_err(|e| RunSvcError::ServiceError(Box::new(e)))?;
+                            match stop {
+                                StopResult::Delivered => {
+                                    log::trace!(target:target, "Service state change requested (graceful)");
+                                    tokio::select! {
+                                        res = svc.run() => {
+                                            log::trace!(target:target, "Service finished running within timeout: {res:?}");
+                                        },
+                                        () = sleep(Duration::from_secs(1)) => {
+                                            log::warn!("timeout");
+                                        }
                                     }
-                                }
-                                state.set(ServiceState::Stopping)?;
-                            } else {
-                                log::trace!(target:target, "Service state change requested: {:?} -> {:?}", state.get(), *rx.borrow());
-                                if *rx.borrow_and_update() == ServiceState::Stopped {
                                     state.set(ServiceState::Stopping)?;
                                 }
+                                StopResult::NotSupported => {
+                                    log::trace!(target:target, "Service state change requested: {:?} -> {:?}", state.get(), *rx.borrow());
+                                    if *rx.borrow_and_update() == ServiceState::Stopped {
+                                        state.set(ServiceState::Stopping)?;
+                                    }
+                                }
                             }
                         }
                     }