1035 lines
40 KiB
Python
1035 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import uuid
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import timedelta
|
|
from typing import Any, Literal
|
|
|
|
from app.core.time import from_iso, to_iso, utc_now
|
|
from app.database.database import Database
|
|
|
|
ROTATION_LEASE_TTL_SECONDS = 300
|
|
|
|
RUN_COLUMNS = """
|
|
id, target_type, target_id, target_name, trigger, status, active_slot,
|
|
current_item_id, lease_owner, lease_until, error_code, error_message,
|
|
total_items, succeeded_items, started_at, updated_at, finished_at
|
|
"""
|
|
|
|
ITEM_COLUMNS = """
|
|
id, run_id, instance_id, position, status, stage, stage_started_at,
|
|
attempt_count, config_version, instance_display_name, aws_region,
|
|
lightsail_instance_name, cloudflare_zone_name, cloudflare_zone_id,
|
|
cloudflare_record_name, socks_port, proxy_health_check,
|
|
health_timeout_seconds, release_grace_seconds, old_static_ip_name, old_ip,
|
|
new_static_ip_name, new_ip, dns_ip_before, dns_ip_after, grace_until,
|
|
aws_operation_id, error_code, error_message, rollback_from_stage,
|
|
rollback_reason_code, rollback_reason_message, started_at, updated_at, finished_at
|
|
"""
|
|
|
|
|
|
class RotationLeaseLostError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FleetRun:
|
|
id: str
|
|
target_type: str
|
|
target_id: str
|
|
target_name: str
|
|
trigger: str
|
|
status: str
|
|
active_slot: int | None
|
|
current_item_id: str | None
|
|
lease_owner: str | None
|
|
lease_until: str | None
|
|
error_code: str | None
|
|
error_message: str | None
|
|
total_items: int
|
|
succeeded_items: int
|
|
started_at: str
|
|
updated_at: str
|
|
finished_at: str | None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FleetRunItem:
|
|
id: str
|
|
run_id: str
|
|
instance_id: str
|
|
position: int
|
|
status: str
|
|
stage: str
|
|
stage_started_at: str
|
|
attempt_count: int
|
|
config_version: int
|
|
instance_display_name: str
|
|
aws_region: str
|
|
lightsail_instance_name: str
|
|
cloudflare_zone_name: str
|
|
cloudflare_zone_id: str
|
|
cloudflare_record_name: str
|
|
socks_port: int
|
|
proxy_health_check: int
|
|
health_timeout_seconds: int
|
|
release_grace_seconds: int
|
|
old_static_ip_name: str | None
|
|
old_ip: str | None
|
|
new_static_ip_name: str | None
|
|
new_ip: str | None
|
|
dns_ip_before: str | None
|
|
dns_ip_after: str | None
|
|
grace_until: str | None
|
|
aws_operation_id: str | None
|
|
error_code: str | None
|
|
error_message: str | None
|
|
rollback_from_stage: str | None
|
|
rollback_reason_code: str | None
|
|
rollback_reason_message: str | None
|
|
started_at: str
|
|
updated_at: str
|
|
finished_at: str | None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["proxy_health_check"] = bool(self.proxy_health_check)
|
|
return data
|
|
|
|
|
|
class RotationRepository:
|
|
def __init__(self, database: Database) -> None:
|
|
self.database = database
|
|
|
|
def create_for_instance(
|
|
self,
|
|
instance_id: str,
|
|
trigger: Literal["manual", "scheduled", "recovery"] = "manual",
|
|
) -> FleetRun:
|
|
return self._create_run("instance", instance_id, trigger)
|
|
|
|
def create_for_group(
|
|
self,
|
|
group_id: str,
|
|
trigger: Literal["manual", "scheduled", "recovery"] = "manual",
|
|
) -> FleetRun:
|
|
return self._create_run("group", group_id, trigger)
|
|
|
|
def _create_run(self, target_type: str, target_id: str, trigger: str) -> FleetRun:
|
|
now = to_iso()
|
|
run_id = str(uuid.uuid4())
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
self._clear_expired_dns_lock(connection, now)
|
|
if connection.execute(
|
|
"""
|
|
SELECT 1 FROM fleet_operation_locks WHERE slot = 1
|
|
UNION ALL
|
|
SELECT 1 FROM fleet_runs WHERE active_slot = 1
|
|
LIMIT 1
|
|
"""
|
|
).fetchone():
|
|
raise RuntimeError("ACTIVE_ROTATION")
|
|
|
|
if target_type == "instance":
|
|
target = connection.execute(
|
|
"""
|
|
SELECT * FROM managed_instances
|
|
WHERE id = ? AND archived_at IS NULL AND enabled = 1
|
|
""",
|
|
(target_id,),
|
|
).fetchone()
|
|
if target is None:
|
|
raise RuntimeError("INSTANCE_NOT_FOUND")
|
|
target_name = str(target["display_name"])
|
|
instances = [target]
|
|
else:
|
|
group = connection.execute(
|
|
"""
|
|
SELECT * FROM instance_groups
|
|
WHERE id = ? AND archived_at IS NULL
|
|
""",
|
|
(target_id,),
|
|
).fetchone()
|
|
if group is None:
|
|
raise RuntimeError("GROUP_NOT_FOUND")
|
|
if trigger == "scheduled" and not bool(group["enabled"]):
|
|
raise RuntimeError("GROUP_SCHEDULE_DISABLED")
|
|
target_name = str(group["name"])
|
|
instances = connection.execute(
|
|
"""
|
|
SELECT instance.*
|
|
FROM instance_group_members AS member
|
|
JOIN managed_instances AS instance ON instance.id = member.instance_id
|
|
WHERE member.group_id = ? AND instance.archived_at IS NULL
|
|
AND instance.enabled = 1
|
|
ORDER BY member.position, instance.created_at
|
|
""",
|
|
(target_id,),
|
|
).fetchall()
|
|
if not instances:
|
|
raise RuntimeError("GROUP_HAS_NO_ENABLED_INSTANCES")
|
|
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_operation_locks(slot, kind, owner_id, lease_until, created_at)
|
|
VALUES (1, 'rotation', ?, NULL, ?)
|
|
""",
|
|
(run_id, now),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_runs(
|
|
id, target_type, target_id, target_name, trigger, status,
|
|
active_slot, total_items, succeeded_items, started_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, 'queued', 1, ?, 0, ?, ?)
|
|
""",
|
|
(
|
|
run_id,
|
|
target_type,
|
|
target_id,
|
|
target_name,
|
|
trigger,
|
|
len(instances),
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
first_item_id: str | None = None
|
|
for position, instance in enumerate(instances):
|
|
item_id = str(uuid.uuid4())
|
|
first_item_id = first_item_id or item_id
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_run_items(
|
|
id, run_id, instance_id, position, status, stage,
|
|
stage_started_at, config_version, instance_display_name,
|
|
aws_region, lightsail_instance_name, cloudflare_zone_name,
|
|
cloudflare_zone_id, cloudflare_record_name, socks_port,
|
|
proxy_health_check, health_timeout_seconds,
|
|
release_grace_seconds, started_at, updated_at
|
|
) VALUES (
|
|
?, ?, ?, ?, 'queued', 'preflight', ?, ?, ?, ?, ?, ?, ?, ?,
|
|
?, ?, ?, ?, ?, ?
|
|
)
|
|
""",
|
|
(
|
|
item_id,
|
|
run_id,
|
|
instance["id"],
|
|
position,
|
|
now,
|
|
instance["config_version"],
|
|
instance["display_name"],
|
|
instance["aws_region"],
|
|
instance["lightsail_instance_name"],
|
|
instance["cloudflare_zone_name"],
|
|
instance["cloudflare_zone_id"],
|
|
instance["cloudflare_record_name"],
|
|
instance["socks_port"],
|
|
instance["proxy_health_check"],
|
|
instance["health_timeout_seconds"],
|
|
instance["release_grace_seconds"],
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
connection.execute(
|
|
"UPDATE fleet_runs SET current_item_id = ? WHERE id = ?",
|
|
(first_item_id, run_id),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id, item_id, occurred_at, stage, level, message, details_json
|
|
) VALUES (?, ?, ?, 'queued', 'info', ?, ?)
|
|
""",
|
|
(
|
|
run_id,
|
|
first_item_id,
|
|
now,
|
|
"轮换批次已创建",
|
|
json.dumps({"target_type": target_type, "items": len(instances)}),
|
|
),
|
|
)
|
|
if target_type == "group":
|
|
connection.execute(
|
|
"""
|
|
UPDATE instance_groups SET next_run_at = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(None, now, target_id),
|
|
)
|
|
connection.commit()
|
|
except sqlite3.IntegrityError as exc:
|
|
connection.rollback()
|
|
raise RuntimeError("ACTIVE_ROTATION") from exc
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
run = self.get(run_id)
|
|
if run is None: # pragma: no cover
|
|
raise RuntimeError("RUN_NOT_FOUND")
|
|
return run
|
|
|
|
def get(self, run_id: str) -> FleetRun | None:
|
|
with self.database.connect() as connection:
|
|
row = connection.execute(
|
|
f"SELECT {RUN_COLUMNS} FROM fleet_runs WHERE id = ?", (run_id,)
|
|
).fetchone()
|
|
return FleetRun(**dict(row)) if row else None
|
|
|
|
def get_active(self) -> FleetRun | None:
|
|
with self.database.connect() as connection:
|
|
row = connection.execute(
|
|
f"SELECT {RUN_COLUMNS} FROM fleet_runs WHERE active_slot = 1"
|
|
).fetchone()
|
|
return FleetRun(**dict(row)) if row else None
|
|
|
|
def get_item(self, item_id: str) -> FleetRunItem | None:
|
|
with self.database.connect() as connection:
|
|
row = connection.execute(
|
|
f"SELECT {ITEM_COLUMNS} FROM fleet_run_items WHERE id = ?", (item_id,)
|
|
).fetchone()
|
|
return FleetRunItem(**dict(row)) if row else None
|
|
|
|
def current_item(self, run_id: str) -> FleetRunItem | None:
|
|
with self.database.connect() as connection:
|
|
row = connection.execute(
|
|
f"""
|
|
SELECT {', '.join(f'item.{part.strip()}' for part in ITEM_COLUMNS.split(','))}
|
|
FROM fleet_runs AS run
|
|
JOIN fleet_run_items AS item ON item.id = run.current_item_id
|
|
WHERE run.id = ?
|
|
""",
|
|
(run_id,),
|
|
).fetchone()
|
|
return FleetRunItem(**dict(row)) if row else None
|
|
|
|
def list_items(self, run_id: str) -> list[FleetRunItem]:
|
|
with self.database.connect() as connection:
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT {ITEM_COLUMNS} FROM fleet_run_items
|
|
WHERE run_id = ? ORDER BY position
|
|
""",
|
|
(run_id,),
|
|
).fetchall()
|
|
return [FleetRunItem(**dict(row)) for row in rows]
|
|
|
|
def list_events(self, run_id: str) -> list[dict[str, Any]]:
|
|
with self.database.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT id, item_id, occurred_at, stage, level, message, details_json
|
|
FROM fleet_events WHERE run_id = ? ORDER BY id
|
|
""",
|
|
(run_id,),
|
|
).fetchall()
|
|
return [
|
|
{
|
|
"id": int(row["id"]),
|
|
"item_id": row["item_id"],
|
|
"occurred_at": row["occurred_at"],
|
|
"stage": row["stage"],
|
|
"level": row["level"],
|
|
"message": row["message"],
|
|
"details": json.loads(row["details_json"]),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def list_runs(
|
|
self,
|
|
page: int,
|
|
per_page: int,
|
|
*,
|
|
instance_id: str | None = None,
|
|
group_id: str | None = None,
|
|
status: str | None = None,
|
|
) -> tuple[list[FleetRun], int]:
|
|
clauses: list[str] = []
|
|
values: list[Any] = []
|
|
if instance_id:
|
|
clauses.append(
|
|
"EXISTS (SELECT 1 FROM fleet_run_items item "
|
|
"WHERE item.run_id = run.id AND item.instance_id = ?)"
|
|
)
|
|
values.append(instance_id)
|
|
if group_id:
|
|
clauses.append("run.target_type = 'group' AND run.target_id = ?")
|
|
values.append(group_id)
|
|
if status:
|
|
clauses.append("run.status = ?")
|
|
values.append(status)
|
|
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
|
|
offset = (page - 1) * per_page
|
|
with self.database.connect() as connection:
|
|
total = int(
|
|
connection.execute(
|
|
f"SELECT COUNT(*) FROM fleet_runs AS run {where}", values
|
|
).fetchone()[0]
|
|
)
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT {', '.join(f'run.{part.strip()}' for part in RUN_COLUMNS.split(','))}
|
|
FROM fleet_runs AS run {where}
|
|
ORDER BY run.started_at DESC LIMIT ? OFFSET ?
|
|
""",
|
|
[*values, per_page, offset],
|
|
).fetchall()
|
|
return [FleetRun(**dict(row)) for row in rows], total
|
|
|
|
def history_summary(self) -> dict[str, int | str | None]:
|
|
with self.database.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT COUNT(*) AS total,
|
|
SUM(CASE WHEN status = 'succeeded' THEN 1 ELSE 0 END) AS succeeded,
|
|
SUM(CASE WHEN status IN ('succeeded','failed','cancelled') THEN 1 ELSE 0 END)
|
|
AS completed,
|
|
MAX(CASE WHEN status = 'succeeded' THEN finished_at END) AS latest_success_at
|
|
FROM fleet_runs
|
|
"""
|
|
).fetchone()
|
|
return {
|
|
"total": int(row["total"] or 0),
|
|
"completed": int(row["completed"] or 0),
|
|
"succeeded": int(row["succeeded"] or 0),
|
|
"latest_success_at": row["latest_success_at"],
|
|
}
|
|
|
|
def acquire_lease(
|
|
self,
|
|
run_id: str,
|
|
owner: str,
|
|
ttl_seconds: int = ROTATION_LEASE_TTL_SECONDS,
|
|
) -> bool:
|
|
now = utc_now()
|
|
now_iso = to_iso(now)
|
|
lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
|
|
with self.database.connect() as connection, connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET lease_owner = ?, lease_until = ?, updated_at = ?
|
|
WHERE id = ? AND active_slot = 1
|
|
AND status IN ('queued','running')
|
|
AND (lease_owner IS NULL OR lease_until < ? OR lease_owner = ?)
|
|
""",
|
|
(owner, lease_until, now_iso, run_id, now_iso, owner),
|
|
)
|
|
return cursor.rowcount == 1
|
|
|
|
def renew_lease(
|
|
self,
|
|
run_id: str,
|
|
owner: str,
|
|
ttl_seconds: int = ROTATION_LEASE_TTL_SECONDS,
|
|
) -> bool:
|
|
now = utc_now()
|
|
now_iso = to_iso(now)
|
|
lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
|
|
with self.database.connect() as connection, connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET lease_until = ?, updated_at = ?
|
|
WHERE id = ? AND active_slot = 1 AND lease_owner = ?
|
|
AND lease_until >= ? AND status IN ('queued','running')
|
|
""",
|
|
(lease_until, now_iso, run_id, owner, now_iso),
|
|
)
|
|
return cursor.rowcount == 1
|
|
|
|
def transition_run(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
status: str,
|
|
expected_owner: str,
|
|
) -> FleetRun:
|
|
now = to_iso()
|
|
with self.database.connect() as connection, connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = ?, updated_at = ?
|
|
WHERE id = ? AND active_slot = 1 AND lease_owner = ? AND lease_until >= ?
|
|
""",
|
|
(status, now, run_id, expected_owner, now),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
raise RotationLeaseLostError("ROTATION_LEASE_LOST")
|
|
run = self.get(run_id)
|
|
if run is None: # pragma: no cover
|
|
raise RuntimeError("RUN_NOT_FOUND")
|
|
return run
|
|
|
|
def transition_item(
|
|
self,
|
|
run_id: str,
|
|
item_id: str,
|
|
*,
|
|
expected_owner: str,
|
|
stage: str | None = None,
|
|
status: str | None = None,
|
|
**fields: Any,
|
|
) -> FleetRunItem:
|
|
allowed = {
|
|
"old_static_ip_name",
|
|
"old_ip",
|
|
"new_static_ip_name",
|
|
"new_ip",
|
|
"dns_ip_before",
|
|
"dns_ip_after",
|
|
"grace_until",
|
|
"aws_operation_id",
|
|
"error_code",
|
|
"error_message",
|
|
"cloudflare_zone_id",
|
|
"rollback_from_stage",
|
|
"rollback_reason_code",
|
|
"rollback_reason_message",
|
|
}
|
|
unknown = set(fields) - allowed
|
|
if unknown:
|
|
raise ValueError(f"Unsupported item fields: {sorted(unknown)}")
|
|
now = to_iso()
|
|
updates = dict(fields)
|
|
if stage is not None:
|
|
updates["stage"] = stage
|
|
updates["stage_started_at"] = now
|
|
if status is not None:
|
|
updates["status"] = status
|
|
updates["updated_at"] = now
|
|
assignments = ", ".join(f"{name} = ?" for name in updates)
|
|
values = [*updates.values(), item_id, run_id, expected_owner, now]
|
|
with self.database.connect() as connection, connection:
|
|
cursor = connection.execute(
|
|
f"""
|
|
UPDATE fleet_run_items SET {assignments}
|
|
WHERE id = ? AND run_id = ? AND EXISTS (
|
|
SELECT 1 FROM fleet_runs run WHERE run.id = fleet_run_items.run_id
|
|
AND run.active_slot = 1 AND run.lease_owner = ?
|
|
AND run.lease_until >= ? AND run.status IN ('queued','running')
|
|
)
|
|
""",
|
|
values,
|
|
)
|
|
if cursor.rowcount != 1:
|
|
raise RotationLeaseLostError("ROTATION_LEASE_LOST")
|
|
item = self.get_item(item_id)
|
|
if item is None: # pragma: no cover
|
|
raise RuntimeError("ITEM_NOT_FOUND")
|
|
return item
|
|
|
|
def add_event(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
stage: str,
|
|
message: str,
|
|
item_id: str | None = None,
|
|
level: str = "info",
|
|
details: dict[str, Any] | None = None,
|
|
expected_owner: str | None = None,
|
|
) -> None:
|
|
now = to_iso()
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
if expected_owner is not None:
|
|
lease = connection.execute(
|
|
"""
|
|
SELECT 1 FROM fleet_runs WHERE id = ? AND active_slot = 1
|
|
AND lease_owner = ? AND lease_until >= ?
|
|
AND status IN ('queued','running')
|
|
""",
|
|
(run_id, expected_owner, now),
|
|
).fetchone()
|
|
if lease is None:
|
|
raise RotationLeaseLostError("ROTATION_LEASE_LOST")
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id, item_id, occurred_at, stage, level, message, details_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
run_id,
|
|
item_id,
|
|
now,
|
|
stage,
|
|
level,
|
|
message,
|
|
json.dumps(details or {}, ensure_ascii=False),
|
|
),
|
|
)
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
|
|
def complete_item(self, run_id: str, item_id: str, expected_owner: str) -> bool:
|
|
now = to_iso()
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
lease = connection.execute(
|
|
"""
|
|
SELECT target_type, target_id FROM fleet_runs
|
|
WHERE id = ? AND active_slot = 1 AND lease_owner = ?
|
|
AND lease_until >= ? AND status IN ('queued','running')
|
|
""",
|
|
(run_id, expected_owner, now),
|
|
).fetchone()
|
|
if lease is None:
|
|
raise RotationLeaseLostError("ROTATION_LEASE_LOST")
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_run_items SET status = 'succeeded', stage = 'succeeded',
|
|
error_code = NULL, error_message = NULL, updated_at = ?, finished_at = ?
|
|
WHERE id = ? AND run_id = ?
|
|
""",
|
|
(now, now, item_id, run_id),
|
|
)
|
|
completed_item = connection.execute(
|
|
"SELECT instance_id, new_ip FROM fleet_run_items WHERE id = ?",
|
|
(item_id,),
|
|
).fetchone()
|
|
connection.execute(
|
|
"""
|
|
UPDATE managed_instances SET last_known_ip = ?, last_checked_at = ?,
|
|
updated_at = ? WHERE id = ?
|
|
""",
|
|
(
|
|
completed_item["new_ip"],
|
|
now,
|
|
now,
|
|
completed_item["instance_id"],
|
|
),
|
|
)
|
|
next_item = connection.execute(
|
|
"""
|
|
SELECT id FROM fleet_run_items
|
|
WHERE run_id = ? AND status = 'queued'
|
|
ORDER BY position LIMIT 1
|
|
""",
|
|
(run_id,),
|
|
).fetchone()
|
|
succeeded = int(
|
|
connection.execute(
|
|
"""
|
|
SELECT COUNT(*) FROM fleet_run_items
|
|
WHERE run_id = ? AND status = 'succeeded'
|
|
""",
|
|
(run_id,),
|
|
).fetchone()[0]
|
|
)
|
|
if next_item is not None:
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET current_item_id = ?, succeeded_items = ?,
|
|
status = 'running', updated_at = ? WHERE id = ?
|
|
""",
|
|
(next_item["id"], succeeded, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id,item_id,occurred_at,stage,level,message,details_json
|
|
) VALUES (?, ?, ?, 'preflight', 'info', '开始下一个实例', '{}')
|
|
""",
|
|
(run_id, next_item["id"], now),
|
|
)
|
|
connection.commit()
|
|
return False
|
|
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = 'succeeded', active_slot = NULL,
|
|
current_item_id = NULL, lease_owner = NULL, lease_until = NULL,
|
|
succeeded_items = ?, error_code = NULL, error_message = NULL,
|
|
updated_at = ?, finished_at = ? WHERE id = ?
|
|
""",
|
|
(succeeded, now, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
|
|
)
|
|
if lease["target_type"] == "group":
|
|
self._advance_group_schedule(connection, str(lease["target_id"]), now)
|
|
connection.commit()
|
|
return True
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
|
|
def fail_execution(
|
|
self,
|
|
run_id: str,
|
|
item_id: str,
|
|
*,
|
|
error_code: str,
|
|
error_message: str,
|
|
outcome: Literal["failed", "needs_attention", "cleanup_pending"],
|
|
expected_owner: str,
|
|
) -> FleetRun:
|
|
now = to_iso()
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"""
|
|
SELECT item.stage, run.target_type, run.target_id FROM fleet_runs run
|
|
JOIN fleet_run_items item ON item.id = run.current_item_id
|
|
WHERE run.id = ? AND item.id = ? AND run.active_slot = 1
|
|
AND run.lease_owner = ? AND run.lease_until >= ?
|
|
""",
|
|
(run_id, item_id, expected_owner, now),
|
|
).fetchone()
|
|
if row is None:
|
|
raise RotationLeaseLostError("ROTATION_LEASE_LOST")
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id,item_id,occurred_at,stage,level,message,details_json
|
|
) VALUES (?, ?, ?, ?, 'error', ?, ?)
|
|
""",
|
|
(
|
|
run_id,
|
|
item_id,
|
|
now,
|
|
row["stage"],
|
|
error_message,
|
|
json.dumps({"code": error_code}, ensure_ascii=False),
|
|
),
|
|
)
|
|
finished_at = now if outcome == "failed" else None
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_run_items SET status = ?, error_code = ?, error_message = ?,
|
|
updated_at = ?, finished_at = ? WHERE id = ?
|
|
""",
|
|
(outcome, error_code, error_message, now, finished_at, item_id),
|
|
)
|
|
if outcome == "failed":
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_run_items SET status = 'cancelled', stage = 'cancelled',
|
|
error_code = 'BATCH_ABORTED',
|
|
error_message = '同一批次中的前序实例失败，未执行此实例',
|
|
updated_at = ?, finished_at = ?
|
|
WHERE run_id = ? AND status = 'queued'
|
|
""",
|
|
(now, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = 'failed', active_slot = NULL,
|
|
lease_owner = NULL, lease_until = NULL, error_code = ?,
|
|
error_message = ?, updated_at = ?, finished_at = ? WHERE id = ?
|
|
""",
|
|
(error_code, error_message, now, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
|
|
)
|
|
if row["target_type"] == "group":
|
|
self._advance_group_schedule(
|
|
connection,
|
|
str(row["target_id"]),
|
|
now,
|
|
)
|
|
else:
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = ?, lease_owner = NULL,
|
|
lease_until = NULL, error_code = ?, error_message = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(outcome, error_code, error_message, now, run_id),
|
|
)
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
run = self.get(run_id)
|
|
if run is None: # pragma: no cover
|
|
raise RuntimeError("RUN_NOT_FOUND")
|
|
return run
|
|
|
|
def prepare_resume(self, run_id: str) -> FleetRun:
|
|
now = to_iso()
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
run = connection.execute(
|
|
"""
|
|
SELECT current_item_id FROM fleet_runs
|
|
WHERE id = ? AND active_slot = 1
|
|
AND status IN ('needs_attention','cleanup_pending')
|
|
""",
|
|
(run_id,),
|
|
).fetchone()
|
|
if run is None:
|
|
raise RuntimeError("RUN_NOT_RESUMABLE")
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = 'running', trigger = 'recovery',
|
|
lease_owner = NULL, lease_until = NULL, error_code = NULL,
|
|
error_message = NULL, updated_at = ? WHERE id = ?
|
|
""",
|
|
(now, run_id),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_run_items SET status = 'running',
|
|
attempt_count = attempt_count + 1, error_code = NULL,
|
|
error_message = NULL, updated_at = ? WHERE id = ?
|
|
""",
|
|
(now, run["current_item_id"]),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id,item_id,occurred_at,stage,level,message,details_json
|
|
) SELECT ?, id, ?, stage, 'info', '管理员已继续当前任务', '{}'
|
|
FROM fleet_run_items WHERE id = ?
|
|
""",
|
|
(run_id, now, run["current_item_id"]),
|
|
)
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
resumed = self.get(run_id)
|
|
if resumed is None: # pragma: no cover
|
|
raise RuntimeError("RUN_NOT_FOUND")
|
|
return resumed
|
|
|
|
def cancel(self, run_id: str) -> FleetRun:
|
|
now = to_iso()
|
|
message = "任务已由管理员放弃，请人工核对 AWS 静态 IP 与 Cloudflare DNS"
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
run = connection.execute(
|
|
"""
|
|
SELECT target_type, target_id, current_item_id FROM fleet_runs
|
|
WHERE id = ? AND active_slot = 1
|
|
AND status IN ('needs_attention','cleanup_pending')
|
|
""",
|
|
(run_id,),
|
|
).fetchone()
|
|
if run is None:
|
|
raise RuntimeError("RUN_NOT_CANCELABLE")
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_run_items SET status = 'cancelled', error_code =
|
|
CASE WHEN id = ? THEN 'ROTATION_CANCELLED' ELSE error_code END,
|
|
error_message = CASE WHEN id = ? THEN ? ELSE error_message END,
|
|
updated_at = ?, finished_at = COALESCE(finished_at, ?)
|
|
WHERE run_id = ? AND status != 'succeeded'
|
|
""",
|
|
(run["current_item_id"], run["current_item_id"], message, now, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
UPDATE fleet_runs SET status = 'cancelled', active_slot = NULL,
|
|
lease_owner = NULL, lease_until = NULL, error_code = 'ROTATION_CANCELLED',
|
|
error_message = ?, updated_at = ?, finished_at = ? WHERE id = ?
|
|
""",
|
|
(message, now, now, run_id),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
|
|
)
|
|
if run["target_type"] == "group":
|
|
connection.execute(
|
|
"""
|
|
UPDATE instance_groups SET enabled = 0, next_run_at = NULL, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(now, run["target_id"]),
|
|
)
|
|
else:
|
|
connection.execute(
|
|
"""
|
|
UPDATE instance_groups SET enabled = 0, next_run_at = NULL, updated_at = ?
|
|
WHERE id IN (
|
|
SELECT group_id FROM instance_group_members WHERE instance_id = ?
|
|
)
|
|
""",
|
|
(now, run["target_id"]),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_events(
|
|
run_id,item_id,occurred_at,stage,level,message,details_json
|
|
) VALUES (?, ?, ?, 'cancelled', 'warning', ?, ?)
|
|
""",
|
|
(
|
|
run_id,
|
|
run["current_item_id"],
|
|
now,
|
|
message,
|
|
json.dumps({"code": "ROTATION_CANCELLED"}),
|
|
),
|
|
)
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
cancelled = self.get(run_id)
|
|
if cancelled is None: # pragma: no cover
|
|
raise RuntimeError("RUN_NOT_FOUND")
|
|
return cancelled
|
|
|
|
def due_group_ids(self) -> list[str]:
|
|
now = to_iso()
|
|
with self.database.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT group_record.id FROM instance_groups AS group_record
|
|
WHERE group_record.archived_at IS NULL AND group_record.enabled = 1
|
|
AND group_record.next_run_at IS NOT NULL
|
|
AND group_record.next_run_at <= ?
|
|
AND EXISTS (
|
|
SELECT 1 FROM instance_group_members member
|
|
JOIN managed_instances instance ON instance.id = member.instance_id
|
|
WHERE member.group_id = group_record.id
|
|
AND instance.archived_at IS NULL AND instance.enabled = 1
|
|
)
|
|
ORDER BY group_record.next_run_at
|
|
""",
|
|
(now,),
|
|
).fetchall()
|
|
return [str(row["id"]) for row in rows]
|
|
|
|
@staticmethod
|
|
def _advance_group_schedule(
|
|
connection: sqlite3.Connection,
|
|
group_id: str,
|
|
finished_at: str,
|
|
) -> None:
|
|
group = connection.execute(
|
|
"""
|
|
SELECT enabled, interval_minutes FROM instance_groups
|
|
WHERE id = ? AND archived_at IS NULL
|
|
""",
|
|
(group_id,),
|
|
).fetchone()
|
|
if group is None:
|
|
return
|
|
next_run_at = (
|
|
to_iso(
|
|
from_iso(finished_at)
|
|
+ timedelta(minutes=int(group["interval_minutes"]))
|
|
)
|
|
if bool(group["enabled"])
|
|
else None
|
|
)
|
|
connection.execute(
|
|
"""
|
|
UPDATE instance_groups
|
|
SET last_run_at = ?, next_run_at = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(finished_at, next_run_at, finished_at, group_id),
|
|
)
|
|
|
|
def defer_group(self, group_id: str) -> None:
|
|
now = utc_now()
|
|
with self.database.connect() as connection, connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT interval_minutes FROM instance_groups
|
|
WHERE id = ? AND archived_at IS NULL AND enabled = 1
|
|
""",
|
|
(group_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return
|
|
connection.execute(
|
|
"""
|
|
UPDATE instance_groups SET next_run_at = ?, updated_at = ? WHERE id = ?
|
|
""",
|
|
(
|
|
to_iso(now + timedelta(minutes=int(row["interval_minutes"]))),
|
|
to_iso(now),
|
|
group_id,
|
|
),
|
|
)
|
|
|
|
def acquire_dns_sync_lock(self, owner_id: str, ttl_seconds: int = 300) -> bool:
|
|
now = utc_now()
|
|
now_iso = to_iso(now)
|
|
lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
|
|
with self.database.connect() as connection:
|
|
try:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
self._clear_expired_dns_lock(connection, now_iso)
|
|
if connection.execute(
|
|
"SELECT 1 FROM fleet_runs WHERE active_slot = 1"
|
|
).fetchone():
|
|
connection.rollback()
|
|
return False
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fleet_operation_locks(slot,kind,owner_id,lease_until,created_at)
|
|
VALUES (1,'dns_sync',?,?,?)
|
|
""",
|
|
(owner_id, lease_until, now_iso),
|
|
)
|
|
connection.commit()
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
connection.rollback()
|
|
return False
|
|
|
|
def renew_operation_lock(self, owner_id: str, ttl_seconds: int = 300) -> bool:
|
|
now = utc_now()
|
|
with self.database.connect() as connection, connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE fleet_operation_locks SET lease_until = ?
|
|
WHERE owner_id = ? AND kind = 'dns_sync' AND lease_until >= ?
|
|
""",
|
|
(to_iso(now + timedelta(seconds=ttl_seconds)), owner_id, to_iso(now)),
|
|
)
|
|
return cursor.rowcount == 1
|
|
|
|
def release_operation_lock(self, owner_id: str) -> None:
|
|
with self.database.connect() as connection, connection:
|
|
connection.execute(
|
|
"DELETE FROM fleet_operation_locks WHERE owner_id = ?", (owner_id,)
|
|
)
|
|
|
|
@staticmethod
|
|
def static_ip_name(instance_name: str, item_id: str) -> str:
|
|
normalized = "".join(
|
|
char.lower() if char.isascii() and char.isalnum() else "-"
|
|
for char in instance_name
|
|
)
|
|
normalized = "-".join(filter(None, normalized.split("-")))[:48] or "instance"
|
|
return f"fluxip-{normalized}-{item_id.replace('-', '')[:12]}"
|
|
|
|
@staticmethod
|
|
def stage_elapsed_seconds(item: FleetRunItem) -> float:
|
|
return (utc_now() - from_iso(item.stage_started_at)).total_seconds()
|
|
|
|
@staticmethod
|
|
def _clear_expired_dns_lock(connection: sqlite3.Connection, now: str) -> None:
|
|
connection.execute(
|
|
"""
|
|
DELETE FROM fleet_operation_locks
|
|
WHERE kind = 'dns_sync' AND lease_until IS NOT NULL AND lease_until < ?
|
|
""",
|
|
(now,),
|
|
)
|