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..421b29d43 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 ( @@ -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 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..305e5e1ae --- /dev/null +++ b/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 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..bf7781225 --- /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: ghcr.io/xlab-uiuc/mariadb-writer:v1 + imagePullPolicy: IfNotPresent + env: + - name: MARIADB_HOST + value: "test-cluster-primary.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" 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") 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/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..6a279d904 100644 --- a/data/strimzi-kafka-operator/v0-45-0/cr.yaml +++ b/data/strimzi-kafka-operator/v0-45-0/cr.yaml @@ -10,10 +10,16 @@ spec: version: 3.9.0 metadataVersion: 3.9-IV0 listeners: + - name: config + port: 9094 + type: internal + tls: false - name: plain port: 9092 type: internal tls: false + authentication: + type: scram-sha-512 - name: tls port: 9093 type: internal @@ -24,6 +30,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..36007555c 100644 --- a/data/strimzi-kafka-operator/v0-45-0/func-only.json +++ b/data/strimzi-kafka-operator/v0-45-0/func-only.json @@ -11,12 +11,18 @@ "apply": { "file": "data/strimzi-kafka-operator/v0-45-0/node-pool.yaml" } + }, + { + "apply": { + "file": "data/strimzi-kafka-operator/v0-45-0/user_topic.yaml" + } } ] }, "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 825e51a74..a480a0e4e 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 +# 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,26 +23,53 @@ 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 = snapshot.input_cr["spec"]["kafka"]["config"] else: return None - + 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, - "acto-namespace", + 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, + self.oracle_handle.namespace, [ "./bin/kafka-configs.sh", "--describe", @@ -50,13 +79,16 @@ def check( "brokers", "--entity-name", "0", - "--all" + "--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 = {} @@ -68,17 +100,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..15c53e3e4 --- /dev/null +++ b/data/strimzi-kafka-operator/v0-45-0/user_topic.yaml @@ -0,0 +1,65 @@ +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: + 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" 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