// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
// macOS ld cannot encode compact-unwind offsets once __eh_frame exceeds 16MB;
// harmless for a binary this size (only slows panic unwinding), so silence it.
#![allow(linker_messages)]
#![recursion_limit = "256"]
#[cfg(test)]
mod tests {
use core::time;
use std::{
env, fs,
net::SocketAddr,
str,
sync::{Arc, Once},
thread,
};
use arrow_flight::flight_service_server::FlightServiceServer;
use axum::{
Router,
body::Body,
http::{HeaderMap, HeaderValue, Method, Request, StatusCode, header},
};
use bytes::{Bytes, BytesMut};
use chrono::{Duration, Utc};
use common::meta::user::UserList;
use config::{
get_config,
meta::{
alerts::{Operator, QueryCondition, TriggerCondition, alert::Alert},
dashboards::{Dashboard, v5},
pipeline::{
Pipeline,
components::{DerivedStream, PipelineSource},
},
stream::StreamType,
triggers::{ScheduledTriggerData, Trigger, TriggerModule, TriggerStatus},
},
utils::{
enrichment_local_cache::{
get_key, get_metadata_content, get_metadata_path, get_table_dir, get_table_path,
},
json,
},
};
use enrichment_data::enrichment::storage::{Values, local};
use infra::schema::{STREAM_SCHEMAS, STREAM_SCHEMAS_LATEST};
use ingestion_common::IngestionResponse;
use openobserve::migration;
use openobserve_api_grpc::handler::grpc::{
auth::check_internal_auth, flight::FlightServiceImpl, request::logs::LogsServer,
};
use openobserve_api_http::handler::http::router::{
basic_routes, config_routes, other_service_routes, service_routes,
};
use openobserve_api_management::{
models::{
alerts::responses::{GetAlertResponseBody, ListAlertsResponseBody},
destinations::{Destination, DestinationType},
},
request::alerts::external_events::inactive_org_response,
};
use openobserve_api_pipelines::models::pipelines::{Pipeline as ApiPipeline, PipelineList};
use openobserve_core::alerts::scheduler::handlers::handle_triggers;
use opentelemetry_proto::tonic::{
collector::logs::v1::{
ExportLogsServiceRequest, logs_service_client::LogsServiceClient,
logs_service_server::LogsServiceServer,
},
common::v1::{AnyValue, any_value},
logs::v1::{LogRecord, ResourceLogs, ScopeLogs},
};
use prost::Message;
use proto::{cluster_rpc::search_server::SearchServer, prometheus_rpc};
use search_service::SEARCH_SERVER;
use serde_json::json;
use tonic::codec::CompressionEncoding;
use tower::ServiceExt;
use transform::enrichment::ENRICHMENT_TABLES;
/// Tables of Task 0.17 that carry an org column, and that column.
const KV_COVERAGE_TABLES: [(&str, &str); 10] = [
("stream_stats", "org"),
("file_list_history", "org"),
("file_list_jobs", "org"),
("pipeline_last_errors", "org_id"),
("kv_store", "org_id"),
("system_settings", "org_id"),
("short_urls", "org_id"),
("backfill_jobs", "org"),
("enrichment_tables", "org"),
("enrichment_table_urls", "org"),
];
/// The stack infra's runtimes give every thread that polls a server future.
const E2E_STACK_SIZE: usize = 16 * 1024 * 1024;
static START: Once = Once::new();
// A sub-test that calls `setup()` after `tear_down` fails fast instead of using deleted state.
static TORN_DOWN: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
struct CoverageRows {
check_id: String,
integration_id: String,
anomaly_id: String,
workflow_id: String,
probe_token: String,
agent_id: String,
location_id: String,
artifact: String,
pricing_model: String,
team_id: String,
response_id: String,
policy_id: String,
dek: Vec,
}
/// The e2e's own data dir, ports, config file and Postgres database, removed by `tear_down`.
struct Sandbox {
dir: tempfile::TempDir,
postgres: Option,
}
impl Sandbox {
/// Runs before the first `get_config`, which reads the env this sets exactly once.
async fn create() -> Self {
let dir = tempfile::Builder::new()
.prefix("o2-e2e-")
.tempdir()
.expect("e2e temp dir");
let config_file = dir.path().join("e2e.env");
fs::write(&config_file, "").expect("e2e config file");
config::config_path_manager::set_config_file_path(config_file)
.expect("e2e config file path");
// The enterprise config also loads a `.env` from the working directory.
env::set_current_dir(dir.path()).expect("enter the e2e temp dir");
let postgres = OwnDatabase::create().await;
let [grpc_port, http_port] = free_ports();
unsafe {
env::set_var("ZO_DATA_DIR", format!("{}/data/", dir.path().display()));
for derived in [
"ZO_DATA_WAL_DIR",
"ZO_DATA_STREAM_DIR",
"ZO_DATA_DB_DIR",
"ZO_DATA_CACHE_DIR",
"ZO_DATA_TMP_DIR",
] {
env::remove_var(derived);
}
env::set_var("ZO_GRPC_ADDR", "127.0.0.1");
env::set_var("ZO_GRPC_PORT", grpc_port.to_string());
env::set_var("ZO_HTTP_ADDR", "127.0.0.1");
env::set_var("ZO_HTTP_PORT", http_port.to_string());
}
Self { dir, postgres }
}
/// Panics unless the config that every pool connects from names this database.
fn assert_config_uses_it(&self) {
let Some(postgres) = &self.postgres else {
return;
};
let cfg = get_config();
let dsn = url::Url::parse(&cfg.common.meta_postgres_dsn).expect("e2e DSN is a URL");
assert_eq!(dsn.path(), format!("/{}", postgres.name));
assert!(cfg.common.meta_postgres_ro_dsn.is_empty());
assert!(cfg.common.meta_ddl_dsn.is_empty());
}
async fn tear_down(self) {
log::info!("Tear Down Invoked");
TORN_DOWN.store(true, std::sync::atomic::Ordering::SeqCst);
let data = self.dir.path().join("data");
// The WAL push job can still be moving the last ingests' files while the dir goes.
let mut result = fs::remove_dir_all(&data);
for _ in 0..20 {
if result.is_ok() {
break;
}
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
result = fs::remove_dir_all(&data);
}
result.expect("Delete local dir failed");
if let Some(postgres) = self.postgres {
postgres.drop_database().await;
}
}
}
/// A database made from the CI DSN, so the e2e never writes into the one the unit tests share.
struct OwnDatabase {
admin_dsn: String,
name: String,
dropped: bool,
}
impl OwnDatabase {
async fn create() -> Option {
use sqlx::Connection;
if env::var("ZO_META_STORE").ok()?.as_str() != "postgres" {
return None;
}
let admin_dsn = env::var("ZO_META_POSTGRES_DSN")
.ok()
.filter(|dsn| !dsn.is_empty())
.expect("the Postgres e2e makes its own database from ZO_META_POSTGRES_DSN");
let name = format!(
"o2_e2e_{}_{}",
std::process::id(),
Utc::now().timestamp_micros()
);
let mut conn = sqlx::PgConnection::connect(&admin_dsn)
.await
.expect("connect to the CI Postgres");
sqlx::query(&format!("CREATE DATABASE {name}"))
.execute(&mut conn)
.await
.expect("create the e2e database");
let mut dsn = url::Url::parse(&admin_dsn).expect("ZO_META_POSTGRES_DSN is a URL");
dsn.set_path(&format!("/{name}"));
unsafe {
env::set_var("ZO_META_POSTGRES_DSN", dsn.as_str());
env::remove_var("ZO_META_POSTGRES_RO_DSN");
env::remove_var("ZO_META_DDL_DSN");
}
Some(Self {
admin_dsn,
name,
dropped: false,
})
}
async fn drop_database(mut self) {
Self::drop_named(&self.admin_dsn, &self.name)
.await
.expect("drop the e2e database");
self.dropped = true;
}
async fn drop_named(admin_dsn: &str, name: &str) -> Result<(), sqlx::Error> {
use sqlx::Connection;
let mut conn = sqlx::PgConnection::connect(admin_dsn).await?;
// FORCE ends the pools this process still holds on the database.
sqlx::query(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)"))
.execute(&mut conn)
.await?;
Ok(())
}
}
impl Drop for OwnDatabase {
/// Also runs on unwind; on a new thread because no runtime starts inside the test's.
fn drop(&mut self) {
if self.dropped {
return;
}
let (admin_dsn, name) = (self.admin_dsn.clone(), self.name.clone());
let dropped = std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| e.to_string())?
.block_on(Self::drop_named(&admin_dsn, &name))
.map_err(|e| e.to_string())
})
.join();
if !matches!(dropped, Ok(Ok(()))) {
eprintln!(
"the e2e database {} was not dropped: {dropped:?}",
self.name
);
}
}
}
fn setup() -> (&'static str, &'static str) {
assert!(
!TORN_DOWN.load(std::sync::atomic::Ordering::SeqCst),
"a sub-test ran after tear_down removed the e2e's data dir and database"
);
START.call_once(|| unsafe {
env::set_var("ZO_ROOT_USER_EMAIL", "root@example.com");
env::set_var("ZO_ROOT_USER_PASSWORD", "Complexpass#123");
env::set_var("ZO_LOCAL_MODE", "true");
env::set_var("ZO_MAX_FILE_SIZE_ON_DISK", "1");
env::set_var("ZO_FILE_PUSH_INTERVAL", "1");
env::set_var("ZO_PAYLOAD_LIMIT", "209715200");
env::set_var("ZO_JSON_LIMIT", "209715200");
env::set_var("ZO_RESULT_CACHE_ENABLED", "false");
env::set_var("ZO_PRINT_KEY_SQL", "true");
env::set_var("ZO_SMTP_ENABLED", "true");
env::set_var("ZO_CREATE_ORG_THROUGH_INGESTION", "true");
env::set_var("ZO_ORG_CLEANUP_POLL_INTERVAL_SECS", "1");
env::set_var("ZO_COMPACT_DELETE_FILES_DELAY_MINUTES", "0");
env::set_var("ZO_SYNTHETICS_ENABLED", "true");
env::set_var("ZO_SKIP_SSRF_CHECKS", "true");
env::set_var("O2_ORG_DELETION_GRACE_PERIOD_DAYS", "0");
env_logger::init_from_env(
env_logger::Env::new().default_filter_or(&get_config().log.level),
);
log::info!("setup Invoked");
});
(
"Authorization",
"Basic cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM=",
)
}
/// Initialize test router with service and basic routes
fn init_test_router() -> Router {
Router::new()
.merge(basic_routes())
.nest("/config", config_routes())
.nest("/api", service_routes())
}
/// Make a test request and return the response
async fn make_request(
app: &Router,
method: Method,
uri: &str,
headers: Option,
body: Option,
) -> (StatusCode, Bytes) {
let mut req_builder = Request::builder().method(method).uri(uri);
if let Some(hdrs) = headers {
for (key, value) in hdrs.iter() {
req_builder = req_builder.header(key, value);
}
}
let req = if let Some(body_str) = body {
req_builder
.body(Body::from(body_str))
.expect("Failed to build request")
} else {
req_builder
.body(Body::empty())
.expect("Failed to build request")
};
let response = app
.clone()
.oneshot(req)
.await
.expect("Failed to execute request");
let status = response.status();
let body_bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.expect("Failed to read response body");
(status, body_bytes)
}
/// Helper to create headers with auth
fn auth_headers(auth: (&str, &str)) -> HeaderMap {
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_str(auth.1).expect("Invalid auth header"),
);
headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/json"),
);
headers
}
async fn init_grpc_server() -> Result<(), anyhow::Error> {
let cfg = get_config();
let ip = if !cfg.grpc.addr.is_empty() {
cfg.grpc.addr.clone()
} else {
"0.0.0.0".to_string()
};
let gaddr: SocketAddr = format!("{}:{}", ip, cfg.grpc.port).parse()?;
let search_svc = SearchServer::new(SEARCH_SERVER.clone())
.send_compressed(CompressionEncoding::Gzip)
.accept_compressed(CompressionEncoding::Gzip);
let flight_svc = FlightServiceServer::new(FlightServiceImpl)
.send_compressed(CompressionEncoding::Gzip)
.accept_compressed(CompressionEncoding::Gzip);
let logs_svc = LogsServiceServer::new(LogsServer);
log::info!("starting gRPC server at {}", gaddr);
tonic::transport::Server::builder()
.layer(tonic::service::InterceptorLayer::new(check_internal_auth))
.add_service(search_svc)
.add_service(flight_svc)
.add_service(logs_svc)
.serve(gaddr)
.await
.expect("gRPC server init failed");
Ok(())
}
/// Ports free now; both listeners stay open until both are picked, so they differ.
fn free_ports() -> [u16; 2] {
let listeners = [(); 2]
.map(|_| std::net::TcpListener::bind("127.0.0.1:0").expect("bind an ephemeral port"));
listeners.map(|l| l.local_addr().expect("ephemeral port").port())
}
/// Helper function to flush memtable to ensure proper test isolation
/// This prevents memtable overflow errors when running sequential ingestion tests
async fn flush_memtable() {
if let Err(e) = ingester::flush_all().await {
log::warn!("Failed to flush memtable: {}", e);
}
}
/// Cleanup any leftover state from previous test runs/retries.
/// Order matters: alerts first (they reference destinations), then destinations, then
/// templates. This ensures idempotency when tests are retried.
async fn cleanup_previous_test_state() {
let auth = setup();
let app = init_test_router();
let headers = auth_headers(auth);
// 1. Delete alerts first (they reference destinations)
// List alerts and delete by ID
let (status, body) = make_request(
&app,
Method::GET,
"/api/v2/e2e/alerts?stream_type=logs",
Some(headers.clone()),
None,
)
.await;
if status.is_success()
&& let Ok(alerts) = serde_json::from_slice::(&body)
&& let Some(list) = alerts.get("list").and_then(|l| l.as_array())
{
let alert_names = ["alertChk", "sns_test_alert", "multirange_alert"];
for alert in list {
if let Some(name) = alert.get("name").and_then(|n| n.as_str())
&& alert_names.contains(&name)
&& let Some(id) = alert.get("alert_id").and_then(|id| id.as_str())
{
let _ = make_request(
&app,
Method::DELETE,
&format!("/api/v2/e2e/alerts/{}?type=logs", id),
Some(headers.clone()),
None,
)
.await;
log::info!("Cleanup: deleted alert {}", name);
}
}
}
// 2. Delete destinations (they reference templates)
for dest in ["slack", "email", "sns_alert"] {
let _ = make_request(
&app,
Method::DELETE,
&format!("/api/e2e/alerts/destinations/{}", dest),
Some(headers.clone()),
None,
)
.await;
}
// 3. Delete templates last
for template in ["slackTemplate", "email_template", "snsTemplate"] {
let _ = make_request(
&app,
Method::DELETE,
&format!("/api/e2e/alerts/templates/{}", template),
Some(headers.clone()),
None,
)
.await;
}
// 4. Delete user if exists
let _ = make_request(
&app,
Method::DELETE,
"/api/e2e/users/nonadmin@example.com",
Some(headers.clone()),
None,
)
.await;
log::info!("Cleanup: finished cleaning up previous test state");
}
// tokio::test polls the body on the 2 MiB harness thread, which the enterprise build overflows.
#[test]
fn e2e_test() {
let body = std::thread::Builder::new()
.name("e2e_test".to_string())
.stack_size(E2E_STACK_SIZE)
.spawn(|| {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.thread_stack_size(E2E_STACK_SIZE)
.enable_all()
.build()
.unwrap()
.block_on(e2e_test_body());
})
.unwrap();
if let Err(panic) = body.join() {
std::panic::resume_unwind(panic);
}
}
async fn e2e_test_body() {
let sandbox = Sandbox::create().await;
setup();
sandbox.assert_config_uses_it();
e2e_boot().await;
for _i in 0..3 {
e2e_1_post_bulk().await;
}
// Flush memtable after bulk ingestion to ensure proper test isolation
// This prevents memtable overflow when subsequent tests try to ingest data
flush_memtable().await;
// ingest
e2e_post_json().await;
e2e_post_multi().await;
e2e_post_trace().await;
e2e_post_metrics().await;
e2e_post_hec().await;
e2e_firehose_and_pubsub_status().await;
// e2e_post_kinesis_data().await;
// streams
e2e_get_stream().await;
e2e_get_stream_schema().await;
e2e_get_org_summary().await;
e2e_post_stream_settings().await;
e2e_get_es_settings().await;
// functions
e2e_post_function().await;
e2e_list_functions().await;
e2e_delete_function().await;
// search
e2e_search().await;
e2e_search_around().await;
// users
e2e_post_user().await;
e2e_update_user().await;
e2e_update_user_with_empty().await;
e2e_add_user_to_org().await;
e2e_add_root_user_to_org().await;
e2e_list_users().await;
e2e_get_organizations().await;
e2e_get_user_passcode().await;
e2e_update_user_passcode().await;
e2e_user_authentication().await;
e2e_user_authentication_with_error().await;
// dashboards
{
let board = e2e_create_dashboard().await;
let list = e2e_list_dashboards().await;
assert_eq!(list[0], board);
let mut v5_board = board.v5.unwrap();
v5_board.title = "e2e test".to_owned();
v5_board.description = "Logs flow downstream".to_owned();
let board = e2e_update_dashboard(v5_board, board.hash).await;
assert_eq!(
e2e_get_dashboard(&board.v5.as_ref().unwrap().dashboard_id).await,
board
);
e2e_delete_dashboard(&board.v5.unwrap().dashboard_id).await;
assert!(e2e_list_dashboards().await.is_empty());
}
// A panel whose column is wrapped in a cast must come back with the
// nesting intact — the model used to flatten it, which silently dropped
// the y column from the query on the next edit.
e2e_dashboard_nested_cast_round_trip().await;
e2e_alerts().await;
e2e_create_test_pipeline().await;
e2e_handle_derived_stream_success().await;
e2e_handle_derived_stream_pipeline_not_found().await;
e2e_handle_derived_stream_max_retries().await;
test_derived_stream_invalid_timerange_delay_scenario().await;
test_derived_stream_invalid_timerange_with_cron_frequency().await;
e2e_handle_derived_stream_evaluation_failure().await;
e2e_cleanup_test_pipeline().await;
test_enrichment_table_integration().await;
test_enrichment_table_local_all_sequential().await;
test_backfill_job_list_and_delete().await;
test_backfill_job_get_nonexistent().await;
test_backfill_job_delete_by_pipeline().await;
test_backfill_job_enable_disable().await;
// Own fn: inline spawns widen e2e_test's frame and overflow the debug stack.
e2e_org_deletion().await;
e2e_health_check().await;
e2e_config().await;
e2e_delete_stream().await;
sandbox.tear_down().await;
}
async fn e2e_boot() {
tokio::task::spawn(async move {
init_grpc_server()
.await
.expect("router gRPC server init failed");
});
common::infra::cluster::register_and_keep_alive()
.await
.unwrap();
config::init().await.unwrap();
migration::init_db().await.unwrap();
infra::db::create_table().await.unwrap();
infra::table::migrate().await.unwrap();
infra::init().await.unwrap();
#[cfg(feature = "enterprise")]
let master_key = Some(
o2_enterprise::enterprise::common::config::get_config()
.encryption
.master_key
.clone(),
);
#[cfg(not(feature = "enterprise"))]
let master_key: Option = None;
infra::table::cipher::boot(master_key.as_deref())
.await
.unwrap();
openobserve_core::bootstrap::init().await.unwrap();
ingester::init().await.unwrap();
openobserve_jobs::job::init().await.unwrap();
// Default user creation runs asynchronously after bootstrap.
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
// A retried run starts from the state the failed attempt left behind.
cleanup_previous_test_state().await;
}
async fn e2e_alerts() {
// alert
e2e_post_alert_template().await;
e2e_get_alert_template().await;
e2e_list_alert_template().await;
e2e_post_alert_destination().await;
e2e_get_alert_destination().await;
e2e_list_alert_destinations().await;
e2e_post_alert_multirange().await;
e2e_delete_alert_multirange().await;
e2e_post_alert().await;
e2e_get_alert().await;
e2e_handle_alert_after_destination_retries().await;
e2e_handle_alert_after_evaluation_retries().await;
e2e_handle_alert_reached_max_retries().await;
e2e_list_alerts().await;
e2e_list_real_time_alerts().await;
e2e_delete_alert().await;
e2e_delete_alert_destination().await;
e2e_delete_alert_template().await;
// Email-specific alert tests
e2e_post_alert_email_template().await;
e2e_get_alert_email_template().await;
e2e_post_alert_email_destination().await;
e2e_get_alert_email_destination().await;
e2e_post_alert_email_destination_should_fail().await;
e2e_delete_alert_email_destination().await;
e2e_delete_alert_email_template().await;
// Clean-up user here after email destinations deleted
e2e_delete_user().await;
// SNS-specific alert tests
// Set up templates
e2e_post_alert_template().await;
e2e_post_sns_alert_template().await;
// SNS destination tests
e2e_post_sns_alert_destination().await;
e2e_get_sns_alert_destination().await;
e2e_list_alert_destinations_with_sns().await;
e2e_update_sns_alert_destination().await;
// Create and test alert with SNS destination
e2e_post_alert_with_sns_destination().await;
// Cleanup
e2e_delete_alert_with_sns_destination().await;
e2e_delete_sns_alert_destination().await;
}
async fn e2e_org_deletion() {
// org deletion: each runs in its own task so one red test does not hide the others
let saved_config = config::CONFIG.load_full();
set_cleanup_test_config(0);
// A worker poll already sleeping when the pause lands still runs once.
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
let mut failed = Vec::new();
let file_list = tokio::spawn(e2e_file_list_step_moves_every_stream_to_deleted()).await;
if file_list.is_err() {
failed.push("e2e_file_list_step_moves_every_stream_to_deleted");
}
if tokio::spawn(e2e_public_paths_refuse_blocked_org())
.await
.is_err()
{
failed.push("e2e_public_paths_refuse_blocked_org");
}
#[cfg(feature = "enterprise")]
if tokio::spawn(e2e_incident_webhook_refuses_inactive_org())
.await
.is_err()
{
failed.push("e2e_incident_webhook_refuses_inactive_org");
}
if tokio::spawn(e2e_org_delete_completes()).await.is_err() {
failed.push("e2e_org_delete_completes");
}
if tokio::spawn(e2e_online_eval_runner_skips_deleting_org())
.await
.is_err()
{
failed.push("e2e_online_eval_runner_skips_deleting_org");
}
#[cfg(feature = "enterprise")]
if tokio::spawn(e2e_host_routed_status_page_refuses_blocked_org())
.await
.is_err()
{
failed.push("e2e_host_routed_status_page_refuses_blocked_org");
}
if tokio::spawn(e2e_org_delete_removes_memberships())
.await
.is_err()
{
failed.push("e2e_org_delete_removes_memberships");
}
if tokio::spawn(e2e_meta_prefix_reads_are_org_exact())
.await
.is_err()
{
failed.push("e2e_meta_prefix_reads_are_org_exact");
}
if tokio::spawn(e2e_org_delete_rejects_old_credentials())
.await
.is_err()
{
failed.push("e2e_org_delete_rejects_old_credentials");
}
if tokio::spawn(e2e_org_setting_writers_do_not_lose_updates())
.await
.is_err()
{
failed.push("e2e_org_setting_writers_do_not_lose_updates");
}
if tokio::spawn(e2e_ingestion_blocked_during_deletion())
.await
.is_err()
{
failed.push("e2e_ingestion_blocked_during_deletion");
}
#[cfg(feature = "enterprise")]
if tokio::spawn(test_initiate_refused_under_super_cluster())
.await
.is_err()
{
failed.push("test_initiate_refused_under_super_cluster");
}
if tokio::spawn(e2e_drain_retry_keeps_attempts_at_zero())
.await
.is_err()
{
failed.push("e2e_drain_retry_keeps_attempts_at_zero");
}
if tokio::spawn(e2e_drain_stalled_row_is_marked_failed())
.await
.is_err()
{
failed.push("e2e_drain_stalled_row_is_marked_failed");
}
if tokio::spawn(e2e_drain_keeps_rows_when_an_object_cannot_be_deleted())
.await
.is_err()
{
failed.push("e2e_drain_keeps_rows_when_an_object_cannot_be_deleted");
}
if tokio::spawn(e2e_drain_refuses_an_unloaded_storage_account())
.await
.is_err()
{
failed.push("e2e_drain_refuses_an_unloaded_storage_account");
}
if tokio::spawn(e2e_drain_failure_spends_an_attempt_each_pass())
.await
.is_err()
{
failed.push("e2e_drain_failure_spends_an_attempt_each_pass");
}
if tokio::spawn(e2e_resurrect_releases_parked_triggers())
.await
.is_err()
{
failed.push("e2e_resurrect_releases_parked_triggers");
}
if tokio::spawn(e2e_blocked_org_reaches_only_its_cleanup_retry())
.await
.is_err()
{
failed.push("e2e_blocked_org_reaches_only_its_cleanup_retry");
}
if tokio::spawn(e2e_compactor_pass_keeps_rows_of_a_deleting_org())
.await
.is_err()
{
failed.push("e2e_compactor_pass_keeps_rows_of_a_deleting_org");
}
if tokio::spawn(e2e_compactor_pass_refuses_an_unloaded_account_of_a_deleting_org())
.await
.is_err()
{
failed.push("e2e_compactor_pass_refuses_an_unloaded_account_of_a_deleting_org");
}
if tokio::spawn(e2e_compactor_removal_waits_for_deletion_start())
.await
.is_err()
{
failed.push("e2e_compactor_removal_waits_for_deletion_start");
}
if tokio::spawn(e2e_retry_backfills_missing_steps_and_resets_attempts())
.await
.is_err()
{
failed.push("e2e_retry_backfills_missing_steps_and_resets_attempts");
}
if tokio::spawn(e2e_scan_failed_sets_gauge_and_emits_once())
.await
.is_err()
{
failed.push("e2e_scan_failed_sets_gauge_and_emits_once");
}
if tokio::spawn(e2e_reconcile_requeues_deleting_org_without_tasks())
.await
.is_err()
{
failed.push("e2e_reconcile_requeues_deleting_org_without_tasks");
}
if tokio::spawn(e2e_delete_org_record_is_idempotent())
.await
.is_err()
{
failed.push("e2e_delete_org_record_is_idempotent");
}
if tokio::spawn(e2e_initiate_rolls_back_status_when_add_batch_fails())
.await
.is_err()
{
failed.push("e2e_initiate_rolls_back_status_when_add_batch_fails");
}
if tokio::spawn(e2e_org_delete_retry_after_exhaustion())
.await
.is_err()
{
failed.push("e2e_org_delete_retry_after_exhaustion");
}
if tokio::spawn(e2e_two_workers_never_run_one_step_twice())
.await
.is_err()
{
failed.push("e2e_two_workers_never_run_one_step_twice");
}
config::CONFIG.store(saved_config);
assert!(failed.is_empty(), "org deletion e2e failed: {failed:?}");
}
fn run_suffix() -> String {
(Utc::now().timestamp_micros() % 1_000_000_000).to_string()
}
/// Ingests one record so the org is auto-created (`ZO_CREATE_ORG_THROUGH_INGESTION`).
async fn ingest_one(app: &Router, headers: &HeaderMap, org: &str) {
let (status, body) = make_request(
app,
Method::POST,
&format!("/api/{org}/orgdel/_json"),
Some(headers.clone()),
Some(r#"[{"a":1}]"#.to_string()),
)
.await;
assert!(
status.is_success(),
"ingest into {org}: {status} {}",
String::from_utf8_lossy(&body)
);
flush_memtable().await;
}
/// Pauses the background cleanup and promotion jobs, so only the test drives them.
fn set_cleanup_test_config(delete_files_delay_minutes: i64) {
let mut cfg = config::config::init();
cfg.common.org_cleanup_poll_interval_secs = 0;
// Set here, not by env: config raises a delay below 1 minute to 120.
cfg.compact.delete_files_delay_minutes = delete_files_delay_minutes;
config::CONFIG.store(Arc::new(cfg));
}
async fn run_cleanup_once() {
let cleanup: Arc =
Arc::new(openobserve_jobs::job::CompactionStreamDataCleanup);
openobserve_core::org_cleanup::run_once(cleanup).await;
}
async fn cleanup_task(org: &str, step: &str) -> infra::table::org_cleanup_tasks::CleanupTask {
infra::table::org_cleanup_tasks::list_by_org_status(org, None)
.await
.unwrap()
.into_iter()
.find(|t| t.step == step)
.unwrap_or_else(|| panic!("{org} has no {step} task"))
}
async fn add_tasks(org: &str, steps: &[(&str, i32)]) {
let tasks: Vec<_> = steps
.iter()
.map(
|(step, order)| infra::table::org_cleanup_tasks::NewCleanupTask {
org_id: org.to_string(),
org_name: org.to_string(),
step: step.to_string(),
step_order: *order,
},
)
.collect();
infra::table::org_cleanup_tasks::add_batch(&tasks)
.await
.unwrap();
}
/// Spends every attempt of the step the way the worker does: claim, then fail.
async fn exhaust_task(org: &str, step: &str) {
let id = cleanup_task(org, step).await.id;
for _ in 0..10 {
let claim = infra::table::org_cleanup_tasks::mark_running(&id)
.await
.unwrap()
.expect("an idle task can be claimed");
assert!(
infra::table::org_cleanup_tasks::mark_failed(&claim, "e2e failure")
.await
.unwrap()
);
}
}
fn deleted_file(org: &str, name: &str) -> config::meta::stream::FileListDeleted {
config::meta::stream::FileListDeleted {
file: format!("files/{org}/logs/s1/2026/01/01/00/{name}.parquet"),
..Default::default()
}
}
async fn remove_deleted_rows(org: &str) {
let rows = infra::file_list::query_deleted(org, i64::MAX, 1000)
.await
.unwrap();
let keys: Vec<_> = rows
.into_iter()
.map(|r| {
config::meta::stream::FileKey::new(
r.id,
r.account,
r.file,
config::meta::stream::FileMeta::default(),
false,
)
})
.collect();
infra::file_list::batch_remove_deleted(&keys).await.unwrap();
}
async fn add_deleting_org(org: &str) {
infra::table::organizations::add(
org,
org,
config::meta::organization::OrganizationType::Default,
)
.await
.unwrap();
infra::table::organizations::set_status(org, "deleting")
.await
.unwrap();
}
async fn org_status_of(org: &str) -> Option {
infra::table::organizations::get_status_uncached(org)
.await
.unwrap()
}
fn failed_steps_gauge(org: &str) -> i64 {
config::metrics::ORG_CLEANUP_FAILED_STEPS
.with_label_values(&[org])
.get()
}
/// A stored file of a stream without a schema, so only steps 200 and 250 can remove it.
/// An object and its file_list row in a stream with no schema or stats row.
async fn seed_orphan_file(org: &str, stream: &str) -> String {
let key = format!("files/{org}/logs/{stream}/2026/01/01/00/a.parquet");
infra::storage::put("", &key, Bytes::from_static(b"orphan"))
.await
.unwrap();
let meta = config::meta::stream::FileMeta {
min_ts: 1,
max_ts: 2,
records: 1,
original_size: 6,
compressed_size: 6,
..Default::default()
};
infra::file_list::add("", &key, &meta).await.unwrap();
key
}
async fn seed_synthetic_with_children(org: &str) -> String {
use config::meta::folder::{Folder, FolderType};
use infra::table::{synthetics_checks, synthetics_jobs, synthetics_refs, synthetics_runs};
let folder = Folder {
folder_id: "default".to_string(),
name: "default".to_string(),
..Default::default()
};
let (folder_pk, ..) =
infra::table::folders::get_or_create(org, folder, FolderType::Synthetics)
.await
.unwrap();
let check = config::meta::synthetics::Synthetic {
name: "orgdel check".to_string(),
target: "https://example.com".to_string(),
folder_id: folder_pk.to_string(),
..Default::default()
};
let conn = infra::db::get_orm_client_rw().await;
let check = synthetics_checks::create(conn, org, check, false)
.await
.unwrap();
let now = Utc::now().timestamp_micros();
let run_id = config::ider::generate();
synthetics_runs::insert_run(
conn,
synthetics_runs::InsertRunParams {
id: &run_id,
synthetics_id: &check.id,
org_id: org,
scheduled_ts: now,
trigger_type: "manual",
job_count: 1,
created_at: now,
},
)
.await
.unwrap();
synthetics_jobs::enqueue(
conn,
synthetics_jobs::EnqueueParams {
synthetics_id: &check.id,
synthetics_name: &check.name,
org_id: org,
location: "local",
env: None,
pool: "default",
scheduled_ts: now,
valid_until: now + 60_000_000,
run_id: &run_id,
browser_devices: None,
steps_configured: 1,
metadata: "{}",
},
)
.await
.unwrap();
synthetics_refs::replace_for_parent(conn, org, &check.id, &["orgdel-child".to_string()])
.await
.unwrap();
check.id
}
async fn count_org_rows(table: &str, org: &str) -> i64 {
use sea_orm::{ConnectionTrait, Statement};
let conn = infra::db::get_orm_client_ro().await;
let row = conn
.query_one(Statement::from_sql_and_values(
conn.get_database_backend(),
format!("SELECT COUNT(*) AS n FROM {table} WHERE org_id = $1"),
[org.into()],
))
.await
.unwrap()
.unwrap();
row.try_get("", "n").unwrap()
}
async fn count_rows(table: &str, column: &str, value: &str) -> i64 {
use sea_orm::{ConnectionTrait, Statement};
let conn = infra::db::get_orm_client_ro().await;
let row = conn
.query_one(Statement::from_sql_and_values(
conn.get_database_backend(),
format!("SELECT COUNT(*) AS n FROM {table} WHERE {column} = $1"),
[value.into()],
))
.await
.unwrap()
.unwrap();
row.try_get("", "n").unwrap()
}
async fn insert_entity_row(row: serde_json::Value)
where
E: sea_orm::EntityTrait,
E::Model: serde::de::DeserializeOwned + sea_orm::IntoActiveModel,
{
use sea_orm::IntoActiveModel;
let model: E::Model = serde_json::from_value(row).unwrap();
E::insert(model.into_active_model())
.exec_without_returning(infra::db::get_orm_client_rw().await)
.await
.unwrap();
}
async fn seed_coverage_rows(org: &str) -> CoverageRows {
let check_id = seed_synthetic_with_children(org).await;
seed_status_page_rows(org, &check_id).await;
let integration_id = seed_incident_rows(org).await;
let (probe_token, agent_id, location_id) = seed_synthetics_registry_rows(org).await;
let workflow_id = seed_workflow_rows(org).await;
let anomaly_id = seed_anomaly_rows(org).await;
let artifact = format!("synthetics/{org}/{check_id}/run1/screenshot.png");
infra::storage::put("", &artifact, Bytes::from_static(b"png"))
.await
.unwrap();
let pricing_model = seed_llm_rows(org).await;
let (team_id, response_id, policy_id) = seed_oncall_rows(org).await;
seed_ai_chat_rows(org).await;
let dek = infra::table::cipher::get_dek(org).await.unwrap();
CoverageRows {
check_id,
integration_id,
anomaly_id,
workflow_id,
probe_token,
agent_id,
location_id,
artifact,
pricing_model,
team_id,
response_id,
policy_id,
dek,
}
}
async fn insert_active(row: A)
where
A: sea_orm::ActiveModelTrait,
::Model: sea_orm::IntoActiveModel,
{
use sea_orm::EntityTrait;
::insert(row)
.exec_without_returning(infra::db::get_orm_client_rw().await)
.await
.unwrap();
}
async fn seed_llm_rows(org: &str) -> String {
use infra::table::entity::{
llm_annotation_queue_bindings, llm_annotation_queue_items, llm_annotation_queues,
llm_dataset_items, llm_datasets, llm_experiment_slot_retries, llm_experiments,
llm_idempotency_records, llm_playground_snapshots, score_configs,
};
use sea_orm::Set;
let now = Utc::now().timestamp_micros();
let id = config::ider::generate;
let (dataset, queue, score_config, experiment) = (id(), id(), id(), id());
insert_active(llm_datasets::ActiveModel {
id: Set(dataset.clone()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {dataset}")),
global_version: Set(1),
created_by: Set("root@example.com".to_string()),
created_at: Set(now),
updated_by: Set("root@example.com".to_string()),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_dataset_items::ActiveModel {
row_id: Set(id()),
logical_id: Set(id()),
org_id: Set(org.to_string()),
dataset_id: Set(dataset.clone()),
input: Set(json!({"q": "a"})),
expected_output: Set(json!({"a": "b"})),
global_version: Set(1),
is_deleted: Set(false),
source: Set("manual".to_string()),
updated_by: Set("root@example.com".to_string()),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_annotation_queues::ActiveModel {
id: Set(queue.clone()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {queue}")),
allowed_ref_types: Set(json!(["trace"])),
created_by: Set("root@example.com".to_string()),
created_at: Set(now),
updated_by: Set("root@example.com".to_string()),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_annotation_queue_items::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
queue_id: Set(queue.clone()),
ref_type: Set("trace".to_string()),
ref_id: Set(id()),
ref_trace_start_time: Set(now),
status: Set("pending".to_string()),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(score_configs::ActiveModel {
id: Set(score_config.clone()),
org_id: Set(org.to_string()),
entity_id: Set(id()),
name: Set(format!("orgdel {score_config}")),
version: Set(1),
data_type: Set("numeric".to_string()),
is_active: Set(true),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_annotation_queue_bindings::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
queue_id: Set(queue),
score_config_row_id: Set(score_config),
created_at: Set(now),
})
.await;
insert_active(llm_experiments::ActiveModel {
id: Set(experiment.clone()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {experiment}")),
dataset_id: Set(dataset),
dataset_version: Set(1),
task_config: Set(json!({})),
scorers: Set(json!([])),
trial_count: Set(1),
status: Set("completed".to_string()),
deadline_at: Set(now),
lifecycle_version: Set(1),
retry_count: Set(0),
is_baseline: Set(false),
created_by: Set("root@example.com".to_string()),
created_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_experiment_slot_retries::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
experiment_id: Set(experiment),
row_id: Set(id()),
trial_index: Set(0),
idempotency_key: Set(id()),
state: Set("done".to_string()),
execution: Set(json!({})),
created_at: Set(now),
updated_at: Set(now),
expires_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_idempotency_records::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
scope: Set("experiment".to_string()),
idempotency_key: Set(id()),
request_hash: Set("h".to_string()),
response: Set(json!({})),
created_at: Set(now),
expires_at: Set(now),
})
.await;
insert_active(llm_playground_snapshots::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
payload: Set(json!({})),
created_by: Set("root@example.com".to_string()),
created_at: Set(now),
last_accessed_at: Set(now),
..Default::default()
})
.await;
seed_eval_job(org, now).await;
seed_ai_rows(org, now).await;
seed_pricing_and_rate_limit(org, now).await
}
async fn seed_eval_job(org: &str, now: i64) -> String {
use infra::table::entity::online_eval_jobs;
use sea_orm::Set;
let eval_job_id = config::ider::generate();
insert_active(online_eval_jobs::ActiveModel {
id: Set(eval_job_id.clone()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {eval_job_id}")),
stream: Set("orgdel".to_string()),
stream_type: Set("traces".to_string()),
target_scope: Set("trace".to_string()),
filter_condition: Set(json!({})),
scorers: Set(json!([])),
sampling_mode: Set("rate".to_string()),
sampling_value: Set(json!(0)),
status: Set("active".to_string()),
version: Set(1),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
eval_job_id
}
async fn seed_ai_rows(org: &str, now: i64) {
use infra::table::entity::{
gen_ai_agents, llm_secrets, org_ai_toolsets, providers, scorers,
};
use sea_orm::Set;
let id = config::ider::generate;
insert_active(scorers::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
entity_id: Set(id()),
name: Set(format!("orgdel {}", id())),
version: Set(1),
scorer_type: Set(scorers::ScorerType::LlmJudge),
template: Set("t".to_string()),
params: Set(json!({})),
is_active: Set(true),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(providers::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {}", id())),
provider_type: Set("openai".to_string()),
default_model: Set("gpt".to_string()),
available_models: Set(json!(["gpt"])),
is_default: Set(false),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(llm_secrets::ActiveModel {
id: Set(id()),
secret_ref: Set(id()),
org_id: Set(org.to_string()),
owner_kind: Set("provider".to_string()),
owner_id: Set(id()),
purpose: Set("api_key".to_string()),
state: Set("active".to_string()),
ciphertext: Set("c".to_string()),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(org_ai_toolsets::ActiveModel {
id: Set(id()),
org: Set(org.to_string()),
name: Set(format!("orgdel {}", id())),
kind: Set("mcp".to_string()),
created_by: Set("root@example.com".to_string()),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(gen_ai_agents::ActiveModel {
agent_key: Set(id()),
org_id: Set(org.to_string()),
stream_type: Set("traces".to_string()),
stream_name: Set("orgdel".to_string()),
identity_source: Set("attribute".to_string()),
first_seen: Set(now),
last_seen: Set(now),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
#[cfg(feature = "enterprise")]
insert_active(infra::table::entity::llm_remote_tasks::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
entity_id: Set(id()),
name: Set(format!("orgdel {}", id())),
version: Set(1),
endpoint: Set("https://dummy/task".to_string()),
http_method: Set("POST".to_string()),
auth: Set(json!({})),
custom_headers: Set(json!({})),
content_type: Set("application/json".to_string()),
response_schema: Set("{}".to_string()),
timeout_ms: Set(1000),
retry_policy: Set(json!({})),
max_concurrency: Set(1),
signing: Set(json!({})),
verification_status: Set("unverified".to_string()),
is_active: Set(false),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
}
async fn seed_pricing_and_rate_limit(org: &str, now: i64) -> String {
use config::meta::model_pricing::{ModelPricingDefinition, PricingTierDefinition};
use sea_orm::Set;
let pricing_model = format!("orgdel-model-{org}");
infra::table::model_pricing::put(ModelPricingDefinition {
org_id: org.to_string(),
name: pricing_model.clone(),
match_pattern: "(?i)^orgdel".to_string(),
enabled: true,
tiers: vec![PricingTierDefinition {
name: "Default".to_string(),
condition: None,
prices: Default::default(),
utc_windows: Vec::new(),
rules: Vec::new(),
priority: 0,
}],
..Default::default()
})
.await
.unwrap();
insert_active(infra::table::entity::rate_limit_rules::ActiveModel {
rule_id: Set(config::ider::generate()),
org: Set(org.to_string()),
rule_type: Set("exact".to_string()),
user_role: Set("admin".to_string()),
user_id: Set(".*".to_string()),
api_group_name: Set("logs".to_string()),
api_group_operation: Set("list".to_string()),
threshold: Set(10),
stat_interval_ms: Set(1000),
created_at: Set(now),
})
.await;
pricing_model
}
/// A chat with a revoked share and a turn record, as a persisted AI chat leaves them.
async fn seed_ai_chat_rows(org: &str) {
use infra::table::{ai_chat_sessions, ai_chat_shares, ai_chat_turns};
let now = Utc::now().timestamp_micros();
let session = config::ider::uuid();
ai_chat_sessions::get_or_create(org, &session, "orgdel-user", "a@x", "o2-ai", now)
.await
.unwrap();
let share = ai_chat_shares::NewShare {
org_id: org,
session_id: &session,
created_by: "orgdel-user",
mode: ai_chat_shares::MODE_LIVE,
snapshot_seq: None,
visibility: ai_chat_shares::VISIBILITY_ORG,
expires_at: None,
redact_tools: false,
};
let share = ai_chat_shares::insert(&share, now).await.unwrap();
ai_chat_shares::revoke(org, &share.id, now).await.unwrap();
ai_chat_turns::admit(org, &session, "orgdel-turn", 0, now)
.await
.unwrap();
// Deleted by their user before: one the sweep still re-purges, one past that window.
let window = ai_chat_sessions::REPLICA_REPURGE_WINDOW_MICROS;
for purged_at in [now, now - window - 1] {
let session = config::ider::uuid();
ai_chat_sessions::get_or_create(org, &session, "orgdel-user", "a@x", "o2-ai", now)
.await
.unwrap();
assert!(
ai_chat_sessions::mark_deleted(org, &session, "orgdel-user", now)
.await
.unwrap()
);
ai_chat_sessions::mark_replica_purged(org, &session, purged_at)
.await
.unwrap();
}
}
/// Chat rows step 300 removes: all but deleted chats the retention sweep still re-purges.
async fn count_unheld_chats(org: &str) -> i64 {
use sea_orm::{ConnectionTrait, Statement};
let repurge_since = Utc::now().timestamp_micros()
- infra::table::ai_chat_sessions::REPLICA_REPURGE_WINDOW_MICROS;
let conn = infra::db::get_orm_client_ro().await;
let row = conn
.query_one(Statement::from_sql_and_values(
conn.get_database_backend(),
"SELECT COUNT(*) AS n FROM ai_chat_sessions WHERE org_id = $1 \
AND (status <> 'deleted' OR replica_purged_at IS NULL OR replica_purged_at <= $2)",
[org.into(), repurge_since.into()],
))
.await
.unwrap()
.unwrap();
row.try_get("", "n").unwrap()
}
async fn seed_oncall_rows(org: &str) -> (String, String, String) {
use infra::table::entity::{
oncall_delivery_reads, oncall_overrides, oncall_ownership_rules, oncall_policies,
oncall_response_events, oncall_response_reports, oncall_responses,
oncall_routing_config, oncall_schedules, oncall_team_members, oncall_teams,
oncall_unavailability, oncall_unrouted_signals, oncall_user_contacts,
};
use sea_orm::Set;
let now = Utc::now().timestamp_micros();
let id = config::ider::generate;
let who = || "oncall@example.com".to_string();
let (team, response, policy) = (id(), id(), id());
insert_active(oncall_teams::ActiveModel {
id: Set(team.clone()),
org_id: Set(org.to_string()),
name: Set(format!("orgdel {team}")),
timezone: Set("UTC".to_string()),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_team_members::ActiveModel {
id: Set(id()),
team_id: Set(team.clone()),
user_email: Set(who()),
created_at: Set(now),
})
.await;
insert_active(oncall_schedules::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
team_id: Set(team.clone()),
timezone: Set("UTC".to_string()),
rotations: Set("[]".to_string()),
created_at: Set(now),
updated_at: Set(now),
})
.await;
insert_active(oncall_overrides::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
team_id: Set(team.clone()),
rotation_id: Set(id()),
user_email: Set(who()),
start_at: Set(now),
end_at: Set(now + 1),
created_by: Set(who()),
created_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_policies::ActiveModel {
id: Set(policy.clone()),
org_id: Set(org.to_string()),
team_id: Set(team.clone()),
rungs: Set("[]".to_string()),
destinations: Set("[]".to_string()),
l0_json: Set("{}".to_string()),
created_at: Set(now),
updated_at: Set(now),
})
.await;
insert_active(oncall_ownership_rules::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
team_id: Set(team.clone()),
path: Set("orgdel".to_string()),
dimensions: Set("{}".to_string()),
created_at: Set(now),
updated_at: Set(now),
})
.await;
insert_active(oncall_routing_config::ActiveModel {
org_id: Set(org.to_string()),
updated_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_responses::ActiveModel {
id: Set(response.clone()),
org_id: Set(org.to_string()),
subject_type: Set(0),
subject_id: Set(id()),
responder_role: Set(0),
priority: Set(0),
state: Set(0),
opened_at: Set(now),
updated_at: Set(now),
..Default::default()
})
.await;
let event = id();
insert_active(oncall_response_events::ActiveModel {
id: Set(event.clone()),
response_id: Set(response.clone()),
kind: Set(0),
at: Set(now),
actor: Set(who()),
body: Set("{}".to_string()),
..Default::default()
})
.await;
insert_active(oncall_response_reports::ActiveModel {
org_id: Set(org.to_string()),
response_id: Set(response.clone()),
report: Set("{}".to_string()),
generated_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_delivery_reads::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
user_email: Set(who()),
event_id: Set(event),
read_at: Set(now),
})
.await;
insert_active(oncall_unavailability::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
user_email: Set(who()),
start_at: Set(now),
end_at: Set(now + 1),
created_by: Set(who()),
created_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_unrouted_signals::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
path: Set(format!("orgdel/{}", id())),
dimensions: Set("{}".to_string()),
occurrences: Set(1),
first_seen_at: Set(now),
last_seen_at: Set(now),
..Default::default()
})
.await;
insert_active(oncall_user_contacts::ActiveModel {
id: Set(id()),
org_id: Set(org.to_string()),
user_email: Set(who()),
updated_at: Set(now),
..Default::default()
})
.await;
(team, response, policy)
}
async fn seed_status_page_rows(org: &str, check_id: &str) {
use infra::table::entity::{
status_page_component_checks, status_page_components, status_page_notice_components,
status_page_notice_updates, status_page_notices,
};
let slug = insert_public_status_page(org).await;
let page = infra::table::status_pages::get_page_by_slug_any(&slug)
.await
.unwrap()
.unwrap();
let now = Utc::now().timestamp_micros();
let (component, notice) = (config::ider::generate(), config::ider::generate());
insert_entity_row::(json!({
"id": component, "status_page_id": page.id, "org_id": org, "name": "api",
"sort_order": 0, "created_at": now, "updated_at": now,
}))
.await;
insert_entity_row::(json!({
"id": config::ider::generate(), "component_id": component,
"synthetics_id": check_id, "org_id": org,
}))
.await;
insert_entity_row::(json!({
"id": notice, "org_id": org, "kind": 0, "impact": 1, "source": 0,
"title": "outage", "body": "", "state": 0, "starts_at": now, "segments": "[]",
"excluded_from_uptime": false, "auto_recovery_streak": 0,
"created_at": now, "updated_at": now,
}))
.await;
insert_entity_row::(json!({
"id": config::ider::generate(), "notice_id": notice, "component_id": component,
"org_id": org,
}))
.await;
insert_entity_row::(json!({
"id": config::ider::generate(), "notice_id": notice, "org_id": org,
"body": "investigating", "created_at": now,
}))
.await;
infra::table::status_pages::upsert_check_snooze(org, check_id, now + 60_000_000, None, now)
.await
.unwrap();
infra::table::status_pages::insert_audit(
org,
"root@example.com",
"create",
Some(¬ice),
now,
)
.await
.unwrap();
let claim = infra::table::entity::status_page_custom_domains::Model {
id: config::ider::generate(),
org_id: org.to_string(),
status_page_id: page.id,
domain: format!("{org}.status.example.com"),
verification_token: config::ider::generate(),
verification_state: 1,
verification_failure_reason: None,
verified_at: Some(now),
released_at: None,
last_checked_at: None,
created_at: now,
updated_at: now,
};
assert!(
infra::table::status_pages::insert_domain_if_unclaimed(&claim)
.await
.unwrap()
);
}
async fn seed_incident_rows(org: &str) -> String {
use infra::table::incident_integrations;
use sea_orm::{EntityTrait, IntoActiveModel};
let now = Utc::now().timestamp_micros();
let integration = incident_integrations::IncidentIntegrationRecord {
id: config::ider::generate(),
org_id: org.to_string(),
name: "orgdel".to_string(),
source_type: "generic".to_string(),
token: incident_integrations::generate_token(),
enabled: true,
config: json!({}),
destinations: vec![],
created_by: "root@example.com".to_string(),
created_at: now,
updated_at: now,
};
incident_integrations::add(&integration).await.unwrap();
incident_integrations::touch_sender(incident_integrations::TouchSenderParams {
integration_id: &integration.id,
detected_source: "generic",
sender_label: None,
now,
accepted: 1,
rejected: 0,
saw_resolved: false,
})
.await
.unwrap();
let alert = infra::table::entity::external_alerts::Model {
id: config::ider::generate(),
org_id: org.to_string(),
integration_id: integration.id.clone(),
detected_source: "generic".to_string(),
dedup_key: "orgdel".to_string(),
title: "orgdel".to_string(),
severity: "critical".to_string(),
state: "firing".to_string(),
labels: "{}".to_string(),
source_url: None,
last_payload: "{}".to_string(),
first_seen_at: now,
last_seen_at: now,
resolved_at: None,
};
infra::table::entity::external_alerts::Entity::insert(alert.into_active_model())
.exec_without_returning(infra::db::get_orm_client_rw().await)
.await
.unwrap();
integration.id
}
async fn seed_synthetics_registry_rows(org: &str) -> (String, String, String) {
use infra::table::{
synthetics_agents, synthetics_environments, synthetics_locations,
synthetics_probe_tokens, synthetics_variables,
};
let now = Utc::now().timestamp_micros();
let has_default = synthetics_probe_tokens::find_default(org).await.unwrap();
let token = synthetics_probe_tokens::SyntheticsProbeTokenRecord {
id: config::ider::uuid(),
org_id: org.to_string(),
name: "orgdel".to_string(),
token: synthetics_probe_tokens::generate_token(),
is_default: has_default.is_none(),
enabled: true,
created_by: "root@example.com".to_string(),
created_at: now,
updated_at: now,
};
synthetics_probe_tokens::add(&token).await.unwrap();
let conn = infra::db::get_orm_client_rw().await;
let (env, _) = synthetics_environments::get_or_create_global(conn, org)
.await
.unwrap();
let variable = synthetics_variables::SyntheticsVariableRecord {
id: config::ider::generate(),
org_id: org.to_string(),
env: env.id,
name: "orgdel".to_string(),
value: "1".to_string(),
kind: synthetics_variables::KIND_PLAIN.to_string(),
description: String::new(),
example: String::new(),
tags: vec![],
owner: None,
created_at: now,
updated_at: now,
};
synthetics_variables::add(conn, &variable).await.unwrap();
let location_id = config::ider::generate();
synthetics_locations::add(&synthetics_locations::SyntheticsLocationRecord {
id: location_id.clone(),
org_id: Some(org.to_string()),
kind: synthetics_locations::KIND_PRIVATE.to_string(),
provider: "self".to_string(),
region: "local".to_string(),
label: format!("{org} private"),
pool: format!("private-{location_id}"),
enabled: false,
created_at: now,
updated_at: now,
})
.await
.unwrap();
let agent_id = config::ider::generate();
synthetics_agents::register(&synthetics_agents::SyntheticsAgentRecord {
id: agent_id.clone(),
org_id: org.to_string(),
location_id: location_id.clone(),
name: "orgdel-agent".to_string(),
version: None,
capabilities: None,
token_id: Some(token.id),
last_seen_at: now,
created_at: now,
})
.await
.unwrap();
(token.token, agent_id, location_id)
}
async fn seed_workflow_rows(org: &str) -> String {
use config::meta::folder::{Folder, FolderType};
use infra::table::workflows;
let folder = Folder {
folder_id: "default".to_string(),
name: "default".to_string(),
..Default::default()
};
let (folder_pk, ..) =
infra::table::folders::get_or_create(org, folder, FolderType::Workflows)
.await
.unwrap();
let now = Utc::now().timestamp_micros();
let workflow = |id: &str| workflows::Workflow {
id: id.to_string(),
org_id: org.to_string(),
folder_id: folder_pk.to_string(),
created_at: now,
updated_at: now,
created_by: "root@example.com".to_string(),
enabled: false,
name: format!("orgdel {id}"),
description: String::new(),
nodes: vec![],
edges: vec![],
};
let id = config::ider::generate();
workflows::save_workflow(workflow(&id)).await.unwrap();
workflows::save_draft(workflow(&config::ider::generate()))
.await
.unwrap();
let run_id = config::ider::generate();
workflows::save_workflow_errors(workflows::WorkflowRunErrors {
id: 0,
cluster: "local".to_string(),
org_id: org.to_string(),
workflow_id: id.clone(),
run_id: run_id.clone(),
ran_at: now,
data: vec![],
input_data: None,
})
.await
.unwrap();
workflows::save_workflow_run_data(workflows::WorkflowRunData {
id: 0,
org_id: org.to_string(),
workflow_id: id.clone(),
run_id,
triggered_at: now,
data: "{}".to_string(),
})
.await
.unwrap();
workflows::add_workflow_association(workflows::WorkflowAssociation {
id: 0,
org_id: org.to_string(),
entity_id: "orgdel-alert".to_string(),
entity_type: "alert".to_string(),
workflow_id: id.clone(),
trigger_type: "manual".to_string(),
created_at: now,
})
.await
.unwrap();
id
}
async fn seed_anomaly_rows(org: &str) -> String {
use config::meta::folder::{Folder, FolderType};
use infra::table::entity::{anomaly_detection_config, anomaly_detection_models};
let folder = Folder {
folder_id: "default".to_string(),
name: "default".to_string(),
..Default::default()
};
let (folder_pk, ..) = infra::table::folders::get_or_create(org, folder, FolderType::Alerts)
.await
.unwrap();
let now = Utc::now().timestamp_micros();
let anomaly_id = config::ider::generate();
insert_entity_row::(json!({
"anomaly_id": anomaly_id, "org_id": org, "stream_name": "orgdel",
"stream_type": "logs", "enabled": false, "name": format!("orgdel {anomaly_id}"),
"query_mode": "filters", "detection_function": "count",
"histogram_interval": "5m", "schedule_interval": "5m",
"detection_window_seconds": 3600, "training_window_days": 7,
"retrain_interval_days": 1, "threshold": 97, "seasonality": "none",
"is_trained": true, "current_model_version": 1, "rcf_num_trees": 50,
"rcf_tree_size": 256, "rcf_shingle_size": 1, "alert_enabled": false,
"folder_id": folder_pk.to_string(), "status": 4, "retries": 0,
"last_updated": now, "created_at": now, "updated_at": now,
}))
.await;
insert_entity_row::(json!({
"anomaly_id": anomaly_id, "version": 1,
"s3_path": format!("s3://bucket/anomaly_models/{org}/{anomaly_id}/1.bin.zst"),
"s3_bucket": "bucket", "model_size_bytes": 1, "training_start_time": now,
"training_end_time": now, "training_data_points": 1, "created_at": now,
}))
.await;
#[cfg(feature = "enterprise")]
infra::storage::put(
"",
&format!("anomaly_models/{org}/{anomaly_id}/1.bin.zst"),
Bytes::from_static(b"model"),
)
.await
.unwrap();
anomaly_id
}
async fn coverage_counts(org: &str, rows: &CoverageRows) -> Vec<(&'static str, i64)> {
let org_tables = [
"ai_chat_shares",
"ai_chat_turns",
"status_pages",
"status_page_snapshots",
"status_page_components",
"status_page_component_checks",
"status_page_notices",
"status_page_notice_components",
"status_page_notice_updates",
"status_page_check_snoozes",
"status_page_audit_log",
"status_page_custom_domains",
"incident_integrations",
"external_alerts",
"synthetics",
"synthetics_runs",
"synthetics_jobs",
"synthetics_refs",
"synthetics_probe_tokens",
"synthetics_variables",
"synthetics_environments",
"synthetics_locations",
"synthetics_agents",
"workflows",
"workflow_drafts",
"workflow_associations",
"workflow_errors",
"workflow_run_data",
"anomaly_detection_config",
"llm_annotation_queue_bindings",
"llm_annotation_queue_items",
"llm_annotation_queues",
"llm_dataset_items",
"llm_datasets",
"llm_experiment_slot_retries",
"llm_experiments",
"llm_idempotency_records",
"llm_playground_snapshots",
"score_configs",
"scorers",
"online_eval_jobs",
"providers",
"llm_secrets",
"gen_ai_agents",
"oncall_teams",
"oncall_schedules",
"oncall_overrides",
"oncall_policies",
"oncall_ownership_rules",
"oncall_routing_config",
"oncall_responses",
"oncall_response_reports",
"oncall_delivery_reads",
"oncall_unavailability",
"oncall_unrouted_signals",
"oncall_user_contacts",
];
let mut counts = vec![("ai_chat_sessions", count_unheld_chats(org).await)];
for table in org_tables {
counts.push((table, count_org_rows(table, org).await));
}
#[cfg(feature = "enterprise")]
counts.push((
"llm_remote_tasks",
count_org_rows("llm_remote_tasks", org).await,
));
for table in [
"org_ai_toolsets",
"model_pricing",
"rate_limit_rules",
"cipher_keys",
] {
counts.push((table, count_rows(table, "org", org).await));
}
counts.push((
"oncall_team_members",
count_rows("oncall_team_members", "team_id", &rows.team_id).await,
));
counts.push((
"oncall_response_events",
count_rows("oncall_response_events", "response_id", &rows.response_id).await,
));
// These two tables carry no org column, so they are counted by the parent row's id.
counts.push((
"incident_integration_senders",
count_rows(
"incident_integration_senders",
"integration_id",
&rows.integration_id,
)
.await,
));
counts.push((
"anomaly_detection_models",
count_rows("anomaly_detection_models", "anomaly_id", &rows.anomaly_id).await,
));
counts
}
async fn object_exists(key: &str) -> bool {
infra::storage::head("", key).await.is_ok()
}
async fn assert_coverage(org: &str, rows: &CoverageRows, present: bool) {
for (table, n) in coverage_counts(org, rows).await {
assert_eq!(n > 0, present, "{table} holds {n} rows of {org}");
}
let workflow = infra::table::workflows::get_by_org_wid(org, &rows.workflow_id).await;
assert_eq!(workflow.unwrap().is_some(), present, "{}", rows.workflow_id);
assert_eq!(
object_exists(&rows.artifact).await,
present,
"{}",
rows.artifact
);
#[cfg(feature = "enterprise")]
{
let model = format!("anomaly_models/{org}/{}/1.bin.zst", rows.anomaly_id);
assert_eq!(object_exists(&model).await, present, "{model}");
}
}
async fn warm_coverage_caches(org: &str, rows: &CoverageRows) {
use infra::table::{
synthetics_agents, synthetics_checks, synthetics_locations, synthetics_probe_tokens,
synthetics_variables,
};
let conn = infra::db::get_orm_client_ro().await;
let cached = synthetics_checks::get_cached(conn, org, &rows.check_id).await;
assert!(cached.unwrap().is_some());
let token = synthetics_probe_tokens::find_global(&rows.probe_token).await;
assert!(token.unwrap().is_some());
let default_token = synthetics_probe_tokens::find_default(org).await;
assert!(default_token.unwrap().is_some());
let variables = synthetics_variables::list_cached(conn, org).await;
assert!(!variables.unwrap().is_empty());
let location = synthetics_locations::get(&rows.location_id).await;
assert!(location.unwrap().is_some());
let agent = synthetics_agents::get_cached(&rows.agent_id).await;
assert!(agent.unwrap().is_some());
#[cfg(feature = "enterprise")]
{
let workflow = db::workflows::get_workflow(org, &rows.workflow_id).await;
assert!(workflow.unwrap().is_some());
}
warm_llm_and_oncall_caches(org, rows).await;
}
async fn warm_llm_and_oncall_caches(org: &str, rows: &CoverageRows) {
use infra::table::{
cipher, oncall_ownership, oncall_policies, oncall_schedules, oncall_teams,
};
db::model_pricing::cache().await.unwrap();
let pricing = db::model_pricing::get_org_pricing_entries(org);
assert!(
pricing
.iter()
.any(|e| e.definition.name == rows.pricing_model)
);
let team = oncall_teams::get_cached(org, &rows.team_id).await;
assert!(team.unwrap().is_some());
assert!(oncall_teams::has_any_cached(org).await.unwrap());
let members = oncall_teams::list_members_cached(&rows.team_id).await;
assert!(!members.unwrap().is_empty());
let schedule = oncall_schedules::get_by_team_cached(org, &rows.team_id).await;
assert!(schedule.unwrap().is_some());
let rules = oncall_ownership::list_cached(org).await;
assert!(!rules.unwrap().is_empty());
let policy = oncall_policies::get_or_create_cached(org, &rows.team_id).await;
assert_eq!(policy.unwrap().id, rows.policy_id);
assert_eq!(cipher::get_dek(org).await.unwrap(), rows.dek);
#[cfg(feature = "enterprise")]
{
let org_rules = infra::table::ratelimit::list(org, None, None)
.await
.unwrap();
assert!(!org_rules.is_empty());
o2_ratelimit::dataresource::db::RATELIMIT_RULES_CACHE
.write()
.await
.extend(org_rules);
}
}
/// Each read below serves a cache hit while the entry lives, so `None` proves the eviction.
async fn assert_coverage_caches_evicted(
org: &str,
rows: &CoverageRows,
warmed: std::time::Instant,
) {
use infra::table::{
synthetics_agents, synthetics_checks, synthetics_locations, synthetics_probe_tokens,
synthetics_variables,
};
let since = warmed.elapsed();
let conn = infra::db::get_orm_client_ro().await;
let cached = synthetics_checks::get_cached(conn, org, &rows.check_id).await;
assert!(
cached.unwrap().is_none(),
"SYNTHETIC_CACHE kept {org} ({since:?})"
);
let token = synthetics_probe_tokens::find_global(&rows.probe_token).await;
assert!(
token.unwrap().is_none(),
"TOKEN_CACHE kept {org} ({since:?})"
);
let default_token = synthetics_probe_tokens::find_default(org).await;
assert!(
default_token.unwrap().is_none(),
"DEFAULT_TOKEN_CACHE kept {org} ({since:?})"
);
let variables = synthetics_variables::list_cached(conn, org).await;
assert!(
variables.unwrap().is_empty(),
"VARIABLE_CACHE kept {org} ({since:?})"
);
let location = synthetics_locations::get(&rows.location_id).await;
assert!(
location.unwrap().is_none(),
"LOCATIONS_CACHE kept {org} ({since:?})"
);
let agent = synthetics_agents::get_cached(&rows.agent_id).await;
assert!(
agent.unwrap().is_none(),
"AGENT_CACHE kept {org} ({since:?})"
);
#[cfg(feature = "enterprise")]
{
let workflow = db::workflows::get_workflow(org, &rows.workflow_id).await;
assert!(workflow.unwrap().is_none(), "workflows CACHE kept {org}");
}
assert_llm_and_oncall_caches_evicted(org, rows, since).await;
}
async fn assert_llm_and_oncall_caches_evicted(
org: &str,
rows: &CoverageRows,
since: std::time::Duration,
) {
use infra::table::{oncall_ownership, oncall_schedules, oncall_teams};
let pricing = db::model_pricing::get_org_pricing_entries(org);
assert!(
!pricing
.iter()
.any(|e| e.definition.name == rows.pricing_model),
"model pricing CACHE kept {org} ({since:?})"
);
let team = oncall_teams::get_cached(org, &rows.team_id).await;
assert!(team.unwrap().is_none(), "TEAM_CACHE kept {org} ({since:?})");
assert!(
!oncall_teams::has_any_cached(org).await.unwrap(),
"ORG_HAS_TEAMS_CACHE kept {org} ({since:?})"
);
let members = oncall_teams::list_members_cached(&rows.team_id).await;
assert!(
members.unwrap().is_empty(),
"MEMBERS_CACHE kept {org} ({since:?})"
);
let schedule = oncall_schedules::get_by_team_cached(org, &rows.team_id).await;
assert!(
schedule.unwrap().is_none(),
"SCHEDULE_CACHE kept {org} ({since:?})"
);
let rules = oncall_ownership::list_cached(org).await;
assert!(
rules.unwrap().is_empty(),
"RULES_CACHE kept {org} ({since:?})"
);
#[cfg(feature = "enterprise")]
assert!(
!o2_ratelimit::dataresource::db::RATELIMIT_RULES_CACHE
.read()
.await
.iter()
.any(|r| r.org == org),
"RATELIMIT_RULES_CACHE kept {org}"
);
}
/// A miss here mints a new policy and DEK, so a different id or key proves the eviction.
async fn assert_reminting_caches_evicted(org: &str, rows: &CoverageRows) {
use infra::table::{cipher, oncall_policies, oncall_teams};
let policy = oncall_policies::get_or_create_cached(org, &rows.team_id).await;
let dek = cipher::get_dek(org).await.unwrap();
oncall_teams::delete_by_org(org).await.unwrap();
cipher::delete_by_org(org).await.unwrap();
assert_ne!(
policy.unwrap().id,
rows.policy_id,
"POLICY_CACHE kept {org}"
);
assert_ne!(dek, rows.dek, "DEK_CACHE kept {org}");
}
async fn drop_coverage_rows(org: &str, rows: &CoverageRows) {
use infra::table::{
anomaly_detection, external_alerts, folders, incident_integrations, status_pages,
synthetics_agents, synthetics_checks, synthetics_environments, synthetics_locations,
synthetics_probe_tokens, synthetics_variables, workflows,
};
let conn = infra::db::get_orm_client_rw().await;
status_pages::delete_by_org(org).await.unwrap();
incident_integrations::delete_by_org(org).await.unwrap();
external_alerts::delete_by_org(org).await.unwrap();
synthetics_checks::delete_by_org(org).await.unwrap();
synthetics_probe_tokens::delete_by_org(org).await.unwrap();
synthetics_variables::delete_by_org(conn, org)
.await
.unwrap();
synthetics_environments::delete_by_org(conn, org)
.await
.unwrap();
synthetics_agents::delete_by_org(org).await.unwrap();
synthetics_locations::delete_by_org(org).await.unwrap();
infra::storage::del_org_prefix_strict(org, &format!("synthetics/{org}/"))
.await
.unwrap();
workflows::delete_children_by_org(org).await.unwrap();
workflows::delete_drafts_by_org(org).await.unwrap();
workflows::delete_by_org(org).await.unwrap();
anomaly_detection::models::delete_by_org(conn, org)
.await
.unwrap();
anomaly_detection::config::delete_by_org(conn, org)
.await
.unwrap();
#[cfg(feature = "enterprise")]
o2_enterprise::enterprise::anomaly_detection::storage::delete_org_models(org)
.await
.unwrap();
folders::delete_by_org(org).await.unwrap();
drop_llm_and_oncall_rows(org).await;
assert_coverage(org, rows, false).await;
}
async fn drop_llm_and_oncall_rows(org: &str) {
use infra::table::{
ai_chat_sessions, ai_chat_shares, ai_chat_turns, cipher, gen_ai_agents, llm_eval,
llm_secrets, oncall_teams, online_eval_jobs, org_ai_toolsets, providers, ratelimit,
score_configs, scorers,
};
#[cfg(feature = "enterprise")]
o2_enterprise::enterprise::llm_evaluations::remote_tasks::delete_all_by_org(org)
.await
.unwrap();
llm_eval::delete_by_org(org).await.unwrap();
score_configs::delete_all_by_org(org).await.unwrap();
scorers::delete_all_by_org(org).await.unwrap();
online_eval_jobs::delete_all_by_org(org).await.unwrap();
providers::delete_all_by_org(org).await.unwrap();
llm_secrets::delete_by_org(org).await.unwrap();
org_ai_toolsets::delete_by_org(org).await.unwrap();
gen_ai_agents::delete_by_org(org).await.unwrap();
db::model_pricing::delete_by_org(org).await.unwrap();
ratelimit::delete_by_org(org).await.unwrap();
oncall_teams::delete_by_org(org).await.unwrap();
cipher::delete_by_org(org).await.unwrap();
ai_chat_turns::delete_by_org(org).await.unwrap();
ai_chat_shares::delete_by_org(org).await.unwrap();
ai_chat_sessions::delete_by_org(org).await.unwrap();
}
/// Rows only the deleted org gets: a dashboard made through the API, and a report on it.
async fn seed_raw_routed_rows(app: &Router, headers: &HeaderMap, org: &str) -> String {
use config::meta::{
dashboards::reports::Report,
folder::{Folder, FolderType},
};
let board = r#"{"version":5,"title":"orgdelBoard","dashboardId":"","description":"","role":"","owner":"root@example.com","created":"2023-03-30T07:49:41.744+00:00","tabs":[{"tabId":"tab1","name":"Main","panels":[]}]}"#;
let (status, body) = make_request(
app,
Method::POST,
&format!("/api/{org}/dashboards"),
Some(headers.clone()),
Some(board.to_string()),
)
.await;
assert!(status.is_success(), "{}", String::from_utf8_lossy(&body));
let dashboard: Dashboard = json::from_slice(&body).unwrap();
let dashboard_id = dashboard.dashboard_id().unwrap().to_string();
common::infra::config::DASHBOARD_ID_TO_ORG
.write()
.await
.insert(dashboard_id.clone(), org.to_string());
let folder = Folder {
folder_id: String::new(),
name: "orgdelReports".to_string(),
description: String::new(),
icon: None,
};
let folder = db::folders::save_folder(org, folder, FolderType::Reports, false)
.await
.unwrap();
let report: Report = serde_json::from_value(json!({
"name": "orgdelReport",
"orgId": org,
"dashboards": [{"dashboard": dashboard_id, "folder": "default", "tabs": ["tab1"]}],
"destinations": []
}))
.unwrap();
let conn = infra::db::get_orm_client_rw().await;
infra::table::reports::create_report(conn, &folder.folder_id, report, None)
.await
.unwrap();
dashboard_id
}
async fn seed_shared_templates(org: &str, kept: &str) -> [String; 4] {
use config::meta::destinations::{Destination, Module};
use infra::table::entity::{destinations, templates};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
let conn = infra::db::get_orm_client_rw().await;
let twin = format!("{org}Twin");
let fallback = format!("{org}Fallback");
let mut ids = Vec::new();
for (owner, name) in [
(config::DEFAULT_ORG, twin.as_str()),
(org, twin.as_str()),
(kept, twin.as_str()),
(config::DEFAULT_ORG, fallback.as_str()),
] {
let id = config::ider::uuid();
templates::ActiveModel {
id: Set(id.clone()),
org: Set(owner.to_string()),
name: Set(name.to_string()),
is_default: Set(false),
r#type: Set("http".to_string()),
body: Set(r#"{"text":"{alert_name}"}"#.to_string()),
title: Set(None),
kind: Set("custom".to_string()),
}
.insert(conn)
.await
.unwrap();
ids.push(id);
}
let dest = infra::table::destinations::get(org, "orgdelDest")
.await
.unwrap()
.unwrap();
let Module::Alert {
destination_type, ..
} = dest.module
else {
panic!("orgdelDest is an alert destination");
};
for (name, template, expected_id) in [
("orgdelOwnedDest", twin, &ids[2]),
("orgdelDefaultDest", fallback, &ids[3]),
] {
infra::table::destinations::put(Destination {
org_id: kept.to_string(),
name: name.to_string(),
module: Module::Alert {
template: Some(template),
destination_type: destination_type.clone(),
},
..Default::default()
})
.await
.unwrap();
let stored = destinations::Entity::find()
.filter(destinations::Column::Org.eq(kept))
.filter(destinations::Column::Name.eq(name))
.one(conn)
.await
.unwrap()
.unwrap();
assert_eq!(stored.template_id.as_ref(), Some(expected_id));
}
ids.try_into().unwrap()
}
async fn assert_shared_templates(org: &str, kept: &str, ids: &[String; 4]) {
let ids_of = |rows: Vec<(String, String)>| -> Vec {
rows.into_iter().map(|(id, _)| id).collect()
};
let defaults = ids_of(
infra::table::templates::list_ids_by_org(config::DEFAULT_ORG)
.await
.unwrap(),
);
assert!(defaults.contains(&ids[0]), "the default twin was deleted");
assert!(
defaults.contains(&ids[3]),
"the default fallback was deleted"
);
assert!(
infra::table::templates::list_ids_by_org(org)
.await
.unwrap()
.is_empty(),
"the deleted organization's templates remain"
);
let survivors = infra::table::templates::list_ids_by_org(kept)
.await
.unwrap();
assert!(
ids_of(survivors).contains(&ids[2]),
"the live twin was deleted"
);
for destination in ["orgdelOwnedDest", "orgdelDefaultDest"] {
assert!(
infra::table::destinations::get(kept, destination)
.await
.unwrap()
.is_some()
);
infra::table::destinations::delete(kept, destination)
.await
.unwrap();
}
assert!(
!infra::table::templates::delete_by_id(org, &ids[0])
.await
.unwrap()
);
for (owner, id) in [
(config::DEFAULT_ORG, &ids[0]),
(kept, &ids[2]),
(config::DEFAULT_ORG, &ids[3]),
] {
assert!(
infra::table::templates::delete_by_id(owner, id)
.await
.unwrap()
);
}
}
/// Asserts the raw-routed rows and the caches their db-layer deletes evict.
async fn assert_raw_routed_rows(org: &str, dashboard_id: &str, present: bool) {
use common::infra::config::{ALERTS_TEMPLATES, DASHBOARD_ID_TO_ORG, DESTINATIONS};
let dashboard = infra::table::dashboards::get_by_id(org, dashboard_id).await;
assert_eq!(dashboard.unwrap().is_some(), present, "dashboard of {org}");
let mapped = DASHBOARD_ID_TO_ORG.read().await.contains_key(dashboard_id);
assert_eq!(mapped, present, "DASHBOARD_ID_TO_ORG for {org}");
for (table, column) in [
("reports", "org"),
("folders", "org"),
("destinations", "org"),
("templates", "org"),
] {
let n = count_rows(table, column, org).await;
assert_eq!(n > 0, present, "{table} holds {n} rows of {org}");
}
let destination = &format!("{org}/orgdelDest");
let template = &format!("{org}/orgdelTemplate");
eventually(
&format!("DESTINATIONS and ALERTS_TEMPLATES for {org}"),
|| async move {
DESTINATIONS.contains_key(destination) == present
&& ALERTS_TEMPLATES.contains_key(template) == present
},
)
.await;
let org_key = format!("/organization/org/{org}");
let coordinator = infra::db::get_coordinator().await;
assert_eq!(
coordinator.get(&org_key).await.is_ok(),
present,
"{org_key}"
);
}
async fn eventually(what: &str, mut check: F)
where
F: FnMut() -> Fut,
Fut: std::future::Future