[feat](load) enable stream load record to audit log system table

be/src/common/config.cpp

@@ -627,6 +627,12 @@ DEFINE_mInt32(slave_replica_writer_rpc_timeout_sec, "60");
 // Whether to enable stream load record function, the default is false.
 // False: disable stream load record
 DEFINE_mBool(enable_stream_load_record, "false");
+// Whether to enable stream load record to audit log table, the default is true.
+DEFINE_mBool(enable_stream_load_record_to_audit_log_table, "false");
+// the maximum bytes of a batch of stream load records to audit log table
+DEFINE_mInt64(stream_load_record_batch_bytes, "104857600"); // 100MB
+// the interval to send a batch of stream load records to audit log table
+DEFINE_mInt64(stream_load_record_batch_interval_secs, "120"); // 2 minutes
 // batch size of stream load record reported to FE
 DEFINE_mInt32(stream_load_record_batch_size, "50");
 // expire time of stream load record in rocksdb.

be/src/common/config.h

@@ -648,6 +648,12 @@ DECLARE_mInt32(slave_replica_writer_rpc_timeout_sec);
 // Whether to enable stream load record function, the default is false.
 // False: disable stream load record
 DECLARE_mBool(enable_stream_load_record);
+// Whether to enable stream load record to audit log table, the default is true.
+DECLARE_mBool(enable_stream_load_record_to_audit_log_table);
+// the maximum bytes of a batch of stream load records to audit log table
+DECLARE_mInt64(stream_load_record_batch_bytes);
+// the interval to send a batch of stream load records to audit log table
+DECLARE_mInt64(stream_load_record_batch_interval_secs);
 // batch size of stream load record reported to FE
 DECLARE_mInt32(stream_load_record_batch_size);
 // expire time of stream load record in rocksdb.

be/src/http/action/http_stream.cpp

