def deploy_env()

in cli/aws_orbit/remote_files/kubectl.py [0:0]

• 79 lines of code
• 4 McCabe index (conditional complexity)

def deploy_env(context: "Context") -> None:
    eks_stack_name: str = f"eksctl-orbit-{context.name}-cluster"
    _logger.debug("EKSCTL stack name: %s", eks_stack_name)
    if cfn.does_stack_exist(stack_name=eks_stack_name):
        k8s_context = get_k8s_context(context=context)
        _logger.debug("k8s_context: %s", k8s_context)

        # orbit-system kustomizations
        output_paths = _generate_orbit_system_kustomizations(context=context)
        for path in output_paths:
            sh.run(f"kubectl apply -k {path} --context {k8s_context} --wait")

        # Wait until cert-manager webhook is available
        _confirm_endpoints(name="cert-manager-webhook", namespace="cert-manager", k8s_context=k8s_context)
        _confirm_readiness(name="cert-manager", namespace="cert-manager", type="Deployment", k8s_context=k8s_context)
        _confirm_readiness(
            name="cert-manager-cainjector", namespace="cert-manager", type="Deployment", k8s_context=k8s_context
        )

        output_path: Optional[str] = _generate_orbit_system_manifest(context=context, clean_up=True)
        sh.run(f"kubectl apply -f {output_path} --context {k8s_context} --wait")

        output_path = _generate_orbit_image_replicator_manifest(context=context, clean_up=True)
        if output_path is not None:
            sh.run(f"kubectl apply -f {output_path} --context {k8s_context} --wait")

        # Commented until we confirm this isn't needed
        # Restart orbit-system deployments and statefulsets to force reload of caches etc
        # sh.run(f"kubectl rollout restart deployments -n orbit-system --context {k8s_context}")

        _confirm_readiness(
            name="podsetting-operator", namespace="orbit-system", type="deployment", k8s_context=k8s_context
        )
        _confirm_readiness(
            name="teamspace-operator", namespace="orbit-system", type="deployment", k8s_context=k8s_context
        )
        _confirm_readiness(
            name="userspace-operator", namespace="orbit-system", type="deployment", k8s_context=k8s_context
        )
        _confirm_endpoints(name="podsetting-pod-webhook", namespace="orbit-system", k8s_context=k8s_context)

        if context.install_image_replicator or not context.networking.data.internet_accessible:
            _confirm_readiness(
                name="imagereplication-operator", namespace="orbit-system", type="deployment", k8s_context=k8s_context
            )
            _confirm_endpoints(name="imagereplication-pod-webhook", namespace="orbit-system", k8s_context=k8s_context)
            sh.run(
                "kubectl rollout restart daemonsets -n orbit-system-ssm-daemons "
                f"ssm-agent-installer --context {k8s_context}"
            )

        # kube-system kustomizations
        output_paths = _generate_kube_system_kustomizations(context=context)
        for output_path in output_paths:
            sh.run(f"kubectl apply -k {output_path} --context {k8s_context} --wait")

        # kube-system manifests
        output_path = _generate_kube_system_manifest(context=context)
        sh.run(f"kubectl apply -f {output_path} --context {k8s_context} --wait")

        # Enable ENIs
        _enable_eni(k8s_context=k8s_context)

        # kubeflow-namespaces
        output_path = _kubeflow_namespaces(context=context)
        sh.run(f"kubectl apply -f {output_path} --context {k8s_context} --wait")

        kubeflow.deploy_kubeflow(context=context)

        # env
        output_paths = _generate_orbit_system_env_kustomizations(context=context)
        for output_path in output_paths:
            sh.run(f"kubectl apply -k {output_path} --context {k8s_context} --wait")

        # Patch Kubeflow
        _logger.debug("Orbit applying KubeFlow patch")
        jupyter_launcher_config_map, patch = _generate_kubeflow_patch(context=context)
        sh.run(f"kubectl apply -f {jupyter_launcher_config_map} --context {k8s_context} --wait")
        sh.run(f'kubectl patch deployment -n kubeflow jupyter-web-app-deployment --patch "{patch}"')
        sh.run("kubectl rollout restart deployment jupyter-web-app-deployment -n kubeflow")

        _apply_deployment_patch_force_env_nodes("istio-system")
        _apply_deployment_patch_force_env_nodes("knative-serving")
        _apply_deployment_patch_force_env_nodes("kube-system")
        _apply_deployment_patch_force_env_nodes("kubeflow")

        # Patch Pods to push into Fargate when deploying in an isolated subnet
        if not context.networking.data.internet_accessible:
            patch = (
                '{"spec":{"template":{"metadata":{"labels":{"orbit/node-type":"fargate"}},'
                '"spec":{"nodeSelector": null}}}}'
            )
            sh.run(f"kubectl patch deployment -n istio-system authzadaptor --patch '{patch}'")

            patch = (
                '{"spec":{"template":{"metadata":{"labels":{"orbit/node-type":"fargate"}},'
                '"spec":{"nodeSelector": null, "containers":[{"name":"alb-ingress-controller","args":'
                '["--ingress-class=alb","--cluster-name=$(CLUSTER_NAME)","--aws-vpc-id=VPC_ID"]}]}}}}'
            )
            patch = patch.replace("VPC_ID", cast(str, context.networking.vpc_id))
            sh.run(f"kubectl patch deployment -n kubeflow alb-ingress-controller --patch '{patch}'")

        # Patch the kubeflow mpi-operator deployment to version lock the images to v0.2.3
        patch = (
            '{"spec":{"template":{"spec":{"containers":[{"name":"mpi-operator","args":["-alsologtostderr",'
            '"--lock-namespace","kubeflow","--kubectl-delivery-image","mpioperator/kubectl-delivery:v0.2.3"],'
            '"image":"mpioperator/mpi-operator:v0.2.3"}]}}}}'
        )
        sh.run(f"kubectl patch deployment -n kubeflow mpi-operator --patch '{patch}'")

        # Confirm env Service Endpoints
        _confirm_endpoints(name="landing-page-service", namespace="orbit-system", k8s_context=k8s_context)