diff --git a/dvc/analytics.py b/dvc/analytics.py index 6e3dc91b9..2ad66cb35 100644 --- a/dvc/analytics.py +++ b/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) diff --git a/dvc/commands/daemon.py b/dvc/commands/daemon.py index 35d6e90e6..cb9fc76ca 100644 --- a/dvc/commands/daemon.py +++ b/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 diff --git a/dvc/daemon.py b/dvc/daemon.py index a97f0b806..73d29a1a3 100644 --- a/dvc/daemon.py +++ b/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 "") diff --git a/dvc/env.py b/dvc/env.py index fc2ad3f15..919b2aefd 100644 --- a/dvc/env.py +++ b/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" diff --git a/dvc/updater.py b/dvc/updater.py index 84e7beae2..906372e21 100644 --- a/dvc/updater.py +++ b/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: diff --git a/tests/func/test_daemon.py b/tests/func/test_daemon.py new file mode 100644 index 000000000..97120a53a --- /dev/null +++ b/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 diff --git a/tests/unit/test_daemon.py b/tests/unit/test_daemon.py index cff3c00c5..20595eec0 100644 --- a/tests/unit/test_daemon.py +++ b/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])