From 99b9aa2a3c24064be3bba786d910e06967f411c8 Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 13 Oct 2026 17:05:38 +0300 Subject: [PATCH] feat(operations): job, claim and worker lease models --- .gitignore | 1 + operations/__init__.py | 0 operations/apps.py | 22 ++++ operations/migrations/0001_initial.py | 150 ++++++++++++++++++++++++++ operations/migrations/__init__.py | 0 operations/models.py | 76 +++++++++++++ operations/tests/__init__.py | 0 operations/tests/test_models.py | 47 ++++++++ webvirtcloud/settings.py.template | 9 ++ 9 files changed, 305 insertions(+) create mode 100644 operations/__init__.py create mode 100644 operations/apps.py create mode 100644 operations/migrations/0001_initial.py create mode 100644 operations/migrations/__init__.py create mode 100644 operations/models.py create mode 100644 operations/tests/__init__.py create mode 100644 operations/tests/test_models.py diff --git a/.gitignore b/.gitignore index 873a8083..2ffa41bd 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,7 @@ venv2 .webvirtcloud *.pyc db.sqlite3* +test_db.sqlite3* console/cert.pem* tags dhcpd.* diff --git a/operations/__init__.py b/operations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/apps.py b/operations/apps.py new file mode 100644 index 00000000..c48a8f9a --- /dev/null +++ b/operations/apps.py @@ -0,0 +1,22 @@ +from django.apps import AppConfig +from django.db.backends.signals import connection_created + + +def _sqlite_wal(sender, connection, **kwargs): + """The worker and gunicorn write the same SQLite file: WAL lets readers + go on during a write. Django's SQLite OPTIONS have no init command.""" + if connection.vendor == "sqlite": + with connection.cursor() as cursor: + cursor.execute("PRAGMA journal_mode=WAL") + + +class OperationsConfig(AppConfig): + default_auto_field = "django.db.models.BigAutoField" + name = "operations" + + def ready(self): + connection_created.connect(_sqlite_wal) + # every installed app's jobs.py registers its kinds + from django.utils.module_loading import autodiscover_modules + + autodiscover_modules("jobs") diff --git a/operations/migrations/0001_initial.py b/operations/migrations/0001_initial.py new file mode 100644 index 00000000..bc53b7cb --- /dev/null +++ b/operations/migrations/0001_initial.py @@ -0,0 +1,150 @@ +# Generated by Django 5.2.18 on 2026-10-13 13:58 + +import django.db.models.deletion +import uuid +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="WorkerLease", + fields=[ + ( + "name", + models.CharField( + default="worker", + max_length=16, + primary_key=True, + serialize=False, + ), + ), + ("holder", models.CharField(max_length=128)), + ("host", models.CharField(max_length=255)), + ("heartbeat_at", models.DateTimeField()), + ], + ), + migrations.CreateModel( + name="Operation", + fields=[ + ( + "id", + models.UUIDField( + default=uuid.uuid4, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ("kind", models.CharField(db_index=True, max_length=64)), + ("username", models.CharField(max_length=150)), + ("submission_key", models.CharField(max_length=64)), + ("resource", models.CharField(db_index=True, max_length=255)), + ("resource_label", models.CharField(max_length=255)), + ( + "state", + models.CharField( + choices=[ + ("queued", "queued"), + ("running", "running"), + ("cancelling", "cancelling"), + ("succeeded", "succeeded"), + ("failed", "failed"), + ("cancelled", "cancelled"), + ("uncertain", "uncertain"), + ], + db_index=True, + default="queued", + max_length=16, + ), + ), + ("params", models.JSONField(default=dict)), + ("intent", models.JSONField(default=dict)), + ("effects_started", models.BooleanField(default=False)), + ("phase", models.CharField(blank=True, max_length=255)), + ("progress_done", models.BigIntegerField(null=True)), + ("progress_total", models.BigIntegerField(null=True)), + ("result", models.JSONField(default=dict)), + ("error", models.TextField(blank=True)), + ("created_at", models.DateTimeField(auto_now_add=True, db_index=True)), + ("started_at", models.DateTimeField(null=True)), + ("finished_at", models.DateTimeField(null=True)), + ("heartbeat_at", models.DateTimeField(null=True)), + ("worker", models.CharField(blank=True, max_length=128)), + ("time_limit", models.PositiveIntegerField(null=True)), + ("cancel_requested", models.BooleanField(default=False)), + ("resolved_at", models.DateTimeField(null=True)), + ("resolution", models.CharField(blank=True, max_length=16)), + ("resolution_note", models.TextField(blank=True)), + ( + "resolved_by", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to=settings.AUTH_USER_MODEL, + ), + ), + ( + "user", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="operations", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "ordering": ["-created_at"], + }, + ), + migrations.CreateModel( + name="Claim", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("resource", models.CharField(db_index=True, max_length=255)), + ( + "mode", + models.CharField( + choices=[("shared", "shared"), ("exclusive", "exclusive")], + max_length=9, + ), + ), + ("holder", models.CharField(blank=True, max_length=128)), + ("expires_at", models.DateTimeField(null=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ( + "operation", + models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.CASCADE, + related_name="claims", + to="operations.operation", + ), + ), + ], + ), + migrations.AddConstraint( + model_name="operation", + constraint=models.UniqueConstraint( + fields=("user", "submission_key"), name="unique_submission" + ), + ), + ] diff --git a/operations/migrations/__init__.py b/operations/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/models.py b/operations/models.py new file mode 100644 index 00000000..e27b337b --- /dev/null +++ b/operations/models.py @@ -0,0 +1,76 @@ +"""Long-running operations: the queue, the history and the resources jobs +hold (spec 2026-10-13, sections 1-2).""" + +import uuid + +from django.conf import settings +from django.db import models + + +class Operation(models.Model): + QUEUED = "queued" + RUNNING = "running" + CANCELLING = "cancelling" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + UNCERTAIN = "uncertain" + ACTIVE = (QUEUED, RUNNING, CANCELLING) + TERMINAL = (SUCCEEDED, FAILED, CANCELLED, UNCERTAIN) + STATES = [(s, s) for s in ACTIVE + TERMINAL] + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + kind = models.CharField(max_length=64, db_index=True) + user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL, related_name="operations") + username = models.CharField(max_length=150) # kept when the user is deleted + submission_key = models.CharField(max_length=64) + resource = models.CharField(max_length=255, db_index=True) # the page that shows the job + resource_label = models.CharField(max_length=255) + state = models.CharField(max_length=16, choices=STATES, default=QUEUED, db_index=True) + params = models.JSONField(default=dict) + intent = models.JSONField(default=dict) # plan facts, then what run is about to change + # set with the intent committed right before the first libvirt effect: + # recovery fails a job without it, it never reached libvirt + effects_started = models.BooleanField(default=False) + phase = models.CharField(max_length=255, blank=True) + progress_done = models.BigIntegerField(null=True) + progress_total = models.BigIntegerField(null=True) + result = models.JSONField(default=dict) + error = models.TextField(blank=True) + created_at = models.DateTimeField(auto_now_add=True, db_index=True) + started_at = models.DateTimeField(null=True) + finished_at = models.DateTimeField(null=True) + heartbeat_at = models.DateTimeField(null=True) + worker = models.CharField(max_length=128, blank=True) # identity of the worker that owns it + time_limit = models.PositiveIntegerField(null=True) # seconds, fixed at admission + cancel_requested = models.BooleanField(default=False) + resolved_by = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL, related_name="+") + resolved_at = models.DateTimeField(null=True) + resolution = models.CharField(max_length=16, blank=True) + resolution_note = models.TextField(blank=True) + + class Meta: + ordering = ["-created_at"] + constraints = [models.UniqueConstraint(fields=["user", "submission_key"], name="unique_submission")] + + def __str__(self): + return f"{self.kind} {self.id}" + + +class Claim(models.Model): + SHARED = "shared" + EXCLUSIVE = "exclusive" + + operation = models.ForeignKey(Operation, null=True, on_delete=models.CASCADE, related_name="claims") + resource = models.CharField(max_length=255, db_index=True) + mode = models.CharField(max_length=9, choices=[(SHARED, SHARED), (EXCLUSIVE, EXCLUSIVE)]) + holder = models.CharField(max_length=128, blank=True) # transient: host:pid:thread + expires_at = models.DateTimeField(null=True) # transient only + created_at = models.DateTimeField(auto_now_add=True) + + +class WorkerLease(models.Model): + name = models.CharField(primary_key=True, max_length=16, default="worker") + holder = models.CharField(max_length=128) + host = models.CharField(max_length=255) + heartbeat_at = models.DateTimeField() diff --git a/operations/tests/__init__.py b/operations/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/tests/test_models.py b/operations/tests/test_models.py new file mode 100644 index 00000000..d5817537 --- /dev/null +++ b/operations/tests/test_models.py @@ -0,0 +1,47 @@ +"""The job record: states, defaults, the per-user submission key, and the +SQLite connection mode every process relies on.""" + +from django.contrib.auth import get_user_model +from django.db import IntegrityError, connection, transaction +from django.test import TestCase + +from operations.models import Claim, Operation + + +def new_op(user, key="k", **kw): + return Operation.objects.create(kind="test.noop", user=user, username=user.username, submission_key=key, + resource="instance:u1", resource_label="c/vm", **kw) + + +class OperationModelTestCase(TestCase): + def setUp(self): + self.user = get_user_model().objects.create_user("jobs_user", "j@example.com", "pw") + + def test_a_new_operation_is_queued_with_empty_records(self): + op = new_op(self.user) + op.refresh_from_db() + self.assertEqual(op.state, Operation.QUEUED) + self.assertEqual((op.params, op.intent, op.result), ({}, {}, {})) + self.assertIn(Operation.QUEUED, Operation.ACTIVE) + self.assertIn(Operation.UNCERTAIN, Operation.TERMINAL) + + def test_a_submission_key_is_unique_per_user(self): + new_op(self.user) + with self.assertRaises(IntegrityError), transaction.atomic(): + new_op(self.user) + + def test_claims_go_with_their_operation(self): + op = new_op(self.user) + Claim.objects.create(operation=op, resource="instance:u1", mode=Claim.EXCLUSIVE) + op.delete() + self.assertFalse(Claim.objects.exists()) + + +class SqliteModeTestCase(TestCase): + def test_sqlite_runs_in_wal_with_immediate_transactions(self): + if connection.vendor != "sqlite": + self.skipTest("SQLite only") + with connection.cursor() as cursor: + cursor.execute("PRAGMA journal_mode") + self.assertEqual(cursor.fetchone()[0].lower(), "wal") + self.assertEqual(connection.transaction_mode, "IMMEDIATE") diff --git a/webvirtcloud/settings.py.template b/webvirtcloud/settings.py.template index 6c39cf97..13d3648f 100644 --- a/webvirtcloud/settings.py.template +++ b/webvirtcloud/settings.py.template @@ -79,6 +79,7 @@ INSTALLED_APPS = [ "storages", "virtsecrets", "logs", + "operations", "rest_framework", "axes", ] @@ -151,6 +152,14 @@ else: "default": { "ENGINE": "django.db.backends.sqlite3", "NAME": Path.joinpath(BASE_DIR, "db.sqlite3"), + # IMMEDIATE takes the write lock when a transaction starts, so a + # check-then-write cannot interleave with another process; timeout + # is how long a writer waits for that lock (seconds). WAL is set + # per connection by the operations app. + "OPTIONS": {"transaction_mode": "IMMEDIATE", "timeout": 20}, + # A file, not shared-cache memory: test threads and worker processes + # open the same database, and shared-cache locks ignore the timeout. + "TEST": {"NAME": os.getenv("WEBVIRTCLOUD_TEST_DB") or Path.joinpath(BASE_DIR, "test_db.sqlite3")}, } }