Implemented measuring CPU usage for queries instrumented with query attributes. Added system table QUERY_PERF_COUNTER to output results of measurements.

Summary: Implemented measuring CPU usage for queries instrumented with query attributes. Added system table QUERY_PERF_COUNTER to output results of measurements.

Reviewed By: anirbanr-fb

Differential Revision: D7479522

fbshipit-source-id: ed64515

Author: Volodymyr Verovkin

client/mysql.cc

@@ -278,7 +278,7 @@ static int com_quit(String *str,char*),
            com_notee(String *str, char*), com_charset(String *str,char*),
            com_prompt(String *str, char*), com_delimiter(String *str, char*),
      com_warnings(String *str, char*), com_nowarnings(String *str, char*),
-     com_resetconnection(String *str, char*);
+     com_resetconnection(String *str, char*), com_attr(String *str,char*);
 
 #ifdef USE_POPEN
 static int com_nopager(String *str, char*), com_pager(String *str, char*),
@@ -329,6 +329,7 @@ typedef struct {
 } COMMANDS;
 
 static COMMANDS commands[] = {
+  { "setattr",'z', com_attr,   1, "Set query attribute."},
   { "?",      '?', com_help,   1, "Synonym for `help'." },
   { "clear",  'c', com_clear,  0, "Clear the current input statement."},
   { "connect",'r', com_connect,1,
@@ -4344,6 +4345,29 @@ com_shell(String *buffer MY_ATTRIBUTE((unused)),
 }
 #endif
 
+static int
+com_attr(String *buffer MY_ATTRIBUTE((unused)), char *line)
+{
+  static const char *delim = " \t";
+  char *ptr = nullptr;
+  char *buf = strdup(line);
+  const char *cmd __attribute__((unused)) = strtok_r(buf, delim, &ptr);
+  const char *key = strtok_r(nullptr, delim, &ptr);
+  const char *val = strtok_r(nullptr, delim, &ptr);
+
+  if (!key || !val)
+  {
+    put_info("Usage: setattr key value", INFO_ERROR);
+    free(buf);
+    return -1;
+  }
+
+  mysql_options4(&mysql, MYSQL_OPT_QUERY_ATTR_ADD,  key, val);
+
+  free(buf);
+  return 0;
+}
+
 
 static int
 com_print(String *buffer,char *line MY_ATTRIBUTE((unused)))

sql/CMakeLists.txt

@@ -119,6 +119,7 @@ SET(SQL_SHARED_SOURCES
   password.c
   procedure.cc 
   protocol.cc
+  query_tag_perf_counter.cc
   records.cc
   rpl_handler.cc
   scheduler.cc 

sql/handler.h

@@ -714,6 +714,7 @@ enum enum_schema_tables
   SCH_TRANSACTION_LIST,
   SCH_CONNECTION_ATTRIBUTES,
   SCH_QUERY_ATTRIBUTES,
+  SCH_QUERY_PERF_COUNTER,
   SCH_PROFILES,
   SCH_REFERENTIAL_CONSTRAINTS,
   SCH_PROCEDURES,

sql/query_tag_perf_counter.cc

@@ -0,0 +1,102 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include <unordered_map>
+#include <mutex>
+#include <memory>
+#include <deque>
+#include <boost/optional.hpp>
+#include <boost/lexical_cast.hpp>
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
+
+
+#include "query_tag_perf_counter.h"
+#include "sql_class.h"
+#include "sql_show.h"
+
+namespace qutils {
+
+namespace pt = boost::property_tree;
+
+static int64_t timespec_diff(const timespec& start, const timespec& stop);
+
+static std::mutex stats_mutex;
+static std::unordered_map<std::string, uint64_t> stats;
+
+query_tag_perf_counter::query_tag_perf_counter(THD* _thd)
+  : started(false), thd(_thd)
+{
+  if(!thd->query_type.empty() && thd->num_queries > 0)
+  {
+      this->started =
+        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &this->starttime) == 0;
+  }
+}
+
+query_tag_perf_counter::~query_tag_perf_counter()
+{
+  if(!this->started)
+    return;
+  struct timespec endtime;
+  if(clock_gettime(CLOCK_THREAD_CPUTIME_ID, &endtime) != 0)
+    return;
+
+  int64_t cputime = timespec_diff(this->starttime, endtime);
+  if(cputime < 0)
+    return; // skip if overflow
+
+  assert(thd->num_queries > 0);
+  int64_t cputime_per_query = cputime / thd->num_queries;
+
+  std::lock_guard<std::mutex> lock(stats_mutex);
+  stats[thd->query_type] += cputime_per_query;
+}
+
+int fill_query_tag_perf_counter(THD* thd, TABLE_LIST* tables, Item* cond)
+{
+  std::lock_guard<std::mutex> lock(stats_mutex);
+
+  DBUG_ENTER("fill_query_tag_perf_counter");
+
+  TABLE* table= tables->table;
+  for (const auto& row : stats)
+  {
+
+    restore_record(table, s->default_values);
+    Field **field = table->field;
+
+    const std::string& query_type = row.first;
+    uint64_t cpu_time = row.second;
+
+    field[0]->store(query_type.c_str(),
+        query_type.length(), system_charset_info);
+    field[1]->store(cpu_time, TRUE);
+
+    if (schema_table_store_record(thd, table))
+    {
+      DBUG_RETURN(-1);
+    }
+  }
+  DBUG_RETURN(0);
+}
+
+const uint query_type_field_length = 254;
+
+ST_FIELD_INFO query_tag_perf_fields_info[]=
+{
+  {"QUERY_TYPE", query_type_field_length, MYSQL_TYPE_STRING,
+		  0, 0, 0, SKIP_OPEN_TABLE},
+  {"CPU", MY_INT64_NUM_DECIMAL_DIGITS, MYSQL_TYPE_LONGLONG,
+		  0, 0, 0, SKIP_OPEN_TABLE},
+  {0, 0, MYSQL_TYPE_STRING, 0, 0, 0, SKIP_OPEN_TABLE}
+};
+
+static int64_t timespec_diff(const timespec& start, const timespec& stop)
+{
+  const int64_t sec = stop.tv_sec - start.tv_sec;
+  const int64_t nsec = stop.tv_nsec - start.tv_nsec;
+  const int64_t diff = sec * 1000000000LL + nsec;
+  return diff;
+}
+} // namespace qutils
+

sql/query_tag_perf_counter.h

@@ -0,0 +1,32 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#ifndef QUERY_TAG_PERF_COUNTER_H_
+#define QUERY_TAG_PERF_COUNTER_H_
+
+#include <ctime>
+#include <string>
+
+class Item;
+class THD;
+struct TABLE_LIST;
+typedef struct st_field_info ST_FIELD_INFO;
+
+namespace qutils {
+
+extern ST_FIELD_INFO query_tag_perf_fields_info[];
+extern int
+fill_query_tag_perf_counter(THD* thd, TABLE_LIST* tables, Item* cond);
+
+class query_tag_perf_counter {
+public:
+  query_tag_perf_counter(THD *thd);
+  ~query_tag_perf_counter();
+private:
+  struct timespec starttime; // query start timestamp
+  bool started; // query_info attribute detected
+  THD* thd; // reference to outer THD
+};
+
+} // namespace qutils
+
+#endif /* QUERY_TAG_PERF_COUNTER_H_ */

sql/sql_class.cc

@@ -65,6 +65,12 @@
 
 #include <mysql/psi/mysql_statement.h>
 
+#include <boost/optional.hpp>
+#include <boost/lexical_cast.hpp>
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
+#include <sstream>
+
 #ifdef TARGET_OS_LINUX
 #include <sys/syscall.h>
 #endif // TARGET_OS_LINUX
@@ -5536,7 +5542,6 @@ bool THD::skip_unique_check()
   return rli_slave && rli_slave->get_skip_unique_check();
 }
 
-
 /**
   This function selects which session tracker to use.  If a Srv_session
   is currently attached to this connection we want to redirect all session
@@ -5548,3 +5553,60 @@ Session_tracker* THD::get_tracker() {
       : &session_tracker;
 }
 
+static std::string net_read_str(const char **ptr)
+{
+  size_t len = net_field_length((uchar**)ptr);
+  const char *str = *ptr;
+  *ptr += len;
+  return std::string(str, len);
+}
+
+void THD::set_query_attrs(const char *attrs, size_t length)
+{
+  query_attrs_string = std::string(attrs, length);
+  const char *ptr = query_attrs_string.c_str();
+  const char *end = ptr + length;
+
+  query_attrs_map.clear();
+  while (ptr < end)
+  {
+    std::string key = net_read_str(&ptr);
+    std::string value = net_read_str(&ptr);
+    query_attrs_map[key] = value;
+  }
+
+}
+
+int THD::parse_query_info_attr()
+{
+  static const std::string query_info_key = "query_info";
+
+  auto it = this->query_attrs_map.find(query_info_key);
+  if (it == this->query_attrs_map.end())
+    return 0;
+  ptree root;
+  try
+  {
+    std::istringstream query_info_attr(it->second);
+    boost::property_tree::read_json(query_info_attr, root);
+  }
+  catch(const boost::property_tree::json_parser::json_parser_error& e)
+  {
+    return -1; // invalid json
+  }
+  try
+  {
+    boost::optional<std::string> trace_id =
+        root.get_optional<std::string>("traceid");
+    if (trace_id)
+      this->trace_id = *trace_id;
+    this->query_type = root.get<std::string>("query_type");
+    this->num_queries = root.get<uint64_t>("num_queries");
+  }
+  catch(const boost::property_tree::ptree_error& e)
+  {
+    return -1; // invalid key or value
+  }
+  return 0;
+
+}

sql/sql_class.h

@@ -4634,19 +4634,20 @@ class THD :public MDL_context_owner,
 
   void mark_transaction_to_rollback(bool all);
 
-  void set_query_attrs(const CSET_STRING &arg)
-  {
-    query_attrs_string= arg;
-    set_attrs_map(query_attrs_string.str(), query_attrs_string.length(),
-        query_attrs_map, true);
-  }
+  void set_query_attrs(const char *attrs, size_t length);
+  int parse_query_info_attr();
   void reset_query_attrs()
   {
-    query_attrs_string= CSET_STRING();
+    query_attrs_string.clear();
     query_attrs_map.clear();
     trace_id.clear();
+    num_queries = 0;
+    query_type.clear();
+  }
+  inline const char *query_attrs() const
+  {
+    return query_attrs_string.c_str();
   }
-  inline char *query_attrs() const { return query_attrs_string.str(); }
   inline uint32 query_attrs_length() const
   {
     return query_attrs_string.length();
@@ -4725,8 +4726,6 @@ class THD :public MDL_context_owner,
 #endif
 
 private:
-  CSET_STRING query_attrs_string;
-
   /** The current internal error handler for this thread, or NULL. */
   Internal_error_handler *m_internal_handler;
 
@@ -4859,6 +4858,9 @@ class THD :public MDL_context_owner,
   std::shared_ptr<utils::PerfCounter> query_perf;
   std::string trace_id;
   uint64_t pc_val;
+  std::string query_attrs_string;
+  uint64_t num_queries;
+  std::string query_type;
 
   void copy_client_charset_settings(const THD* other) {
     variables.character_set_client= other->variables.character_set_client;

sql/sql_parse.cc

@@ -123,6 +123,7 @@
 #ifndef EMBEDDED_LIBRARY
 #include "perf_counters.h"
 #endif
+#include "query_tag_perf_counter.h"
 
 #ifdef HAVE_JEMALLOC
 #ifndef EMBEDDED_LIBRARY
@@ -1584,7 +1585,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd, char* packet,
       Save query attributes.
      */
     size_t attrs_len= net_field_length((uchar**) &packet_ptr);
-    thd->set_query_attrs(CSET_STRING(packet_ptr, attrs_len, thd->charset()));
+    thd->set_query_attrs(packet_ptr, attrs_len);
 
     auto bytes_to_skip = attrs_len + net_length_size(attrs_len);
     packet_length -= bytes_to_skip;
@@ -1593,12 +1594,8 @@ bool dispatch_command(enum enum_server_command command, THD *thd, char* packet,
 
 #ifndef EMBEDDED_LIBRARY
   // do collection of samples for passed in "traceid"
-  std::string traceId;
-  std::unordered_map<std::string, std::string>::const_iterator tid_iter;
-  if (!thd->query_attrs_map.empty() &&
-      (tid_iter= thd->query_attrs_map.find("traceid")) !=
-       thd->query_attrs_map.end()) {
-    thd->trace_id = tid_iter->second;
+  thd->parse_query_info_attr();
+  if (!thd->trace_id.empty()) {
     if (!thd->query_perf) {
       thd->query_perf= pc_factory->makeSharedPerfCounter(
         utils::PerfCounterMode::PCM_THREAD,
@@ -1868,6 +1865,9 @@ bool dispatch_command(enum enum_server_command command, THD *thd, char* packet,
     size_t query_len= packet_length;
     if (alloc_query(thd, query_ptr, query_len))
       break;					// fatal error is set
+
+    qutils::query_tag_perf_counter counter(thd);
+
     MYSQL_QUERY_START(thd->query(), thd->thread_id,
                       (char *) (thd->db ? thd->db : ""),
                       &thd->security_ctx->priv_user[0],

sql/sql_show.cc

@@ -67,6 +67,8 @@
 using std::max;
 using std::min;
 
+#include "query_tag_perf_counter.h"
+
 #define STR_OR_NIL(S) ((S) ? (S) : "<nil>")
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
@@ -9523,6 +9525,9 @@ ST_SCHEMA_TABLE schema_tables[]=
   {"QUERY_ATTRIBUTES", query_attrs_fields_info,
    create_schema_table, fill_schema_query_attrs,
    NULL, NULL, -1, -1, false, 0},
+  {"QUERY_PERF_COUNTER", qutils::query_tag_perf_fields_info,
+    create_schema_table, qutils::fill_query_tag_perf_counter,
+    NULL, NULL, -1, -1, false, 0},
   {"PROFILING", query_profile_statistics_info, create_schema_table,
     fill_query_profile_statistics_info, make_profile_table_for_show, 
     NULL, -1, -1, false, 0},