@@ -113,9 +113,11 @@ void HttpStreamAction::handle(HttpRequest* req) {
     // add new line at end
     str = str + '\n';
     HttpChannel::send_reply(req, str);
-    if (config::enable_stream_load_record) {
-        str = ctx->prepare_stream_load_record(str);
-        _save_stream_load_record(ctx, str);
+    if (config::enable_stream_load_record || config::enable_stream_load_record_to_audit_log_table) {
+        if (req->header(HTTP_SKIP_RECORD_TO_AUDIT_LOG_TABLE).empty()) {
+            str = ctx->prepare_stream_load_record(str);
+            _save_stream_load_record(ctx, str);
+        }
     }
     // update statistics
     http_stream_requests_total->increment(1);
@@ -176,9 +178,12 @@ int HttpStreamAction::on_header(HttpRequest* req) {
         // add new line at end
         str = str + '\n';
         HttpChannel::send_reply(req, str);
-        if (config::enable_stream_load_record) {
-            str = ctx->prepare_stream_load_record(str);
-            _save_stream_load_record(ctx, str);
+        if (config::enable_stream_load_record ||
+            config::enable_stream_load_record_to_audit_log_table) {
+            if (req->header(HTTP_SKIP_RECORD_TO_AUDIT_LOG_TABLE).empty()) {
+                str = ctx->prepare_stream_load_record(str);
+                _save_stream_load_record(ctx, str);
+            }
         }
         return -1;
     }

be/src/http/action/stream_load.cpp

@@ -131,9 +131,11 @@ void StreamLoadAction::handle(HttpRequest* req) {
     str = str + '\n';
     HttpChannel::send_reply(req, str);
 #ifndef BE_TEST
-    if (config::enable_stream_load_record) {
-        str = ctx->prepare_stream_load_record(str);
-        _save_stream_load_record(ctx, str);
+    if (config::enable_stream_load_record || config::enable_stream_load_record_to_audit_log_table) {
+        if (req->header(HTTP_SKIP_RECORD_TO_AUDIT_LOG_TABLE).empty()) {
+            str = ctx->prepare_stream_load_record(str);
+            _save_stream_load_record(ctx, str);
+        }
     }
 #endif
 
@@ -238,9 +240,12 @@ int StreamLoadAction::on_header(HttpRequest* req) {
         str = str + '\n';
         HttpChannel::send_reply(req, str);
 #ifndef BE_TEST
-        if (config::enable_stream_load_record) {
-            str = ctx->prepare_stream_load_record(str);
-            _save_stream_load_record(ctx, str);
+        if (config::enable_stream_load_record ||
+            config::enable_stream_load_record_to_audit_log_table) {
+            if (req->header(HTTP_SKIP_RECORD_TO_AUDIT_LOG_TABLE).empty()) {
+                str = ctx->prepare_stream_load_record(str);
+                _save_stream_load_record(ctx, str);
+            }
         }
 #endif
         return -1;

be/src/http/http_client.cpp

@@ -409,6 +409,12 @@ Status HttpClient::execute_post_request(const std::string& payload, std::string*
     return execute(response);
 }
 
+Status HttpClient::execute_put_request(const std::string& payload, std::string* response) {
+    set_payload(payload);
+    curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "PUT");
+    return execute(response);
+}
+
 Status HttpClient::execute_delete_request(const std::string& payload, std::string* response) {
     set_method(DELETE);
     set_payload(payload);

be/src/http/http_client.h

@@ -158,6 +158,8 @@ class HttpClient {
 
     Status execute_post_request(const std::string& payload, std::string* response);
 
+    Status execute_put_request(const std::string& payload, std::string* response);
+
     Status execute_delete_request(const std::string& payload, std::string* response);
 
     // execute a simple method, and its response is saved in response argument

be/src/http/http_common.h

@@ -73,5 +73,6 @@ static const std::string HTTP_GROUP_COMMIT = "group_commit";
 static const std::string HTTP_CLOUD_CLUSTER = "cloud_cluster";
 static const std::string HTTP_EMPTY_FIELD_AS_NULL = "empty_field_as_null";
 static const std::string HTTP_COMPUTE_GROUP = "compute_group";
+static const std::string HTTP_SKIP_RECORD_TO_AUDIT_LOG_TABLE = "skip_record_to_audit_log_table";
 
 } // namespace doris

be/src/runtime/exec_env.h

@@ -30,6 +30,7 @@
 
 #include "common/config.h"
 #include "common/status.h"
+#include "exec/schema_scanner/schema_routine_load_job_scanner.h"
 #include "io/cache/fs_file_cache_storage.h"
 #include "olap/memtable_memory_limiter.h"
 #include "olap/options.h"
@@ -105,6 +106,7 @@ class LoadStreamMgr;
 class LoadStreamMapPool;
 class StreamLoadExecutor;
 class RoutineLoadTaskExecutor;
+class StreamLoadRecorderManager;
 class SmallFileMgr;
 class BackendServiceClient;
 class TPaloBrokerServiceClient;
@@ -490,6 +492,7 @@ class ExecEnv {
 
     std::unique_ptr<StreamLoadExecutor> _stream_load_executor;
     RoutineLoadTaskExecutor* _routine_load_task_executor = nullptr;
+    StreamLoadRecorderManager* _stream_load_recorder_manager = nullptr;
     SmallFileMgr* _small_file_mgr = nullptr;
     HeartbeatFlags* _heartbeat_flags = nullptr;
     vectorized::ScannerScheduler* _scanner_scheduler = nullptr;

be/src/runtime/exec_env_init.cpp

@@ -93,6 +93,7 @@
 #include "runtime/small_file_mgr.h"
 #include "runtime/stream_load/new_load_stream_mgr.h"
 #include "runtime/stream_load/stream_load_executor.h"
+#include "runtime/stream_load/stream_load_recorder_manager.h"
 #include "runtime/thread_context.h"
 #include "runtime/user_function_cache.h"
 #include "runtime/workload_group/workload_group_manager.h"
@@ -397,6 +398,10 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
         return st;
     }
 
+    // should start after storage_engine->open()
+    _stream_load_recorder_manager = new StreamLoadRecorderManager();
+    _stream_load_recorder_manager->start();
+
     // create internal workload group should be after storage_engin->open()
     RETURN_IF_ERROR(_create_internal_workload_group());
     _workload_sched_mgr = new WorkloadSchedPolicyMgr();
@@ -770,6 +775,7 @@ void ExecEnv::destroy() {
     SAFE_STOP(_group_commit_mgr);
     // _routine_load_task_executor should be stopped before _new_load_stream_mgr.
     SAFE_STOP(_routine_load_task_executor);
+    SAFE_STOP(_stream_load_recorder_manager);
     // stop workload scheduler
     SAFE_STOP(_workload_sched_mgr);
     // stop pipline step 2, cgroup execution
@@ -837,6 +843,7 @@ void ExecEnv::destroy() {
     SAFE_DELETE(_file_meta_cache);
     SAFE_DELETE(_group_commit_mgr);
     SAFE_DELETE(_routine_load_task_executor);
+    SAFE_DELETE(_stream_load_recorder_manager);
     // _stream_load_executor
     SAFE_DELETE(_function_client_cache);
     SAFE_DELETE(_streaming_client_cache);

be/src/runtime/stream_load/stream_load_recorder.cpp

@@ -22,7 +22,6 @@
 #include <rocksdb/status.h>
 
 #include <memory>
-#include <ostream>
 
 #include "common/config.h"
 #include "common/status.h"
@@ -63,7 +62,9 @@ Status StreamLoadRecorder::init() {
     std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
     // default column family is required
     column_families.emplace_back(DEFAULT_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions());
-    std::vector<int32_t> ttls = {config::stream_load_record_expire_time_secs};
+    // meta column family is used for bookkeeping data such as last fetch keys and should not expire
+    column_families.emplace_back(META_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions());
+    std::vector<int32_t> ttls = {config::stream_load_record_expire_time_secs, 0};
 
     rocksdb::DBWithTTL* tmp = _db.release();
     rocksdb::Status s =
@@ -78,8 +79,9 @@ Status StreamLoadRecorder::init() {
     return Status::OK();
 }
 
-Status StreamLoadRecorder::put(const std::string& key, const std::string& value) {
-    rocksdb::ColumnFamilyHandle* handle = _handles[0];
+Status StreamLoadRecorder::put(const std::string& key, const std::string& value, bool use_meta_cf) {
+    rocksdb::ColumnFamilyHandle* handle =
+            _handles[use_meta_cf ? META_COLUMN_FAMILY_INDEX : DEFAULT_COLUMN_FAMILY_INDEX];
     rocksdb::WriteOptions write_options;
     write_options.sync = false;
     rocksdb::Status s = _db->Put(write_options, handle, rocksdb::Slice(key), rocksdb::Slice(value));
@@ -89,20 +91,41 @@ Status StreamLoadRecorder::put(const std::string& key, const std::string& value)
                                      s.ToString());
     }
 
+    if (!use_meta_cf) {
+        return Status::OK();
+    }
+
     if ((UnixMillis() - _last_compaction_time) / 1000 >
         config::clean_stream_load_record_interval_secs) {
         rocksdb::CompactRangeOptions options;
-        s = _db->CompactRange(options, _handles[0], nullptr, nullptr);
+        s = _db->CompactRange(options, _handles[DEFAULT_COLUMN_FAMILY_INDEX], nullptr, nullptr);
         if (s.ok()) {
             _last_compaction_time = UnixMillis();
         }
     }
+
+    return Status::OK();
+}
+
+Status StreamLoadRecorder::get(const std::string& key, std::string* value, bool use_meta_cf) {
+    rocksdb::ColumnFamilyHandle* handle =
+            _handles[use_meta_cf ? META_COLUMN_FAMILY_INDEX : DEFAULT_COLUMN_FAMILY_INDEX];
+    rocksdb::ReadOptions read_options;
+    rocksdb::Status s = _db->Get(read_options, handle, rocksdb::Slice(key), value);
+    if (s.IsNotFound()) {
+        return Status::NotFound("Key not found: {}", key);
+    }
+    if (!s.ok()) {
+        LOG(WARNING) << "rocks db get key:" << key << " failed, reason:" << s.ToString();
+        return Status::InternalError("Stream load record rocksdb get failed, reason: {}",
+                                     s.ToString());
+    }
     return Status::OK();
 }
 
 Status StreamLoadRecorder::get_batch(const std::string& start, int batch_size,
                                      std::map<std::string, std::string>* stream_load_records) {
-    rocksdb::ColumnFamilyHandle* handle = _handles[0];
+    rocksdb::ColumnFamilyHandle* handle = _handles[DEFAULT_COLUMN_FAMILY_INDEX];
     std::unique_ptr<rocksdb::Iterator> it(_db->NewIterator(rocksdb::ReadOptions(), handle));
     if (start == "-1") {
         it->SeekToFirst();

be/src/runtime/stream_load/stream_load_recorder.h

@@ -45,7 +45,9 @@ class StreamLoadRecorder {
 
     Status init();
 
-    Status put(const std::string& key, const std::string& value);
+    Status put(const std::string& key, const std::string& value, bool use_meta_cf = false);
+
+    Status get(const std::string& key, std::string* value, bool use_meta_cf = false);
 
     Status get_batch(const std::string& start, int batch_size,
                      std::map<std::string, std::string>* stream_load_records);
@@ -59,9 +61,11 @@ class StreamLoadRecorder {
 
     enum ColumnFamilyIndex {
         DEFAULT_COLUMN_FAMILY_INDEX = 0,
+        META_COLUMN_FAMILY_INDEX = 1,
     };
 
     const std::string DEFAULT_COLUMN_FAMILY = "default";
+    const std::string META_COLUMN_FAMILY = "meta";
 };
 
 } // namespace doris