1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00

feat(operations): job, claim and worker lease models

This commit is contained in:
catborise 2026-10-13 17:05:38 +03:00
parent 5f8f7a8643
commit 99b9aa2a3c
9 changed files with 305 additions and 0 deletions

1
.gitignore vendored
View file

@ -8,6 +8,7 @@ venv2
.webvirtcloud
*.pyc
db.sqlite3*
test_db.sqlite3*
console/cert.pem*
tags
dhcpd.*

0
operations/__init__.py Normal file
View file

22
operations/apps.py Normal file
View file

@ -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")

View file

@ -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"
),
),
]

View file

76
operations/models.py Normal file
View file

@ -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()

View file

View file

@ -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")

View file

@ -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")},
}
}