diff --git a/bin/mounter/mounter.py b/bin/mounter/mounter.py index 83c26d7608c..7aa39481a80 100644 --- a/bin/mounter/mounter.py +++ b/bin/mounter/mounter.py @@ -154,9 +154,15 @@ def is_mounted(path): def _responds(path): - """True if `path` can be stat()ed — false for a mount whose FUSE server is gone.""" + """True if `path` can be listed — false for a mount whose FUSE server is gone. + + stat() alone is not enough: the kernel can answer it from attributes cached while the + server was alive, so a dead mount may stat fine and fail with ENOTCONN only once it is + opened. Reporting that mount as live would hand it back as "already mounted" for ever. + """ try: - os.stat(path) + with os.scandir(path) as entries: + next(entries, None) return True except OSError: return False diff --git a/bin/mounter/tests/test_mounter.py b/bin/mounter/tests/test_mounter.py index c7b84472929..3122acfd55d 100644 --- a/bin/mounter/tests/test_mounter.py +++ b/bin/mounter/tests/test_mounter.py @@ -95,6 +95,32 @@ def test_a_dead_mount_is_still_reported_as_mounted(mounter, cu_dir): assert mounter.is_mounted(target) # what it should have asked +# ─────────────────── _responds() ─────────────────── + +def test_a_listable_directory_responds(mounter, tmp_path): + assert mounter._responds(str(tmp_path)) + + +def test_a_missing_path_does_not_respond(mounter, tmp_path): + assert not mounter._responds(str(tmp_path / "gone")) + + +def test_a_mount_that_stats_but_cannot_be_listed_does_not_respond(mounter, tmp_path, monkeypatch): + """A dead FUSE mount can still stat() from attributes the kernel cached, and fail only + when opened. Such a mount must not count as live, or it is never remounted.""" + real_scandir = os.scandir + + def dead_mount(path): + if os.fspath(path) == str(tmp_path): + raise OSError(107, "Transport endpoint is not connected") + return real_scandir(path) + + monkeypatch.setattr(os, "scandir", dead_mount) + os.stat(tmp_path) # stat still succeeds + + assert not mounter._responds(str(tmp_path)) + + # ─────────────────── _cuid_of() ─────────────────── @pytest.mark.parametrize( diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala index d24b267bf95..9227e02fad2 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala @@ -27,7 +27,7 @@ import java.net.{HttpURLConnection, URI} import java.nio.charset.StandardCharsets import java.nio.file.{Files, Path, Paths} import scala.io.Source -import scala.util.Using +import scala.util.{Try, Using} /** * Makes a versioned LakeFS repository readable inside this computing unit's pod. @@ -141,7 +141,7 @@ object RepositoryMountManager 35000 ) -private object InPodMount { +private[amber] object InPodMount { def postJson(url: String, body: String, jwt: String): Unit = { val connection = URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection] @@ -165,9 +165,20 @@ private object InPodMount { } } + /** + * True if a live FUSE mount is at `mountPoint`. + * + * Being listed is not enough. A mount whose GeeseFS process has died keeps its entry, and + * the kernel can still answer stat() from attributes it cached, but every read fails with + * "Transport endpoint is not connected". Counted as mounted, it would never be requested + * again -- and the mounter, which replaces a dead mount when asked, would never be asked. + */ + def isFuseMounted(mountPoint: Path): Boolean = + isListedAsFuse(mountPoint) && responds(mountPoint) + // /proc/mounts is the kernel's read-only view of this pod's mount table: the kernel adds the // entry when a propagated GeeseFS mount arrives and removes it on unmount, so nothing writes it. - def isFuseMounted(mountPoint: Path): Boolean = { + private def isListedAsFuse(mountPoint: Path): Boolean = { if (!Files.exists(mountPoint)) { return false } @@ -179,4 +190,8 @@ private object InPodMount { } }.getOrElse(false) } + + /** True if `directory` can be listed, which a dead FUSE mount cannot. */ + def responds(directory: Path): Boolean = + Try(Using.resource(Files.newDirectoryStream(directory))(_.iterator().hasNext)).isSuccess } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala index 46273d96ac0..59cda79c41f 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala @@ -19,12 +19,12 @@ package org.apache.texera.amber.storage -import org.apache.texera.amber.core.storage.RepositoryMountManager +import org.apache.texera.amber.core.storage.{InPodMount, RepositoryMountManager} import org.apache.texera.common.config.EnvironmentalVariable import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.nio.file.{Path, Paths} +import java.nio.file.{Files, Path, Paths} import scala.collection.mutable class RepositoryMountManagerSpec extends AnyFlatSpec with Matchers { @@ -115,4 +115,32 @@ class RepositoryMountManagerSpec extends AnyFlatSpec with Matchers { fixture.requests shouldBe empty } } + + // A mount whose GeeseFS process died stays in /proc/mounts and can even stat, but cannot be + // listed. Telling it apart is what lets ensureMounted ask again instead of handing the + // operator an unreadable directory. A real dead FUSE mount cannot be made in a unit test, so + // these pin the rule itself: whatever cannot be listed is not live. + "InPodMount.responds" should "accept a directory that can be listed" in { + val dir = Files.createTempDirectory("mount-live") + try InPodMount.responds(dir) shouldBe true + finally Files.delete(dir) + } + + it should "reject a path that cannot be listed" in { + val dir = Files.createTempDirectory("mount-dead") + val file = Files.createFile(dir.resolve("not-a-directory")) + try { + InPodMount.responds(dir.resolve("missing")) shouldBe false + InPodMount.responds(file) shouldBe false + } finally { + Files.delete(file) + Files.delete(dir) + } + } + + "InPodMount.isFuseMounted" should "not count a plain directory as a mount" in { + val dir = Files.createTempDirectory("mount-none") + try InPodMount.isFuseMounted(dir) shouldBe false + finally Files.delete(dir) + } }