BUG#35410360 Admin API X Protocol Port changes

The AdminAPI stores in the Metadata schema information about each
instance's endpoints, including the X Plugin port endpoint. This
particular endpoint is used by Router when the GR notifications are
enabled.

If, for some reason, the X Port is changed on a Cluster member and the
member is restarted, the information stored in the Metadata will be
outdated and the AdminAPI is not detecting it or attempting to correct
it. This can lead to MySQL Router not being able to connect for the GR
notifications or connect to a wrong Cluster.

This patch fixes that by including a check for a possible change of the
X Port in Cluster.status() and the possibility of updating the Metadata
schema with Cluster.rescan().

Change-Id: I46c96fd397b1f7dd48c325448adb2f1437385da4

Author: miguelaraujo

modules/adminapi/cluster/rescan.cc

@@ -31,6 +31,7 @@
 #include "modules/adminapi/common/metadata_storage.h"
 #include "modules/adminapi/common/preconditions.h"
 #include "modules/adminapi/common/validations.h"
+#include "mysqlshdk/include/scripting/types.h"
 #include "mysqlshdk/include/shellcore/console.h"
 #include "mysqlshdk/libs/config/config.h"
 #include "mysqlshdk/libs/config/config_server_handler.h"
@@ -43,6 +44,8 @@ namespace mysqlsh {
 namespace dba {
 namespace cluster {
 
+constexpr const char *k_xplugin_disabled = "<disabled>";
+
 Rescan::Rescan(const Rescan_options &options, Cluster_impl *cluster)
     : m_options(options), m_cluster(cluster) {
   assert(cluster);
@@ -251,38 +254,73 @@ void Rescan::check_mismatched_hostnames_addresses(
 
         auto address = instance->get_canonical_address();
 
+        std::optional<int> xport = instance->get_xport();
+
         auto res = instance->query(
             "select @@hostname, @@report_host, @@port, @@report_port");
         auto row = res->fetch_one();
 
         log_debug(
             "Hostname metadata check for %s: address=%s @@hostname=%s "
-            "@@report_host=%s @@port=%s @@report_port=%s md.instance_id=%u "
-            "md.endpoint=%s md.address=%s md.label=%s md.xendpoint=%s "
-            "md.grendpoint=%s member_id=%s member_host=%s member_port=%i",
+            "@@report_host=%s @@port=%s @@report_port=%s @@mysqlx_port=%s "
+            " md.instance_id=%u md.endpoint=%s md.address=%s md.label=%s "
+            "md.xendpoint=%s md.grendpoint=%s member_id=%s member_host=%s "
+            "member_port=%i",
             instance->get_uuid().c_str(), address.c_str(),
             row->get_as_string(0).c_str(), row->get_as_string(1).c_str(),
             row->get_as_string(2).c_str(), row->get_as_string(3).c_str(),
+            xport.has_value() ? std::to_string(*xport).c_str() : "null",
             info.first.id, info.first.endpoint.c_str(),
             info.first.address.c_str(), info.first.label.c_str(),
             info.first.xendpoint.c_str(), info.first.grendpoint.c_str(),
             info.second.uuid.c_str(), info.second.host.c_str(),
             info.second.port);
 
+        bool xport_equal = false;
+        bool address_equal = false;
+        std::string x_address;
+
+        auto canonical_hostname = instance->get_canonical_hostname();
+
+        if (xport.has_value()) {
+          x_address =
+              mysqlshdk::utils::make_host_and_port(canonical_hostname, *xport);
+        }
+
+        if (mysqlshdk::utils::are_endpoints_equal(info.first.xendpoint,
+                                                  x_address)) {
+          xport_equal = true;
+        }
+
         if (grendpoint_valid &&
             mysqlshdk::utils::are_endpoints_equal(info.first.endpoint,
                                                   address) &&
             mysqlshdk::utils::are_endpoints_equal(info.first.address,
                                                   address)) {
-          return true;
+          address_equal = true;
         }
 
-        instances->emplace_back(shcore::Value(
-            shcore::make_dict("member_id", shcore::Value(instance->get_uuid()),
-                              "id", shcore::Value(info.first.id), "label",
-                              shcore::Value(info.first.label), "host",
-                              shcore::Value(instance->get_canonical_address()),
-                              "old_host", shcore::Value(info.first.endpoint))));
+        if (xport_equal && address_equal) return true;
+
+        shcore::Dictionary_t dict = shcore::make_dict();
+
+        dict->set("member_id", shcore::Value(instance->get_uuid()));
+        dict->set("id", shcore::Value(info.first.id));
+        dict->set("label", shcore::Value(info.first.label));
+        dict->set("host", shcore::Value(instance->get_canonical_address()));
+        dict->set(
+            "xport_endpoint",
+            shcore::Value(x_address.empty() ? k_xplugin_disabled : x_address));
+
+        instances->emplace_back(shcore::Value(dict));
+
+        if (!address_equal) {
+          dict->set("old_host", shcore::Value(info.first.endpoint));
+        }
+
+        if (!xport_equal) {
+          dict->set("old_xport_endpoint", shcore::Value(info.first.xendpoint));
+        }
 
         return true;
       },
@@ -615,6 +653,7 @@ void Rescan::update_metadata_for_instances(
   for (const auto &instance : *instance_list) {
     auto instance_map = instance.as_map();
     std::string label;
+    std::string xport_endpoint_str;
 
     std::string instance_address = instance_map->get_string("host");
     // Report that an obsolete instance was found (in the metadata).
@@ -628,8 +667,7 @@ void Rescan::update_metadata_for_instances(
           instance_map->get_string("member_id").c_str()));
     } else {
       // member address changed
-      if (instance_map->get_string("old_host") !=
-          instance_map->get_string("host")) {
+      if (instance_map->has_key("old_host")) {
         console->print_info(shcore::str_format(
             "The instance '%s' is part of the cluster but its "
             "reported address has changed. "
@@ -641,8 +679,22 @@ void Rescan::update_metadata_for_instances(
 
       // update instance label if it's the default value
       if (instance_map->get_string("label") ==
-          instance_map->get_string("old_host"))
+          instance_map->get_string("old_host")) {
         label = instance_map->get_string("host");
+      }
+
+      // xport has changed
+      if (instance_map->has_key("old_xport_endpoint")) {
+        xport_endpoint_str = instance_map->get_string("xport_endpoint");
+
+        console->print_info(shcore::str_format(
+            "The instance '%s' is part of the Cluster but the reported X "
+            "plugin port has changed. "
+            "Old xport: %s. Current xport: %s.",
+            instance_address.c_str(),
+            instance_map->get_string("old_xport_endpoint").c_str(),
+            xport_endpoint_str.c_str()));
+      }
     }
 
     // Decide to add/remove instance based on given operation options.
@@ -651,6 +703,13 @@ void Rescan::update_metadata_for_instances(
 
     update_metadata_for_instance(instance_cnx_opts,
                                  instance_map->get_uint("id"), label);
+
+    // The xplugin was disabled, remove the endpoint from the metadata
+    if (xport_endpoint_str == k_xplugin_disabled) {
+      m_cluster->get_metadata_storage()->update_instance_addresses(
+          instance_map->get_string("member_id"), "mysqlX",
+          shcore::Value::Null());
+    }
   }
 }
 

modules/adminapi/cluster/status.cc

@@ -24,6 +24,8 @@
 #include "modules/adminapi/cluster/status.h"
 
 #include <algorithm>
+#include <optional>
+#include <string>
 
 #include "modules/adminapi/cluster/api_options.h"
 #include "modules/adminapi/cluster_set/cluster_set_impl.h"
@@ -954,6 +956,25 @@ void check_host_metadata(shcore::Array_t issues, Instance *instance,
                       instance_md.md.endpoint + ", actual=" + address +
                       "). Use rescan() to update the metadata."));
   }
+
+  // Check if the instance's X plugin port matches the metadata
+  std::optional<int> xport = instance->get_xport();
+  std::string x_address;
+
+  if (xport.has_value()) {
+    x_address = mysqlshdk::utils::make_host_and_port(
+        instance->get_canonical_hostname(), *xport);
+  }
+
+  if (!mysqlshdk::utils::are_endpoints_equal(instance_md.md.xendpoint,
+                                             x_address)) {
+    issues->push_back(shcore::Value(
+        "ERROR: Metadata for this instance does not match "
+        "X plugin port reported by instance (metadata=" +
+        instance_md.md.xendpoint +
+        ", actual=" + (x_address.empty() ? "<disabled>" : x_address) +
+        "). Use rescan() to update the metadata."));
+  }
 }
 
 void check_transaction_size_limit(shcore::Array_t issues, Instance *instance,

modules/adminapi/common/metadata_storage.cc

@@ -1048,22 +1048,35 @@ bool MetadataStorage::query_instance_attribute(std::string_view uuid,
   return false;
 }
 
-void MetadataStorage::update_instance_attribute(std::string_view uuid,
-                                                std::string_view attribute,
-                                                const shcore::Value &value,
-                                                Transaction_undo *undo) {
+void MetadataStorage::update_instance_metadata(std::string_view uuid,
+                                               Instance_column_md column,
+                                               std::string_view field,
+                                               const shcore::Value &value,
+                                               Transaction_undo *undo) {
+  std::string column_str;
+
+  switch (column) {
+    case Instance_column_md::ATTRIBUTES:
+      column_str = "attributes";
+      break;
+    case Instance_column_md::ADDRESSES:
+      column_str = "addresses";
+      break;
+  }
+
   if (undo) {
     undo->add_snapshot_for_update(
-        "mysql_innodb_cluster_metadata.instances", "attributes",
-        *get_md_server(), shcore::sqlformat("mysql_server_uuid = ?", uuid));
+        "mysql_innodb_cluster_metadata.instances", column_str, *get_md_server(),
+        shcore::sqlformat("mysql_server_uuid = ?", uuid));
   }
 
   if (value) {
     auto stmt = shcore::str_format(
-        "UPDATE mysql_innodb_cluster_metadata.instances SET attributes = "
-        "json_set(attributes, '$.%.*s', CAST(? as JSON)) WHERE "
+        "UPDATE mysql_innodb_cluster_metadata.instances SET %s = "
+        "json_set(%s, '$.%.*s', CAST(? as JSON)) WHERE "
         "mysql_server_uuid = ?",
-        static_cast<int>(attribute.length()), attribute.data());
+        column_str.c_str(), column_str.c_str(),
+        static_cast<int>(field.length()), field.data());
 
     shcore::sqlstring query(stmt, 0);
     query << value.repr() << uuid;
@@ -1072,9 +1085,10 @@ void MetadataStorage::update_instance_attribute(std::string_view uuid,
     execute_sql(query);
   } else {
     auto stmt = shcore::str_format(
-        "UPDATE mysql_innodb_cluster_metadata.instances SET attributes = "
-        "json_remove(attributes, '$.%.*s') WHERE mysql_server_uuid = ?",
-        static_cast<int>(attribute.length()), attribute.data());
+        "UPDATE mysql_innodb_cluster_metadata.instances SET %s = "
+        "json_remove(%s, '$.%.*s') WHERE mysql_server_uuid = ?",
+        column_str.c_str(), column_str.c_str(),
+        static_cast<int>(field.length()), field.data());
 
     shcore::sqlstring query(stmt, 0);
     query << uuid;
@@ -1084,6 +1098,22 @@ void MetadataStorage::update_instance_attribute(std::string_view uuid,
   }
 }
 
+void MetadataStorage::update_instance_attribute(std::string_view uuid,
+                                                std::string_view attribute,
+                                                const shcore::Value &value,
+                                                Transaction_undo *undo) {
+  update_instance_metadata(uuid, Instance_column_md::ATTRIBUTES, attribute,
+                           value, undo);
+}
+
+void MetadataStorage::update_instance_addresses(std::string_view uuid,
+                                                std::string_view address,
+                                                const shcore::Value &value,
+                                                Transaction_undo *undo) {
+  update_instance_metadata(uuid, Instance_column_md::ADDRESSES, address, value,
+                           undo);
+}
+
 void MetadataStorage::set_instance_tag(const std::string &uuid,
                                        const std::string &tagname,
                                        const shcore::Value &value) {

modules/adminapi/common/metadata_storage.h

@@ -52,6 +52,8 @@ inline constexpr const char *kNotifyDataClusterSetId = "CLUSTER_SET_ID";
 
 enum class Instance_type { GROUP_MEMBER, ASYNC_MEMBER, READ_REPLICA, NONE };
 
+enum class Instance_column_md { ATTRIBUTES, ADDRESSES };
+
 std::string to_string(const Instance_type instance_type);
 Instance_type to_instance_type(const std::string &instance_type);
 
@@ -276,6 +278,11 @@ class MetadataStorage {
                                  const shcore::Value &value,
                                  Transaction_undo *undo = nullptr);
 
+  void update_instance_addresses(std::string_view uuid,
+                                 std::string_view address,
+                                 const shcore::Value &value,
+                                 Transaction_undo *undo = nullptr);
+
   /**
    * Update the information on the metadata about a tag set on the instance with
    * the given uuid.
@@ -699,6 +706,12 @@ class MetadataStorage {
 
   bool cluster_sets_supported() const;
 
+  void update_instance_metadata(std::string_view uuid,
+                                Instance_column_md column,
+                                std::string_view field,
+                                const shcore::Value &value,
+                                Transaction_undo *undo);
+
   friend class Transaction;
 
   std::shared_ptr<Instance> m_md_server;

modules/adminapi/common/sql.cc

@@ -265,20 +265,13 @@ std::vector<std::pair<std::string, int>> get_open_sessions(
 Instance_metadata query_instance_info(
     const mysqlshdk::mysql::IInstance &instance, bool validate_gr_endpoint,
     Instance_type instance_type) {
-  int xport = -1;
+  std::optional<int> xport;
   std::string local_gr_address;
 
   // Get the required data from the joining instance to store in the metadata
 
   // Get the MySQL X port.
-  try {
-    xport = instance.queryf_one_int(0, -1, "SELECT @@mysqlx_port");
-  } catch (const std::exception &e) {
-    log_info(
-        "The X plugin is not enabled on instance '%s'. No value will be "
-        "assumed for the X protocol address.",
-        instance.descr().c_str());
-  }
+  xport = instance.get_xport();
 
   if (instance_type == Instance_type::GROUP_MEMBER) {
     // Get the local GR host data.
@@ -311,9 +304,10 @@ Instance_metadata query_instance_info(
 
   instance_def.address = instance.get_canonical_address();
   instance_def.endpoint = instance_def.address;
-  if (xport != -1)
+  if (xport.has_value()) {
     instance_def.xendpoint = mysqlshdk::utils::make_host_and_port(
-        instance.get_canonical_hostname(), xport);
+        instance.get_canonical_hostname(), *xport);
+  }
 
   if (instance_type == Instance_type::GROUP_MEMBER) {
     instance_def.grendpoint = local_gr_address;

mysqlshdk/libs/mysql/instance.cc

@@ -98,6 +98,24 @@ int Instance::get_canonical_port() const {
   return m_port;
 }
 
+std::optional<int> Instance::get_xport() const {
+  if (!m_xport.has_value()) {
+    try {
+      auto result = query("SELECT @@mysqlx_port");
+      auto row = result->fetch_one();
+
+      m_xport = row->get_int(0);
+    } catch (const std::exception &) {
+      log_info("The X plugin is not enabled on instance '%s'.",
+               descr().c_str());
+
+      m_xport = std::nullopt;
+    }
+  }
+
+  return m_xport;
+}
+
 std::string Instance::get_canonical_address() const {
   // returns the canonical address that should be used to reach this instance in
   // the format: canonical_hostname + ':' + canonical_port

mysqlshdk/libs/mysql/instance.h

@@ -90,6 +90,7 @@ class IInstance {
   virtual std::string descr() const = 0;
   virtual std::string get_canonical_hostname() const = 0;
   virtual int get_canonical_port() const = 0;
+  virtual std::optional<int> get_xport() const = 0;
   /**
    * @brief returns the canonical address that should be used to reach this
    * instance in the format: canonical_hostname + ':' + canonical_port
@@ -373,6 +374,7 @@ class Instance : public IInstance {
   std::string get_canonical_hostname() const override;
   std::string get_canonical_address() const override;
   int get_canonical_port() const override;
+  std::optional<int> get_xport() const override;
 
   const std::string &get_uuid() const override;
   uint32_t get_id() const override;
@@ -528,6 +530,7 @@ class Instance : public IInstance {
   mutable std::string m_group_name;
   mutable std::string m_hostname;
   mutable int m_port = 0;
+  mutable std::optional<int> m_xport;
   mutable uint32_t m_server_id = 0;
   int m_sql_binlog_suppress_count = 0;
   Warnings_callback m_warnings_callback = nullptr;