masakari — Async processing

Asynchronous processing — masakari — branch stable/2026.2

Branch stable/2026.2 @ ae41528e1ebc — repository https://opendev.org/openstack/masakari — analysed on 2026-10-28

All the non-HTTP behaviour lives in masakari-engine (masakari/cmd/engine.py). It runs one oslo.messaging RPC server on topic [DEFAULT]/masakari_topic = ha_engine (masakari/conf/engine.py:20), with server = [DEFAULT]/host (masakari/service.py:91), the eventlet executor (masakari/rpc.py:147) and the endpoint MasakariManager. Its periodic tasks are scheduled by an oslo.service dynamic timer (masakari/service.py:103). The engine has no heartbeat, no service table and no leader election: every engine instance consumes the same queue and runs every periodic task.

Periodic tasks start after a random delay of 0–[DEFAULT]/periodic_fuzzy_delay (60 s) and run when [DEFAULT]/periodic_enable is true (default, masakari/conf/service.py:37). The two task intervals are read at import time (spacing=CONF.…, masakari/engine/manager.py:354), so a configuration reload does not change them.

process_notification (masakari-engine)

• Trigger: RPC cast process_notification(notification=<Notification 1.3>), version 1.0, emitted by masakari-api after the INSERT (masakari/engine/rpcapi.py:48). Exchange openstack (default control exchange), queue ha_engine, round-robin between engines.
• Emitter: NotificationAPI._create_notification (masakari/ha/api.py:333).
• Purpose: turn a failure report into recovery actions on Nova.
• Flow:
  1. Load the source host. If its segment is disabled: status = ignored, message = "Recovery aboart: the segment HA is disabled.", then raise FailoverSegmentDisabled (logged by oslo.messaging, masakari/engine/manager.py:339).
  2. Emit notification.process.start (masakari/engine/manager.py:328).
  3. Take a process-local lock named after the source host UUID (blocking, masakari/engine/manager.py:274, masakari/utils.py:243): notifications of the same host are serialised within one engine only.
  4. Re-read the notification. If the message says new but the database no longer does, skip it (bug 1773132, masakari/engine/manager.py:288). This protects against a replayed message, but not against two engines receiving two different messages for the same row [inferred].
  5. status = running, then dispatch on the type (below), then write the final status (masakari/engine/manager.py:322) and emit notification.process.end (or .error with the exception and traceback).
• Concurrency and idempotency:
  • Two engines can process two notifications of the same host at the same time. The API refuses a duplicate only within 180 s with an identical payload, and the lock is per process.
  • Retries replay the whole workflow, because each run creates a new taskflow flow detail in the same logbook (masakari/engine/drivers/taskflow/base.py:124). Nothing is resumed.
• Error behaviour: the cast has no reply. MasakariException subclasses become error; any other exception escapes and leaves the row running until it expires [inferred].

COMPUTE_HOST: _handle_notification_type_host (masakari/engine/manager.py:174)

Recovery of a stopped host:

1. Set the host on_maintenance = true, and reserved = false if it was a reserved host (masakari/engine/manager.py:207).
2. If the segment's method is not auto, list its reserved hosts that are not on maintenance (masakari/engine/manager.py:223), **in the default order created_at desc, id desc**.
3. Run the method (masakari/engine/drivers/taskflow/driver.py:134):
   • auto: auto flow;
   • reserved_host: RH flow, and with an empty list → ReservedHostsUnavailable (masakari/engine/drivers/taskflow/driver.py:68);
   • auto_priority: auto flow, and on any exception other than SkipHostRecoveryException → RH flow (masakari/engine/drivers/taskflow/driver.py:83);
   • rh_priority: RH flow, then auto flow on failure (masakari/engine/drivers/taskflow/driver.py:109).

The auto flow is a taskflow linear flow pre → main → post whose tasks are loaded by stevedore from [taskflow_driver_recovery_flows]/host_auto_failure_recovery_tasks (default pre: disable_compute_service_task, main: prepare_HA_enabled_instances_task, post: evacuate_instances_task, masakari/conf/engine_driver.py:128). The RH flow (host_rh_failure_recovery_tasks, default pre: disable_compute_service_task, main: prepare_HA_enabled_instances_task, evacuate_instances_task) wraps main in a ParameterizedForEach retry over reserved_host_list, so the main tasks are re-run with the next reserved host when one fails (masakari/engine/drivers/taskflow/host_failure.py:537).

