j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00
Code Issues Projects Releases Wiki Activity

fix(logs): a job's rows carry the address it was asked from, and a cancel its own user

Browse source 
A job keeps the client address of its submission (Operation.client_ip,
migration operations.0003); the rows the worker writes for it, including the
migration's and the VM delete's own rows, carry it. A cancel is written under
the user who asked and from their address, also when an administrator
cancels another user's job; a running job's cancel gets a row when asked,
and its end a row when it stops. A resolution carries the administrator's
address.
This commit is contained in:
catborise 2026-11-07 14:25:18 +03:00
parent 528b4492f6
commit c5c6de6ba8
9 changed files with 85 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

10
instances/jobs.py
View file

@ -226,7 +226,7 @@ def _finish_clone(op, clone, warning):
except libvirtError as err:
notes += say(_("The clone is made, but not as asked: %(error)s"), error=err)
write(op.username, c.name, op.intent["name"], f"VM {op.intent['name']}: resize or start the clone: "
f"{safe_text(err)}", False, uuid=op.intent["uuid"])
f"{safe_text(err)}", False, op.client_ip, op.intent["uuid"])
return {"warning": notes} if notes else {}
@ -972,7 +972,7 @@ def _moved(op, dest):
target = Compute.objects.get(pk=op.params["target"]).name
def failed(action, err): # as migrate_instance logs its later steps
write(op.username, source, name, f"VM {name}: {action}: {safe_text(err)}", False, uuid=op.intent["uuid"])
write(op.username, source, name, f"VM {name}: {action}: {safe_text(err)}", False, op.client_ip, op.intent["uuid"])
if op.intent.get("autostart"):
try:
@ -1043,7 +1043,7 @@ class MigrateInstance(Kind):
ctx.set_phase(gettext_noop("Migrating the VM"))
try:
utils.migrate_instance(Compute.objects.get(pk=p["target"]), instance, op.user,
disk_pools=p["disk_pools"],
disk_pools=p["disk_pools"], ip=op.client_ip,
on_moved=lambda: ctx.record_intent(outcome_logged=True),
**{k: p[k] for k in MIGRATE_OPTIONS})
except Exception as err:
@ -1276,7 +1276,7 @@ def _removed(op, recovering=False):
continue
notes += say(DISK_NOT_DELETED, path=path, err=err)
write(op.username, compute.name, op.intent["name"], f"VM {op.intent['name']}: delete volume {path}: "
f"{safe_text(err)}", False, uuid=op.intent["uuid"])
f"{safe_text(err)}", False, op.client_ip, op.intent["uuid"])
try:
while (path := conn.backing_path(path)) and path not in kept:
kept.add(path)
@ -1284,7 +1284,7 @@ def _removed(op, recovering=False):
kept.update(op.intent["to_delete"]) # the files it needs are unknown: none of the rest goes
else:
write(op.username, compute.name, op.intent["name"], f"VM {op.intent['name']}: delete volume {path}",
True, uuid=op.intent["uuid"])
True, op.client_ip, op.intent["uuid"])
for path in op.intent["shared"]:
notes += say(DISK_SHARED, path=path)
return {"open": reverse("instances:index"), **({"warning": notes} if notes else {})}

3
instances/tests/test_clone_job.py
View file

