From 09fe535bdca548e0445ffb07d86a1c3cf656d5e5 Mon Sep 17 00:00:00 2001 From: wendrul <53628737+wendrul@users.noreply.github.com> Date: Wed, 27 Nov 2024 18:57:23 +0100 Subject: [PATCH] fix: service logs: aggregation query + killpill on index pull + retention period for logs on s3 (#4795) * Search across all hosts in index. killpill on index pull * Add retention period for service logs * Remove console log * monitor log_file when older than biggest retention period (job / log) * Retention period deletes service and job logs * Add setting to monitor logs out of s3 * Prepare sqlx * Update ee repo ref * Prepare sqlx * Unused variables --- ...43061932dc5b4b17cf13d5cf8d833aedbe6d5.json | 2 +- ...c5139ee535e23c8b3362c52eef49998349f15.json | 22 +++ ...5d1096bdfd6c539d8c31462f8f2bd711ab0fa.json | 14 -- ...63708ef7fe2c88b28d0667988a47b9e995c0e.json | 23 --- ...7f5033b9c9afc344d9c3e385ba20a3ad2197a.json | 2 +- ...62c3c267ca336a8b6bec5b29d4409030ed561.json | 4 +- ...95fbe9c6b59e27c368eaf78b33ecdfa95140f.json | 23 --- ...8cce4a406b74c9a8c8590a3bae323d59d046a.json | 14 -- ...633c1531170ded1fdf09114718b941f5e1db.json} | 5 +- ...6f956d47810a94cb052fdcffb5cfe440f8033.json | 28 ++++ ...eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json | 4 +- ...110d4702a80f455c63af6fcbd7cf45bbc4f7a.json | 23 --- ...160ad1962668fc3305d8e80ae91ef73614a80.json | 2 +- backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 50 +++++-- backend/src/monitor.rs | 133 +++++++++++++++--- backend/windmill-api/openapi.yaml | 5 - .../windmill-common/src/global_settings.rs | 1 + backend/windmill-common/src/lib.rs | 3 + .../lib/components/ServiceLogsInner.svelte | 20 ++- .../src/lib/components/instanceSettings.ts | 42 +++--- 21 files changed, 246 insertions(+), 176 deletions(-) create mode 100644 backend/.sqlx/query-288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15.json delete mode 100644 backend/.sqlx/query-467c563cdc81dd57a01735a9f535d1096bdfd6c539d8c31462f8f2bd711ab0fa.json delete mode 100644 backend/.sqlx/query-4a764cdcb847b71183425a7a0e863708ef7fe2c88b28d0667988a47b9e995c0e.json delete mode 100644 backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json delete mode 100644 backend/.sqlx/query-845120388af12c2b2f57fedf95c8cce4a406b74c9a8c8590a3bae323d59d046a.json rename backend/.sqlx/{query-dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6.json => query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json} (85%) create mode 100644 backend/.sqlx/query-94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033.json delete mode 100644 backend/.sqlx/query-ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a.json diff --git a/backend/.sqlx/query-02424907504848e983bfa89eec343061932dc5b4b17cf13d5cf8d833aedbe6d5.json b/backend/.sqlx/query-02424907504848e983bfa89eec343061932dc5b4b17cf13d5cf8d833aedbe6d5.json index c8b5e3086f..043d236a6b 100644 --- a/backend/.sqlx/query-02424907504848e983bfa89eec343061932dc5b4b17cf13d5cf8d833aedbe6d5.json +++ b/backend/.sqlx/query-02424907504848e983bfa89eec343061932dc5b4b17cf13d5cf8d833aedbe6d5.json @@ -5,7 +5,7 @@ "columns": [ { "ordinal": 0, - "name": "?column?", + "name": "bool", "type_info": "Bool" } ], diff --git a/backend/.sqlx/query-288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15.json b/backend/.sqlx/query-288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15.json new file mode 100644 index 0000000000..eb2b1979d9 --- /dev/null +++ b/backend/.sqlx/query-288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "log_file_index", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + true + ] + }, + "hash": "288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15" +} diff --git a/backend/.sqlx/query-467c563cdc81dd57a01735a9f535d1096bdfd6c539d8c31462f8f2bd711ab0fa.json b/backend/.sqlx/query-467c563cdc81dd57a01735a9f535d1096bdfd6c539d8c31462f8f2bd711ab0fa.json deleted file mode 100644 index 0b47ce7a7d..0000000000 --- a/backend/.sqlx/query-467c563cdc81dd57a01735a9f535d1096bdfd6c539d8c31462f8f2bd711ab0fa.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM job_logs WHERE job_id = ANY($1)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "UuidArray" - ] - }, - "nullable": [] - }, - "hash": "467c563cdc81dd57a01735a9f535d1096bdfd6c539d8c31462f8f2bd711ab0fa" -} diff --git a/backend/.sqlx/query-4a764cdcb847b71183425a7a0e863708ef7fe2c88b28d0667988a47b9e995c0e.json b/backend/.sqlx/query-4a764cdcb847b71183425a7a0e863708ef7fe2c88b28d0667988a47b9e995c0e.json deleted file mode 100644 index 9fde573b4d..0000000000 --- a/backend/.sqlx/query-4a764cdcb847b71183425a7a0e863708ef7fe2c88b28d0667988a47b9e995c0e.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT flow_version.value \n FROM flow \n LEFT JOIN flow_version \n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 AND flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "value", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "4a764cdcb847b71183425a7a0e863708ef7fe2c88b28d0667988a47b9e995c0e" -} diff --git a/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json b/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json index 75b8108281..1fa370e682 100644 --- a/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json +++ b/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json @@ -5,7 +5,7 @@ "columns": [ { "ordinal": 0, - "name": "?column?", + "name": "bool", "type_info": "Bool" } ], diff --git a/backend/.sqlx/query-6f4817fad2739a11d89b6704edf62c3c267ca336a8b6bec5b29d4409030ed561.json b/backend/.sqlx/query-6f4817fad2739a11d89b6704edf62c3c267ca336a8b6bec5b29d4409030ed561.json index 5605518c9c..a3da2f6bc9 100644 --- a/backend/.sqlx/query-6f4817fad2739a11d89b6704edf62c3c267ca336a8b6bec5b29d4409030ed561.json +++ b/backend/.sqlx/query-6f4817fad2739a11d89b6704edf62c3c267ca336a8b6bec5b29d4409030ed561.json @@ -26,7 +26,9 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow" + "singlescriptflow", + "flowscript", + "flownode" ] } } diff --git a/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json b/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json deleted file mode 100644 index 9dcfbc697b..0000000000 --- a/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) AS \"value!: sqlx::types::Json>\" \n FROM flow \n LEFT JOIN flow_version \n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n LEFT JOIN flow_version_lite \n ON flow_version_lite.id = flow_version.id\n WHERE flow.path = $1 AND flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "value!: sqlx::types::Json>", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f" -} diff --git a/backend/.sqlx/query-845120388af12c2b2f57fedf95c8cce4a406b74c9a8c8590a3bae323d59d046a.json b/backend/.sqlx/query-845120388af12c2b2f57fedf95c8cce4a406b74c9a8c8590a3bae323d59d046a.json deleted file mode 100644 index afede37f44..0000000000 --- a/backend/.sqlx/query-845120388af12c2b2f57fedf95c8cce4a406b74c9a8c8590a3bae323d59d046a.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [] - }, - "hash": "845120388af12c2b2f57fedf95c8cce4a406b74c9a8c8590a3bae323d59d046a" -} diff --git a/backend/.sqlx/query-dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6.json b/backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json similarity index 85% rename from backend/.sqlx/query-dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6.json rename to backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json index b64db838d3..d0b96444ee 100644 --- a/backend/.sqlx/query-dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6.json +++ b/backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n hostname,\n mode::text,\n worker_group,\n log_ts,\n file_path,\n ok_lines,\n err_lines,\n json_fmt\n FROM log_file\n ORDER BY log_ts ASC LIMIT $1", + "query": "SELECT\n hostname,\n mode::text,\n worker_group,\n log_ts,\n file_path,\n ok_lines,\n err_lines,\n json_fmt\n FROM log_file\n WHERE log_ts > NOW() - make_interval(secs => $1)\n ORDER BY log_ts ASC LIMIT $2", "describe": { "columns": [ { @@ -46,6 +46,7 @@ ], "parameters": { "Left": [ + "Float8", "Int8" ] }, @@ -60,5 +61,5 @@ true ] }, - "hash": "dd967c5983fa0ff05e2b320ad0e0b5a152784826cb8fb4381c1ffe228cb7feb6" + "hash": "8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db" } diff --git a/backend/.sqlx/query-94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033.json b/backend/.sqlx/query-94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033.json new file mode 100644 index 0000000000..24a5ee62dd --- /dev/null +++ b/backend/.sqlx/query-94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval RETURNING file_path, hostname", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "file_path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "hostname", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033" +} diff --git a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json index 2393bc83df..7bf8ca9496 100644 --- a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json +++ b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json @@ -26,7 +26,9 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow" + "singlescriptflow", + "flowscript", + "flownode" ] } } diff --git a/backend/.sqlx/query-ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a.json b/backend/.sqlx/query-ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a.json deleted file mode 100644 index c97780ba06..0000000000 --- a/backend/.sqlx/query-ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT flow_version.value AS \"value: sqlx::types::Json>\" FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 AND flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "value: sqlx::types::Json>", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a" -} diff --git a/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json b/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json index 271c395f9a..103219fe05 100644 --- a/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json +++ b/backend/.sqlx/query-d6615719bf8db4b333ed55c9a3c160ad1962668fc3305d8e80ae91ef73614a80.json @@ -5,7 +5,7 @@ "columns": [ { "ordinal": 0, - "name": "?column?", + "name": "bool", "type_info": "Bool" } ], diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index f5b5a34d5e..fd88bdf0a0 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -446e4fbc59048bb11c18648ac094e57c0bfa4a28 \ No newline at end of file +3fa1a6bc51f39e65d441afed31825347b24a521e diff --git a/backend/src/main.rs b/backend/src/main.rs index a74a6d3be3..4e9850ead8 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -8,8 +8,7 @@ use anyhow::Context; use monitor::{ - reload_indexer_config, reload_timeout_wait_result_setting, - send_current_log_file_to_object_store, send_logs_to_object_store, + reload_delete_logs_periodically_setting, reload_indexer_config, reload_timeout_wait_result_setting, send_current_log_file_to_object_store, send_logs_to_object_store }; use rand::Rng; use sqlx::{postgres::PgListener, Pool, Postgres}; @@ -30,15 +29,7 @@ use windmill_common::ee::{maybe_renew_license_key_on_start, LICENSE_KEY_ID, LICE use windmill_common::{ global_settings::{ - BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, - CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, - DEFAULT_TAGS_WORKSPACES_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, - EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING, - JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, - LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, - REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, - TIMEOUT_WAIT_RESULT_SETTING, + BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING }, scripts::ScriptLang, stats_ee::schedule_stats, @@ -497,8 +488,21 @@ Windmill Community Edition {GIT_VERSION} #[cfg(feature = "tantivy")] let (index_reader, index_writer) = if should_index_jobs { - let (r, w) = windmill_indexer::completed_runs_ee::init_index(&db).await?; - (Some(r), Some(w)) + let mut indexer_rx = killpill_rx.resubscribe(); + + let (mut reader, mut writer) = (None, None); + tokio::select! { + _ = indexer_rx.recv() => { + tracing::info!("Received killpill, aborting index initialization"); + }, + res = windmill_indexer::completed_runs_ee::init_index(&db) => { + let res = res?; + reader = Some(res.0); + writer = Some(res.1); + } + + } + (reader, writer) } else { (None, None) }; @@ -522,8 +526,21 @@ Windmill Community Edition {GIT_VERSION} #[cfg(all(feature = "tantivy", feature = "parquet"))] let (log_index_reader, log_index_writer) = if should_index_jobs { - let (r, w) = windmill_indexer::service_logs_ee::init_index(&db).await?; - (Some(r), Some(w)) + let mut indexer_rx = killpill_rx.resubscribe(); + + let (mut reader, mut writer) = (None, None); + tokio::select! { + _ = indexer_rx.recv() => { + tracing::info!("Received killpill, aborting index initialization"); + }, + res = windmill_indexer::service_logs_ee::init_index(&db) => { + let res = res?; + reader = Some(res.0); + writer = Some(res.1); + } + + } + (reader, writer) } else { (None, None) }; @@ -700,6 +717,9 @@ Windmill Community Edition {GIT_VERSION} RETENTION_PERIOD_SECS_SETTING => { reload_retention_period_setting(&db).await }, + MONITOR_LOGS_ON_OBJECT_STORE_SETTING => { + reload_delete_logs_periodically_setting(&db).await + }, JOB_DEFAULT_TIMEOUT_SECS_SETTING => { reload_job_default_timeout_setting(&db).await }, diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 321eff4609..6fec44f0fe 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -11,6 +11,7 @@ use std::{ }; use chrono::{NaiveDateTime, Utc}; +use futures::{stream::FuturesUnordered, StreamExt}; use serde::de::DeserializeOwned; use sqlx::{Pool, Postgres}; use tokio::{ @@ -33,14 +34,7 @@ use windmill_common::{ error, flow_status::FlowStatusModule, global_settings::{ - BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, - CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, - DEFAULT_TAGS_WORKSPACES_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, - EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, - JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, - OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, - REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, - SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING, + BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING }, indexer::load_indexer_config, jobs::QueuedJob, @@ -52,10 +46,11 @@ use windmill_common::{ worker::{ load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting, update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, - SMTP_CONFIG, WORKER_CONFIG, WORKER_GROUP, + SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP, }, BASE_URL, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED, + MONITOR_LOGS_ON_OBJECT_STORE, SERVICE_LOG_RETENTION_SECS, }; use windmill_queue::cancel_job; use windmill_worker::{ @@ -567,6 +562,11 @@ pub async fn load_require_preexisting_user(db: &DB) { }; } +struct LogFile { + file_path: String, + hostname: String, +} + pub async fn delete_expired_items(db: &DB) -> () { let tokens_deleted_r: std::result::Result, _> = sqlx::query_scalar( "DELETE FROM token WHERE expiration <= now() @@ -629,6 +629,25 @@ pub async fn delete_expired_items(db: &DB) -> () { Err(e) => tracing::error!("Error deleting cache resource {}", e.to_string()), } + match sqlx::query_as!( + LogFile, + "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval RETURNING file_path, hostname", + SERVICE_LOG_RETENTION_SECS, + ) + .fetch_all(db) + .await + { + Ok(log_files_to_delete) => { + let paths = log_files_to_delete + .iter() + .map(|f| format!("{}/{}", f.hostname, f.file_path)) + .collect(); + delete_log_files_from_disk_and_store(paths, TMP_WINDMILL_LOGS_SERVICE, windmill_common::tracing_init::LOGS_SERVICE).await; + + } + Err(e) => tracing::error!("Error deleting log file: {:?}", e), + } + let job_retention_secs = *JOB_RETENTION_SECS.read().await; if job_retention_secs > 0 { match db.begin().await { @@ -658,14 +677,22 @@ pub async fn delete_expired_items(db: &DB) -> () { { tracing::error!("Error deleting job stats: {:?}", e); } - if let Err(e) = sqlx::query!( - "DELETE FROM job_logs WHERE job_id = ANY($1)", + match sqlx::query_scalar!( + "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index", &deleted_jobs ) - .execute(&mut *tx) + .fetch_all(&mut *tx) .await { - tracing::error!("Error deleting job stats: {:?}", e); + Ok(log_file_index) => { + let paths = log_file_index + .into_iter() + .filter_map(|opt| opt) + .flat_map(|inner_vec| inner_vec.into_iter()) + .collect(); + delete_log_files_from_disk_and_store(paths, TMP_DIR, "").await; + } + Err(e) => tracing::error!("Error deleting job stats: {:?}", e), } if let Err(e) = sqlx::query!( "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ", @@ -676,15 +703,7 @@ pub async fn delete_expired_items(db: &DB) -> () { { tracing::error!("Error deleting custom concurrency key: {:?}", e); } - if let Err(e) = sqlx::query!( - "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval ", - job_retention_secs - ) - .execute(&mut *tx) - .await - { - tracing::error!("Error deleting log file: {:?}", e); - } + if let Err(e) = sqlx::query!("DELETE FROM job WHERE id = ANY($1)", &deleted_jobs) .execute(&mut *tx) @@ -711,6 +730,62 @@ pub async fn delete_expired_items(db: &DB) -> () { } } +async fn delete_log_files_from_disk_and_store( + paths_to_delete: Vec, + tmp_dir: &str, + _s3_prefix: &str, +) { + #[cfg(feature = "parquet")] + let os = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone(); + #[cfg(not(feature = "parquet"))] + let os: Option<()> = None; + + let _should_del_from_store = MONITOR_LOGS_ON_OBJECT_STORE.read().await.clone(); + + let delete_futures = FuturesUnordered::new(); + + for path in paths_to_delete { + let _os2 = &os; + + delete_futures.push(async move { + let disk_path = std::path::Path::new(tmp_dir).join(&path); + if tokio::fs::metadata(&disk_path) + .await + .is_ok() + { + if let Err(e) = tokio::fs::remove_file(&disk_path).await { + tracing::error!( + "Failed to delete from disk {}: {e}", + disk_path.to_string_lossy() + ); + } else { + tracing::debug!( + "Succesfully deleted {} from disk", + disk_path.to_string_lossy() + ); + } + } + + #[cfg(feature = "parquet")] + if _should_del_from_store { + if let Some(os) = _os2 { + let p = object_store::path::Path::from(format!("{}{}", _s3_prefix, path)); + if let Err(e) = os.delete(&p).await { + tracing::error!("Failed to delete from object store {}: {e}", p.to_string()) + } else { + tracing::debug!("Succesfully deleted {} from object store", p.to_string()); + } + } + } + }); + } + + let _: Vec<_> = delete_futures.collect().await; +} + pub async fn reload_scim_token_setting(db: &DB) { reload_option_setting_with_tracing(db, SCIM_TOKEN_SETTING, "SCIM_TOKEN", SCIM_TOKEN.clone()) .await; @@ -789,6 +864,20 @@ pub async fn reload_retention_period_setting(db: &DB) { tracing::error!("Error reloading retention period: {:?}", e) } } +pub async fn reload_delete_logs_periodically_setting(db: &DB) { + if let Err(e) = reload_setting( + db, + MONITOR_LOGS_ON_OBJECT_STORE_SETTING, + "MONITOR_LOGS_ON_OBJECT_STORE", + false, + MONITOR_LOGS_ON_OBJECT_STORE.clone(), + |x| x, + ) + .await + { + tracing::error!("Error reloading retention period: {:?}", e) + } +} #[cfg(feature = "parquet")] pub async fn reload_s3_cache_setting(db: &DB) { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 4a5d070044..10f5438bda 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -10098,11 +10098,6 @@ paths: required: true schema: type: string - - name: hosts - in: query - required: true - schema: - type: string - name: min_ts in: query required: false diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index f24153ec38..7b932d8f50 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -4,6 +4,7 @@ pub const DEFAULT_TAGS_WORKSPACES_SETTING: &str = "default_tags_workspaces"; pub const BASE_URL_SETTING: &str = "base_url"; pub const OAUTH_SETTING: &str = "oauths"; pub const RETENTION_PERIOD_SECS_SETTING: &str = "retention_period_secs"; +pub const MONITOR_LOGS_ON_OBJECT_STORE_SETTING: &str = "monitor_logs_on_s3"; pub const JOB_DEFAULT_TIMEOUT_SECS_SETTING: &str = "job_default_timeout"; pub const REQUEST_SIZE_LIMIT_SETTING: &str = "request_size_limit_mb"; pub const LICENSE_KEY_SETTING: &str = "license_key"; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 91f49da937..20a8a71e63 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -54,6 +54,7 @@ pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5; pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5; pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev"; +pub const SERVICE_LOG_RETENTION_SECS: i64 = 60 * 24 * 14; // 2 weeks retention period for logs #[macro_export] macro_rules! add_time { @@ -99,6 +100,8 @@ lazy_static::lazy_static! { pub static ref JOB_RETENTION_SECS: Arc> = Arc::new(RwLock::new(0)); + pub static ref MONITOR_LOGS_ON_OBJECT_STORE: Arc> = Arc::new(RwLock::new(false)); + pub static ref INSTANCE_NAME: String = rd_string(5); } diff --git a/frontend/src/lib/components/ServiceLogsInner.svelte b/frontend/src/lib/components/ServiceLogsInner.svelte index 6bccebc5d6..0b78ddb688 100644 --- a/frontend/src/lib/components/ServiceLogsInner.svelte +++ b/frontend/src/lib/components/ServiceLogsInner.svelte @@ -309,21 +309,17 @@ debounceTimeout && clearTimeout(debounceTimeout) debounceTimeout = setTimeout(async () => { if (allLogs) { - const hostnames: string[] = [] - Object.entries(allLogs).forEach(([mode, wgs]) => { - Object.entries(wgs).forEach(([wg, keys]) => { - Object.entries(keys).forEach(([key, _log_files]) => { - hostnames.push(`${mode},${wg},${key}`) - }) - }) - }) const countLogsResponse = await IndexSearchService.countSearchLogsIndex({ searchQuery: searchTerm, - hosts: hostnames.join(','), minTs, maxTs }) - countsPerHost = countLogsResponse.count_per_host + const buckets = (countLogsResponse.count_per_host as any)['count_per_host']['buckets'] + countsPerHost = new Map(buckets.map(({ key, doc_count }) => [key, doc_count])); + countsPerHost = buckets.reduce((acc: any, { key, doc_count }) => { + acc[key] = {doc_count}; + return acc; + }, {} as Record) queryParseErrors = countLogsResponse.query_parse_errors ?? [] loadingLogCounts = false } @@ -539,7 +535,7 @@ return true } const hostKey = `${mode},${wg},${hn}` - if (countsPerHost && countsPerHost[hostKey] && countsPerHost[hostKey].count === 0) { + if (countsPerHost && (countsPerHost[hostKey] == undefined || countsPerHost[hostKey].doc_count === 0)) { return false } return true @@ -570,7 +566,7 @@ {:else if countsPerHost}
- {countsPerHost[hostKey]?.count} matches + {countsPerHost[hostKey]?.doc_count ?? 0} matches
{:else}
diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 2a8dd95aad..a4584e8582 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -7,20 +7,20 @@ export interface Setting { tooltip?: string key: string fieldType: - | 'text' - | 'number' - | 'boolean' - | 'password' - | 'select' - | 'textarea' - | 'seconds' - | 'email' - | 'license_key' - | 'object_store_config' - | 'critical_error_channels' - | 'slack_connect' - | 'smtp_connect' - | 'indexer_rates' + | 'text' + | 'number' + | 'boolean' + | 'password' + | 'select' + | 'textarea' + | 'seconds' + | 'email' + | 'license_key' + | 'object_store_config' + | 'critical_error_channels' + | 'slack_connect' + | 'smtp_connect' + | 'indexer_rates' storage: SettingStorage advancedToggle?: { label: string @@ -50,9 +50,9 @@ export const settings: Record = { isValid: (value: string | undefined) => value ? value?.startsWith('http') && - value.includes('://') && - !value?.endsWith('/') && - !value?.endsWith(' ') + value.includes('://') && + !value?.endsWith('/') && + !value?.endsWith(' ') : false }, { @@ -120,6 +120,14 @@ export const settings: Record = { ee_only: 'You can only adjust this setting to above 30 days in the EE version', cloudonly: false }, + { + label: 'Delete logs from s3 periodically', + description: 'Job and service logs are periodically deleted from disk. When this setting is on, they will also be deleted from the object storage.', + key: 'monitor_logs_on_s3', + fieldType: 'boolean', + storage: 'setting', + ee_only: '' + }, { label: 'Expose metrics', description: