redirect fds rather than closing (#10028)

log exit code and pid on verbose logging

Is useful for debugging on daemon.

dvc/cli/__init__.py

@@ -173,10 +173,10 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
     args = None
 
     outer_log_level = logger.level
+    level = None
     try:
         args = parse_args(argv)
 
-        level = None
         if args.quiet:
             level = logging.CRITICAL
         elif args.verbose == 1:
@@ -235,11 +235,17 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         ret = _log_exceptions(exc) or 255
 
     try:
+        import os
+
         from dvc import analytics
 
         if analytics.is_enabled():
             analytics.collect_and_send_report(args, ret)
 
+        logger.trace(  # type: ignore[attr-defined]
+            "Process %s exiting with %s", os.getpid(), ret
+        )
+
         return ret
     finally:
         logger.setLevel(outer_log_level)

dvc/config_schema.py

@@ -256,6 +256,9 @@ SCHEMA = {
                     "exclude_shared_token_cache_credential": Bool,
                     "exclude_managed_identity_credential": Bool,
                     Optional("verify", default=False): Bool,
+                    "timeout": int,
+                    "read_timeout": int,
+                    "connection_timeout": int,
                     **REMOTE_COMMON,
                 },
                 "oss": {

dvc/daemon.py

@@ -3,6 +3,7 @@
 import inspect
 import logging
 import os
+import platform
 import sys
 from subprocess import Popen  # nosec B404
 from typing import List
@@ -54,6 +55,14 @@ def _spawn_windows(cmd, env):
         _suppress_resource_warning(popen)
 
 
+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 _spawn_posix(cmd, env):
     from dvc.cli import main
 
@@ -87,13 +96,14 @@ def _spawn_posix(cmd, env):
         logger.exception("failed at second fork")
         os._exit(1)  # pylint: disable=protected-access
 
-    sys.stdin.close()
-    sys.stdout.close()
-    sys.stderr.close()
-    os.closerange(0, 3)
-
-    os.environ.update(env)
-    main(cmd)
+    _redirect_streams_to_null()
+    if platform.system() == "Darwin":
+        # workaround for MacOS bug
+        # https://github.com/iterative/dvc/issues/4294
+        _popen(cmd, env=env).communicate()
+    else:
+        os.environ.update(env)
+        main(cmd)
 
     os._exit(0)  # pylint: disable=protected-access
 

dvc/render/converter/vega.py

@@ -114,10 +114,6 @@ class VegaConverter(Converter):
         self.plot_id = plot_id
         self.inferred_properties: Dict = {}
 
-        # TODO we should be handling that in `convert`,
-        #      to avoid stateful `self.inferred_properties`
-        self._infer_x_y()
-
     def _infer_y_from_data(self):
         if self.plot_id in self.data:
             for lst in _lists(self.data[self.plot_id]):
@@ -296,16 +292,11 @@ class VegaConverter(Converter):
     ):
         """
         Convert the data. Fill necessary fields ('x', 'y') and return both
-        generated datapoints and updated properties. If `x` is not provided,
-        leave it as None, fronteds should handle it.
-
-        NOTE: Studio uses this method.
-              The only thing studio FE handles is filling `x` and `y`.
-              `x/y_label` should be filled here.
-
-              Datapoints are not stripped according to config, because users
-              might be utilizing other fields in their custom plots.
+        generated datapoints and updated properties. `x`, `y` values and labels
+        are inferred and always provided.
         """
+        self._infer_x_y()
+
         datapoints = self._find_datapoints()
         properties = {**self.properties, **self.inferred_properties}
 

dvc/repo/worktree.py

@@ -210,7 +210,9 @@ def _merge_push_meta(
         assert isinstance(old_tree, Tree)
         entry.hash_info = old_tree.hash_info
         entry.meta = out.meta
+        entries = [entry]
         for subkey, entry in index.iteritems(key):
+            entries.append(entry)
             if entry.meta is not None and entry.meta.isdir:
                 continue
             fs_path = repo.fs.path.join(repo.root_dir, *subkey)
@@ -225,6 +227,10 @@ def _merge_push_meta(
                 # this dir (entry will have the latest remote version
                 # ID after checkout)
                 entry.meta = meta
+
+        for entry in entries:
+            index.add(entry)
+
         tree_meta, new_tree = build_tree(index, key)
         out.obj = new_tree
         out.hash_info = new_tree.hash_info

dvc/testing/fixtures.py

@@ -1,9 +1,12 @@
 import os
+import pathlib
 import subprocess  # nosec B404
 from typing import Dict, Tuple
 
 import pytest
 
+from .scripts import copy_script
+
 # pylint: disable=redefined-outer-name,unused-argument
 
 __all__ = [
@@ -28,6 +31,8 @@ __all__ = [
     "local_workspace",
     "docker_compose_project_name",
     "docker_services",
+    "copy_script",
+    "run_copy",
 ]
 
 CACHE: Dict[Tuple[bool, bool, bool], str] = {}
@@ -261,3 +266,20 @@ def docker_services(
         executor.execute(docker_setup)
         # note: we are not tearing down the containers here
         return Services(executor)
+
+
+@pytest.fixture
+def run_copy(tmp_dir, copy_script, dvc):  # noqa: ARG001
+    def run_copy(src, dst, **run_kwargs):
+        wdir = pathlib.Path(run_kwargs.get("wdir", "."))
+        wdir = pathlib.Path("../" * len(wdir.parts))
+        script_path = wdir / "copy.py"
+
+        return dvc.run(
+            cmd=f"python {script_path} {src} {dst}",
+            outs=[dst],
+            deps=[src, f"{script_path}"],
+            **run_kwargs,
+        )
+
+    return run_copy

dvc/testing/remote_tests.py

@@ -169,22 +169,40 @@ class TestRemote:
 
 
 class TestRemoteVersionAware:
-    def test_file(self, tmp_dir, dvc, remote_version_aware):  # pylint: disable=W0613
+    def test_file(
+        self, tmp_dir, dvc, run_copy, remote_version_aware
+    ):  # pylint: disable=W0613
         (stage,) = tmp_dir.dvc_gen("foo", "foo")
+        run_copy("foo", "foo_copy", name="copy")
 
         dvc.push()
-        assert "version_id" in (tmp_dir / "foo.dvc").read_text()
+        foo_dvc = (tmp_dir / "foo.dvc").read_text()
+        assert "version_id" in foo_dvc
         stage = stage.reload()
         out = stage.outs[0]
         assert out.meta.version_id
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
         remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
 
         dvc.pull()
         assert (tmp_dir / "foo").read_text() == "foo"
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-    def test_dir(self, tmp_dir, dvc, remote_version_aware):  # pylint: disable=W0613
+        dvc.push()
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+        dvc.reproduce()
+        dvc.push()
+        assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+    def test_dir(
+        self, tmp_dir, dvc, run_copy, remote_version_aware
+    ):  # pylint: disable=W0613
         (stage,) = tmp_dir.dvc_gen(
             {
                 "data_dir": {
@@ -196,8 +214,10 @@ class TestRemoteVersionAware:
         )
 
         dvc.push()
-        assert "files" in (tmp_dir / "data_dir.dvc").read_text()
-        assert "version_id" in (tmp_dir / "data_dir.dvc").read_text()
+
+        data_dir_dvc = (tmp_dir / "data_dir.dvc").read_text()
+        assert "files" in data_dir_dvc
+        assert "version_id" in data_dir_dvc
         stage = stage.reload()
         out = stage.outs[0]
         assert out.files
@@ -213,6 +233,19 @@ class TestRemoteVersionAware:
         assert (
             tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
         ).read_text() == "data_sub"
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+
+        run_copy("data_dir", "data_dir_copy", name="copy")
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
+
+        dvc.push()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() != dvc_lock
+        dvc_lock = (tmp_dir / "dvc.lock").read_text()
+
+        dvc.push()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
 
 class TestRemoteWorktree:

dvc/testing/scripts.py

@@ -0,0 +1,28 @@
+import pytest
+
+COPY_SCRIPT = """
+import os
+import shutil
+import sys
+
+if os.path.isfile(sys.argv[1]):
+    shutil.copyfile(sys.argv[1], sys.argv[2])
+else:
+    shutil.copytree(sys.argv[1], sys.argv[2])
+""".replace(
+    "\r\n", "\n"
+)
+
+
+def _add_script(tmp_dir, path, contents=""):
+    script, *_ = tmp_dir.gen(path, contents.encode("utf-8"))
+    if hasattr(tmp_dir, "scm"):
+        tmp_dir.scm_add(path, commit=f"add {path}")
+    return script.fs_path
+
+
+@pytest.fixture
+def copy_script(
+    tmp_dir,
+):
+    return _add_script(tmp_dir, "copy.py", COPY_SCRIPT)

pyproject.toml

@@ -70,7 +70,7 @@ dependencies = [
 
 [project.optional-dependencies]
 all = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs]"]
-azure = ["dvc-azure>=2.21.2"]
+azure = ["dvc-azure>=2.23.0"]
 dev = ["dvc[azure,gdrive,gs,hdfs,lint,oss,s3,ssh,tests,webdav,webhdfs]"]
 gdrive = ["dvc-gdrive==2.20"]
 gs = ["dvc-gs==2.22.1"]

tests/dir_helpers.py

@@ -45,12 +45,10 @@ from global repo template to creating everything inplace, which:
 
 
 import os
-import pathlib
 
 import pytest
 
 __all__ = [
-    "run_copy",
     "run_head",
     "erepo_dir",
     "git_dir",
@@ -59,23 +57,6 @@ __all__ = [
 ]
 
 
-@pytest.fixture
-def run_copy(tmp_dir, copy_script, dvc):
-    def run_copy(src, dst, **run_kwargs):
-        wdir = pathlib.Path(run_kwargs.get("wdir", "."))
-        wdir = pathlib.Path("../" * len(wdir.parts))
-        script_path = wdir / "copy.py"
-
-        return dvc.run(
-            cmd=f"python {script_path} {src} {dst}",
-            outs=[dst],
-            deps=[src, f"{script_path}"],
-            **run_kwargs,
-        )
-
-    return run_copy
-
-
 @pytest.fixture
 def run_head(tmp_dir, head_script, dvc):
     script = os.path.abspath(tmp_dir / "head.py")

tests/func/experiments/test_experiments.py

@@ -25,8 +25,8 @@ from dvc.repo.experiments.refs import CELERY_STASH
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.scm import SCMError, resolve_rev
 from dvc.stage.exceptions import StageFileDoesNotExistError
+from dvc.testing.scripts import COPY_SCRIPT
 from dvc.utils.serialize import PythonFileCorruptedError
-from tests.scripts import COPY_SCRIPT
 
 
 @pytest.mark.parametrize("name", [None, "foo"])

tests/scripts.py

@@ -1,24 +1,6 @@
 import pytest
 
-COPY_SCRIPT = """
-import os
-import shutil
-import sys
-
-if os.path.isfile(sys.argv[1]):
-    shutil.copyfile(sys.argv[1], sys.argv[2])
-else:
-    shutil.copytree(sys.argv[1], sys.argv[2])
-""".replace(
-    "\r\n", "\n"
-)
-
-
-def _add_script(tmp_dir, path, contents=""):
-    script, *_ = tmp_dir.gen(path, contents.encode("utf-8"))
-    if hasattr(tmp_dir, "scm"):
-        tmp_dir.scm_add(path, commit=f"add {path}")
-    return script.fs_path
+from dvc.testing.scripts import _add_script
 
 
 @pytest.fixture
@@ -38,13 +20,6 @@ with Path(sys.argv[1]).open("a+", encoding="utf-8") as f:
     )
 
 
-@pytest.fixture
-def copy_script(
-    tmp_dir,
-):
-    return _add_script(tmp_dir, "copy.py", COPY_SCRIPT)
-
-
 @pytest.fixture
 def head_script(tmp_dir):
     """Output first line of each file to different file with '-1' appended.

tests/unit/render/test_vega_converter.py

@@ -24,6 +24,7 @@ def test_finding_lists(dictionary, expected_result):
     assert list(result) == expected_result
 
 
+@pytest.mark.studio
 @pytest.mark.parametrize(
     "input_data,properties,expected_datapoints,expected_properties",
     [