mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 16:10:43 +00:00
sktrpathi
75fc81cecf
### **User description** A significant chunk of API unit tests were not testing any actual business logic, but performing trivial actions such as struct-field-access, derive trait validation such as `Eq`, `PartialEq`, `Clone`, `Serialize` etc, and many of the tests were also just comparing constant values. They added no value to the test suite and have now been removed from the codebase, making the tests more focused. Additionally, - A lot of flaky tests have been corrected. These used to fail due to occasional race conditions when ran locally. - Some mock setup and teardown functions have also been cleaned up, which makes tests a lot more consistent. ___ ### **PR Type** Tests, Enhancement ___ ### **Description** - Remove trivial, non-value unit tests - Relax flaky assertions and panics handling - Improve test isolation via DB/cache resets - Minor test output formatting improvements ___ ### Diagram Walkthrough ```mermaid flowchart LR tests["Remove trivial/boilerplate tests"] -- cleaner suite --> stability["Stabilize assertions and panics"] stability -- reduce flakiness --> isolation["DB/Cache resets in tests"] isolation -- consistent runs --> outputs["Improve test log formatting"] ``` <details> <summary><h3> File Walkthrough</h3></summary> <table><thead><tr><th></th><th align="left">Relevant files</th></tr></thead><tbody><tr><td><strong>Tests</strong></td><td><details><summary>11 files</summary><table> <tr> <td><strong>organization.rs</strong><dd><code>Remove struct-field and default-value trivial tests</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-4f089715a7ac4f78b45039d773f2fccbb4fbc6a409e0ab9ceee1fcf7911bc44b">+4/-325</a> </td> </tr> <tr> <td><strong>mod.rs</strong><dd><code>Drop JSON roundtrip serialization tests; tidy imports</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-c4bd542e4e07ed7706b0643086f5a7d4edbb3647c7ceedbe4bea516c5274d8bd">+1/-183</a> </td> </tr> <tr> <td><strong>auth.rs</strong><dd><code>Remove Debug/PartialEq trivial tests; format strings</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-9bc2b1841504a435f66ad094f72ddbe7b7fa75e5e1557b8adbfe36f7f0592e1b">+2/-104</a> </td> </tr> <tr> <td><strong>main.rs</strong><dd><code>Make tracing test tolerant; ignore enterprise tests</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-42cb6807ad74b3e201c5a7ca98b911c5fa08380e942be6e4ac5807f8377f87fc">+23/-60</a> </td> </tr> <tr> <td><strong>event.rs</strong><dd><code>Remove constants and simple struct creation tests</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-689769712cbd6bd02beba078c1e1dfc188e2be39ab19a0e232b8bd12bc7f1e78">+1/-95</a> </td> </tr> <tr> <td><strong>validator.rs</strong><dd><code>Ensure test isolation with ORM init and cache clears</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-f4d4b51574b378a8c372345ea48875834a4eadacfac33fa4d0ceca1e0ef96b35">+28/-22</a> </td> </tr> <tr> <td><strong>mod.rs</strong><dd><code>Make base URI tests robust to env errors</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-d7987e62dcc7b1ec1bd4aeabef94f85d8c41b2ea33de5ab78660794fa4d51a9e">+25/-4</a> </td> </tr> <tr> <td><strong>users.rs</strong><dd><code>Use best-effort cleanup; clear caches; improve logs</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-98da2588a28d62bb2e301d989836125f89f89b76272cd7f75e08ec4deeea5f38">+12/-9</a> </td> </tr> <tr> <td><strong>integration_test.rs</strong><dd><code>Compact println formatting and panic messages</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-7ff815b68943f08bef79f083c09e55ca8c18db8e25033034b12fa31f1d43dab4">+3/-3</a> </td> </tr> <tr> <td><strong>config.rs</strong><dd><code>Remove strict zero-length assertions on globals</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-cda0d9179ed982fa5f0989c9b26f21cfa49dd1de28195a98135650ffcc7c602a">+1/-13</a> </td> </tr> <tr> <td><strong>schema.rs</strong><dd><code>Remove unused constant import in tests</code> </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-a5ea87a7766cb24619f64a6100f385f4b42c47af7d33b8e2f060229c536d7418">+1/-8</a> </td> </tr> </table></details></td></tr><tr><td><strong>Additional files</strong></td><td><details><summary>26 files</summary><table> <tr> <td><strong>authz.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-28a815d1c4728aff854a9996de0ba3d3dafe1bea6bfb0cf2a9f4186c40f985b6">+0/-52</a> </td> </tr> <tr> <td><strong>loki.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-121c166af5eab8b9cb83716afa7b65fa7856c006c809b3d910ced184c52ec587">+0/-7</a> </td> </tr> <tr> <td><strong>proxy.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-7f12a4f2483e737671ac41f9c35b411e7323e0ff2731ad92a3b309e144d1316a">+0/-46</a> </td> </tr> <tr> <td><strong>service.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-61d456e7217d19bcec3920a4d1c17fd396a257c81e5394655db3632a057bdd9e">+0/-7</a> </td> </tr> <tr> <td><strong>service_account.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-878ad520e1c84500cc21ff6fc4088c8da64ff2febd79e60c0d4cc8c695c49798">+0/-80</a> </td> </tr> <tr> <td><strong>traces.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-bda8182c1f17a8360337c769edade0cb9478ff1443dd6aedce7f526afbe97afe">+0/-106</a> </td> </tr> <tr> <td><strong>user.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-e13b3a981fd70182b9738a6b0ecfdee915c5b36757703a87741e1eddf3f6e786">+0/-101</a> </td> </tr> <tr> <td><strong>auth_tests.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-2688cbdc0b009b3f0dd829c69ab86c6a7d0778ab3e65e916a410a76aa6ae6480">+0/-2961</a></td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-2aa59ce4464e7dbabdb9367fc8bb8c1b489c914b8690577d4916870463ab12db">+0/-1</a> </td> </tr> <tr> <td><strong>action.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-fc87c75a7df9539d7e7ee6184dfab02d4a8932323e81f95d57c05363aa44752d">+0/-18</a> </td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-5bf716db7e1d91dfbd427a2e320ce407c4f0c92d55d68f8e636daed1b02abd56">+0/-13</a> </td> </tr> <tr> <td><strong>reports.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-d2117a3d67d4ec47b0017861845c92143baf917e10e6746df5dd97f925f81eeb">+0/-10</a> </td> </tr> <tr> <td><strong>destinations.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-e0ba33d95ef7f0eac13bdac7e750462907a0308f01c027ea206d058e282c977b">+0/-103</a> </td> </tr> <tr> <td><strong>inverted_index.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-8bf9d9403f05f53f0c6fd5e6a6dbf0204ea80465b233567552cb7395d08d1c3c">+0/-22</a> </td> </tr> <tr> <td><strong>meta_store.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-45018c838da377fa9ded9305a1ee752adde06acdb607addf26c95f5747537615">+0/-6</a> </td> </tr> <tr> <td><strong>ratelimit.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-fa7dc03fb526c09d5eb58021872d821fcd72eeb4d6a670224f5286012485bc3d">+0/-10</a> </td> </tr> <tr> <td><strong>timed_annotations.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-f94b4d6d12d68e7054ad33601e666b08e4594586013dc34384fa21b1df5fe6e8">+0/-12</a> </td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-ca09c9c9b020049bdbc3382c466bfcca80daac92e98760d9a9a76172abf34049">+0/-34</a> </td> </tr> <tr> <td><strong>m20250611_000002_populate_reports_table.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-8247af6d6f8ccb6a5a51daff4cb1e352585be2d1c785d86ae5ebfea1dbc1099c">+0/-1</a> </td> </tr> <tr> <td><strong>folders.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-402482c22fda5dff8f33382ac11d0908f693975dec7c08b37b1c5fe24a20486a">+0/-1</a> </td> </tr> <tr> <td><strong>bulk.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-2017dbe9076c78a7e7919ed5c8738729123d630038803cef0ce2c356d6fa2c48">+0/-9</a> </td> </tr> <tr> <td><strong>ingestion.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-3edd5f4ed1eed0f765b7b450ef0fbf587632e6b22302006a5e397049e297057f">+0/-7</a> </td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-2f7f0bac01afc6e37908521d1d33d05fcd650b7a08792062976e1e7465b4a1f9">+0/-42</a> </td> </tr> <tr> <td><strong>footer_cache.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-f9278e53ed7131c953fc30f8b15cda801b74b81fb2d570fe7f80a3b7fc1de872">+0/-7</a> </td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-1602f85399080c1e2838d1ef1d62af23d96f353379e49a7770026e892c954018">+0/-20</a> </td> </tr> <tr> <td><strong>mod.rs</strong></td> <td><a href="https://github.com/openobserve/openobserve/pull/8172/files#diff-66f3b4f5e7927092fd6e57738687a1091b6f316638f7c58f8d7be9addb72bc30">+0/-31</a> </td> </tr> </table></details></td></tr></tr></tbody></table> </details> ___
2964 lines
112 KiB
Rust
2964 lines
112 KiB
Rust
// Copyright 2025 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/>.
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use core::time;
|
|
use std::{env, fs, net::SocketAddr, str, sync::Once, thread};
|
|
|
|
use actix_web::{App, http::header::ContentType, test, web};
|
|
use arrow_flight::flight_service_server::FlightServiceServer;
|
|
use bytes::{Bytes, BytesMut};
|
|
use chrono::{Duration, Utc};
|
|
use config::{
|
|
get_config,
|
|
meta::{
|
|
alerts::{Operator, QueryCondition, TriggerCondition, alert::Alert},
|
|
dashboards::{Dashboard, v1},
|
|
pipeline::{
|
|
Pipeline,
|
|
components::{DerivedStream, PipelineSource},
|
|
},
|
|
stream::StreamType,
|
|
triggers::{ScheduledTriggerData, Trigger, TriggerModule, TriggerStatus},
|
|
},
|
|
utils::json,
|
|
};
|
|
use openobserve::{
|
|
common::meta::ingestion::IngestionResponse,
|
|
handler::{
|
|
grpc::{auth::check_auth, flight::FlightServiceImpl},
|
|
http::{
|
|
models::{
|
|
alerts::responses::{GetAlertResponseBody, ListAlertsResponseBody},
|
|
destinations::{Destination, DestinationType},
|
|
},
|
|
router::*,
|
|
},
|
|
},
|
|
migration,
|
|
service::{alerts::scheduler::handlers::handle_triggers, search::SEARCH_SERVER},
|
|
};
|
|
use prost::Message;
|
|
use proto::{cluster_rpc::search_server::SearchServer, prometheus_rpc};
|
|
use tonic::codec::CompressionEncoding;
|
|
|
|
static START: Once = Once::new();
|
|
|
|
fn setup() -> (&'static str, &'static str) {
|
|
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_logger::init_from_env(
|
|
env_logger::Env::new().default_filter_or(&get_config().log.level),
|
|
);
|
|
|
|
log::info!("setup Invoked");
|
|
});
|
|
(
|
|
"Authorization",
|
|
"Basic cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM=",
|
|
)
|
|
}
|
|
|
|
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);
|
|
|
|
log::info!("starting gRPC server at {}", gaddr);
|
|
tonic::transport::Server::builder()
|
|
.layer(tonic::service::interceptor(check_auth))
|
|
.add_service(search_svc)
|
|
.add_service(flight_svc)
|
|
.serve(gaddr)
|
|
.await
|
|
.expect("gRPC server init failed");
|
|
Ok(())
|
|
}
|
|
|
|
async fn e2e_100_tear_down() {
|
|
log::info!("Tear Down Invoked");
|
|
fs::remove_dir_all("./data").expect("Delete local dir failed");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn e2e_test() {
|
|
// make sure data dir is deleted before we run integration tests
|
|
fs::remove_dir_all("./data")
|
|
.unwrap_or_else(|e| log::info!("Error deleting local dir: {}", e));
|
|
|
|
setup();
|
|
|
|
// start gRPC server
|
|
tokio::task::spawn(async move {
|
|
init_grpc_server()
|
|
.await
|
|
.expect("router gRPC server init failed");
|
|
});
|
|
|
|
// register node
|
|
openobserve::common::infra::cluster::register_and_keep_alive()
|
|
.await
|
|
.unwrap();
|
|
// init config
|
|
config::init().await.unwrap();
|
|
// init infra
|
|
migration::init_db().await.unwrap();
|
|
infra::init().await.unwrap();
|
|
// ensure database tables are created
|
|
infra::db::create_table().await.unwrap();
|
|
// db migration steps, since it's separated out
|
|
infra::table::migrate().await.unwrap();
|
|
openobserve::common::infra::init().await.unwrap();
|
|
// ingester init
|
|
ingester::init().await.unwrap();
|
|
// init job
|
|
openobserve::job::init().await.unwrap();
|
|
|
|
for _i in 0..3 {
|
|
e2e_1_post_bulk().await;
|
|
}
|
|
|
|
// ingest
|
|
e2e_post_json().await;
|
|
e2e_post_multi().await;
|
|
e2e_post_trace().await;
|
|
e2e_post_metrics().await;
|
|
e2e_post_hec().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_org().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_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.clone());
|
|
|
|
let board = e2e_update_dashboard(
|
|
v1::Dashboard {
|
|
title: "e2e test".to_owned(),
|
|
description: "Logs flow downstream".to_owned(),
|
|
..board.v1.unwrap()
|
|
},
|
|
board.hash,
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
e2e_get_dashboard(&board.clone().v1.unwrap().dashboard_id).await,
|
|
board
|
|
);
|
|
e2e_delete_dashboard(&board.v1.unwrap().dashboard_id).await;
|
|
assert!(e2e_list_dashboards().await.is_empty());
|
|
}
|
|
|
|
// 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;
|
|
|
|
// derived streams
|
|
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;
|
|
e2e_handle_derived_stream_evaluation_failure().await;
|
|
e2e_cleanup_test_pipeline().await;
|
|
|
|
// syslog
|
|
e2e_post_syslog_route().await;
|
|
e2e_list_syslog_routes().await;
|
|
|
|
// others
|
|
e2e_health_check().await;
|
|
e2e_config().await;
|
|
e2e_100_tear_down().await;
|
|
|
|
// clear
|
|
e2e_delete_stream().await;
|
|
}
|
|
|
|
async fn e2e_1_post_bulk() {
|
|
let auth = setup();
|
|
let path = "./tests/input.json";
|
|
let body_str = fs::read_to_string(path).expect("Read file failed");
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_bulk", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_json() {
|
|
let auth = setup();
|
|
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
|
|
// timestamp in past
|
|
let body_str = "[{\"Year\": 1896, \"City\": \"Athens\", \"Sport\": \"Aquatics\", \"Discipline\": \"Swimming\", \"Athlete\": \"HERSCHMANN, Otto\", \"Country\": \"AUT\", \"Gender\": \"Men\", \"Event\": \"100M Freestyle\", \"Medal\": \"Silver\", \"Season\": \"summer\",\"_timestamp\":1665136888163792}]";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/_json", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let res: IngestionResponse = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(res.code, 200);
|
|
assert_eq!(res.status.len(), 1);
|
|
assert_eq!(res.status[0].status.successful, 0);
|
|
assert_eq!(res.status[0].status.failed, 1);
|
|
assert!(res.status[0].status.error.contains("Too old data"));
|
|
assert!(
|
|
res.status[0]
|
|
.status
|
|
.error
|
|
.contains("ZO_INGEST_ALLOWED_UPTO=")
|
|
);
|
|
|
|
// timestamp in future
|
|
let body_str = "[{\"Year\": 1896, \"City\": \"Athens\", \"Sport\": \"Aquatics\", \"Discipline\": \"Swimming\", \"Athlete\": \"HERSCHMANN, Otto\", \"Country\": \"AUT\", \"Gender\": \"Men\", \"Event\": \"100M Freestyle\", \"Medal\": \"Silver\", \"Season\": \"summer\",\"_timestamp\":9999999999999999}]";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/_json", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let res: IngestionResponse = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(res.code, 200);
|
|
assert_eq!(res.status.len(), 1);
|
|
assert_eq!(res.status[0].status.successful, 0);
|
|
assert_eq!(res.status[0].status.failed, 1);
|
|
assert!(res.status[0].status.error.contains("Too far data"));
|
|
assert!(
|
|
res.status[0]
|
|
.status
|
|
.error
|
|
.contains("ZO_INGEST_ALLOWED_IN_FUTURE=")
|
|
);
|
|
|
|
// timestamp not present
|
|
let body_str = "[{\"Year\": 1896, \"City\": \"Athens\", \"Sport\": \"Aquatics\", \"Discipline\": \"Swimming\", \"Athlete\": \"HERSCHMANN, Otto\", \"Country\": \"AUT\", \"Gender\": \"Men\", \"Event\": \"100M Freestyle\", \"Medal\": \"Silver\", \"Season\": \"summer\"}]";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/_json", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let res: IngestionResponse = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(res.code, 200);
|
|
assert_eq!(res.status.len(), 1);
|
|
assert_eq!(res.status[0].status.successful, 1);
|
|
assert_eq!(res.status[0].status.failed, 0);
|
|
|
|
// timestamp just right
|
|
let ts = chrono::Utc::now().timestamp_micros();
|
|
let body_str = format!(
|
|
"[{{\"Year\": 1896, \"City\": \"Athens\", \"Sport\": \"Aquatics\", \"Discipline\": \"Swimming\", \"Athlete\": \"HERSCHMANN, Otto\", \"Country\": \"AUT\", \"Gender\": \"Men\", \"Event\": \"100M Freestyle\", \"Medal\": \"Silver\", \"Season\": \"summer\",\"_timestamp\":{ts}}}]"
|
|
);
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/_json", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let res: IngestionResponse = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(res.code, 200);
|
|
assert_eq!(res.status.len(), 1);
|
|
assert_eq!(res.status[0].status.successful, 1);
|
|
assert_eq!(res.status[0].status.failed, 0);
|
|
}
|
|
|
|
async fn e2e_post_hec() {
|
|
let auth = setup();
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
|
|
// test case : missing index in metadata
|
|
let body_str = "{\"event\":\"hello\"}";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_hec", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// test case : valid payload
|
|
let body_str = "{\"event\":\"hello\",\"index\":\"hec_test\"}";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_hec", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// test case : json event
|
|
let body_str =
|
|
"{\"event\":{\"log\":\"hello\",\"severity\":\"info\"},\"index\":\"hec_test\"}";
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_hec", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// test case : ndjson
|
|
let body_str = r#"
|
|
{ "index": "hec_test", "event": "test log", "time": 1749113798091 }
|
|
{ "index": "hec_test", "event": {"log":"test log","severity":"info"}, "fields": {"cluster":"c1", "namespace":"n1"} }
|
|
{ "index": "hec_test", "event": {"log":"test log","severity":"info"}, "source" : "e2e_test", "fields": {"cluster":"c1", "namespace":"n1"}}
|
|
"#;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_hec", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_multi() {
|
|
let auth = setup();
|
|
let body_str = "{\"Year\": 1896, \"City\": \"Athens\", \"Sport\": \"Aquatics\", \"Discipline\": \"Swimming\", \"Athlete\": \"HERSCHMANN, Otto\", \"Country\": \"AUT\", \"Gender\": \"Men\", \"Event\": \"100M Freestyle\", \"Medal\": \"Silver\", \"Season\": \"summer\",\"_timestamp\":1665136888163792}";
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/_multi", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_stream() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/streams", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_stream_schema() {
|
|
let auth = setup();
|
|
let one_sec = time::Duration::from_secs(2);
|
|
thread::sleep(one_sec);
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!(
|
|
"/api/{}/streams/{}/schema",
|
|
"e2e", "olympics_schema"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_stream_settings() {
|
|
let auth = setup();
|
|
let body_str = r#"{"partition_keys":{"add":[{"field":"test_key"}],"remove":[]}, "full_text_search_keys":{"add":["city"],"remove":[]}}"#;
|
|
// app
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!(
|
|
"/api/{}/streams/{}/settings",
|
|
"e2e", "olympics_schema"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_stream() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/streams/{}", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_org() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_function() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "e2etestfn",
|
|
"function":".sqNew,err = .Year*.Year \n .",
|
|
"params":"row"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/functions", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
if !resp.status().is_success() {
|
|
let status = resp.status();
|
|
let body = test::read_body(resp).await;
|
|
let body_str = String::from_utf8_lossy(&body);
|
|
println!("e2e_post_function response status: {status:?}");
|
|
println!("e2e_post_function response body: {body_str:?}");
|
|
|
|
// If function already exists, that's OK for our test
|
|
if status == actix_web::http::StatusCode::BAD_REQUEST
|
|
&& body_str.contains("Function already exist")
|
|
{
|
|
println!("Function already exists, continuing with test");
|
|
return;
|
|
}
|
|
|
|
panic!("e2e_post_function failed with status: {status:?}");
|
|
}
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_functions() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/functions", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_function() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/functions/{}", "e2e", "e2etestfn"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_search() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"query": {
|
|
"sql": "select * from olympics_schema",
|
|
"from": 0,
|
|
"size": 100,
|
|
"start_time": 1714857600000,
|
|
"end_time": 1714944000000
|
|
}
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/_search", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_search_around() {
|
|
let auth = setup();
|
|
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let ts = chrono::Utc::now().timestamp_micros();
|
|
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!(
|
|
"/api/{}/{}/_around?key={}&size=10",
|
|
"e2e", "olympics_schema", ts
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_users() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/users", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_organizations() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/api/organizations")
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_user_passcode() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/passcode", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_update_user_passcode() {
|
|
let auth = setup();
|
|
let body_str = "";
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!("/api/{}/passcode", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_user_authentication() {
|
|
let _auth = setup();
|
|
let body_str = r#"{
|
|
"name": "root@example.com",
|
|
"password": "Complexpass#123"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri("/auth/login")
|
|
.insert_header(ContentType::json())
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_user_authentication_with_error() {
|
|
let _auth = setup();
|
|
let body_str = r#"{
|
|
"name": "root2@example.com",
|
|
"password": "Complexpass#123"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri("/auth/login")
|
|
.insert_header(ContentType::json())
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(!resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_user() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"email": "nonadmin@example.com",
|
|
"password": "Abcd12345",
|
|
"role": "admin"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/users", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
println!("post user resp: {resp:?}");
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_update_user() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"email": "nonadmin@example.com",
|
|
"new_password": "12345678",
|
|
"change_password": true
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!("/api/{}/users/{}", "e2e", "nonadmin@example.com"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_update_user_with_empty() {
|
|
let auth = setup();
|
|
let body_str = r#"{}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!("/api/{}/users/{}", "e2e", "nonadmin@example.com"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(!resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_add_user_to_org() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"role":"member"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/users/{}", "e2e", "root@example.com"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_user() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/users/{}", "e2e", "nonadmin@example.com"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_create_dashboard() -> Dashboard {
|
|
let auth = setup();
|
|
let body_str = r##"{"title":"b2","dashboardId":"","description":"desc2","role":"","owner":"root@example.com","created":"2023-03-30T07:49:41.744+00:00","panels":[{"id":"Panel_ID7857010","type":"bar","fields":{"stream":"default","stream_type":"logs","x":[{"label":"Timestamp","alias":"x_axis_1","column":"_timestamp","color":null,"aggregationFunction":"histogram"}],"y":[{"label":"Kubernetes Host","alias":"y_axis_1","column":"kubernetes_host","color":"#5960b2","aggregationFunction":"count"}],"filter":[{"type":"condition","values":[],"column":"method","operator":"Is Not Null","value":null}]},"config":{"title":"p5","description":"sample config blah blah blah","show_legends":true},"query":"SELECT histogram(_timestamp) as \"x_axis_1\", count(kubernetes_host) as \"y_axis_1\" FROM \"default\" WHERE method IS NOT NULL GROUP BY \"x_axis_1\" ORDER BY \"x_axis_1\"","customQuery":false}],"layouts":[{"x":0,"y":0,"w":12,"h":13,"i":1,"panelId":"Panel_ID7857010","static":false}]}"##;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/dashboards", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
|
|
let body = test::call_and_read_body(&app, req).await;
|
|
json::from_slice(&body).unwrap()
|
|
}
|
|
|
|
async fn e2e_list_dashboards() -> Vec<Dashboard> {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/dashboards", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
|
|
// Try to parse the response body as a list of dashboards.
|
|
let body = test::call_and_read_body(&app, req).await;
|
|
let mut body_json: json::Value = json::from_slice(&body).unwrap();
|
|
let list_json = body_json
|
|
.as_object_mut()
|
|
.unwrap()
|
|
.remove("dashboards")
|
|
.unwrap();
|
|
let dashboards: Vec<Dashboard> = json::from_value(list_json).unwrap();
|
|
|
|
dashboards
|
|
}
|
|
|
|
async fn e2e_update_dashboard(dashboard: v1::Dashboard, hash: String) -> Dashboard {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!(
|
|
"/api/{}/dashboards/{}?hash={}",
|
|
"e2e", dashboard.dashboard_id, hash
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(json::to_string(&dashboard).unwrap())
|
|
.to_request();
|
|
|
|
let body = test::call_and_read_body(&app, req).await;
|
|
json::from_slice(&body).unwrap()
|
|
}
|
|
|
|
async fn e2e_get_dashboard(dashboard_id: &str) -> Dashboard {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/dashboards/{dashboard_id}", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
|
|
let body = test::call_and_read_body(&app, req).await;
|
|
json::from_slice(&body).unwrap()
|
|
}
|
|
|
|
async fn e2e_delete_dashboard(dashboard_id: &str) {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/dashboards/{dashboard_id}", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_trace() {
|
|
let auth = setup();
|
|
let path = "./tests/trace_input.json";
|
|
let body_str = fs::read_to_string(path).expect("Read file failed");
|
|
|
|
// app
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/traces", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_metrics() {
|
|
let auth = setup();
|
|
|
|
let loc_label: Vec<prometheus_rpc::Label> = vec![
|
|
prometheus_rpc::Label {
|
|
name: "__name__".to_string(),
|
|
value: "grafana_api_dashboard_save_milliseconds_count".to_string(),
|
|
},
|
|
prometheus_rpc::Label {
|
|
name: "cluster".to_string(),
|
|
value: "prom-k8s".to_string(),
|
|
},
|
|
prometheus_rpc::Label {
|
|
name: "__replica__".to_string(),
|
|
value: "prom-k8s-0".to_string(),
|
|
},
|
|
];
|
|
|
|
let mut loc_samples: Vec<prometheus_rpc::Sample> = vec![];
|
|
|
|
for i in 1..2 {
|
|
loc_samples.push(prometheus_rpc::Sample {
|
|
value: i as f64,
|
|
timestamp: Utc::now().timestamp_micros(),
|
|
});
|
|
}
|
|
loc_samples.push(prometheus_rpc::Sample {
|
|
value: f64::NEG_INFINITY,
|
|
timestamp: Utc::now().timestamp_micros(),
|
|
});
|
|
loc_samples.push(prometheus_rpc::Sample {
|
|
value: f64::INFINITY,
|
|
timestamp: Utc::now().timestamp_micros(),
|
|
});
|
|
|
|
loc_samples.push(prometheus_rpc::Sample {
|
|
value: f64::NAN,
|
|
timestamp: Utc::now().timestamp_micros(),
|
|
});
|
|
let loc_exemp: Vec<prometheus_rpc::Exemplar> = vec![];
|
|
let loc_hist: Vec<prometheus_rpc::Histogram> = vec![];
|
|
|
|
let ts = prometheus_rpc::TimeSeries {
|
|
labels: loc_label,
|
|
samples: loc_samples,
|
|
exemplars: loc_exemp,
|
|
histograms: loc_hist,
|
|
};
|
|
|
|
let metadata: Vec<prometheus_rpc::MetricMetadata> = vec![];
|
|
let wr_req: prometheus_rpc::WriteRequest = prometheus_rpc::WriteRequest {
|
|
timeseries: vec![ts],
|
|
metadata,
|
|
};
|
|
let mut out = BytesMut::with_capacity(wr_req.encoded_len());
|
|
wr_req.encode(&mut out).expect("Out of memory");
|
|
let data: Bytes = out.into();
|
|
let body = snap::raw::Encoder::new()
|
|
.compress_vec(&data)
|
|
.expect("Out of memory");
|
|
|
|
// app
|
|
let thread_id: usize = 0;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.app_data(web::Data::new(thread_id))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/prometheus/api/v1/write", "e2e"))
|
|
.insert_header(("X-Prometheus-Remote-Write-Version", "0.1.0"))
|
|
.insert_header(("Content-Encoding", "snappy"))
|
|
.insert_header(("Content-Type", "application/x-protobuf"))
|
|
.append_header(auth)
|
|
.set_payload(body)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_org_summary() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/summary", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
log::info!("{:?}", resp.status());
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_template() {
|
|
let auth = setup();
|
|
let body_str = r#"{"name":"slackTemplate","body":"{\"text\":\"For stream {stream_name} of organization {org_name} alert {alert_name} of type {alert_type} is active app_name {app_name}\"}"}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/templates", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_alert_template() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/templates/{}",
|
|
"e2e", "slackTemplate"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_alert_template() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/templates/{}",
|
|
"e2e", "slackTemplate"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_email_template() {
|
|
let auth = setup();
|
|
let body_str = r#"{"name":"email_template","body":"This is email for {alert_name}.","type":"email","title":"Email Subject"}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/templates", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_alert_email_template() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/templates/{}",
|
|
"e2e", "email_template"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_alert_email_template() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/templates/{}",
|
|
"e2e", "email_template"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_alert_template() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts/templates", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_destination() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "slack",
|
|
"url": "https://dummy/alert",
|
|
"method": "post",
|
|
"template": "slackTemplate",
|
|
"headers":{
|
|
"x_org_id":"Test_header"
|
|
}
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_alert_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts/destinations/{}", "e2e", "slack"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_alert_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/alerts/destinations/{}", "e2e", "slack"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
log::info!("{:?}", resp.status());
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_email_destination() {
|
|
let auth = setup();
|
|
let body_str = r#"{"url":"","method":"post","skip_tls_verify":false,"template":"email_template","headers":{},"name":"email","type":"email","emails":["nonadmin@example.com"]}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_email_destination_should_fail() {
|
|
let auth = setup();
|
|
let body_str = r#"{"url":"","method":"post","skip_tls_verify":false,"template":"email_template","headers":{},"name":"email_fail","type":"email","emails":["nonadmin2@example.com"]}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_client_error());
|
|
}
|
|
|
|
async fn e2e_get_alert_email_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts/destinations/{}", "e2e", "email"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_alert_email_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/{}/alerts/destinations/{}", "e2e", "email"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_alert_destinations() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_sns_alert_template() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "snsTemplate",
|
|
"body": "{\"default\": \"SNS alert {alert_name} triggered for {stream_name} in {org_name}\"}"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/templates", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_sns_alert_destination() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "sns_alert",
|
|
"type": "sns",
|
|
"sns_topic_arn": "arn:aws:sns:us-east-1:123456789012:MyTopic",
|
|
"aws_region": "us-east-1",
|
|
"template": "snsTemplate"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_get_sns_alert_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/destinations/{}",
|
|
"e2e", "sns_alert"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// Optionally, deserialize and check the response body
|
|
let body = test::read_body(resp).await;
|
|
let destination: Destination = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(destination.destination_type, DestinationType::Sns);
|
|
assert_eq!(
|
|
destination.sns_topic_arn,
|
|
Some("arn:aws:sns:us-east-1:123456789012:MyTopic".to_string())
|
|
);
|
|
assert_eq!(destination.aws_region, Some("us-east-1".to_string()));
|
|
}
|
|
|
|
async fn e2e_list_alert_destinations_with_sns() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts/destinations", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// Optionally, deserialize and check the response body
|
|
let body = test::read_body(resp).await;
|
|
let destinations: Vec<Destination> = serde_json::from_slice(&body).unwrap();
|
|
assert!(
|
|
destinations
|
|
.iter()
|
|
.any(|d| d.destination_type == DestinationType::Sns)
|
|
);
|
|
}
|
|
|
|
async fn e2e_update_sns_alert_destination() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "sns_alert",
|
|
"type": "sns",
|
|
"sns_topic_arn": "arn:aws:sns:us-west-2:123456789012:UpdatedTopic",
|
|
"aws_region": "us-west-2",
|
|
"template": "snsTemplate"
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::put()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/destinations/{}",
|
|
"e2e", "sns_alert"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_sns_alert_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!(
|
|
"/api/{}/alerts/destinations/{}",
|
|
"e2e", "sns_alert"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_with_sns_destination() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "sns_test_alert",
|
|
"stream_type": "logs",
|
|
"stream_name": "olympics_schema",
|
|
"is_real_time": false,
|
|
"query_condition": {
|
|
"conditions": [{
|
|
"column": "level",
|
|
"operator": "=",
|
|
"value": "error"
|
|
}]
|
|
},
|
|
"trigger_condition": {
|
|
"period": 5,
|
|
"threshold": 1,
|
|
"silence": 10
|
|
},
|
|
"destinations": ["sns_alert"],
|
|
"context_attributes": {
|
|
"app_name": "TestApp"
|
|
}
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/alerts", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_delete_alert_with_sns_destination() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!(
|
|
"/api/{}/{}/alerts/{}",
|
|
"e2e", "olympics_schema", "sns_test_alert"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_alert_multirange() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "alert_multi_range",
|
|
"stream_type": "logs",
|
|
"stream_name": "olympics_schema",
|
|
"is_real_time": false,
|
|
"query_condition": {
|
|
"conditions": [{
|
|
"column": "country",
|
|
"operator": "=",
|
|
"value": "USA"
|
|
}],
|
|
"multi_time_range": [{
|
|
"offSet": "1440m"
|
|
}]
|
|
},
|
|
"trigger_condition": {
|
|
"period": 5,
|
|
"threshold": 1,
|
|
"silence": 0,
|
|
"frequency": 1
|
|
},
|
|
"destinations": ["slack"],
|
|
"context_attributes":{
|
|
"app_name":"App1"
|
|
}
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/{}/alerts", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
// Get the alert with the same stream name
|
|
let alert = openobserve::service::db::alerts::alert::get_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"alert_multi_range",
|
|
)
|
|
.await;
|
|
assert!(alert.is_ok());
|
|
let alert = alert.unwrap();
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
assert_eq!(alert.stream_type, config::meta::stream::StreamType::Logs);
|
|
assert_eq!(alert.stream_name, "olympics_schema");
|
|
assert_eq!(alert.name, "alert_multi_range");
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
// Check the trigger
|
|
let trigger = openobserve::service::db::scheduler::exists(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(trigger);
|
|
}
|
|
|
|
async fn e2e_delete_alert_multirange() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
|
|
// Get the alert with the same stream name
|
|
let alert = openobserve::service::db::alerts::alert::get_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"alert_multi_range",
|
|
)
|
|
.await;
|
|
assert!(alert.is_ok());
|
|
let alert = alert.unwrap();
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
|
|
// Use the v2 api to delete the alert
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!("/api/v2/{}/alerts/{}", "e2e", id))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
|
|
let trigger = openobserve::service::db::scheduler::exists(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(!trigger);
|
|
}
|
|
|
|
async fn e2e_post_alert() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"name": "alertChk",
|
|
"stream_type": "logs",
|
|
"stream_name": "olympics_schema",
|
|
"is_real_time": false,
|
|
"enabled": true,
|
|
"query_condition": {
|
|
"conditions": [{
|
|
"column": "country",
|
|
"operator": "NotContains",
|
|
"value": "AUT"
|
|
}]
|
|
},
|
|
"trigger_condition": {
|
|
"period": 60,
|
|
"threshold": 1,
|
|
"silence": 0,
|
|
"frequency": 60,
|
|
"operator": ">="
|
|
},
|
|
"destinations": ["slack"],
|
|
"context_attributes":{
|
|
"app_name":"App1"
|
|
}
|
|
}"#;
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/v2/{}/alerts", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
println!("{:?}", resp.status());
|
|
println!("{:?}", resp.response().body());
|
|
assert!(resp.status().is_success());
|
|
|
|
// Get the alert list
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/v2/{}/alerts", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let alert_list_response: ListAlertsResponseBody = serde_json::from_slice(&body).unwrap();
|
|
assert!(!alert_list_response.list.is_empty());
|
|
let alert = alert_list_response
|
|
.list
|
|
.iter()
|
|
.find(|a| a.name == "alertChk");
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
assert_eq!(alert.name, "alertChk");
|
|
assert!(alert.enabled);
|
|
let id = alert.alert_id;
|
|
let id = id.to_string();
|
|
|
|
// Check the trigger
|
|
let trigger = openobserve::service::db::scheduler::exists(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(trigger);
|
|
}
|
|
|
|
async fn e2e_get_alert() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
|
|
// Get the alert list
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/v2/{}/alerts", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let alert_list_response: ListAlertsResponseBody = serde_json::from_slice(&body).unwrap();
|
|
assert!(!alert_list_response.list.is_empty());
|
|
let alert = alert_list_response
|
|
.list
|
|
.iter()
|
|
.find(|a| a.name == "alertChk");
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
assert_eq!(alert.name, "alertChk");
|
|
assert!(alert.enabled);
|
|
let id = alert.alert_id;
|
|
let id = id.to_string();
|
|
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/v2/{}/alerts/{}", "e2e", id))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
log::info!("{:?}", resp.status());
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let alert_response: GetAlertResponseBody = serde_json::from_slice(&body).unwrap();
|
|
assert_eq!(alert_response.0.name, "alertChk");
|
|
assert_eq!(
|
|
alert_response.0.stream_type,
|
|
openobserve::handler::http::models::alerts::StreamType::Logs
|
|
);
|
|
assert_eq!(alert_response.0.stream_name, "olympics_schema");
|
|
assert!(alert_response.0.enabled);
|
|
}
|
|
|
|
async fn e2e_handle_alert_after_destination_retries() {
|
|
let alert = openobserve::service::db::alerts::alert::get_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"alertChk",
|
|
)
|
|
.await;
|
|
assert!(alert.is_ok());
|
|
let alert = alert.unwrap();
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_3_later = now
|
|
+ Duration::try_minutes(3)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let trigger = Trigger {
|
|
id: 1,
|
|
org: "e2e".to_string(),
|
|
module: config::meta::triggers::TriggerModule::Alert,
|
|
module_key: id.to_string(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_3_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 2,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// This alert has an invalid destination
|
|
assert!(res.is_ok());
|
|
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
assert!(trigger.next_run_at > now && trigger.retries == 0);
|
|
}
|
|
|
|
async fn e2e_handle_alert_reached_max_retries() {
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_3_later = now
|
|
+ Duration::try_minutes(3)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let alert = openobserve::service::db::alerts::alert::get_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"alertChk",
|
|
)
|
|
.await;
|
|
assert!(alert.is_ok());
|
|
let alert = alert.unwrap();
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
let trigger = Trigger {
|
|
id: 1,
|
|
org: "e2e".to_string(),
|
|
module: config::meta::triggers::TriggerModule::Alert,
|
|
module_key: id.to_string(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_3_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 3,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// This alert has an invalid destination
|
|
assert!(res.is_ok());
|
|
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
assert!(trigger.next_run_at > now && trigger.retries == 0);
|
|
}
|
|
|
|
async fn e2e_handle_alert_after_evaluation_retries() {
|
|
let mut alert: Alert = Default::default();
|
|
alert.name = "test_alert_wrong_sql".to_string();
|
|
alert.stream_type = "logs".into();
|
|
alert.stream_name = "olympics_schema".to_string();
|
|
alert.is_real_time = false;
|
|
alert.enabled = true;
|
|
alert.query_condition = QueryCondition {
|
|
query_type: "sql".into(),
|
|
conditions: None,
|
|
sql: Some("SELEC country FROM \"olympics_schema\"".to_string()),
|
|
..Default::default()
|
|
};
|
|
alert.trigger_condition = TriggerCondition {
|
|
period: 60,
|
|
threshold: 1,
|
|
silence: 0,
|
|
frequency: 3600,
|
|
operator: Operator::GreaterThanEquals,
|
|
..Default::default()
|
|
};
|
|
alert.destinations = vec!["slack".to_string()];
|
|
|
|
let res = openobserve::service::db::alerts::alert::set("e2e", alert, true).await;
|
|
assert!(res.is_ok());
|
|
let alert = res.unwrap();
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_3_later = now
|
|
+ Duration::try_minutes(3)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let trigger = Trigger {
|
|
id: 1,
|
|
org: "e2e".to_string(),
|
|
module: config::meta::triggers::TriggerModule::Alert,
|
|
module_key: id.to_string(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_3_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 2,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// In case of alert evaluation errors, this error is returned
|
|
assert!(res.is_err());
|
|
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
assert!(trigger.next_run_at > now && trigger.retries == 0);
|
|
|
|
let res = openobserve::service::db::alerts::alert::delete_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"test_alert_wrong_sql",
|
|
)
|
|
.await;
|
|
assert!(res.is_ok());
|
|
}
|
|
|
|
async fn e2e_delete_alert() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
|
|
let alert = openobserve::service::db::alerts::alert::get_by_name(
|
|
"e2e",
|
|
config::meta::stream::StreamType::Logs,
|
|
"olympics_schema",
|
|
"alertChk",
|
|
)
|
|
.await;
|
|
assert!(alert.is_ok());
|
|
let alert = alert.unwrap();
|
|
assert!(alert.is_some());
|
|
let alert = alert.unwrap();
|
|
let id = alert.id;
|
|
assert!(id.is_some());
|
|
let id = id.unwrap();
|
|
|
|
let req = test::TestRequest::delete()
|
|
.uri(&format!(
|
|
"/api/{}/{}/alerts/{}",
|
|
"e2e", "olympics_schema", "alertChk"
|
|
))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
log::info!("{:?}", resp.status());
|
|
assert!(resp.status().is_success());
|
|
|
|
let trigger = openobserve::service::db::scheduler::exists(
|
|
"e2e",
|
|
config::meta::triggers::TriggerModule::Alert,
|
|
&id.to_string(),
|
|
)
|
|
.await;
|
|
assert!(!trigger);
|
|
}
|
|
|
|
async fn e2e_list_alerts() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/alerts", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_real_time_alerts() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/{}/alerts", "e2e", "olympics_schema"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_health_check() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/healthz")
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_config() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_config_routes)
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/config")
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_post_syslog_route() {
|
|
let auth = setup();
|
|
let body_str = r#"{
|
|
"orgId": "acceptLogs",
|
|
"streamName":"syslog",
|
|
"subnets": [
|
|
"192.168.0.0/16",
|
|
"127.0.0.0/8",
|
|
"172.16.0.0/12"
|
|
]
|
|
}"#;
|
|
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/syslog-routes", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
async fn e2e_list_syslog_routes() {
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes)
|
|
.configure(get_basic_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri(&format!("/api/{}/syslog-routes", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
}
|
|
|
|
// Helper function to create pipeline via API
|
|
async fn e2e_post_pipeline(pipeline_data: Pipeline) {
|
|
let auth = setup();
|
|
let body_str = serde_json::to_string(&pipeline_data).unwrap();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::post()
|
|
.uri(&format!("/api/{}/pipelines", "e2e"))
|
|
.insert_header(ContentType::json())
|
|
.append_header(auth)
|
|
.set_payload(body_str)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
if !resp.status().is_success() {
|
|
let body = test::read_body(resp).await;
|
|
println!("Response body: {}", String::from_utf8_lossy(&body));
|
|
panic!("Failed to create pipeline");
|
|
}
|
|
}
|
|
|
|
// Derived Stream Integration Tests
|
|
async fn e2e_create_test_pipeline() {
|
|
// Create a test pipeline with derived stream for testing
|
|
let pipeline_data = Pipeline {
|
|
id: "test_derived_stream_pipeline".to_string(),
|
|
version: 1,
|
|
enabled: true,
|
|
org: "e2e".to_string(),
|
|
name: "test_derived_stream".to_string(),
|
|
description: "Test pipeline for derived stream integration tests".to_string(),
|
|
source: PipelineSource::Scheduled(DerivedStream {
|
|
org_id: "e2e".to_string(),
|
|
stream_type: StreamType::Logs,
|
|
query_condition: QueryCondition {
|
|
query_type: config::meta::alerts::QueryType::SQL,
|
|
sql: Some("SELECT _timestamp FROM \"olympics_schema\"".to_string()),
|
|
..Default::default()
|
|
},
|
|
trigger_condition: TriggerCondition {
|
|
period: 5, // 5 minutes
|
|
frequency: 300, // 5 minutes in seconds
|
|
..Default::default()
|
|
},
|
|
tz_offset: 0,
|
|
start_at: None,
|
|
delay: None,
|
|
}),
|
|
nodes: vec![
|
|
// Source node (query node for scheduled pipeline)
|
|
config::meta::pipeline::components::Node::new(
|
|
"source-node-1".to_string(),
|
|
config::meta::pipeline::components::NodeData::Query(DerivedStream {
|
|
org_id: "e2e".to_string(),
|
|
stream_type: StreamType::Logs,
|
|
query_condition: QueryCondition {
|
|
query_type: config::meta::alerts::QueryType::SQL,
|
|
sql: Some("SELECT _timestamp FROM \"olympics_schema\"".to_string()),
|
|
..Default::default()
|
|
},
|
|
trigger_condition: TriggerCondition {
|
|
period: 5,
|
|
frequency: 300,
|
|
..Default::default()
|
|
},
|
|
tz_offset: 0,
|
|
start_at: None,
|
|
delay: None,
|
|
}),
|
|
100.0,
|
|
50.0,
|
|
"input".to_string(),
|
|
),
|
|
// Destination node (output stream)
|
|
config::meta::pipeline::components::Node::new(
|
|
"dest-node-1".to_string(),
|
|
config::meta::pipeline::components::NodeData::Stream(
|
|
config::meta::stream::StreamParams {
|
|
org_id: "e2e".to_string().into(),
|
|
stream_name: "test_derived_output_stream".to_string().into(),
|
|
stream_type: StreamType::Logs,
|
|
},
|
|
),
|
|
100.0,
|
|
200.0,
|
|
"output".to_string(),
|
|
),
|
|
],
|
|
edges: vec![config::meta::pipeline::components::Edge::new(
|
|
"source-node-1".to_string(),
|
|
"dest-node-1".to_string(),
|
|
)],
|
|
};
|
|
|
|
// Save pipeline using API call
|
|
e2e_post_pipeline(pipeline_data).await;
|
|
}
|
|
|
|
async fn e2e_handle_derived_stream_success() {
|
|
// list the pipelines and choose the first one using API
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/api/e2e/pipelines")
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let pipeline_list: openobserve::handler::http::models::pipelines::PipelineList =
|
|
json::from_slice(&body).unwrap();
|
|
let pipeline = pipeline_list.list.first();
|
|
assert!(pipeline.is_some());
|
|
let pipeline = pipeline.unwrap();
|
|
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_5_later = now
|
|
+ Duration::try_minutes(5)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let module_key = format!("logs/e2e/test_derived_stream/{}", pipeline.id);
|
|
|
|
let trigger = Trigger {
|
|
id: 1,
|
|
org: "e2e".to_string(),
|
|
module: TriggerModule::DerivedStream,
|
|
module_key: module_key.clone(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_5_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 0,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_derived_stream_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// Should succeed even with empty data
|
|
assert!(res.is_ok());
|
|
|
|
// Verify trigger was updated
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
TriggerModule::DerivedStream,
|
|
&module_key,
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
let scheduled_trigger_data: ScheduledTriggerData =
|
|
serde_json::from_str(&trigger.data).unwrap();
|
|
assert!(scheduled_trigger_data.period_end_time.is_some());
|
|
assert!(scheduled_trigger_data.period_end_time.unwrap() > 0);
|
|
assert!(trigger.status == TriggerStatus::Waiting);
|
|
assert!(trigger.next_run_at > now && trigger.retries == 0);
|
|
}
|
|
|
|
async fn e2e_handle_derived_stream_pipeline_not_found() {
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_5_later = now
|
|
+ Duration::try_minutes(5)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let module_key = "logs/e2e/nonexistent_pipeline/invalid_id".to_string();
|
|
|
|
let trigger = Trigger {
|
|
id: 2,
|
|
org: "e2e".to_string(),
|
|
module: TriggerModule::DerivedStream,
|
|
module_key,
|
|
start_time: Some(now),
|
|
end_time: Some(mins_5_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 0,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_derived_stream_not_found_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// Should fail with pipeline not found error
|
|
assert!(res.is_err());
|
|
assert!(
|
|
res.unwrap_err()
|
|
.to_string()
|
|
.contains("Pipeline associated with trigger not found")
|
|
);
|
|
}
|
|
|
|
async fn e2e_handle_derived_stream_max_retries() {
|
|
// list pipelines using API
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/api/e2e/pipelines")
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let pipeline_list: openobserve::handler::http::models::pipelines::PipelineList =
|
|
json::from_slice(&body).unwrap();
|
|
let pipelines = pipeline_list.list.first();
|
|
assert!(pipelines.is_some());
|
|
let pipeline = pipelines.unwrap();
|
|
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_5_later = now
|
|
+ Duration::try_minutes(5)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let module_key = format!("logs/e2e/test_derived_stream/{}", pipeline.id);
|
|
|
|
let trigger = Trigger {
|
|
id: 3,
|
|
org: "e2e".to_string(),
|
|
module: TriggerModule::DerivedStream,
|
|
module_key: module_key.clone(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_5_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 5, // Max retries reached
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_derived_stream_max_retries_trace_id";
|
|
let res = handle_triggers(trace_id, trigger).await;
|
|
// Should succeed but skip to next run due to max retries
|
|
assert!(res.is_ok());
|
|
|
|
// Verify trigger was updated with next run time and retries reset
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
TriggerModule::DerivedStream,
|
|
&module_key,
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
assert!(trigger.next_run_at > now && trigger.retries == 0);
|
|
}
|
|
|
|
async fn e2e_handle_derived_stream_evaluation_failure() {
|
|
let auth = setup();
|
|
// Create a pipeline with invalid SQL to cause evaluation failure
|
|
let pipeline_data = Pipeline {
|
|
id: "test_derived_stream_pipeline_invalid".to_string(),
|
|
version: 1,
|
|
enabled: true,
|
|
org: "e2e".to_string(),
|
|
name: "test_derived_stream_invalid".to_string(),
|
|
description: "Test pipeline with invalid SQL".to_string(),
|
|
source: PipelineSource::Scheduled(DerivedStream {
|
|
org_id: "e2e".to_string(),
|
|
stream_type: StreamType::Logs,
|
|
query_condition: QueryCondition {
|
|
query_type: config::meta::alerts::QueryType::SQL,
|
|
sql: Some("SELECT _timestamp, city FROM \"olympics_schema\"".to_string()), /* Invalid
|
|
* SQL */
|
|
..Default::default()
|
|
},
|
|
trigger_condition: TriggerCondition {
|
|
period: 5,
|
|
frequency: 300,
|
|
..Default::default()
|
|
},
|
|
tz_offset: 0,
|
|
start_at: None,
|
|
delay: None,
|
|
}),
|
|
nodes: vec![
|
|
// Source node (query node for scheduled pipeline with invalid SQL)
|
|
config::meta::pipeline::components::Node::new(
|
|
"source-node-2".to_string(),
|
|
config::meta::pipeline::components::NodeData::Query(DerivedStream {
|
|
org_id: "e2e".to_string(),
|
|
stream_type: StreamType::Logs,
|
|
query_condition: QueryCondition {
|
|
query_type: config::meta::alerts::QueryType::SQL,
|
|
sql: Some(
|
|
"SELECT _timestamp, city FROM \"olympics_schema\"".to_string(),
|
|
),
|
|
..Default::default()
|
|
},
|
|
trigger_condition: TriggerCondition {
|
|
period: 5,
|
|
frequency: 300,
|
|
..Default::default()
|
|
},
|
|
tz_offset: 0,
|
|
start_at: None,
|
|
delay: None,
|
|
}),
|
|
150.0,
|
|
50.0,
|
|
"input".to_string(),
|
|
),
|
|
// Destination node (output stream)
|
|
config::meta::pipeline::components::Node::new(
|
|
"dest-node-2".to_string(),
|
|
config::meta::pipeline::components::NodeData::Stream(
|
|
config::meta::stream::StreamParams {
|
|
org_id: "e2e".to_string().into(),
|
|
stream_name: "test_invalid_pipeline_output".to_string().into(),
|
|
stream_type: StreamType::Logs,
|
|
},
|
|
),
|
|
150.0,
|
|
200.0,
|
|
"output".to_string(),
|
|
),
|
|
],
|
|
edges: vec![config::meta::pipeline::components::Edge::new(
|
|
"source-node-2".to_string(),
|
|
"dest-node-2".to_string(),
|
|
)],
|
|
};
|
|
|
|
// Save pipeline using API call
|
|
e2e_post_pipeline(pipeline_data.clone()).await;
|
|
|
|
// Check if pipeline was saved successfully by doing a list using API
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes),
|
|
)
|
|
.await;
|
|
|
|
// delete the city field from the stream
|
|
// Make a PUT call to `e2e/streams/olympics_schema/delete_fields` api
|
|
// with `{"fields":["method"]}` as the payload
|
|
let delete_fields_url = "/api/e2e/streams/olympics_schema/delete_fields";
|
|
let delete_fields_payload = serde_json::json!({"fields": ["city"]});
|
|
|
|
let req = test::TestRequest::put()
|
|
.uri(delete_fields_url)
|
|
.append_header(auth)
|
|
.set_json(&delete_fields_payload)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success(), "Failed to delete fields");
|
|
|
|
let req = test::TestRequest::get()
|
|
.uri("/api/e2e/pipelines")
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success(), "Failed to list pipelines");
|
|
let body = test::read_body(resp).await;
|
|
let pipeline_response: openobserve::handler::http::models::pipelines::PipelineList =
|
|
json::from_slice(&body).unwrap();
|
|
// Get the pipeline that matches the pipeline name
|
|
let pipeline = pipeline_response
|
|
.list
|
|
.iter()
|
|
.find(|p| p.name == pipeline_data.name);
|
|
assert!(pipeline.is_some(), "Pipeline not found");
|
|
let pipeline = pipeline.unwrap();
|
|
|
|
let now = Utc::now().timestamp_micros();
|
|
let mins_5_later = now
|
|
+ Duration::try_minutes(5)
|
|
.unwrap()
|
|
.num_microseconds()
|
|
.unwrap();
|
|
let module_key = format!("logs/e2e/test_derived_stream_invalid/{}", pipeline.id);
|
|
|
|
let trigger = Trigger {
|
|
id: 4,
|
|
org: "e2e".to_string(),
|
|
module: TriggerModule::DerivedStream,
|
|
module_key: module_key.clone(),
|
|
start_time: Some(now),
|
|
end_time: Some(mins_5_later),
|
|
next_run_at: now,
|
|
is_realtime: false,
|
|
is_silenced: false,
|
|
status: config::meta::triggers::TriggerStatus::Processing,
|
|
retries: 0,
|
|
data: "{}".to_string(),
|
|
};
|
|
|
|
let trace_id = "test_derived_stream_eval_failure_trace_id";
|
|
let _ = handle_triggers(trace_id, trigger).await;
|
|
// Should succeed (handler handles errors gracefully) but increment retries
|
|
// Verify trigger retries were incremented
|
|
let trigger = openobserve::service::db::scheduler::get(
|
|
"e2e",
|
|
TriggerModule::DerivedStream,
|
|
&module_key,
|
|
)
|
|
.await;
|
|
assert!(trigger.is_ok());
|
|
let trigger = trigger.unwrap();
|
|
assert!(trigger.retries > 0);
|
|
|
|
// Clean up the invalid pipeline
|
|
let _ = openobserve::service::db::pipeline::delete(&pipeline.id).await;
|
|
}
|
|
|
|
async fn e2e_cleanup_test_pipeline() {
|
|
// list the pipelines and choose the first one using API
|
|
let auth = setup();
|
|
let app = test::init_service(
|
|
App::new()
|
|
.app_data(web::JsonConfig::default().limit(get_config().limit.req_json_limit))
|
|
.app_data(web::PayloadConfig::new(
|
|
get_config().limit.req_payload_limit,
|
|
))
|
|
.configure(get_service_routes),
|
|
)
|
|
.await;
|
|
let req = test::TestRequest::get()
|
|
.uri("/api/e2e/pipelines")
|
|
.append_header(auth)
|
|
.to_request();
|
|
let resp = test::call_service(&app, req).await;
|
|
assert!(resp.status().is_success());
|
|
let body = test::read_body(resp).await;
|
|
let pipeline_list: openobserve::handler::http::models::pipelines::PipelineList =
|
|
json::from_slice(&body).unwrap();
|
|
let pipeline = pipeline_list.list.first();
|
|
assert!(pipeline.is_some());
|
|
let pipeline = pipeline.unwrap();
|
|
|
|
// Clean up test pipelines
|
|
let _ = openobserve::service::db::pipeline::delete(&pipeline.id).await;
|
|
}
|
|
}
|