@ -152,6 +152,9 @@ class CloneJobTestCase(TransactionTestCase):
# the job succeeded: Logs has the step that failed too
self.assertTrue(Logs.objects.filter(instance_uuid=op.intent["uuid"], success=False,
message="VM copy: resize or start the clone: no memory").exists())
# the worker writes them with the address the form came from
self.assertEqual(set(Logs.objects.filter(instance_uuid=op.intent["uuid"]).values_list("ip", flat=True)),
{"127.0.0.1"})
def queued(self):
self.client.post(reverse("instances:clone", args=[self.source.pk]),

3
instances/utils.py
View file

@ -214,6 +214,7 @@ def migrate_instance(
compress=False,
delete_source=False,
disk_pools=None,
ip=None,
on_moved=None,
):
"""Migrates instance to new_compute and logs the outcome (the mode
@ -226,7 +227,7 @@ def migrate_instance(
def log(action, error=None):
message = f"VM {instance.name}: {action}" + (f": {reason(error, user)}" if error is not None else "")
write(user.username, source.name, instance.name, message, error is None, uuid=instance.uuid)
write(user.username, source.name, instance.name, message, error is None, ip, instance.uuid)
conn_migrate = None
try:

25
operations/jobs.py
View file

@ -17,6 +17,7 @@ from django.utils.translation import gettext, gettext_noop
from logs.audit import write
from vrtManager.util import OperationError
from webvirtcloud.log_context import request_ip
from . import claims, utils
from .models import Claim, LeaseLost, Operation, fenced
@ -108,7 +109,8 @@ class Kind:
def log_entry(self, op, state):
"""(host, instance, instance UUID, message) of the Logs row written
with each end of the job (in the same transaction); its host, instance
with each end of the job (in the same transaction) and with a cancel
asked while it runs (state CANCELLING); its host, instance
and UUID also go on the row of a resolution. None: no row."""
return None
@ -174,7 +176,7 @@ def submit(kind_name, user, raw_params, submission_key):
op = Operation(
kind=kind.name, user=user, username=user.username, submission_key=submission_key,
resource=plan.resource, resource_label=plan.label[:255], params=params, intent=plan.intent,
time_limit=kind.time_limit,
time_limit=kind.time_limit, client_ip=_client_ip(),
)
if kind.runs_in_request:
now = timezone.now()
@ -197,11 +199,14 @@ def cancel(op_id, user):
Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLED, finished_at=timezone.now())
Claim.objects.filter(operation=op).delete()
if (entry := _entry(op, Operation.CANCELLED)) is not None:
_log(op.username, *entry, None)
_log(user.username, *entry, None, _client_ip()) # an administrator may cancel another's job
elif op.state == Operation.RUNNING:
if not get_kind(op.kind).cancellable:
raise OperationError(gettext_noop("This job cannot be cancelled once it has started"))
Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLING, cancel_requested=True)
# who asked; the job's end row comes when it stops
if (entry := _entry(op, Operation.CANCELLING)) is not None:
_log(user.username, *entry, None, _client_ip())
elif op.state != Operation.CANCELLING:
raise OperationError(gettext_noop("The job has already ended"))
op.refresh_from_db()
@ -220,8 +225,13 @@ def _entry(op, state):
return None
def _log(user, host, instance, uuid, message, success):
write(str(user), str(host), str(instance), utils.safe_text(message), success, uuid=str(uuid))
def _log(user, host, instance, uuid, message, success, ip=None):
write(str(user), str(host), str(instance), utils.safe_text(message), success, ip, str(uuid))
def _client_ip():
"""The address of the web request this runs in; None in the worker."""
return request_ip.get()
def finish(op, outcome, *, worker=None):
@ -241,7 +251,8 @@ def finish(op, outcome, *, worker=None):
if outcome.state != Operation.UNCERTAIN:
Claim.objects.filter(operation=op).delete()
if (entry := _entry(op, outcome.state)) is not None:
_log(op.username, *entry, {Operation.SUCCEEDED: True, Operation.FAILED: False}.get(outcome.state))
_log(op.username, *entry, {Operation.SUCCEEDED: True, Operation.FAILED: False}.get(outcome.state),
op.client_ip)
logger.info("job %s %s: %s", op.pk, op.kind, outcome.state)
op.refresh_from_db()
return True
@ -267,7 +278,7 @@ def resolve(op_id, user, outcome_state, note):
entry = _entry(op, outcome_state)
host, instance, uuid = entry[:3] if entry is not None else ("-", op.resource_label, op.intent.get("uuid", ""))
_log(user.username, host, instance, uuid, f"Job {op.pk} resolved as {outcome_state}: {note}",
outcome_state == Operation.SUCCEEDED)
outcome_state == Operation.SUCCEEDED, _client_ip())
op.refresh_from_db()
return op

18
operations/migrations/0003_client_ip.py Normal file
View file

@ -0,0 +1,18 @@
# Generated by Django 5.2.18 on 2026-11-07 11:23
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('operations', '0002_error_messages'),
]
operations = [
migrations.AddField(
model_name='operation',
name='client_ip',
field=models.GenericIPAddressField(null=True),
),
]

1
operations/models.py
View file

@ -32,6 +32,7 @@ class Operation(models.Model):
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
client_ip = models.GenericIPAddressField(null=True) # of the submission, for the job's Logs rows
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)

29
operations/tests/test_service.py
View file

@ -14,6 +14,7 @@ from operations import claims, jobs
from operations.jobs import Outcome
from operations.models import Claim, LeaseLost, Operation, WorkerLease, acquire
from operations.tests import fakes
from webvirtcloud.log_context import request_ip
KEY = "key-00000001"
@ -112,6 +113,34 @@ class CancelFinishResolveTestCase(TransactionTestCase):
jobs.resolve(op.pk, self.admin, Operation.FAILED, "checked")
self.assertIs(Logs.objects.latest("pk").success, False)
def test_a_job_s_rows_name_who_did_each_step_and_from_where(self):
def from_ip(ip): # as the middleware sets it while it handles a request
token = request_ip.set(ip)
self.addCleanup(request_ip.reset, token)
def last_row():
row = Logs.objects.latest("pk")
return row.user, row.ip, row.message
from_ip("192.0.2.7")
op = jobs.submit("test.noop", self.user, {}, KEY)
queued = jobs.submit("test.noop", self.user, {"uuid": "u2"}, "key-00000002")
running = jobs.submit("test.scripted", self.user, {"uuid": "u3"}, "key-00000003")
Operation.objects.filter(pk=running.pk).update(state=Operation.RUNNING)
request_ip.set(None) # the worker
jobs.finish(op, Outcome(Operation.SUCCEEDED))
self.assertEqual(last_row(), ("svc2", "192.0.2.7", "No-op succeeded"))
from_ip("198.51.100.9") # an administrator cancels the user's jobs
jobs.cancel(queued.pk, self.admin)
self.assertEqual(last_row(), ("svc_admin", "198.51.100.9", "No-op cancelled"))
jobs.cancel(running.pk, self.admin)
self.assertEqual(last_row(), ("svc_admin", "198.51.100.9", "No-op cancelling"))
def test_a_submission_after_a_request_does_not_take_its_address(self):
# the middleware keeps the request for Django's log line, not its address
self.client.get("/accounts/login/")
self.assertIsNone(jobs.submit("test.noop", self.user, {}, KEY).client_ip)
def test_cancel_decides_from_the_current_state(self):
op = jobs.submit("test.scripted", self.user, {}, KEY)
Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING) # the worker took it meanwhile

3
webvirtcloud/log_context.py
View file

@ -5,6 +5,9 @@ import contextvars
import logging
current_request = contextvars.ContextVar("current_request", default=None)
# the client address while a request is handled, reset when it is answered
# (current_request stays for Django's log line): the Logs rows of the jobs it submits
request_ip = contextvars.ContextVar("request_ip", default=None)
class RequestFilter(logging.Filter):

7
webvirtcloud/middleware.py
View file

@ -13,6 +13,7 @@ from operations import claims, jobs
from operations.utils import busy_message, public_error_text, said, summary, words
from vrtManager.util import ConnectionFailed, InUse
from webvirtcloud import log_context
from webvirtcloud.log_context import current_request
API_PREFIX = "/api/"
@ -83,7 +84,11 @@ class RequestIdMiddleware:
request.client_ip = request_ip(request)
# not reset after the response: Django logs a 4xx or 5xx answer after the middleware returns
current_request.set(request)
response = self.get_response(request)
token = log_context.request_ip.set(request.client_ip)
try:
response = self.get_response(request)
finally:
log_context.request_ip.reset(token)
response["X-Request-ID"] = request.request_id
return response