Second iteration on PostgreSQL monitoring POC

This commit packs multiple fixes/improvements:

- Added READONLY support for PostgreSQL.
- Major rework for queries and statements used on monitoring tables:
    + Checks/Actions rewrote for single instance checks.
    + Reuse of prepared statements instance of re-preparation.
- Fixed missing error handling in connection creation state machine.
- Added support for configurable batching in scheduler thread, via
  '*_interval_window' variables. These variables allows to define the
  burstiness of the scheduling within the processing interval.
- Several fixes for 'poll' timeout computation for worker threads.
- Fixed edge cases for current interval detection.
- Reduced deviation in scheduling intervals computation.
- Refactored and simplified connection event handling.
- Honor '-M' argument for disabling monitoring support.

include/PgSQL_Monitor.hpp

@@ -14,6 +14,8 @@
 
 #define MONITOR_SQLITE_TABLE_PGSQL_SERVER_PING_LOG "CREATE TABLE pgsql_server_ping_log ( hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , time_start_us INT NOT NULL DEFAULT 0 , ping_success_time_us INT DEFAULT 0 , ping_error VARCHAR , PRIMARY KEY (hostname, port, time_start_us))"
 
+#define MONITOR_SQLITE_TABLE_PGSQL_SERVER_READ_ONLY_LOG "CREATE TABLE pgsql_server_read_only_log ( hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , time_start_us INT NOT NULL DEFAULT 0 , success_time_us INT DEFAULT 0 , read_only INT DEFAULT 1 , error VARCHAR , PRIMARY KEY (hostname, port, time_start_us))"
+
 #define MONITOR_SQLITE_TABLE_PGSQL_SERVERS "CREATE TABLE pgsql_servers (hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 3306 , status INT CHECK (status IN (0, 1, 2, 3, 4)) NOT NULL DEFAULT 0 , use_ssl INT CHECK (use_ssl IN(0,1)) NOT NULL DEFAULT 0 , PRIMARY KEY (hostname, port) )"
 
 #define MONITOR_SQLITE_TABLE_PROXYSQL_SERVERS "CREATE TABLE proxysql_servers (hostname VARCHAR NOT NULL , port INT NOT NULL DEFAULT 6032 , weight INT CHECK (weight >= 0) NOT NULL DEFAULT 0 , comment VARCHAR NOT NULL DEFAULT '' , PRIMARY KEY (hostname, port) )"
