[Enhancement](load) Limit the number of incorrect data drops

Author: felixwluo

be/src/common/config.cpp

@@ -470,6 +470,8 @@ DEFINE_Int32(single_replica_load_download_num_workers, "64");
 DEFINE_Int64(load_data_reserve_hours, "4");
 // log error log will be removed after this time
 DEFINE_mInt64(load_error_log_reserve_hours, "48");
+// error log size limit, default 200MB
+DEFINE_mInt64(load_error_log_limit_bytes, "209715200");
 
 DEFINE_Int32(brpc_heavy_work_pool_threads, "-1");
 DEFINE_Int32(brpc_light_work_pool_threads, "-1");

be/src/common/config.h

@@ -521,6 +521,8 @@ DECLARE_Int32(single_replica_load_download_num_workers);
 DECLARE_Int64(load_data_reserve_hours);
 // log error log will be removed after this time
 DECLARE_mInt64(load_error_log_reserve_hours);
+// error log size limit, default 200MB
+DECLARE_mInt64(load_error_log_limit_bytes);
 
 // be brpc interface is classified into two categories: light and heavy
 // each category has diffrent thread number

be/src/runtime/runtime_state.cpp

@@ -414,8 +414,16 @@ Status RuntimeState::append_error_msg_to_file(std::function<std::string()> line,
         }
     }
 
-    if (out.size() > 0) {
-        (*_error_log_file) << fmt::to_string(out) << std::endl;
+    size_t error_row_size = out.size();
+    if (error_row_size > 0) {
+        if (error_row_size > config::load_error_log_limit_bytes) {
+            fmt::memory_buffer limit_byte_out;
+            limit_byte_out.append(out.data(), out.data() + config::load_error_log_limit_bytes);
+            (*_error_log_file) << fmt::to_string(limit_byte_out) + "error log is too long"
+                               << std::endl;
+        } else {
+            (*_error_log_file) << fmt::to_string(out) << std::endl;
+        }
     }
     return Status::OK();
 }

regression-test/data/load_p0/stream_load/test_stream_load_err_log_limit.csv

@@ -0,0 +1,50 @@
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc
+1,abc

regression-test/data/load_p0/stream_load/test_stream_load_err_log_limit.out

@@ -0,0 +1,3 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+0

regression-test/suites/load_p0/stream_load/test_stream_load_err_log_limit.groovy

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_stream_load_err_log_limit", "p0") {
+    sql "show tables"
+
+    def tableName = "test_stream_load_err_log_limit_table"
+
+    sql """ DROP TABLE IF EXISTS ${tableName} """
+    sql """
+        CREATE TABLE IF NOT EXISTS ${tableName} (
+            `k1` int NOT NULL,
+            `k2` varchar(20) NOT NULL
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`k1`)
+        DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+        PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+    """
+
+    def backendId_to_backendIP = [:]
+    def backendId_to_backendHttpPort = [:]
+    getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+    def set_be_param = { paramName, paramValue ->
+        for (String id in backendId_to_backendIP.keySet()) {
+            def beIp = backendId_to_backendIP.get(id)
+            def bePort = backendId_to_backendHttpPort.get(id)
+            def (code, out, err) = curl("POST", String.format("http://%s:%s/api/update_config?%s=%s", beIp, bePort, paramName, paramValue))
+            assertTrue(out.contains("OK"))
+        }
+    }
+
+    try {
+        set_be_param.call("load_error_log_limit_bytes", "100")
+
+        streamLoad {
+            table "${tableName}"
+            set 'column_separator', ','
+            set 'columns', 'k1, k2, k3'
+            file 'test_stream_load_err_log_limit.csv'
+
+            check { result, exception, startTime, endTime ->
+                if (exception != null) {
+                    throw exception
+                }
+                log.info("Stream load result: ${result}".toString())
+                def json = parseJson(result)
+                def (code, out, err) = curl("GET", json.ErrorURL)
+                log.info("error result: " + out)
+                def checkError = out.contains("error log is too long")
+                assertTrue(checkError)
+                log.info("url: " + json.ErrorURL)
+            }
+        }
+    } finally {
+        set_be_param.call("load_error_log_limit_bytes", "209715200")
+    }
+
+    sql "sync"
+    qt_sql "select count(*) from ${tableName}"
+    sql """ DROP TABLE IF EXISTS ${tableName} """
+}