Merge pull request #4920 from wazir-ahmed/purge-stats-query-digest

feat(stats): Purge query digest based on last seen

Author: renecannao

include/gen_utils.h

@@ -516,6 +516,9 @@ bool mywildcmp(const char *p, const char *str);
 std::string trim(const std::string& s);
 char* escape_string_single_quotes_and_backslashes(char* input, bool free_it);
 const char* escape_string_backslash_spaces(const char* input);
+time_t monotonic_time_to_realtime(time_t mt);
+time_t realtime_to_monotonic_time(time_t rt);
+
 std::string strip_schema_from_query(const char* query, const char* schema,
                                     const std::vector<std::string>& tables = {}, bool ansi_quotes = false);
 /**

include/query_processor.h

@@ -81,6 +81,7 @@ class QP_query_digest_stats {
 		unsigned long long t, unsigned long long n, unsigned long long ra, unsigned long long rs,
 		unsigned long long cnt = 1
 	);
+	void merge(const QP_query_digest_stats *other);
 	~QP_query_digest_stats();
 	char *get_digest_text(const umap_query_digest_text *digest_text_umap) const;
 	char **get_row(umap_query_digest_text *digest_text_umap, query_digest_stats_pointers_t *qdsp);
@@ -345,7 +346,7 @@ class Query_Processor {
 		uint32_t max_window
 	);
 	void get_query_digests_reset(umap_query_digest* uqd, umap_query_digest_text* uqdt);
-	unsigned long long purge_query_digests(bool async_purge, bool parallel, char** msg);
+	unsigned long long purge_query_digests(bool async_purge, bool parallel, time_t last_seen = 0);
 
 	void save_query_rules(SQLite3_result* resultset);
 
@@ -457,8 +458,8 @@ class Query_Processor {
 	DEFINE_HAS_METHOD_STRUCT(query_parser_first_comment_extended);
 	DEFINE_HAS_METHOD_STRUCT(process_query_extended);
 
-	unsigned long long purge_query_digests_async(char** msg);
-	unsigned long long purge_query_digests_sync(bool parallel);
+	unsigned long long purge_query_digests_async(time_t last_seen = 0);
+	unsigned long long purge_query_digests_sync(bool parallel, time_t last_seen = 0);
 
 	/**
 	 * @brief Searches for a matching rule in the supplied map, returning the destination hostgroup.

lib/Admin_Bootstrap.cpp

@@ -143,9 +143,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;

lib/Admin_FlushVariables.cpp

@@ -118,9 +118,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;

lib/Admin_Handler.cpp

@@ -120,9 +120,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;
@@ -327,6 +324,17 @@ const std::vector<std::string> LOAD_COREDUMP_FROM_MEMORY = {
 	"LOAD COREDUMP TO RUNTIME" ,
 	"LOAD COREDUMP TO RUN" };
 
+const std::vector<std::string> CMD_PREFIX_PURGE_QUERY_DIGESTS = {
+	"PURGE TABLE stats.stats_mysql_query_digest TO ",
+	"PURGE TABLE stats_mysql_query_digest TO ",
+	"PURGE stats.stats_mysql_query_digest TO ",
+	"PURGE stats_mysql_query_digest TO ",
+	"PURGE TABLE stats.stats_pgsql_query_digest TO ",
+	"PURGE TABLE stats_pgsql_query_digest TO ",
+	"PURGE stats.stats_pgsql_query_digest TO ",
+	"PURGE stats_pgsql_query_digest TO ",
+};
+
 extern unordered_map<string,std::tuple<string, vector<string>, vector<string>>> load_save_disk_commands;
 
 // Helper function: Extract pattern from c.relname OPERATOR or LIKE clause
@@ -518,6 +526,17 @@ bool is_admin_command_or_alias(const std::vector<std::string>& cmds, char *query
 	return false;
 }
 
+const char * match_command_prefix(const std::vector<std::string>& cmd_prefix, char *query, int query_len) {
+	for (auto &prefix : cmd_prefix) {
+		if ((unsigned int) query_len >= prefix.length()
+			&& !strncasecmp(prefix.c_str(), query, prefix.length()))
+		{
+			return prefix.c_str();
+		}
+	}
+
+	return nullptr;
+}
 
 
 template <typename S>
@@ -553,6 +572,38 @@ bool FlushCommandWrapper(S* sess, const string& modname, char *query_no_space, i
 	return false;
 }
 
+std::tuple<bool, enum SERVER_TYPE, time_t> parse_command_purge_query_digests(char *query, int query_len) {
+	bool match = false;
+	enum SERVER_TYPE server_type = SERVER_TYPE_MYSQL;
+	time_t last_seen = 0;
+
+	const char *prefix = match_command_prefix(CMD_PREFIX_PURGE_QUERY_DIGESTS, query, query_len);
+	if (prefix) {
+		match = true;
+
+		if (strstr(prefix, "_pgsql_") != nullptr) {
+			server_type = SERVER_TYPE_PGSQL;
+		}
+
+		// parse timestamp
+		mf_unique_ptr<char> ts_str(strdup(query + strlen(prefix)));
+		char *ts_end = nullptr;
+		long long ts = strtoll(trim_spaces_in_place(ts_str.get()), &ts_end, 10);
+
+		// ts_str should only contain digits and respresent a valid timestamp
+		if ((*ts_end == 0) && (ts > 0)) {
+			last_seen = realtime_to_monotonic_time(ts);
+			if (last_seen <= 0) {
+				// valid timestamp, but older than the monotonic clock epoch (system
+				// boot): no entry can match, make the command a successful no-op
+				last_seen = 1;
+			}
+		}
+	}
+
+	return std::make_tuple(match, server_type, last_seen);
+}
+
 template <typename S>
 bool admin_handler_command_kill_pgsql_connection(uint32_t session_thd_id, S* sess, ProxySQL_Admin* pa) {
 	proxy_debug(PROXY_DEBUG_ADMIN, 4, "Trying to kill pgsql session %u\n", session_thd_id);
@@ -3442,10 +3493,8 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 					SPA->admindb->execute("DELETE FROM stats.stats_mysql_query_digest_reset");
 					SPA->vacuum_stats(true);
 					// purge the digest map, asynchronously, in single thread
-					char *msg = NULL;
-					int r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, &msg);
-					SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-					free(msg);
+					int r1 = GloMyQPro->purge_query_digests(true, false);
+					SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 					run_query=false;
 					goto __run_query;
 				}
@@ -3478,16 +3527,45 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 					SPA->admindb->execute("DELETE FROM stats.stats_pgsql_query_digest_reset");
 					SPA->vacuum_stats(true);
 					// purge the digest map, asynchronously, in single thread
-					char* msg = NULL;
-					int r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(true, false, &msg);
-					SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-					free(msg);
+					int r1 = GloPgQPro->purge_query_digests(true, false);
+					SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 					run_query = false;
 					goto __run_query;
 				}
 			}
 		}
 	}
+
+	// handles 'PURGE stats_mysql_query_digest TO <value>'.
+	// any entry in stats_mysql_query_digest where last_seen is less than <value> will be deleted.
+	if (!strncasecmp("PURGE ", query_no_space, strlen("PURGE "))
+		&& sess->session_type == PROXYSQL_SESSION_ADMIN
+	) {
+		auto result = parse_command_purge_query_digests(query_no_space, query_no_space_length);
+		bool match = std::get<0>(result);
+
+		if (match == true) {
+			int ret = 0;
+			enum SERVER_TYPE type = std::get<1>(result);
+			time_t last_seen = std::get<2>(result);
+
+			if (last_seen > 0) {
+				if (type == SERVER_TYPE_MYSQL) {
+					ret = GloMyQPro->purge_query_digests(true, false, last_seen);
+				} else if (type == SERVER_TYPE_PGSQL) {
+					ret = GloPgQPro->purge_query_digests(true, false, last_seen);
+				}
+
+				pa->send_ok_msg_to_client(sess, NULL, ret, query_no_space);
+			} else {
+				pa->send_error_msg_to_client(sess, "Invalid timestamp");
+			}
+
+			run_query = false;
+			goto __run_query;
+		}
+	}
+
 #ifdef DEBUG
 	/**
 	 * @brief Handles the 'PROXYSQL_SIMULATOR' command. Performing the operation specified in the payload

lib/ProxySQL_Admin.cpp

@@ -123,9 +123,6 @@ struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;

lib/ProxySQL_Admin_Tests.cpp

@@ -1,3 +1,4 @@
+#include <ctime>
 #include <iostream>     // std::cout
 #include <sstream>      // std::stringstream
 #include <fstream>
@@ -56,12 +57,12 @@ bool ProxySQL_Test___Refresh_MySQL_Variables(unsigned int cnt) {
 }
 
 template <enum SERVER_TYPE ST>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg) {
+int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, time_t last_seen) {
 	int r = 0;
 	if constexpr (ST == SERVER_TYPE_MYSQL) {
-		r = GloMyQPro->purge_query_digests(async_purge, parallel, msg);
+		r = GloMyQPro->purge_query_digests(async_purge, parallel, last_seen);
 	} else if constexpr (ST == SERVER_TYPE_PGSQL) {
-		r = GloPgQPro->purge_query_digests(async_purge, parallel, msg);
+		r = GloPgQPro->purge_query_digests(async_purge, parallel, last_seen);
 	}
 	return r;
 }
@@ -147,5 +148,5 @@ int ProxySQL_Test___GenerateRandomQueryInDigestTable(int n) {
 	return n*1000;
 }
 
-template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(bool async_purge, bool parallel, char** msg);
-template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(bool async_purge, bool parallel, char** msg);
+template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(bool async_purge, bool parallel, time_t last_seen);
+template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(bool async_purge, bool parallel, time_t last_seen);

lib/ProxySQL_Admin_Tests2.cpp

@@ -46,7 +46,7 @@ static void init_rand_del() {
 int ProxySQL_Test___GetDigestTable(bool reset, bool use_swap);
 bool ProxySQL_Test___Refresh_MySQL_Variables(unsigned int cnt);
 template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
+int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, time_t last_seen);
 int ProxySQL_Test___GenerateRandomQueryInDigestTable(int n);
 
 void ProxySQL_Admin::map_test_mysql_firewall_whitelist_rules_cleanup() {
@@ -1179,21 +1179,20 @@ void ProxySQL_Admin::ProxySQL_Test_Handler(ProxySQL_Admin *SPA, S* sess, char *q
 				break;
 			case 4:
 				// purge the digest map, synchronously, in single thread
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, false, NULL);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, false, test_arg1);
 				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 5:
 				// purge the digest map, synchronously, in multiple threads
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, true, NULL);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, true, test_arg1);
 				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 6:
 				// purge the digest map, asynchronously, in single thread
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, &msg);
-				SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-				free(msg);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, test_arg1);
+				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 7:

lib/QP_query_digest_stats.cpp

@@ -1,3 +1,4 @@
+#include "gen_utils.h"
 #include "query_processor.h"
 
 // reverse:  reverse string s in place
@@ -79,6 +80,26 @@ void QP_query_digest_stats::add_time(
 	}
 	last_seen=n;
 }
+// Merges the counters of 'other' into this entry. Used when reconciling stats
+// collected while a purge operation was running (see purge_query_digests_async()).
+void QP_query_digest_stats::merge(const QP_query_digest_stats *other) {
+	count_star += other->count_star;
+	sum_time += other->sum_time;
+	rows_affected += other->rows_affected;
+	rows_sent += other->rows_sent;
+	if (other->min_time && (min_time == 0 || other->min_time < min_time)) {
+		min_time = other->min_time;
+	}
+	if (other->max_time > max_time) {
+		max_time = other->max_time;
+	}
+	if (other->first_seen && (first_seen == 0 || other->first_seen < first_seen)) {
+		first_seen = other->first_seen;
+	}
+	if (other->last_seen > last_seen) {
+		last_seen = other->last_seen;
+	}
+}
 QP_query_digest_stats::~QP_query_digest_stats() {
 	if (digest_text) {
 		free(digest_text);
@@ -151,16 +172,13 @@ char **QP_query_digest_stats::get_row(umap_query_digest_text *digest_text_umap,
 	my_itoa(qdsp->count_star, count_star);
 	pta[5]=qdsp->count_star;
 
-	time_t __now;
-	time(&__now);
-	unsigned long long curtime=monotonic_time();
 	time_t seen_time;
-	seen_time= __now - curtime/1000000 + first_seen/1000000;
+	seen_time=monotonic_time_to_realtime(first_seen);
 	//sprintf(qdsp->first_seen,"%ld", seen_time);
 	my_itoa(qdsp->first_seen, seen_time);
 	pta[6]=qdsp->first_seen;
 
-	seen_time= __now - curtime/1000000 + last_seen/1000000;
+	seen_time=monotonic_time_to_realtime(last_seen);
 	//sprintf(qdsp->last_seen,"%ld", seen_time);
 	my_itoa(qdsp->last_seen, seen_time);
 	pta[7]=qdsp->last_seen;