mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
feat(license): licence-gate the search routes that skipped the search check
This commit is contained in:
Ashish Kolhe
Generated
+1
@@ -8139,6 +8139,7 @@ dependencies = [
|
||||
"reqwest 0.13.4",
|
||||
"sea-orm",
|
||||
"search",
|
||||
"search_service",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"snap",
|
||||
|
||||
@@ -21,6 +21,7 @@ enterprise = [
|
||||
"openobserve-api-search/enterprise",
|
||||
"openobserve-api-common/enterprise",
|
||||
"openobserve-core/enterprise",
|
||||
"search_service/enterprise",
|
||||
"stream/enterprise",
|
||||
]
|
||||
cloud = [
|
||||
@@ -54,6 +55,7 @@ openobserve-tls.workspace = true
|
||||
ingestion-common.workspace = true
|
||||
stream.workspace = true
|
||||
search.workspace = true
|
||||
search_service.workspace = true
|
||||
openobserve-api-ingest.workspace = true
|
||||
openobserve-api-management.workspace = true
|
||||
openobserve-api-pipelines.workspace = true
|
||||
@@ -98,6 +100,7 @@ futures-util.workspace = true
|
||||
base64.workspace = true
|
||||
sea-orm.workspace = true
|
||||
svix-ksuid.workspace = true
|
||||
o2_enterprise = { workspace = true, default-features = false, features = ["test-utils"] }
|
||||
tracing.workspace = true
|
||||
tracing-subscriber.workspace = true
|
||||
zstd.workspace = true
|
||||
|
||||
@@ -17,8 +17,10 @@ mod ingest_rejections;
|
||||
mod org_blocking;
|
||||
mod password_policy;
|
||||
mod root_only;
|
||||
mod search_license;
|
||||
|
||||
pub use ingest_rejections::ingest_rejections_middleware;
|
||||
pub use org_blocking::{blocked_org_param, blocked_orgs_middleware};
|
||||
pub use password_policy::password_policy_middleware;
|
||||
pub use root_only::root_only_middleware;
|
||||
pub use search_license::search_license_middleware;
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use axum::{
|
||||
extract::{FromRequestParts, RawPathParams, Request},
|
||||
http::request::Parts,
|
||||
middleware::Next,
|
||||
response::Response,
|
||||
};
|
||||
use common::meta::http::HttpResponse as MetaHttpResponse;
|
||||
use config::{META_ORG_ID, utils::base64};
|
||||
|
||||
/// Answers the search handlers' licence 429 before a gated route reads any data.
|
||||
pub async fn search_license_middleware(request: Request, next: Next) -> Response {
|
||||
let (mut parts, body) = request.into_parts();
|
||||
if let Some(refused) = refusal(&mut parts).await {
|
||||
return refused;
|
||||
}
|
||||
next.run(Request::from_parts(parts, body)).await
|
||||
}
|
||||
|
||||
async fn refusal(parts: &mut Parts) -> Option<Response> {
|
||||
// Unreadable params fall through to the closed-gate answer for an unknown org and stream.
|
||||
let (org_id, streams) = match RawPathParams::from_request_parts(parts, &()).await {
|
||||
Ok(params) => (
|
||||
param(¶ms, "org_id").unwrap_or_default().to_string(),
|
||||
path_streams(¶ms),
|
||||
),
|
||||
Err(_) => (String::new(), Vec::new()),
|
||||
};
|
||||
refusal_for(&org_id, &streams, search_service::check_search_allowed)
|
||||
}
|
||||
|
||||
fn param<'a>(params: &'a RawPathParams, name: &str) -> Option<&'a str> {
|
||||
params
|
||||
.iter()
|
||||
.find_map(|(key, value)| (key == name).then_some(value))
|
||||
}
|
||||
|
||||
fn path_streams(params: &RawPathParams) -> Vec<String> {
|
||||
if let Some(stream) = param(params, "stream_name") {
|
||||
return vec![stream.to_string()];
|
||||
}
|
||||
param(params, "stream_names")
|
||||
.and_then(|encoded| base64::decode_url(encoded).ok())
|
||||
.map(|names| names.split(',').map(str::to_string).collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn refusal_for(
|
||||
org_id: &str,
|
||||
streams: &[String],
|
||||
gate: impl Fn(&str, Option<&str>) -> Result<(), infra::errors::Error>,
|
||||
) -> Option<Response> {
|
||||
if org_id == META_ORG_ID {
|
||||
return None;
|
||||
}
|
||||
let verdict = if streams.is_empty() {
|
||||
gate(org_id, None)
|
||||
} else {
|
||||
streams
|
||||
.iter()
|
||||
.try_for_each(|stream| gate(org_id, Some(stream)))
|
||||
};
|
||||
verdict.err().map(MetaHttpResponse::too_many_requests)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use axum::{Router, body::Body, http::StatusCode, routing::get};
|
||||
use infra::errors::Error;
|
||||
use tower::ServiceExt;
|
||||
|
||||
use super::*;
|
||||
|
||||
const BLOCKED: &str = "Search is temporarily disabled due to exceeding allotted ingestion limit. Please contact your administrator.";
|
||||
|
||||
fn open(_: &str, _: Option<&str>) -> Result<(), Error> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn closed(_: &str, _: Option<&str>) -> Result<(), Error> {
|
||||
Err(Error::Message(BLOCKED.to_string()))
|
||||
}
|
||||
|
||||
fn names(streams: &[&str]) -> Vec<String> {
|
||||
streams.iter().map(|s| s.to_string()).collect()
|
||||
}
|
||||
|
||||
async fn body_bytes(response: Response) -> Vec<u8> {
|
||||
axum::body::to_bytes(response.into_body(), usize::MAX)
|
||||
.await
|
||||
.unwrap()
|
||||
.to_vec()
|
||||
}
|
||||
|
||||
async fn streams_at(encoded: &str) -> String {
|
||||
let app = Router::new().route(
|
||||
"/{org_id}/{stream_names}/_around_multi",
|
||||
get(|params: RawPathParams| async move { path_streams(¶ms).join("|") }),
|
||||
);
|
||||
let request = axum::http::Request::get(format!("/acme/{encoded}/_around_multi"))
|
||||
.body(Body::empty())
|
||||
.unwrap();
|
||||
String::from_utf8(body_bytes(app.oneshot(request).await.unwrap()).await).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn open_gate_refuses_nothing() {
|
||||
assert!(refusal_for("acme", &[], open).is_none());
|
||||
assert!(refusal_for("acme", &names(&["logs"]), open).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn closed_gate_answers_the_search_handler_429() {
|
||||
let refused = refusal_for("acme", &[], closed).unwrap();
|
||||
assert_eq!(refused.status(), StatusCode::TOO_MANY_REQUESTS);
|
||||
assert!(refused.headers().get("retry-after").is_none());
|
||||
let body = body_bytes(refused).await;
|
||||
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
|
||||
assert_eq!(
|
||||
json,
|
||||
serde_json::json!({"code": 429, "message": format!("Error# {BLOCKED}")})
|
||||
);
|
||||
let handler_body = body_bytes(MetaHttpResponse::too_many_requests(Error::Message(
|
||||
BLOCKED.to_string(),
|
||||
)))
|
||||
.await;
|
||||
assert_eq!(body, handler_body);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn meta_org_is_never_refused_and_never_asks_the_gate() {
|
||||
let calls = AtomicUsize::new(0);
|
||||
let counting = |org: &str, stream: Option<&str>| {
|
||||
calls.fetch_add(1, Ordering::SeqCst);
|
||||
closed(org, stream)
|
||||
};
|
||||
assert!(refusal_for(META_ORG_ID, &names(&["logs"]), counting).is_none());
|
||||
assert!(refusal_for(META_ORG_ID, &[], counting).is_none());
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn each_path_stream_reaches_the_gate() {
|
||||
let only_triggers = |_: &str, stream: Option<&str>| match stream {
|
||||
Some("triggers") => Ok(()),
|
||||
_ => Err(Error::Message(BLOCKED.to_string())),
|
||||
};
|
||||
assert!(refusal_for("acme", &names(&["triggers"]), only_triggers).is_none());
|
||||
assert!(refusal_for("acme", &names(&["logs"]), only_triggers).is_some());
|
||||
assert!(refusal_for("acme", &names(&["triggers", "logs"]), only_triggers).is_some());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn around_multi_stream_names_are_decoded() {
|
||||
assert_eq!(
|
||||
streams_at(&base64::encode_url("triggers,logs")).await,
|
||||
"triggers|logs"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn undecodable_stream_names_are_gated_as_unknown() {
|
||||
assert_eq!(streams_at("!!not-base64!!").await, "");
|
||||
let asked = std::sync::Mutex::new(Vec::new());
|
||||
let recording = |_: &str, stream: Option<&str>| {
|
||||
asked.lock().unwrap().push(stream.map(str::to_string));
|
||||
closed("", None)
|
||||
};
|
||||
assert!(refusal_for("acme", &[], recording).is_some());
|
||||
assert_eq!(*asked.lock().unwrap(), vec![None]);
|
||||
}
|
||||
}
|
||||
@@ -85,7 +85,7 @@ use crate::{
|
||||
},
|
||||
router::middlewares::{
|
||||
blocked_org_param, blocked_orgs_middleware, ingest_rejections_middleware,
|
||||
password_policy_middleware, root_only_middleware,
|
||||
password_policy_middleware, root_only_middleware, search_license_middleware,
|
||||
},
|
||||
},
|
||||
};
|
||||
@@ -99,6 +99,8 @@ mod one_ack_tests;
|
||||
pub mod openapi;
|
||||
#[cfg(test)]
|
||||
mod openapi_coverage;
|
||||
#[cfg(test)]
|
||||
mod search_gate_tests;
|
||||
|
||||
use common::meta::http::ERROR_HEADER;
|
||||
|
||||
@@ -1292,9 +1294,13 @@ pub fn service_routes() -> Router {
|
||||
#[cfg(not(feature = "enterprise"))]
|
||||
let server = cfg.common.instance_name_short.to_string();
|
||||
|
||||
// Downtimes, Slack and service streams live in their own tables to keep this function under the line limit.
|
||||
let mut router =
|
||||
public_dashboard_admin_routes(Router::new().merge(downtime_routes()).merge(slack_routes()));
|
||||
// Downtimes, Slack, search and service streams live in their own tables to keep this function under the line limit.
|
||||
let mut router = public_dashboard_admin_routes(
|
||||
Router::new()
|
||||
.merge(downtime_routes())
|
||||
.merge(slack_routes())
|
||||
.merge(search_routes()),
|
||||
);
|
||||
router = router.merge(service_streams_routes());
|
||||
// Full UI configuration — authenticated counterpart of the unauthenticated
|
||||
// `/config` bootstrap in config_routes()
|
||||
@@ -1402,7 +1408,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/{org_id}/{stream_name}/traces/session/details", get(traces::session::get_session_details))
|
||||
.route("/{org_id}/{stream_name}/traces/user", get(traces::user::get_latest_users))
|
||||
.route("/{org_id}/{stream_name}/traces/time_range", get(traces::time_index::get_trace_time_range))
|
||||
.route("/{org_id}/{stream_name}/traces/{trace_id}/details", get(traces::details::get_trace_details))
|
||||
.route("/{org_id}/{stream_name}/traces/{trace_id}/dag", get(traces::dag::get_trace_dag))
|
||||
|
||||
// Profiles query (ingest remains under /v1/profiles)
|
||||
@@ -1412,43 +1417,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/{org_id}/{stream_name}/profiles/merge", get(profiles_query::merge_profiles_get).post(profiles_query::merge_profiles))
|
||||
.route("/{org_id}/{stream_name}/profiles/diff", post(profiles_query::diff_profiles))
|
||||
|
||||
// Database Monitoring — its own top-level module, not under /traces.
|
||||
// The path segment must stay `db_monitoring` to match the OFGA resource
|
||||
// these routes authorize against (`db_monitoring:{org}`, see o2_openfga
|
||||
// ROUTE_PERMISSIONS).
|
||||
//
|
||||
// Every route below is registered in both builds; the build-type gate
|
||||
// lives in the handlers, three of which (deadlocks/blocking/
|
||||
// table_health) are dual-implemented and answer 403 on OSS. Registering
|
||||
// them in the enterprise-gated block instead would 404 them, losing the
|
||||
// distinction between "not licensed" and "no such endpoint". Runtime
|
||||
// off-switch is ZO_DB_MONITORING_ENABLED.
|
||||
.route("/{org_id}/db_monitoring/databases", get(db_monitoring::handler::get_dbm_databases))
|
||||
.route("/{org_id}/db_monitoring/queries", get(db_monitoring::handler::get_dbm_queries))
|
||||
.route("/{org_id}/db_monitoring/query/history", get(db_monitoring::handler::get_dbm_query_history))
|
||||
.route("/{org_id}/db_monitoring/query/endpoints", get(db_monitoring::handler::get_dbm_query_endpoints))
|
||||
.route("/{org_id}/db_monitoring/samples", get(db_monitoring::handler::get_dbm_samples))
|
||||
// Server-vantage events (read the canonical o2_dbm_* columns)
|
||||
.route("/{org_id}/db_monitoring/deadlocks", get(db_monitoring::handler::get_dbm_deadlocks))
|
||||
.route("/{org_id}/db_monitoring/blocking", get(db_monitoring::handler::get_dbm_blocking))
|
||||
.route("/{org_id}/db_monitoring/activity", get(db_monitoring::handler::get_dbm_activity))
|
||||
// `query/insights` returns the query-detail page's Logs-side pair in one
|
||||
// round trip; `query/plans` and `query/server_metrics` are superseded by
|
||||
// it (same sections, same envelopes) and stay registered for
|
||||
// compatibility.
|
||||
.route("/{org_id}/db_monitoring/query/insights", get(db_monitoring::handler::get_dbm_query_insights))
|
||||
.route("/{org_id}/db_monitoring/query/plans", get(db_monitoring::handler::get_dbm_query_plans))
|
||||
.route("/{org_id}/db_monitoring/query/server_metrics", get(db_monitoring::handler::get_dbm_query_server_metrics))
|
||||
.route("/{org_id}/db_monitoring/server_queries", get(db_monitoring::handler::get_dbm_server_queries))
|
||||
.route("/{org_id}/db_monitoring/server_samples", get(db_monitoring::handler::get_dbm_server_samples))
|
||||
.route("/{org_id}/db_monitoring/table_health", get(db_monitoring::handler::get_dbm_table_health))
|
||||
.route("/{org_id}/db_monitoring/instances", get(db_monitoring::handler::get_dbm_instances))
|
||||
.route(
|
||||
"/{org_id}/db_monitoring/instance_metrics",
|
||||
get(db_monitoring::handler::get_dbm_instance_metrics),
|
||||
)
|
||||
.route("/{org_id}/db_monitoring/badges", get(db_monitoring::handler::get_dbm_badges))
|
||||
|
||||
// LLM Model Pricing
|
||||
.route("/{org_id}/llm/models", get(model_pricing::list).post(model_pricing::create))
|
||||
// NOTE: named routes MUST be registered before {model_id} or they match as a model ID
|
||||
@@ -1471,7 +1439,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/{org_id}/prometheus/api/v1/metadata", get(promql::metadata))
|
||||
.route("/{org_id}/prometheus/api/v1/series", get(promql::series_get).post(promql::series_post))
|
||||
.route("/{org_id}/prometheus/api/v1/labels", get(promql::labels_get).post(promql::labels_post))
|
||||
.route("/{org_id}/prometheus/api/v1/label/{label_name}/values", get(promql::label_values))
|
||||
.route("/{org_id}/prometheus/api/v1/format_query", get(promql::format_query_get).post(promql::format_query_post))
|
||||
.route("/{org_id}/prometheus/api/v1/parse_tree", post(promql::parse_tree))
|
||||
|
||||
@@ -1484,12 +1451,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/{org_id}/_search_history", post(search::search_history))
|
||||
.route("/{org_id}/result_schema", post(search::result_schema))
|
||||
|
||||
// Multi-stream search
|
||||
.route("/{org_id}/_search_multi", post(search::multi_streams::search_multi))
|
||||
.route("/{org_id}/_search_multi_stream", post(search::multi_streams::search_multi_stream))
|
||||
.route("/{org_id}/_search_partition_multi", post(search::multi_streams::_search_partition_multi))
|
||||
.route("/{org_id}/{stream_names}/_around_multi", get(search::multi_streams::around_multi))
|
||||
|
||||
// HTTP/2 streaming
|
||||
.route("/{org_id}/_search_stream", post(search::search_stream::search_http2_stream))
|
||||
.route("/{org_id}/_values_stream", post(search::search_stream::values_http2_stream))
|
||||
@@ -1643,8 +1604,7 @@ pub fn service_routes() -> Router {
|
||||
// Anomaly Detection
|
||||
.route("/{org_id}/anomaly_detection", get(anomaly_detection::list_configs).post(anomaly_detection::create_config))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}", get(anomaly_detection::get_config).put(anomaly_detection::update_config).delete(anomaly_detection::delete_config))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/train", post(anomaly_detection::train_model).delete(anomaly_detection::cancel_training))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/detect", post(anomaly_detection::detect_anomalies))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/train", delete(anomaly_detection::cancel_training))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/history", get(anomaly_detection::get_detection_history))
|
||||
.route("/{org_id}/anomaly_detection/history", get(alerts::history::get_all_anomaly_history));
|
||||
router = alert_calibration_routes(router);
|
||||
@@ -1755,21 +1715,11 @@ pub fn service_routes() -> Router {
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/annotation_queues/{queue_id}/items",
|
||||
post(annotation_queues::enqueue_annotation_queue_item)
|
||||
.delete(annotation_queues::clear_annotation_queue_items),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}",
|
||||
get(annotation_queues::get_annotation_queue_item),
|
||||
delete(annotation_queues::clear_annotation_queue_items),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/reviews",
|
||||
get(annotation_queues::list_annotation_queue_item_reviews)
|
||||
.post(annotation_queues::review_annotation_queue_item),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/push_to_dataset",
|
||||
post(datasets::push_annotation_queue_item_to_dataset),
|
||||
post(annotation_queues::review_annotation_queue_item),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/annotation_queues/{queue_id}/items/archive",
|
||||
@@ -1791,9 +1741,7 @@ pub fn service_routes() -> Router {
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/datasets/{dataset_id}/items",
|
||||
get(datasets::list_dataset_items)
|
||||
.post(datasets::push_dataset_item)
|
||||
.put(datasets::upsert_dataset_items),
|
||||
get(datasets::list_dataset_items).put(datasets::upsert_dataset_items),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/datasets/{dataset_id}/rows",
|
||||
@@ -1819,27 +1767,15 @@ pub fn service_routes() -> Router {
|
||||
"/{org_id}/experiments",
|
||||
get(experiments::list_experiments).post(experiments::create_experiment),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/compare",
|
||||
get(experiments::compare_experiments),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/{experiment_id}",
|
||||
get(experiments::get_experiment).delete(experiments::delete_experiment),
|
||||
delete(experiments::delete_experiment),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/{experiment_id}/baseline",
|
||||
put(experiments::set_experiment_baseline)
|
||||
.delete(experiments::clear_experiment_baseline),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/{experiment_id}/rows",
|
||||
get(experiments::list_experiment_result_rows),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/{experiment_id}/rows/{row_id}",
|
||||
get(experiments::get_experiment_row),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/experiments/{experiment_id}/rows/{row_id}/trials/{trial_index}/retry",
|
||||
post(experiments::retry_experiment_slot),
|
||||
@@ -1872,9 +1808,6 @@ pub fn service_routes() -> Router {
|
||||
// On-demand human annotation from Discovery
|
||||
.route("/{org_id}/annotations", post(annotations::annotate_target))
|
||||
|
||||
// LLM score-based Discovery
|
||||
.route("/{org_id}/discovery", get(discovery::list_discovery_items))
|
||||
|
||||
// LLM Providers (Online Eval Phase 2)
|
||||
.route("/{org_id}/providers", get(providers::list_providers).post(providers::create_provider))
|
||||
.route("/{org_id}/providers/test", post(providers::test_provider_config))
|
||||
@@ -1892,7 +1825,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/{org_id}/tasks/{entity_id}/signing/end_grace", post(remote_tasks::end_remote_task_signing_grace))
|
||||
.route("/{org_id}/tasks/{entity_id}/signing", get(remote_tasks::get_remote_task_signing_status).delete(remote_tasks::revoke_remote_task_signing_secret))
|
||||
.route("/{org_id}/tasks/{entity_id}/versions", get(remote_tasks::list_remote_task_versions))
|
||||
.route("/{org_id}/tasks/{entity_id}/stats", get(remote_tasks::get_remote_task_stats))
|
||||
.route("/{org_id}/tasks/{entity_id}/draft", get(remote_tasks::get_remote_task_draft).delete(remote_tasks::discard_remote_task_draft))
|
||||
.route("/{org_id}/tasks/{entity_id}/test_connection", post(remote_tasks::publish_remote_task))
|
||||
.route("/{org_id}/tasks/{entity_id}/test_run", post(remote_tasks::test_run_remote_task))
|
||||
@@ -1934,11 +1866,9 @@ pub fn service_routes() -> Router {
|
||||
{
|
||||
router = fleet_routes(router)
|
||||
// Search jobs
|
||||
.route("/{org_id}/search_jobs", get(search::search_job::list_status).post(search::search_job::submit_job))
|
||||
.route("/{org_id}/search_jobs", get(search::search_job::list_status))
|
||||
.route("/{org_id}/search_jobs/{job_id}", get(search::search_job::get_status).delete(search::search_job::delete_job))
|
||||
.route("/{org_id}/search_jobs/{job_id}/result", get(search::search_job::get_job_result))
|
||||
.route("/{org_id}/search_jobs/{job_id}/cancel", post(search::search_job::cancel_job))
|
||||
.route("/{org_id}/search_jobs/{job_id}/retry", post(search::search_job::retry_job))
|
||||
|
||||
// Query manager
|
||||
.route("/{org_id}/query_manager/status", get(search::query_manager::query_status))
|
||||
@@ -1976,16 +1906,6 @@ pub fn service_routes() -> Router {
|
||||
.route("/license", get(license::get_license_info).post(license::store_license))
|
||||
.route("/license/refresh", post(license::refresh_license_limits))
|
||||
|
||||
// Topology
|
||||
.route("/{org_id}/traces/service_graph/topology/current", get(traces::get_current_topology))
|
||||
.route("/{org_id}/traces/service_graph/edge/history", get(traces::get_edge_history))
|
||||
// Agent behavior signals (loop / failure / cost) — reads the _agent_signals stream
|
||||
.route("/{org_id}/traces/agent_signals", get(traces::get_agent_signals))
|
||||
.route("/{org_id}/traces/agent_signals/compare", post(traces::compare_agent_versions))
|
||||
|
||||
// Patterns
|
||||
.route("/{org_id}/streams/{stream_name}/patterns/extract", post(patterns::extract_patterns))
|
||||
|
||||
// Service streams
|
||||
.route("/{org_id}/service_streams", get(service_streams::list_services))
|
||||
.route("/{org_id}/service_streams/config/identity", get(service_streams::get_identity_config).put(service_streams::save_identity_config))
|
||||
@@ -2378,8 +2298,6 @@ pub fn downtime_routes() -> Router {
|
||||
Router::new()
|
||||
.route("/v2/{org_id}/downtimes", get(downtimes::list_downtimes).post(downtimes::create_downtime))
|
||||
.route("/v2/{org_id}/downtimes/preview", post(downtimes::preview_downtime))
|
||||
.route("/v2/{org_id}/downtimes/resources", post(downtimes::downtime_resources))
|
||||
.route("/v2/{org_id}/downtimes/values", post(downtimes::downtime_values))
|
||||
.route("/v2/{org_id}/downtimes/move", patch(downtimes::move_downtimes))
|
||||
.route("/v2/{org_id}/downtimes/{downtime_id}", get(downtimes::get_downtime).put(downtimes::update_downtime).delete(downtimes::delete_downtime))
|
||||
.route("/v2/{org_id}/downtimes/{downtime_id}/cancel", post(downtimes::cancel_downtime))
|
||||
@@ -2405,6 +2323,101 @@ fn report_routes() -> Router {
|
||||
.route("/v2/{org_id}/reports/{report_id}/trigger", put(dashboards::reports::trigger_report_v2))
|
||||
}
|
||||
|
||||
/// Routes that read customer data through search; the licence gate answers 429 while search is blocked.
|
||||
#[rustfmt::skip]
|
||||
fn search_routes() -> Router {
|
||||
#[cfg(feature = "enterprise")]
|
||||
let llm_eval = get_o2_config().llm_eval_config.enabled;
|
||||
#[cfg(not(feature = "enterprise"))]
|
||||
let llm_eval = false;
|
||||
search_routes_with(llm_eval)
|
||||
}
|
||||
|
||||
#[rustfmt::skip]
|
||||
#[cfg_attr(not(feature = "enterprise"), allow(unused_variables))]
|
||||
fn search_routes_with(llm_eval: bool) -> Router {
|
||||
let router = Router::new()
|
||||
// Multi-stream search
|
||||
.route("/{org_id}/_search_multi", post(search::multi_streams::search_multi))
|
||||
.route("/{org_id}/_search_multi_stream", post(search::multi_streams::search_multi_stream))
|
||||
.route("/{org_id}/_search_partition_multi", post(search::multi_streams::_search_partition_multi))
|
||||
.route("/{org_id}/{stream_names}/_around_multi", get(search::multi_streams::around_multi))
|
||||
.route("/{org_id}/{stream_name}/traces/{trace_id}/details", get(traces::details::get_trace_details))
|
||||
.route("/{org_id}/prometheus/api/v1/label/{label_name}/values", get(promql::label_values))
|
||||
.route("/v2/{org_id}/downtimes/resources", post(downtimes::downtime_resources))
|
||||
.route("/v2/{org_id}/downtimes/values", post(downtimes::downtime_values))
|
||||
|
||||
// Database Monitoring — its own top-level module, not under /traces.
|
||||
// The path segment must stay `db_monitoring` to match the OFGA resource
|
||||
// these routes authorize against (`db_monitoring:{org}`, see o2_openfga
|
||||
// ROUTE_PERMISSIONS).
|
||||
//
|
||||
// Every route below is registered in both builds; the build-type gate
|
||||
// lives in the handlers, three of which (deadlocks/blocking/
|
||||
// table_health) are dual-implemented and answer 403 on OSS. Registering
|
||||
// them in the enterprise-gated block instead would 404 them, losing the
|
||||
// distinction between "not licensed" and "no such endpoint". Runtime
|
||||
// off-switch is ZO_DB_MONITORING_ENABLED.
|
||||
.route("/{org_id}/db_monitoring/databases", get(db_monitoring::handler::get_dbm_databases))
|
||||
.route("/{org_id}/db_monitoring/queries", get(db_monitoring::handler::get_dbm_queries))
|
||||
.route("/{org_id}/db_monitoring/query/history", get(db_monitoring::handler::get_dbm_query_history))
|
||||
.route("/{org_id}/db_monitoring/query/endpoints", get(db_monitoring::handler::get_dbm_query_endpoints))
|
||||
.route("/{org_id}/db_monitoring/samples", get(db_monitoring::handler::get_dbm_samples))
|
||||
// Server-vantage events (read the canonical o2_dbm_* columns)
|
||||
.route("/{org_id}/db_monitoring/deadlocks", get(db_monitoring::handler::get_dbm_deadlocks))
|
||||
.route("/{org_id}/db_monitoring/blocking", get(db_monitoring::handler::get_dbm_blocking))
|
||||
.route("/{org_id}/db_monitoring/activity", get(db_monitoring::handler::get_dbm_activity))
|
||||
// `query/insights` returns the query-detail page's Logs-side pair in one
|
||||
// round trip; `query/plans` and `query/server_metrics` are superseded by
|
||||
// it (same sections, same envelopes) and stay registered for
|
||||
// compatibility.
|
||||
.route("/{org_id}/db_monitoring/query/insights", get(db_monitoring::handler::get_dbm_query_insights))
|
||||
.route("/{org_id}/db_monitoring/query/plans", get(db_monitoring::handler::get_dbm_query_plans))
|
||||
.route("/{org_id}/db_monitoring/query/server_metrics", get(db_monitoring::handler::get_dbm_query_server_metrics))
|
||||
.route("/{org_id}/db_monitoring/server_queries", get(db_monitoring::handler::get_dbm_server_queries))
|
||||
.route("/{org_id}/db_monitoring/server_samples", get(db_monitoring::handler::get_dbm_server_samples))
|
||||
.route("/{org_id}/db_monitoring/table_health", get(db_monitoring::handler::get_dbm_table_health))
|
||||
.route("/{org_id}/db_monitoring/instances", get(db_monitoring::handler::get_dbm_instances))
|
||||
.route(
|
||||
"/{org_id}/db_monitoring/instance_metrics",
|
||||
get(db_monitoring::handler::get_dbm_instance_metrics),
|
||||
)
|
||||
.route("/{org_id}/db_monitoring/badges", get(db_monitoring::handler::get_dbm_badges));
|
||||
#[cfg(feature = "enterprise")]
|
||||
let router = {
|
||||
let router = router
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/train", post(anomaly_detection::train_model))
|
||||
.route("/{org_id}/anomaly_detection/{config_id}/detect", post(anomaly_detection::detect_anomalies))
|
||||
.route("/{org_id}/search_jobs", post(search::search_job::submit_job))
|
||||
.route("/{org_id}/search_jobs/{job_id}/result", get(search::search_job::get_job_result))
|
||||
.route("/{org_id}/search_jobs/{job_id}/retry", post(search::search_job::retry_job))
|
||||
.route("/{org_id}/traces/service_graph/topology/current", get(traces::get_current_topology))
|
||||
.route("/{org_id}/traces/service_graph/edge/history", get(traces::get_edge_history))
|
||||
// Agent behavior signals (loop / failure / cost) — reads the _agent_signals stream
|
||||
.route("/{org_id}/traces/agent_signals", get(traces::get_agent_signals))
|
||||
.route("/{org_id}/traces/agent_signals/compare", post(traces::compare_agent_versions))
|
||||
.route("/{org_id}/streams/{stream_name}/patterns/extract", post(patterns::extract_patterns));
|
||||
if llm_eval {
|
||||
router
|
||||
.route("/{org_id}/annotation_queues/{queue_id}/items", post(annotation_queues::enqueue_annotation_queue_item))
|
||||
.route("/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}", get(annotation_queues::get_annotation_queue_item))
|
||||
.route("/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/reviews", get(annotation_queues::list_annotation_queue_item_reviews))
|
||||
.route("/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/push_to_dataset", post(datasets::push_annotation_queue_item_to_dataset))
|
||||
.route("/{org_id}/datasets/{dataset_id}/items", post(datasets::push_dataset_item))
|
||||
.route("/{org_id}/experiments/compare", get(experiments::compare_experiments))
|
||||
.route("/{org_id}/experiments/{experiment_id}", get(experiments::get_experiment))
|
||||
.route("/{org_id}/experiments/{experiment_id}/rows", get(experiments::list_experiment_result_rows))
|
||||
.route("/{org_id}/experiments/{experiment_id}/rows/{row_id}", get(experiments::get_experiment_row))
|
||||
.route("/{org_id}/discovery", get(discovery::list_discovery_items))
|
||||
.route("/{org_id}/tasks/{entity_id}/stats", get(remote_tasks::get_remote_task_stats))
|
||||
} else {
|
||||
router
|
||||
}
|
||||
};
|
||||
// A route layer, not a layer: it must not reach the ungated methods merged onto the same paths.
|
||||
router.route_layer(middleware::from_fn(search_license_middleware))
|
||||
}
|
||||
|
||||
/// On-call management routes; the caller mounts them only when on-call is enabled.
|
||||
#[cfg(feature = "enterprise")]
|
||||
fn oncall_routes(router: Router) -> Router {
|
||||
|
||||
@@ -1683,6 +1683,9 @@ mod tests {
|
||||
#[cfg(feature = "enterprise")]
|
||||
const SERVICE_ROUTES_SIGNATURE: &str = "pub fn service_routes(";
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
const SEARCH_ROUTES_SIGNATURE: &str = "fn search_routes_with(";
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
const HTTP_METHODS: &[&str] = &[
|
||||
"get", "post", "put", "delete", "patch", "head", "options", "trace",
|
||||
@@ -1909,13 +1912,18 @@ mod tests {
|
||||
// Ends at the column-0 close brace, so an item after the body is never scraped into it.
|
||||
#[cfg(feature = "enterprise")]
|
||||
fn service_routes_body() -> &'static str {
|
||||
router_fn_body(SERVICE_ROUTES_SIGNATURE)
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
fn router_fn_body(signature: &str) -> &'static str {
|
||||
let start = ROUTER_SOURCE
|
||||
.find(SERVICE_ROUTES_SIGNATURE)
|
||||
.expect("the router must define service_routes()");
|
||||
let after_signature = &ROUTER_SOURCE[start + SERVICE_ROUTES_SIGNATURE.len()..];
|
||||
.find(signature)
|
||||
.unwrap_or_else(|| panic!("the router must define {signature}"));
|
||||
let after_signature = &ROUTER_SOURCE[start + signature.len()..];
|
||||
let end = after_signature
|
||||
.find("\n}\n")
|
||||
.expect("service_routes() must end at a column-0 close brace");
|
||||
.unwrap_or_else(|| panic!("{signature} must end at a column-0 close brace"));
|
||||
&after_signature[..end]
|
||||
}
|
||||
|
||||
@@ -2014,7 +2022,13 @@ mod tests {
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
fn registered_gated_routes() -> BTreeSet<(String, String)> {
|
||||
let live = normalize(&strip_comments(service_routes_body()));
|
||||
// The licence-gated search routes are a separate group that service_routes() merges in.
|
||||
let routes = [
|
||||
service_routes_body(),
|
||||
router_fn_body(SEARCH_ROUTES_SIGNATURE),
|
||||
]
|
||||
.concat();
|
||||
let live = normalize(&strip_comments(&routes));
|
||||
let mut routes = BTreeSet::new();
|
||||
for call in live.split(".route(").skip(1) {
|
||||
let Some(arguments) = balanced_arguments(call) else {
|
||||
|
||||
@@ -0,0 +1,618 @@
|
||||
// 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use axum::body::Body;
|
||||
use tower::ServiceExt;
|
||||
|
||||
use super::*;
|
||||
|
||||
/// Every route `search_routes_with` registers, as (method, template, handler).
|
||||
#[rustfmt::skip]
|
||||
const GATED: [(&str, &str, &str); 46] = [
|
||||
("POST", "/{org_id}/_search_multi", "search::multi_streams::search_multi"),
|
||||
("POST", "/{org_id}/_search_multi_stream", "search::multi_streams::search_multi_stream"),
|
||||
("POST", "/{org_id}/_search_partition_multi", "search::multi_streams::_search_partition_multi"),
|
||||
("GET", "/{org_id}/{stream_names}/_around_multi", "search::multi_streams::around_multi"),
|
||||
("GET", "/{org_id}/{stream_name}/traces/{trace_id}/details", "traces::details::get_trace_details"),
|
||||
("GET", "/{org_id}/prometheus/api/v1/label/{label_name}/values", "promql::label_values"),
|
||||
("POST", "/v2/{org_id}/downtimes/resources", "downtimes::downtime_resources"),
|
||||
("POST", "/v2/{org_id}/downtimes/values", "downtimes::downtime_values"),
|
||||
("GET", "/{org_id}/db_monitoring/databases", "db_monitoring::handler::get_dbm_databases"),
|
||||
("GET", "/{org_id}/db_monitoring/queries", "db_monitoring::handler::get_dbm_queries"),
|
||||
("GET", "/{org_id}/db_monitoring/query/history", "db_monitoring::handler::get_dbm_query_history"),
|
||||
("GET", "/{org_id}/db_monitoring/query/endpoints", "db_monitoring::handler::get_dbm_query_endpoints"),
|
||||
("GET", "/{org_id}/db_monitoring/samples", "db_monitoring::handler::get_dbm_samples"),
|
||||
("GET", "/{org_id}/db_monitoring/deadlocks", "db_monitoring::handler::get_dbm_deadlocks"),
|
||||
("GET", "/{org_id}/db_monitoring/blocking", "db_monitoring::handler::get_dbm_blocking"),
|
||||
("GET", "/{org_id}/db_monitoring/activity", "db_monitoring::handler::get_dbm_activity"),
|
||||
("GET", "/{org_id}/db_monitoring/query/insights", "db_monitoring::handler::get_dbm_query_insights"),
|
||||
("GET", "/{org_id}/db_monitoring/query/plans", "db_monitoring::handler::get_dbm_query_plans"),
|
||||
("GET", "/{org_id}/db_monitoring/query/server_metrics", "db_monitoring::handler::get_dbm_query_server_metrics"),
|
||||
("GET", "/{org_id}/db_monitoring/server_queries", "db_monitoring::handler::get_dbm_server_queries"),
|
||||
("GET", "/{org_id}/db_monitoring/server_samples", "db_monitoring::handler::get_dbm_server_samples"),
|
||||
("GET", "/{org_id}/db_monitoring/table_health", "db_monitoring::handler::get_dbm_table_health"),
|
||||
("GET", "/{org_id}/db_monitoring/instances", "db_monitoring::handler::get_dbm_instances"),
|
||||
("GET", "/{org_id}/db_monitoring/instance_metrics", "db_monitoring::handler::get_dbm_instance_metrics"),
|
||||
("GET", "/{org_id}/db_monitoring/badges", "db_monitoring::handler::get_dbm_badges"),
|
||||
("POST", "/{org_id}/anomaly_detection/{config_id}/train", "anomaly_detection::train_model"),
|
||||
("POST", "/{org_id}/anomaly_detection/{config_id}/detect", "anomaly_detection::detect_anomalies"),
|
||||
("POST", "/{org_id}/search_jobs", "search::search_job::submit_job"),
|
||||
("GET", "/{org_id}/search_jobs/{job_id}/result", "search::search_job::get_job_result"),
|
||||
("POST", "/{org_id}/search_jobs/{job_id}/retry", "search::search_job::retry_job"),
|
||||
("GET", "/{org_id}/traces/service_graph/topology/current", "traces::get_current_topology"),
|
||||
("GET", "/{org_id}/traces/service_graph/edge/history", "traces::get_edge_history"),
|
||||
("GET", "/{org_id}/traces/agent_signals", "traces::get_agent_signals"),
|
||||
("POST", "/{org_id}/traces/agent_signals/compare", "traces::compare_agent_versions"),
|
||||
("POST", "/{org_id}/streams/{stream_name}/patterns/extract", "patterns::extract_patterns"),
|
||||
("POST", "/{org_id}/annotation_queues/{queue_id}/items", "annotation_queues::enqueue_annotation_queue_item"),
|
||||
("GET", "/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}", "annotation_queues::get_annotation_queue_item"),
|
||||
("GET", "/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/reviews", "annotation_queues::list_annotation_queue_item_reviews"),
|
||||
("POST", "/{org_id}/annotation_queues/{queue_id}/items/{queue_item_id}/push_to_dataset", "datasets::push_annotation_queue_item_to_dataset"),
|
||||
("POST", "/{org_id}/datasets/{dataset_id}/items", "datasets::push_dataset_item"),
|
||||
("GET", "/{org_id}/experiments/compare", "experiments::compare_experiments"),
|
||||
("GET", "/{org_id}/experiments/{experiment_id}", "experiments::get_experiment"),
|
||||
("GET", "/{org_id}/experiments/{experiment_id}/rows", "experiments::list_experiment_result_rows"),
|
||||
("GET", "/{org_id}/experiments/{experiment_id}/rows/{row_id}", "experiments::get_experiment_row"),
|
||||
("GET", "/{org_id}/discovery", "discovery::list_discovery_items"),
|
||||
("GET", "/{org_id}/tasks/{entity_id}/stats", "remote_tasks::get_remote_task_stats"),
|
||||
];
|
||||
|
||||
/// Modules whose every registered handler reads data through search.
|
||||
const FULLY_GATED_MODULES: [&str; 5] = [
|
||||
"search::multi_streams::",
|
||||
"db_monitoring::handler::",
|
||||
"patterns::",
|
||||
"discovery::",
|
||||
"traces::details::",
|
||||
];
|
||||
|
||||
/// Modules that hold gated and ungated handlers side by side.
|
||||
const MIXED_MODULES: [&str; 9] = [
|
||||
"search::search_job",
|
||||
"anomaly_detection",
|
||||
"experiments",
|
||||
"annotation_queues",
|
||||
"datasets",
|
||||
"remote_tasks",
|
||||
"downtimes",
|
||||
"promql",
|
||||
"traces",
|
||||
];
|
||||
|
||||
/// Each ungated handler of a mixed module, with why it stays open.
|
||||
#[rustfmt::skip]
|
||||
const UNGATED_SIBLINGS: &[(&str, &str)] = &[
|
||||
("annotation_queues::archive_annotation_queue_items", "queue bookkeeping, no search"),
|
||||
("annotation_queues::clear_annotation_queue_items", "queue bookkeeping, no search"),
|
||||
("annotation_queues::create_annotation_queue", "queue bookkeeping, no search"),
|
||||
("annotation_queues::delete_annotation_queue", "queue bookkeeping, no search"),
|
||||
("annotation_queues::get_annotation_queue", "queue bookkeeping, no search"),
|
||||
("annotation_queues::list_annotation_queue_items", "stored queue items, no search"),
|
||||
("annotation_queues::list_annotation_queues", "queue bookkeeping, no search"),
|
||||
("annotation_queues::review_annotation_queue_item", "stores a human review, no search"),
|
||||
("annotation_queues::update_annotation_queue", "queue bookkeeping, no search"),
|
||||
("anomaly_detection::cancel_training", "stops work, no search"),
|
||||
("anomaly_detection::create_config", "create-time training waits in the scheduler while blocked"),
|
||||
("anomaly_detection::delete_config", "config bookkeeping, no search"),
|
||||
("anomaly_detection::get_config", "config bookkeeping, no search"),
|
||||
("anomaly_detection::get_detection_history", "operational history of the org's own detections"),
|
||||
("anomaly_detection::list_configs", "config bookkeeping, no search"),
|
||||
("anomaly_detection::update_config", "config bookkeeping, no search"),
|
||||
("datasets::create_dataset", "dataset bookkeeping, no search"),
|
||||
("datasets::delete_dataset", "dataset bookkeeping, no search"),
|
||||
("datasets::delete_dataset_item", "dataset bookkeeping, no search"),
|
||||
("datasets::get_dataset", "dataset bookkeeping, no search"),
|
||||
("datasets::get_dataset_item_versions", "stored dataset rows, no search"),
|
||||
("datasets::get_dataset_snapshot_rows", "stored dataset rows, no search"),
|
||||
("datasets::import_dataset_items", "rows come from the uploaded CSV, not telemetry"),
|
||||
("datasets::list_dataset_items", "stored dataset rows, no search"),
|
||||
("datasets::list_datasets", "dataset bookkeeping, no search"),
|
||||
("datasets::update_dataset", "dataset bookkeeping, no search"),
|
||||
("datasets::update_dataset_item", "rows come from the caller, not telemetry"),
|
||||
("datasets::upsert_dataset_items", "rows come from the caller, not telemetry"),
|
||||
("downtimes::cancel_downtime", "downtime bookkeeping, no search"),
|
||||
("downtimes::create_downtime", "downtime bookkeeping, no search"),
|
||||
("downtimes::delete_downtime", "downtime bookkeeping, no search"),
|
||||
("downtimes::extend_downtime", "downtime bookkeeping, no search"),
|
||||
("downtimes::get_downtime", "downtime bookkeeping, no search"),
|
||||
("downtimes::list_downtimes", "downtime bookkeeping, no search"),
|
||||
("downtimes::move_downtimes", "downtime bookkeeping, no search"),
|
||||
("downtimes::preview_downtime", "inventory match only"),
|
||||
("downtimes::update_downtime", "downtime bookkeeping, no search"),
|
||||
("experiments::cancel_experiment", "stops work, no search"),
|
||||
("experiments::clear_experiment_baseline", "experiment bookkeeping, no search"),
|
||||
("experiments::clone_experiment", "experiment bookkeeping; runs score as background eval work"),
|
||||
("experiments::create_experiment", "runs score as background eval work"),
|
||||
("experiments::delete_experiment", "experiment bookkeeping, no search"),
|
||||
("experiments::finalize_experiment", "experiment bookkeeping, no search"),
|
||||
("experiments::list_experiment_slots", "slot bookkeeping, no search"),
|
||||
("experiments::list_experiments", "summary skipped while blocked (S4)"),
|
||||
("experiments::preview_experiment", "previews the dataset, not telemetry"),
|
||||
("experiments::retry_experiment", "runs score as background eval work"),
|
||||
("experiments::retry_experiment_slot", "runs score as background eval work"),
|
||||
("experiments::set_experiment_baseline", "experiment bookkeeping, no search"),
|
||||
("experiments::submit_experiment_records", "records come from the caller"),
|
||||
("promql::format_query_get", "formats the query text, no data"),
|
||||
("promql::format_query_post", "formats the query text, no data"),
|
||||
("promql::labels_get", "schema only"),
|
||||
("promql::labels_post", "schema only"),
|
||||
("promql::metadata", "schema only"),
|
||||
("promql::parse_tree", "parses the query text, no data"),
|
||||
("promql::query_exemplars_get", "checks the licence in the handler (query_range)"),
|
||||
("promql::query_exemplars_post", "checks the licence in the handler (query_range)"),
|
||||
("promql::query_get", "checks the licence in the handler"),
|
||||
("promql::query_post", "checks the licence in the handler"),
|
||||
("promql::query_range_get", "checks the licence in the handler"),
|
||||
("promql::query_range_post", "checks the licence in the handler"),
|
||||
("promql::remote_write", "ingestion"),
|
||||
("promql::series_get", "checks the licence in the handler"),
|
||||
("promql::series_post", "checks the licence in the handler"),
|
||||
("remote_tasks::activate_remote_task_signing_candidate", "task bookkeeping, no search"),
|
||||
("remote_tasks::create_remote_task", "task bookkeeping, no search"),
|
||||
("remote_tasks::delete_remote_task", "task bookkeeping, no search"),
|
||||
("remote_tasks::discard_remote_task_draft", "task bookkeeping, no search"),
|
||||
("remote_tasks::end_remote_task_signing_grace", "task bookkeeping, no search"),
|
||||
("remote_tasks::get_remote_task", "task bookkeeping, no search"),
|
||||
("remote_tasks::get_remote_task_draft", "task bookkeeping, no search"),
|
||||
("remote_tasks::get_remote_task_signing_status", "task bookkeeping, no search"),
|
||||
("remote_tasks::list_remote_task_versions", "task bookkeeping, no search"),
|
||||
("remote_tasks::list_remote_tasks", "task bookkeeping, no search"),
|
||||
("remote_tasks::publish_remote_task", "calls the remote endpoint, no search"),
|
||||
("remote_tasks::replace_remote_task_auth_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::replace_remote_task_header_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::revoke_remote_task_auth_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::revoke_remote_task_header_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::revoke_remote_task_signing_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::rotate_remote_task_signing_secret", "task bookkeeping, no search"),
|
||||
("remote_tasks::save_remote_task_draft", "task bookkeeping, no search"),
|
||||
("remote_tasks::test_remote_task", "calls the remote endpoint, no search"),
|
||||
("remote_tasks::test_remote_task_signing_candidate", "calls the remote endpoint, no search"),
|
||||
("remote_tasks::test_run_remote_task", "calls the remote endpoint on caller input"),
|
||||
("search::search_job::cancel_job", "job bookkeeping, no search"),
|
||||
("search::search_job::delete_job", "job bookkeeping, no search"),
|
||||
("search::search_job::get_status", "job bookkeeping, no search"),
|
||||
("search::search_job::list_status", "job bookkeeping, no search"),
|
||||
("traces::get_latest_traces", "checks the licence in the handler"),
|
||||
("traces::get_latest_traces_stream", "checks the licence in the handler"),
|
||||
("traces::traces_write", "ingestion"),
|
||||
];
|
||||
|
||||
/// Pages an over-limit customer needs to understand and fix their state; none may be gated.
|
||||
const OPERATIONAL_READS: [&str; 22] = [
|
||||
"/v2/{org_id}/alerts/history",
|
||||
"/v2/{org_id}/reports/history",
|
||||
"/{org_id}/pipelines/history",
|
||||
"/{org_id}/workflows/{id}/history",
|
||||
"/v2/{org_id}/alert_calibration/digests",
|
||||
"/v2/{org_id}/alert_calibration/digests/alerts",
|
||||
"/v2/{org_id}/alert_calibration/digests/records",
|
||||
"/v2/{org_id}/alert_calibration/digests/rules",
|
||||
"/v2/{org_id}/alert_calibration/digests/run",
|
||||
"/{org_id}/ai/chats",
|
||||
"/{org_id}/ai/chats/{session_id}",
|
||||
"/{org_id}/synthetics",
|
||||
"/{org_id}/synthetics/{id}/runs",
|
||||
"/{org_id}/_search_history",
|
||||
"/{org_id}/search/profile",
|
||||
"/license",
|
||||
"/license/refresh",
|
||||
"/{org_id}/config",
|
||||
"/{org_id}/anomaly_detection/{config_id}/history",
|
||||
"/{org_id}/anomaly_detection/history",
|
||||
"/v2/{org_id}/alerts/incidents/{incident_id}",
|
||||
"/{org_id}/search_jobs/{job_id}",
|
||||
];
|
||||
|
||||
const SEARCH_ENTRY: &str = r"(?:search_service|SearchService|crate::search|core_promql::search)::(?:cache::)?(?:search|search_multi|search_partition|search_partition_multi)\(|\bgrpc_search(?:_partition)?\(|process_search_stream_request(?:_multi)?\(|eval_jobs::search::execute|execute_training_query\(";
|
||||
|
||||
/// Every production file that calls a search entry point, and how its search is licence-gated.
|
||||
#[rustfmt::skip]
|
||||
const SEARCH_CALLERS: [(&str, Caller); 39] = [
|
||||
("api/management/src/request/alert_calibration/digests.rs", Caller::Operational),
|
||||
("api/management/src/request/alerts/history.rs", Caller::Exempt("the org's own alert and anomaly run history; its routes stay open")),
|
||||
("api/management/src/request/dashboards/reports_history.rs", Caller::Exempt("the org's own report run history; its route stays open")),
|
||||
("api/management/src/request/db_monitoring/service/common.rs", Caller::Routed("db_monitoring::handler::")),
|
||||
("api/management/src/request/db_monitoring/service/instance_metrics.rs", Caller::Routed("db_monitoring::handler::")),
|
||||
("api/management/src/request/synthetics/last_check.rs", Caller::Exempt("last check on the synthetics list; its route stays open")),
|
||||
("api/management/src/request/workflows/mod.rs", Caller::Exempt("the org's own workflow run history; its route stays open")),
|
||||
("api/pipelines/src/request/pipelines/history.rs", Caller::Exempt("the org's own pipeline run history; its route stays open")),
|
||||
("api/search/src/profiles/mod.rs", Caller::InHandler),
|
||||
("api/search/src/promql/mod.rs", Caller::InHandler),
|
||||
("api/search/src/rum/lookup/query.rs", Caller::Helper("called by rum lookup, which checks the licence in rum/lookup/mod.rs")),
|
||||
("api/search/src/search/around.rs", Caller::Helper("around v1/v2 check in search/mod.rs; around_multi is in the gated group")),
|
||||
("api/search/src/search/mod.rs", Caller::InHandler),
|
||||
("api/search/src/search/multi_streams.rs", Caller::Routed("search::multi_streams::")),
|
||||
("api/search/src/search/patterns.rs", Caller::Routed("patterns::")),
|
||||
("api/search/src/search/search_inspector.rs", Caller::Exempt("search/profile stays open (D13)")),
|
||||
("api/search/src/search/search_stream.rs", Caller::InHandler),
|
||||
("api/search/src/traces/dag.rs", Caller::InHandler),
|
||||
("api/search/src/traces/details.rs", Caller::Routed("traces::details::")),
|
||||
("api/search/src/traces/list/query.rs", Caller::Helper("latest traces; traces/mod.rs checks the licence")),
|
||||
("api/search/src/traces/list/stream.rs", Caller::Helper("latest traces; traces/mod.rs checks the licence")),
|
||||
("api/search/src/traces/session.rs", Caller::InHandler),
|
||||
("api/search/src/traces/time_index/mod.rs", Caller::InHandler),
|
||||
("api/search/src/traces/user.rs", Caller::InHandler),
|
||||
("core/src/ai_chat.rs", Caller::Operational),
|
||||
("core/src/alerts/alert_calibration/mod.rs", Caller::Background("calibration runs")),
|
||||
("core/src/alerts/mod.rs", Caller::Background("alert evaluation")),
|
||||
("core/src/alerts/notifications/chart/mod.rs", Caller::Background("notification charts")),
|
||||
("core/src/anomaly_detection/partitioned_search.rs", Caller::Background("anomaly detection runs")),
|
||||
("core/src/db_monitoring/rollup.rs", Caller::Background("DBM rollup")),
|
||||
("core/src/downtimes/values.rs", Caller::Routed("downtimes::")),
|
||||
("core/src/metrics/prom.rs", Caller::Routed("promql::label_values")),
|
||||
("core/src/self_reporting/search.rs", Caller::Exempt("licence usage from the meta org")),
|
||||
("core/src/slo/job.rs", Caller::Background("SLO job")),
|
||||
("core/src/slo/reconcile.rs", Caller::Background("SLO reconcile")),
|
||||
("core/src/traces/agent_signals/api.rs", Caller::Routed("traces::get_agent_signals")),
|
||||
("core/src/traces/service_graph/mod.rs", Caller::Routed("traces::get_current_topology")),
|
||||
("jobs/src/job/mod.rs", Caller::Background("eval-job scoring")),
|
||||
("jobs/src/job/red_insights.rs", Caller::Background("red_insights")),
|
||||
];
|
||||
|
||||
/// How a search call site is licence-gated.
|
||||
enum Caller {
|
||||
/// The file calls `check_search_allowed` itself.
|
||||
InHandler,
|
||||
/// The file serves handlers under this prefix, which sit in the gated group.
|
||||
Routed(&'static str),
|
||||
/// The file reads only an operational stream.
|
||||
Operational,
|
||||
Background(&'static str),
|
||||
Helper(&'static str),
|
||||
Exempt(&'static str),
|
||||
}
|
||||
|
||||
type Registration = (String, String, String);
|
||||
|
||||
fn production_source() -> &'static str {
|
||||
include_str!("mod.rs")
|
||||
.split_once("\n#[cfg(test)]\nmod tests {")
|
||||
.unwrap()
|
||||
.0
|
||||
}
|
||||
|
||||
fn fn_body<'a>(source: &'a str, signature: &str) -> &'a str {
|
||||
let start = source.find(signature).unwrap();
|
||||
&source[start..start + source[start..].find("\n}\n").unwrap()]
|
||||
}
|
||||
|
||||
/// The `/api` tree the router serves, minus the gated group.
|
||||
fn api_span_outside_the_group() -> String {
|
||||
let source = production_source();
|
||||
let section = |from: &str, to: &str| {
|
||||
let start = source.find(from).unwrap();
|
||||
&source[start..start + source[start..].find(to).unwrap()]
|
||||
};
|
||||
let api = section("fn ai_chat_routes(", "pub fn other_service_routes(");
|
||||
let group = fn_body(source, "fn search_routes_with(");
|
||||
format!(
|
||||
"{}{}",
|
||||
api.replace(group, ""),
|
||||
section("fn es_compat_routes(", "\n}\n")
|
||||
)
|
||||
}
|
||||
|
||||
/// Every (method, template, handler) the `.route(` calls in `text` register.
|
||||
fn registrations(text: &str) -> BTreeSet<Registration> {
|
||||
let handler = regex::Regex::new(
|
||||
r"(?:^|[\s(.,])(get|post|put|delete|patch|head)\(([A-Za-z_][A-Za-z0-9_:]*)\)",
|
||||
)
|
||||
.unwrap();
|
||||
let mut found = BTreeSet::new();
|
||||
for (at, _) in text.match_indices(".route(") {
|
||||
let args = &text[at + ".route(".len()..];
|
||||
let mut depth = 1;
|
||||
let end = args
|
||||
.char_indices()
|
||||
.find(|(_, c)| {
|
||||
depth += match c {
|
||||
'(' => 1,
|
||||
')' => -1,
|
||||
_ => 0,
|
||||
};
|
||||
depth == 0
|
||||
})
|
||||
.unwrap()
|
||||
.0;
|
||||
let args = &args[..end];
|
||||
let template = args.split('"').nth(1).unwrap().to_string();
|
||||
for c in handler.captures_iter(args) {
|
||||
found.insert((c[1].to_uppercase(), template.clone(), c[2].to_string()));
|
||||
}
|
||||
}
|
||||
found
|
||||
}
|
||||
|
||||
fn gated() -> BTreeSet<Registration> {
|
||||
GATED
|
||||
.iter()
|
||||
.map(|(m, t, h)| (m.to_string(), t.to_string(), h.to_string()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn module_of(handler: &str) -> &str {
|
||||
handler.rsplit_once("::").map_or("", |(module, _)| module)
|
||||
}
|
||||
|
||||
fn search_call_files() -> BTreeSet<String> {
|
||||
let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
|
||||
let mut roots: Vec<_> = std::fs::read_dir(src.join("api"))
|
||||
.unwrap()
|
||||
.map(|e| e.unwrap().path().join("src"))
|
||||
.collect();
|
||||
roots.extend(["core/src", "mcp/src", "synthetics/src", "jobs/src"].map(|d| src.join(d)));
|
||||
let entry = regex::Regex::new(SEARCH_ENTRY).unwrap();
|
||||
let mut hits = BTreeSet::new();
|
||||
let mut pending = roots;
|
||||
while let Some(path) = pending.pop() {
|
||||
if path.is_dir() {
|
||||
pending.extend(std::fs::read_dir(&path).unwrap().map(|e| e.unwrap().path()));
|
||||
continue;
|
||||
}
|
||||
let rel = path
|
||||
.strip_prefix(&src)
|
||||
.unwrap()
|
||||
.to_string_lossy()
|
||||
.replace('\\', "/");
|
||||
// This file quotes the entry pattern, so scanning it would find itself.
|
||||
if !rel.ends_with(".rs") || rel.ends_with("router/search_gate_tests.rs") {
|
||||
continue;
|
||||
}
|
||||
let text = std::fs::read_to_string(&path).unwrap();
|
||||
let production = text.split("\n#[cfg(test)]\nmod tests").next().unwrap();
|
||||
if entry.is_match(production) {
|
||||
hits.insert(rel);
|
||||
}
|
||||
}
|
||||
hits
|
||||
}
|
||||
|
||||
fn caller_source(file: &str) -> String {
|
||||
let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../..")
|
||||
.join(file);
|
||||
let text = std::fs::read_to_string(path).unwrap();
|
||||
text.split("\n#[cfg(test)]\nmod tests")
|
||||
.next()
|
||||
.unwrap()
|
||||
.to_string()
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
fn filled(template: &str) -> String {
|
||||
template
|
||||
.split('/')
|
||||
.map(|segment| match segment {
|
||||
"{org_id}" => "acme".to_string(),
|
||||
"{stream_name}" => "s1".to_string(),
|
||||
"{job_id}" => "j1".to_string(),
|
||||
"{stream_names}" => config::utils::base64::encode_url("s1"),
|
||||
s if s.starts_with('{') => "p1".to_string(),
|
||||
s => s.to_string(),
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join("/")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn search_routes_hold_exactly_the_gated_routes() {
|
||||
let group = registrations(fn_body(production_source(), "fn search_routes_with("));
|
||||
assert_eq!(group, gated());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_gated_handler_is_registered_outside_search_routes() {
|
||||
let outside = registrations(&api_span_outside_the_group());
|
||||
let gated_handlers: BTreeSet<String> = GATED.iter().map(|(_, _, h)| h.to_string()).collect();
|
||||
let siblings: BTreeSet<&str> = UNGATED_SIBLINGS.iter().map(|(h, _)| *h).collect();
|
||||
let mut seen_siblings = BTreeSet::new();
|
||||
for (method, template, handler) in &outside {
|
||||
let route = format!("{method} {template} {handler}");
|
||||
assert!(
|
||||
!gated_handlers.contains(handler),
|
||||
"gated handler outside the group: {route}"
|
||||
);
|
||||
assert!(
|
||||
!FULLY_GATED_MODULES.iter().any(|m| handler.starts_with(m)),
|
||||
"a fully gated module registers outside the group: {route}"
|
||||
);
|
||||
if MIXED_MODULES.contains(&module_of(handler)) {
|
||||
assert!(
|
||||
siblings.contains(handler.as_str()),
|
||||
"unclassified handler: {route}"
|
||||
);
|
||||
seen_siblings.insert(handler.as_str());
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
seen_siblings, siblings,
|
||||
"UNGATED_SIBLINGS lists a handler no route registers"
|
||||
);
|
||||
assert!(
|
||||
UNGATED_SIBLINGS
|
||||
.iter()
|
||||
.all(|(_, reason)| !reason.is_empty())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn operational_reads_stay_outside_the_gate() {
|
||||
let group: BTreeSet<String> =
|
||||
registrations(fn_body(production_source(), "fn search_routes_with("))
|
||||
.into_iter()
|
||||
.map(|(_, template, _)| template)
|
||||
.collect();
|
||||
let outside: BTreeSet<String> = registrations(&api_span_outside_the_group())
|
||||
.into_iter()
|
||||
.map(|(_, template, _)| template)
|
||||
.collect();
|
||||
for template in OPERATIONAL_READS {
|
||||
assert!(
|
||||
outside.contains(template),
|
||||
"{template} is no longer registered"
|
||||
);
|
||||
assert!(
|
||||
!group.contains(template),
|
||||
"{template} must stay outside the gate"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_gate_is_a_route_layer_on_the_group_only() {
|
||||
let source = production_source();
|
||||
assert_eq!(
|
||||
source.matches("from_fn(search_license_middleware)").count(),
|
||||
1
|
||||
);
|
||||
assert!(
|
||||
fn_body(source, "fn search_routes_with(")
|
||||
.contains(".route_layer(middleware::from_fn(search_license_middleware))")
|
||||
);
|
||||
assert!(!source.contains(".layer(middleware::from_fn(search_license_middleware"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn route_layer_does_not_reach_methods_merged_on_the_same_path() {
|
||||
const PATH: &str = "/v2/{org_id}/downtimes/values";
|
||||
let _built = Router::new()
|
||||
.merge(search_routes_with(false))
|
||||
.route(PATH, get(|| async { "stub" }));
|
||||
let closed = middleware::from_fn(|_: Request, _: Next| async {
|
||||
common::meta::http::HttpResponse::too_many_requests("blocked")
|
||||
});
|
||||
let app = Router::new()
|
||||
.merge(
|
||||
Router::new()
|
||||
.route(PATH, post(|| async { "gated" }))
|
||||
.route_layer(closed),
|
||||
)
|
||||
.route(PATH, get(|| async { "open" }));
|
||||
let send = |method: Method| {
|
||||
app.clone().oneshot(
|
||||
axum::http::Request::builder()
|
||||
.method(method)
|
||||
.uri("/v2/acme/downtimes/values")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
};
|
||||
assert_eq!(send(Method::GET).await.unwrap().status(), StatusCode::OK);
|
||||
assert_eq!(
|
||||
send(Method::POST).await.unwrap().status(),
|
||||
StatusCode::TOO_MANY_REQUESTS
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_search_call_site_is_classified() {
|
||||
let listed: BTreeSet<String> = SEARCH_CALLERS.iter().map(|(f, _)| f.to_string()).collect();
|
||||
assert_eq!(
|
||||
search_call_files(),
|
||||
listed,
|
||||
"classify every search call site in SEARCH_CALLERS"
|
||||
);
|
||||
let operational = ["DIGEST_STREAM", "AI_CHAT_EVENTS_STREAM"]
|
||||
.into_iter()
|
||||
.chain(search_service::OPERATIONAL_STREAMS);
|
||||
let operational: Vec<&str> = operational.collect();
|
||||
for (file, class) in SEARCH_CALLERS {
|
||||
match class {
|
||||
Caller::InHandler => {
|
||||
assert!(
|
||||
caller_source(file).contains("check_search_allowed("),
|
||||
"{file}"
|
||||
)
|
||||
}
|
||||
Caller::Routed(prefix) => {
|
||||
assert!(
|
||||
GATED.iter().any(|(_, _, h)| h.starts_with(prefix)),
|
||||
"{file}"
|
||||
)
|
||||
}
|
||||
Caller::Operational => {
|
||||
let text = caller_source(file);
|
||||
assert!(operational.iter().any(|name| text.contains(name)), "{file}")
|
||||
}
|
||||
Caller::Background(reason) | Caller::Helper(reason) | Caller::Exempt(reason) => {
|
||||
assert!(!reason.is_empty(), "{file}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
#[tokio::test]
|
||||
async fn closed_gate_refuses_every_gated_route() {
|
||||
use o2_enterprise::enterprise::license::{TEST_STATICS_LOCK, close_search_gate_for_tests};
|
||||
|
||||
let _gate = close_search_gate_for_tests(TEST_STATICS_LOCK.lock().await);
|
||||
let app = search_routes_with(true);
|
||||
for (method, template, handler) in GATED {
|
||||
let request = axum::http::Request::builder()
|
||||
.method(method)
|
||||
.uri(filled(template))
|
||||
.body(Body::empty())
|
||||
.unwrap();
|
||||
let response = app.clone().oneshot(request).await.unwrap();
|
||||
assert_eq!(
|
||||
response.status(),
|
||||
StatusCode::TOO_MANY_REQUESTS,
|
||||
"{method} {template} {handler}"
|
||||
);
|
||||
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
|
||||
.await
|
||||
.unwrap();
|
||||
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
|
||||
assert_eq!(json["code"], 429, "{template}");
|
||||
let message = json["message"].as_str().unwrap();
|
||||
assert!(
|
||||
message.starts_with(
|
||||
"Error# Search is temporarily disabled due to exceeding allotted ingestion limit."
|
||||
) && message.ends_with("Please contact your administrator."),
|
||||
"{template}: {message}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
#[tokio::test]
|
||||
async fn meta_org_passes_the_closed_gate() {
|
||||
use o2_enterprise::enterprise::license::{TEST_STATICS_LOCK, close_search_gate_for_tests};
|
||||
|
||||
let _gate = close_search_gate_for_tests(TEST_STATICS_LOCK.lock().await);
|
||||
let request = axum::http::Request::post(format!("/{}/_search_multi", config::META_ORG_ID))
|
||||
.body(Body::empty())
|
||||
.unwrap();
|
||||
let response = search_routes_with(true).oneshot(request).await.unwrap();
|
||||
assert_ne!(response.status(), StatusCode::TOO_MANY_REQUESTS);
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
#[tokio::test]
|
||||
async fn open_gate_lets_a_gated_route_through() {
|
||||
let _held = o2_enterprise::enterprise::license::TEST_STATICS_LOCK
|
||||
.lock()
|
||||
.await;
|
||||
let app = Router::new()
|
||||
.route("/{org_id}/{stream_name}/probe", get(|| async { "ok" }))
|
||||
.route_layer(middleware::from_fn(search_license_middleware));
|
||||
let request = axum::http::Request::get("/acme/logs/probe")
|
||||
.body(Body::empty())
|
||||
.unwrap();
|
||||
assert_eq!(app.oneshot(request).await.unwrap().status(), StatusCode::OK);
|
||||
}
|
||||
Reference in New Issue
Block a user