feature: Standalone watcher

changes/1560.feature.md

@@ -0,0 +1 @@
+Implement storage watcher and agent watcher with new standalone watcher interface and remove legacy watchers and their configs.

configs/agent/halfstack.toml

@@ -18,8 +18,6 @@ var-base-path = "./var/lib/backend.ai"
 # allow-compute-plugins = []
 # block-compute-plugins = []
 image-commit-path = "./tmp/backend.ai/commit/"
-mount-path = "./vfroot/local"
-cohabiting-storage-proxy = true
 
 
 [container]

configs/agent/sample.toml

@@ -52,13 +52,6 @@ agent-sock-port = 6007
 # This affects the per-sgroup configuration scope.
 scaling-group = "default"
 
-# Set the volume mount path for the agent node.
-# mount-path = "/vfroot"
-
-# A boolean flag to indicate if there is any cohabiting storage proxy within the same node.
-# If is there any cohabiting storage proxy, then events such as mount or umount will be ignored. [default: true]
-# cohabiting-storage-proxy = true
-
 # Create a PID file so that daemon managers could keep track of us.
 # If set to an empty string, it does NOT create the PID file.
 # pid-file = "./agent.pid"             # env: BACKEND_PID_FILE

configs/watcher/sample.toml

@@ -0,0 +1,63 @@
+[etcd]
+namespace = "local"                         # env: BACKEND_NAMESPACE
+addr = { host = "127.0.0.1", port = 8121 }  # env: BACKEND_ETCD_ADDR (host:port)
+user = ""                                   # env: BACKEND_ETCD_USER
+password = ""				    # env: BACKEND_ETCD_PASSWORD
+
+
+[logging]
+# One of: "NOTSET", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
+# Set the global logging level.
+level = "INFO"
+drivers = ["console"]
+
+[logging.pkg-ns]
+"" = "WARNING"
+"aiotools" = "INFO"
+"aiohttp" = "INFO"
+"ai.backend" = "INFO"
+
+[logging.console]
+# If set true, use ANSI colors if the console is a terminal.
+# If set false, always disable the colored output in console logs.
+colored = true
+
+# One of: "simple", "verbose"
+format = "simple"
+
+[debug]
+# Enable the debug mode by overriding the global loglevel and "ai.backend" loglevel.
+enabled = false
+
+[watcher]
+# Override the name of this watcher node.
+# If empty or unspecified, the config builds this from the hostname by prefixing it with "i-",
+# like "i-watcher-hostname".  The "i-watcher-" prefix is not mandatory, though.
+# Explicit configuration may be required if the hostname changes frequently,
+# to handle the event bus messages consistently.
+# This affects the per-node configuration scope.
+# node-id = "i-watcher-storage-proxy-01"
+
+# One of: "asyncio", "uvloop"
+# This changes the event loop backend.
+# uvloop is a fast libuv-based implementation but sometimes has
+# compatibility issues.
+# event-loop = "uvloop"
+
+# The base directory to put domain sockets for IPC.
+# Normally you don't have to change it.
+# NOTE: If Docker is installed via Snap (https://snapcraft.io/docker),
+#       you must change this to a directory under your *home* directory.
+# ipc-base-path = "/tmp/backend.ai/ipc"
+
+# Set the SSL certificate chain and the private keys used for serving the API requests.
+ssl-enabled = false
+#ssl-cert = "/etc/backend.ai/ssl/apiserver-fullchain.pem"   # env: BACKNED_SSL_CERT
+#ssl-privkey = "/etc/backend.ai/ssl/apiserver-privkey.pem"  # env: BACKNED_SSL_KEY
+
+
+# Set module's config.
+# <MODULE_NAME> should be the `name` property of watcher plugin.
+# It is recommended to define a watcher config that inherits from BaseWatcherConfig
+# for better config validation and strict typing.
+# [module.<MODULE_NAME>]

scripts/install-dev.sh

