bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: harden demo flows and observability

Browse Source
This commit is contained in:
a.tolmachev
2026-03-25 21:31:54 +03:00
parent 1a6d7c5ddc
commit 2d3abb9f3d
7 changed files with 320 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/admin-api/src/app.rs
+113
View File
@@ -375,6 +375,119 @@ mod tests {
assert_eq!(imported["import_mode"], "upsert");
}
#[tokio::test]
async fn roundtrips_graphql_operation_through_yaml_upsert() {
let registry = test_registry().await;
let storage_root = test_storage_root("yaml_graphql");
let upstream_base_url = spawn_graphql_server().await;
let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
let client = reqwest::Client::new();
let created = client
.post(format!("{base_url}/operations"))
.json(&test_graphql_operation_payload(
&upstream_base_url,
"crm_yaml_graphql",
))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let operation_id = created["operation_id"].as_str().unwrap().to_owned();
let yaml = client
.get(format!("{base_url}/operations/{operation_id}/export"))
.send()
.await
.unwrap()
.text()
.await
.unwrap();
let imported = client
.post(format!("{base_url}/operations/import?mode=upsert"))
.body(yaml)
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let test_run = client
.post(format!("{base_url}/operations/{operation_id}/test-runs"))
.json(&json!({
"version": 2,
"input": { "email": "user@example.com" }
}))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
assert_eq!(imported["operation_id"], operation_id);
assert_eq!(imported["version"], 2);
assert_eq!(test_run["ok"], true);
assert_eq!(test_run["response_preview"]["id"], "lead_123");
}
#[tokio::test]
async fn roundtrips_grpc_operation_through_yaml_upsert() {
let registry = test_registry().await;
let storage_root = test_storage_root("yaml_grpc");
let server_addr = grpc_test_support::spawn_unary_echo_server().await;
let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
let client = reqwest::Client::new();
let created = client
.post(format!("{base_url}/operations"))
.json(&test_grpc_operation_payload(&server_addr, "echo_yaml_grpc"))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let operation_id = created["operation_id"].as_str().unwrap().to_owned();
let yaml = client
.get(format!("{base_url}/operations/{operation_id}/export"))
.send()
.await
.unwrap()
.text()
.await
.unwrap();
let imported = client
.post(format!("{base_url}/operations/import?mode=upsert"))
.body(yaml)
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let test_run = client
.post(format!("{base_url}/operations/{operation_id}/test-runs"))
.json(&json!({
"version": 2,
"input": { "message": "hello" }
}))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
assert_eq!(imported["operation_id"], operation_id);
assert_eq!(imported["version"], 2);
assert_eq!(test_run["ok"], true);
assert_eq!(test_run["response_preview"]["message"], "hello");
}
#[tokio::test]
async fn uploads_samples_and_generates_draft() {
let registry = test_registry().await;
apps/admin-api/src/error.rs
+25 -1
View File
@@ -9,6 +9,7 @@ use mcpaas_runtime::RuntimeError;
use mcpaas_schema::SchemaError;
use serde_json::{Value, json};
use thiserror::Error;
use tracing::{error, warn};
use crate::storage::StorageError;
@@ -70,6 +71,17 @@ impl ApiError {
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
match &self {
Self::Internal { message } => {
error!(error_code = self.code(), error_message = %message)
}
Self::Validation { message }
| Self::NotFound { message }
| Self::Conflict { message } => {
warn!(error_code = self.code(), error_message = %message)
}
}
let body = Json(json!({
"error": {
"code": self.code(),
@@ -137,7 +149,19 @@ impl From<StorageError> for ApiError {
pub fn runtime_test_failure(error: &RuntimeError) -> Value {
json!({
"code": "runtime_test_error",
"code": runtime_test_failure_code(error),
"message": error.to_string()
})
}
fn runtime_test_failure_code(error: &RuntimeError) -> &'static str {
match error {
RuntimeError::Schema(_) => "runtime_schema_error",
RuntimeError::Mapping(_) => "runtime_mapping_error",
RuntimeError::GraphqlAdapter(_) => "runtime_graphql_error",
RuntimeError::GrpcAdapter(_) => "runtime_grpc_error",
RuntimeError::RestAdapter(_) => "runtime_rest_error",
RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error",
RuntimeError::InvalidPreparedRequest { .. } => "runtime_request_error",
}
}
apps/admin-api/src/service.rs
+61 -6
View File
@@ -16,6 +16,7 @@ use mcpaas_schema::Schema;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use tracing::{info, instrument};
use uuid::Uuid;
use crate::{error::ApiError, storage::LocalArtifactStorage};
@@ -170,10 +171,12 @@ impl AdminService {
}
}
#[instrument(skip(self))]
pub async fn list_operations(&self) -> Result<Vec<OperationSummary>, ApiError> {
Ok(self.registry.list_operations().await?)
}
#[instrument(skip(self))]
pub async fn get_operation(
&self,
operation_id: &OperationId,
@@ -186,6 +189,7 @@ impl AdminService {
})
}
#[instrument(skip(self))]
pub async fn get_operation_version(
&self,
operation_id: &OperationId,
@@ -202,6 +206,7 @@ impl AdminService {
})
}
#[instrument(skip(self, payload), fields(protocol = ?payload.protocol, operation_name = %payload.name))]
pub async fn create_operation(
&self,
payload: OperationPayload,
@@ -243,6 +248,7 @@ impl AdminService {
};
self.registry.create_operation(&snapshot, None).await?;
info!(operation_id = %operation_id.as_str(), version = 1, "operation created");
Ok(CreatedOperationResponse {
operation_id: operation_id.as_str().to_owned(),
@@ -251,6 +257,7 @@ impl AdminService {
})
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), protocol = ?payload.operation.protocol))]
pub async fn create_version(
&self,
operation_id: &OperationId,
@@ -293,6 +300,7 @@ impl AdminService {
created_by: None,
})
.await?;
info!(operation_id = %operation_id.as_str(), version, "operation version created");
Ok(CreatedOperationResponse {
operation_id: operation_id.as_str().to_owned(),
@@ -301,6 +309,7 @@ impl AdminService {
})
}
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
pub async fn publish_operation(
&self,
operation_id: &OperationId,
@@ -315,6 +324,7 @@ impl AdminService {
published_by: None,
})
.await?;
info!(operation_id = %operation_id.as_str(), version, "operation published");
Ok(PublishResponse {
operation_id: operation_id.as_str().to_owned(),
@@ -323,6 +333,7 @@ impl AdminService {
})
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))]
pub async fn run_test(
&self,
operation_id: &OperationId,
@@ -363,10 +374,12 @@ impl AdminService {
}
}
#[instrument(skip(self))]
pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, ApiError> {
Ok(self.registry.list_auth_profiles().await?)
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("descriptor-set.bin")))]
pub async fn upload_descriptor_set(
&self,
operation_id: &OperationId,
@@ -412,6 +425,12 @@ impl AdminService {
},
})
.await?;
info!(
operation_id = %operation_id.as_str(),
descriptor_id = %descriptor_id.as_str(),
version = summary.current_draft_version,
"descriptor set uploaded"
);
Ok(DescriptorUploadResponse {
descriptor_id: descriptor_id.as_str().to_owned(),
@@ -419,6 +438,7 @@ impl AdminService {
})
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("schema.proto")))]
pub async fn upload_proto_file(
&self,
operation_id: &OperationId,
@@ -459,6 +479,12 @@ impl AdminService {
},
})
.await?;
info!(
operation_id = %operation_id.as_str(),
descriptor_id = %descriptor_id.as_str(),
version = summary.current_draft_version,
"proto file uploaded"
);
Ok(DescriptorUploadResponse {
descriptor_id: descriptor_id.as_str().to_owned(),
@@ -466,6 +492,7 @@ impl AdminService {
})
}
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
pub async fn list_grpc_services(
&self,
operation_id: &OperationId,
@@ -499,6 +526,7 @@ impl AdminService {
.collect::<Result<Vec<_>, _>>()
}
#[instrument(skip(self))]
pub async fn get_auth_profile(
&self,
auth_profile_id: &AuthProfileId,
@@ -514,6 +542,7 @@ impl AdminService {
})
}
#[instrument(skip(self, payload), fields(auth_profile_name = %payload.name, auth_kind = ?payload.kind))]
pub async fn create_auth_profile(
&self,
payload: AuthProfilePayload,
@@ -533,10 +562,12 @@ impl AdminService {
self.registry
.save_auth_profile(SaveAuthProfileRequest { profile: &profile })
.await?;
info!(auth_profile_id = %profile.id.as_str(), "auth profile created");
Ok(profile)
}
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))]
pub async fn export_operation(
&self,
operation_id: &OperationId,
@@ -566,6 +597,7 @@ impl AdminService {
serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string()))
}
#[instrument(skip(self, yaml_document), fields(mode = ?query.mode))]
pub async fn import_operation(
&self,
query: ImportQuery,
@@ -615,25 +647,38 @@ impl AdminService {
)
.await?;
Ok(ImportResponse {
let response = ImportResponse {
operation_id: created.operation_id,
version: created.version,
import_mode: ImportMode::Upsert,
warnings: Vec::new(),
})
};
info!(
operation_id = %response.operation_id,
version = response.version,
"operation imported by upsert"
);
Ok(response)
} else {
let created = self.create_operation(payload).await?;
Ok(ImportResponse {
let response = ImportResponse {
operation_id: created.operation_id,
version: created.version,
import_mode: ImportMode::Upsert,
warnings: Vec::new(),
})
};
info!(
operation_id = %response.operation_id,
version = response.version,
"operation imported by upsert"
);
Ok(response)
}
}
}
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))]
pub async fn save_json_sample(
&self,
operation_id: &OperationId,
@@ -667,10 +712,17 @@ impl AdminService {
self.registry
.save_sample_metadata(SaveSampleMetadataRequest { sample: &metadata })
.await?;
info!(
operation_id = %operation_id.as_str(),
sample_id = %metadata.id.as_str(),
version,
"json sample saved"
);
Ok(metadata)
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str()))]
pub async fn generate_draft(
&self,
operation_id: &OperationId,
@@ -722,13 +774,16 @@ impl AdminService {
warnings: Vec::new(),
};
Ok(DraftGenerationResult {
let result = DraftGenerationResult {
generated_draft,
input_schema,
output_schema,
input_mapping,
output_mapping,
})
};
info!(operation_id = %operation_id.as_str(), "draft generated from samples");
Ok(result)
}
fn validate_operation_payload(&self, payload: &OperationPayload) -> Result<(), ApiError> {