fix(broker): if errno==0, we considerer the error as a server error

broker/core/sql/src/mysql_connection.cc

@@ -16,6 +16,7 @@
  * For more information : [email protected]
  */
 #include <errmsg.h>
+#include <mysqld_error.h>
 
 #include "com/centreon/broker/config/applier/init.hh"
 #include "com/centreon/broker/misc/misc.hh"
@@ -462,10 +463,10 @@ void mysql_connection::_statement(mysql_task* t) {
       if (mysql_stmt_execute(stmt)) {
         int32_t err_code = ::mysql_stmt_errno(stmt);
         if (err_code == 0) {
-          SPDLOG_LOGGER_TRACE(
-              _logger,
-              "mysql_connection: errno=0, so we simulate a server error 1213");
-          err_code = 1213;
+          SPDLOG_LOGGER_TRACE(_logger,
+                              "mysql_connection: errno=0, so we simulate a "
+                              "server error CR_SERVER_LOST");
+          err_code = CR_SERVER_LOST;
         }
         std::string err_msg(fmt::format("{} errno={} {}",
                                         mysql_error::msg[task->error_code],
@@ -477,8 +478,8 @@ void mysql_connection::_statement(mysql_task* t) {
           set_error_message(err_msg);
           break;
         }
-        if (mysql_stmt_errno(stmt) != 1213 &&
-            mysql_stmt_errno(stmt) != 1205)  // Dead Lock error
+        if (err_code != ER_LOCK_DEADLOCK &&
+            err_code != ER_LOCK_WAIT_TIMEOUT)  // Dead Lock error
           attempts = MAX_ATTEMPTS;
 
         if (mysql_commit(_conn)) {

tests/broker-engine/services-and-bulk-stmt.robot

@@ -496,6 +496,84 @@ EBMSSMDBD
 	Log To Console    ${result}
     END
 
+EBMSSMPART
+    [Documentation]    1000 services are configured with 100 metrics each.
+    ...    The rrd output is removed from the broker configuration.
+    ...    The data_bin table is configured with two partitions p1 and p2 such
+    ...    that p1 contains old data and p2 contains current data.
+    ...    While metrics are written in the database, we remove the p2 partition.
+    ...    Once the p2 partition is recreated, broker must recover its connection
+    ...    to the database and continue to write metrics.
+    ...    To check that last point, we force a last service check and we check
+    ...    that its metrics are written in the database.
+    [Tags]    broker    engine    unified_sql    MON-152743
+    Ctn Clear Metrics
+    Ctn Config Engine    ${1}    ${1}    ${1000}
+    # We want all the services to be passive to avoid parasite checks during our test.
+    Ctn Set Services Passive    ${0}    service_.*
+    Ctn Config Broker    central
+    Ctn Config Broker    rrd
+    Ctn Config Broker    module    ${1}
+    Ctn Config BBDO3    1
+    Ctn Broker Config Log    central    core    error
+    Ctn Broker Config Log    central    tcp    error
+    Ctn Broker Config Log    central    sql    trace
+    Ctn Config Broker Sql Output    central    unified_sql
+    Ctn Config Broker Remove Rrd Output    central
+    Ctn Clear Retention
+
+    Ctn Prepare Partitions For Data Bin
+    ${start}    Get Current Date
+    Ctn Start Broker
+    Ctn Start Engine
+
+    Ctn Wait For Engine To Be Ready    ${start}    1
+
+    ${start}    Ctn Get Round Current Date
+    # Let's wait for one "INSERT INTO data_bin" to appear in stats.
+    Log To Console    Many service checks with 100 metrics each are processed.
+    FOR    ${i}    IN RANGE    ${1000}
+        Ctn Process Service Check Result With Metrics    host_1    service_${i+1}    1    warning${i}    100
+    END
+
+    Log To Console    We wait for at least one metric to be written in the database.
+    # Let's wait for all force checks to be in the storage database.
+    Connect To Database    pymysql    ${DBName}    ${DBUser}    ${DBPass}    ${DBHost}    ${DBPort}
+    FOR    ${i}    IN RANGE    ${500}
+        ${output}    Query
+        ...    SELECT COUNT(s.last_check) FROM metrics m LEFT JOIN index_data i ON m.index_id = i.id LEFT JOIN services s ON s.host_id = i.host_id AND s.service_id = i.service_id WHERE metric_name LIKE "metric_%%" AND s.last_check >= ${start}
+        IF    ${output[0][0]} >= 1    BREAK
+        Sleep    1s
+    END
+
+    Log To Console    Let's start some database manipulation...
+    Ctn Remove P2 From Data Bin
+    ${start}    Get Current Date
+
+    ${content}    Create List    errno=
+    FOR    ${i}    IN RANGE    ${6}
+	${result}    Ctn Find In Log With Timeout    ${centralLog}    ${start}    ${content}    10
+	IF    ${result}    BREAK
+    END
+
+    Log To Console    Let's recreate the p2 partition...
+    Ctn Add P2 To Data Bin
+
+    ${start}    Ctn Get Round Current Date
+    Ctn Process Service Check Result With Metrics    host_1    service_1    0    Last Output OK    100
+
+    Log To Console    Let's wait for the last service check to be in the database...
+    Connect To Database    pymysql    ${DBName}    ${DBUser}    ${DBPass}    ${DBHost}    ${DBPort}
+    FOR    ${i}    IN RANGE    ${120}
+        ${output}    Query    SELECT count(*) FROM data_bin WHERE ctime >= ${start} - 10
+	Log To Console    ${output}
+        IF    ${output[0][0]} == 100    BREAK
+        Sleep    1s
+    END
+    Log To Console    ${output}
+    Should Be True    ${output[0][0]} == 100
+
+
 *** Keywords ***
 Ctn Test Clean
     Ctn Stop engine

tests/resources/Broker.py

@@ -1689,7 +1689,7 @@ def ctn_get_service_index(host_id: int, service_id: int, timeout: int = 60):
                 my_id = [r['id'] for r in result]
                 if len(my_id) > 0:
                     logger.console(
-                            f"Index data {id} found for service {host_id}:{service_id}")
+                        f"Index data {id} found for service {host_id}:{service_id}")
                     return my_id[0]
                 time.sleep(2)
     logger.console(f"no index data found for service {host_id}:{service_id}")
@@ -2024,20 +2024,35 @@ def ctn_get_indexes_to_rebuild(count: int, nb_day=180):
             cursor.execute(sql)
             result = cursor.fetchall()
             for r in result:
-                if int(r['metric_id']) in ids:
-                    index_id = int(r['index_id'])
-                    logger.console(
-                        "building data for metric {} index_id {}".format(r['metric_id'], index_id))
-                    # We go back to 180 days with steps of 5 mn
-                    start = int(time.time() / 86400) * 86400 - \
-                        24 * 60 * 60 * nb_day
-                    value = int(r['metric_id']) // 2
-                    status_value = index_id % 3
-                    cursor.execute("DELETE FROM data_bin WHERE id_metric={} AND ctime >= {}".format(
-                        r['metric_id'], start))
-                    # We set the value to a constant on 180 days
-                    for i in range(0, 24 * 60 * 60 * nb_day, 60 * 5):
-                        cursor.execute(
+                index_id = int(r['index_id'])
+
+                if index_id not in ids:
+                    continue
+
+                # We must rebuild all the metrics of a given index.
+                if last_index != index_id and len(retval) == count:
+                    return retval
+
+                logger.console(
+                    f"building data for metric {r['metric_id']} index_id {index_id}")
+                # We go back to 180 days with steps of 5 mn
+                now = datetime.datetime.now()
+                dt = now.replace(hour=0, minute=0, second=0, microsecond=0)
+                start = dt - datetime.timedelta(days=nb_day)
+                start = int(start.timestamp())
+                logger.console(
+                    f">>>>>>>>>> start = {datetime.datetime.fromtimestamp(start)}")
+                value = int(r['metric_id']) // 2
+                status_value = index_id % 3
+                cursor.execute("DELETE FROM data_bin WHERE id_metric={} AND ctime >= {}".format(
+                    r['metric_id'], start))
+                # We set the value to a constant on 180 days
+                now = int(now.timestamp())
+                logger.console(
+                    f">>>>>>>>>> end = {datetime.datetime.fromtimestamp(now)}")
+                for i in range(start, now, 60 * 5):
+                    if i == start:
+                        logger.console(
                             "INSERT INTO data_bin (id_metric, ctime, value, status) VALUES ({},{},{},'{}')".format(
                                 r['metric_id'], start + i, value, status_value))
                     connection.commit()
@@ -2911,3 +2926,100 @@ def ctn_get_broker_log_info(port, log, timeout=TIMEOUT):
             except:
                 logger.console("gRPC server not ready")
     return str(res)
+
+
+def ctn_prepare_partitions_for_data_bin():
+    """
+    Create two partitions for the data_bin table.
+    The first one named p1 contains data with ctime older than now - 60.
+    The second one named p2 contains data with ctime older than now + 3600.
+    """
+    connection = pymysql.connect(host=DB_HOST,
+                                 user=DB_USER,
+                                 password=DB_PASS,
+                                 database=DB_NAME_STORAGE,
+                                 charset='utf8mb4',
+                                 cursorclass=pymysql.cursors.DictCursor)
+
+    now = int(time.time())
+    before = now - 60
+    after = now + 3600
+    with connection:
+        with connection.cursor() as cursor:
+            cursor.execute("DROP TABLE IF EXISTS data_bin")
+            sql = f"""CREATE TABLE `data_bin` (
+  `id_metric` int(11) DEFAULT NULL,
+  `ctime` int(11) DEFAULT NULL,
+  `value` float DEFAULT NULL,
+  `status` enum('0','1','2','3','4') DEFAULT NULL,
+  KEY `index_metric` (`id_metric`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`ctime`)
+(PARTITION `p1` VALUES LESS THAN ({before}) ENGINE = InnoDB,
+ PARTITION `p2` VALUES LESS THAN ({after}) ENGINE = InnoDB)"""
+            cursor.execute(sql)
+            connection.commit()
+
+
+def ctn_remove_p2_from_data_bin():
+    """
+    Remove the partition p2 from the data_bin table.
+    """
+    connection = pymysql.connect(host=DB_HOST,
+                                 user=DB_USER,
+                                 password=DB_PASS,
+                                 database=DB_NAME_STORAGE,
+                                 charset='utf8mb4',
+                                 cursorclass=pymysql.cursors.DictCursor)
+
+    with connection:
+        with connection.cursor() as cursor:
+            cursor.execute("ALTER TABLE data_bin DROP PARTITION p2")
+            connection.commit()
+
+
+def ctn_add_p2_to_data_bin():
+    """
+    Add the partition p2 the the data_bin table.
+    """
+    connection = pymysql.connect(host=DB_HOST,
+                                 user=DB_USER,
+                                 password=DB_PASS,
+                                 database=DB_NAME_STORAGE,
+                                 charset='utf8mb4',
+                                 cursorclass=pymysql.cursors.DictCursor)
+
+    after = int(time.time()) + 3600
+    with connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"ALTER TABLE data_bin ADD PARTITION (PARTITION p2 VALUES LESS THAN ({after}))")
+            connection.commit()
+
+
+def ctn_init_data_bin_without_partition():
+    """
+    Recreate the data_bin table without partition.
+    """
+    connection = pymysql.connect(host=DB_HOST,
+                                 user=DB_USER,
+                                 password=DB_PASS,
+                                 database=DB_NAME_STORAGE,
+                                 charset='utf8mb4',
+                                 cursorclass=pymysql.cursors.DictCursor)
+
+    now = int(time.time())
+    before = now - 60
+    after = now + 3600
+    with connection:
+        with connection.cursor() as cursor:
+            cursor.execute("DROP TABLE IF EXISTS data_bin")
+            sql = f"""CREATE TABLE `data_bin` (
+  `id_metric` int(11) DEFAULT NULL,
+  `ctime` int(11) DEFAULT NULL,
+  `value` float DEFAULT NULL,
+  `status` enum('0','1','2','3','4') DEFAULT NULL,
+  KEY `index_metric` (`id_metric`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1"""
+            cursor.execute(sql)
+            connection.commit()