@@ -888,7 +888,6 @@ configure_backendai() {
     sed '/scratch-nfs-/d' ./agent.toml > ./agent.toml.sed
     mv ./agent.toml.sed ./agent.toml
   fi
-  sed_inplace "s@\(# \)\{0,1\}mount-path = .*@mount-path = "'"'"${ROOT_PATH}/${VFOLDER_REL_PATH}"'"'"@" ./agent.toml
   if [ $ENABLE_CUDA -eq 1 ]; then
     sed_inplace "s/# allow-compute-plugins =.*/allow-compute-plugins = [\"ai.backend.accelerator.cuda_open\"]/" ./agent.toml
   elif [ $ENABLE_CUDA_MOCK -eq 1 ]; then

src/ai/backend/agent/BUILD

@@ -14,19 +14,6 @@ python_sources(
         "src/ai/backend/helpers:src",
         "//:reqs#backend.ai-krunner-static-gnu",  # not auto-inferred
         ":resources",
-        "!./__init__.py:src-watcher",
-    ],
-)
-python_sources(
-    name="src-watcher",
-    sources=[
-        "__init__.py",
-        "watcher.py",
-    ],
-    dependencies=[
-        ":resources",
-        # Exclude transitive dependencies of the agent itself!
-        "!./__init__.py:src",
     ],
 )
 
