[Iris] Replace txn_log audit table with structured log lines

lib/iris/OPS.md

@@ -118,9 +118,14 @@ SELECT slice_id, lifecycle, scale_group, worker_ids FROM slices WHERE lifecycle=
 -- Task attempt history (debugging retries)
 SELECT task_id, attempt_id, state, exit_code, error FROM task_attempts
 WHERE task_id LIKE '%<job_fragment>%' ORDER BY attempt_id;
+```
+
+Controller audit events (`event=<kind> action=<action> entity=<id> ...`) are
+emitted as structured `logger.info` lines — query them through
+`iris process logs` with a substring filter, not via SQL. Example:
 
--- What the controller has been doing
-SELECT kind, count(*) FROM txn_log GROUP BY kind ORDER BY count(*) DESC LIMIT 10;
+```bash
+iris process logs --since 24h | grep 'action=worker_heartbeat_failed'
 ```
 
 Full table list: `iris query "SELECT name FROM sqlite_master WHERE type='table'"`.

lib/iris/dashboard/src/App.vue

@@ -20,7 +20,6 @@ const WORKER_TABS: Tab[] = [
   { key: 'fleet', label: 'Workers', to: '/fleet' },
   { key: 'endpoints', label: 'Endpoints', to: '/endpoints' },
   { key: 'autoscaler', label: 'Autoscaler', to: '/autoscaler' },
-  { key: 'transactions', label: 'Transactions', to: '/transactions' },
   { key: 'account', label: 'Account', to: '/account' },
   { key: 'status', label: 'Status', to: '/status' },
 ]
@@ -30,7 +29,6 @@ const KUBERNETES_TABS: Tab[] = [
   { key: 'scheduler', label: 'Scheduler', to: '/scheduler' },
   { key: 'cluster', label: 'Cluster', to: '/cluster' },
   { key: 'endpoints', label: 'Endpoints', to: '/endpoints' },
-  { key: 'transactions', label: 'Transactions', to: '/transactions' },
   { key: 'account', label: 'Account', to: '/account' },
   { key: 'status', label: 'Status', to: '/status' },
 ]
@@ -46,7 +44,6 @@ const PATH_TO_TAB: Record<string, string> = {
   '/cluster': 'cluster',
   '/endpoints': 'endpoints',
   '/autoscaler': 'autoscaler',
-  '/transactions': 'transactions',
   '/account': 'account',
   '/status': 'status',
 }

lib/iris/dashboard/src/router.ts

@@ -30,10 +30,6 @@ const routes = [
     path: '/status',
     component: () => import('./components/controller/StatusTab.vue'),
   },
-  {
-    path: '/transactions',
-    component: () => import('./components/controller/TransactionsTab.vue'),
-  },
   {
     path: '/scheduler',
     component: () => import('./components/controller/SchedulerTab.vue'),

lib/iris/dashboard/src/types/rpc.ts

@@ -458,19 +458,6 @@ export interface GetProcessStatusResponse {
   logEntries?: LogEntry[]
 }
 
-// -- Transactions --
-
-export interface TransactionAction {
-  timestamp?: ProtoTimestamp
-  action?: string
-  entityId?: string
-  details?: string
-}
-
-export interface GetTransactionsResponse {
-  actions: TransactionAction[]
-}
-
 // -- Task State Counts (used in job summaries and user summaries) --
 
 /** Mapping from lowercase state name to count, e.g. { running: 2, pending: 5 } */

lib/iris/scripts/benchmark_db_queries.py

@@ -102,8 +102,6 @@
     "task_resource_history",
     "endpoints",
     "reservation_claims",
-    "txn_log",
-    "txn_actions",
     "meta",
     "schema_migrations",
 ]

lib/iris/src/iris/cluster/controller/auth.py

@@ -56,6 +56,13 @@ def create_api_key(
         "VALUES (?, ?, ?, ?, ?, ?, ?)",
         (key_id, key_hash, key_prefix, user_id, name, now.epoch_ms(), expires_at.epoch_ms() if expires_at else None),
     )
+    logger.info(
+        "event=api_key_created entity=%s trigger=- user=%s name=%s expires_at_ms=%s",
+        key_id,
+        user_id,
+        name,
+        expires_at.epoch_ms() if expires_at else "-",
+    )
 
 
 def lookup_api_key_by_hash(db: ControllerDB, key_hash: str) -> ApiKeyRow | None:
@@ -81,7 +88,10 @@ def revoke_api_key(db: ControllerDB, key_id: str, now: Timestamp) -> bool:
             f"UPDATE {db.api_keys_table} SET revoked_at_ms = ? WHERE key_id = ? AND revoked_at_ms IS NULL",
             (now.epoch_ms(), key_id),
         )
-        return cur._cursor.rowcount > 0
+        revoked = cur._cursor.rowcount > 0
+    if revoked:
+        logger.info("event=api_key_revoked entity=%s trigger=-", key_id)
+    return revoked
 
 
 def list_api_keys(db: ControllerDB, user_id: str | None = None) -> list[ApiKeyRow]:
@@ -111,6 +121,11 @@ def revoke_login_keys_for_user(db: ControllerDB, user_id: str, now: Timestamp) -
                 " WHERE user_id = ? AND name LIKE 'login-%' AND revoked_at_ms IS NULL",
                 (now.epoch_ms(), user_id),
             )
+        logger.info(
+            "event=login_keys_revoked entity=%s trigger=- count=%d",
+            user_id,
+            len(revoked_ids),
+        )
     return revoked_ids
 
 

lib/iris/src/iris/cluster/controller/controller.py

@@ -936,9 +936,6 @@ class ControllerConfig:
     worker_retention: Duration = field(default_factory=lambda: Duration.from_seconds(86400))
     """Delete inactive/unhealthy workers whose last heartbeat exceeds this (default: 24 hours)."""
 
-    txn_action_retention: Duration = field(default_factory=lambda: Duration.from_seconds(3 * 86400))
-    """Delete txn_actions older than this (default: 3 days)."""
-
     profile_retention: Duration = field(default_factory=lambda: Duration.from_seconds(86400))
     """Delete task_profiles older than this (default: 24 hours)."""
 
@@ -1419,7 +1416,6 @@ def _run_prune_loop(self, stop_event: threading.Event) -> None:
                     self._transitions.prune_old_data(
                         job_retention=self._config.job_retention,
                         worker_retention=self._config.worker_retention,
-                        txn_action_retention=self._config.txn_action_retention,
                         profile_retention=self._config.profile_retention,
                         stop_event=stop_event,
                     )

lib/iris/src/iris/cluster/controller/migrations/0004_worker_indexes.py

@@ -5,18 +5,9 @@
 
 
 def migrate(conn: sqlite3.Connection) -> None:
-    conn.executescript(
-        """
-DROP TRIGGER IF EXISTS trg_txn_log_retention;
-CREATE TRIGGER IF NOT EXISTS trg_txn_log_retention
-AFTER INSERT ON txn_log
-WHEN (SELECT COUNT(*) FROM txn_log) > 1100
-BEGIN
-  DELETE FROM txn_log WHERE id <= (
-    SELECT id FROM txn_log ORDER BY id DESC LIMIT 1 OFFSET 1000
-  );
-END;
-
-CREATE INDEX IF NOT EXISTS idx_workers_healthy_active ON workers(healthy, active);
-"""
-    )
+    # Originally this migration also rewrote the `trg_txn_log_retention`
+    # trigger; those statements were removed once migration 0037 dropped the
+    # `txn_log` / `txn_actions` tables entirely. On DBs that already ran the
+    # old form the trigger survives until 0037 executes; 0037 is idempotent
+    # (`DROP TRIGGER IF EXISTS`) so no fixup is needed here.
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_workers_healthy_active ON workers(healthy, active)")

lib/iris/src/iris/cluster/controller/migrations/0037_drop_txn_log_and_txn_actions.py

@@ -0,0 +1,19 @@
+# Copyright The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Drop the `txn_log` and `txn_actions` audit tables.
+
+The controller no longer records mutating transitions to SQLite; every
+state change now emits a semi-structured `logger.info` line via
+``transitions.log_event`` that is captured by the Iris log server and
+queried through the normal log-store DuckDB interface.
+"""
+
+import sqlite3
+
+
+def migrate(conn: sqlite3.Connection) -> None:
+    conn.execute("DROP TRIGGER IF EXISTS trg_txn_log_retention")
+    conn.execute("DROP INDEX IF EXISTS idx_txn_actions_txn")
+    conn.execute("DROP TABLE IF EXISTS txn_actions")
+    conn.execute("DROP TABLE IF EXISTS txn_log")

lib/iris/src/iris/cluster/controller/schema.py

@@ -1112,62 +1112,6 @@ def generate_full_ddl(tables: Sequence[Table]) -> str:
     indexes=("CREATE INDEX IF NOT EXISTS idx_dispatch_worker ON dispatch_queue(worker_id, id)",),
 )
 
-TXN_LOG = Table(
-    "txn_log",
-    "tl",
-    columns=(
-        Column("id", "INTEGER", "PRIMARY KEY AUTOINCREMENT"),
-        Column("kind", "TEXT", "NOT NULL", python_type=str, decoder=str),
-        Column("payload_json", "TEXT", "NOT NULL", python_name="payload", python_type=dict, decoder=_decode_json_dict),
-        Column(
-            "created_at_ms",
-            "INTEGER",
-            "NOT NULL",
-            python_name="created_at",
-            python_type=Timestamp,
-            decoder=decode_timestamp_ms,
-        ),
-    ),
-    triggers=(
-        # Migration 0004_worker_indexes rewrote the trigger from 0001
-        """CREATE TRIGGER IF NOT EXISTS trg_txn_log_retention
-AFTER INSERT ON txn_log
-WHEN (SELECT COUNT(*) FROM txn_log) > 1100
-BEGIN
-  DELETE FROM txn_log WHERE id <= (
-    SELECT id FROM txn_log ORDER BY id DESC LIMIT 1 OFFSET 1000
-  );
-END;""",
-    ),
-)
-
-TXN_ACTIONS = Table(
-    "txn_actions",
-    "ta2",
-    columns=(
-        Column("id", "INTEGER", "PRIMARY KEY AUTOINCREMENT"),
-        Column(
-            "txn_id",
-            "INTEGER",
-            "NOT NULL REFERENCES txn_log(id) ON DELETE CASCADE",
-            python_type=int,
-            decoder=int,
-        ),
-        Column("action", "TEXT", "NOT NULL", python_type=str, decoder=str),
-        Column("entity_id", "TEXT", "NOT NULL", python_type=str, decoder=str),
-        Column("details_json", "TEXT", "NOT NULL", python_name="details", python_type=dict, decoder=_decode_json_dict),
-        Column(
-            "created_at_ms",
-            "INTEGER",
-            "NOT NULL",
-            python_name="timestamp",
-            python_type=Timestamp,
-            decoder=decode_timestamp_ms,
-        ),
-    ),
-    indexes=("CREATE INDEX IF NOT EXISTS idx_txn_actions_txn ON txn_actions(txn_id, id)",),
-)
-
 # Migration 0003: restructured scaling_groups
 SCALING_GROUPS = Table(
     "scaling_groups",
@@ -1370,8 +1314,6 @@ def generate_full_ddl(tables: Sequence[Table]) -> str:
     TASK_RESOURCE_HISTORY,
     ENDPOINTS,
     DISPATCH_QUEUE,
-    TXN_LOG,
-    TXN_ACTIONS,
     SCALING_GROUPS,
     SLICES,
     RESERVATION_CLAIMS,
@@ -1632,16 +1574,6 @@ class EndpointRow:
     registered_at: Timestamp
 
 
-@dataclass(frozen=True, slots=True)
-class TransactionActionRow:
-    """Transaction action log entry."""
-
-    timestamp: Timestamp
-    action: str
-    entity_id: str
-    details: dict
-
-
 @dataclass(frozen=True, slots=True)
 class ApiKeyRow:
     """API key record."""
@@ -1947,15 +1879,6 @@ def _job_columns(*names: str) -> tuple[tuple[Column, ...], tuple[str, ...]]:
     row_cls=EndpointRow,
 )
 
-# Transaction action row.
-TXN_ACTION_PROJECTION = TXN_ACTIONS.projection(
-    "created_at_ms",
-    "action",
-    "entity_id",
-    "details_json",
-    row_cls=TransactionActionRow,
-)
-
 # API key row.
 API_KEY_PROJECTION = AUTH_API_KEYS.projection(
     "key_id",