Monitor child CA versions and send version to parent/repo (#622, #620)

src/cli/client.rs

@@ -9,8 +9,8 @@ use rpki::uri;
 use crate::cli::options::KrillUserDetails;
 use crate::cli::report::{ApiResponse, ReportError};
 use crate::commons::api::{
-    AllCertAuthIssues, CaRepoDetails, CertAuthIssues, ChildCaInfo, ParentCaContact, ParentStatuses, PublisherDetails,
-    PublisherList, RepoStatus, Token,
+    AllCertAuthIssues, CaRepoDetails, CertAuthIssues, ChildCaInfo, ChildrenConnectionStats, ParentCaContact,
+    ParentStatuses, PublisherDetails, PublisherList, RepoStatus, Token,
 };
 use crate::commons::bgp::BgpAnalysisAdvice;
 use crate::commons::remote::rfc8183;
@@ -258,6 +258,11 @@ impl KrillClient {
                 delete(&self.server, &self.token, &uri).await?;
                 Ok(ApiResponse::Empty)
             }
+            CaCommand::ChildConnections(handle) => {
+                let uri = format!("api/v1/cas/{}/stats/children/connections", handle);
+                let stats: ChildrenConnectionStats = get_json(&self.server, &self.token, &uri).await?;
+                Ok(ApiResponse::ChildrenStats(stats))
+            }
 
             CaCommand::KeyRollInit(handle) => {
                 let uri = format!("api/v1/cas/{}/keys/roll_init", handle);

src/cli/options.rs

@@ -9,9 +9,9 @@ use std::{env, fmt};
 use bytes::Bytes;
 use clap::{App, Arg, ArgMatches, SubCommand};
 
-use rpki::uri;
 use rpki::repository::crypto::KeyIdentifier;
 use rpki::repository::x509::Time;
+use rpki::uri;
 
 use crate::commons::crypto::{IdCert, SignSupport};
 use crate::commons::remote::rfc8183;
@@ -469,6 +469,15 @@ impl Options {
         app.subcommand(sub)
     }
 
+    fn make_cas_children_connections_sc<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> {
+        let mut sub = SubCommand::with_name("connections").about("Show connections stats for children of a CA");
+
+        sub = Self::add_general_args(sub);
+        sub = Self::add_my_ca_arg(sub);
+
+        app.subcommand(sub)
+    }
+
     fn make_cas_children_sc<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> {
         let mut sub = SubCommand::with_name("children").about("Manage children for a CA");
 
@@ -477,6 +486,7 @@ impl Options {
         sub = Self::make_cas_children_info_sc(sub);
         sub = Self::make_cas_children_remove_sc(sub);
         sub = Self::make_cas_children_response_sc(sub);
+        sub = Self::make_cas_children_connections_sc(sub);
 
         app.subcommand(sub)
     }
@@ -1409,6 +1419,14 @@ impl Options {
         Ok(Options::make(general_args, command))
     }
 
+    fn parse_matches_cas_children_connections(matches: &ArgMatches) -> Result<Options, Error> {
+        let general_args = GeneralArgs::from_matches(matches)?;
+        let my_ca = Self::parse_my_ca(matches)?;
+
+        let command = Command::CertAuth(CaCommand::ChildConnections(my_ca));
+        Ok(Options::make(general_args, command))
+    }
+
     fn parse_matches_cas_children(matches: &ArgMatches) -> Result<Options, Error> {
         if let Some(m) = matches.subcommand_matches("add") {
             Self::parse_matches_cas_children_add(m)
@@ -1420,6 +1438,8 @@ impl Options {
             Self::parse_matches_cas_children_update(m)
         } else if let Some(m) = matches.subcommand_matches("remove") {
             Self::parse_matches_cas_children_remove(m)
+        } else if let Some(m) = matches.subcommand_matches("connections") {
+            Self::parse_matches_cas_children_connections(m)
         } else {
             Err(Error::UnrecognizedSubCommand)
         }
@@ -2087,6 +2107,7 @@ pub enum CaCommand {
     ChildAdd(Handle, AddChildRequest),
     ChildUpdate(Handle, ChildHandle, UpdateChildRequest),
     ChildDelete(Handle, ChildHandle),
+    ChildConnections(Handle),
 
     // Key Management
     KeyRollInit(Handle),

src/cli/report.rs

@@ -5,8 +5,8 @@ use serde::Serialize;
 
 use crate::commons::api::{
     AllCertAuthIssues, CaCommandDetails, CaRepoDetails, CertAuthInfo, CertAuthIssues, CertAuthList, ChildCaInfo,
-    CommandHistory, ParentCaContact, ParentStatuses, PublisherDetails, PublisherList, RepoStatus, RoaDefinitions,
-    RtaList, RtaPrepResponse, ServerInfo,
+    ChildrenConnectionStats, CommandHistory, ParentCaContact, ParentStatuses, PublisherDetails, PublisherList,
+    RepoStatus, RoaDefinitions, RtaList, RtaPrepResponse, ServerInfo,
 };
 use crate::commons::bgp::{BgpAnalysisAdvice, BgpAnalysisReport, BgpAnalysisSuggestion};
 use crate::commons::remote::api::ClientInfos;
@@ -36,6 +36,7 @@ pub enum ApiResponse {
     ParentStatuses(ParentStatuses),
 
     ChildInfo(ChildCaInfo),
+    ChildrenStats(ChildrenConnectionStats),
 
     PublisherDetails(PublisherDetails),
     PublisherList(PublisherList),
@@ -81,6 +82,7 @@ impl ApiResponse {
                 ApiResponse::ParentCaContact(contact) => Ok(Some(contact.report(fmt)?)),
                 ApiResponse::ParentStatuses(statuses) => Ok(Some(statuses.report(fmt)?)),
                 ApiResponse::ChildInfo(info) => Ok(Some(info.report(fmt)?)),
+                ApiResponse::ChildrenStats(stats) => Ok(Some(stats.report(fmt)?)),
                 ApiResponse::PublisherList(list) => Ok(Some(list.report(fmt)?)),
                 ApiResponse::PublisherDetails(details) => Ok(Some(details.report(fmt)?)),
                 ApiResponse::RepoStats(stats) => Ok(Some(stats.report(fmt)?)),
@@ -177,6 +179,7 @@ impl Report for CaCommandDetails {}
 impl Report for PublisherList {}
 
 impl Report for RepoStats {}
+impl Report for ChildrenConnectionStats {}
 
 impl Report for PublisherDetails {}
 

src/commons/api/ca.rs

@@ -1355,7 +1355,7 @@ impl ParentStatus {
         self.last_exchange = Some(ParentExchange {
             timestamp: Time::now().timestamp(),
             uri,
-            result: ParentExchangeResult::Failure(error),
+            result: ExchangeResult::Failure(error),
         });
         self.set_next_exchange_plus_seconds(next_seconds);
     }
@@ -1386,7 +1386,7 @@ impl ParentStatus {
         self.last_exchange = Some(ParentExchange {
             timestamp: Time::now().timestamp(),
             uri,
-            result: ParentExchangeResult::Success,
+            result: ExchangeResult::Success,
         });
         self.set_next_exchange_plus_seconds(next_run_seconds);
     }
@@ -1443,7 +1443,7 @@ impl RepoStatus {
         self.last_exchange = Some(ParentExchange {
             timestamp: Time::now().timestamp(),
             uri,
-            result: ParentExchangeResult::Failure(error),
+            result: ExchangeResult::Failure(error),
         });
         self.next_exchange_before = (Time::now() + Duration::minutes(5)).timestamp();
     }
@@ -1452,7 +1452,7 @@ impl RepoStatus {
         self.last_exchange = Some(ParentExchange {
             timestamp: Time::now().timestamp(),
             uri,
-            result: ParentExchangeResult::Success,
+            result: ExchangeResult::Success,
         });
         self.published = published;
         self.next_exchange_before = Self::now_plus_hours(next_hours);
@@ -1462,7 +1462,7 @@ impl RepoStatus {
         self.last_exchange = Some(ParentExchange {
             timestamp: Time::now().timestamp(),
             uri,
-            result: ParentExchangeResult::Success,
+            result: ExchangeResult::Success,
         });
         self.next_exchange_before = Self::now_plus_hours(next_hours);
     }
@@ -1491,7 +1491,7 @@ impl fmt::Display for RepoStatus {
 pub struct ParentExchange {
     timestamp: i64,
     uri: ServiceUri,
-    result: ParentExchangeResult,
+    result: ExchangeResult,
 }
 
 impl ParentExchange {
@@ -1503,41 +1503,139 @@ impl ParentExchange {
         &self.uri
     }
 
-    pub fn result(&self) -> &ParentExchangeResult {
+    pub fn result(&self) -> &ExchangeResult {
         &self.result
     }
 
     pub fn was_success(&self) -> bool {
         match &self.result {
-            ParentExchangeResult::Success => true,
-            ParentExchangeResult::Failure(_) => false,
+            ExchangeResult::Success => true,
+            ExchangeResult::Failure(_) => false,
         }
     }
 
     pub fn into_failure_opt(self) -> Option<ErrorResponse> {
         match self.result {
-            ParentExchangeResult::Success => None,
-            ParentExchangeResult::Failure(error) => Some(error),
+            ExchangeResult::Success => None,
+            ExchangeResult::Failure(error) => Some(error),
         }
     }
 }
 
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
 #[allow(clippy::large_enum_variant)]
-pub enum ParentExchangeResult {
+pub enum ExchangeResult {
     Success,
     Failure(ErrorResponse),
 }
 
-impl fmt::Display for ParentExchangeResult {
+impl fmt::Display for ExchangeResult {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
-            ParentExchangeResult::Success => write!(f, "success"),
-            ParentExchangeResult::Failure(e) => write!(f, "failure: {}", e.msg()),
+            ExchangeResult::Success => write!(f, "success"),
+            ExchangeResult::Failure(e) => write!(f, "failure: {}", e.msg()),
         }
     }
 }
 
+//------------ ChildConnectionStats ------------------------------------------
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ChildrenConnectionStats {
+    children: Vec<ChildConnectionStats>,
+}
+
+impl ChildrenConnectionStats {
+    pub fn new(children: Vec<ChildConnectionStats>) -> Self {
+        ChildrenConnectionStats { children }
+    }
+}
+
+impl fmt::Display for ChildrenConnectionStats {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        if !self.children.is_empty() {
+            writeln!(f, "handle, user_agent, last_exchange, result")?;
+            for child in &self.children {
+                match &child.last_exchange {
+                    None => {
+                        writeln!(f, "{},n/a,never,n/a", child.handle)?;
+                    }
+                    Some(exchange) => {
+                        let agent = exchange.user_agent.as_deref().unwrap_or("");
+                        let time = Time::new(Utc.timestamp(exchange.timestamp, 0));
+
+                        writeln!(
+                            f,
+                            "{},{},{},{}",
+                            child.handle,
+                            agent,
+                            time.to_rfc3339(),
+                            exchange.result
+                        )?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ChildConnectionStats {
+    handle: ChildHandle,
+    last_exchange: Option<ChildExchange>,
+}
+
+impl ChildConnectionStats {
+    pub fn new(handle: ChildHandle, last_exchange: Option<ChildExchange>) -> Self {
+        ChildConnectionStats { handle, last_exchange }
+    }
+}
+
+//------------ ChildStatus ---------------------------------------------------
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ChildStatus {
+    last_exchange: Option<ChildExchange>,
+}
+
+impl ChildStatus {
+    pub fn set_success(&mut self, user_agent: Option<String>) {
+        self.last_exchange = Some(ChildExchange {
+            timestamp: Time::now().timestamp(),
+            result: ExchangeResult::Success,
+            user_agent,
+        });
+    }
+
+    pub fn set_failure(&mut self, user_agent: Option<String>, error_response: ErrorResponse) {
+        self.last_exchange = Some(ChildExchange {
+            timestamp: Time::now().timestamp(),
+            result: ExchangeResult::Failure(error_response),
+            user_agent,
+        });
+    }
+}
+
+impl Default for ChildStatus {
+    fn default() -> Self {
+        ChildStatus { last_exchange: None }
+    }
+}
+
+impl From<ChildStatus> for Option<ChildExchange> {
+    fn from(status: ChildStatus) -> Self {
+        status.last_exchange
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ChildExchange {
+    timestamp: i64,
+    result: ExchangeResult,
+    user_agent: Option<String>,
+}
+
 //------------ CertAuthInfo --------------------------------------------------
 
 /// This type represents the details of a CertAuth that need

src/commons/util/httpclient.rs

@@ -12,7 +12,7 @@ use serde::Serialize;
 
 use crate::commons::api::{ErrorResponse, Token};
 use crate::commons::util::file;
-use crate::constants::{HTTP_CLIENT_TIMEOUT_SECS, KRILL_CLI_API_ENV, KRILL_HTTPS_ROOT_CERTS_ENV};
+use crate::constants::{HTTP_CLIENT_TIMEOUT_SECS, KRILL_CLI_API_ENV, KRILL_HTTPS_ROOT_CERTS_ENV, KRILL_VERSION};
 
 const JSON_CONTENT: &str = "application/json";
 
@@ -184,17 +184,22 @@ pub async fn post_empty(uri: &str, token: Option<&Token>) -> Result<(), Error> {
     }
 }
 
-/// Posts binary data, and expects a binary response.
+/// Posts binary data, and expects a binary response. Includes the full krill version
+/// as the user agent. Intended for sending RFC 6492 (provisioning) and 8181 (publication)
+/// to the trusted parent or publication server.
 ///
 /// Note: Bytes may be empty if the post was successful, but the response was
 /// empty.
-pub async fn post_binary(uri: &str, data: &Bytes, content_type: &str) -> Result<Bytes, Error> {
+pub async fn post_binary_with_full_ua(uri: &str, data: &Bytes, content_type: &str) -> Result<Bytes, Error> {
     let body = data.to_vec();
     if env::var(KRILL_CLI_API_ENV).is_ok() {
         report_post_and_exit(uri, None, None, PostBody::Bytes(&body));
     }
 
-    let headers = headers(Some(content_type), None)?;
+    let mut headers = HeaderMap::new();
+    headers.insert(USER_AGENT, HeaderValue::from_str(&format!("krill/{}", KRILL_VERSION))?);
+    headers.insert(CONTENT_TYPE, HeaderValue::from_str(content_type)?);
+
     let res = client(uri).await?.post(uri).headers(headers).body(body).send().await?;
 
     match res.status() {

src/constants.rs

@@ -76,6 +76,7 @@ pub const ID_CERTIFICATE_VALIDITY_YEARS: i32 = 15;
 pub const BGP_RIS_REFRESH_MINUTES: i64 = 60;
 
 pub const HTTP_CLIENT_TIMEOUT_SECS: u64 = 120;
+pub const HTTP_USER_AGENT_TRUNCATE: usize = 256; // Will truncate received user-agent values at this size.
 pub const OPENID_CONNECT_HTTP_CLIENT_TIMEOUT_SECS: u64 = 30;
 
 pub const NO_RESOURCE: NoResourceType = NoResourceType;

src/daemon/ca/manager.rs

@@ -15,7 +15,6 @@ use rpki::uri;
 use crate::{
     commons::{
         actor::Actor,
-        api::rrdp::PublishElement,
         api::{
             self, AddChildRequest, Base64, CaCommandDetails, CaCommandResult, CertAuthList, CertAuthSummary,
             ChildCaInfo, ChildHandle, CommandHistory, CommandHistoryCriteria, Entitlements, Handle, IssuanceRequest,
@@ -23,6 +22,7 @@ use crate::{
             RcvdCert, RepoStatus, RepositoryContact, ResourceClassName, ResourceSet, RevocationRequest,
             RevocationResponse, RtaName, StoredEffect, UpdateChildRequest,
         },
+        api::{rrdp::PublishElement, ChildrenConnectionStats},
         crypto::{IdCert, KrillSigner, ProtocolCms, ProtocolCmsBuilder},
         error::Error,
         eventsourcing::{Aggregate, AggregateStore, Command, CommandKey},
@@ -445,6 +445,11 @@ impl CaManager {
         ca.get_child(child).map(|details| details.clone().into())
     }
 
+    /// Show the (connection) stats for children under a CA.
+    pub async fn ca_stats_child_connections(&self, ca: &Handle) -> KrillResult<ChildrenConnectionStats> {
+        self.status_store.lock().await.get_stats_child_connections(ca).await
+    }
+
     /// Show a contact for a child.
     pub async fn ca_parent_contact(
         &self,
@@ -515,7 +520,13 @@ impl CaManager {
     /// - validates the request
     /// - processes the child request
     /// - signs a response and returns the bytes
-    pub async fn rfc6492(&self, ca_handle: &Handle, msg_bytes: Bytes, actor: &Actor) -> KrillResult<Bytes> {
+    pub async fn rfc6492(
+        &self,
+        ca_handle: &Handle,
+        msg_bytes: Bytes,
+        user_agent: Option<String>,
+        actor: &Actor,
+    ) -> KrillResult<Bytes> {
         let ca = self.get_ca(ca_handle).await?;
 
         let msg = match ProtocolCms::decode(msg_bytes.as_ref(), false) {
@@ -540,22 +551,23 @@ impl CaManager {
         let (res, should_log_cms) = match content {
             rfc6492::Content::Qry(rfc6492::Qry::Revoke(req)) => {
                 let res = self.revoke(ca_handle, child.clone(), req, actor).await?;
-                let msg = rfc6492::Message::revoke_response(child, recipient, res);
+                let msg = rfc6492::Message::revoke_response(child.clone(), recipient, res);
                 (self.wrap_rfc6492_response(ca_handle, msg).await, true)
             }
             rfc6492::Content::Qry(rfc6492::Qry::List) => {
                 let entitlements = self.list(ca_handle, &child).await?;
-                let msg = rfc6492::Message::list_response(child, recipient, entitlements);
+                let msg = rfc6492::Message::list_response(child.clone(), recipient, entitlements);
                 (self.wrap_rfc6492_response(ca_handle, msg).await, false)
             }
             rfc6492::Content::Qry(rfc6492::Qry::Issue(req)) => {
                 let res = self.issue(ca_handle, &child, req, actor).await?;
-                let msg = rfc6492::Message::issue_response(child, recipient, res);
+                let msg = rfc6492::Message::issue_response(child.clone(), recipient, res);
                 (self.wrap_rfc6492_response(ca_handle, msg).await, true)
             }
             _ => (Err(Error::custom("Unsupported RFC6492 message")), true),
         };
 
+        // Log CMS messages if needed, and if enabled by config (this is a no-op if it isn't)
         match &res {
             Ok(reply_bytes) => {
                 if should_log_cms {
@@ -569,6 +581,24 @@ impl CaManager {
             }
         }
 
+        // Set child status
+        match &res {
+            Ok(_) => {
+                self.status_store
+                    .lock()
+                    .await
+                    .set_child_success(ca.handle(), &child, user_agent)
+                    .await?;
+            }
+            Err(e) => {
+                self.status_store
+                    .lock()
+                    .await
+                    .set_child_failure(ca.handle(), &child, user_agent, e)
+                    .await?;
+            }
+        }
+
         res
     }
 
@@ -1589,7 +1619,7 @@ impl CaManager {
 
         let uri = service_uri.to_string();
 
-        let res = httpclient::post_binary(&uri, &signed_msg, content_type)
+        let res = httpclient::post_binary_with_full_ua(&uri, &signed_msg, content_type)
             .await
             .map_err(Error::HttpClientError)?;
 

src/daemon/ca/status.rs

@@ -1,15 +1,18 @@
-use std::path::Path;
+use std::{collections::HashMap, path::Path};
 
 use tokio::sync::RwLock;
 
-use crate::commons::api::{
-    rrdp::PublishElement, Entitlements, ErrorResponse, Handle, ParentHandle, ParentStatuses, RepoStatus,
+use crate::commons::{
+    api::{
+        rrdp::PublishElement, ChildConnectionStats, ChildHandle, ChildStatus, ChildrenConnectionStats, Entitlements,
+        ErrorResponse, Handle, ParentHandle, ParentStatuses, RepoStatus,
+    },
+    error::Error,
+    eventsourcing::{KeyStoreKey, KeyValueStore},
+    remote::rfc8183::ServiceUri,
+    util::httpclient,
+    KrillResult,
 };
-use crate::commons::error::Error;
-use crate::commons::eventsourcing::{KeyStoreKey, KeyValueStore};
-use crate::commons::remote::rfc8183::ServiceUri;
-use crate::commons::util::httpclient;
-use crate::commons::KrillResult;
 
 //------------ CaStatus ------------------------------------------------------
 
@@ -17,6 +20,20 @@ use crate::commons::KrillResult;
 struct CaStatus {
     repo: RepoStatus,
     parents: ParentStatuses,
+    #[serde(skip_serializing_if = "HashMap::is_empty", default = "HashMap::new")]
+    children: HashMap<ChildHandle, ChildStatus>,
+}
+
+impl CaStatus {
+    pub fn get_children_connection_stats(&self) -> ChildrenConnectionStats {
+        let children = self
+            .children
+            .clone()
+            .into_iter()
+            .map(|(handle, status)| ChildConnectionStats::new(handle, status.into()))
+            .collect();
+        ChildrenConnectionStats::new(children)
+    }
 }
 
 impl Default for CaStatus {
@@ -24,6 +41,7 @@ impl Default for CaStatus {
         CaStatus {
             repo: RepoStatus::default(),
             parents: ParentStatuses::default(),
+            children: HashMap::new(),
         }
     }
 }
@@ -51,12 +69,6 @@ impl StatusStore {
         Ok(self.store.get(&Self::status_key(ca))?.unwrap_or_default())
     }
 
-    /// Save the status for a CA
-    fn set_ca_status(&self, ca: &Handle, status: &CaStatus) -> KrillResult<()> {
-        self.store.store(&Self::status_key(ca), status)?;
-        Ok(())
-    }
-
     pub async fn get_parent_statuses(&self, ca: &Handle) -> KrillResult<ParentStatuses> {
         let _lock = self.lock.read().await;
         let status = self.get_ca_status(ca)?;
@@ -77,15 +89,14 @@ impl StatusStore {
         error: &Error,
         next_run_seconds: i64,
     ) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-
         let error_response = Self::error_to_error_res(&error);
 
-        status
-            .parents
-            .set_failure(parent, uri, error_response, next_run_seconds);
-        self.set_ca_status(ca, &status)
+        self.update_ca_status(ca, |status| {
+            status
+                .parents
+                .set_failure(parent, uri, error_response, next_run_seconds)
+        })
+        .await
     }
 
     pub async fn set_parent_last_updated(
@@ -95,10 +106,10 @@ impl StatusStore {
         uri: &ServiceUri,
         next_run_seconds: i64,
     ) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-        status.parents.set_last_updated(parent, uri, next_run_seconds);
-        self.set_ca_status(ca, &status)
+        self.update_ca_status(ca, |status| {
+            status.parents.set_last_updated(parent, uri, next_run_seconds)
+        })
+        .await
     }
 
     pub async fn set_parent_entitlements(
@@ -109,37 +120,57 @@ impl StatusStore {
         entitlements: &Entitlements,
         next_run_seconds: i64,
     ) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-        status
-            .parents
-            .set_entitlements(parent, uri, entitlements, next_run_seconds);
-        self.set_ca_status(ca, &status)
+        self.update_ca_status(ca, |status| {
+            status
+                .parents
+                .set_entitlements(parent, uri, entitlements, next_run_seconds)
+        })
+        .await
     }
 
     pub async fn remove_parent(&self, ca: &Handle, parent: &ParentHandle) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-        status.parents.remove(parent);
+        self.update_ca_status(ca, |status| status.parents.remove(parent)).await
+    }
 
-        Ok(())
+    pub async fn set_child_success(
+        &self,
+        ca: &Handle,
+        child: &ChildHandle,
+        user_agent: Option<String>,
+    ) -> KrillResult<()> {
+        self.update_ca_child_status(ca, child, |status| status.set_success(user_agent))
+            .await
+    }
+
+    pub async fn set_child_failure(
+        &self,
+        ca: &Handle,
+        child: &ChildHandle,
+        user_agent: Option<String>,
+        error: &Error,
+    ) -> KrillResult<()> {
+        let error_response = Self::error_to_error_res(&error);
+
+        self.update_ca_child_status(ca, child, |status| status.set_failure(user_agent, error_response))
+            .await
+    }
+
+    pub async fn get_stats_child_connections(&self, ca: &Handle) -> KrillResult<ChildrenConnectionStats> {
+        let _lock = self.lock.read().await;
+        let status = self.get_ca_status(ca)?;
+
+        Ok(status.get_children_connection_stats())
     }
 
     pub async fn set_status_repo_failure(&self, ca: &Handle, uri: ServiceUri, error: &Error) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-
         let error_response = Self::error_to_error_res(&error);
-
-        status.repo.set_failure(uri, error_response);
-        self.set_ca_status(ca, &status)
+        self.update_ca_status(ca, |status| status.repo.set_failure(uri, error_response))
+            .await
     }
 
     pub async fn set_status_repo_success(&self, ca: &Handle, uri: ServiceUri, next_hours: i64) -> KrillResult<()> {
-        let _lock = self.lock.write().await;
-        let mut status = self.get_ca_status(ca)?;
-        status.repo.set_last_updated(uri, next_hours);
-        self.set_ca_status(ca, &status)
+        self.update_ca_status(ca, |status| status.repo.set_last_updated(uri, next_hours))
+            .await
     }
 
     pub async fn set_status_repo_published(
@@ -149,10 +180,38 @@ impl StatusStore {
         published: Vec<PublishElement>,
         next_hours: i64,
     ) -> KrillResult<()> {
+        self.update_ca_status(ca, |status| status.repo.set_published(uri, published, next_hours))
+            .await
+    }
+
+    async fn update_ca_status<F>(&self, ca: &Handle, op: F) -> KrillResult<()>
+    where
+        F: FnOnce(&mut CaStatus),
+    {
         let _lock = self.lock.write().await;
         let mut status = self.get_ca_status(ca)?;
-        status.repo.set_published(uri, published, next_hours);
-        self.set_ca_status(ca, &status)
+        op(&mut status);
+
+        self.store.store(&Self::status_key(ca), &status)?;
+
+        Ok(())
+    }
+
+    async fn update_ca_child_status<F>(&self, ca: &Handle, child: &ChildHandle, op: F) -> KrillResult<()>
+    where
+        F: FnOnce(&mut ChildStatus),
+    {
+        self.update_ca_status(ca, |status| {
+            let mut child_status = match status.children.get_mut(child) {
+                Some(child_status) => child_status,
+                None => {
+                    status.children.insert(child.clone(), ChildStatus::default());
+                    status.children.get_mut(child).unwrap()
+                }
+            };
+            op(&mut child_status)
+        })
+        .await
     }
 
     fn error_to_error_res(error: &Error) -> ErrorResponse {

src/daemon/http/mod.rs

@@ -1,3 +1,4 @@
+use hyper::header::USER_AGENT;
 use serde::de::DeserializeOwned;
 use std::io;
 use std::str::FromStr;
@@ -16,6 +17,7 @@ use crate::commons::{
     actor::{Actor, ActorDef},
     KrillResult,
 };
+use crate::constants::HTTP_USER_AGENT_TRUNCATE;
 use crate::daemon::auth::LoggedInUser;
 use crate::daemon::http::server::State;
 
@@ -353,6 +355,21 @@ impl Request {
         self.request.headers()
     }
 
+    pub fn user_agent(&self) -> Option<String> {
+        match self.headers().get(&USER_AGENT) {
+            None => None,
+            Some(value) => value.to_str().ok().map(|s| {
+                // Note: HeaderValue.to_str() only returns ok in case the value is plain
+                //       ascii so it's safe to treat bytes as characters here.
+                if s.len() > HTTP_USER_AGENT_TRUNCATE {
+                    s[..HTTP_USER_AGENT_TRUNCATE].to_string()
+                } else {
+                    s.to_string()
+                }
+            }),
+        }
+    }
+
     pub async fn upgrade_from_anonymous(&mut self, actor_def: ActorDef) {
         if self.actor.is_anonymous() {
             self.actor = self.state.actor_from_def(actor_def);

src/daemon/http/server.rs

@@ -77,23 +77,19 @@ fn test_data_dir_or_die(config_item: &str, dir: &Path) {
     let test_file = dir.join("test");
 
     if let Err(e) = file::save(b"test", &test_file) {
-        print_write_error_hint_and_die(
-            format!(
-                "Cannot write to dir '{}' for configuration setting '{}', Error: {}",
-                dir.to_string_lossy(),
-                config_item,
-                e
-            )
-        );
+        print_write_error_hint_and_die(format!(
+            "Cannot write to dir '{}' for configuration setting '{}', Error: {}",
+            dir.to_string_lossy(),
+            config_item,
+            e
+        ));
     } else if let Err(e) = file::delete_file(&test_file) {
-        print_write_error_hint_and_die(
-            format!(
-                "Cannot delete test file '{}' in dir for configuration setting '{}', Error: {}",
-                test_file.to_string_lossy(),
-                config_item,
-                e
-            )
-        );
+        print_write_error_hint_and_die(format!(
+            "Cannot delete test file '{}' in dir for configuration setting '{}', Error: {}",
+            test_file.to_string_lossy(),
+            config_item,
+            e
+        ));
     }
 }
 
@@ -665,13 +661,14 @@ pub async fn rfc6492(req: Request) -> RoutingResult {
 
         let actor = req.actor();
         let state = req.state().clone();
+        let user_agent = req.user_agent();
 
         let bytes = match req.rfc6492_bytes().await {
             Ok(bytes) => bytes,
             Err(e) => return render_error(e),
         };
         let krill_server = state;
-        match krill_server.rfc6492(ca, bytes, &actor).await {
+        match krill_server.rfc6492(ca, bytes, user_agent, &actor).await {
             Ok(bytes) => Ok(HttpResponse::rfc6492(bytes.to_vec())),
             Err(e) => render_error(e),
         }
@@ -871,7 +868,10 @@ async fn api_cas(req: Request, path: &mut RequestPath) -> RoutingResult {
                 Some("parents") => api_ca_parents(req, path, ca).await,
                 Some("repo") => api_ca_repo(req, path, ca).await,
                 Some("routes") => api_ca_routes(req, path, ca).await,
+                Some("stats") => api_ca_stats(req, path, ca).await,
+
                 Some("rta") => api_ca_rta(req, path, ca).await,
+
                 _ => render_unknown_method(),
             }
         }),
@@ -939,6 +939,16 @@ async fn api_ca_routes(req: Request, path: &mut RequestPath, ca: Handle) -> Rout
     }
 }
 
+async fn api_ca_stats(req: Request, path: &mut RequestPath, ca: Handle) -> RoutingResult {
+    match path.next() {
+        Some("children") => match path.next() {
+            Some("connections") => api_ca_stats_child_connections(req, ca).await,
+            _ => render_unknown_method(),
+        },
+        _ => render_unknown_method(),
+    }
+}
+
 async fn api_publication_server(req: Request, path: &mut RequestPath) -> RoutingResult {
     match path.next() {
         Some("publishers") => api_publishers(req, path).await,
@@ -991,7 +1001,8 @@ pub async fn api_stale_publishers(req: Request, seconds: Option<&str>) -> Routin
         let seconds = seconds.unwrap_or("");
         match i64::from_str(seconds) {
             Ok(seconds) => render_json_res(
-                req.state().repo_stats()
+                req.state()
+                    .repo_stats()
                     .map(|stats| PublisherList::build(&stats.stale_publishers(seconds))),
             ),
             Err(_) => render_error(Error::ApiInvalidSeconds),
@@ -1003,7 +1014,9 @@ pub async fn api_stale_publishers(req: Request, seconds: Option<&str>) -> Routin
 pub async fn api_list_pbl(req: Request) -> RoutingResult {
     aa!(req, Permission::PUB_LIST, {
         render_json_res(
-            req.state().publishers().map(|publishers| PublisherList::build(&publishers)),
+            req.state()
+                .publishers()
+                .map(|publishers| PublisherList::build(&publishers)),
         )
     })
 }
@@ -1101,6 +1114,15 @@ async fn api_ca_child_show(req: Request, ca: Handle, child: ChildHandle) -> Rout
     )
 }
 
+async fn api_ca_stats_child_connections(req: Request, ca: Handle) -> RoutingResult {
+    aa!(
+        req,
+        Permission::CA_READ,
+        ca.clone(),
+        render_json_res(req.state().ca_stats_child_connections(&ca).await)
+    )
+}
+
 async fn api_ca_parent_contact(req: Request, ca: Handle, child: ChildHandle) -> RoutingResult {
     aa!(
         req,
@@ -1431,9 +1453,7 @@ async fn api_ca_parent_add_or_update(req: Request, ca: Handle, parent_override:
         };
 
         match extract_parent_ca_req(&ca, bytes, parent_override) {
-            Ok(parent_req) => render_empty_res(
-                server.ca_parent_add_or_update(ca, parent_req, &actor).await,
-            ),
+            Ok(parent_req) => render_empty_res(server.ca_parent_add_or_update(ca, parent_req, &actor).await),
             Err(e) => render_error(e),
         }
     })
@@ -1577,9 +1597,7 @@ async fn api_ca_routes_analysis(req: Request, path: &mut RequestPath, ca: Handle
                     let server = req.state().clone();
                     match req.json().await {
                         Err(e) => render_error(e),
-                        Ok(resources) => {
-                            render_json_res(server.ca_routes_bgp_suggest(&ca, Some(resources)).await)
-                        }
+                        Ok(resources) => render_json_res(server.ca_routes_bgp_suggest(&ca, Some(resources)).await),
                     }
                 }
                 _ => render_unknown_method(),

src/daemon/krillserver.rs

@@ -7,14 +7,10 @@ use bytes::Bytes;
 use chrono::Duration;
 
 use rpki::{
-    repository::{
-        cert::Cert,
-        x509::Time,
-    },
+    repository::{cert::Cert, x509::Time},
     uri,
 };
 
-use crate::commons::actor::{Actor, ActorDef};
 use crate::commons::api::{
     AddChildRequest, AllCertAuthIssues, CaCommandDetails, CaRepoDetails, CertAuthInfo, CertAuthInit, CertAuthIssues,
     CertAuthList, CertAuthStats, ChildCaInfo, ChildHandle, CommandHistory, CommandHistoryCriteria, Handle, ListReply,
@@ -26,6 +22,10 @@ use crate::commons::bgp::{BgpAnalyser, BgpAnalysisReport, BgpAnalysisSuggestion}
 use crate::commons::crypto::KrillSigner;
 use crate::commons::eventsourcing::CommandKey;
 use crate::commons::remote::rfc8183;
+use crate::commons::{
+    actor::{Actor, ActorDef},
+    api::ChildrenConnectionStats,
+};
 use crate::commons::{KrillEmptyResult, KrillResult};
 use crate::constants::*;
 #[cfg(feature = "multi-user")]
@@ -433,11 +433,16 @@ impl KrillServer {
         Ok(())
     }
 
-    /// Show details for a child under the TA.
-    pub async fn ca_child_show(&self, parent: &ParentHandle, child: &ChildHandle) -> KrillResult<ChildCaInfo> {
-        let child = self.ca_manager.ca_show_child(parent, child).await?;
+    /// Show details for a child under the CA.
+    pub async fn ca_child_show(&self, ca: &Handle, child: &ChildHandle) -> KrillResult<ChildCaInfo> {
+        let child = self.ca_manager.ca_show_child(ca, child).await?;
         Ok(child)
     }
+
+    /// Show children stats under the CA.
+    pub async fn ca_stats_child_connections(&self, ca: &Handle) -> KrillResult<ChildrenConnectionStats> {
+        self.ca_manager.ca_stats_child_connections(ca).await
+    }
 }
 
 /// # Being a child
@@ -457,15 +462,13 @@ impl KrillServer {
     ) -> KrillEmptyResult {
         let parent = parent_req.handle();
         let contact = parent_req.contact();
-        self.ca_manager.get_entitlements_from_contact(&ca, parent, contact, false).await?;
+        self.ca_manager
+            .get_entitlements_from_contact(&ca, parent, contact, false)
+            .await?;
 
-        Ok(self
-            .ca_manager
-            .ca_parent_add_or_update(ca, parent_req, actor)
-            .await?)
+        Ok(self.ca_manager.ca_parent_add_or_update(ca, parent_req, actor).await?)
     }
 
-
     pub async fn ca_parent_remove(&self, handle: Handle, parent: ParentHandle, actor: &Actor) -> KrillEmptyResult {
         Ok(self.ca_manager.ca_parent_remove(handle, parent, actor).await?)
     }
@@ -646,8 +649,14 @@ impl KrillServer {
             .await?)
     }
 
-    pub async fn rfc6492(&self, handle: Handle, msg_bytes: Bytes, actor: &Actor) -> KrillResult<Bytes> {
-        Ok(self.ca_manager.rfc6492(&handle, msg_bytes, actor).await?)
+    pub async fn rfc6492(
+        &self,
+        handle: Handle,
+        msg_bytes: Bytes,
+        user_agent: Option<String>,
+        actor: &Actor,
+    ) -> KrillResult<Bytes> {
+        Ok(self.ca_manager.rfc6492(&handle, msg_bytes, user_agent, actor).await?)
     }
 }