@@ -80,14 +67,12 @@ pex_binary(
         "//src/ai/backend/accelerator/mock:lib",
         "//src/ai/backend/accelerator/mock:buildscript",
         "!!stubs/trafaret:stubs",
-        "!!./__init__.py:src-watcher",
     ],
 )
 pex_binary(
     name="pex-watcher",
     entry_point="watcher.py",
     dependencies=[
-        ":src-watcher",
         "!!stubs/trafaret:stubs",
         "!!./__init__.py:src",
     ],

src/ai/backend/agent/agent.py

@@ -67,7 +67,7 @@
 
 from ai.backend.common import msgpack, redis_helper
 from ai.backend.common.config import model_definition_iv
-from ai.backend.common.defs import REDIS_STAT_DB, REDIS_STREAM_DB
+from ai.backend.common.defs import MOUNT_MAP_KEY, REDIS_STAT_DB, REDIS_STREAM_DB
 from ai.backend.common.docker import MAX_KERNELSPEC, MIN_KERNELSPEC, ImageRef
 from ai.backend.common.events import (
     AbstractEvent,
@@ -2388,41 +2388,40 @@ async def handle_volume_mount(
     source: AgentId,
     event: DoVolumeMountEvent,
 ) -> None:
-    if context.local_config["agent"]["cohabiting-storage-proxy"]:
-        log.debug("Storage proxy is in the same node. Skip the volume task.")
+    # TODO: This should be removed after agent-watcher is fully implemented
+    err_msg: str | None = None
+
+    mountpoint = Path(event.dir_name)
+    if mountpoint.is_mount():
+        log.debug("Volume is already mounted. Skip the volume task.")
         await context.event_producer.produce_event(
             VolumeMounted(
+                "already-mounted",
                 str(context.id),
                 VolumeMountableNodeType.AGENT,
+                str(mountpoint),
                 "",
-                event.quota_scope_id,
             )
         )
         return
-    mount_prefix = await context.etcd.get("volumes/_mount")
-    volume_mount_prefix: str | None = context.local_config["agent"]["mount-path"]
-    if volume_mount_prefix is None:
-        volume_mount_prefix = "./"
-    real_path = Path(volume_mount_prefix, event.dir_name)
-    err_msg: str | None = None
     try:
         await mount(
-            str(real_path),
+            str(mountpoint),
             event.fs_location,
             event.fs_type,
             event.cmd_options,
             event.edit_fstab,
             event.fstab_path,
-            mount_prefix,
         )
+        await context.etcd.put(f"{MOUNT_MAP_KEY}/{str(mountpoint)}", event.fs_location)
     except VolumeMountFailed as e:
         err_msg = str(e)
     await context.event_producer.produce_event(
         VolumeMounted(
+            "mount-event",
             str(context.id),
             VolumeMountableNodeType.AGENT,
-            str(real_path),
-            event.quota_scope_id,
+            str(mountpoint),
             err_msg,
         )
     )
@@ -2433,40 +2432,41 @@ async def handle_volume_umount(
     source: AgentId,
     event: DoVolumeUnmountEvent,
 ) -> None:
-    if context.local_config["agent"]["cohabiting-storage-proxy"]:
-        log.debug("Storage proxy is in the same node. Skip the volume task.")
+    # TODO: This should be removed after agent-watcher is fully implemented
+    timeout = await context.etcd.get("config/watcher/file-io-timeout")
+    err_msg: str | None = None
+
+    mountpoint = Path(event.dir_name)
+    if not mountpoint.is_mount():
+        log.debug("Volume is already umounted. Skip the volume task.")
         await context.event_producer.produce_event(
             VolumeUnmounted(
+                "already-umounted",
                 str(context.id),
                 VolumeMountableNodeType.AGENT,
+                str(mountpoint),
                 "",
-                event.quota_scope_id,
             )
         )
         return
-    mount_prefix = await context.etcd.get("volumes/_mount")
-    timeout = await context.etcd.get("config/watcher/file-io-timeout")
-    volume_mount_prefix = context.local_config["agent"]["mount-path"]
-    real_path = Path(volume_mount_prefix, event.dir_name)
-    err_msg: str | None = None
     try:
         did_umount = await umount(
-            str(real_path),
-            mount_prefix,
+            str(mountpoint),
             event.edit_fstab,
             event.fstab_path,
             timeout_sec=float(timeout) if timeout is not None else None,
         )
+        await context.etcd.delete(f"{MOUNT_MAP_KEY}/{str(mountpoint)}")
     except VolumeMountFailed as e:
         err_msg = str(e)
     if not did_umount:
-        log.warning(f"{real_path} does not exist. Skip umount")
+        log.warning(f"{mountpoint} does not exist. Skip umount")
     await context.event_producer.produce_event(
         VolumeUnmounted(
+            "umount-event",
             str(context.id),
             VolumeMountableNodeType.AGENT,
-            str(real_path),
-            event.quota_scope_id,
+            str(mountpoint),
             err_msg,
         )
     )

src/ai/backend/agent/config.py

@@ -32,8 +32,6 @@
             t.Key("var-base-path", default="./var/lib/backend.ai"): tx.Path(
                 type="dir", auto_create=True
             ),
-            t.Key("mount-path", default=None): t.Null | tx.Path(type="dir", auto_create=True),
-            t.Key("cohabiting-storage-proxy", default=True): t.Bool(),
             t.Key("public-host", default=None): t.Null | t.String,
             t.Key("region", default=None): t.Null | t.String,
             t.Key("instance-type", default=None): t.Null | t.String,

src/ai/backend/agent/server.py

@@ -883,38 +883,33 @@ def main(
     # Determine where to read configuration.
     try:
         raw_cfg, cfg_src_path = config.read_from_file(config_path, "agent")
-    except config.ConfigurationError as e:
-        print(
-            "ConfigurationError: Could not read or validate the storage-proxy local config:",
-            file=sys.stderr,
-        )
-        print(pformat(e.invalid_data), file=sys.stderr)
-        raise click.Abort()
 
-    # Override the read config with environment variables (for legacy).
-    config.override_with_env(raw_cfg, ("etcd", "namespace"), "BACKEND_NAMESPACE")
-    config.override_with_env(raw_cfg, ("etcd", "addr"), "BACKEND_ETCD_ADDR")
-    config.override_with_env(raw_cfg, ("etcd", "user"), "BACKEND_ETCD_USER")
-    config.override_with_env(raw_cfg, ("etcd", "password"), "BACKEND_ETCD_PASSWORD")
-    config.override_with_env(
-        raw_cfg, ("agent", "rpc-listen-addr", "host"), "BACKEND_AGENT_HOST_OVERRIDE"
-    )
-    config.override_with_env(raw_cfg, ("agent", "rpc-listen-addr", "port"), "BACKEND_AGENT_PORT")
-    config.override_with_env(raw_cfg, ("agent", "pid-file"), "BACKEND_PID_FILE")
-    config.override_with_env(raw_cfg, ("container", "port-range"), "BACKEND_CONTAINER_PORT_RANGE")
-    config.override_with_env(raw_cfg, ("container", "bind-host"), "BACKEND_BIND_HOST_OVERRIDE")
-    config.override_with_env(raw_cfg, ("container", "sandbox-type"), "BACKEND_SANDBOX_TYPE")
-    config.override_with_env(raw_cfg, ("container", "scratch-root"), "BACKEND_SCRATCH_ROOT")
-
-    if debug:
-        log_level = LogSeverity.DEBUG
-    config.override_key(raw_cfg, ("debug", "enabled"), log_level == LogSeverity.DEBUG)
-    config.override_key(raw_cfg, ("logging", "level"), log_level)
-    config.override_key(raw_cfg, ("logging", "pkg-ns", "ai.backend"), log_level)
-
-    # Validate and fill configurations
-    # (allow_extra will make configs to be forward-copmatible)
-    try:
+        # Override the read config with environment variables (for legacy).
+        config.override_with_env(raw_cfg, ("etcd", "namespace"), "BACKEND_NAMESPACE")
+        config.override_with_env(raw_cfg, ("etcd", "addr"), "BACKEND_ETCD_ADDR")
+        config.override_with_env(raw_cfg, ("etcd", "user"), "BACKEND_ETCD_USER")
+        config.override_with_env(raw_cfg, ("etcd", "password"), "BACKEND_ETCD_PASSWORD")
+        config.override_with_env(
+            raw_cfg, ("agent", "rpc-listen-addr", "host"), "BACKEND_AGENT_HOST_OVERRIDE"
+        )
+        config.override_with_env(
+            raw_cfg, ("agent", "rpc-listen-addr", "port"), "BACKEND_AGENT_PORT"
+        )
+        config.override_with_env(raw_cfg, ("agent", "pid-file"), "BACKEND_PID_FILE")
+        config.override_with_env(
+            raw_cfg, ("container", "port-range"), "BACKEND_CONTAINER_PORT_RANGE"
+        )
+        config.override_with_env(raw_cfg, ("container", "bind-host"), "BACKEND_BIND_HOST_OVERRIDE")
+        config.override_with_env(raw_cfg, ("container", "sandbox-type"), "BACKEND_SANDBOX_TYPE")
+        config.override_with_env(raw_cfg, ("container", "scratch-root"), "BACKEND_SCRATCH_ROOT")
+        if debug:
+            log_level = LogSeverity.DEBUG
+        config.override_key(raw_cfg, ("debug", "enabled"), log_level == LogSeverity.DEBUG)
+        config.override_key(raw_cfg, ("logging", "level"), log_level)
+        config.override_key(raw_cfg, ("logging", "pkg-ns", "ai.backend"), log_level)
+
+        # Validate and fill configurations
+        # (allow_extra will make configs to be forward-copmatible)
         cfg = config.check(raw_cfg, agent_local_config_iv)
         if cfg["agent"]["backend"] == AgentBackend.KUBERNETES:
             if cfg["container"]["scratch-type"] == "k8s-nfs" and (
@@ -931,7 +926,10 @@ def main(
             pprint(cfg)
         cfg["_src"] = cfg_src_path
     except config.ConfigurationError as e:
-        print("ConfigurationError: Validation of agent local config has failed:", file=sys.stderr)
+        print(
+            "ConfigurationError: Could not read or validate the agent local config:",
+            file=sys.stderr,
+        )
         print(pformat(e.invalid_data), file=sys.stderr)
         raise click.Abort()
 
@@ -1000,7 +998,7 @@ def main(
                 log.info("runtime: {0}", utils.env_info())
 
                 log_config = logging.getLogger("ai.backend.agent.config")
-                if log_level == "DEBUG":
+                if log_level == LogSeverity.DEBUG:
                     log_config.debug("debug mode enabled.")
 
                 if cfg["agent"]["event-loop"] == "uvloop":