From 37fc79c9ec6bf974e31f6844da703119815f78e0 Mon Sep 17 00:00:00 2001 From: github-ops Date: Sun, 21 Jun 2026 01:50:44 +0000 Subject: [PATCH] Split admin observability service module --- apps/admin-api/src/service.rs | 161 ++------------------ apps/admin-api/src/service/observability.rs | 155 +++++++++++++++++++ 2 files changed, 164 insertions(+), 152 deletions(-) create mode 100644 apps/admin-api/src/service/observability.rs diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index d0060fe..d946a6e 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -20,14 +20,14 @@ use crank_mapping::{JsonPathRoot, MappingRule, MappingSet, infer_mapping_from_sa use crank_registry::{ AgentSummary, AgentVersionRecord, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, - CreateVersionRequest, CreateWorkspaceRequest, InvocationLogRecord, ListInvocationLogsQuery, - OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, - OperationVersionRecord, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, - PublishRequest, RegistryError, RegistryOperation, RotateSecretRequest, SampleKind, - SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveSampleMetadataRequest, - SaveWorkspaceUpstreamRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, - UsageOperationBreakdown, UsageQuery, UsageSummary, UsageTimelinePoint, - WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, + CreateVersionRequest, CreateWorkspaceRequest, ListInvocationLogsQuery, OperationAgentRef, + OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord, + PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryError, + RegistryOperation, RotateSecretRequest, SampleKind, SaveAgentBindingsRequest, + SaveAuthProfileRequest, SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, + UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageQuery, + UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, + WorkspaceUpstream, WorkspaceUpstreamId, }; use crank_runtime::{ PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, RuntimeOperation, @@ -42,6 +42,7 @@ use tracing::{info, instrument}; use uuid::Uuid; mod import_export; +mod observability; use crate::{ auth::{ @@ -1115,150 +1116,6 @@ impl AdminService { Ok(()) } - #[instrument(skip(self))] - pub async fn list_logs( - &self, - workspace_id: &WorkspaceId, - query: LogsQuery, - ) -> Result, 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 { - 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 { - 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 { - 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 { - 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() }), - ) - }) - } - pub async fn list_protocol_capabilities(&self) -> Vec { let capabilities = self.get_capabilities().await; capabilities diff --git a/apps/admin-api/src/service/observability.rs b/apps/admin-api/src/service/observability.rs new file mode 100644 index 0000000..8cdf010 --- /dev/null +++ b/apps/admin-api/src/service/observability.rs @@ -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, 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 { + 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 { + 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 { + 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 { + 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() }), + ) + }) + } +}