starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 00:09:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: removed index on original_url for short_url service (#4720)

Browse Source
This commit is contained in:
Loakesh Indiran
2024-10-07 20:48:39 +08:00
committed by GitHub
parent f69593ba26
commit 290daf4fe5
8 changed files with 105 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/handler/http/router/mod.rs
+3 -4
View File
@@ -226,10 +226,9 @@ pub fn get_basic_routes(cfg: &mut web::ServiceConfig) {
.service(status::flush_node),
);
cfg.service(
web::scope("/short")
.service(short_url::retrieve),
);
cfg.service(web::scope("/short")
.wrap(cors.clone())
.service(short_url::retrieve));
if get_config().common.swagger_enabled {
cfg.service(
src/infra/src/scheduler/mysql.rs
+51 -17
View File
@@ -47,7 +47,9 @@ impl super::Scheduler for MySqlScheduler {
/// Creates the Scheduled Jobs table
async fn create_table(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["create", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["create", "scheduled_jobs"])
.inc();
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS scheduled_jobs
@@ -72,7 +74,9 @@ CREATE TABLE IF NOT EXISTS scheduled_jobs
.await?;
// create data column for old version <= 0.10.9
DB_QUERY_NUMS.with_label_values(&["select", "information_schema.columns"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "information_schema.columns"])
.inc();
let has_data_column = sqlx::query_scalar::<_,i64>("SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='scheduled_jobs' AND column_name='data';")
.fetch_one(&pool)
.await?;
@@ -111,7 +115,9 @@ CREATE TABLE IF NOT EXISTS scheduled_jobs
/// The count of jobs for the given module (Report/Alert etc.)
async fn len_module(&self, module: TriggerModule) -> usize {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let ret = match sqlx::query(
r#"
SELECT CAST(COUNT(*) AS SIGNED) AS num FROM scheduled_jobs WHERE module = ?;"#,
@@ -137,7 +143,9 @@ SELECT CAST(COUNT(*) AS SIGNED) AS num FROM scheduled_jobs WHERE module = ?;"#,
let pool = CLIENT.clone();
let mut tx = pool.begin().await?;
DB_QUERY_NUMS.with_label_values(&["insert", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["insert", "scheduled_jobs"])
.inc();
if let Err(e) = sqlx::query(
r#"
INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, status, retries, next_run_at, start_time, end_time, data)
@@ -186,7 +194,9 @@ INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_sile
/// Deletes the Trigger job matching the given parameters
async fn delete(&self, org: &str, module: TriggerModule, key: &str) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
sqlx::query(
r#"DELETE FROM scheduled_jobs WHERE org = ? AND module = ? AND module_key = ?;"#,
)
@@ -218,7 +228,9 @@ INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_sile
retries: i32,
) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
sqlx::query(
r#"UPDATE scheduled_jobs SET status = ?, retries = ? WHERE org = ? AND module_key = ? AND module = ?;"#
)
@@ -237,7 +249,9 @@ INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_sile
async fn update_trigger(&self, trigger: Trigger) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
sqlx::query(
r#"UPDATE scheduled_jobs
SET status = ?, retries = ?, next_run_at = ?, is_realtime = ?, is_silenced = ?, data = ?
@@ -298,7 +312,9 @@ WHERE org = ? AND module_key = ? AND module = ?;"#,
.num_microseconds()
.unwrap();
let mut tx = pool.begin().await?;
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let job_ids: Vec<TriggerId> = match sqlx::query_as::<_, TriggerId>(
r#"SELECT id
FROM scheduled_jobs
@@ -339,7 +355,9 @@ FOR UPDATE;
}
let job_ids: Vec<String> = job_ids.into_iter().map(|id| id.id.to_string()).collect();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
let query = format!(
"UPDATE scheduled_jobs
SET status = ?, start_time = ?,
@@ -376,7 +394,9 @@ WHERE id IN ({});",
job_ids.join(",")
);
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let jobs: Vec<Trigger> = sqlx::query_as::<_, Trigger>(query.as_str())
.fetch_all(&pool)
.await?;
@@ -386,7 +406,9 @@ WHERE id IN ({});",
async fn get(&self, org: &str, module: TriggerModule, key: &str) -> Result<Trigger> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let query =
r#"SELECT * FROM scheduled_jobs WHERE org = ? AND module = ? AND module_key = ?;"#;
let job = match sqlx::query_as::<_, Trigger>(query)
@@ -409,7 +431,9 @@ WHERE id IN ({});",
async fn list(&self, module: Option<TriggerModule>) -> Result<Vec<Trigger>> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let jobs: Vec<Trigger> = if let Some(module) = module {
let query = r#"SELECT * FROM scheduled_jobs WHERE module = ? ORDER BY id;"#;
sqlx::query_as::<_, Trigger>(query)
@@ -427,7 +451,9 @@ WHERE id IN ({});",
/// retries >= threshold set through environment
async fn clean_complete(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
sqlx::query(r#"DELETE FROM scheduled_jobs WHERE status = ? OR retries >= ?;"#)
.bind(TriggerStatus::Completed)
.bind(config::get_config().limit.scheduler_max_retries)
@@ -445,7 +471,9 @@ WHERE id IN ({});",
async fn watch_timeout(&self) -> Result<()> {
let pool = CLIENT.clone();
let now = chrono::Utc::now().timestamp_micros();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
sqlx::query(
r#"UPDATE scheduled_jobs
SET status = ?, retries = retries + 1
@@ -462,7 +490,9 @@ WHERE status = ? AND end_time <= ?
async fn len(&self) -> usize {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let ret = match sqlx::query(
r#"
SELECT CAST(COUNT(*) AS SIGNED) AS num FROM scheduled_jobs;"#,
@@ -488,7 +518,9 @@ SELECT CAST(COUNT(*) AS SIGNED) AS num FROM scheduled_jobs;"#,
async fn clear(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
match sqlx::query(r#"DELETE FROM scheduled_jobs;"#)
.execute(&pool)
.await
@@ -504,7 +536,9 @@ SELECT CAST(COUNT(*) AS SIGNED) AS num FROM scheduled_jobs;"#,
async fn add_data_column() -> Result<()> {
log::info!("[MYSQL] Adding data column to scheduled_jobs table");
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["alter", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["alter", "scheduled_jobs"])
.inc();
if let Err(e) = sqlx::query(r#"ALTER TABLE scheduled_jobs ADD COLUMN data LONGTEXT NOT NULL;"#)
.execute(&pool)
.await
src/infra/src/scheduler/postgres.rs
+39 -13
View File
@@ -115,7 +115,9 @@ CREATE TABLE IF NOT EXISTS scheduled_jobs
/// The count of jobs for the given module (Report/Alert etc.)
async fn len_module(&self, module: TriggerModule) -> usize {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let ret = match sqlx::query(
r#"
SELECT COUNT(*)::BIGINT AS num FROM scheduled_jobs WHERE module = $1;"#,
@@ -141,7 +143,9 @@ SELECT COUNT(*)::BIGINT AS num FROM scheduled_jobs WHERE module = $1;"#,
// let db = db::get_db().await;
let pool = CLIENT.clone();
let mut tx = pool.begin().await?;
DB_QUERY_NUMS.with_label_values(&["insert", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["insert", "scheduled_jobs"])
.inc();
if let Err(e) = sqlx::query(
r#"
INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, status, retries, next_run_at, start_time, end_time, data)
@@ -191,7 +195,9 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s
/// Deletes the Trigger job matching the given parameters
async fn delete(&self, org: &str, module: TriggerModule, key: &str) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
sqlx::query(
r#"DELETE FROM scheduled_jobs WHERE org = $1 AND module_key = $2 AND module = $3;"#,
)
@@ -223,7 +229,9 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s
retries: i32,
) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
sqlx::query(
r#"UPDATE scheduled_jobs SET status = $1, retries = $2 WHERE org = $3 AND module_key = $4 AND module = $5;"#
)
@@ -242,7 +250,9 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s
async fn update_trigger(&self, trigger: Trigger) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
sqlx::query(
r#"UPDATE scheduled_jobs
SET status = $1, retries = $2, next_run_at = $3, is_realtime = $4, is_silenced = $5, data = $6
@@ -301,7 +311,9 @@ WHERE org = $7 AND module_key = $8 AND module = $9;"#,
.unwrap()
.num_microseconds()
.unwrap();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
let query = r#"UPDATE scheduled_jobs
SET status = $1, start_time = $2,
end_time = CASE
@@ -352,7 +364,9 @@ RETURNING *;"#;
async fn get(&self, org: &str, module: TriggerModule, key: &str) -> Result<Trigger> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let query = r#"
SELECT * FROM scheduled_jobs
WHERE org = $1 AND module = $2 AND module_key = $3;"#;
@@ -376,7 +390,9 @@ WHERE org = $1 AND module = $2 AND module_key = $3;"#;
async fn list(&self, module: Option<TriggerModule>) -> Result<Vec<Trigger>> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let jobs: Vec<Trigger> = if let Some(module) = module {
let query = r#"SELECT * FROM scheduled_jobs WHERE module = $1 ORDER BY id;"#;
sqlx::query_as::<_, Trigger>(query)
@@ -394,7 +410,9 @@ WHERE org = $1 AND module = $2 AND module_key = $3;"#;
/// retries >= threshold set through environment
async fn clean_complete(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
sqlx::query(r#"DELETE FROM scheduled_jobs WHERE status = $1 OR retries >= $2;"#)
.bind(TriggerStatus::Completed)
.bind(config::get_config().limit.scheduler_max_retries)
@@ -411,7 +429,9 @@ WHERE org = $1 AND module = $2 AND module_key = $3;"#;
/// - Update their status back to "Waiting" and increase their "retries" by 1
async fn watch_timeout(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["update", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["update", "scheduled_jobs"])
.inc();
let now = chrono::Utc::now().timestamp_micros();
sqlx::query(
r#"UPDATE scheduled_jobs
@@ -429,7 +449,9 @@ WHERE status = $2 AND end_time <= $3;
async fn len(&self) -> usize {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["select", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["select", "scheduled_jobs"])
.inc();
let ret = match sqlx::query(
r#"
SELECT COUNT(*)::BIGINT AS num FROM scheduled_jobs;"#,
@@ -455,7 +477,9 @@ SELECT COUNT(*)::BIGINT AS num FROM scheduled_jobs;"#,
async fn clear(&self) -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["delete", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["delete", "scheduled_jobs"])
.inc();
match sqlx::query(r#"DELETE FROM scheduled_jobs;"#)
.execute(&pool)
.await
@@ -471,7 +495,9 @@ SELECT COUNT(*)::BIGINT AS num FROM scheduled_jobs;"#,
async fn add_data_column() -> Result<()> {
log::info!("[POSTGRES] Adding data column to scheduled_jobs table");
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["alter", "scheduled_jobs"]).inc();
DB_QUERY_NUMS
.with_label_values(&["alter", "scheduled_jobs"])
.inc();
if let Err(e) = sqlx::query(
r#"ALTER TABLE scheduled_jobs ADD COLUMN IF NOT EXISTS data TEXT NOT NULL DEFAULT '';"#,
)
src/infra/src/schema/history/mysql.rs
+6 -2
View File
@@ -56,7 +56,9 @@ impl super::SchemaHistory for MysqlSchemaHistory {
) -> Result<()> {
let value = json::to_string(&schema)?;
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["insert", "schema_history"]).inc();
DB_QUERY_NUMS
.with_label_values(&["insert", "schema_history"])
.inc();
match sqlx::query(
r#"
INSERT IGNORE INTO schema_history (org, stream_type, stream_name, start_dt, value)
@@ -86,7 +88,9 @@ INSERT IGNORE INTO schema_history (org, stream_type, stream_name, start_dt, valu
pub async fn create_table() -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["create", "schema_history"]).inc();
DB_QUERY_NUMS
.with_label_values(&["create", "schema_history"])
.inc();
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS schema_history
src/infra/src/schema/history/postgres.rs
+6 -2
View File
@@ -56,7 +56,9 @@ impl super::SchemaHistory for PostgresSchemaHistory {
) -> Result<()> {
let value = json::to_string(&schema)?;
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["insert", "schema_history"]).inc();
DB_QUERY_NUMS
.with_label_values(&["insert", "schema_history"])
.inc();
match sqlx::query(
r#"
INSERT INTO schema_history (org, stream_type, stream_name, start_dt, value)
@@ -87,7 +89,9 @@ INSERT INTO schema_history (org, stream_type, stream_name, start_dt, value)
pub async fn create_table() -> Result<()> {
let pool = CLIENT.clone();
DB_QUERY_NUMS.with_label_values(&["create", "schema_history"]).inc();
DB_QUERY_NUMS
.with_label_values(&["create", "schema_history"])
.inc();
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS schema_history
src/infra/src/short_url/mysql.rs
-7
View File
@@ -57,13 +57,6 @@ impl ShortUrl for MysqlShortUrl {
/// Create index for short_urls at short_id and original_url
async fn create_table_index(&self) -> Result<()> {
create_index("short_urls_short_id_idx", "short_urls", true, &["short_id"]).await?;
create_index(
"short_urls_original_url_idx",
"short_urls",
false,
&["original_url(32)"],
)
.await?;
create_index(
"short_urls_created_at_idx",
"short_urls",
src/infra/src/short_url/postgres.rs
-7
View File
@@ -57,13 +57,6 @@ impl ShortUrl for PostgresShortUrl {
/// Create index for short_urls at short_id and original_url
async fn create_table_index(&self) -> Result<()> {
create_index("short_urls_short_id_idx", "short_urls", true, &["short_id"]).await?;
create_index(
"short_urls_original_url_idx",
"short_urls",
false,
&["original_url"],
)
.await?;
create_index(
"short_urls_created_at_idx",
"short_urls",
src/infra/src/short_url/sqlite.rs
-7
View File
@@ -64,13 +64,6 @@ impl ShortUrl for SqliteShortUrl {
/// Creates indexes on the short_urls table
async fn create_table_index(&self) -> Result<()> {
create_index("short_urls_short_id_idx", "short_urls", true, &["short_id"]).await?;
create_index(
"short_urls_original_url_idx",
"short_urls",
false,
&["original_url"],
)
.await?;
create_index(
"short_urls_created_at_idx",
"short_urls",