def validate()

in pbspro/src/pbspro/cli.py [0:0]

• 72 lines of code
• 13 McCabe index (conditional complexity)

    def validate(self, config: Dict) -> None:
        """
        Best-effort validation of your PBS environment's compatibility with this autoscaler.
        """
        pbs_driver = PBSProDriver(config)
        pbs_env = self._pbs_env(pbs_driver)
        sched = pbs_env.default_scheduler
        if not sched:
            print("Could not find a default server.", file=sys.stderr)
            sys.exit(1)

        exit = 0

        for attr in ["ungrouped", "group_id"]:
            if attr not in sched.resources_for_scheduling:
                print(
                    "{} is not defined for line 'resources:' in {}/sched_priv.".format(
                        attr, sched.sched_priv
                    )
                    + " Please add this and restart PBS"
                )
                exit = 1

        if sched.node_group_key and not sched.node_group_enable:
            print(
                "node_group_key is set to '{}' but node_group_enable is false".format(
                    sched.node_group_key
                ),
                file=sys.stderr,
            )
            exit = 1
        elif not sched.node_group_enable:
            print(
                "node_group_enable is false, so MPI/parallel jobs may not work if multiple placement groups are created.",
                file=sys.stderr,
            )
            exit = 1

        if not sched.only_explicit_psets:
            print(
                "only_explicit_psets should be set to true in your sched_config if you are using MPI or colocated jobs.",
                file=sys.stderr,
            )
            exit = 1

        if not sched.do_not_span_psets:
            print(
                "do_not_span_psets should be set to true in your sched_config if you are using MPI or colocated jobs.",
                file=sys.stderr,
            )
            exit = 1

        jetpack_path = which("jetpack")
        if jetpack_path:
            key = "cyclecloud.hosts.standalone_dns.enabled"
            jetpack_config = json.loads(
                check_output(["jetpack", "config", "--json", key]).decode()
            )
            if jetpack_config.get(key):
                dcalc, _ = self._demand_calc(config, pbs_driver)

                for bucket in dcalc.node_mgr.get_buckets():
                    if not is_standalone_dns(bucket):
                        print(
                            "Nodearray %s has %s=false, but this host has %s=true. Because of this, /etc/hosts was generated with static entries for every possible address in this subnet."
                            % (bucket.nodearray, key, key),
                            file=sys.stderr,
                            end=" ",
                        )
                        print(
                            "Please ensure that all entries after '#The following was autogenerated for Cloud environments.  (Subnet: ...)' are either commented out or deleted.",
                            file=sys.stderr,
                            end=" ",
                        )
                        print(
                            "For future clusters, set %s=false under the scheduler's configuration section in the template."
                            % (key),
                            file=sys.stderr,
                        )
                        exit = 1
                        break

        sys.exit(exit)