425 lines
16 KiB
Python
425 lines
16 KiB
Python
"""
|
|
TaskService — creates tasks, dispatches them as asyncio background tasks,
|
|
and persists every state change to the DB so restarts don't lose history.
|
|
|
|
Flow:
|
|
create() → INSERT tasks (status=pending) → asyncio.create_task(run) → return task_id
|
|
run() → UPDATE running → per-device execute → UPDATE success/failed
|
|
"""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.models.rack import Rack
|
|
from app.models.task import Task, TaskEvent, TaskResult
|
|
from app.models.user import User
|
|
from app.plugins.registry import ActionRegistry, action_registry
|
|
from app.config import settings
|
|
from app.services.database import AsyncSessionLocal
|
|
from app.services.sse import SSEEvent, sse_manager
|
|
|
|
# Max parallel devices within a single task
|
|
_TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY
|
|
|
|
|
|
async def _quick_ping(ip: str, attempts: int = 4) -> bool:
|
|
"""Quick reachability check: ping -c 1 -W 1, up to `attempts` times."""
|
|
for _ in range(attempts):
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"ping", "-c", "1", "-W", "1", ip,
|
|
stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.DEVNULL,
|
|
)
|
|
await asyncio.wait_for(proc.wait(), timeout=3)
|
|
if proc.returncode == 0:
|
|
return True
|
|
except (asyncio.TimeoutError, OSError):
|
|
continue
|
|
return False
|
|
|
|
|
|
async def _emit(task_id: str, message: str, level: str = "info") -> None:
|
|
"""Write a progress line to DB and push to SSE subscribers."""
|
|
now = datetime.now(timezone.utc)
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
db.add(TaskEvent(task_id=task_id, message=message, level=level, created_at=now))
|
|
|
|
await sse_manager.publish(SSEEvent(
|
|
type="task_event",
|
|
task_id=task_id,
|
|
data={"message": message, "level": level, "ts": now.isoformat()},
|
|
))
|
|
|
|
|
|
async def _resolve_scope(scope: dict) -> list[tuple[Device, Object]]:
|
|
"""Expand target_scope to a list of (device, parent_object) pairs."""
|
|
async with AsyncSessionLocal() as db:
|
|
query = select(Device).options(
|
|
selectinload(Device.object).selectinload(Object.ssh_credentials),
|
|
selectinload(Device.object).selectinload(Object.category_credentials),
|
|
).where(Device.is_active == True)
|
|
|
|
scope_type = scope.get("type")
|
|
if scope_type == "device":
|
|
query = query.where(Device.id == scope["id"])
|
|
elif scope_type == "device_list":
|
|
query = query.where(Device.id.in_(scope["ids"]))
|
|
elif scope_type == "object":
|
|
query = query.where(Device.object_id == scope["id"])
|
|
elif scope_type == "rack":
|
|
query = query.where(Device.rack_id == scope["id"])
|
|
elif scope_type == "zone":
|
|
# Expand zone → rack IDs → devices
|
|
rack_rows = await db.execute(
|
|
select(Rack.id).where(Rack.zone_id == scope["id"])
|
|
)
|
|
rack_ids = [r[0] for r in rack_rows.all()]
|
|
if not rack_ids:
|
|
return []
|
|
query = query.where(Device.rack_id.in_(rack_ids))
|
|
elif scope_type == "category":
|
|
# support both "value" (legacy) and "category" key
|
|
cat = scope.get("category") or scope.get("value")
|
|
if cat:
|
|
query = query.where(Device.category == cat)
|
|
if "object_ids" in scope:
|
|
query = query.where(Device.object_id.in_(scope["object_ids"]))
|
|
elif "object_id" in scope:
|
|
query = query.where(Device.object_id == scope["object_id"])
|
|
elif scope_type == "role":
|
|
role = scope.get("role") or scope.get("value")
|
|
if role:
|
|
query = query.where(Device.role == role)
|
|
if "object_ids" in scope:
|
|
query = query.where(Device.object_id.in_(scope["object_ids"]))
|
|
elif "object_id" in scope:
|
|
query = query.where(Device.object_id == scope["object_id"])
|
|
elif scope_type == "filter":
|
|
# Cross-object filter scope (from WorkDevices global page)
|
|
if scope.get("category"):
|
|
cats = scope["category"]
|
|
if isinstance(cats, list):
|
|
query = query.where(Device.category.in_(cats))
|
|
else:
|
|
query = query.where(Device.category == cats)
|
|
if scope.get("object_id"):
|
|
query = query.where(Device.object_id == scope["object_id"])
|
|
if scope.get("city") or scope.get("client"):
|
|
query = query.join(Object, Device.object_id == Object.id)
|
|
if scope.get("city"):
|
|
query = query.where(Object.city == scope["city"])
|
|
if scope.get("client"):
|
|
query = query.where(Object.client == scope["client"])
|
|
else:
|
|
return []
|
|
|
|
result = await db.execute(query)
|
|
devices = result.scalars().all()
|
|
return [(d, d.object) for d in devices]
|
|
|
|
|
|
async def _run_on_device(
|
|
task_id: str,
|
|
device: Device,
|
|
obj: Object,
|
|
action,
|
|
params: dict,
|
|
sem: asyncio.Semaphore,
|
|
) -> bool | None:
|
|
async with sem:
|
|
# Session 1: cancellation check + create TaskResult
|
|
started_at = datetime.now(timezone.utc)
|
|
result_id: int | None = None
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task_row = await db.get(Task, task_id)
|
|
if task_row and task_row.cancel_requested:
|
|
return None # None = skipped due to cancellation
|
|
|
|
# Pre-ping reachability check (skip for ping action itself)
|
|
if action.name != "ping":
|
|
reachable = await _quick_ping(device.ip)
|
|
if not reachable:
|
|
await _emit(task_id, f"{device.ip}: unreachable, skipping", "warn")
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
tr = TaskResult(
|
|
task_id=task_id,
|
|
device_id=device.id,
|
|
status="skipped",
|
|
executed_at=datetime.now(timezone.utc),
|
|
duration_ms=0,
|
|
stderr="Device unreachable (ping pre-check failed)",
|
|
)
|
|
db.add(tr)
|
|
await sse_manager.publish(SSEEvent(
|
|
type="task_result",
|
|
task_id=task_id,
|
|
data={
|
|
"device_id": device.id,
|
|
"device_ip": device.ip,
|
|
"device_hostname": device.hostname,
|
|
"status": "skipped",
|
|
"duration_ms": 0,
|
|
"stdout": None,
|
|
"stderr": "Device unreachable (ping pre-check failed)",
|
|
"error": None,
|
|
},
|
|
))
|
|
return None
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
tr = TaskResult(
|
|
task_id=task_id,
|
|
device_id=device.id,
|
|
status="running",
|
|
executed_at=started_at,
|
|
)
|
|
db.add(tr)
|
|
await db.flush()
|
|
result_id = tr.id
|
|
|
|
emit = lambda msg, level="info": _emit(task_id, msg, level)
|
|
|
|
try:
|
|
action_result = await action.execute(device, obj, params, emit, db=None)
|
|
except Exception as exc:
|
|
action_result_success = False
|
|
action_stdout = ""
|
|
action_stderr = ""
|
|
action_exit_code = -1
|
|
action_data = {"error": str(exc)}
|
|
action_error = str(exc)
|
|
else:
|
|
action_result_success = action_result.success
|
|
action_stdout = action_result.stdout
|
|
action_stderr = action_result.stderr
|
|
action_exit_code = action_result.exit_code
|
|
action_data = action_result.data
|
|
action_error = action_result.error
|
|
|
|
finished_at = datetime.now(timezone.utc)
|
|
duration_ms = int((finished_at - started_at).total_seconds() * 1000)
|
|
|
|
# Session 2: update TaskResult + device status (ping) in one transaction
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
tr = await db.get(TaskResult, result_id)
|
|
if tr:
|
|
tr.status = "success" if action_result_success else "failed"
|
|
tr.stdout = action_stdout
|
|
tr.stderr = action_stderr
|
|
tr.exit_code = action_exit_code
|
|
tr.data = {**(action_data or {}), "error": action_error} if action_error else (action_data or None)
|
|
tr.duration_ms = duration_ms
|
|
|
|
if action.name == "ping":
|
|
new_status = "online" if action_result_success else "offline"
|
|
device_row = await db.get(Device, device.id)
|
|
if device_row:
|
|
device_row.status = new_status
|
|
device_row.last_ping_at = finished_at
|
|
if action_result_success:
|
|
device_row.last_seen = finished_at
|
|
ping_ms = (action_data or {}).get("ping_ms")
|
|
if ping_ms is not None:
|
|
device_row.last_ping_ms = ping_ms
|
|
|
|
# For ping action: publish SSE device_status after DB commit
|
|
if action.name == "ping":
|
|
new_status = "online" if action_result_success else "offline"
|
|
await sse_manager.publish(SSEEvent(
|
|
type="device_status",
|
|
task_id="__monitor__",
|
|
data={
|
|
"device_id": device.id,
|
|
"object_id": device.object_id,
|
|
"ip": device.ip,
|
|
"status": new_status,
|
|
"ping_ms": (action_data or {}).get("ping_ms") if action_result_success else None,
|
|
"ts": finished_at.isoformat(),
|
|
},
|
|
))
|
|
|
|
await sse_manager.publish(SSEEvent(
|
|
type="task_result",
|
|
task_id=task_id,
|
|
data={
|
|
"device_id": device.id,
|
|
"device_ip": device.ip,
|
|
"device_hostname": device.hostname,
|
|
"status": "success" if action_result_success else "failed",
|
|
"duration_ms": duration_ms,
|
|
"stdout": action_stdout or None,
|
|
"stderr": action_stderr or None,
|
|
"error": action_error or None,
|
|
},
|
|
))
|
|
|
|
return action_result_success
|
|
|
|
|
|
async def _heartbeat_writer(task_id: str, interval: int = 30) -> None:
|
|
"""Write task.heartbeat_at every `interval` seconds while task is running."""
|
|
while True:
|
|
await asyncio.sleep(interval)
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = await db.get(Task, task_id)
|
|
if task and task.status == "running":
|
|
task.heartbeat_at = datetime.now(timezone.utc)
|
|
|
|
|
|
async def _run_task(task_id: str) -> None:
|
|
await _emit(task_id, "Task started", "info")
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = await db.get(Task, task_id)
|
|
if task is None:
|
|
return
|
|
task.status = "running"
|
|
task.started_at = datetime.now(timezone.utc)
|
|
task.heartbeat_at = datetime.now(timezone.utc)
|
|
scope = dict(task.target_scope)
|
|
action_type = task.type
|
|
params = dict(task.params)
|
|
|
|
action = action_registry.get(action_type)
|
|
if action is None:
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = await db.get(Task, task_id)
|
|
if task:
|
|
task.status = "failed"
|
|
task.error_message = f"Unknown action: {action_type}"
|
|
task.finished_at = datetime.now(timezone.utc)
|
|
await _emit(task_id, f"Unknown action: {action_type}", "error")
|
|
return
|
|
|
|
targets = await _resolve_scope(scope)
|
|
|
|
# Filter devices by action compatibility (compatible_categories)
|
|
if action.compatible_categories:
|
|
compatible = [(d, o) for d, o in targets if action.is_compatible(d.category)]
|
|
skipped = [(d, o) for d, o in targets if not action.is_compatible(d.category)]
|
|
if skipped:
|
|
skipped_info = ", ".join(
|
|
f"{d.ip} ({d.category})" for d, _ in skipped[:10]
|
|
)
|
|
suffix = f" и ещё {len(skipped) - 10}" if len(skipped) > 10 else ""
|
|
await _emit(
|
|
task_id,
|
|
f"Пропущено {len(skipped)} несовместимых устройств: {skipped_info}{suffix}",
|
|
"warn",
|
|
)
|
|
targets = compatible
|
|
|
|
if not targets:
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = await db.get(Task, task_id)
|
|
if task:
|
|
task.status = "success"
|
|
task.result = {"total": 0, "ok": 0, "failed": 0}
|
|
task.finished_at = datetime.now(timezone.utc)
|
|
await _emit(task_id, "No devices matched the scope", "warn")
|
|
return
|
|
|
|
await _emit(task_id, f"Running on {len(targets)} device(s)...", "info")
|
|
await sse_manager.publish(SSEEvent(
|
|
type="task_meta",
|
|
task_id=task_id,
|
|
data={"total": len(targets)},
|
|
))
|
|
sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY)
|
|
|
|
# Start heartbeat writer
|
|
heartbeat_task = asyncio.create_task(_heartbeat_writer(task_id))
|
|
|
|
try:
|
|
results = await asyncio.gather(
|
|
*[_run_on_device(task_id, device, obj, action, params, sem) for device, obj in targets],
|
|
return_exceptions=True,
|
|
)
|
|
finally:
|
|
heartbeat_task.cancel()
|
|
try:
|
|
await heartbeat_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
ok = sum(1 for r in results if r is True)
|
|
skipped = sum(1 for r in results if r is None)
|
|
failed = len(results) - ok - skipped
|
|
ran = len(results) - skipped
|
|
|
|
# Check if cancellation was requested
|
|
was_cancelled = False
|
|
async with AsyncSessionLocal() as db:
|
|
task_row = await db.get(Task, task_id)
|
|
was_cancelled = bool(task_row and task_row.cancel_requested)
|
|
|
|
if was_cancelled:
|
|
final_status = "cancelled"
|
|
else:
|
|
final_status = "success" if failed == 0 else ("failed" if ok == 0 else "partial")
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = await db.get(Task, task_id)
|
|
if task:
|
|
task.status = final_status
|
|
task.result = {"total": ran, "ok": ok, "failed": failed}
|
|
task.finished_at = datetime.now(timezone.utc)
|
|
|
|
summary = f"Done: {ok}/{ran} succeeded" + (f" ({skipped} skipped)" if skipped else "")
|
|
await _emit(task_id, summary, "info")
|
|
await sse_manager.publish(SSEEvent(
|
|
type="task_status",
|
|
task_id=task_id,
|
|
data={"status": final_status, "total": ran, "ok": ok, "failed": failed},
|
|
))
|
|
|
|
|
|
async def create_task(
|
|
action_type: str,
|
|
target_scope: dict,
|
|
params: dict,
|
|
user_id: int | None = None,
|
|
) -> Task:
|
|
if action_registry.get(action_type) is None:
|
|
raise ValueError(f"Unknown action: {action_type}")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
task_id = str(uuid4())
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
task = Task(
|
|
id=task_id,
|
|
type=action_type,
|
|
status="pending",
|
|
target_scope=target_scope,
|
|
params=params,
|
|
created_by=user_id,
|
|
created_at=now,
|
|
)
|
|
db.add(task)
|
|
|
|
# Dispatch — fire and forget; state is already persisted
|
|
asyncio.create_task(_run_task(task_id))
|
|
|
|
# Return a fresh copy to the caller
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.get(Task, task_id)
|
|
return result
|