Merge pull request #3828 from mariadb-corporation/feat/MCOL-6194-failover-conf

Configurable failover interval

Author: mariadb-AlexanderPresniakov

cmapi/cmapi_server/__main__.py

@@ -31,6 +31,7 @@
 from cmapi_server.managers.certificate import CertificateManager
 from cmapi_server.managers.process import MCSProcessManager
 from failover.node_monitor import NodeMonitor
+from failover.config import Config
 from mcs_node_control.models.dbrm_socket import SOCK_TIMEOUT, DBRMSocketHandler
 from mcs_node_control.models.node_config import NodeConfig
 from tracing.trace_tool import register_tracing_tools
@@ -97,7 +98,8 @@ class FailoverBackgroundThread(plugins.SimplePlugin):
 
     def __init__(self, bus, turned_on):
         super().__init__(bus)
-        self.node_monitor = NodeMonitor(agent=FailoverAgent())
+        sampling_interval = Config().getFailoverTimeoutSeconds()
+        self.node_monitor = NodeMonitor(agent=FailoverAgent(), samplingInterval=sampling_interval)
         self.running = False
         self.turned_on = turned_on
         if self.turned_on:

cmapi/cmapi_server/constants.py

@@ -42,6 +42,8 @@
 # TOTP secret key
 SECRET_KEY = 'MCSIsTheBestEver'  # not just a random string! (base32)
 
+DEFAULT_FAILOVER_SAMPLING_INTERVAL_SECS = 30
+
 
 # network constants
 # according to https://www.ibm.com/docs/en/storage-sentinel/1.1.2?topic=installation-map-your-local-host-loopback-address

cmapi/cmapi_server/controllers/dispatcher.py

@@ -9,6 +9,7 @@
     CommitController, ConfigController, ExtentMapController,
     LoggingConfigController, NodeController, NodeProcessController,
     RollbackController, ShutdownController, StartController, StatusController,
+    CmapiConfigController,
 )
 from cmapi_server.controllers.s3dataload import S3DataLoadController
 
@@ -205,7 +206,7 @@
 )
 
 
-# /_version/cluster/node/ (POST, PUT)
+# /_version/cluster/load_s3data (POST, PUT)
 dispatcher.connect(name = 'cluster_load_s3data',
                    route = f'/cmapi/{_version}/cluster/load_s3data',
                    action = 'load_s3data',
@@ -482,8 +483,16 @@
     conditions = {'method': ['PUT']}
 )
 
+# /_version/cmapi_config (PATCH)
+dispatcher.connect(
+    name = 'cmapi_config',
+    route = f'/cmapi/{_version}/cmapi_config',
+    action = 'patch_cmapi_config',
+    controller = CmapiConfigController(),
+    conditions = {'method': ['PATCH']}
+)
 
