Cache agent inventory manifests by modification time
This commit is contained in:
+31
-5
@@ -32,7 +32,7 @@ import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
VERSION = "1.4.10"
|
||||
VERSION = "1.4.11"
|
||||
LABEL = "io.electric-cherry"
|
||||
MAX_FILE = 32 * 1024 * 1024
|
||||
MANIFEST_MAX_ENTRIES = 10000
|
||||
@@ -918,6 +918,8 @@ class Agent:
|
||||
self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
|
||||
self.library_progress = {}
|
||||
self.library_transfers = {}
|
||||
self.inventory_manifest_cache = {}
|
||||
self.inventory_manifest_lock = threading.Lock()
|
||||
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
|
||||
self.executing_id = None
|
||||
|
||||
@@ -1191,6 +1193,25 @@ class Agent:
|
||||
continue
|
||||
return samples
|
||||
|
||||
def inventory_manifest(self, marker):
|
||||
# Inventory only: launch/download verification must still read disk.
|
||||
# Include file identity so atomic replacements invalidate even when
|
||||
# their size and modification time match the previous manifest.
|
||||
with self.inventory_manifest_lock:
|
||||
try:
|
||||
metadata = marker.stat()
|
||||
signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
|
||||
cached = self.inventory_manifest_cache.get(marker)
|
||||
if cached is not None and cached[0] == signature:
|
||||
return cached[1]
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
value = json.loads(marker.read_text())
|
||||
self.inventory_manifest_cache[marker] = (signature, value)
|
||||
return value
|
||||
except (OSError, ValueError):
|
||||
self.inventory_manifest_cache.pop(marker, None)
|
||||
raise
|
||||
|
||||
def inventory(self):
|
||||
cpu_usage = self.cpu_usage_percent()
|
||||
usage = shutil.disk_usage(self.root)
|
||||
@@ -1210,13 +1231,18 @@ class Agent:
|
||||
libraries, caches = [], []
|
||||
progress = {key: value.copy() for key, value in list(self.library_progress.items())}
|
||||
progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
|
||||
for marker in (self.root / "libraries").glob("*/*/manifest.json"):
|
||||
manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
|
||||
partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
|
||||
with self.inventory_manifest_lock:
|
||||
for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
|
||||
del self.inventory_manifest_cache[marker]
|
||||
for marker in manifests:
|
||||
# A crash between writing a manifest and its atomic promotion must
|
||||
# never expose the staging directory as a ready library.
|
||||
if marker.parent.name.startswith("."):
|
||||
continue
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
# Rechecking an existing cache must revoke readiness until its
|
||||
# entire contents have passed verification again.
|
||||
if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
|
||||
@@ -1233,9 +1259,9 @@ class Agent:
|
||||
except (OSError, ValueError, KeyError):
|
||||
continue
|
||||
ready = {item["id"] for item in libraries}
|
||||
for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
|
||||
for marker in partial_manifests:
|
||||
try:
|
||||
value = json.loads(marker.read_text())
|
||||
value = self.inventory_manifest(marker)
|
||||
library_id = identifier(value["libraryId"])
|
||||
if library_id in ready or library_id in progress:
|
||||
continue
|
||||
|
||||
@@ -386,6 +386,68 @@ class AgentTests(unittest.TestCase):
|
||||
self.agent.journal.close()
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_inventory_reuses_manifests_until_mtime_changes(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
original_read = Path.read_text
|
||||
reads = []
|
||||
|
||||
def read(path, *args, **kwargs):
|
||||
if path == marker:
|
||||
reads.append(path)
|
||||
return original_read(path, *args, **kwargs)
|
||||
|
||||
with patch.object(Path, "read_text", read):
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
|
||||
self.assertEqual(len(reads), 1)
|
||||
metadata = marker.stat()
|
||||
marker.write_text(json.dumps({**saved, "size": 8}))
|
||||
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000))
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
self.assertEqual(len(reads), 2)
|
||||
|
||||
def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
self.agent.inventory()
|
||||
metadata = marker.stat()
|
||||
module.atomic_json(marker, {**saved, "size": 8})
|
||||
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
|
||||
self.assertEqual(marker.stat().st_size, metadata.st_size)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
|
||||
def test_inventory_discards_invalid_and_removed_manifests(self):
|
||||
marker = self.library / "manifest.json"
|
||||
saved = json.loads(marker.read_text())
|
||||
self.agent.inventory()
|
||||
marker.write_text("invalid JSON")
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
|
||||
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
|
||||
module.atomic_json(marker, saved)
|
||||
self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1)
|
||||
marker.unlink()
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
|
||||
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
|
||||
module.atomic_json(marker, {**saved, "size": 8})
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
|
||||
|
||||
def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self):
|
||||
self.agent.inventory()
|
||||
progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3}
|
||||
module.atomic_json(self.library / "transfer.json", progress)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing")
|
||||
module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5})
|
||||
cache = self.agent.inventory()["libraryCaches"][0]
|
||||
self.assertEqual(cache["status"], "failed")
|
||||
self.assertEqual(cache["bytesVerified"], 5)
|
||||
(self.library / "content/game.ff").write_bytes(b"bad-data!")
|
||||
with self.assertRaises(AgentError):
|
||||
self.agent.verify_library(self.library)
|
||||
(self.library / "manifest.json").write_text("invalid JSON")
|
||||
with self.assertRaises(ValueError):
|
||||
self.agent.verify_library(self.library)
|
||||
|
||||
def test_library_remove_preserves_other_versions_and_is_idempotent(self):
|
||||
other = self.agent.library_dir("t5", "other-library")
|
||||
other.mkdir()
|
||||
@@ -1179,6 +1241,17 @@ class AgentTests(unittest.TestCase):
|
||||
self.assertEqual(caches[0]["status"], "missing")
|
||||
self.assertEqual(caches[0]["bytesDownloaded"], len(content))
|
||||
self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"])
|
||||
(staging / "content/zone/test.ff").write_bytes(content[:5])
|
||||
original_read = Path.read_text
|
||||
|
||||
def read(path, *args, **kwargs):
|
||||
if path == staging / "source.json":
|
||||
self.fail("Unchanged partial manifest should be cached")
|
||||
return original_read(path, *args, **kwargs)
|
||||
|
||||
with patch.object(Path, "read_text", read):
|
||||
partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
|
||||
self.assertEqual(partial["bytesDownloaded"], 5)
|
||||
self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3}
|
||||
caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
|
||||
self.assertEqual(len(caches), 1)
|
||||
|
||||
Reference in New Issue
Block a user