From 02509a2327ca58be894d556c9969e2e0e5510c0f Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Thu, 26 Jun 2025 23:24:00 -0700 Subject: [PATCH 1/2] Fix compatibility issues in reproduce and the ZK custom oracle Signed-off-by: Tyler Gu --- acto/reproduce.py | 66 +++++++++--- data/zookeeper-operator/oracle.py | 167 +++++++++++++++++++----------- 2 files changed, 159 insertions(+), 74 deletions(-) diff --git a/acto/reproduce.py b/acto/reproduce.py index 791f26418d..82dab6749d 100644 --- a/acto/reproduce.py +++ b/acto/reproduce.py @@ -7,7 +7,7 @@ from datetime import datetime from functools import partial from glob import glob -from typing import List, Optional +from typing import List, Optional, Tuple import jsonpatch import yaml @@ -15,7 +15,11 @@ from acto import DEFAULT_KUBERNETES_VERSION from acto.engine import Acto from acto.input import k8s_schemas, property_attribute -from acto.input.input import CustomKubernetesMapping, DeterministicInputModel +from acto.input.input import ( + CustomKubernetesMapping, + CustomPropertySchemaMapping, + DeterministicInputModel, +) from acto.input.testcase import TestCase from acto.input.testplan import TestGroup from acto.input.value_with_schema import ValueWithSchema @@ -23,6 +27,7 @@ from acto.post_process.post_diff_test import PostDiffTest from acto.result import OracleResults from acto.schema.base import BaseSchema +from acto.schema.opaque import OpaqueSchema from acto.schema.schema import extract_schema from acto.utils import get_thread_logger @@ -106,7 +111,6 @@ def __init__( logger.info("%d steps for reproducing", len(self.testcases)) self.num_total_cases = len(cr_list) - 1 self.num_workers = 1 - self.metadata = {} override_matches: Optional[list[tuple[BaseSchema, str]]] = None if custom_module_path is not None: @@ -141,6 +145,43 @@ def __init__( f"but got {type(custom_mapping)}" ) + if hasattr(custom_module, "CUSTOM_PROPERTY_SCHEMA_MAPPING"): + custom_property_schema_mapping = ( + custom_module.CUSTOM_PROPERTY_SCHEMA_MAPPING + ) + if isinstance(custom_property_schema_mapping, list): + for custom_mapping in custom_property_schema_mapping: + if isinstance( + custom_mapping, CustomPropertySchemaMapping + ): + try: + schema = self.get_schema_by_path( + custom_mapping.schema_path + ) + except KeyError: + logger.warning( + "Specified schema path does not exist: %s, using opaque schema", + custom_mapping.schema_path, + ) + schema = OpaqueSchema( + custom_mapping.schema_path, {} + ) + self.set_schema_by_path( + custom_mapping.schema_path, + custom_mapping.custom_schema.from_original_schema( + schema + ), + ) + logger.info( + "Applying custom schema to property %s", + custom_mapping.schema_path, + ) + else: + raise TypeError( + "Expected CustomPropertySchemaMapping in " + f"CUSTOM_PROPERTY_SCHEMA_MAPPING, but got {type(custom_mapping)}" + ) + # Do the matching from CRD to Kubernetes schemas # Match the Kubernetes schemas to subproperties of the root schema kubernetes_schema_matcher = k8s_schemas.K8sSchemaMatcher.from_version( @@ -190,13 +231,15 @@ def get_seed_input(self) -> dict: return self.seed_input def generate_test_plan( - self, delta_from: str = None, focus_fields: list = None + self, + focus_fields: Optional[list] = None, ) -> dict: - _ = delta_from _ = focus_fields return {} - def next_test(self) -> list: + def next_test( + self, + ) -> Optional[list[Tuple[TestGroup, tuple[str, TestCase]]]]: """ Returns: - a list of tuples, containing @@ -204,7 +247,7 @@ def next_test(self) -> list: """ test = self.testcases.pop(0) return [ - (TestGroup([test]), ('["spec"]', test)) + (TestGroup([('["spec"]', test)]), ('["spec"]', test)) ] # return the first test case def apply_k8s_schema(self, k8s_field): @@ -259,7 +302,6 @@ def reproduce( workdir_path=workdir_path, operator_config=config, cluster_runtime=kwargs["cluster_runtime"], - preload_images_=[], context_file=context_cache, helper_crd=None, num_workers=1, @@ -272,7 +314,7 @@ def reproduce( acto_namespace=acto_namespace, ) - errors = acto.run(modes=["normal"]) + errors = acto.run() return [error for error in errors if error is not None] @@ -336,14 +378,10 @@ def reproduce_postdiff( parser.add_argument("--context", dest="context", help="Cached context data") args = parser.parse_args() - workdir_path_ = f"testrun-{datetime.now().strftime('%Y-%m-%d-%H-%M')}" - - start_time = datetime.now() reproduce( - workdir_path=workdir_path_, + workdir_path=f"testrun-{datetime.now().strftime('%Y-%m-%d-%H-%M')}", reproduce_dir=args.reproduce_dir, operator_config=args.config, acto_namespace=args.acto_namespace, cluster_runtime=args.cluster_runtime, ) - end_time = datetime.now() diff --git a/data/zookeeper-operator/oracle.py b/data/zookeeper-operator/oracle.py index fe5e128a07..c0bc4e8dbe 100644 --- a/data/zookeeper-operator/oracle.py +++ b/data/zookeeper-operator/oracle.py @@ -1,119 +1,166 @@ import json -from typing import List -from acto.checker.checker import Checker +from typing import Callable, Optional + +from acto.checker.checker import CheckerInterface +from acto.common import canonicalize from acto.oracle_handle import OracleHandle -from acto.common import ErrorResult, Oracle, PassResult, RunResult, canonicalize +from acto.result import OracleResult +from acto.snapshot import Snapshot + +class ZookeeperOracle(CheckerInterface): + """Custom oracle for checking Zookeeper cluster health""" -class ZookeeperOracle(Checker): - name = 'custom' + name = "custom" def __init__(self, oracle_handle: OracleHandle, **kwargs): super().__init__(**kwargs) self.oracle_handle = oracle_handle - def check(self, **kwargs) -> RunResult: - '''Checks the health of the Zookeeper cluster''' + def check( + self, generation: int, snapshot: Snapshot, prev_snapshot: Snapshot + ) -> Optional[OracleResult]: + """Checks the health of the Zookeeper cluster""" + _, _, _ = generation, snapshot, prev_snapshot + handle = self.oracle_handle cr = handle.get_cr() - if 'config' in cr['spec']: - config = cr['spec']['config'] - if 'additionalConfig' in config and config['additionalConfig'] != None: - for key, value in config['additionalConfig'].items(): + if "config" in cr["spec"]: + config = cr["spec"]["config"] + if ( + "additionalConfig" in config + and config["additionalConfig"] is not None + ): + for key, value in config["additionalConfig"].items(): config[key] = value - del config['additionalConfig'] + del config["additionalConfig"] else: config = None sts_list = handle.get_stateful_sets() if len(sts_list) != 1: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper cluster has more than one stateful set') + return OracleResult( + message="Zookeeper cluster has more than one stateful set", + ) pod_list = handle.get_pods_in_stateful_set(sts_list[0]) leaders = 0 for pod in pod_list: - if pod.status.pod_ip == None: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper pod does not have an IP assigned') + if pod.status.pod_ip is None: + return OracleResult( + message="Zookeeper pod does not have an IP assigned", + ) p = handle.kubectl_client.exec( pod.metadata.name, - pod.metadata.namespace, ['curl', 'http://' + pod.status.pod_ip + ':8080/commands/ruok'], + pod.metadata.namespace, + ["curl", "http://" + pod.status.pod_ip + ":8080/commands/ruok"], capture_output=True, - text=True) + text=True, + ) if p.returncode != 0: - return ErrorResult(oracle=Oracle.CUSTOM, msg='Zookeeper pod is not responding') + return OracleResult(message="Zookeeper pod is not responding") else: result = json.loads(p.stdout) - if result['error'] != None: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper cluster curl has error ' + result['error']) + if result["error"] is not None: + return OracleResult( + message="Zookeeper cluster curl has error " + + result["error"], + ) p = handle.kubectl_client.exec( pod.metadata.name, - pod.metadata.namespace, ['curl', 'http://' + pod.status.pod_ip + ':8080/commands/stat'], + pod.metadata.namespace, + ["curl", "http://" + pod.status.pod_ip + ":8080/commands/stat"], capture_output=True, - text=True) + text=True, + ) if p.returncode != 0: - return ErrorResult(oracle=Oracle.CUSTOM, msg='Zookeeper pod is not responding') + return OracleResult(message="Zookeeper pod is not responding") else: result = json.loads(p.stdout) - if result['error'] != None: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper cluster curl has error ' + result['error']) - elif result['server_stats']['server_state'] == 'leader': + if result["error"] is not None: + return OracleResult( + message="Zookeeper cluster curl has error " + + result["error"], + ) + elif result["server_stats"]["server_state"] == "leader": leaders += 1 - if config != None: + if config is not None: p = handle.kubectl_client.exec( pod.metadata.name, pod.metadata.namespace, - ['curl', 'http://' + pod.status.pod_ip + ':8080/commands/conf'], + [ + "curl", + "http://" + pod.status.pod_ip + ":8080/commands/conf", + ], capture_output=True, - text=True) + text=True, + ) if p.returncode != 0: - return ErrorResult(oracle=Oracle.CUSTOM, msg='Zookeeper pod is not responding') + return OracleResult( + message="Zookeeper pod is not responding", + ) else: result = json.loads(p.stdout) - if result['error'] != None: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper cluster curl has error ' + result['error']) + if result["error"] is not None: + return OracleResult( + message="Zookeeper cluster curl has error " + + result["error"], + ) for key, value in config.items(): canonicalize_key = canonicalize(key) if canonicalize_key not in result: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper config does not contain key ' + key) + return OracleResult( + message="Zookeeper config does not contain key " + + key, + ) elif result[canonicalize_key] != value: - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper cluster has incorrect config') + return OracleResult( + message="Zookeeper cluster has incorrect config", + ) if leaders > 1: - return ErrorResult(oracle=Oracle.CUSTOM, msg='Zookeeper cluster has more than one leader') - - p = handle.kubectl_client.exec('zkapp', - handle.namespace, ['request'], - capture_output=True, - text=True) + return OracleResult( + message="Zookeeper cluster has more than one leader", + ) + + p = handle.kubectl_client.exec( + "zkapp", + handle.namespace, + ["request"], + capture_output=True, + text=True, + ) if p.returncode != 0: - return ErrorResult(oracle=Oracle.CUSTOM, msg='Zookeeper app request failed') - elif p.stdout != 'test': - return ErrorResult(oracle=Oracle.CUSTOM, - msg='Zookeeper app request result wrong %s' % p.stdout) + return OracleResult(message="Zookeeper app request failed") + elif p.stdout != "test": + return OracleResult( + message=f"Zookeeper app request result wrong {p.stdout}", + ) - return PassResult() + return None def deploy_zk_app(handle: OracleHandle): - handle.kubectl_client.kubectl([ - 'run', 'zkapp', '--image=tylergu1998/zkapp:v1', '-l', 'acto/tag=custom-oracle', '-n', - handle.namespace - ]) - - -CUSTOM_CHECKER: List[callable] = [ZookeeperOracle] -ON_INIT: List[callable] = [deploy_zk_app] + """Deploys a Zookeeper app for testing purposes""" + handle.kubectl_client.kubectl( + [ + "run", + "zkapp", + "--image=tylergu1998/zkapp:v1", + "-l", + "acto/tag=custom-oracle", + "-n", + handle.namespace, + ] + ) + + +CUSTOM_CHECKER: type[CheckerInterface] = ZookeeperOracle +ON_INIT: Callable = deploy_zk_app From 6cbcf61649b7be3f8298f245eaae993f5d7a2201 Mon Sep 17 00:00:00 2001 From: Tyler Gu Date: Sat, 28 Jun 2025 10:32:28 -0700 Subject: [PATCH 2/2] Fix the type of repro_apply_testcase function and add type validation Signed-off-by: Tyler Gu --- acto/engine.py | 16 ++++++++++++++-- acto/reproduce.py | 10 ++++++++-- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/acto/engine.py b/acto/engine.py index d7d01b4d0c..845c65af3e 100644 --- a/acto/engine.py +++ b/acto/engine.py @@ -57,9 +57,21 @@ RECOVERY_SNAPSHOT = -2 # the immediate snapshot before the error +ApplyTestCaseFunctionType = Callable[ + [ + ValueWithSchema, + list[str], + TestCase, + bool, + Optional[list[XorCondition]], + ], + jsonpatch.JsonPatch, +] + + def apply_testcase( value_with_schema: ValueWithSchema, - path: list, + path: list[str], testcase: TestCase, setup: bool = False, constraints: Optional[list[XorCondition]] = None, @@ -824,7 +836,7 @@ def __init__( analysis_only: bool, is_reproduce: bool, input_model: type[DeterministicInputModel], - apply_testcase_f: Callable, + apply_testcase_f: ApplyTestCaseFunctionType, mount: Optional[list] = None, focus_fields: Optional[list] = None, acto_namespace: int = 0, diff --git a/acto/reproduce.py b/acto/reproduce.py index 82dab6749d..d1b490f80d 100644 --- a/acto/reproduce.py +++ b/acto/reproduce.py @@ -15,10 +15,12 @@ from acto import DEFAULT_KUBERNETES_VERSION from acto.engine import Acto from acto.input import k8s_schemas, property_attribute +from acto.input.constraint import XorCondition from acto.input.input import ( CustomKubernetesMapping, CustomPropertySchemaMapping, DeterministicInputModel, + InputMetadata, ) from acto.input.testcase import TestCase from acto.input.testplan import TestGroup @@ -34,11 +36,13 @@ def apply_repro_testcase( value_with_schema: ValueWithSchema, - _: list, + path: list[str], testcase: TestCase, - __: bool = False, + setup: bool = False, + constraints: Optional[list[XorCondition]] = None, ) -> jsonpatch.JsonPatch: """apply_testcase function for reproducing""" + _, _, _ = path, setup, constraints logger = get_thread_logger(with_prefix=True) next_cr = testcase.mutator(None) # next cr in yaml format @@ -112,6 +116,8 @@ def __init__( self.num_total_cases = len(cr_list) - 1 self.num_workers = 1 + self.metadata = InputMetadata() + override_matches: Optional[list[tuple[BaseSchema, str]]] = None if custom_module_path is not None: custom_module = importlib.import_module(custom_module_path)