CHANGELOG.md

@@ -2,24 +2,6 @@
 
 ## [Unreleased]
 
-## 0.9.0
-
-Date: 2025-02-18
-
-- 🐛 — Fix worker timeout for old results cleaning in recurring tasks (#84)
-
-💥 Old results are now removed by their age, not based on their number.
-
-💥 Warning: `max_results` setting has been replaced by `max_results_age`, which is a duration.
-Use `argos server generate-config > /etc/argos/config.yaml-dist` to generate
-a new example configuration file.
-
-## 0.8.2
-
-Date: 2025-02-18
-
-- 🐛 — Fix recurring tasks with gunicorn
-
 ## 0.8.1
 
 Date: 2025-02-18
@@ -38,15 +20,13 @@ Date: 2025-02-18
 - 🐛 — Automatically reconnect to LDAP if unreachable (#81)
 - 🐛 — Better httpx.RequestError handling (#83)
 
-💥 Warning: there is new settings to add to your configuration file.
+💥 Warning: there is now new settings to add to your configuration file.
 Use `argos server generate-config > /etc/argos/config.yaml-dist` to generate
 a new example configuration file.
 
 💥 You don’t need cron tasks anymore!
 Remove your old cron tasks as they will now do nothing but generating errors.
 
-NB: You may want to add `--enqueue` to `reload-config` command in your systemd file.
-
 ## 0.7.4
 
 Date: 2025-02-12

argos/__init__.py

@@ -1 +1 @@
-VERSION = "0.9.0"
+VERSION = "0.8.1"

argos/config-example.yaml

@@ -152,11 +152,9 @@ ssl:
 # Argos will execute some tasks in the background for you
 # every 2 minutes and needs some configuration for that
 recurring_tasks:
-  # Maximum age of results
-  # Use m for minutes, h for hours, d for days
-  # w for weeks, M for months, y for years
-  # See https://github.com/timwedde/durations_nlp#scales-reference for details
-  max_results_age: "1d"
+  # Max number of results per tasks you want to keep
+  # Minimum value is 1, default is 100
+  max_results: 100
   # Max number of seconds a task can be locked
   # Minimum value is 61, default is 100
   max_lock_seconds: 100

argos/schemas/config.py

@@ -49,14 +49,17 @@ class SSL(BaseModel):
 
 
 class RecurringTasks(BaseModel):
-    max_results_age: float
+    max_results: int
     max_lock_seconds: int
     time_without_agent: int
 
-    @field_validator("max_results_age", mode="before")
-    def parse_max_results_age(cls, value):
-        """Convert the configured maximum results age to seconds"""
-        return Duration(value).to_seconds()
+    @field_validator("max_results", mode="before")
+    def parse_max_results(cls, value):
+        """Ensure that max_results is higher than 0"""
+        if value >= 1:
+            return value
+
+        return 100
 
     @field_validator("max_lock_seconds", mode="before")
     def parse_max_lock_seconds(cls, value):

argos/server/main.py

@@ -6,7 +6,6 @@ from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 from fastapi_login import LoginManager
 from fastapi_utils.tasks import repeat_every
-from psutil import Process
 from sqlalchemy import create_engine, event
 from sqlalchemy.orm import sessionmaker
 
@@ -117,41 +116,25 @@ def create_manager(cookie_secret: str) -> LoginManager:
 @repeat_every(seconds=120, logger=logger)
 async def recurring_tasks() -> None:
     """Recurring DB cleanup and watch-agents tasks"""
-    # If we are using gunicorn
-    if not hasattr(app.state, "SessionLocal"):
-        parent_process = Process(os.getppid())
-        children = parent_process.children(recursive=True)
-        # Start the task only once, not for every worker
-        if children[0].pid == os.getpid():
-            # and we need to setup database engine
-            setup_database(app)
-        else:
-            return None
-
     set_log_level("info", quiet=True)
     logger.info("Start background recurring tasks")
-
     with app.state.SessionLocal() as db:
         config = app.state.config.recurring_tasks
-
-        agents = await queries.get_recent_agents_count(db, config.time_without_agent)
-        if agents == 0:
-            no_agent_alert(app.state.config)
-        logger.info("Agent presence checked")
-
-        removed = await queries.remove_old_results(db, config.max_results_age)
+        removed = await queries.remove_old_results(db, config.max_results)
         logger.info("%i result(s) removed", removed)
 
         updated = await queries.release_old_locks(db, config.max_lock_seconds)
         logger.info("%i lock(s) released", updated)
 
+        agents = await queries.get_recent_agents_count(db, config.time_without_agent)
+        if agents == 0:
+            no_agent_alert(app.state.config)
+
         processed_jobs = await queries.process_jobs(db)
         logger.info("%i job(s) processed", processed_jobs)
 
     logger.info("Background recurring tasks ended")
 
-    return None
-
 
 @asynccontextmanager
 async def lifespan(appli: FastAPI):

argos/server/queries.py

@@ -4,7 +4,7 @@ from hashlib import sha256
 from typing import List
 from urllib.parse import urljoin
 
-from sqlalchemy import asc, func, Select
+from sqlalchemy import asc, desc, func, Select
 from sqlalchemy.orm import Session
 
 from argos import schemas
@@ -409,13 +409,28 @@ async def reschedule_all(db: Session):
     db.commit()
 
 
-async def remove_old_results(db: Session, max_results_age: float):
-    """Remove old results, base on age"""
-    max_acceptable_time = datetime.now() - timedelta(seconds=max_results_age)
-    deleted = (
-        db.query(Result).filter(Result.submitted_at < max_acceptable_time).delete()
-    )
-    db.commit()
+async def remove_old_results(db: Session, max_results: int):
+    tasks = db.query(Task).all()
+    deleted = 0
+    for task in tasks:
+        # Get the id of the oldest result to keep
+        subquery = (
+            db.query(Result.id)
+            .filter(Result.task_id == task.id)
+            .order_by(desc(Result.id))
+            .limit(max_results)
+            .subquery()
+        )
+        min_id = db.query(func.min(subquery.c.id)).scalar()  #  pylint: disable-msg=not-callable
+
+        # Delete all the results older than min_id
+        if min_id:
+            deleted += (
+                db.query(Result)
+                .where(Result.id < min_id, Result.task_id == task.id)
+                .delete()
+            )
+            db.commit()
 
     return deleted
 

docs/cli.md

@@ -60,7 +60,7 @@ Options:
   --max-tasks INTEGER             Number of concurrent tasks this agent can run
   --wait-time INTEGER             Waiting time between two polls on the server
                                   (seconds)
-  --log-level [debug|info|warning|error|critical]
+  --log-level [DEBUG|INFO|WARNING|ERROR|CRITICAL]
   --user-agent TEXT               A custom string to append to the User-Agent
                                   header
   --help                          Show this message and exit.

docs/installation/tl-dr.md

@@ -90,7 +90,7 @@ User=argos
 WorkingDirectory=/opt/argos/
 EnvironmentFile=/etc/default/argos-server
 ExecStartPre=/opt/argos/venv/bin/argos server migrate
-ExecStartPre=/opt/argos/venv/bin/argos server reload-config --enqueue
+ExecStartPre=/opt/argos/venv/bin/argos server reload-config
 ExecStart=/opt/argos/venv/bin/gunicorn "argos.server.main:get_application()" \\
                                        --workers \$ARGOS_SERVER_WORKERS \\
                                        --worker-class uvicorn.workers.UvicornWorker \\

pyproject.toml

@@ -33,7 +33,6 @@ dependencies = [
   "Jinja2>=3.0,<4",
   "jsonpointer>=3.0,<4",
   "passlib>=1.7.4,<2",
-  "psutil>=5.9.8,<6",
   "psycopg2-binary>=2.9,<3",
   "pydantic[email]>=2.4,<3",
   "pydantic-settings>=2.0,<3",

tests/config.yaml

@@ -43,11 +43,9 @@ ssl:
 # Argos will execute some tasks in the background for you
 # every 2 minutes and needs some configuration for that
 recurring_tasks:
-  # Maximum age of results
-  # Use m for minutes, h for hours, d for days
-  # w for weeks, M for months, y for years
-  # See https://github.com/timwedde/durations_nlp#scales-reference for details
-  max_results_age: "1d"
+  # Max number of results per tasks you want to keep
+  # Minimum value is 1, default is 100
+  max_results: 100
   # Max number of seconds a task can be locked
   # Minimum value is 61, default is 100
   max_lock_seconds: 100

tests/test_queries.py

@@ -10,9 +10,9 @@ from argos.server.models import Result, Task, User
 @pytest.mark.asyncio
 async def test_remove_old_results(db, ten_tasks):  #  pylint: disable-msg=redefined-outer-name
     for _task in ten_tasks:
-        for iterator in range(5):
+        for _ in range(5):
             result = Result(
-                submitted_at=datetime.now() - timedelta(seconds=iterator * 2),
+                submitted_at=datetime.now(),
                 status="success",
                 context={"foo": "bar"},
                 task=_task,
@@ -24,12 +24,12 @@ async def test_remove_old_results(db, ten_tasks):  #  pylint: disable-msg=redefi
 
     # So we have 5 results per tasks
     assert db.query(Result).count() == 50
-    # Keep only those newer than 1 second ago
-    deleted = await queries.remove_old_results(db, 6)
-    assert deleted == 20
-    assert db.query(Result).count() == 30
+    # Keep only 2
+    deleted = await queries.remove_old_results(db, 2)
+    assert deleted == 30
+    assert db.query(Result).count() == 20
     for _task in ten_tasks:
-        assert db.query(Result).filter(Result.task == _task).count() == 3
+        assert db.query(Result).filter(Result.task == _task).count() == 2
 
 
 @pytest.mark.asyncio
@@ -245,7 +245,7 @@ def empty_config():
         ),
         ssl=schemas.config.SSL(thresholds=[]),
         recurring_tasks=schemas.config.RecurringTasks(
-            max_results_age="6s",
+            max_results=100,
             max_lock_seconds=120,
             time_without_agent=300,
         ),