fix: log table transform should keep UTC time (#18059)

* fix: log table transform should keep UTC time

* fix: log table transform should keep UTC time

* fix: log table transform should keep UTC time

Author: dqhl76

src/common/tracing/src/predefined_tables/history_tables.toml

@@ -2,26 +2,26 @@
 name = "log_history"
 target = ""
 create = "CREATE TABLE IF NOT EXISTS system_history.log_history (timestamp TIMESTAMP NULL, path STRING NULL, target STRING NULL, log_level STRING NULL, cluster_id STRING NULL, node_id STRING NULL, warehouse_id STRING NULL, query_id STRING NULL, message STRING NULL, fields VARIANT NULL, batch_number Int64) CLUSTER BY LINEAR(timestamp, query_id)"
-transform = "COPY INTO system_history.log_history FROM (SELECT timestamp, path, target, log_level, cluster_id,node_id, warehouse_id, query_id, message, fields, {batch_number} FROM @{stage_name}) file_format = (TYPE = PARQUET) PURGE = TRUE"
+transform = "settings (timezone='Etc/UTC') COPY INTO system_history.log_history FROM (SELECT timestamp, path, target, log_level, cluster_id,node_id, warehouse_id, query_id, message, fields, {batch_number} FROM @{stage_name}) file_format = (TYPE = PARQUET) PURGE = TRUE"
 delete = "DELETE FROM system_history.log_history WHERE timestamp < subtract_hours(NOW(), {retention_hours})"
 
 [[tables]]
 name = "query_history"
 target = "databend::log::query"
 create = "CREATE TABLE IF NOT EXISTS system_history.query_history (log_type Int8 NULL, log_type_name STRING NULL, handler_type STRING NULL, tenant_id STRING NULL, cluster_id STRING NULL, node_id STRING NULL, sql_user STRING NULL, sql_user_quota STRING NULL, sql_user_privileges STRING NULL, query_id STRING NULL, query_kind STRING NULL, query_text STRING NULL, query_hash STRING NULL, query_parameterized_hash STRING NULL, event_date DATE NULL, event_time TIMESTAMP NULL, query_start_time TIMESTAMP NULL, query_duration_ms Int64 NULL, query_queued_duration_ms Int64 NULL, current_database STRING NULL, written_rows UInt64 NULL, written_bytes UInt64 NULL, join_spilled_rows UInt64 NULL, join_spilled_bytes UInt64 NULL, agg_spilled_rows UInt64 NULL, agg_spilled_bytes UInt64 NULL, group_by_spilled_rows UInt64 NULL, group_by_spilled_bytes UInt64 NULL, written_io_bytes UInt64 NULL, written_io_bytes_cost_ms UInt64 NULL, scan_rows UInt64 NULL, scan_bytes UInt64 NULL, scan_io_bytes UInt64 NULL, scan_io_bytes_cost_ms UInt64 NULL, scan_partitions UInt64 NULL, total_partitions UInt64 NULL, result_rows UInt64 NULL, result_bytes UInt64 NULL, bytes_from_remote_disk UInt64 NULL, bytes_from_local_disk UInt64 NULL, bytes_from_memory UInt64 NULL, client_address STRING NULL, user_agent STRING NULL, exception_code Int32 NULL, exception_text STRING NULL, server_version STRING NULL, query_tag STRING NULL, has_profile BOOLEAN NULL, peek_memory_usage VARIANT NULL) CLUSTER BY LINEAR(event_time, query_id)"
-transform = "INSERT INTO system_history.query_history FROM (SELECT m['log_type'], m['log_type_name'], m['handler_type'], m['tenant_id'], m['cluster_id'], m['node_id'], m['sql_user'], m['sql_user_quota'], m['sql_user_privileges'], m['query_id'], m['query_kind'], m['query_text'], m['query_hash'], m['query_parameterized_hash'], m['event_date'], m['event_time'], m['query_start_time'], m['query_duration_ms'], m['query_queued_duration_ms'], m['current_database'], m['written_rows'], m['written_bytes'], m['join_spilled_rows'], m['join_spilled_bytes'], m['agg_spilled_rows'], m['agg_spilled_bytes'], m['group_by_spilled_rows'], m['group_by_spilled_bytes'], m['written_io_bytes'], m['written_io_bytes_cost_ms'], m['scan_rows'], m['scan_bytes'], m['scan_io_bytes'], m['scan_io_bytes_cost_ms'], m['scan_partitions'], m['total_partitions'], m['result_rows'], m['result_bytes'], m['bytes_from_remote_disk'], m['bytes_from_local_disk'], m['bytes_from_memory'], m['client_address'], m['user_agent'], m['exception_code'], m['exception_text'], m['server_version'], m['query_tag'], m['has_profile'], m['peek_memory_usage'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::query' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
+transform = "settings (timezone='Etc/UTC') INSERT INTO system_history.query_history FROM (SELECT m['log_type'], m['log_type_name'], m['handler_type'], m['tenant_id'], m['cluster_id'], m['node_id'], m['sql_user'], m['sql_user_quota'], m['sql_user_privileges'], m['query_id'], m['query_kind'], m['query_text'], m['query_hash'], m['query_parameterized_hash'], m['event_date'], m['event_time'], m['query_start_time'], m['query_duration_ms'], m['query_queued_duration_ms'], m['current_database'], m['written_rows'], m['written_bytes'], m['join_spilled_rows'], m['join_spilled_bytes'], m['agg_spilled_rows'], m['agg_spilled_bytes'], m['group_by_spilled_rows'], m['group_by_spilled_bytes'], m['written_io_bytes'], m['written_io_bytes_cost_ms'], m['scan_rows'], m['scan_bytes'], m['scan_io_bytes'], m['scan_io_bytes_cost_ms'], m['scan_partitions'], m['total_partitions'], m['result_rows'], m['result_bytes'], m['bytes_from_remote_disk'], m['bytes_from_local_disk'], m['bytes_from_memory'], m['client_address'], m['user_agent'], m['exception_code'], m['exception_text'], m['server_version'], m['query_tag'], m['has_profile'], m['peek_memory_usage'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::query' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
 delete = "DELETE FROM system_history.query_history WHERE event_time < subtract_hours(NOW(), {retention_hours})"
 
 [[tables]]
 name = "profile_history"
 target = "databend::log::profile"
 create = "CREATE TABLE IF NOT EXISTS system_history.profile_history (timestamp TIMESTAMP NULL, query_id VARCHAR NULL, profiles VARIANT NULL, statistics_desc VARIANT NULL) CLUSTER BY (timestamp, query_id)"
-transform = "INSERT INTO system_history.profile_history FROM (SELECT timestamp, m['query_id'], m['profiles'], m['statistics_desc'] FROM (SELECT timestamp, parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::profile' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
+transform = "settings (timezone='Etc/UTC') INSERT INTO system_history.profile_history FROM (SELECT timestamp, m['query_id'], m['profiles'], m['statistics_desc'] FROM (SELECT timestamp, parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::profile' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
 delete = "DELETE FROM system_history.profile_history WHERE timestamp < subtract_hours(NOW(), {retention_hours})"
 
 [[tables]]
 name = "login_history"
 target = "databend::log::login"
 create = "CREATE TABLE IF NOT EXISTS system_history.login_history (event_time TIMESTAMP NULL, handler STRING NULL, event_type STRING NULL, connection_uri STRING NULL, auth_type STRING NULL, user_name STRING NULL, client_ip STRING NULL, user_agent STRING NULL, session_id STRING NULL, node_id STRING NULL, error_message STRING NULL)"
-transform = "INSERT INTO system_history.login_history FROM (SELECT m['event_time'], m['handler'], m['event_type'], m['connection_uri'], m['auth_type'], m['user_name'], m['client_ip'], m['user_agent'], m['session_id'], m['node_id'], m['error_message'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::login' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
+transform = "settings (timezone='Etc/UTC') INSERT INTO system_history.login_history FROM (SELECT m['event_time'], m['handler'], m['event_type'], m['connection_uri'], m['auth_type'], m['user_name'], m['client_ip'], m['user_agent'], m['session_id'], m['node_id'], m['error_message'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::login' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
 delete = "DELETE FROM system_history.login_history WHERE event_time < subtract_hours(NOW(), {retention_hours})"

src/common/tracing/src/remote_log.rs

@@ -217,7 +217,7 @@ impl RemoteLog {
         let fields = serde_json::to_string(&fields).unwrap_or_default();
 
         let log_level = record.level().to_string();
-        let timestamp = chrono::Local::now().timestamp_micros();
+        let timestamp = chrono::Utc::now().timestamp_micros();
 
         let path = format!(
             "{}: {}:{}",
@@ -266,7 +266,7 @@ impl LogBuffer {
     pub fn new(sender: Sender<LogMessage>, interval: u64) -> Self {
         Self {
             queue: ConcurrentQueue::unbounded(),
-            last_collect: AtomicU64::new(chrono::Local::now().timestamp_micros() as u64),
+            last_collect: AtomicU64::new(chrono::Utc::now().timestamp_micros() as u64),
             sender,
             interval,
         }
@@ -277,12 +277,12 @@ impl LogBuffer {
         self.queue.push(log_element)?;
         if self.queue.len() >= Self::MAX_BUFFER_SIZE {
             self.last_collect.store(
-                chrono::Local::now().timestamp_micros() as u64,
+                chrono::Utc::now().timestamp_micros() as u64,
                 Ordering::SeqCst,
             );
             self.collect()?;
         }
-        let now = chrono::Local::now().timestamp_micros() as u64;
+        let now = chrono::Utc::now().timestamp_micros() as u64;
         let mut current_last_collect = 0;
         loop {
             match self.last_collect.compare_exchange_weak(

src/query/service/src/history_tables/global_history_log.rs

@@ -219,7 +219,7 @@ impl GlobalHistoryLog {
         {
             Some(v) => {
                 let last: u64 = serde_json::from_slice(&v.data)?;
-                chrono::Local::now().timestamp_millis() as u64
+                chrono::Utc::now().timestamp_millis() as u64
                     - Duration::from_secs(interval).as_millis() as u64
                     > last
             }
@@ -238,9 +238,7 @@ impl GlobalHistoryLog {
             .upsert_kv(UpsertKV::new(
                 format!("{}/last_timestamp", meta_key),
                 MatchSeq::Any,
-                Operation::Update(serde_json::to_vec(
-                    &chrono::Local::now().timestamp_millis(),
-                )?),
+                Operation::Update(serde_json::to_vec(&chrono::Utc::now().timestamp_millis())?),
                 None,
             ))
             .await?;