The three tasks:

• DisableComputeServiceTask (masakari/engine/drivers/taskflow/host_failure.py:44): Nova disables nova-compute on the host with reason [host_failure]/service_disable_reason ("Masakari detected host failed."), then **sleeps wait_period_after_service_update (180 s)** "until nova recognizes the node down". Masakari never forces the service down (forced_down is not used), so Nova accepts the evacuation only once its own service group sees the node down [inferred from Nova's evacuate preconditions].
• PrepareHAEnabledInstancesTask (masakari/engine/drivers/taskflow/host_failure.py:64):
  1. List all servers of the host (GET /servers/detail?all_tenants=true&host=<name>, one page only: no limit, so Nova's max_limit (1000) caps it, masakari/compute/nova.py:147).
  2. Keep those with metadata [host_failure]/ha_enabled_instance_metadata_key (HA_Enabled) true. Skip error instances when ignore_instances_in_error_state. Append the others when evacuate_all_instances (default true).
  3. With nothing left, raise SkipHostRecoveryException → finished.
  4. Otherwise create one vmoves row per instance (pending, evacuation).
• EvacuateInstancesTask (masakari/engine/drivers/taskflow/host_failure.py:153):
  1. Read the vmoves still pending, in the default order created_at desc, id desc. Since the HA-enabled instances were inserted first, non-HA instances are evacuated first, against the help text of evacuate_all_instances ("First preference will be given to those instances which have … True"). Observed in host-failure-auto.
  2. With a reserved host: optionally add it to every aggregate that contains the failed host ([host_failure]/add_reserved_host_to_aggregate; a Nova 409 is ignored), enable its compute service, and set it reserved = false in Masakari. This runs under a non-blocking, process-local lock named after the reserved host (masakari/engine/drivers/taskflow/host_failure.py:458).
  3. Evacuate in a green thread pool of [DEFAULT]/host_failure_recovery_threads (3). For each instance (_evacuate_and_confirm, masakari/engine/drivers/taskflow/host_failure.py:193):
     1. GET the server twice. Lock it unless it is already locked.
     2. Normalise its state:
        • a vm_state other than active/stopped/error → os-resetState to error;
        • an active instance with a pending task_state → reset to error and do not stop it afterwards;
        • a resized instance that is not shut down → do not stop it afterwards.
     3. vmove ongoing, POST evacuate (with host only in the RH flow; Nova chooses otherwise).
     4. Poll until the instance is on another host (and vm_state back to the old value, or error → active).
     5. Instances that were not active are stopped again (os-stop, then poll until stopped), and an instance that was in error is reset to error again.
     6. vmove succeeded with dest_host. Any exception or timeout → vmove failed with the message.
     7. In all cases set end_time and unlock; an unlock failure is only logged (masakari/engine/drivers/taskflow/host_failure.py:341).
  4. Re-read all vmoves of the notification. Any failed, or any still pending/ongoing → HostRecoveryFailureException (masakari/engine/drivers/taskflow/host_failure.py:424).

**Defect (observed, scenario host-failure-auto-priority-fallback).** The check of step 4 counts the vmoves of earlier attempts too. Under auto_priority, the auto attempt leaves a failed vmove. The RH fallback runs PrepareHAEnabledInstancesTask again (a second vmove for the same instance) and evacuates the instance successfully to the reserved host, but the notification still ends error ("Failed to evacuate instances 'X,X'"). The same applies to rh_priority, to the ParameterizedForEach retry of the next reserved host, and to the periodic retry of an error notification. In the observed run, the periodic retry later found no instance left on the host and ended finished.

VM: _handle_notification_type_instance (masakari/engine/manager.py:130)

• Only these (event, vir_domain_event) pairs are processed (masakari/engine/instance_events.py:22): (LIFECYCLE, STOPPED_FAILED), (QEMU_GUEST_AGENT_ERROR, STOPPED_FAILED) and (IO_ERROR, IO_ERROR_REPORT). Every other pair → ignored.
• The instance flow (instance_failure_recovery_tasks, default stop → start → confirm active):
  • StopInstanceTask (masakari/engine/drivers/taskflow/instance_failure.py:40): skip it if it is not HA-enabled and [instance_failure]/process_all_instances is false (→ finished); paused/rescued → ignored; resized → reset to active; then os-stop (a Nova 409 is tolerated if the instance is stopped) and poll.
  • StartInstanceTask: os-start if stopped, otherwise InstanceRecoveryFailureException.
  • ConfirmInstanceActiveTask: poll until active.
• The flow works on the source host only: it does not check that the instance is still on the notifying host. It also starts an instance that was deliberately stopped: an IO_ERROR on a stopped HA instance ends with the instance active (observed, instance-failure:vm-io-error).

PROCESS: _handle_notification_type_process (masakari/engine/manager.py:67)

• STARTED → finished, nothing done.
• STOPPED → the host is set **on_maintenance = true first** (masakari/engine/manager.py:90), then:
  • nova-compute → disable the compute service unless it is already disabled (reason [process_failure]/service_disable_reason), and poll until disabled;
  • any other process name → SkipProcessRecoveryException → finished, with the host left on maintenance: later host failures of that node are refused with 409 (observed, process-failure/libvirtd).

get_notification_recovery_workflow_details (masakari-engine)

• Trigger: RPC call, version 1.1, from GET /v1/notifications/{id} at microversion ≥ 1.1 (masakari/engine/rpcapi.py:53).
• Flow: open the taskflow persistence backend, read the logbook named after the notification UUID, and list its flows and atoms. Then order the atoms by the configured task list of the notification type and method (masakari/engine/drivers/taskflow/driver.py:258), and return each atom that stored progress_details in its meta (masakari/engine/drivers/taskflow/driver.py:349).
• Defect: the comparison recovery_method in ['AUTO_PRIORITY', 'RH_PRIORITY'] (masakari/engine/drivers/taskflow/driver.py:319) uses upper-case values while the database stores auto_priority/rh_priority, so the "which flow ran" detection never applies. auto_priority is then ordered by the auto task list. With the default lists the RH tasks have the same names, so the effect is limited to reordering when operators customise the lists [inferred].
• Error: any exception → MasakariException returned to the API → 500.

_process_unfinished_notifications (masakari-engine)

• Trigger: every process_unfinished_notifications_interval (120 s) in every engine.
• Data read: SELECT * FROM notifications WHERE deleted = 0 AND status IN ('error','new') ORDER BY created_at DESC, id DESC (masakari/db/sqlalchemy/api.py:496), with no limit.
• Flow: for each row:
  1. If it is error, or new with generated_time older than retry_notification_new_status_interval (60 s, mutable), run _process_notification directly. This bypasses the "segment disabled" check of process_notification [inferred from masakari/engine/manager.py:370].
  2. Re-read the row; if it is still error, set it to failed (masakari/engine/manager.py:376).
• Concurrency: no lock across engines. Two engines can retry the same notification at the same time, and the per-host lock is per process. The guard at masakari/engine/manager.py:288 does not stop it, because the in-memory status is already error/new [inferred]. With N engines, a failed host recovery can be replayed N times.
• Cost: one query, plus one full workflow per retried row, sequentially inside the periodic task. A host flow sleeps at least 180 s in DisableComputeServiceTask, so a backlog delays the other periodic task.

_check_expired_notifications (masakari-engine)

• Trigger: every 600 s in every engine.
• Data read: notifications WHERE status IN ('running','error','new'). Each row whose generated_time (sender clock) is older than notifications_expired_interval (86400 s) is set to failed (masakari/engine/manager.py:404). A running workflow that is still alive is marked failed without being stopped [inferred].

Maintenance commands (not triggers, run by the operator)

• masakari-manage db sync / db version (masakari/cmd/manage.py:67).
• masakari-manage db purge --age_in_days N --max_rows M: hard-deletes soft-deleted segments and hosts older than N days, and notifications that are finished/failed/ignored with updated_at older than N days (masakari/db/sqlalchemy/api.py:752). **vmoves rows are never purged**: they are soft-deleted only by notification_delete (masakari/db/sqlalchemy/api.py:623), which no API or engine path calls, and the purge selects other tables on deleted_at (masakari/db/sqlalchemy/api.py:791). The purge then hard-deletes the notifications, leaving orphan vmoves.
• masakari-status upgrade check (masakari/cmd/status.py:26).

Source: reverse-spec-masakari/05-async-processing_stable-2026.2.md. The _rewrite/ and _inventory/ files are not published on the site.