daemon: add logging support, tests and generalize popen based runs for future (#10033)

* use popen to run updater/analytics in detached mode

subprocess does not provide a way to avoid a zombie process. Not to confuse with orphaned processes, which will be `wait`-ed by `pid 1`.

It might not work for long-running processes, but I don't think zombie processes are going to be a problem for us, as the parent process immediately exits after spawning a child process.

Even in case of `queue` based experiment runner, parent will exit before than the child process. So I don't think it's a requirement here. But it does feel wrong to leak the processes, so I have to sleep on that.

---

This PR:

1. makes `daemonized` process inherit the logging verbosity, so if the parent process used `-vv`, the daemonized process will run the same way.
2. Add `DVC_DAEMON_LOGFILE` envvar, that the daemon will log to the file. By default, the verbosity is the same as running parent processes. Without `DVC_DAEMON_LOGFILE`, the output will be redirected to `/dev/null`.
3. Adds debug/trace for updater/analytics.

I hope by doing this, we can make our life easier on debugging and make daemon more transparent/observable.

* add tests

* fixes

* use kwargs

* pass all os.environ with some modified

dvc/analytics.py

@@ -2,7 +2,7 @@ import json
 import logging
 import os
 
-from .env import DVC_NO_ANALYTICS
+from .env import DVC_ANALYTICS_ENDPOINT, DVC_NO_ANALYTICS
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +34,10 @@ def collect_and_send_report(args=None, return_code=None):
 
     with tempfile.NamedTemporaryFile(delete=False, mode="w") as fobj:
         json.dump(report, fobj)
+
+    logger.trace(  # type: ignore[attr-defined]
+        "Saving analytics report to %s", fobj.name
+    )
     daemon(["analytics", fobj.name])
 
 
@@ -65,7 +69,7 @@ def send(path):
     """
     import requests
 
-    url = "https://analytics.dvc.org"
+    url = os.environ.get(DVC_ANALYTICS_ENDPOINT, "https://analytics.dvc.org")
     headers = {"content-type": "application/json"}
 
     with open(path, encoding="utf-8") as fobj:
@@ -73,11 +77,16 @@ def send(path):
 
     report.update(_runtime_info())
 
+    logger.debug("uploading report to %s", url)
+    logger.trace("Sending %s to %s", report, url)  # type: ignore[attr-defined]
+
     try:
         requests.post(url, json=report, headers=headers, timeout=5)
-    except requests.exceptions.RequestException:
-        logger.debug("failed to send analytics report", exc_info=True)
+    except requests.exceptions.RequestException as e:
+        logger.trace("", exc_info=True)  # type: ignore[attr-defined]
+        logger.debug("failed to send analytics report %s", str(e))
 
+    logger.trace("removing report %s", path)  # type: ignore[attr-defined]
     os.remove(path)
 
 

dvc/commands/daemon.py

@@ -1,7 +1,11 @@
+import logging
+
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import fix_subparsers
 
+logger = logging.getLogger(__name__)
+
 
 class CmdDaemonBase(CmdBaseNoRepo):
     pass
@@ -21,6 +25,8 @@ class CmdDaemonUpdater(CmdDaemonBase):
         config = Config(dvc_dir, validate=False)
         hardlink_lock = config.get("core", {}).get("hardlink_lock", False)
         updater = Updater(tmp_dir, hardlink_lock=hardlink_lock)
+
+        logger.info("Starting updater to fetch the latest version")
         updater.fetch(detach=False)
 
         return 0
@@ -30,6 +36,7 @@ class CmdDaemonAnalytics(CmdDaemonBase):
     def run(self):
         from dvc import analytics
 
+        logger.info("Sending analytics")
         analytics.send(self.args.target)
 
         return 0

dvc/daemon.py

@@ -4,36 +4,50 @@ import inspect
 import logging
 import os
 import platform
+import subprocess  # nosec B404
 import sys
-from subprocess import Popen  # nosec B404
-from typing import List
+from contextlib import nullcontext
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Iterable,
+    List,
+    Mapping,
+    Optional,
+    Sequence,
+    Union,
+)
 
-from dvc.env import DVC_DAEMON
+from dvc.env import DVC_DAEMON, DVC_DAEMON_LOGFILE
 from dvc.utils import fix_env, is_binary
+from dvc.utils.collections import ensure_list
+
+if TYPE_CHECKING:
+    from typing import IO, ContextManager
 
 logger = logging.getLogger(__name__)
 
+_FILE = Union[None, int, "IO[Any]"]
 
-def _suppress_resource_warning(popen: Popen):
+
+def _suppress_resource_warning(popen: subprocess.Popen) -> None:
     """Sets the returncode to avoid ResourceWarning when popen is garbage collected."""
     # only use for daemon processes.
     # See https://bugs.python.org/issue38890.
     popen.returncode = 0
 
 
-def _popen(cmd, **kwargs) -> Popen:
-    prefix = [sys.executable]
-    if not is_binary():
-        main_entrypoint = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "__main__.py"
-        )
-        prefix += [main_entrypoint]
-    return Popen(
-        prefix + cmd, close_fds=True, shell=False, **kwargs  # nosec B603  # noqa: S603
-    )
-
-
-def _spawn_windows(cmd, env):
+def run_detached(
+    args: Sequence[str],
+    env: Optional[Mapping[str, str]] = None,
+    stdin: _FILE = subprocess.DEVNULL,
+    stdout: _FILE = subprocess.DEVNULL,
+    stderr: _FILE = subprocess.DEVNULL,
+    **kwargs: Any,
+) -> int:
+    # NOTE: this may create zombie processes on unix
+    startupinfo = None
+    creationflags = 0
     if sys.platform == "win32":
         from subprocess import (  # nosec B404
             CREATE_NEW_PROCESS_GROUP,
@@ -49,21 +63,38 @@ def _spawn_windows(cmd, env):
         startupinfo = STARTUPINFO()
         startupinfo.dwFlags |= STARTF_USESHOWWINDOW
 
-        popen = _popen(
-            cmd, env=env, creationflags=creationflags, startupinfo=startupinfo
-        )
-        _suppress_resource_warning(popen)
+    popen = subprocess.Popen(
+        args,
+        stdin=stdin,
+        stdout=stdout,
+        stderr=stderr,
+        close_fds=True,
+        shell=False,  # noqa: S603 # nosec B603
+        env=env,
+        startupinfo=startupinfo,
+        creationflags=creationflags,
+        start_new_session=True,
+        **kwargs,
+    )
+    _suppress_resource_warning(popen)
+    return popen.pid
 
 
-def _redirect_streams_to_null():
-    # See discussion in https://github.com/iterative/dvc/pull/10026
-    fd = os.open(os.devnull, os.O_RDWR)
-    for fd2 in range(3):
-        os.dup2(fd, fd2)
-    os.close(fd)
+def _get_dvc_args() -> List[str]:
+    args = [sys.executable]
+    if not is_binary():
+        root_dir = os.path.abspath(os.path.dirname(__file__))
+        main_entrypoint = os.path.join(root_dir, "__main__.py")
+        args.append(main_entrypoint)
+    return args
 
 
-def _spawn_posix(cmd, env):
+def _spawn_posix(
+    executable: List[str],
+    args: List[str],
+    env: Optional[Mapping[str, str]] = None,
+    output_file: Optional[str] = None,
+) -> None:
     from dvc.cli import main
 
     # `fork` will copy buffers, so we need to flush them before forking.
@@ -96,49 +127,91 @@ def _spawn_posix(cmd, env):
         logger.exception("failed at second fork")
         os._exit(1)  # pylint: disable=protected-access
 
-    _redirect_streams_to_null()
+    # disconnect from the terminal
+    fd = os.open(os.devnull, os.O_RDWR)
+    os.dup2(fd, sys.stdin.fileno())
+    os.close(fd)
+
+    with open(output_file or os.devnull, "ab") as f:
+        os.dup2(f.fileno(), sys.stdout.fileno())
+        os.dup2(f.fileno(), sys.stderr.fileno())
+
     if platform.system() == "Darwin":
         # workaround for MacOS bug
         # https://github.com/iterative/dvc/issues/4294
-        _popen(cmd, env=env).communicate()
+        subprocess.Popen(
+            executable + args, env=env, shell=False  # noqa: S603 # nosec B603
+        ).communicate()
     else:
-        os.environ.update(env)
-        main(cmd)
+        os.environ.update(env or {})
+        main(args)
 
     os._exit(0)  # pylint: disable=protected-access
 
 
-def _spawn(cmd, env):
-    logger.debug("Trying to spawn '%s'", cmd)
-
-    if os.name == "nt":
-        _spawn_windows(cmd, env)
-    elif os.name == "posix":
-        _spawn_posix(cmd, env)
-    else:
-        raise NotImplementedError
-
-    logger.debug("Spawned '%s'", cmd)
-
-
-def daemon(args):
+def daemon(args: Iterable[str]) -> None:
     """Launch a `dvc daemon` command in a detached process.
 
     Args:
         args (list): list of arguments to append to `dvc daemon` command.
     """
-    daemonize(["daemon", "-q", *args])
+    flags = {
+        logging.CRITICAL: "-q",
+        logging.DEBUG: "-v",
+        logging.TRACE: "-vv",  # type: ignore[attr-defined]
+    }
+    args = list(args)
+    if flag := flags.get(logger.getEffectiveLevel()):
+        args.append(flag)
+    daemonize(["daemon", *args])
 
 
-def daemonize(cmd: List[str]):
+def _spawn_subprocess(
+    executable: List[str],
+    args: List[str],
+    env: Optional[Mapping[str, str]] = None,
+    output_file: Optional[str] = None,
+) -> Optional[int]:
+    # adapt run_detached to _spawn's interface
+    file: "ContextManager[Any]" = nullcontext()
+    kw = {}
+    if output_file:
+        file = open(output_file, "ab")  # noqa: SIM115
+        kw = {"stdout": file, "stderr": file}
+
+    with file:
+        return run_detached(executable + args, env, **kw)
+
+
+def _spawn(
+    executable: List[str],
+    args: List[str],
+    env: Optional[Mapping[str, str]] = None,
+    output_file: Optional[str] = None,
+) -> Optional[int]:
+    if os.name == "nt":
+        return _spawn_subprocess(executable, args, env, output_file=output_file)
+
+    if os.name == "posix":
+        _spawn_posix(executable, args, env, output_file=output_file)
+        return None
+
+    raise NotImplementedError
+
+
+def daemonize(args: List[str], executable: Union[None, str, List[str]] = None) -> None:
     if os.environ.get(DVC_DAEMON):
         logger.debug("skipping launching a new daemon.")
         return
 
+    executable = _get_dvc_args() if executable is None else ensure_list(executable)
+
     env = fix_env()
+    env[DVC_DAEMON] = "1"
     if not is_binary():
         file_path = os.path.abspath(inspect.stack()[0][1])
         env["PYTHONPATH"] = os.path.dirname(os.path.dirname(file_path))
-    env[DVC_DAEMON] = "1"
 
-    _spawn(cmd, env)
+    logger.debug("Trying to spawn %r", args)
+    pid = _spawn(executable, args, env, output_file=env.get(DVC_DAEMON_LOGFILE))
+    logger.debug("Spawned %r%s", args, f" with pid {pid}" if pid else "")

dvc/env.py

@@ -1,4 +1,6 @@
+DVC_ANALYTICS_ENDPOINT = "DVC_ANALYTICS_ENDPOINT"
 DVC_DAEMON = "DVC_DAEMON"
+DVC_DAEMON_LOGFILE = "DVC_DAEMON_LOGFILE"
 DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
 DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
 DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
@@ -15,3 +17,4 @@ DVC_STUDIO_REPO_URL = "DVC_STUDIO_REPO_URL"
 DVC_STUDIO_TOKEN = "DVC_STUDIO_TOKEN"  # noqa: S105 # nosec B105
 DVC_STUDIO_URL = "DVC_STUDIO_URL"
 DVC_SYSTEM_CONFIG_DIR = "DVC_SYSTEM_CONFIG_DIR"
+DVC_UPDATER_ENDPOINT = "DVC_UPDATER_ENDPOINT"

dvc/updater.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Optional
 from packaging import version
 
 from dvc import PKG, __version__
+from dvc.env import DVC_UPDATER_ENDPOINT
 
 if TYPE_CHECKING:
     from dvc.ui import RichText
@@ -46,6 +47,7 @@ class Updater:
             with self.lock:
                 func()
         except LockError:
+            logger.trace("", exc_info=True)  # type: ignore[attr-defined]
             logger.debug(
                 "Failed to acquire '%s' before %s updates",
                 self.lock.lockfile,
@@ -77,6 +79,7 @@ class Updater:
                 info = json.load(fobj)
                 latest = info["version"]
             except Exception as e:  # noqa: BLE001  # pylint: disable=W0703
+                logger.trace("", exc_info=True)  # type: ignore[attr-defined]
                 logger.debug("'%s' is not a valid json: %s", self.updater_file, e)
                 self.fetch()
                 return
@@ -98,14 +101,23 @@ class Updater:
 
         import requests
 
+        url = os.environ.get(DVC_UPDATER_ENDPOINT, self.URL)
+        logger.debug("Checking updates in %s", url)
         try:
-            resp = requests.get(self.URL, timeout=self.TIMEOUT_GET)
+            resp = requests.get(url, timeout=self.TIMEOUT_GET)
             info = resp.json()
         except requests.exceptions.RequestException as exc:
+            logger.trace("", exc_info=True)  # type: ignore[attr-defined]
             logger.debug("Failed to retrieve latest version: %s", exc)
             return
 
+        logger.trace(  # type: ignore[attr-defined]
+            "received payload: %s (status=%s)", info, resp.status_code
+        )
         with open(self.updater_file, "w+", encoding="utf-8") as fobj:
+            logger.trace(  # type: ignore[attr-defined]
+                "Saving latest version info to %s", self.updater_file
+            )
             json.dump(info, fobj)
 
     def _notify(self, latest: str, pkg: Optional[str] = PKG) -> None:

tests/func/test_daemon.py

@@ -0,0 +1,142 @@
+import json
+import os
+import re
+import subprocess
+import time
+from collections import defaultdict
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+from pathlib import Path
+from threading import Thread
+from typing import Dict
+
+import pytest
+
+from dvc import version_tuple
+from dvc.daemon import _get_dvc_args
+from dvc.env import (
+    DVC_ANALYTICS_ENDPOINT,
+    DVC_DAEMON_LOGFILE,
+    DVC_NO_ANALYTICS,
+    DVC_UPDATER_ENDPOINT,
+)
+from dvc.updater import Updater
+
+version = ".".join(map(str, version_tuple[:3]))
+UPDATER_INFO = {
+    "version": version,
+    "packages": {
+        "linux": {
+            "deb": f"https://dvc.org/download/linux-deb/dvc-{version}",
+            "rpm": f"https://dvc.org/download/linux-rpm/dvc-{version}",
+        },
+        "windows": {"exe": f"https://dvc.org/download/win/dvc-{version}"},
+        "osx": {"pkg": f"https://dvc.org/download/osx/dvc-{version}"},
+    },
+}
+
+UPDATER_INFO_STR = json.dumps(UPDATER_INFO).encode("utf8")
+
+
+def make_request_handler():
+    class RequestHandler(SimpleHTTPRequestHandler):
+        # save requests count for each method
+        hits: Dict[str, int] = defaultdict(lambda: 0)
+
+        def do_POST(self):  # noqa: N802
+            # analytics endpoint
+            self.hits["POST"] += 1
+            self.send_response(200)
+            super().end_headers()
+
+        def do_GET(self):  # noqa: N802
+            # updater endpoint
+            self.hits["GET"] += 1
+            self.send_response(200)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+            self.wfile.write(UPDATER_INFO_STR)
+
+    return RequestHandler
+
+
+@pytest.fixture
+def server():
+    with HTTPServer(("localhost", 0), make_request_handler()) as httpd:
+        thread = Thread(target=httpd.serve_forever)
+        thread.daemon = True
+        thread.start()
+        yield httpd
+
+
+def retry_until(pred, timeout):
+    timeout_ns = timeout * 1e9
+    start = time.perf_counter_ns()
+    while time.perf_counter_ns() - start < timeout_ns:
+        if pred():
+            return
+        time.sleep(0.01)
+    raise RuntimeError(f"timed out after {timeout}s")
+
+
+def test_analytics(tmp_path, server):
+    addr = server.server_address
+    logfile = tmp_path / "logfile"
+
+    env = {
+        **os.environ,
+        DVC_DAEMON_LOGFILE: str(logfile),
+        DVC_ANALYTICS_ENDPOINT: "http://{}:{}".format(*addr),
+    }
+    env.pop("DVC_TEST", None)
+    env.pop("DVC_NO_ANALYTICS", None)
+
+    output = subprocess.check_output(
+        [*_get_dvc_args(), "config", "-l", "-vv"],
+        env=env,
+        text=True,
+    )
+
+    pattern = r".*Saving analytics report to (.*)"
+    for line in output.splitlines():
+        if match := re.search(pattern, line):
+            report_file = match.group(1).strip()
+            break
+    else:
+        raise AssertionError("no match for the report file")
+
+    # wait until the file disappears
+    retry_until(lambda: not os.path.exists(report_file), 10)
+    # wait till the daemon exits
+    retry_until(lambda: "exiting with 0" in logfile.read_text(encoding="utf8"), 5)
+    assert server.RequestHandlerClass.hits == {"POST": 1}
+
+
+def test_updater(tmp_dir, dvc, server):
+    addr = server.server_address
+    logfile = tmp_dir / "logfile"
+
+    env = {
+        **os.environ,
+        DVC_DAEMON_LOGFILE: str(logfile),
+        DVC_UPDATER_ENDPOINT: "http://{}:{}".format(*addr),
+        # prevent running analytics daemon
+        DVC_NO_ANALYTICS: "true",
+    }
+    env.pop("DVC_TEST", None)
+    env.pop("CI", None)
+
+    subprocess.check_output(
+        [*_get_dvc_args(), "version", "-vv"],
+        env=env,
+        text=True,
+    )
+
+    updater_file = Path(dvc.tmp_dir) / Updater.UPDATER_FILE
+
+    # wait until the updater file appears
+    retry_until(updater_file.is_file, 10)
+    # wait till the daemon exits
+    retry_until(lambda: "exiting with 0" in logfile.read_text(encoding="utf8"), 5)
+    assert server.RequestHandlerClass.hits == {"GET": 1}
+    # check that the file is saved correctly
+    assert json.loads(updater_file.read_text(encoding="utf8")) == UPDATER_INFO

tests/unit/test_daemon.py

@@ -5,7 +5,7 @@ from dvc import daemon
 
 
 def test_daemon(mocker):
-    mock_windows = mocker.patch("dvc.daemon._spawn_windows")
+    mock_windows = mocker.patch("dvc.daemon._spawn_subprocess")
     mock_posix = mocker.patch("dvc.daemon._spawn_posix")
     daemon.daemon(["updater"])
 
@@ -18,7 +18,7 @@ def test_daemon(mocker):
         mock_posix.assert_called()
         args = mock_posix.call_args[0]
 
-    env = args[1]
+    env = args[2]
     assert "PYTHONPATH" in env
 
     file_path = os.path.abspath(inspect.stack()[0][1])