From 36028601320acdd710766968e16a0d49f93be95c Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Sun, 9 Mar 2025 23:54:57 -0500 Subject: [PATCH 1/8] Adding MariaDB availability oracle Signed-off-by: Tyler Gu Fix MariaDB database name Signed-off-by: Tyler Gu --- data/mariadb-operator/v0-30-0/func-only.json | 1 + data/mariadb-operator/v0-30-0/oracle.py | 145 ++++++++++++++---- data/mariadb-operator/v0-30-0/writer_pod.yaml | 26 ++++ 3 files changed, 145 insertions(+), 27 deletions(-) create mode 100644 data/mariadb-operator/v0-30-0/writer_pod.yaml diff --git a/data/mariadb-operator/v0-30-0/func-only.json b/data/mariadb-operator/v0-30-0/func-only.json index 39783fd5f..b0ccb88be 100644 --- a/data/mariadb-operator/v0-30-0/func-only.json +++ b/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'\\]" diff --git a/data/mariadb-operator/v0-30-0/oracle.py b/data/mariadb-operator/v0-30-0/oracle.py index e3036898e..2d0833040 100644 --- a/data/mariadb-operator/v0-30-0/oracle.py +++ b/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 \ No newline at end of file +ON_INIT: Callable = deploy_writer diff --git a/data/mariadb-operator/v0-30-0/writer_pod.yaml b/data/mariadb-operator/v0-30-0/writer_pod.yaml new file mode 100644 index 000000000..a4f95670d --- /dev/null +++ b/data/mariadb-operator/v0-30-0/writer_pod.yaml @@ -0,0 +1,26 @@ +apiVersion: v1 +kind: Pod +metadata: + name: mariadb-writer + labels: + app: mariadb-writer + acto/tag: custom-oracle +spec: + containers: + - name: mariadb-writer + image: docker.io/tylergu1998/mariadb-writer:v1 + imagePullPolicy: IfNotPresent + env: + - name: MARIADB_HOST + value: "test-cluster.acto-namespace.svc.cluster.local" + - name: MARIADB_PORT + value: "3306" + - name: MARIADB_USER + value: "root" + - name: MARIADB_PASSWORD + valueFrom: + secretKeyRef: + name: mariadb + key: root-password + - name: MARIADB_DATABASE + value: "mariadb" From 982cd75ed4928be504c22412b47dfb30708f66ee Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Mon, 10 Mar 2025 19:17:28 -0500 Subject: [PATCH 2/8] Add MongoDB writer oracle Signed-off-by: Tyler Gu Add custom oracle to MongoDB's config file Signed-off-by: Tyler Gu Fix MongoDB writer pod on_init Signed-off-by: Tyler Gu Switch from docker to ghcr Signed-off-by: Tyler Gu Update mongodb preload images Signed-off-by: Tyler Gu Update MongoDB config Signed-off-by: Tyler Gu --- data/mariadb-operator/v0-30-0/writer_pod.yaml | 2 +- .../v1-16-0/context.json | 3 +- .../v1-16-0/cr.yaml | 4 +- .../v1-16-0/func-only.json | 1 + .../v1-16-0/oracle.py | 132 +++++++++++++++--- .../v1-16-0/writer_pod.yaml | 24 ++++ data/tidb-operator/v1-6-0/writer_pod.yaml | 2 +- 7 files changed, 145 insertions(+), 23 deletions(-) create mode 100644 data/percona-server-mongodb-operator/v1-16-0/writer_pod.yaml diff --git a/data/mariadb-operator/v0-30-0/writer_pod.yaml b/data/mariadb-operator/v0-30-0/writer_pod.yaml index a4f95670d..409f1df5a 100644 --- a/data/mariadb-operator/v0-30-0/writer_pod.yaml +++ b/data/mariadb-operator/v0-30-0/writer_pod.yaml @@ -8,7 +8,7 @@ metadata: spec: containers: - name: mariadb-writer - image: docker.io/tylergu1998/mariadb-writer:v1 + image: ghcr.io/xlab-uiuc/mariadb-writer:v1 imagePullPolicy: IfNotPresent env: - name: MARIADB_HOST diff --git a/data/percona-server-mongodb-operator/v1-16-0/context.json b/data/percona-server-mongodb-operator/v1-16-0/context.json index 3745ffc58..bdd5ddeb2 100644 --- a/data/percona-server-mongodb-operator/v1-16-0/context.json +++ b/data/percona-server-mongodb-operator/v1-16-0/context.json @@ -28441,7 +28441,8 @@ "preload_images": [ "docker.io/percona/percona-server-mongodb:7.0.8-5", "docker.io/perconalab/percona-server-mongodb-operator:1.16.0", - "docker.io/perconalab/percona-server-mongodb-operator:main" + "docker.io/perconalab/percona-server-mongodb-operator:main", + "docker.io/percona/percona-backup-mongodb:2.4.1" ], "static_analysis_time": 1.5735626220703125e-05 } diff --git a/data/percona-server-mongodb-operator/v1-16-0/cr.yaml b/data/percona-server-mongodb-operator/v1-16-0/cr.yaml index 2f1a53bab..395a11f2c 100644 --- a/data/percona-server-mongodb-operator/v1-16-0/cr.yaml +++ b/data/percona-server-mongodb-operator/v1-16-0/cr.yaml @@ -1,7 +1,7 @@ apiVersion: psmdb.percona.com/v1 kind: PerconaServerMongoDB metadata: - name: my-cluster-name + name: test-cluster spec: crVersion: 1.16.0 image: percona/percona-server-mongodb:7.0.8-5 @@ -148,7 +148,7 @@ spec: storage: 3Gi mongos: - size: 3 + size: 2 affinity: antiAffinityTopologyKey: "kubernetes.io/hostname" podDisruptionBudget: diff --git a/data/percona-server-mongodb-operator/v1-16-0/func-only.json b/data/percona-server-mongodb-operator/v1-16-0/func-only.json index 21e137e13..a0056b000 100644 --- a/data/percona-server-mongodb-operator/v1-16-0/func-only.json +++ b/data/percona-server-mongodb-operator/v1-16-0/func-only.json @@ -16,6 +16,7 @@ "example_dir": "data/percona-server-mongodb-operator/v1-16-0/examples", "num_nodes": 5, "custom_module": "data.percona-server-mongodb-operator.v1-16-0.custom_mapping", + "custom_oracle": "data.percona-server-mongodb-operator.v1-16-0.oracle", "diff_ignore_fields": [ "\\['secret'\\]\\['default\\-token\\-.*'\\]", "\\['secret'\\]\\['percona\\-server\\-mongodb\\-operator\\-token-.*'\\]", diff --git a/data/percona-server-mongodb-operator/v1-16-0/oracle.py b/data/percona-server-mongodb-operator/v1-16-0/oracle.py index 748a57049..f465f56f3 100644 --- a/data/percona-server-mongodb-operator/v1-16-0/oracle.py +++ b/data/percona-server-mongodb-operator/v1-16-0/oracle.py @@ -1,41 +1,48 @@ import base64 -from typing import Optional +import datetime +import json +import re +from typing import Callable, Optional import kubernetes +import yaml +# pylint: disable=import-error from acto.checker.checker import CheckerInterface +from acto.checker.impl.state_compare import CustomCompareMethods 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 json -import yaml -import re + +# pylint: enable=import-error class MongoDBConfigChecker(CheckerInterface): - """Custom oracle for checking Cassandra config""" + """Custom oracle for checking MongoDB config""" name = "mongodb-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 MongoDB config""" + _, _ = generation, prev_snapshot if ( "replsets" in snapshot.input_cr["spec"] and len(snapshot.input_cr["spec"]["replsets"]) > 0 and "configuration" in snapshot.input_cr["spec"]["replsets"][0] ): - mongo_yaml = yaml.load(snapshot.input_cr["spec"]["replsets"][0][ - "configuration"], Loader=yaml.FullLoader) + mongo_yaml = yaml.load( + snapshot.input_cr["spec"]["replsets"][0]["configuration"], + Loader=yaml.FullLoader, + ) else: return None @@ -52,8 +59,12 @@ def check( "my-cluster-name-secrets", self.oracle_handle.namespace ).data - username = base64.b64decode(secret["MONGODB_DATABASE_ADMIN_USER"]).decode("utf-8") - password = base64.b64decode(secret["MONGODB_DATABASE_ADMIN_PASSWORD"]).decode("utf-8") + username = base64.b64decode( + secret["MONGODB_DATABASE_ADMIN_USER"] + ).decode("utf-8") + password = base64.b64decode( + secret["MONGODB_DATABASE_ADMIN_PASSWORD"] + ).decode("utf-8") pod = pod_list[0] p = self.oracle_handle.kubectl_client.exec( @@ -84,19 +95,104 @@ def check( mark = 0 for i in range(len(config_data)): - config_data[i] = re.sub("(\w+):", '"\g<1>":', config_data[i]) - config_data[i] = re.sub("'(.+)'", '"\g<1>"', config_data[i]) - if '\"ok\":' in config_data[i]: + config_data[i] = re.sub(r"(\w+):", r'"\g<1>":', config_data[i]) + config_data[i] = re.sub(r"'(.+)'", r'"\g<1>"', config_data[i]) + if '"ok":' in config_data[i]: mark = i break - config_data[mark - 1] = re.sub("(\w*)},", '\g<1>}', config_data[mark - 1]) + config_data[mark - 1] = re.sub( + r"(\w*)},", r"\g<1>}", config_data[mark - 1] + ) config_data = json.loads("".join(config_data[:mark] + ["}"]))["parsed"] compare_methods = CustomCompareMethods() if not compare_methods.equals(mongo_yaml, config_data): - return OracleResult(message="MongoDB config check failed due to missing keys or wrong values") + return OracleResult( + message="MongoDB 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 MongoDB 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="mongodb-writer", + namespace=self.oracle_handle.namespace, + container="mongodb-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="MongoDB writer availability rate is below 90%" + ) + else: + logger.warning( + "Unable to parse MongoDB writer log line: %s", line + ) + continue + + if min_avail_rate is not None: + logger.info( + "MongoDB writer lowest availability rate: %f", min_avail_rate + ) + else: + logger.error("MongoDB writer availability rate not found") + + return None + + def check( + self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot + ) -> Optional[OracleResult]: + """Check the MongoDB 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""" + + handle.kubectl_client.kubectl( + [ + "apply", + "-f", + "data/percona-server-mongodb-operator/v1-16-0/writer_pod.yaml", + "-n", + handle.namespace, + ] + ) + CUSTOM_CHECKER: type[CheckerInterface] = MongoDBConfigChecker + +ON_INIT: Callable = deploy_writer diff --git a/data/percona-server-mongodb-operator/v1-16-0/writer_pod.yaml b/data/percona-server-mongodb-operator/v1-16-0/writer_pod.yaml new file mode 100644 index 000000000..46e2e3598 --- /dev/null +++ b/data/percona-server-mongodb-operator/v1-16-0/writer_pod.yaml @@ -0,0 +1,24 @@ +apiVersion: v1 +kind: Pod +metadata: + name: mongodb-writer +spec: + containers: + - name: mongodb-writer + image: ghcr.io/xlab-uiuc/mongodb-writer:v1 + imagePullPolicy: IfNotPresent + env: + - name: MONGO_HOST + value: "test-cluster-mongos.acto-namespace.svc.cluster.local" + - name: MONGO_PORT + value: "27017" + - name: MONGO_USER + valueFrom: + secretKeyRef: + name: internal-test-cluster-users + key: MONGODB_DATABASE_ADMIN_USER + - name: MONGO_PASSWORD + valueFrom: + secretKeyRef: + name: internal-test-cluster-users + key: MONGODB_DATABASE_ADMIN_PASSWORD diff --git a/data/tidb-operator/v1-6-0/writer_pod.yaml b/data/tidb-operator/v1-6-0/writer_pod.yaml index fb4234d87..7d3c0db94 100644 --- a/data/tidb-operator/v1-6-0/writer_pod.yaml +++ b/data/tidb-operator/v1-6-0/writer_pod.yaml @@ -8,7 +8,7 @@ metadata: spec: containers: - name: writer - image: docker.io/tylergu1998/tidb-writer:v1 + image: ghcr.io/xlab-uiuc/tidb-writer:v1 imagePullPolicy: IfNotPresent env: - name: TIDB_HOST From f5f4c2bf57b0b4d540584388af82cc237616dada Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Fri, 21 Mar 2025 01:43:23 -0500 Subject: [PATCH 3/8] Add Cassandra writer oracle Signed-off-by: Tyler Gu Fix namespace Signed-off-by: Tyler Gu Add error log Signed-off-by: Tyler Gu --- data/cass-operator/v1-22/func-only.json | 1 + data/cass-operator/v1-22/oracle.py | 99 ++++++++++++++++++++++-- data/cass-operator/v1-22/writer_pod.yaml | 24 ++++++ 3 files changed, 118 insertions(+), 6 deletions(-) create mode 100644 data/cass-operator/v1-22/writer_pod.yaml diff --git a/data/cass-operator/v1-22/func-only.json b/data/cass-operator/v1-22/func-only.json index ab5f7cf86..a9894ffdc 100644 --- a/data/cass-operator/v1-22/func-only.json +++ b/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\\'\\]", diff --git a/data/cass-operator/v1-22/oracle.py b/data/cass-operator/v1-22/oracle.py index 067d17d01..a25bfcfb7 100644 --- a/data/cass-operator/v1-22/oracle.py +++ b/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 ( @@ -79,13 +85,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 +108,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="acto-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 diff --git a/data/cass-operator/v1-22/writer_pod.yaml b/data/cass-operator/v1-22/writer_pod.yaml new file mode 100644 index 000000000..3be2dffcd --- /dev/null +++ b/data/cass-operator/v1-22/writer_pod.yaml @@ -0,0 +1,24 @@ +apiVersion: v1 +kind: Pod +metadata: + name: cassandra-writer +spec: + containers: + - name: cassandra-writer + image: docker.io/tylergu1998/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: username From bd2c151797b4a42c07571cb2b670ada27b85eea0 Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Fri, 21 Mar 2025 20:08:34 -0500 Subject: [PATCH 4/8] Fix cassandra writer pod yaml Signed-off-by: Tyler Gu Fix Cassandra oracle Signed-off-by: Tyler Gu Tolerate when there is no Cassandra pod available Signed-off-by: Tyler Gu --- data/cass-operator/v1-22/oracle.py | 5 ++++- data/cass-operator/v1-22/writer_pod.yaml | 6 ++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/data/cass-operator/v1-22/oracle.py b/data/cass-operator/v1-22/oracle.py index a25bfcfb7..421b29d43 100644 --- a/data/cass-operator/v1-22/oracle.py +++ b/data/cass-operator/v1-22/oracle.py @@ -46,6 +46,9 @@ def __check_config( 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( @@ -124,7 +127,7 @@ def __check_availability( ) logs = v1.read_namespaced_pod_log( name="cassandra-writer", - namespace="acto-namespace", + namespace=self.oracle_handle.namespace, container="cassandra-writer", since_seconds=since_seconds, ).split("\n") diff --git a/data/cass-operator/v1-22/writer_pod.yaml b/data/cass-operator/v1-22/writer_pod.yaml index 3be2dffcd..305e5e1ae 100644 --- a/data/cass-operator/v1-22/writer_pod.yaml +++ b/data/cass-operator/v1-22/writer_pod.yaml @@ -2,10 +2,12 @@ apiVersion: v1 kind: Pod metadata: name: cassandra-writer + labels: + acto/tag: custom-oracle spec: containers: - name: cassandra-writer - image: docker.io/tylergu1998/cassandra-writer:v1 + image: ghcr.io/xlab-uiuc/cassandra-writer:v1 imagePullPolicy: IfNotPresent env: - name: CASSANDRA_HOST @@ -21,4 +23,4 @@ spec: valueFrom: secretKeyRef: name: development-superuser - key: username + key: password From 56e035720818f99267be208e4b60f8153b568266 Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Sun, 23 Mar 2025 15:04:53 -0500 Subject: [PATCH 5/8] Add Kafka sample user and topic Signed-off-by: Tyler Gu Fix Kafka CR's authorization config Signed-off-by: Tyler Gu Fix Kafka CR's authorization config Signed-off-by: Tyler Gu Port Kafka oracle Signed-off-by: Tyler Gu Try scram-sha-512 Signed-off-by: Tyler Gu SASL Plain Signed-off-by: Tyler Gu Kafka workload working Signed-off-by: Tyler Gu --- .../v0-45-0/config-only.json | 5 + data/strimzi-kafka-operator/v0-45-0/cr.yaml | 6 + .../v0-45-0/func-only.json | 5 + data/strimzi-kafka-operator/v0-45-0/oracle.py | 123 +++++++++++++++--- .../v0-45-0/user_topic.yaml | 37 ++++++ .../v0-45-0/writer_pod.yaml | 23 ++++ 6 files changed, 184 insertions(+), 15 deletions(-) create mode 100644 data/strimzi-kafka-operator/v0-45-0/user_topic.yaml create mode 100644 data/strimzi-kafka-operator/v0-45-0/writer_pod.yaml diff --git a/data/strimzi-kafka-operator/v0-45-0/config-only.json b/data/strimzi-kafka-operator/v0-45-0/config-only.json index cb4fcf391..29ceba4c0 100644 --- a/data/strimzi-kafka-operator/v0-45-0/config-only.json +++ b/data/strimzi-kafka-operator/v0-45-0/config-only.json @@ -11,6 +11,11 @@ "apply": { "file": "data/strimzi-kafka-operator/v0-45-0/node-pool.yaml" } + }, + { + "apply": { + "file": "data/strimzi-kafka-operator/v0-45-0/user_topic.yaml" + } } ] }, diff --git a/data/strimzi-kafka-operator/v0-45-0/cr.yaml b/data/strimzi-kafka-operator/v0-45-0/cr.yaml index f1ffa8035..cc128a88c 100644 --- a/data/strimzi-kafka-operator/v0-45-0/cr.yaml +++ b/data/strimzi-kafka-operator/v0-45-0/cr.yaml @@ -14,6 +14,8 @@ spec: port: 9092 type: internal tls: false + authentication: + type: scram-sha-512 - name: tls port: 9093 type: internal @@ -24,6 +26,10 @@ spec: transaction.state.log.min.isr: 2 default.replication.factor: 3 min.insync.replicas: 2 + authorization: + type: simple + superUsers: + - admin entityOperator: topicOperator: {} userOperator: {} diff --git a/data/strimzi-kafka-operator/v0-45-0/func-only.json b/data/strimzi-kafka-operator/v0-45-0/func-only.json index c05174197..c3a56bfb8 100644 --- a/data/strimzi-kafka-operator/v0-45-0/func-only.json +++ b/data/strimzi-kafka-operator/v0-45-0/func-only.json @@ -11,6 +11,11 @@ "apply": { "file": "data/strimzi-kafka-operator/v0-45-0/node-pool.yaml" } + }, + { + "apply": { + "file": "data/strimzi-kafka-operator/v0-45-0/user_topic.yaml" + } } ] }, diff --git a/data/strimzi-kafka-operator/v0-45-0/oracle.py b/data/strimzi-kafka-operator/v0-45-0/oracle.py index 825e51a74..024dc096d 100644 --- a/data/strimzi-kafka-operator/v0-45-0/oracle.py +++ b/data/strimzi-kafka-operator/v0-45-0/oracle.py @@ -1,16 +1,18 @@ -import base64 -from typing import Optional +import datetime +import re +from typing import Callable, Optional import kubernetes +import yaml +# 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 -from acto.checker.impl.state_compare import CustomCompareMethods -import yaml -import re + +# pylint: enable=import-error class KafkaConfigChecker(CheckerInterface): @@ -21,21 +23,26 @@ class KafkaConfigChecker(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""" + """Check the Kafka config""" + _, _ = generation, prev_snapshot logger = get_thread_logger() logger.info("Checking Kafka config") if ( "kafka" in snapshot.input_cr["spec"] and "config" in snapshot.input_cr["spec"]["kafka"] ): - kafka_config = yaml.load(snapshot.input_cr["spec"]["kafka"]["config"], Loader=yaml.FullLoader) + kafka_config = yaml.load( + snapshot.input_cr["spec"]["kafka"]["config"], + Loader=yaml.FullLoader, + ) else: return None - + pod_name = "test-cluster-dual-role-0" p = self.oracle_handle.kubectl_client.exec( @@ -50,7 +57,7 @@ def check( "brokers", "--entity-name", "0", - "--all" + "--all", ], capture_output=True, text=True, @@ -68,17 +75,103 @@ def check( value = True elif value == "false": value = False - elif re.match(r"^-?\d+$", value) or re.match(r"^-?\d+\.\d+$", value): + elif re.match(r"^-?\d+$", value) or re.match( + r"^-?\d+\.\d+$", value + ): value = float(value) runtime_config[name] = value - + for n in kafka_config: if not n in runtime_config: - return OracleResult(message=f"Kafka config check failed due to missing keys") + return OracleResult( + message=f"Kafka config check failed due to missing keys {n}" + ) elif runtime_config[n] != kafka_config[n]: - return OracleResult(message=f"Kafka config check failed due to inconsistent value of the {name}") + return OracleResult( + message=f"Kafka config check failed due to inconsistent value of the {n}" + ) return None + def __check_availability( + self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot + ) -> Optional[OracleResult]: + """Check the Kafka 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="kafka-writer", + namespace=self.oracle_handle.namespace, + container="kafka-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="Kafka writer availability rate is below 90%" + ) + else: + logger.warning( + "Unable to parse Kafka writer log line: %s", line + ) + continue + + if min_avail_rate is not None: + logger.info( + "Kafka writer lowest availability rate: %f", min_avail_rate + ) + else: + logger.error("Kafka writer availability rate not found") + + return None + + def check( + self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot + ) -> Optional[OracleResult]: + """Check the Kafka 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""" + + handle.kubectl_client.kubectl( + [ + "apply", + "-f", + "data/strimzi-kafka-operator/v0-45-0/writer_pod.yaml", + "-n", + handle.namespace, + ] + ) + -CUSTOM_CHECKER: type[CheckerInterface] = KafkaConfigChecker \ No newline at end of file +CUSTOM_CHECKER: type[CheckerInterface] = KafkaConfigChecker +ON_INIT: Callable = deploy_writer diff --git a/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml b/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml new file mode 100644 index 000000000..f4705a728 --- /dev/null +++ b/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml @@ -0,0 +1,37 @@ +apiVersion: kafka.strimzi.io/v1beta2 +kind: KafkaUser +metadata: + name: my-user + labels: + strimzi.io/cluster: test-cluster +spec: + authentication: + type: scram-sha-512 + authorization: + type: simple + acls: + # Example consumer Acls for topic my-topic using consumer group my-group + - resource: + type: topic + name: "*" + patternType: literal + operations: + - All + host: "*" + - resource: + type: group + name: my-group + patternType: literal + operations: + - Read + host: "*" +--- +apiVersion: kafka.strimzi.io/v1beta2 +kind: KafkaTopic +metadata: + name: my-topic + labels: + strimzi.io/cluster: test-cluster +spec: + partitions: 1 + replicas: 3 diff --git a/data/strimzi-kafka-operator/v0-45-0/writer_pod.yaml b/data/strimzi-kafka-operator/v0-45-0/writer_pod.yaml new file mode 100644 index 000000000..1eb5d2c0b --- /dev/null +++ b/data/strimzi-kafka-operator/v0-45-0/writer_pod.yaml @@ -0,0 +1,23 @@ +apiVersion: v1 +kind: Pod +metadata: + name: kafka-writer +spec: + containers: + - name: kafka-writer + image: ghcr.io/xlab-uiuc/kafka-writer:v1 + imagePullPolicy: Always + env: + - name: KAFKA_HOST + value: "test-cluster-kafka-bootstrap.acto-namespace.svc.cluster.local" + - name: KAFKA_PORT + value: "9092" + - name: KAFKA_USER + value: "my-user" + - name: KAFKA_PASSWORD + valueFrom: + secretKeyRef: + name: my-user + key: password + - name: KAFKA_TOPIC + value: "my-topic" From eb6a1ba50ae2b12dcdddf2ad6e3640b26b31aa32 Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Fri, 28 Mar 2025 15:15:57 -0500 Subject: [PATCH 6/8] Update Kafka config to use the custom oracle Signed-off-by: Tyler Gu Fix Kafka oracle Signed-off-by: Tyler Gu --- data/strimzi-kafka-operator/v0-45-0/func-only.json | 1 + data/strimzi-kafka-operator/v0-45-0/oracle.py | 6 +----- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/data/strimzi-kafka-operator/v0-45-0/func-only.json b/data/strimzi-kafka-operator/v0-45-0/func-only.json index c3a56bfb8..36007555c 100644 --- a/data/strimzi-kafka-operator/v0-45-0/func-only.json +++ b/data/strimzi-kafka-operator/v0-45-0/func-only.json @@ -22,6 +22,7 @@ "crd_name": "kafkas.kafka.strimzi.io", "seed_custom_resource": "data/strimzi-kafka-operator/v0-45-0/cr.yaml", "example_dir": "data/strimzi-kafka-operator/v0-45-0/examples", + "custom_oracle": "data.strimzi-kafka-operator.v0-45-0.oracle", "wait_time": 60, "diff_ignore_fields": [ "\\['metadata']\\['annotations']\\['strimzi\\.io\/revision']", diff --git a/data/strimzi-kafka-operator/v0-45-0/oracle.py b/data/strimzi-kafka-operator/v0-45-0/oracle.py index 024dc096d..2d6bcc07f 100644 --- a/data/strimzi-kafka-operator/v0-45-0/oracle.py +++ b/data/strimzi-kafka-operator/v0-45-0/oracle.py @@ -3,7 +3,6 @@ from typing import Callable, Optional import kubernetes -import yaml # pylint: disable=import-error from acto.checker.checker import CheckerInterface @@ -36,10 +35,7 @@ def __check_config( "kafka" in snapshot.input_cr["spec"] and "config" in snapshot.input_cr["spec"]["kafka"] ): - kafka_config = yaml.load( - snapshot.input_cr["spec"]["kafka"]["config"], - Loader=yaml.FullLoader, - ) + kafka_config = snapshot.input_cr["spec"]["kafka"]["config"] else: return None From 82d32549cf79f0f8532fa2b2ecb18b18a77bfd98 Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Sat, 29 Mar 2025 14:26:23 -0500 Subject: [PATCH 7/8] Add a non-auth port for config check Signed-off-by: Tyler Gu Fix port Signed-off-by: Tyler Gu Use scram sha 512 for authentication Signed-off-by: Tyler Gu Debug Signed-off-by: Tyler Gu Debug Signed-off-by: Tyler Gu Fix client properties creation Signed-off-by: Tyler Gu Fix client properties creation Signed-off-by: Tyler Gu Use primary svc for workload Signed-off-by: Tyler Gu --- data/mariadb-operator/v0-30-0/writer_pod.yaml | 2 +- data/strimzi-kafka-operator/v0-45-0/cr.yaml | 4 +++ data/strimzi-kafka-operator/v0-45-0/oracle.py | 33 +++++++++++++++++-- .../v0-45-0/user_topic.yaml | 28 ++++++++++++++++ 4 files changed, 64 insertions(+), 3 deletions(-) diff --git a/data/mariadb-operator/v0-30-0/writer_pod.yaml b/data/mariadb-operator/v0-30-0/writer_pod.yaml index 409f1df5a..bf7781225 100644 --- a/data/mariadb-operator/v0-30-0/writer_pod.yaml +++ b/data/mariadb-operator/v0-30-0/writer_pod.yaml @@ -12,7 +12,7 @@ spec: imagePullPolicy: IfNotPresent env: - name: MARIADB_HOST - value: "test-cluster.acto-namespace.svc.cluster.local" + value: "test-cluster-primary.acto-namespace.svc.cluster.local" - name: MARIADB_PORT value: "3306" - name: MARIADB_USER diff --git a/data/strimzi-kafka-operator/v0-45-0/cr.yaml b/data/strimzi-kafka-operator/v0-45-0/cr.yaml index cc128a88c..6a279d904 100644 --- a/data/strimzi-kafka-operator/v0-45-0/cr.yaml +++ b/data/strimzi-kafka-operator/v0-45-0/cr.yaml @@ -10,6 +10,10 @@ spec: version: 3.9.0 metadataVersion: 3.9-IV0 listeners: + - name: config + port: 9094 + type: internal + tls: false - name: plain port: 9092 type: internal diff --git a/data/strimzi-kafka-operator/v0-45-0/oracle.py b/data/strimzi-kafka-operator/v0-45-0/oracle.py index 2d6bcc07f..a480a0e4e 100644 --- a/data/strimzi-kafka-operator/v0-45-0/oracle.py +++ b/data/strimzi-kafka-operator/v0-45-0/oracle.py @@ -1,3 +1,4 @@ +import base64 import datetime import re from typing import Callable, Optional @@ -41,9 +42,34 @@ def __check_config( pod_name = "test-cluster-dual-role-0" + core_v1 = kubernetes.client.CoreV1Api(self.oracle_handle.k8s_client) + secret = core_v1.read_namespaced_secret( + "admin", self.oracle_handle.namespace + ).data + jaas_config = base64.b64decode(secret["sasl.jaas.config"]).decode( + "utf-8" + ) + + p = self.oracle_handle.kubectl_client.exec( + pod_name, + self.oracle_handle.namespace, + [ + "bash", + "-c", + "bash -c 'echo \"security.protocol=SASL_PLAINTEXT\n" + "sasl.mechanism=SCRAM-SHA-512\n" + f"sasl.jaas.config={jaas_config}\" > /tmp/client.properties'", + ], + capture_output=True, + text=True, + ) + if p.returncode != 0: + logger.error("Failed to create client.properties%s", p.stderr) + return OracleResult(message="Failed to create client.properties") + p = self.oracle_handle.kubectl_client.exec( pod_name, - "acto-namespace", + self.oracle_handle.namespace, [ "./bin/kafka-configs.sh", "--describe", @@ -54,12 +80,15 @@ def __check_config( "--entity-name", "0", "--all", + "--command-config", + "/tmp/client.properties", ], capture_output=True, text=True, ) if p.returncode != 0: - return OracleResult(message="Kafka config check failed") + logger.error("Failed to get Kafka config%s", p.stderr) + return OracleResult(message="Failed to get Kafka config for check") lines = p.stdout.split("\n")[1:] runtime_config = {} diff --git a/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml b/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml index f4705a728..15c53e3e4 100644 --- a/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml +++ b/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml @@ -1,5 +1,33 @@ apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaUser +metadata: + name: admin + labels: + strimzi.io/cluster: test-cluster +spec: + authentication: + type: scram-sha-512 + authorization: + type: simple + acls: + # Example consumer Acls for topic my-topic using consumer group my-group + - resource: + type: topic + name: "*" + patternType: literal + operations: + - All + host: "*" + - resource: + type: group + name: my-group + patternType: literal + operations: + - Read + host: "*" +--- +apiVersion: kafka.strimzi.io/v1beta2 +kind: KafkaUser metadata: name: my-user labels: From 4dda6a7dc627b728efafa63a466118b74d16367b Mon Sep 17 00:00:00 2001 From: TZ-zzz Date: Mon, 7 Apr 2025 10:06:45 +0800 Subject: [PATCH 8/8] fix field not found bug in minio oracle --- data/minio-operator/v7-0-0/oracle.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data/minio-operator/v7-0-0/oracle.py b/data/minio-operator/v7-0-0/oracle.py index 0314b3859..e15fc3c58 100644 --- a/data/minio-operator/v7-0-0/oracle.py +++ b/data/minio-operator/v7-0-0/oracle.py @@ -31,7 +31,7 @@ def check( if ( "storage-configuration" in snapshot.system_state["secret"] ): - minio_config = snapshot.system_state["secret"]["data"]["config.env"].strip().split("\n") + minio_config = snapshot.system_state["secret"]["storage-configuration"]["data"]["config.env"].strip().split("\n") else: return OracleResult(message=f"Missing the Secret that contains the configuration")