Split admin observability service module

2026-06-21 01:50:44 +00:00
parent 2dab9d666e
commit 37fc79c9ec
2 changed files with 164 additions and 152 deletions
@@ -0,0 +1,155 @@
+use crank_core::{AgentId, InvocationLogId, OperationId, UsagePeriod, WorkspaceId};
+use crank_registry::{InvocationLogRecord, ListInvocationLogsQuery, UsageQuery, UsageRollupRecord};
+use serde_json::json;
+use tracing::instrument;
+
+use crate::{
+    error::ApiError,
+    service::{AdminService, LogsQuery, UsageOverviewResponse, UsageRequestQuery, usage_window},
+};
+
+impl AdminService {
+    #[instrument(skip(self))]
+    pub async fn list_logs(
+        &self,
+        workspace_id: &WorkspaceId,
+        query: LogsQuery,
+    ) -> Result<Vec<InvocationLogRecord>, ApiError> {
+        self.ensure_workspace_exists(workspace_id).await?;
+        let operation_id = query.operation_id.as_deref().map(OperationId::new);
+        let agent_id = query.agent_id.as_deref().map(AgentId::new);
+        let (_, created_after, _) = usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
+
+        self.registry
+            .list_invocation_logs(ListInvocationLogsQuery {
+                workspace_id,
+                level: query.level,
+                search_text: query.search.as_deref(),
+                source: query.source,
+                operation_id: operation_id.as_ref(),
+                agent_id: agent_id.as_ref(),
+                created_after: Some(&created_after),
+                limit: query.limit.unwrap_or(100),
+            })
+            .await
+            .map_err(ApiError::from)
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_log(
+        &self,
+        workspace_id: &WorkspaceId,
+        log_id: &InvocationLogId,
+    ) -> Result<InvocationLogRecord, ApiError> {
+        self.registry
+            .get_invocation_log(workspace_id, log_id)
+            .await?
+            .ok_or_else(|| {
+                ApiError::not_found_with_context(
+                    format!("invocation log {} was not found", log_id.as_str()),
+                    json!({ "log_id": log_id.as_str() }),
+                )
+            })
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_usage_overview(
+        &self,
+        workspace_id: &WorkspaceId,
+        query: UsageRequestQuery,
+    ) -> Result<UsageOverviewResponse, ApiError> {
+        self.ensure_workspace_exists(workspace_id).await?;
+        let (period, created_after, bucket) =
+            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
+        let usage_query = UsageQuery {
+            workspace_id,
+            period,
+            source: query.source,
+            created_after: &created_after,
+            bucket,
+        };
+
+        let summary = self.registry.summarize_usage(usage_query.clone()).await?;
+        let timeline = self
+            .registry
+            .list_usage_timeline(usage_query.clone())
+            .await?;
+        let operations = self
+            .registry
+            .list_usage_by_operation(usage_query.clone())
+            .await?;
+        let agents = self.registry.list_usage_by_agent(usage_query).await?;
+
+        Ok(UsageOverviewResponse {
+            summary,
+            timeline,
+            operations,
+            agents,
+        })
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_operation_usage(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+        query: UsageRequestQuery,
+    ) -> Result<UsageRollupRecord, ApiError> {
+        self.get_operation(workspace_id, operation_id).await?;
+        let (period, created_after, bucket) =
+            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
+
+        self.registry
+            .get_usage_for_operation(
+                UsageQuery {
+                    workspace_id,
+                    period,
+                    source: query.source,
+                    created_after: &created_after,
+                    bucket,
+                },
+                operation_id,
+            )
+            .await?
+            .ok_or_else(|| {
+                ApiError::not_found_with_context(
+                    format!(
+                        "usage for operation {} was not found",
+                        operation_id.as_str()
+                    ),
+                    json!({ "operation_id": operation_id.as_str() }),
+                )
+            })
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_agent_usage(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+        query: UsageRequestQuery,
+    ) -> Result<UsageRollupRecord, ApiError> {
+        self.get_agent(workspace_id, agent_id).await?;
+        let (period, created_after, bucket) =
+            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
+
+        self.registry
+            .get_usage_for_agent(
+                UsageQuery {
+                    workspace_id,
+                    period,
+                    source: query.source,
+                    created_after: &created_after,
+                    bucket,
+                },
+                agent_id,
+            )
+            .await?
+            .ok_or_else(|| {
+                ApiError::not_found_with_context(
+                    format!("usage for agent {} was not found", agent_id.as_str()),
+                    json!({ "agent_id": agent_id.as_str() }),
+                )
+            })
+    }
+}