import argparse
import json
import logging
import os

from ray._private.runtime_env.context import RuntimeEnvContext
from ray.core.generated.common_pb2 import Language

logger = logging.getLogger(__name__)

parser = argparse.ArgumentParser(
    description=(
        "Set up the environment for a Ray worker and launch the worker."))

parser.add_argument(
    "--serialized-runtime-env",
    type=str,
    help="the serialized parsed runtime env dict")

parser.add_argument(
    "--serialized-runtime-env-context",
    type=str,
    help="the serialized runtime env context")

parser.add_argument(
    "--allocated-instances-serialized-json",
    type=str,
    help="the worker allocated resource")

parser.add_argument(
    "--language", type=str, help="the language type of the worker")


def get_tmp_dir(remaining_args):
    for arg in remaining_args:
        if arg.startswith("--temp-dir="):
            return arg[11:]
    return None


def parse_allocated_resource(allocated_instances_serialized_json):
    container_resource_args = []
    allocated_resource = json.loads(allocated_instances_serialized_json)
    if "CPU" in allocated_resource.keys():
        cpu_resource = allocated_resource["CPU"]
        if isinstance(cpu_resource, list):
            # cpuset: because we may split one cpu core into some pieces,
            # we need set cpuset.cpu_exclusive=0 and set cpuset-cpus
            cpu_ids = []
            cpu_shares = 0
            for idx, val in enumerate(cpu_resource):
                if val > 0:
                    cpu_ids.append(idx)
                    cpu_shares += val
            container_resource_args.append("--cpu-shares=" +
                                           str(int(cpu_shares / 10000 * 1024)))
            container_resource_args.append("--cpuset-cpus=" + ",".join(
                str(e) for e in cpu_ids))
        else:
            # cpushare
            container_resource_args.append(
                "--cpu-shares=" + str(int(cpu_resource / 10000 * 1024)))
    if "memory" in allocated_resource.keys():
        container_resource_args.append(
            "--memory=" + str(int(allocated_resource["memory"] / 10000)))
    return container_resource_args


def start_worker_in_container(container_option, args, remaining_args):
    worker_setup_hook = args.worker_setup_hook
    last_period_idx = worker_setup_hook.rfind(".")
    module_name = worker_setup_hook[:last_period_idx]
    # python -m ray.workers.setup_runtime_env --session-dir=
    # default_worker.py --node-ip-address= ...
    entrypoint_args = ["-m"]
    entrypoint_args.append(module_name)
    # replace default_worker.py path
    if container_option.get("worker_path"):
        remaining_args[1] = container_option.get("worker_path")
    entrypoint_args.extend(remaining_args)
    # now we will start a container, add argument worker-shim-pid
    entrypoint_args.append("--worker-shim-pid={}".format(os.getpid()))

    tmp_dir = get_tmp_dir(remaining_args)
    if not tmp_dir:
        logger.error(
            "failed to get tmp_dir, the args: {}".format(remaining_args))

    container_driver = "podman"
    # todo add cgroup config
    # todo flag "--rm"
    container_command = [
        container_driver, "run", "-v", tmp_dir + ":" + tmp_dir,
        "--cgroup-manager=cgroupfs", "--network=host", "--pid=host",
        "--ipc=host", "--env-host"
    ]
    container_command.append("--env")
    container_command.append("RAY_RAYLET_PID=" + str(os.getppid()))
    if container_option.get("run_options"):
        container_command.extend(container_option.get("run_options"))
    container_command.extend(
        parse_allocated_resource(args.allocated_instances_serialized_json))

    container_command.append("--entrypoint")
    container_command.append("python")
    container_command.append(container_option.get("image"))
    container_command.extend(entrypoint_args)
    logger.warning("start worker in container: {}".format(container_command))
    os.execvp(container_driver, container_command)


if __name__ == "__main__":
    args, remaining_args = parser.parse_known_args()
    runtime_env: dict = json.loads(args.serialized_runtime_env or "{}")
    container_option = runtime_env.get("container")
    if container_option and container_option.get("image"):
        start_worker_in_container(container_option, args, remaining_args)
    else:
        # NOTE(edoakes): args.serialized_runtime_env_context is only None when
        # we're starting the main Ray client proxy server. That case should
        # probably not even go through this codepath.
        runtime_env_context = RuntimeEnvContext.deserialize(
            args.serialized_runtime_env_context or "{}")
        runtime_env_context.exec_worker(remaining_args,
                                        Language.Value(args.language))
