kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: service logs: aggregation query + killpill on index pull + retention period for logs on s3 (#4795)

Browse Source 
* 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
This commit is contained in:
wendrul
2024-11-27 18:57:23 +01:00
committed by GitHub
parent c998102cf3
commit 09fe535bdc
21 changed files with 246 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
backend/src/main.rs
View File

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

133
backend/src/monitor.rs
View File

@@ -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<Vec<String>, _> = 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<String>,
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) {