You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/07/22 14:22:33 UTC
(tooling-trusted-releases) branch sbp updated: Remove the temporary performance monitoring code
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
The following commit(s) were added to refs/heads/sbp by this push:
new c63be6d8 Remove the temporary performance monitoring code
c63be6d8 is described below
commit c63be6d8b09bccf5ce236bffd7da7a4abad41e84
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Wed Jul 22 15:22:03 2026 +0100
Remove the temporary performance monitoring code
---
atr/archives.py | 15 ++-------------
atr/config.py | 2 --
atr/log.py | 13 -------------
atr/manager.py | 22 ----------------------
atr/server.py | 33 ---------------------------------
atr/tasks/maintenance.py | 38 --------------------------------------
atr/tasks/quarantine.py | 7 ++-----
atr/util.py | 16 ----------------
atr/worker.py | 38 --------------------------------------
tests/unit/test_quarantine_task.py | 18 +++++++++---------
10 files changed, 13 insertions(+), 189 deletions(-)
diff --git a/atr/archives.py b/atr/archives.py
index fb13a202..4044ac8b 100644
--- a/atr/archives.py
+++ b/atr/archives.py
@@ -17,7 +17,6 @@
import asyncio
import contextlib
import os
-import time
from typing import Final
import aiofiles.os
@@ -58,23 +57,13 @@ def extract(
extracted_paths.append(entry.path)
try:
- bytes_written = extract_measured(archive_path, extract_dir, cfg)
+ report = exarch.extract_archive(str(archive_path), extract_dir, cfg)
except exarch.QuotaExceededError as exc:
raise ExtractionError(f"Extraction exceeded size limit of {max_size} bytes: {exc}") from exc
except Exception as exc:
raise ExtractionError(f"Failed to extract archive: {exc}") from exc
- return bytes_written, extracted_paths
-
-
-def extract_measured(archive_path: safe.StatePath, extract_dir: str, cfg: exarch.SecurityConfig) -> int:
- size = os.path.getsize(archive_path)
- started = time.monotonic()
- report = exarch.extract_archive(str(archive_path), extract_dir, cfg)
- log.resources(
- f"extract {archive_path} in={size} out={report.bytes_written} seconds={time.monotonic() - started:.1f}"
- )
- return report.bytes_written
+ return report.bytes_written, extracted_paths
def extraction_config() -> exarch.SecurityConfig:
diff --git a/atr/config.py b/atr/config.py
index 5cbfb312..1411eb03 100644
--- a/atr/config.py
+++ b/atr/config.py
@@ -125,7 +125,6 @@ class AppConfig:
AUTH_AUDIT_LOG_FILE = os.path.join(STATE_DIR, "audit", "auth-audit.log")
PERFORMANCE_LOG_FILE = os.path.join(STATE_DIR, "logs", "route-performance.log")
REQUEST_LOG_FILE = os.path.join(STATE_DIR, "logs", "requests.log")
- RESOURCES_LOG_FILE = os.path.join(STATE_DIR, "logs", "resources.log")
TASK_LOG_FILE = os.path.join(STATE_DIR, "logs", "tasks.log")
# Apache RAT configuration
@@ -265,7 +264,6 @@ def validate() -> None:
(conf.STORAGE_AUDIT_LOG_FILE, "STORAGE_AUDIT_LOG_FILE"),
(conf.AUTH_AUDIT_LOG_FILE, "AUTH_AUDIT_LOG_FILE"),
(conf.PERFORMANCE_LOG_FILE, "PERFORMANCE_LOG_FILE"),
- (conf.RESOURCES_LOG_FILE, "RESOURCES_LOG_FILE"),
(conf.TASK_LOG_FILE, "TASK_LOG_FILE"),
]
relative_paths = [
diff --git a/atr/log.py b/atr/log.py
index 6a5c3561..57c4413d 100644
--- a/atr/log.py
+++ b/atr/log.py
@@ -29,7 +29,6 @@ from typing import Any, Final
import structlog
PERFORMANCE: logging.Logger | None = None
-RESOURCES: logging.Logger | None = None
_global_recent_logs: collections.deque[str] | None = None
_global_recent_logs_lock = threading.Lock()
@@ -224,18 +223,6 @@ def request_context_fields() -> dict[str, str]:
return {"request_id": request_id}
-def resources(msg: str) -> None:
- if RESOURCES is not None:
- RESOURCES.info(msg)
-
-
-def resources_init() -> None:
- import atr.config as config
-
- global RESOURCES
- RESOURCES = _line_logger("log.resources", config.get().RESOURCES_LOG_FILE)
-
-
def set_asf_uid(asfuid: str | None) -> None:
add_context(asfuid=asfuid)
diff --git a/atr/manager.py b/atr/manager.py
index 72b7e341..a8b40edb 100644
--- a/atr/manager.py
+++ b/atr/manager.py
@@ -25,8 +25,6 @@ import io
import os
import signal
import sys
-import time
-from typing import Final
import sqlalchemy
import sqlalchemy.engine as engine
@@ -45,8 +43,6 @@ global_worker_debug: bool = False
# Can't use "StringClass" | None, must use Optional["StringClass"] for forward references
global_worker_manager: WorkerManager | None = None
-_POOL_RESOURCES_INTERVAL_SECONDS: Final = 60.0
-
class WorkerManager:
"""Manager for a pool of worker processes."""
@@ -65,7 +61,6 @@ class WorkerManager:
self.budget_index = 0
self.running = False
self.check_task: asyncio.Task | None = None
- self.resources_logged_at = 0.0
async def start(self) -> None:
"""Start the worker manager."""
@@ -197,9 +192,6 @@ class WorkerManager:
while self.running:
try:
await self.check_workers()
- if (time.monotonic() - self.resources_logged_at) >= _POOL_RESOURCES_INTERVAL_SECONDS:
- self.resources_logged_at = time.monotonic()
- await self.log_pool_resources()
await asyncio.sleep(self.check_interval_seconds)
except asyncio.CancelledError:
break
@@ -254,20 +246,6 @@ class WorkerManager:
# Reset any tasks that were being processed by now inactive workers
await self.reset_broken_tasks()
- async def log_pool_resources(self) -> None:
- async with db.session() as data:
- count_query = (
- sqlmodel.select(sqlalchemy.func.count())
- .select_from(sql.Task)
- .where(sql.Task.status == sql.TaskStatus.QUEUED)
- )
- queued = (await data.execute(count_query)).scalar_one()
- rss_parts = []
- for pid in sorted(self.workers):
- rss, _ = await asyncio.to_thread(util.proc_memory_kb, pid)
- rss_parts.append(f"{pid}:{'?' if (rss is None) else rss}")
- log.resources(f"pool queued={queued} workers={len(self.workers)} rss={','.join(rss_parts)}")
-
async def terminate_long_running_task(self, task: sql.Task, worker: WorkerProcess, task_id: int, pid: int) -> None:
"""
Terminate a task that has been running for too long.
diff --git a/atr/server.py b/atr/server.py
index 756cfb46..4058a213 100644
--- a/atr/server.py
+++ b/atr/server.py
@@ -26,7 +26,6 @@ import multiprocessing
import os
import pathlib
import queue
-import resource
import stat
import sys
import time
@@ -81,8 +80,6 @@ import atr.web as web
# We should probably find a cleaner way to do this
app: base.QuartApp | None = None
-_RESOURCES_GAUGE_INTERVAL_SECONDS: Final = 300
-
# The order of these migrations must be checked carefully to avoid conflicts
_MIGRATIONS: Final[list[tuple[str, str]]] = [
# Archives
@@ -366,9 +363,6 @@ def _app_setup_lifecycle(app: base.QuartApp, app_config: type[config.AppConfig])
ssh_rate_limit_task = asyncio.create_task(ssh.rate_limit_cleanup_loop())
app.extensions["ssh_rate_limit_task"] = ssh_rate_limit_task
- resources_gauge_task = asyncio.create_task(_resources_gauge_loop())
- app.extensions["resources_gauge_task"] = resources_gauge_task
-
@app.after_serving
async def shutdown() -> None:
"""Clean up services after the app stops serving requests."""
@@ -389,11 +383,6 @@ def _app_setup_lifecycle(app: base.QuartApp, app_config: type[config.AppConfig])
with contextlib.suppress(asyncio.CancelledError):
await ssh_cleanup_task
- if resources_gauge_task := app.extensions.get("resources_gauge_task"):
- resources_gauge_task.cancel()
- with contextlib.suppress(asyncio.CancelledError):
- await resources_gauge_task
-
ssh_server = app.extensions.get("ssh_server")
if ssh_server:
await ssh.server_stop(ssh_server)
@@ -706,7 +695,6 @@ def _create_app(app_config: type[config.AppConfig]) -> base.QuartApp:
_app_dirs_setup(app_config.STATE_DIR, hot_reload)
_validate_secrets_permissions(pathlib.Path(app_config.STATE_DIR))
log.performance_init()
- log.resources_init()
app = _app_create_base(app_config)
app.sessions = sessions.Store()
jwtoken.setup_signing_key(app)
@@ -1221,27 +1209,6 @@ async def _reset_request_log_context():
log.add_context(user_id=asf_uid)
-async def _resources_gauge_loop() -> None:
- while True:
- await asyncio.sleep(_RESOURCES_GAUGE_INTERVAL_SECONDS)
- try:
- rss, hwm = await asyncio.to_thread(util.proc_memory_kb, os.getpid())
- fds: int | str
- try:
- fds = len(await asyncio.to_thread(os.listdir, "/proc/self/fd"))
- except OSError:
- fds = "?"
- children = resource.getrusage(resource.RUSAGE_CHILDREN)
- log.resources(
- f"app rss={'?' if (rss is None) else rss} hwm={'?' if (hwm is None) else hwm} fds={fds}"
- f" cutime={children.ru_utime:.3f} cstime={children.ru_stime:.3f}"
- )
- except asyncio.CancelledError:
- raise
- except Exception:
- log.exception("Failed to log app resources gauge")
-
-
def _set_file_permissions_to_read_only() -> None:
"""Set permissions of all files in the unfinished and finished directories to read only."""
# TODO: After a migration period, incorrect permissions should be an error
diff --git a/atr/tasks/maintenance.py b/atr/tasks/maintenance.py
index 09c0c689..2be8d10c 100644
--- a/atr/tasks/maintenance.py
+++ b/atr/tasks/maintenance.py
@@ -15,16 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-import asyncio
import datetime
-import os
import time
from typing import Final
-import sqlalchemy
import sqlmodel
-import atr.config as config
import atr.db as db
import atr.log as log
import atr.models.args as args
@@ -36,7 +32,6 @@ import atr.tasks.inactivity as inactivity
_EXPIRED_TOKEN_RETENTION_DAYS: Final[int] = 30
_FORM_ERROR_TTL_SECONDS: Final[int] = 24 * 60 * 60
-_RESOURCES_LOG_MAX_BYTES: Final[int] = 64 * 1024 * 1024
class MaintenanceError(Exception):
@@ -58,8 +53,6 @@ async def run(task_args: args.MaintenanceArgs) -> results.Results | None:
await _storage_maintenance()
await _workflow_auth_maintenance()
await _inactivity_maintenance()
- await _database_survey_maintenance()
- await _resources_log_maintenance()
log.info(
"Storage maintenance completed successfully",
@@ -74,24 +67,6 @@ async def run(task_args: args.MaintenanceArgs) -> results.Results | None:
raise MaintenanceError(error_msg) from e
-async def _database_survey_maintenance() -> None:
- try:
- async with db.session() as data:
- pages = (await data.execute(sqlalchemy.text("PRAGMA page_count"))).scalar_one()
- page_size = (await data.execute(sqlalchemy.text("PRAGMA page_size"))).scalar_one()
- freelist = (await data.execute(sqlalchemy.text("PRAGMA freelist_count"))).scalar_one()
- counts = {}
- for table in ("task", "checkresult", "signingcertificate"):
- counts[table] = (await data.execute(sqlalchemy.text(f"SELECT COUNT(*) FROM {table}"))).scalar_one()
- except Exception:
- log.exception("Database survey failed")
- return
- log.resources(
- f"db pages={pages} page_size={page_size} freelist={freelist} task={counts['task']}"
- f" checkresult={counts['checkresult']} keys={counts['signingcertificate']}"
- )
-
-
async def _expired_pats_maintenance() -> None:
via = sql.validate_instrumented_attribute
cutoff = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=_EXPIRED_TOKEN_RETENTION_DAYS)
@@ -119,19 +94,6 @@ async def _inactivity_maintenance() -> None:
log.exception(f"Inactivity action failed for release {plan.release_key!r}")
-async def _resources_log_maintenance() -> None:
- path = config.get().RESOURCES_LOG_FILE
- try:
- size = await asyncio.to_thread(os.path.getsize, path)
- if size < _RESOURCES_LOG_MAX_BYTES:
- return
- await asyncio.to_thread(os.replace, path, path + ".1")
- except OSError:
- log.exception("Resources log rotation failed")
- return
- log.info(f"Rotated resources log at {size} bytes")
-
-
async def _session_data_maintenance() -> None:
via = sql.validate_instrumented_attribute
cutoff = time.time() - _FORM_ERROR_TTL_SECONDS
diff --git a/atr/tasks/quarantine.py b/atr/tasks/quarantine.py
index a8ec9a37..b5f6d45a 100644
--- a/atr/tasks/quarantine.py
+++ b/atr/tasks/quarantine.py
@@ -25,10 +25,10 @@ import pathlib
import shutil
import time
import uuid
-from typing import TYPE_CHECKING
import aiofiles.os
import aioshutil
+import exarch
import atr.archives as archives
import atr.attestable as attestable
@@ -49,9 +49,6 @@ import atr.swhid as swhid
import atr.tasks.checks as checks
import atr.util as util
-if TYPE_CHECKING:
- import exarch
-
def backfill_archive_cache() -> list[tuple[str, safe.StatePath, float]]:
done_file = _backfill_done_file()
@@ -230,7 +227,7 @@ def _extract_archive_to_dir(
try:
staging_dir_path.mkdir(parents=False, exist_ok=False)
start = time.monotonic()
- archives.extract_measured(archive_path, str(staging_dir), extraction_cfg)
+ exarch.extract_archive(str(archive_path), str(staging_dir), extraction_cfg)
if compute_swhid:
try:
root = archives.single_root_dir(staging_dir)
diff --git a/atr/util.py b/atr/util.py
index 5467d22d..c713701e 100644
--- a/atr/util.py
+++ b/atr/util.py
@@ -1187,22 +1187,6 @@ def plural(count: int, singular: str, plural_form: str | None = None, *, include
return word
-def proc_memory_kb(pid: int) -> tuple[int | None, int | None]:
- try:
- with open(f"/proc/{pid}/status", encoding="utf-8") as f:
- text = f.read()
- except OSError:
- return None, None
- rss = None
- hwm = None
- for line in text.splitlines():
- if line.startswith("VmRSS:"):
- rss = int(line.split()[1])
- elif line.startswith("VmHWM:"):
- hwm = int(line.split()[1])
- return rss, hwm
-
-
def publication_check_url(
committee: sql.Committee,
suffix: safe.RelPath | None,
diff --git a/atr/worker.py b/atr/worker.py
index 3da1e76c..4057cee0 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -30,8 +30,6 @@ import logging
import os
import resource
import signal
-import sys
-import time
import traceback
from collections.abc import Awaitable, Callable
from typing import Any, Final
@@ -54,9 +52,6 @@ import atr.tasks.task as task
_CPU_LIMIT_SECONDS: Final = 300
_MEMORY_LIMIT_BYTES: Final = 3 * 1024 * 1024 * 1024
-# The ru_maxrss unit is bytes on macOS, and kilobytes on Linux
-_MAXRSS_KILOBYTES_DIVISOR: Final = 1024 if (sys.platform == "darwin") else 1
-
# Successful recurring tasks are appended here as JSON lines instead of kept in the database
_TASK_LOG_LOGGER: Final = "atr.tasks.log"
@@ -73,7 +68,6 @@ def main() -> None:
os.chdir(conf.STATE_DIR)
_setup_logging()
- log.resources_init()
log.add_context(worker_pid=os.getpid())
log.info(f"Starting worker process with pid {os.getpid()}")
@@ -106,7 +100,6 @@ def main() -> None:
# If the worker decides to stop running (see #230 in _worker_loop_run()), shutdown the database gracefully
asyncio.run(db.shutdown_database())
log.info("Exiting worker process")
- log.listeners_stop()
async def _execute_check_task(
@@ -321,10 +314,6 @@ async def _task_process(task_id: int, task_type: str, task_args: list[str] | dic
await _task_result_process(task_id, None, task.FAILED, str(e))
return
- wall_started = time.monotonic()
- rusage_self_before = resource.getrusage(resource.RUSAGE_SELF)
- rusage_children_before = resource.getrusage(resource.RUSAGE_CHILDREN)
-
task_results: results.Results | None
try:
if (
@@ -362,36 +351,9 @@ async def _task_process(task_id: int, task_type: str, task_args: list[str] | dic
error_details = traceback.format_exc()
log.error(f"Task {task_id} failed processing: {error_details}")
error = str(e)
- _task_resources_log(
- task_id, task_type, status, time.monotonic() - wall_started, rusage_self_before, rusage_children_before
- )
await _task_result_process(task_id, task_results, status, error)
-def _task_resources_log(
- task_id: int,
- task_type: str,
- status: sql.TaskStatus,
- wall: float,
- rusage_self_before: resource.struct_rusage,
- rusage_children_before: resource.struct_rusage,
-) -> None:
- rusage_self = resource.getrusage(resource.RUSAGE_SELF)
- rusage_children = resource.getrusage(resource.RUSAGE_CHILDREN)
- maxrss = rusage_self.ru_maxrss // _MAXRSS_KILOBYTES_DIVISOR
- cmaxrss = rusage_children.ru_maxrss // _MAXRSS_KILOBYTES_DIVISOR
- rose_self = int(rusage_self.ru_maxrss > rusage_self_before.ru_maxrss)
- rose_children = int(rusage_children.ru_maxrss > rusage_children_before.ru_maxrss)
- log.resources(
- f"task {task_id} {task_type} {status} wall={wall:.3f}"
- f" utime={rusage_self.ru_utime - rusage_self_before.ru_utime:.3f}"
- f" stime={rusage_self.ru_stime - rusage_self_before.ru_stime:.3f}"
- f" cutime={rusage_children.ru_utime - rusage_children_before.ru_utime:.3f}"
- f" cstime={rusage_children.ru_stime - rusage_children_before.ru_stime:.3f}"
- f" maxrss={maxrss} cmaxrss={cmaxrss} rose={rose_self}{rose_children}"
- )
-
-
async def _task_result_process(
task_id: int, task_results: results.Results | None, status: sql.TaskStatus, error: str | None = None
) -> None:
diff --git a/tests/unit/test_quarantine_task.py b/tests/unit/test_quarantine_task.py
index 29fd58e9..58a484be 100644
--- a/tests/unit/test_quarantine_task.py
+++ b/tests/unit/test_quarantine_task.py
@@ -215,7 +215,7 @@ async def test_extract_archives_discards_staging_dir_on_enotempty_collision(
tmp_root = temp_dir / "temporary"
recorded: dict[str, pathlib.Path] = {}
- def extract_measured(_archive_path: safe.StatePath, extract_dir: str, _config: object) -> None:
+ def extract_archive(_archive_path: str, extract_dir: str, _config: object) -> None:
staging_dir = pathlib.Path(extract_dir)
recorded["staging_dir"] = staging_dir
(staging_dir / "content.txt").write_text("staged")
@@ -228,7 +228,7 @@ async def test_extract_archives_discards_staging_dir_on_enotempty_collision(
monkeypatch.setattr(quarantine.paths, "get_archives_dir", lambda: cache_root)
monkeypatch.setattr(quarantine.paths, "get_tmp_dir", lambda: tmp_root)
- monkeypatch.setattr(quarantine.archives, "extract_measured", extract_measured)
+ monkeypatch.setattr(quarantine.exarch, "extract_archive", extract_archive)
monkeypatch.setattr(quarantine.os, "rename", rename)
entries = [sql.QuarantineFileEntryV1(rel_path=archive_rel_path, size_bytes=7, content_hash="blake3:ghi", errors=[])]
@@ -261,7 +261,7 @@ async def test_extract_archives_discards_staging_dir_when_other_worker_wins(
tmp_root = temp_dir / "temporary"
recorded: dict[str, pathlib.Path] = {}
- def extract_measured(_archive_path: safe.StatePath, extract_dir: str, _config: object) -> None:
+ def extract_archive(_archive_path: str, extract_dir: str, _config: object) -> None:
staging_dir = pathlib.Path(extract_dir)
recorded["staging_dir"] = staging_dir
(staging_dir / "content.txt").write_text("staged")
@@ -274,7 +274,7 @@ async def test_extract_archives_discards_staging_dir_when_other_worker_wins(
monkeypatch.setattr(quarantine.paths, "get_archives_dir", lambda: cache_root)
monkeypatch.setattr(quarantine.paths, "get_tmp_dir", lambda: tmp_root)
- monkeypatch.setattr(quarantine.archives, "extract_measured", extract_measured)
+ monkeypatch.setattr(quarantine.exarch, "extract_archive", extract_archive)
monkeypatch.setattr(quarantine.os, "rename", rename)
entries = [sql.QuarantineFileEntryV1(rel_path=archive_rel_path, size_bytes=7, content_hash="blake3:def", errors=[])]
@@ -306,12 +306,12 @@ async def test_extract_archives_propagates_exarch_error_to_file_entry(
cache_root = temp_dir / "cache"
tmp_root = temp_dir / "temporary"
- def extract_measured(_archive_path: safe.StatePath, _extract_dir: str, _config: object) -> None:
+ def extract_archive(_archive_path: str, _extract_dir: str, _config: object) -> None:
raise RuntimeError("unsafe zip detected")
monkeypatch.setattr(quarantine.paths, "get_archives_dir", lambda: cache_root)
monkeypatch.setattr(quarantine.paths, "get_tmp_dir", lambda: tmp_root)
- monkeypatch.setattr(quarantine.archives, "extract_measured", extract_measured)
+ monkeypatch.setattr(quarantine.exarch, "extract_archive", extract_archive)
entries = [sql.QuarantineFileEntryV1(rel_path=archive_rel_path, size_bytes=7, content_hash="blake3:bad", errors=[])]
@@ -341,15 +341,15 @@ async def test_extract_archives_stages_in_temporary_then_promotes(
tmp_root = temp_dir / "temporary"
recorded: dict[str, str] = {}
- def extract_measured(archive_path: safe.StatePath, extract_dir: str, _config: object) -> None:
- recorded["archive_path"] = str(archive_path)
+ def extract_archive(archive_path: str, extract_dir: str, _config: object) -> None:
+ recorded["archive_path"] = archive_path
recorded["extract_dir"] = extract_dir
extract_path = pathlib.Path(extract_dir)
(extract_path / "content.txt").write_text("cached")
monkeypatch.setattr(quarantine.paths, "get_archives_dir", lambda: cache_root)
monkeypatch.setattr(quarantine.paths, "get_tmp_dir", lambda: tmp_root)
- monkeypatch.setattr(quarantine.archives, "extract_measured", extract_measured)
+ monkeypatch.setattr(quarantine.exarch, "extract_archive", extract_archive)
entries = [sql.QuarantineFileEntryV1(rel_path=archive_rel_path, size_bytes=7, content_hash="blake3:abc", errors=[])]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org