-def jsonify_error(status, message, traceback, version): \
+def jsonify_error(status, message, traceback, version):
     # pylint: disable=unused-argument
     """JSONify all CherryPy error responses (created by raising the
     cherrypy.HTTPError exception)

cmapi/cmapi_server/controllers/endpoints.py

@@ -6,15 +6,16 @@
 from copy import deepcopy
 from datetime import datetime
 from pathlib import Path
+from typing import Optional
 
 import cherrypy
 import pyotp
 import requests
+from pydantic import BaseModel, Field, ValidationError, model_validator
+
 from mcs_node_control.models.dbrm import set_cluster_mode
 from mcs_node_control.models.node_config import NodeConfig
 from mcs_node_control.models.node_status import NodeStatus
-from pydantic import ValidationError
-
 from cmapi_server.constants import (
     CMAPI_PACKAGE_NAME,
     CMAPI_PORT,
@@ -30,8 +31,10 @@
     SECRET_KEY,
 )
 from cmapi_server.controllers.api_clients import NodeControllerClient
+from cmapi_server import helpers
 from cmapi_server.controllers.error import APIError
 from cmapi_server.exceptions import CMAPIBasicError, cmapi_error_to_422
+from cmapi_server.exceptions import validate_or_422, exc_to_422
 from cmapi_server.controllers.request_models import (
     ConfigPutRequestRootModel, StatefulConfigPutRequestModel,
 )
@@ -1997,12 +2000,13 @@ def put_stateful_config(self):
         log_begin(module_logger, func_name)
 
         request_body = cherrypy.request.json
-        try:
-            request_stateful_config = StatefulConfigModel.model_validate(
-                request_body.get('stateful_config_dict')
-            )
-        except ValidationError as exp:
-            raise_422_error(module_logger, func_name,f'Invalid request body: {exp.errors()}')
+        request_stateful_config = validate_or_422(
+            StatefulConfigModel,
+            request_body.get('stateful_config_dict'),
+            module_logger,
+            func_name,
+            prefix='Invalid request body',
+        )
 
         success = AppStatefulConfig.apply_update(request_stateful_config)
         if not success:
@@ -2014,3 +2018,52 @@ def put_stateful_config(self):
             )
 
         return {'timestamp': str(datetime.now()), 'success': success}
+
+
+class CmapiConfigPatchModel(BaseModel):
+    failover_sampling_interval_seconds: Optional[int] = Field(default=None, ge=1)
+
+    @model_validator(mode='after')
+    def ensure_any_present(self):
+        if self.failover_sampling_interval_seconds is None:
+            raise ValueError('At least one field must be provided')
+        return self
+
+
+class CmapiConfigController:
+    @cherrypy.tools.timeit()
+    @cherrypy.tools.json_in()
+    @cherrypy.tools.json_out()
+    @cherrypy.tools.validate_api_key()  # pylint: disable=no-member
+    def patch_cmapi_config(self):
+        """Update our own CMAPI config section in Columnstore.xml"""
+        func_name = 'patch_cmapi_config'
+        log_begin(module_logger, func_name)
+
+        req_model = validate_or_422(
+            CmapiConfigPatchModel,
+            cherrypy.request.json,
+            module_logger,
+            func_name,
+            prefix='Invalid payload',
+        )
+
+        # Update Columnstore.xml under <CMAPIConfig>
+        nc = NodeConfig()
+        with nc.modify_config(DEFAULT_MCS_CONF_PATH) as root:
+            cmapi_node = helpers.get_or_create_child_xml_node(root, 'CMAPIConfig')
+
+            # Failover sampling interval
+            if req_model.failover_sampling_interval_seconds is not None:
+                node = helpers.get_or_create_child_xml_node(cmapi_node, 'FailoverSamplingIntervalSeconds')
+                node.text = str(req_model.failover_sampling_interval_seconds)
+
+        with exc_to_422(module_logger, func_name, prefix='Failed to bump config revision'):
+            helpers.update_revision_and_manager(input_config_filename=DEFAULT_MCS_CONF_PATH)
+
+        # Broadcast updated config
+        with cmapi_error_to_422(module_logger, func_name):
+            with TransactionManager() as txn:
+                helpers.broadcast_new_config(nodes=txn.success_txn_nodes)
+
+        return {'timestamp': str(datetime.now())}

cmapi/cmapi_server/exceptions.py

@@ -2,7 +2,9 @@
 
 from collections.abc import Iterator
 from contextlib import contextmanager
-from typing import Optional
+from typing import Optional, TypeVar, Any
+
+from pydantic import BaseModel, ValidationError
 
 from cmapi_server.controllers.error import APIError
 
@@ -58,3 +60,26 @@ def cmapi_error_to_422(logger, func_name: str) -> Iterator[None]:
         # mirror raise_422_error behavior locally to avoid circular imports
         logger.error(f'{func_name} {err.message}', exc_info=False)
         raise APIError(422, err.message) from err
+
+
+T = TypeVar('T', bound=BaseModel)
+
+def validate_or_422(
+    model: type[T], payload: Any, logger, func_name: str,
+    prefix: Optional[str] = 'Invalid request body',
+) -> T:
+    """Validate payload with Pydantic model or raise HTTP 422 APIError."""
+    try:
+        return model.model_validate(payload)
+    except ValidationError as exp:
+        msg = f"{prefix}: {exp.errors()}" if prefix else str(exp.errors())
+        logger.error(f"{func_name} {msg}", exc_info=False)
+        raise APIError(422, msg) from exp
+
+
+@contextmanager
+def exc_to_422(logger, func_name: str, prefix: Optional[str] = None) -> Iterator[None]:
+    """Convert any exception into HTTP 422"""
+    with cmapi_error_to_422(logger, func_name):
+        with exc_to_cmapi_error(prefix=prefix):
+            yield

cmapi/cmapi_server/helpers.py

@@ -18,6 +18,7 @@
 
 import aiohttp
 import lxml.objectify
+from lxml import etree
 import requests
 from tracing.traced_session import get_traced_session
 from tracing.traced_aiohttp import create_traced_async_session
@@ -54,6 +55,14 @@ def get_id() -> int:
     return int(random() * 1000000)
 
 
+def get_or_create_child_xml_node(parent, name: str):
+    """Get a direct child by tag name or create it if missing."""
+    node = parent.find(f'./{name}')
+    if node is None:
+        node = etree.SubElement(parent, name)
+    return node
+
+
 def start_transaction(
     config_filename: str = CMAPI_CONF_PATH,
     cs_config_filename: str = DEFAULT_MCS_CONF_PATH,

cmapi/cmapi_server/test/test_cmapi_config_endpoint.py

@@ -0,0 +1,62 @@
+import json
+from contextlib import ExitStack
+from unittest.mock import Mock, patch
+
+import requests
+from lxml import etree
+
+from cmapi_server.constants import _version
+from cmapi_server.test.unittest_global import TEST_API_KEY, BaseServerTestCase
+
+requests.packages.urllib3.disable_warnings()
+
+
+class TestCmapiConfigEndpoint(BaseServerTestCase):
+    def setUp(self):
+        super().setUp()
+        self._stack = ExitStack()
+        self.addCleanup(self._stack.close)
+
+        # Mock broadcast_new_config, TransactionManager and config path
+        self.broadcast_mock = Mock()
+        self._stack.enter_context(patch('cmapi_server.helpers.broadcast_new_config', new=self.broadcast_mock))
+        self._stack.enter_context(patch('cmapi_server.controllers.endpoints.DEFAULT_MCS_CONF_PATH', new=self.mcs_config_filename))
+        self._stack.enter_context(patch('cmapi_server.constants.DEFAULT_MCS_CONF_PATH', new=self.mcs_config_filename))
+
+        class _FakeTxn:
+            def __enter__(self):
+                self.success_txn_nodes = ['n1', 'n2']
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+        self._stack.enter_context(patch('cmapi_server.controllers.endpoints.TransactionManager', new=lambda: _FakeTxn()))
+
+    def test_sampling_interval_written(self):
+        r = self._request_patch(
+            {
+                'failover_sampling_interval_seconds': 45,
+            }
+        )
+        self.assertEqual(r.status_code, 200)
+
+        tree = etree.parse(str(self.mcs_config_filename))
+        self.assertEqual(tree.findtext('./CMAPIConfig/FailoverSamplingIntervalSeconds'), '45')
+
+    def test_interval_bounds_rejected(self):
+        r = self._request_patch({'failover_sampling_interval_seconds': 0})
+        self.assertEqual(r.status_code, 422)
+        body = r.json()
+        self.assertIn('error', body)
+
+    def test_broadcast_called(self):
+        r = self._request_patch({'failover_sampling_interval_seconds': 30})
+        self.assertEqual(r.status_code, 200)
+        self.broadcast_mock.assert_called_once()
+        self.assertEqual(self.broadcast_mock.call_args.kwargs.get('nodes'), ['n1', 'n2'])
+
+    def _request_patch(self, payload: dict):
+        url = f'https://localhost:8640/cmapi/{_version}/cmapi_config'
+        headers = {'x-api-key': TEST_API_KEY, 'Content-Type': 'application/json'}
+        return requests.patch(url, verify=False, headers=headers, data=json.dumps(payload))

cmapi/cmapi_server/test/test_em_endpoints.py

@@ -26,12 +26,14 @@ def run_server():
     CertificateManager.create_self_signed_certificate_if_not_exist()
     cherrypy.engine.start()
     cherrypy.engine.wait(cherrypy.engine.states.STARTED)
-    yield
-    cherrypy.engine.exit()
-    cherrypy.engine.block()
+    try:
+        yield
+    finally:
+        cherrypy.engine.exit()
+        cherrypy.engine.block()
 
 
-def get_current_key():
+def get_current_key() -> str:
     app_config = configparser.ConfigParser()
     try:
         with open(cmapi_config_filename, 'r') as _config_file:

cmapi/failover/config.py

@@ -1,11 +1,8 @@
-import configparser
 import logging
 import threading
 from os.path import getmtime
 
-import lxml
-
-from cmapi_server.constants import DEFAULT_MCS_CONF_PATH, DEFAULT_SM_CONF_PATH
+from cmapi_server.constants import DEFAULT_FAILOVER_SAMPLING_INTERVAL_SECS, DEFAULT_MCS_CONF_PATH
 from mcs_node_control.models.node_config import NodeConfig
 
 
@@ -18,6 +15,7 @@ class Config:
     _inactive_nodes = []
     _primary_node = ''
     _my_name = None  # derived from config file
+    _failover_sampling_interval = DEFAULT_FAILOVER_SAMPLING_INTERVAL_SECS
 
     config_lock = threading.Lock()
     last_mtime = 0
@@ -69,6 +67,13 @@ def getPrimaryNode(self):
         self.config_lock.release()
         return ret
 
+    def getFailoverTimeoutSeconds(self) -> int:
+        self.config_lock.acquire()
+        self.check_reload()
+        ret = self._failover_sampling_interval
+        self.config_lock.release()
+        return ret
+
     def check_reload(self):
         """Check config reload.
 
@@ -157,4 +162,11 @@ def load_config(self):
         self._primary_node = primary_node
         self.last_mtime = last_mtime
         self._my_name = my_name
+
+        sampling_interval_node = root.find('./CMAPIConfig/FailoverSamplingIntervalSeconds')
+        if sampling_interval_node is not None and sampling_interval_node.text is not None:
+            self._failover_sampling_interval = int(sampling_interval_node.text)
+        else:
+            self._failover_sampling_interval = DEFAULT_FAILOVER_SAMPLING_INTERVAL_SECS
+
         return True

cmapi/integration_tests/ssh.py

@@ -2,11 +2,10 @@
 import json
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any, Callable, Optional
 
 from fabric import Connection
 
-
 from dataclasses import dataclass, field
 
 @dataclass
@@ -75,8 +74,8 @@ def replicas(self) -> list[RemoteHost]:
 def run_on_all_hosts_parallel(
       hosts: list[RemoteHost],
       func: Callable[[RemoteHost], Any],
-      timeout: float | None = None,
-      max_workers: int | None = None,
+      timeout: Optional[float] = None,
+      max_workers: Optional[int] = None,
     ) -> dict[str, Any]:
     """
     Run a function on all hosts in parallel.