Add client workload oracle for TiDB, MariaDB, MongoDB, Cassandra, and Kafka

data/cass-operator/v1-22/func-only.json

@@ -30,6 +30,7 @@
     "seed_custom_resource": "data/cass-operator/v1-22/cr.yaml",
     "example_dir": "data/cass-operator/v1-22/examples",
     "custom_module": "data.cass-operator.v1-22.custom_mapping",
+    "custom_oracle": "data.cass-operator.v1-22.oracle",
     "diff_ignore_fields": [
         "\\[\\'secret\\'\\]\\[\\'cluster1\\-superuser\\'\\]\\[\\'data\\'\\]\\[\\'password\\'\\]",
         "\\[\\'secret\\'\\]\\[\\'test\\-cluster\\-ca\\-keystore\\'\\]\\[\\'data\\'\\]\\[\\'cert\\'\\]",

data/cass-operator/v1-22/oracle.py

@@ -1,15 +1,19 @@
 import base64
+import datetime
 import re
-from typing import Optional
+from typing import Callable, Optional
 
 import kubernetes
 
+# pylint: disable=import-error
 from acto.checker.checker import CheckerInterface
 from acto.oracle_handle import OracleHandle
 from acto.result import OracleResult
 from acto.snapshot import Snapshot
 from acto.utils.thread_logger import get_thread_logger
 
+# pylint: enable=import-error
+
 
 class CassandraConfigChecker(CheckerInterface):
     """Custom oracle for checking Cassandra config"""
@@ -19,11 +23,13 @@ class CassandraConfigChecker(CheckerInterface):
     def __init__(self, oracle_handle: OracleHandle, **kwargs):
         super().__init__(**kwargs)
         self.oracle_handle = oracle_handle
+        self._previous_ts: Optional[datetime.datetime] = None
 
-    def check(
+    def __check_config(
         self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
     ) -> Optional[OracleResult]:
         """Check the Cassandra config"""
+        _, _ = generation, prev_snapshot
         logger = get_thread_logger()
 
         if (
@@ -40,6 +46,9 @@ def check(
         sts_list = self.oracle_handle.get_stateful_sets()
         pod_list = self.oracle_handle.get_pods_in_stateful_set(sts_list[0])
 
+        if len(pod_list) == 0:
+            return None
+
         core_v1 = kubernetes.client.CoreV1Api(self.oracle_handle.k8s_client)
 
         secret = core_v1.read_namespaced_secret(
@@ -79,13 +88,13 @@ def check(
                 key = matches.group(1)
                 if matches.group(3):
                     value = matches.group(3)
-                    if re.match(r'^-?\d+(?:\.\d+)?$', value):
+                    if re.match(r"^-?\d+(?:\.\d+)?$", value):
                         value = float(value)
-                    elif re.match(r'true|false', value):
-                        value = eval(value[0].upper() + value[1:])
+                    elif re.match(r"true|false", value):
+                        value = value == "true"
                 else:
                     value = None
-                
+
                 config_data[key] = value
             else:
                 logger.error("Failed to parse line: %s", line)
@@ -102,5 +111,86 @@ def check(
 
         return None
 
+    def __check_availability(
+        self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
+    ) -> Optional[OracleResult]:
+        """Check the Cassandra writer availability rate parsing its logs"""
+        _, _, _ = generation, snapshot, prev_snapshot
+        logger = get_thread_logger()
+
+        v1 = kubernetes.client.CoreV1Api(self.oracle_handle.k8s_client)
+        now = datetime.datetime.now()
+        since_seconds: Optional[int] = (
+            None
+            if self._previous_ts is None
+            else int((now - self._previous_ts).total_seconds())
+        )
+        logs = v1.read_namespaced_pod_log(
+            name="cassandra-writer",
+            namespace=self.oracle_handle.namespace,
+            container="cassandra-writer",
+            since_seconds=since_seconds,
+        ).split("\n")
+        self._previous_ts = now
+
+        min_avail_rate: Optional[float] = None
+        for line in logs:
+            match = re.search(r"TS: \[(.*?)\].*?Success Rate: \[(.*?)\]", line)
+            if match:
+                _ = match.group(1)
+                success_rate = float(match.group(2))
+                min_avail_rate = (
+                    success_rate
+                    if min_avail_rate is None
+                    else min(min_avail_rate, success_rate)
+                )
+
+                if float(success_rate) < 0.9:
+                    return OracleResult(
+                        message="Cassandra writer availability rate is below 90%"
+                    )
+            else:
+                logger.warning(
+                    "Unable to parse Cassandra writer log line: %s", line
+                )
+                continue
+
+        if min_avail_rate is not None:
+            logger.info(
+                "Cassandra writer lowest availability rate: %f", min_avail_rate
+            )
+        else:
+            logger.error("Cassandra writer availability rate not found")
+
+        return None
+
+    def check(
+        self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
+    ) -> Optional[OracleResult]:
+        """Check the Cassandra config and availability"""
+        if result := self.__check_config(generation, snapshot, prev_snapshot):
+            return result
+        if result := self.__check_availability(
+            generation, snapshot, prev_snapshot
+        ):
+            return result
+        return None
+
+
+def deploy_writer(handle: OracleHandle):
+    """Deploy the Writer Pod for Oracle"""
+    p = handle.kubectl_client.kubectl(
+        [
+            "apply",
+            "-f",
+            "data/cass-operator/v1-22/writer_pod.yaml",
+            "-n",
+            handle.namespace,
+        ]
+    )
+    if p.returncode != 0:
+        raise RuntimeError("Failed to deploy the writer pod", p.stderr)
+
 
 CUSTOM_CHECKER: type[CheckerInterface] = CassandraConfigChecker
+ON_INIT: Callable = deploy_writer

data/cass-operator/v1-22/writer_pod.yaml

@@ -0,0 +1,26 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: cassandra-writer
+  labels:
+    acto/tag: custom-oracle
+spec:
+  containers:
+    - name: cassandra-writer
+      image: ghcr.io/xlab-uiuc/cassandra-writer:v1
+      imagePullPolicy: IfNotPresent
+      env:
+        - name: CASSANDRA_HOST
+          value: "development-test-cluster-service.cass-operator.svc.cluster.local"
+        - name: CASSANDRA_PORT
+          value: "9042"
+        - name: CASSANDRA_USER
+          valueFrom:
+            secretKeyRef:
+              name: development-superuser
+              key: username
+        - name: CASSANDRA_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: development-superuser
+              key: password

data/mariadb-operator/v0-30-0/func-only.json

@@ -24,6 +24,7 @@
     "example_dir": "data/mariadb-operator/v0-30-0/examples",
     "seed_custom_resource": "data/mariadb-operator/v0-30-0/cr.yaml",
     "custom_module": "data.mariadb-operator.v0-30-0.custom_mapping",
+    "custom_oracle": "data.mariadb-operator.v0-30-0.oracle",
     "num_nodes": 5,
     "diff_ignore_fields": [
         "\\['test\\-cluster\\-primary'\\]\\['spec'\\]\\['selector'\\]\\['statefulset\\.kubernetes\\.io\\\/pod\\-name'\\]"

data/mariadb-operator/v0-30-0/oracle.py

@@ -1,18 +1,25 @@
 import base64
-from typing import Optional
+import configparser
+import datetime
+import re
+from typing import Callable, Optional
 
 import kubernetes
 
+# pylint: disable=import-error
 from acto.checker.checker import CheckerInterface
+from acto.checker.impl.state_compare import CustomCompareMethods
+from acto.k8s_util.k8sutil import canonicalize_quantity
 from acto.oracle_handle import OracleHandle
 from acto.result import OracleResult
 from acto.snapshot import Snapshot
 from acto.utils.thread_logger import get_thread_logger
-from acto.checker.impl.state_compare import CustomCompareMethods
-import configparser
-from acto.k8s_util.k8sutil import canonicalize_quantity
+
+# pylint: enable=import-error
+
 
 def decode(value: str) -> dict:
+    """Decode the config string to dictionary"""
     config = configparser.ConfigParser()
     config.read_string(value)
 
@@ -41,32 +48,35 @@ def decode(value: str) -> dict:
 
 
 class MariaDBConfigChecker(CheckerInterface):
-    """Custom oracle for checking TiDB config"""
+    """Custom oracle for checking MariaDB config"""
 
-    name = "tidb-config-checker"
+    name = "mariadb-config-checker"
 
     def __init__(self, oracle_handle: OracleHandle, **kwargs):
         super().__init__(**kwargs)
         self.oracle_handle = oracle_handle
+        self._previous_ts: Optional[datetime.datetime] = None
 
-    def check(
+    def __check_config(
         self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
     ) -> Optional[OracleResult]:
-        """Check the Cassandra config"""
-        logger = get_thread_logger()
+        """Check the MariaDB config"""
+        _, _ = generation, prev_snapshot
 
-        if (
-            "myCnf" in snapshot.input_cr["spec"]
-        ):
+        if "myCnf" in snapshot.input_cr["spec"]:
             mairadb_spec_config = snapshot.input_cr["spec"]["myCnf"]
         else:
             return None
-        
+
         mariadb_spec_config = decode(mairadb_spec_config)["mariadb"]
 
         for key, value in mariadb_spec_config.items():
-            mariadb_spec_config[key] = canonicalize_quantity(value) if type(canonicalize_quantity(value)) != str else canonicalize_quantity(value).lower()
-
+            mariadb_spec_config[key] = (
+                canonicalize_quantity(value)
+                if not isinstance(canonicalize_quantity(value), str)
+                else canonicalize_quantity(value).lower()
+            )
+
         # Get the configuration from the cqlsh
         sts_list = self.oracle_handle.get_stateful_sets()
         pod_list = self.oracle_handle.get_pods_in_stateful_set(sts_list[0])
@@ -84,14 +94,7 @@ def check(
         p = self.oracle_handle.kubectl_client.exec(
             pod.metadata.name,
             pod.metadata.namespace,
-            [
-                "mariadb",
-                "-u", 
-                "root", 
-                f"-p{password}", 
-                "-e", 
-                "SHOW VARIABLES;"
-            ],
+            ["mariadb", "-u", "root", f"-p{password}", "-e", "SHOW VARIABLES;"],
             capture_output=True,
             text=True,
         )
@@ -104,14 +107,102 @@ def check(
             compoents = line.split("\t")
             if len(compoents) < 2:
                 continue
-            else:
-                mariadb_config[compoents[0]] = canonicalize_quantity(compoents[1]) if type(canonicalize_quantity(compoents[1])) != str else canonicalize_quantity(compoents[1]).lower()
+
+            mariadb_config[compoents[0]] = (
+                canonicalize_quantity(compoents[1])
+                if not isinstance(canonicalize_quantity(compoents[1]), str)
+                else canonicalize_quantity(compoents[1]).lower()
+            )
 
         compare_methods = CustomCompareMethods()
         if not compare_methods.equals(mariadb_spec_config, mariadb_config):
-            return OracleResult(message=f"MariaDB config check failed due to missing keys or wrong values.")
+            return OracleResult(
+                message="MariaDB config check failed due to missing keys or wrong values."
+            )
+
+        return None
+
+    def __check_availability(
+        self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
+    ) -> Optional[OracleResult]:
+        """Check the MariaDB writer availability rate parsing its logs"""
+        _, _, _ = generation, snapshot, prev_snapshot
+        logger = get_thread_logger()
+
+        v1 = kubernetes.client.CoreV1Api(self.oracle_handle.k8s_client)
+        now = datetime.datetime.now()
+        since_seconds: Optional[int] = (
+            None
+            if self._previous_ts is None
+            else int((now - self._previous_ts).total_seconds())
+        )
+        logs = v1.read_namespaced_pod_log(
+            name="mariadb-writer",
+            namespace=self.oracle_handle.namespace,
+            container="mariadb-writer",
+            since_seconds=since_seconds,
+        ).split("\n")
+        self._previous_ts = now
+
+        min_avail_rate: Optional[float] = None
+        for line in logs:
+            match = re.search(r"TS: \[(.*?)\].*?Success Rate: \[(.*?)\]", line)
+            if match:
+                _ = match.group(1)
+                success_rate = float(match.group(2))
+                min_avail_rate = (
+                    success_rate
+                    if min_avail_rate is None
+                    else min(min_avail_rate, success_rate)
+                )
+
+                if float(success_rate) < 0.9:
+                    return OracleResult(
+                        message="MariaDB writer availability rate is below 90%"
+                    )
+            else:
+                logger.warning(
+                    "Unable to parse MariaDB writer log line: %s", line
+                )
+                continue
+
+        if min_avail_rate is not None:
+            logger.info(
+                "MariaDB writer lowest availability rate: %f", min_avail_rate
+            )
+        else:
+            logger.error("MariaDB writer availability rate not found")
 
         return None
 
+    def check(
+        self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot
+    ) -> Optional[OracleResult]:
+        """Check the MariaDB config and availability"""
+        if result := self.__check_config(generation, snapshot, prev_snapshot):
+            return result
+        if result := self.__check_availability(
+            generation, snapshot, prev_snapshot
+        ):
+            return result
+        return None
+
+
+CUSTOM_CHECKER: type[CheckerInterface] = MariaDBConfigChecker
+
+
+def deploy_writer(handle: OracleHandle):
+    """Deploy the Writer Pod for Oracle"""
+
+    handle.kubectl_client.kubectl(
+        [
+            "apply",
+            "-f",
+            "data/mariadb-operator/v0-30-0/writer_pod.yaml",
+            "-n",
+            handle.namespace,
+        ]
+    )
+
 
-CUSTOM_CHECKER: type[CheckerInterface] = MariaDBConfigChecker
+ON_INIT: Callable = deploy_writer