<reponame>srinivasreddych/aws-orbit-workbench<gh_stars>10-100
#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License").
#    You may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

from typing import Any, Dict, List, Optional

import kopf
from kubernetes import dynamic
from kubernetes.client.rest import ApiException
from kubernetes.dynamic.client import DynamicClient

KUBEFLOW_API_GROUP = "kubeflow.org"
KUBEFLOW_API_VERSION = "v1alpha1"


def construct(
    name: str,
    desc: str,
    owner_reference: Optional[Dict[str, str]] = None,
    labels: Optional[Dict[str, str]] = None,
    annnotations: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    poddefault: Dict[str, Any] = {
        "apiVersion": f"{KUBEFLOW_API_GROUP}/{KUBEFLOW_API_VERSION}",
        "kind": "PodDefault",
        "metadata": {
            "name": name,
            "labels": labels,
            "annotations": annnotations,
        },
        "spec": {"selector": {"matchLabels": {f"orbit/{name}": ""}}, "desc": desc},
    }
    if owner_reference is not None:
        poddefault["metadata"]["ownerReferences"] = [owner_reference]
    return poddefault


def create_poddefault(
    namespace: str,
    poddefault: Dict[str, Any],
    client: dynamic.DynamicClient,
    logger: kopf.Logger,
) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    api.create(namespace=namespace, body=poddefault)
    logger.debug(
        "Created PodDefault: %s in Namespace: %s",
        poddefault["metadata"]["name"],
        namespace,
    )


def modify_poddefault(
    namespace: str,
    name: str,
    desc: str,
    client: dynamic.DynamicClient,
    logger: kopf.Logger,
) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    patch = {"spec": {"desc": desc}}
    api.patch(namespace=namespace, name=name, body=patch)
    logger.debug("Modified PodDefault: %s in Namespace: %s", name, namespace)


def delete_poddefault(namespace: str, name: str, client: dynamic.DynamicClient, logger: kopf.Logger) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    api.delete(namespace=namespace, name=name, body={})
    logger.debug("Deleted PodDefault: %s in Namesapce: %s", name, namespace)


def copy_poddefaults_to_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Copying PodDefaults %s to user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                kwargs = {
                    "name": poddefault["metadata"]["name"],
                    "desc": poddefault["spec"]["desc"],
                    "labels": {
                        "orbit/space": "user",
                        "orbit/team": poddefault["metadata"]["labels"].get("orbit/team", None),
                    },
                }
                create_poddefault(
                    namespace=namespace,
                    poddefault=construct(**kwargs),
                    client=client,
                    logger=logger,
                )
            except ApiException as e:
                logger.warning(
                    "Unable to create PodDefault %s in Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e.body),
                )
            except Exception as e:
                logger.warning(
                    "Failed to create PodDefault",
                    str(e),
                )


def modify_poddefaults_in_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Modifying PodDefaults %s in user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                modify_poddefault(
                    namespace=namespace,
                    name=poddefault["metadata"]["name"],
                    desc=poddefault["spec"]["desc"],
                    client=client,
                    logger=logger,
                )
            except Exception as e:
                logger.warn(
                    "Unable to delete PodDefault %s from Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e),
                )


def delete_poddefaults_from_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Deleting PodDefaults %s from user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                delete_poddefault(
                    namespace=namespace,
                    name=poddefault["metadata"]["name"],
                    client=client,
                    logger=logger,
                )
            except Exception as e:
                logger.warn(
                    "Unable to delete PodDefault %s from Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e),
                )

#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License").
#    You may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

from typing import Any, Dict, List, Optional

import kopf
from kubernetes import dynamic
from kubernetes.client.rest import ApiException
from kubernetes.dynamic.client import DynamicClient

KUBEFLOW_API_GROUP = "kubeflow.org"
KUBEFLOW_API_VERSION = "v1alpha1"


def construct(
    name: str,
    desc: str,
    owner_reference: Optional[Dict[str, str]] = None,
    labels: Optional[Dict[str, str]] = None,
    annnotations: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    poddefault: Dict[str, Any] = {
        "apiVersion": f"{KUBEFLOW_API_GROUP}/{KUBEFLOW_API_VERSION}",
        "kind": "PodDefault",
        "metadata": {
            "name": name,
            "labels": labels,
            "annotations": annnotations,
        },
        "spec": {"selector": {"matchLabels": {f"orbit/{name}": ""}}, "desc": desc},
    }
    if owner_reference is not None:
        poddefault["metadata"]["ownerReferences"] = [owner_reference]
    return poddefault


def create_poddefault(
    namespace: str,
    poddefault: Dict[str, Any],
    client: dynamic.DynamicClient,
    logger: kopf.Logger,
) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    api.create(namespace=namespace, body=poddefault)
    logger.debug(
        "Created PodDefault: %s in Namespace: %s",
        poddefault["metadata"]["name"],
        namespace,
    )


def modify_poddefault(
    namespace: str,
    name: str,
    desc: str,
    client: dynamic.DynamicClient,
    logger: kopf.Logger,
) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    patch = {"spec": {"desc": desc}}
    api.patch(namespace=namespace, name=name, body=patch)
    logger.debug("Modified PodDefault: %s in Namespace: %s", name, namespace)


def delete_poddefault(namespace: str, name: str, client: dynamic.DynamicClient, logger: kopf.Logger) -> None:
    api = client.resources.get(api_version=KUBEFLOW_API_VERSION, group=KUBEFLOW_API_GROUP, kind="PodDefault")
    api.delete(namespace=namespace, name=name, body={})
    logger.debug("Deleted PodDefault: %s in Namesapce: %s", name, namespace)


def copy_poddefaults_to_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Copying PodDefaults %s to user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                kwargs = {
                    "name": poddefault["metadata"]["name"],
                    "desc": poddefault["spec"]["desc"],
                    "labels": {
                        "orbit/space": "user",
                        "orbit/team": poddefault["metadata"]["labels"].get("orbit/team", None),
                    },
                }
                create_poddefault(
                    namespace=namespace,
                    poddefault=construct(**kwargs),
                    client=client,
                    logger=logger,
                )
            except ApiException as e:
                logger.warning(
                    "Unable to create PodDefault %s in Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e.body),
                )
            except Exception as e:
                logger.warning(
                    "Failed to create PodDefault",
                    str(e),
                )


def modify_poddefaults_in_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Modifying PodDefaults %s in user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                modify_poddefault(
                    namespace=namespace,
                    name=poddefault["metadata"]["name"],
                    desc=poddefault["spec"]["desc"],
                    client=client,
                    logger=logger,
                )
            except Exception as e:
                logger.warn(
                    "Unable to delete PodDefault %s from Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e),
                )


def delete_poddefaults_from_user_namespaces(
    poddefaults: List[Dict[str, Any]],
    user_namespaces: List[str],
    client: DynamicClient,
    logger: kopf.Logger,
) -> None:
    logger.debug(
        "Deleting PodDefaults %s from user Namespaces %s",
        [pd["metadata"]["name"] for pd in poddefaults],
        user_namespaces,
    )
    for poddefault in poddefaults:
        for namespace in user_namespaces:
            try:
                delete_poddefault(
                    namespace=namespace,
                    name=poddefault["metadata"]["name"],
                    client=client,
                    logger=logger,
                )
            except Exception as e:
                logger.warn(
                    "Unable to delete PodDefault %s from Namespace %s: %s",
                    poddefault["metadata"]["name"],
                    namespace,
                    str(e),
                )
#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # #    Licensed under the Apache License, Version 2.0 (the "License"). #    You may not use this file except in compliance with the License. #    You may obtain a copy of the License at # #        http://www.apache.org/licenses/LICENSE-2.0 # #    Unless required by applicable law or agreed to in writing, software #    distributed under the License is distributed on an "AS IS" BASIS, #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #    See the License for the specific language governing permissions and #    limitations under the License.