fix: service logs: aggregation query + killpill on index pull + reten…

…tion 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

backend/ee-repo-ref.txt

@@ -1 +1 @@
-446e4fbc59048bb11c18648ac094e57c0bfa4a28
+3fa1a6bc51f39e65d441afed31825347b24a521e

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
                                                 },