@@ -35,6 +37,8 @@ struct PgSQL_Monitor {
 	uint64_t connect_check_OK { 0 };
 	uint64_t ping_check_ERR { 0 };
 	uint64_t ping_check_OK { 0 };
+	uint64_t readonly_check_ERR { 0 };
+	uint64_t readonly_check_OK { 0 };
 	///////////////////////////////////////////////////////////////////////////
 
 	std::vector<table_def_t> tables_defs_monitor {
@@ -45,7 +49,11 @@ struct PgSQL_Monitor {
 		{
 			const_cast<char*>("pgsql_server_ping_log"),
 			const_cast<char*>(MONITOR_SQLITE_TABLE_PGSQL_SERVER_PING_LOG)
-		}
+		},
+		{
+			const_cast<char*>("pgsql_server_read_only_log"),
+			const_cast<char*>(MONITOR_SQLITE_TABLE_PGSQL_SERVER_READ_ONLY_LOG)
+		},
 	};
 
 	std::vector<table_def_t> tables_defs_monitor_internal {

include/PgSQL_Thread.h

@@ -801,16 +801,19 @@ class PgSQL_Threads_Handler
 
 		int monitor_history;
 		int monitor_connect_interval;
+		int monitor_connect_interval_window;
 		int monitor_connect_timeout;
 		//! Monitor ping interval. Unit: 'ms'.
 		int monitor_ping_interval;
+		int monitor_ping_interval_window;
 		int monitor_ping_max_failures;
 		//! Monitor ping timeout. Unit: 'ms'.
 		int monitor_ping_timeout;
 		//! Monitor aws rds topology discovery interval. Unit: 'one discovery check per X monitor_read_only checks'.
 		int monitor_aws_rds_topology_discovery_interval;
 		//! Monitor read only timeout. Unit: 'ms'.
 		int monitor_read_only_interval;
+		int monitor_read_only_interval_window;
 		//! Monitor read only timeout. Unit: 'ms'.
 		int monitor_read_only_timeout;
 		int monitor_read_only_max_timeout_count;

include/proxysql_glovars.hpp

@@ -86,7 +86,8 @@ class ProxySQL_GlobalVariables {
 		unsigned long long start_time;
 		bool gdbg;
 		bool nostart;
-		bool monitor;
+		bool my_monitor;
+		bool pg_monitor;
 		bool version_check;
 #ifdef SO_REUSEPORT
 		bool reuseport;

include/proxysql_structs.h

@@ -648,6 +648,7 @@ enum PROXYSQL_MYSQL_ERR {
 	ER_PROXYSQL_AWS_HEALTH_CHECK_TIMEOUT              = 9018,
 	ER_PROXYSQL_SRV_NULL_REPLICATION_LAG              = 9019,
 	ER_PROXYSQL_CONNECT_TIMEOUT                       = 9020,
+	ER_PROXYSQL_READONLY_TIMEOUT                      = 9021,
 };
 
 enum proxysql_session_type {
@@ -1084,11 +1085,14 @@ __thread int pgsql_thread___query_processor_regex;
 __thread bool pgsql_thread___monitor_enabled;
 __thread int pgsql_thread___monitor_history;
 __thread int pgsql_thread___monitor_connect_interval;
+__thread int pgsql_thread___monitor_connect_interval_window;
 __thread int pgsql_thread___monitor_connect_timeout;
 __thread int pgsql_thread___monitor_ping_interval;
+__thread int pgsql_thread___monitor_ping_interval_window;
 __thread int pgsql_thread___monitor_ping_max_failures;
 __thread int pgsql_thread___monitor_ping_timeout;
 __thread int pgsql_thread___monitor_read_only_interval;
+__thread int pgsql_thread___monitor_read_only_interval_window;
 __thread int pgsql_thread___monitor_read_only_timeout;
 __thread int pgsql_thread___monitor_read_only_max_timeout_count;
 __thread int pgsql_thread___monitor_threads;
@@ -1371,11 +1375,14 @@ extern __thread int pgsql_thread___query_processor_regex;
 extern __thread bool pgsql_thread___monitor_enabled;
 extern __thread int pgsql_thread___monitor_history;
 extern __thread int pgsql_thread___monitor_connect_interval;
+extern __thread int pgsql_thread___monitor_connect_interval_window;
 extern __thread int pgsql_thread___monitor_connect_timeout;
 extern __thread int pgsql_thread___monitor_ping_interval;
+extern __thread int pgsql_thread___monitor_ping_interval_window;
 extern __thread int pgsql_thread___monitor_ping_max_failures;
 extern __thread int pgsql_thread___monitor_ping_timeout;
 extern __thread int pgsql_thread___monitor_read_only_interval;
+extern __thread int pgsql_thread___monitor_read_only_interval_window;
 extern __thread int pgsql_thread___monitor_read_only_timeout;
 extern __thread int pgsql_thread___monitor_read_only_max_timeout_count;
 extern __thread int pgsql_thread___monitor_threads;

lib/PgSQL_Connection.cpp

@@ -1841,7 +1841,12 @@ PG_ASYNC_ST PgSQL_Connection::handler(short event) {
 					if (error_category != PGSQL_ERROR_CATEGORY::ERRCATEGORY_SYNTAX_ERROR &&
 						error_category != PGSQL_ERROR_CATEGORY::ERRCATEGORY_STATUS &&
 						error_category != PGSQL_ERROR_CATEGORY::ERRCATEGORY_DATA_ERROR) {
-						proxy_error("Error: %s, Multi-Statement: %d\n", get_error_code_with_message().c_str(), processing_multi_statement);
+						proxy_error(
+							"Error: %s, Cat: %hhu, Multi-Statement: %d\n",
+							get_error_code_with_message().c_str(),
+							static_cast<uint8_t>(error_category),
+							processing_multi_statement
+						);
 					}
 					//}
 					NEXT_IMMEDIATE(ASYNC_USE_RESULT_CONT);
@@ -2738,6 +2743,7 @@ bool PgSQL_Connection::IsServerOffline() {
 bool PgSQL_Connection::is_connection_in_reusable_state() const {
 	const PGTransactionStatusType txn_status = PQtransactionStatus(pgsql_conn);
 	const bool conn_usable = !(txn_status == PQTRANS_UNKNOWN || txn_status == PQTRANS_ACTIVE);
+	proxy_info("pgsql_conn '%p', txn_status %d\n", pgsql_conn, txn_status);
 	assert(!(conn_usable == false && is_error_present() == false));
 	return conn_usable;
 }

lib/PgSQL_Data_Stream.cpp

@@ -1414,4 +1414,4 @@ int PgSQL_Data_Stream::buffer2array() {
 		queueIN.pkt.ptr = NULL;
 	}
 	return ret;
-}
+}