diff --git a/krkn/scenario_plugins/native/network/ingress_shaping.py b/krkn/scenario_plugins/native/network/ingress_shaping.py index cf74828f..2aa9d84f 100644 --- a/krkn/scenario_plugins/native/network/ingress_shaping.py +++ b/krkn/scenario_plugins/native/network/ingress_shaping.py @@ -18,17 +18,14 @@ from kubernetes.client.api.batch_v1_api import BatchV1Api as BatchV1Api @dataclass class NetworkScenarioConfig: - node_interface_name: typing.Dict[ - str, typing.List[str] - ] = field( + node_interface_name: typing.Dict[str, typing.List[str]] = field( default=None, metadata={ "name": "Node Interface Name", - "description": - "Dictionary with node names as key and values as a list of " - "their test interfaces. " - "Required if label_selector is not set.", - } + "description": "Dictionary with node names as key and values as a list of " + "their test interfaces. " + "Required if label_selector is not set.", + }, ) label_selector: typing.Annotated[ @@ -37,93 +34,76 @@ class NetworkScenarioConfig: default=None, metadata={ "name": "Label selector", - "description": - "Kubernetes label selector for the target nodes. " - "Required if node_interface_name is not set.\n" - "See https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/ " # noqa - "for details.", - } - ) - - test_duration: typing.Annotated[ - typing.Optional[int], - validation.min(1) - ] = field( - default=120, - metadata={ - "name": "Test duration", - "description": - "Duration for which each step of the ingress chaos testing " - "is to be performed.", + "description": "Kubernetes label selector for the target nodes. " + "Required if node_interface_name is not set.\n" + "See https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/ " # noqa + "for details.", }, ) - wait_duration: typing.Annotated[ - typing.Optional[int], - validation.min(1) - ] = field( + test_duration: typing.Annotated[typing.Optional[int], validation.min(1)] = field( + default=120, + metadata={ + "name": "Test duration", + "description": "Duration for which each step of the ingress chaos testing " + "is to be performed.", + }, + ) + + wait_duration: typing.Annotated[typing.Optional[int], validation.min(1)] = field( default=30, metadata={ "name": "Wait Duration", - "description": - "Wait duration for finishing a test and its cleanup." - "Ensure that it is significantly greater than wait_duration" - } + "description": "Wait duration for finishing a test and its cleanup." + "Ensure that it is significantly greater than wait_duration", + }, ) - instance_count: typing.Annotated[ - typing.Optional[int], - validation.min(1) - ] = field( + instance_count: typing.Annotated[typing.Optional[int], validation.min(1)] = field( default=1, metadata={ "name": "Instance Count", - "description": - "Number of nodes to perform action/select that match " - "the label selector.", - } + "description": "Number of nodes to perform action/select that match " + "the label selector.", + }, ) kubeconfig_path: typing.Optional[str] = field( default=None, metadata={ "name": "Kubeconfig path", - "description": - "Path to your Kubeconfig file. Defaults to ~/.kube/config.\n" - "See https://kubernetes.io/docs/concepts/configuration/organize-cluster-access-kubeconfig/ " # noqa - "for details.", - } + "description": "Path to your Kubeconfig file. Defaults to ~/.kube/config.\n" + "See https://kubernetes.io/docs/concepts/configuration/organize-cluster-access-kubeconfig/ " # noqa + "for details.", + }, ) execution_type: typing.Optional[str] = field( - default='parallel', + default="parallel", metadata={ "name": "Execution Type", - "description": - "The order in which the ingress filters are applied. " - "Execution type can be 'serial' or 'parallel'" - } + "description": "The order in which the ingress filters are applied. " + "Execution type can be 'serial' or 'parallel'", + }, ) network_params: typing.Dict[str, str] = field( default=None, metadata={ "name": "Network Parameters", - "description": - "The network filters that are applied on the interface. " - "The currently supported filters are latency, " - "loss and bandwidth" - } + "description": "The network filters that are applied on the interface. " + "The currently supported filters are latency, " + "loss and bandwidth", + }, ) kraken_config: typing.Optional[str] = field( - default='', + default="", metadata={ "name": "Kraken Config", - "description": - "Path to the config file of Kraken. " - "Set this field if you wish to publish status onto Cerberus" - } + "description": "Path to the config file of Kraken. " + "Set this field if you wish to publish status onto Cerberus", + }, ) @@ -132,33 +112,30 @@ class NetworkScenarioSuccessOutput: filter_direction: str = field( metadata={ "name": "Filter Direction", - "description": - "Direction in which the traffic control filters are applied " - "on the test interfaces" + "description": "Direction in which the traffic control filters are applied " + "on the test interfaces", } ) test_interfaces: typing.Dict[str, typing.List[str]] = field( metadata={ "name": "Test Interfaces", - "description": - "Dictionary of nodes and their interfaces on which " - "the chaos experiment was performed" + "description": "Dictionary of nodes and their interfaces on which " + "the chaos experiment was performed", } ) network_parameters: typing.Dict[str, str] = field( metadata={ "name": "Network Parameters", - "description": - "The network filters that are applied on the interfaces" + "description": "The network filters that are applied on the interfaces", } ) execution_type: str = field( metadata={ "name": "Execution Type", - "description": "The order in which the filters are applied" + "description": "The order in which the filters are applied", } ) @@ -168,18 +145,13 @@ class NetworkScenarioErrorOutput: error: str = field( metadata={ "name": "Error", - "description": - "Error message when there is a run-time error during " - "the execution of the scenario" + "description": "Error message when there is a run-time error during " + "the execution of the scenario", } ) -def get_default_interface( - node: str, - pod_template, - cli: CoreV1Api -) -> str: +def get_default_interface(node: str, pod_template, cli: CoreV1Api) -> str: """ Function that returns a random interface from a node @@ -210,9 +182,9 @@ def get_default_interface( logging.error("Exception occurred while executing command in pod") sys.exit(1) - routes = output.split('\n') + routes = output.split("\n") for route in routes: - if 'default' in route: + if "default" in route: default_route = route break @@ -226,10 +198,7 @@ def get_default_interface( def verify_interface( - input_interface_list: typing.List[str], - node: str, - pod_template, - cli: CoreV1Api + input_interface_list: typing.List[str], node: str, pod_template, cli: CoreV1Api ) -> typing.List[str]: """ Function that verifies whether a list of interfaces is present in the node. @@ -258,22 +227,15 @@ def verify_interface( try: if input_interface_list == []: cmd = ["ip", "r"] - output = kube_helper.exec_cmd_in_pod( - cli, - cmd, - "fedtools", - "default" - ) + output = kube_helper.exec_cmd_in_pod(cli, cmd, "fedtools", "default") if not output: - logging.error( - "Exception occurred while executing command in pod" - ) + logging.error("Exception occurred while executing command in pod") sys.exit(1) - routes = output.split('\n') + routes = output.split("\n") for route in routes: - if 'default' in route: + if "default" in route: default_route = route break @@ -281,20 +243,13 @@ def verify_interface( else: cmd = ["ip", "-br", "addr", "show"] - output = kube_helper.exec_cmd_in_pod( - cli, - cmd, - "fedtools", - "default" - ) + output = kube_helper.exec_cmd_in_pod(cli, cmd, "fedtools", "default") if not output: - logging.error( - "Exception occurred while executing command in pod" - ) + logging.error("Exception occurred while executing command in pod") sys.exit(1) - interface_ip = output.split('\n') + interface_ip = output.split("\n") node_interface_list = [ interface.split()[0] for interface in interface_ip[:-1] ] @@ -302,12 +257,12 @@ def verify_interface( for interface in input_interface_list: if interface not in node_interface_list: logging.error( - "Interface %s not found in node %s interface list %s" % - (interface, node, node_interface_list) + "Interface %s not found in node %s interface list %s" + % (interface, node, node_interface_list) ) raise Exception( - "Interface %s not found in node %s interface list %s" % - (interface, node, node_interface_list) + "Interface %s not found in node %s interface list %s" + % (interface, node, node_interface_list) ) finally: logging.info("Deleteing pod to query interface on node") @@ -321,9 +276,8 @@ def get_node_interfaces( label_selector: str, instance_count: int, pod_template, - cli: CoreV1Api + cli: CoreV1Api, ) -> typing.Dict[str, typing.List[str]]: - """ Function that is used to process the input dictionary with the nodes and its test interfaces. @@ -364,11 +318,7 @@ def get_node_interfaces( nodes = kube_helper.get_node(None, label_selector, instance_count, cli) node_interface_dict = {} for node in nodes: - node_interface_dict[node] = get_default_interface( - node, - pod_template, - cli - ) + node_interface_dict[node] = get_default_interface(node, pod_template, cli) else: node_name_list = node_interface_dict.keys() filtered_node_list = [] @@ -395,9 +345,8 @@ def apply_ingress_filter( batch_cli: BatchV1Api, cli: CoreV1Api, create_interfaces: bool = True, - param_selector: str = 'all' + param_selector: str = "all", ) -> str: - """ Function that applies the filters to shape incoming traffic to the provided node's interfaces. @@ -438,22 +387,18 @@ def apply_ingress_filter( """ network_params = cfg.network_params - if param_selector != 'all': + if param_selector != "all": network_params = {param_selector: cfg.network_params[param_selector]} if create_interfaces: create_virtual_interfaces(cli, interface_list, node, pod_template) exec_cmd = get_ingress_cmd( - interface_list, network_params, duration=cfg.test_duration - ) + interface_list, network_params, duration=cfg.test_duration + ) logging.info("Executing %s on node %s" % (exec_cmd, node)) job_body = yaml.safe_load( - job_template.render( - jobname=str(hash(node))[:5], - nodename=node, - cmd=exec_cmd - ) + job_template.render(jobname=str(hash(node))[:5], nodename=node, cmd=exec_cmd) ) api_response = kube_helper.create_job(batch_cli, job_body) @@ -464,10 +409,7 @@ def apply_ingress_filter( def create_virtual_interfaces( - cli: CoreV1Api, - interface_list: typing.List[str], - node: str, - pod_template + cli: CoreV1Api, interface_list: typing.List[str], node: str, pod_template ) -> None: """ Function that creates a privileged pod and uses it to create @@ -488,25 +430,20 @@ def create_virtual_interfaces( - The YAML template used to instantiate a pod to create virtual interfaces on the node """ - pod_body = yaml.safe_load( - pod_template.render(nodename=node) - ) + pod_body = yaml.safe_load(pod_template.render(nodename=node)) kube_helper.create_pod(cli, pod_body, "default", 300) logging.info( "Creating {0} virtual interfaces on node {1} using a pod".format( - len(interface_list), - node + len(interface_list), node ) ) - create_ifb(cli, len(interface_list), 'modtools') + create_ifb(cli, len(interface_list), "modtools") logging.info("Deleting pod used to create virtual interfaces") kube_helper.delete_pod(cli, "modtools", "default") def delete_virtual_interfaces( - cli: CoreV1Api, - node_list: typing.List[str], - pod_template + cli: CoreV1Api, node_list: typing.List[str], pod_template ): """ Function that creates a privileged pod and uses it to delete all @@ -529,14 +466,10 @@ def delete_virtual_interfaces( """ for node in node_list: - pod_body = yaml.safe_load( - pod_template.render(nodename=node) - ) + pod_body = yaml.safe_load(pod_template.render(nodename=node)) kube_helper.create_pod(cli, pod_body, "default", 300) - logging.info( - "Deleting all virtual interfaces on node {0}".format(node) - ) - delete_ifb(cli, 'modtools') + logging.info("Deleting all virtual interfaces on node {0}".format(node)) + delete_ifb(cli, "modtools") kube_helper.delete_pod(cli, "modtools", "default") @@ -546,21 +479,13 @@ def create_ifb(cli: CoreV1Api, number: int, pod_name: str): Makes use of modprobe commands """ - exec_command = [ - 'chroot', '/host', - 'modprobe', 'ifb', 'numifbs=' + str(number) - ] - kube_helper.exec_cmd_in_pod(cli, exec_command, pod_name, 'default') + exec_command = ["chroot", "/host", "modprobe", "ifb", "numifbs=" + str(number)] + kube_helper.exec_cmd_in_pod(cli, exec_command, pod_name, "default") for i in range(0, number): - exec_command = ['chroot', '/host', 'ip', 'link', 'set', 'dev'] - exec_command += ['ifb' + str(i), 'up'] - kube_helper.exec_cmd_in_pod( - cli, - exec_command, - pod_name, - 'default' - ) + exec_command = ["chroot", "/host", "ip", "link", "set", "dev"] + exec_command += ["ifb" + str(i), "up"] + kube_helper.exec_cmd_in_pod(cli, exec_command, pod_name, "default") def delete_ifb(cli: CoreV1Api, pod_name: str): @@ -569,8 +494,8 @@ def delete_ifb(cli: CoreV1Api, pod_name: str): Makes use of modprobe command """ - exec_command = ['chroot', '/host', 'modprobe', '-r', 'ifb'] - kube_helper.exec_cmd_in_pod(cli, exec_command, pod_name, 'default') + exec_command = ["chroot", "/host", "modprobe", "-r", "ifb"] + kube_helper.exec_cmd_in_pod(cli, exec_command, pod_name, "default") def get_job_pods(cli: CoreV1Api, api_response): @@ -591,18 +516,14 @@ def get_job_pods(cli: CoreV1Api, api_response): controllerUid = api_response.metadata.labels["controller-uid"] pod_label_selector = "controller-uid=" + controllerUid pods_list = kube_helper.list_pods( - cli, - label_selector=pod_label_selector, - namespace="default" + cli, label_selector=pod_label_selector, namespace="default" ) return pods_list[0] def wait_for_job( - batch_cli: BatchV1Api, - job_list: typing.List[str], - timeout: int = 300 + batch_cli: BatchV1Api, job_list: typing.List[str], timeout: int = 300 ) -> None: """ Function that waits for a list of jobs to finish within a time period @@ -625,13 +546,11 @@ def wait_for_job( for job_name in job_list: try: api_response = kube_helper.get_job_status( - batch_cli, - job_name, - namespace="default" + batch_cli, job_name, namespace="default" ) if ( - api_response.status.succeeded is not None or - api_response.status.failed is not None + api_response.status.succeeded is not None + or api_response.status.failed is not None ): count += 1 job_list.remove(job_name) @@ -645,11 +564,7 @@ def wait_for_job( time.sleep(5) -def delete_jobs( - cli: CoreV1Api, - batch_cli: BatchV1Api, - job_list: typing.List[str] -): +def delete_jobs(cli: CoreV1Api, batch_cli: BatchV1Api, job_list: typing.List[str]): """ Function that deletes jobs @@ -667,38 +582,28 @@ def delete_jobs( for job_name in job_list: try: api_response = kube_helper.get_job_status( - batch_cli, - job_name, - namespace="default" + batch_cli, job_name, namespace="default" ) if api_response.status.failed is not None: pod_name = get_job_pods(cli, api_response) - pod_stat = kube_helper.read_pod( - cli, - name=pod_name, - namespace="default" - ) + pod_stat = kube_helper.read_pod(cli, name=pod_name, namespace="default") logging.error(pod_stat.status.container_statuses) pod_log_response = kube_helper.get_pod_log( - cli, - name=pod_name, - namespace="default" + cli, name=pod_name, namespace="default" ) pod_log = pod_log_response.data.decode("utf-8") logging.error(pod_log) except Exception as e: logging.warn("Exception in getting job status: %s" % str(e)) api_response = kube_helper.delete_job( - batch_cli, - name=job_name, - namespace="default" + batch_cli, name=job_name, namespace="default" ) def get_ingress_cmd( interface_list: typing.List[str], network_parameters: typing.Dict[str, str], - duration: int = 300 + duration: int = 300, ): """ Function that returns the commands to the ingress traffic shaping on @@ -736,9 +641,7 @@ def get_ingress_cmd( for i, interface in enumerate(interface_list): if not interface_pattern.match(interface): - logging.error( - "Interface name can only consist of alphanumeric characters" - ) + logging.error("Interface name can only consist of alphanumeric characters") raise Exception( "Interface '{0}' does not match the required regex pattern :" r" ^[a-z0-9\-\@\_]+$".format(interface) @@ -752,33 +655,23 @@ def get_ingress_cmd( "follow the regex pattern ^ifb[0-9]+$".format(ifb_name) ) - tc_set += "tc qdisc add dev {0} handle ffff: ingress;".format( - interface - ) + tc_set += "tc qdisc add dev {0} handle ffff: ingress;".format(interface) tc_set += "tc filter add dev {0} parent ffff: protocol ip u32 match u32 0 0 action mirred egress redirect dev {1};".format( # noqa - interface, - ifb_name + interface, ifb_name ) tc_set = "{0} tc qdisc add dev {1} root netem".format(tc_set, ifb_name) tc_unset = "{0} tc qdisc del dev {1} root ;".format(tc_unset, ifb_name) - tc_unset += "tc qdisc del dev {0} handle ffff: ingress;".format( - interface - ) + tc_unset += "tc qdisc del dev {0} handle ffff: ingress;".format(interface) tc_ls = "{0} tc qdisc ls dev {1} ;".format(tc_ls, ifb_name) for parameter in network_parameters.keys(): tc_set += " {0} {1} ".format( - param_map[parameter], - network_parameters[parameter] + param_map[parameter], network_parameters[parameter] ) tc_set += ";" exec_cmd = "{0} {1} sleep {2};{3} sleep 20;{4}".format( - tc_set, - tc_ls, - duration, - tc_unset, - tc_ls + tc_set, tc_ls, duration, tc_unset, tc_ls ) return exec_cmd @@ -790,17 +683,14 @@ def get_ingress_cmd( description="Applies filters to ihe ingress side of node(s) interfaces", outputs={ "success": NetworkScenarioSuccessOutput, - "error": NetworkScenarioErrorOutput + "error": NetworkScenarioErrorOutput, }, ) -def network_chaos(cfg: NetworkScenarioConfig) -> typing.Tuple[ - str, - typing.Union[ - NetworkScenarioSuccessOutput, - NetworkScenarioErrorOutput - ] +def network_chaos( + cfg: NetworkScenarioConfig, +) -> typing.Tuple[ + str, typing.Union[NetworkScenarioSuccessOutput, NetworkScenarioErrorOutput] ]: - """ Function that performs the ingress network chaos scenario based on the provided configuration @@ -826,12 +716,10 @@ def network_chaos(cfg: NetworkScenarioConfig) -> typing.Tuple[ cfg.label_selector, cfg.instance_count, pod_interface_template, - cli + cli, ) except Exception: - return "error", NetworkScenarioErrorOutput( - format_exc() - ) + return "error", NetworkScenarioErrorOutput(format_exc()) job_list = [] publish = False if cfg.kraken_config: @@ -840,16 +728,12 @@ def network_chaos(cfg: NetworkScenarioConfig) -> typing.Tuple[ with open(cfg.kraken_config, "r") as f: config = yaml.full_load(f) except Exception: - logging.error( - "Error reading Kraken config from %s" % cfg.kraken_config - ) - return "error", NetworkScenarioErrorOutput( - format_exc() - ) + logging.error("Error reading Kraken config from %s" % cfg.kraken_config) + return "error", NetworkScenarioErrorOutput(format_exc()) publish = True try: - if cfg.execution_type == 'parallel': + if cfg.execution_type == "parallel": for node in node_interface_dict: job_list.append( apply_ingress_filter( @@ -859,22 +743,19 @@ def network_chaos(cfg: NetworkScenarioConfig) -> typing.Tuple[ pod_module_template, job_template, batch_cli, - cli + cli, ) ) logging.info("Waiting for parallel job to finish") start_time = int(time.time()) - wait_for_job(batch_cli, job_list[:], cfg.test_duration+100) + wait_for_job(batch_cli, job_list[:], cfg.test_duration + 100) end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, - failed_post_scenarios, - start_time, - end_time + config, failed_post_scenarios, start_time, end_time ) - elif cfg.execution_type == 'serial': + elif cfg.execution_type == "serial": create_interfaces = True for param in cfg.network_params: for node in node_interface_dict: @@ -888,50 +769,39 @@ def network_chaos(cfg: NetworkScenarioConfig) -> typing.Tuple[ batch_cli, cli, create_interfaces=create_interfaces, - param_selector=param + param_selector=param, ) ) logging.info("Waiting for serial job to finish") start_time = int(time.time()) - wait_for_job(batch_cli, job_list[:], cfg.test_duration+100) + wait_for_job(batch_cli, job_list[:], cfg.test_duration + 100) logging.info("Deleting jobs") delete_jobs(cli, batch_cli, job_list[:]) job_list = [] - logging.info( - "Waiting for wait_duration : %ss" % cfg.wait_duration - ) + logging.info("Waiting for wait_duration : %ss" % cfg.wait_duration) time.sleep(cfg.wait_duration) end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, - failed_post_scenarios, - start_time, - end_time + config, failed_post_scenarios, start_time, end_time ) create_interfaces = False else: return "error", NetworkScenarioErrorOutput( - "Invalid execution type - serial and parallel are " - "the only accepted types" - ) + "Invalid execution type - serial and parallel are " + "the only accepted types" + ) return "success", NetworkScenarioSuccessOutput( filter_direction="ingress", test_interfaces=node_interface_dict, network_parameters=cfg.network_params, - execution_type=cfg.execution_type + execution_type=cfg.execution_type, ) except Exception as e: logging.error("Network Chaos exiting due to Exception - %s" % e) - return "error", NetworkScenarioErrorOutput( - format_exc() - ) + return "error", NetworkScenarioErrorOutput(format_exc()) finally: - delete_virtual_interfaces( - cli, - node_interface_dict.keys(), - pod_module_template - ) + delete_virtual_interfaces(cli, node_interface_dict.keys(), pod_module_template) logging.info("Deleting jobs(if any)") delete_jobs(cli, batch_cli, job_list[:]) diff --git a/krkn/scenario_plugins/native/pod_network_outage/pod_network_outage_plugin.py b/krkn/scenario_plugins/native/pod_network_outage/pod_network_outage_plugin.py index 410a58b7..87d53464 100755 --- a/krkn/scenario_plugins/native/pod_network_outage/pod_network_outage_plugin.py +++ b/krkn/scenario_plugins/native/pod_network_outage/pod_network_outage_plugin.py @@ -42,8 +42,7 @@ def get_test_pods( pod names (string) in the namespace """ pods_list = [] - pods_list = kubecli.list_pods( - label_selector=pod_label, namespace=namespace) + pods_list = kubecli.list_pods(label_selector=pod_label, namespace=namespace) if pod_name and pod_name not in pods_list: raise Exception("pod name not found in namespace ") elif pod_name and pod_name in pods_list: @@ -92,8 +91,7 @@ def delete_jobs(kubecli: KrknKubernetes, job_list: typing.List[str]): for job_name in job_list: try: - api_response = kubecli.get_job_status( - job_name, namespace="default") + api_response = kubecli.get_job_status(job_name, namespace="default") if api_response.status.failed is not None: pod_name = get_job_pods(kubecli, api_response) pod_stat = kubecli.read_pod(name=pod_name, namespace="default") @@ -131,8 +129,7 @@ def wait_for_job( while count != job_len: for job_name in job_list: try: - api_response = kubecli.get_job_status( - job_name, namespace="default") + api_response = kubecli.get_job_status(job_name, namespace="default") if ( api_response.status.succeeded is not None or api_response.status.failed is not None @@ -149,8 +146,7 @@ def wait_for_job( time.sleep(5) -def get_bridge_name(cli: ApiextensionsV1Api, - custom_obj: CustomObjectsApi) -> str: +def get_bridge_name(cli: ApiextensionsV1Api, custom_obj: CustomObjectsApi) -> str: """ Function that gets OVS bridge present in node. @@ -328,16 +324,13 @@ def apply_ingress_policy( create_virtual_interfaces(kubecli, len(ips), node, pod_template) for count, pod_ip in enumerate(set(ips)): - pod_inf = get_pod_interface( - node, pod_ip, pod_template, bridge_name, kubecli) + pod_inf = get_pod_interface(node, pod_ip, pod_template, bridge_name, kubecli) exec_cmd = get_ingress_cmd( test_execution, pod_inf, mod, count, network_params, duration ) - logging.info("Executing %s on pod %s in node %s" % - (exec_cmd, pod_ip, node)) + logging.info("Executing %s on pod %s in node %s" % (exec_cmd, pod_ip, node)) job_body = yaml.safe_load( - job_template.render(jobname=mod + str(pod_ip), - nodename=node, cmd=exec_cmd) + job_template.render(jobname=mod + str(pod_ip), nodename=node, cmd=exec_cmd) ) job_list.append(job_body["metadata"]["name"]) api_response = kubecli.create_job(job_body) @@ -405,16 +398,13 @@ def apply_net_policy( job_list = [] for pod_ip in set(ips): - pod_inf = get_pod_interface( - node, pod_ip, pod_template, bridge_name, kubecli) + pod_inf = get_pod_interface(node, pod_ip, pod_template, bridge_name, kubecli) exec_cmd = get_egress_cmd( test_execution, pod_inf, mod, network_params, duration ) - logging.info("Executing %s on pod %s in node %s" % - (exec_cmd, pod_ip, node)) + logging.info("Executing %s on pod %s in node %s" % (exec_cmd, pod_ip, node)) job_body = yaml.safe_load( - job_template.render(jobname=mod + str(pod_ip), - nodename=node, cmd=exec_cmd) + job_template.render(jobname=mod + str(pod_ip), nodename=node, cmd=exec_cmd) ) job_list.append(job_body["metadata"]["name"]) api_response = kubecli.create_job(job_body) @@ -456,18 +446,16 @@ def get_ingress_cmd( Returns: str: ingress filter """ - ifb_dev = 'ifb{0}'.format(count) + ifb_dev = "ifb{0}".format(count) tc_set = tc_unset = tc_ls = "" param_map = {"latency": "delay", "loss": "loss", "bandwidth": "rate"} tc_set = "tc qdisc add dev {0} ingress ;".format(test_interface) tc_set = "{0} tc filter add dev {1} ingress matchall action mirred egress redirect dev {2} ;".format( - tc_set, test_interface, ifb_dev) - tc_set = "{0} tc qdisc replace dev {1} root netem".format( - tc_set, ifb_dev) - tc_unset = "{0} tc qdisc del dev {1} root ;".format( - tc_unset, ifb_dev) - tc_unset = "{0} tc qdisc del dev {1} ingress".format( - tc_unset, test_interface) + tc_set, test_interface, ifb_dev + ) + tc_set = "{0} tc qdisc replace dev {1} root netem".format(tc_set, ifb_dev) + tc_unset = "{0} tc qdisc del dev {1} root ;".format(tc_unset, ifb_dev) + tc_unset = "{0} tc qdisc del dev {1} ingress".format(tc_unset, test_interface) tc_ls = "{0} tc qdisc ls dev {1} ;".format(tc_ls, ifb_dev) if execution == "parallel": for val in vallst.keys(): @@ -475,8 +463,7 @@ def get_ingress_cmd( tc_set += ";" else: tc_set += " {0} {1} ;".format(param_map[mod], vallst[mod]) - exec_cmd = "{0} {1} sleep {2};{3}".format( - tc_set, tc_ls, duration, tc_unset) + exec_cmd = "{0} {1} sleep {2};{3}".format(tc_set, tc_ls, duration, tc_unset) return exec_cmd @@ -512,10 +499,8 @@ def get_egress_cmd( """ tc_set = tc_unset = tc_ls = "" param_map = {"latency": "delay", "loss": "loss", "bandwidth": "rate"} - tc_set = "{0} tc qdisc replace dev {1} root netem".format( - tc_set, test_interface) - tc_unset = "{0} tc qdisc del dev {1} root ;".format( - tc_unset, test_interface) + tc_set = "{0} tc qdisc replace dev {1} root netem".format(tc_set, test_interface) + tc_unset = "{0} tc qdisc del dev {1} root ;".format(tc_unset, test_interface) tc_ls = "{0} tc qdisc ls dev {1} ;".format(tc_ls, test_interface) if execution == "parallel": for val in vallst.keys(): @@ -523,17 +508,13 @@ def get_egress_cmd( tc_set += ";" else: tc_set += " {0} {1} ;".format(param_map[mod], vallst[mod]) - exec_cmd = "{0} {1} sleep {2};{3}".format( - tc_set, tc_ls, duration, tc_unset) + exec_cmd = "{0} {1} sleep {2};{3}".format(tc_set, tc_ls, duration, tc_unset) return exec_cmd def create_virtual_interfaces( - kubecli: KrknKubernetes, - nummber: int, - node: str, - pod_template + kubecli: KrknKubernetes, nummber: int, node: str, pod_template ) -> None: """ Function that creates a privileged pod and uses it to create @@ -554,25 +535,18 @@ def create_virtual_interfaces( - The YAML template used to instantiate a pod to create virtual interfaces on the node """ - pod_body = yaml.safe_load( - pod_template.render(nodename=node) - ) + pod_body = yaml.safe_load(pod_template.render(nodename=node)) kubecli.create_pod(pod_body, "default", 300) logging.info( - "Creating {0} virtual interfaces on node {1} using a pod".format( - nummber, - node - ) + "Creating {0} virtual interfaces on node {1} using a pod".format(nummber, node) ) - create_ifb(kubecli, nummber, 'modtools') + create_ifb(kubecli, nummber, "modtools") logging.info("Deleting pod used to create virtual interfaces") kubecli.delete_pod("modtools", "default") def delete_virtual_interfaces( - kubecli: KrknKubernetes, - node_list: typing.List[str], - pod_template + kubecli: KrknKubernetes, node_list: typing.List[str], pod_template ): """ Function that creates a privileged pod and uses it to delete all @@ -595,14 +569,10 @@ def delete_virtual_interfaces( """ for node in node_list: - pod_body = yaml.safe_load( - pod_template.render(nodename=node) - ) + pod_body = yaml.safe_load(pod_template.render(nodename=node)) kubecli.create_pod(pod_body, "default", 300) - logging.info( - "Deleting all virtual interfaces on node {0}".format(node) - ) - delete_ifb(kubecli, 'modtools') + logging.info("Deleting all virtual interfaces on node {0}".format(node)) + delete_ifb(kubecli, "modtools") kubecli.delete_pod("modtools", "default") @@ -612,24 +582,14 @@ def create_ifb(kubecli: KrknKubernetes, number: int, pod_name: str): Makes use of modprobe commands """ - exec_command = [ - '/host', - 'modprobe', 'ifb', 'numifbs=' + str(number) - ] - kubecli.exec_cmd_in_pod( - exec_command, - pod_name, - 'default', - base_command="chroot") + exec_command = ["/host", "modprobe", "ifb", "numifbs=" + str(number)] + kubecli.exec_cmd_in_pod(exec_command, pod_name, "default", base_command="chroot") for i in range(0, number): - exec_command = ['/host', 'ip', 'link', 'set', 'dev'] - exec_command += ['ifb' + str(i), 'up'] + exec_command = ["/host", "ip", "link", "set", "dev"] + exec_command += ["ifb" + str(i), "up"] kubecli.exec_cmd_in_pod( - exec_command, - pod_name, - 'default', - base_command="chroot" + exec_command, pod_name, "default", base_command="chroot" ) @@ -639,17 +599,11 @@ def delete_ifb(kubecli: KrknKubernetes, pod_name: str): Makes use of modprobe command """ - exec_command = ['/host', 'modprobe', '-r', 'ifb'] - kubecli.exec_cmd_in_pod( - exec_command, - pod_name, - 'default', - base_command="chroot") + exec_command = ["/host", "modprobe", "-r", "ifb"] + kubecli.exec_cmd_in_pod(exec_command, pod_name, "default", base_command="chroot") -def list_bridges( - node: str, pod_template, kubecli: KrknKubernetes -) -> typing.List[str]: +def list_bridges(node: str, pod_template, kubecli: KrknKubernetes) -> typing.List[str]: """ Function that returns a list of bridges on the node @@ -787,7 +741,7 @@ def get_pod_interface( find_ip = f"external-ids:ip_addresses={ip}/23" else: find_ip = f"external-ids:ip={ip}" - + cmd = [ "/host", "ovs-vsctl", @@ -797,24 +751,20 @@ def get_pod_interface( "interface", find_ip, ] - + output = kubecli.exec_cmd_in_pod( cmd, "modtools", "default", base_command="chroot" ) if not output: - cmd= [ - "/host", - "ip", - "addr", - "show" - ] + cmd = ["/host", "ip", "addr", "show"] output = kubecli.exec_cmd_in_pod( - cmd, "modtools", "default", base_command="chroot") + cmd, "modtools", "default", base_command="chroot" + ) for if_str in output.split("\n"): - if re.search(ip,if_str): - inf = if_str.split(' ')[-1] + if re.search(ip, if_str): + inf = if_str.split(" ")[-1] else: - inf = output + inf = output finally: logging.info("Deleting pod to query interface on node") kubecli.delete_pod("modtools", "default") @@ -927,11 +877,11 @@ class InputParams: }, ) - kraken_config: typing.Optional[str] = field( + kraken_config: typing.Dict[str, typing.Any] = field( default=None, metadata={ "name": "Kraken Config", - "description": "Path to the config file of Kraken. " + "description": "Kraken config file dictionary " "Set this field if you wish to publish status onto Cerberus", }, ) @@ -1043,14 +993,6 @@ def pod_outage( publish = False if params.kraken_config: - failed_post_scenarios = "" - try: - with open(params.kraken_config, "r") as f: - config = yaml.full_load(f) - except Exception: - logging.error("Error reading Kraken config from %s" % - params.kraken_config) - return "error", PodOutageErrorOutput(format_exc()) publish = True for i in params.direction: @@ -1106,7 +1048,7 @@ def pod_outage( end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, failed_post_scenarios, start_time, end_time + params.kraken_config, "", start_time, end_time ) return "success", PodOutageSuccessOutput( @@ -1116,8 +1058,7 @@ def pod_outage( egress_ports=params.egress_ports, ) except Exception as e: - logging.error( - "Pod network outage scenario exiting due to Exception - %s" % e) + logging.error("Pod network outage scenario exiting due to Exception - %s" % e) return "error", PodOutageErrorOutput(format_exc()) finally: logging.info("Deleting jobs(if any)") @@ -1179,11 +1120,11 @@ class EgressParams: }, ) - kraken_config: typing.Optional[str] = field( + kraken_config: typing.Dict[str, typing.Any] = field( default=None, metadata={ "name": "Kraken Config", - "description": "Path to the config file of Kraken. " + "description": "Krkn config file dictionary " "Set this field if you wish to publish status onto Cerberus", }, ) @@ -1276,8 +1217,7 @@ class PodEgressNetShapingErrorOutput: def pod_egress_shaping( params: EgressParams, ) -> typing.Tuple[ - str, typing.Union[PodEgressNetShapingSuccessOutput, - PodEgressNetShapingErrorOutput] + str, typing.Union[PodEgressNetShapingSuccessOutput, PodEgressNetShapingErrorOutput] ]: """ Function that performs egress pod traffic shaping based @@ -1302,14 +1242,6 @@ def pod_egress_shaping( publish = False if params.kraken_config: - failed_post_scenarios = "" - try: - with open(params.kraken_config, "r") as f: - config = yaml.full_load(f) - except Exception: - logging.error("Error reading Kraken config from %s" % - params.kraken_config) - return "error", PodEgressNetShapingErrorOutput(format_exc()) publish = True try: @@ -1344,30 +1276,30 @@ def pod_egress_shaping( for mod in mod_lst: for node, ips in node_dict.items(): - job_list.extend( apply_net_policy( - mod, - node, - ips, - job_template, - pod_module_template, - params.network_params, - params.test_duration, - br_name, - kubecli, - params.execution_type, - )) + job_list.extend( + apply_net_policy( + mod, + node, + ips, + job_template, + pod_module_template, + params.network_params, + params.test_duration, + br_name, + kubecli, + params.execution_type, + ) + ) if params.execution_type == "serial": logging.info("Waiting for serial job to finish") start_time = int(time.time()) - wait_for_job(job_list[:], kubecli, - params.test_duration + 20) - logging.info("Waiting for wait_duration %s" % - params.test_duration) + wait_for_job(job_list[:], kubecli, params.test_duration + 20) + logging.info("Waiting for wait_duration %s" % params.test_duration) time.sleep(params.test_duration) end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, failed_post_scenarios, start_time, end_time + params.kraken_config, "", start_time, end_time ) if params.execution_type == "parallel": break @@ -1380,7 +1312,7 @@ def pod_egress_shaping( end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, failed_post_scenarios, start_time, end_time + params.kraken_config, "", start_time, end_time ) return "success", PodEgressNetShapingSuccessOutput( @@ -1389,8 +1321,7 @@ def pod_egress_shaping( execution_type=params.execution_type, ) except Exception as e: - logging.error( - "Pod network Shaping scenario exiting due to Exception - %s" % e) + logging.error("Pod network Shaping scenario exiting due to Exception - %s" % e) return "error", PodEgressNetShapingErrorOutput(format_exc()) finally: logging.info("Deleting jobs(if any)") @@ -1452,7 +1383,7 @@ class IngressParams: }, ) - kraken_config: typing.Optional[str] = field( + kraken_config: typing.Dict[str, typing.Any] = field( default=None, metadata={ "name": "Kraken Config", @@ -1549,8 +1480,8 @@ class PodIngressNetShapingErrorOutput: def pod_ingress_shaping( params: IngressParams, ) -> typing.Tuple[ - str, typing.Union[PodIngressNetShapingSuccessOutput, - PodIngressNetShapingErrorOutput] + str, + typing.Union[PodIngressNetShapingSuccessOutput, PodIngressNetShapingErrorOutput], ]: """ Function that performs ingress pod traffic shaping based @@ -1575,14 +1506,6 @@ def pod_ingress_shaping( publish = False if params.kraken_config: - failed_post_scenarios = "" - try: - with open(params.kraken_config, "r") as f: - config = yaml.full_load(f) - except Exception: - logging.error("Error reading Kraken config from %s" % - params.kraken_config) - return "error", PodIngressNetShapingErrorOutput(format_exc()) publish = True try: @@ -1617,30 +1540,30 @@ def pod_ingress_shaping( for mod in mod_lst: for node, ips in node_dict.items(): - job_list.extend(apply_ingress_policy( - mod, - node, - ips, - job_template, - pod_module_template, - params.network_params, - params.test_duration, - br_name, - kubecli, - params.execution_type, - )) + job_list.extend( + apply_ingress_policy( + mod, + node, + ips, + job_template, + pod_module_template, + params.network_params, + params.test_duration, + br_name, + kubecli, + params.execution_type, + ) + ) if params.execution_type == "serial": logging.info("Waiting for serial job to finish") start_time = int(time.time()) - wait_for_job(job_list[:], kubecli, - params.test_duration + 20) - logging.info("Waiting for wait_duration %s" % - params.test_duration) + wait_for_job(job_list[:], kubecli, params.test_duration + 20) + logging.info("Waiting for wait_duration %s" % params.test_duration) time.sleep(params.test_duration) end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, failed_post_scenarios, start_time, end_time + params.kraken_config, "", start_time, end_time ) if params.execution_type == "parallel": break @@ -1653,7 +1576,7 @@ def pod_ingress_shaping( end_time = int(time.time()) if publish: cerberus.publish_kraken_status( - config, failed_post_scenarios, start_time, end_time + params.kraken_config, "", start_time, end_time ) return "success", PodIngressNetShapingSuccessOutput( @@ -1662,14 +1585,9 @@ def pod_ingress_shaping( execution_type=params.execution_type, ) except Exception as e: - logging.error( - "Pod network Shaping scenario exiting due to Exception - %s" % e) + logging.error("Pod network Shaping scenario exiting due to Exception - %s" % e) return "error", PodIngressNetShapingErrorOutput(format_exc()) finally: - delete_virtual_interfaces( - kubecli, - node_dict.keys(), - pod_module_template - ) + delete_virtual_interfaces(kubecli, node_dict.keys(), pod_module_template) logging.info("Deleting jobs(if any)") delete_jobs(kubecli, job_list[:])