code: def master_up(
      port: int,
      master_config_path: Optional[pathlib.Path],
      storage_host_path: pathlib.Path,
      master_name: Optional[str],
      image_repo_prefix: str,
      version: str,
      db_password: str,
      delete_db: bool,
      autorestart: bool,
      cluster_name: str,
      auto_work_dir: Optional[pathlib.Path],
  ) -> None:
      # Some cli flags for det deploy local will cause us to write a temporary master.yaml.
      make_temp_conf = False
  
      if master_name is None:
          master_name = f"{cluster_name}_{MASTER_NAME}"
  
      if master_config_path is not None:
          with master_config_path.open() as f:
              master_conf = util.yaml_safe_load(f)
      else:
          master_conf = MASTER_CONF_DEFAULT
          make_temp_conf = True
  
      if storage_host_path is not None:
          master_conf["checkpoint_storage"] = {
              "type": "shared_fs",
              "host_path": str(storage_host_path.resolve()),
          }
          make_temp_conf = True
  
      # Ensure checkpoint storage directory exists.
      storage_info = master_conf.get("checkpoint_storage", {})
      final_storage_host_path = storage_info.get("host_path")
      container_storage_path = (
          storage_info.get("container_path")
          or constants.SHARED_FS_CONTAINER_PATH
      )
  
      if final_storage_host_path is not None:
          final_storage_host_path = pathlib.Path(
              final_storage_host_path
          )
          if not final_storage_host_path.exists():
              final_storage_host_path.mkdir(parents=True)
  
      if auto_work_dir is not None:
          work_dir = str(auto_work_dir.resolve())
          master_conf.setdefault("task_container_defaults", {})[
              "work_dir"
          ] = work_dir
          master_conf["task_container_defaults"].setdefault(
              "bind_mounts", []
          ).append({"host_path": work_dir, "container_path": work_dir})
          make_temp_conf = True
  
      if make_temp_conf:
          fd, temp_path = tempfile.mkstemp(
              prefix="det-deploy-local-master-config-"
          )
          with open(fd, "w") as f:
              util.yaml_safe_dump(master_conf, f)
          master_config_path = pathlib.Path(temp_path)
  
      # This is always true by now, but mypy needs help.
      assert master_config_path is not None
      restart_policy = "unless-stopped" if autorestart else "no"
      env = {
          "DET_MASTER_HTTP_PORT": str(port),
          "DET_DB_PASSWORD": db_password,
          "DET_LOG_INFO": "info",
      }
  
      # Kill existing master container if exists.
      master_down(
          master_name=master_name,
          delete_db=delete_db,
          cluster_name=cluster_name,
      )
  
      # Create network.
      client = docker_client()
  
      # Start db and wait for healthcheck.
      db_name = f"{cluster_name}_{DB_NAME}"
      volume_name = f"{cluster_name}_{VOLUME_NAME}"
  
      @contextlib.contextmanager
      def defer_cleanup(fn: Callable[[], None]) -> Generator:
          """
          Defer cleanup tasks for each resource if Exceptions are caught.
          """
          try:
              yield
          except Exception as ex:
              fn()
              raise ex
  
      with contextlib.ExitStack() as exit_stack:
          # Create network used by DB and master.
          print(f"Creating network {NETWORK_NAME}...")
          exit_stack.enter_context(
              defer_cleanup(lambda: remove_network(NETWORK_NAME))
          )
          client.networks.create(name=NETWORK_NAME, attachable=True)
  
          # Start up db.
          exit_stack.enter_context(
              defer_cleanup(
                  lambda: db_down(db_name, VOLUME_NAME, delete_db)
              )
          )
          db_up(
              name=db_name,
              password=db_password,
              network_name=NETWORK_NAME,
              cluster_name=cluster_name,
              volume_name=volume_name,
          )
  
          # Wait for db to reach a healthy state.
          _wait_for_container(db_name, timeout=5)
  
          # Remove cleanup methods from ExitStack after DB successfully starts.
          exit_stack.pop_all()
  
          # Start master instance.
          print(f"Creating {master_name}...")
          exit_stack.enter_context(
              defer_cleanup(lambda: master_down(master_name, delete_db, cluster_name))  # type: ignore
          )
          volumes = [
              f"{os.path.abspath(master_config_path)}:/etc/determined/master.yaml",
              f"{final_storage_host_path}:{container_storage_path}",
          ]
          client.containers.run(
              image=f"{image_repo_prefix}/determined-master:{version}",
              environment=env,
              init=True,
              mounts=[],
              volumes=volumes,
              name=master_name,
              detach=True,
              labels={},
              restart_policy={"Name": restart_policy},
              device_requests=None,
              ports={"8080": f"{port}"},
              network=NETWORK_NAME,
              hostname="determined-master",
          )
  
          _wait_for_master("localhost", port, cluster_name)
  
          # Remove all cleanup methods from ExitStack.
          exit_stack.pop_all()
  
imports: import contextlib
  import os
  import pathlib
  import re
  import socket
  import subprocess
  import sys
  import tempfile
  import threading
  import time
  from typing import Any, Callable, Dict, Generator, List, Optional, Sequence, Type
  import appdirs
  import docker
  from determined.common import constants, util
  from determined.deploy.errors import MasterTimeoutExpired
  from determined.deploy.healthcheck import wait_for_master
label_desc: Use image node_id 7 for calling a global function with example usage: master_up(port, master_config_path, storage_host_path, master_name, image_repo_prefix, version, db_password, delete_db, autorestart, cluster_name, auto_work_dir) without return types