mirror of
https://github.com/vatesfr/xen-orchestra.git
synced 2026-09-10 22:14:48 -05:00
feat(xo-server): add crash-resistant on-disk observability for RPU/RPR (#10078)
* feat(xo-server): add crash-resistant on-disk observability for RPU/RPR (XO-2640) * fix(xo-server): reconcile RPU heartbeats left pending at boot (XO-2640) * chore(CHANGELOG): add RPU observability entry (XO-2640) * docs(rpu): document on-disk traces and heartbeat for troubleshooting * refactor(xo-server): address RPU observability review feedback
This commit is contained in:
Mathieu
@@ -22,6 +22,7 @@
|
||||
- [vm stats] Reduce the memory consumption of the rrd stats (PR [#10039](https://github.com/vatesfr/xen-orchestra/pull/10039))
|
||||
- [XO6/Site] Add "New VM" and "Connect pool" actions to the site infrastructure tree (PR [#10047](https://github.com/vatesfr/xen-orchestra/pull/10047))
|
||||
- [VM]: Add possibility to duplicate a VM (PR [#9580](https://github.com/vatesfr/xen-orchestra/pull/9580))
|
||||
- [RPU] Trace rolling pool updates/reboots to disk to allow diagnosis even after xo-server restarts (PR [#10078](https://github.com/vatesfr/xen-orchestra/pull/10078))
|
||||
|
||||
### Bug fixes
|
||||
|
||||
|
||||
@@ -349,19 +349,19 @@ Activating "Auto Power on" for a VM will also configure the pool accordingly. If
|
||||
|
||||
#### What's a VIF?
|
||||
|
||||
In the Vates stack, each VM connects to the network through a **Virtual Interface**, or **VIF**.
|
||||
In the Vates stack, each VM connects to the network through a **Virtual Interface**, or **VIF**.
|
||||
|
||||
Think of it as the VM’s virtual network card —it handles all incoming and outgoing traffic on a specific virtual network, which is linked to a physical NIC on the host via a bridge. Every VIF comes with a fixed MAC address and can be tied to a particular VLAN (or virtual network), depending on how your environment is set up.
|
||||
|
||||
#### What does VIF locking mode do?
|
||||
|
||||
**VIF locking mode** is a security feature that helps prevent unauthorized or spoofed traffic from getting in or out of a VM.
|
||||
**VIF locking mode** is a security feature that helps prevent unauthorized or spoofed traffic from getting in or out of a VM.
|
||||
|
||||
In practical terms, enabling locking mode means **the VM can’t send traffic using a fake MAC address** — and if an IP address is set, it won’t be able to use anything else either. This is **especially helpful in multi-tenant setups** or anytime you need tight control over which VM is allowed to do what on the network.
|
||||
|
||||
#### Default behavior
|
||||
|
||||
**By default, VIFs are not locked.** Users can assign any IP address to their VM, and they will work without restriction.
|
||||
**By default, VIFs are not locked.** Users can assign any IP address to their VM, and they will work without restriction.
|
||||
However, if you add restricted IPs, the VIF becomes locked, meaning only the specified IP addresses are allowed to send traffic outside the VM.
|
||||
|
||||
#### Adding restricted IP addresses
|
||||
@@ -382,17 +382,18 @@ To add a restricted IP address:
|
||||
1. Click the pencil icon (with the **Edit locking mode** label). A dropdown menu appears.
|
||||
1. Choose your desired mode from that menu:
|
||||
|
||||
|
||||
|
||||
If Xen Orchestra knows the VM’s IP address — either through the guest agent or DHCP — it will automatically apply IP-level locking when the mode is enabled.
|
||||
|
||||
#### How do I automate VIF locking mode?
|
||||
|
||||
If you’re automating things or working with scripts, you can also control VIF locking mode through the XO [command-line interface](architecture#xo-cli-cli).
|
||||
If you’re automating things or working with scripts, you can also control VIF locking mode through the XO [command-line interface](architecture#xo-cli-cli).
|
||||
|
||||
### VM high availability (HA)
|
||||
|
||||
If your pool supports HA (must have shared storage), you can activate "HA". Read the XCP-ng documentation for more details on [High Availability with XCP-ng](https://docs.xcp-ng.org/management/ha/).
|
||||
In the VM's Advanced tab, you can choose between three HA modes:
|
||||
|
||||
- Restart: a protected VM cannot be immediately restarted after a server failure. HA will attempt to restart the VM when additional capacity becomes available in the pool.
|
||||
- Best-Effort: for VMs configured with best-effort, HA will try to restart them on another host if their original host goes offline. This attempt occurs only after all VMs set to the "restart" mode have been successfully restarted. HA will make only one attempt to restart a best-effort VM; if it fails, no further attempts will be made.
|
||||
- Disabled: if an unprotected VM or its host is stopped, HA does not attempt to restart the VM.
|
||||
@@ -407,7 +408,7 @@ In Xen Orchestra, you can filter to see which VMs are in which mode:
|
||||
You can use containers (such as Docker) inside your VMs. To know more on installing Docker inside a VM, check out the official [Docker documentation](https://docs.docker.com/).
|
||||
|
||||
:::tip
|
||||
You can also run containers using Kubernetes, using the dedicated [Kubernetes recipe](https://docs.vates.tech/devops-tools/kubernetes).
|
||||
You can also run containers using Kubernetes, using the dedicated [Kubernetes recipe](https://docs.vates.tech/devops-tools/kubernetes).
|
||||
:::
|
||||
|
||||
### VM CPU priority
|
||||
@@ -514,7 +515,6 @@ To prevent a virtual machine from migrating:
|
||||
2. Go to the **Advanced** tab.
|
||||
3. Activate the **Prevent migration** toggle switch.
|
||||
|
||||
|
||||
When this option is enabled, the VM won't be able to migrate to another host.
|
||||
|
||||
## Migrating from VMware with V2V
|
||||
@@ -525,7 +525,6 @@ You can migrate your VM from VMware vSphere to a Vates environment (Xen Orchestr
|
||||
|
||||
To know more on using V2V in Xen Orchestra to migrate your environment from VMware, refer to the [XO V2V section in the XCP-ng documentation](https://docs.xcp-ng.org/installation/migrate-to-xcp-ng/#xo-v2v).
|
||||
|
||||
|
||||
## Hosts management
|
||||
|
||||
Outside updates (see next section), you can also do host management via Xen Orchestra. Basic operations are supported, like reboot, shutdown and so on.
|
||||
@@ -585,7 +584,7 @@ On XCP-ng, there's multiple way to update your pools and hosts.
|
||||
|
||||
#### ⚙️ How it works
|
||||
|
||||
Xen Orchestra will request a plugin, bundled and hosted within your XCP-ng hosts. This plugin will query the status of updates. Then, when the update will be applied, it's also the plugin that will download and apply them.
|
||||
Xen Orchestra will request a plugin, bundled and hosted within your XCP-ng hosts. This plugin will query the status of updates. Then, when the update will be applied, it's also the plugin that will download and apply them.
|
||||
So unlike with Citrix Hypervisor, Xen Orchestra will not fetch or download updates, but it will order the host to do it by itself. Be sure that your host(s) can access the update repositories.
|
||||
|
||||
#### Rolling Pool Updates (RPU)
|
||||
@@ -605,6 +604,7 @@ XO will restart the hosts one by one and wait for each host to be up and running
|
||||
[xapiOptions]
|
||||
restartHostTimeout = '40 minutes'
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
|
||||
@@ -617,12 +617,39 @@ You can schedule rolling pool updates to automate the patching and rebooting of
|
||||
Of course, schedule the updates outside production hours. Although RPUs are designed to be transparent, it is always wise to schedule some buffer time in case something goes wrong.
|
||||
:::
|
||||
|
||||
1. **Create a job**: Navigate to the **Jobs** screen and create a new job, using the `pool.rollingUpdate` method. Determine which pools you want to include in the update:
|
||||
1. **Create a job**: Navigate to the **Jobs** screen and create a new job, using the `pool.rollingUpdate` method. Determine which pools you want to include in the update:
|
||||
|
||||
1. **Set the schedule**: Assign a schedule to the job (e.g., every Sunday at 3 AM) and apply it to your new job:
|
||||
|
||||
1. **Monitor the process**: Once scheduled, the RPU will run automatically at the set time, applying updates and rebooting hosts as necessary.
|
||||
|
||||
##### Troubleshooting a RPU
|
||||
|
||||
XO keeps a full trace of each rolling pool update/reboot on disk. So even if `xo-server` crashed in the middle of a run, you can still find out what happened and where it stopped.
|
||||
|
||||
Each run writes two files in `<datadir>/rpu-traces` (on XOA: `/var/lib/xo-server/data/rpu-traces`):
|
||||
|
||||
- `rpu-<poolId>-<timestamp>.ndjson`: the trace. One JSON line per event: steps, progress, VM migrations, errors. Lines are written synchronously, so the file stays readable even after a crash.
|
||||
- `rpu-<poolId>-<timestamp>.heartbeat.json`: rewritten every 5 seconds with the current status, as long as the run is alive.
|
||||
|
||||
The path of the trace file is logged in `journalctl -u xo-server` when the RPU starts. You'll also find it on the task itself, as the `traceFile` property in the tasks view.
|
||||
|
||||
A heartbeat stuck on `"status": "pending"` with an old `lastUpdated` means the run died mid-flight. When `xo-server` comes back, it marks the task as `interrupted`, stamps the heartbeat with the last time the run was alive, and logs a line pointing at the trace.
|
||||
|
||||
There's no resume yet: just launch the RPU again. Hosts that were already patched have nothing left to install, so the new run goes through them quickly.
|
||||
|
||||
:::tip
|
||||
Only one RPU or rolling pool reboot can run at a time on a given pool. A second attempt is rejected with an explicit error, not queued.
|
||||
:::
|
||||
|
||||
Traces are deleted after 31 days. Both the folder and the retention can be changed in your `xo-server` config:
|
||||
|
||||
```toml
|
||||
[rpu]
|
||||
#tracesDir = '/var/lib/xo-server/data/rpu-traces'
|
||||
tracesRetention = '31 days'
|
||||
```
|
||||
|
||||
#### Pool updates
|
||||
|
||||
If you can't use RPU (Rolling Pool Updates), you can still use "Install pool patches" button. This will simply install updates on all hosts on your pool and restart the toolstack, **without doing any host reboot**:
|
||||
@@ -646,7 +673,8 @@ We do NOT recommend to install updates to individual hosts. Obviously except if
|
||||
### XenServer/Citrix Hypervisor
|
||||
|
||||
#### ⚙️ How it works
|
||||
Xen Orchestra will directly request a specific XML, hosted by Citrix. It will be analyzed and compared to the patch level on your hosts.
|
||||
|
||||
Xen Orchestra will directly request a specific XML, hosted by Citrix. It will be analyzed and compared to the patch level on your hosts.
|
||||
If there's available/missing updates, XO will download it directly, then send it to the pool, and finally ask the pool to apply it. In that scenario, you should check if Xen Orchestra can access outside to get those updates.
|
||||
|
||||
#### Pool updates
|
||||
@@ -677,9 +705,10 @@ As specified in the [documentation](https://xcp-ng.org/docs/requirements.html#po
|
||||
:::
|
||||
|
||||
:::warning
|
||||
|
||||
- Even with matching CPU vendors, in the case of different CPU models, XCP-ng/Citrix Hypervisor will "level" down to use the CPU having the least instructions.
|
||||
- All the hosts in a pool must run the same XCP-ng version.
|
||||
:::
|
||||
:::
|
||||
|
||||
### Creating a pool
|
||||
|
||||
@@ -709,9 +738,10 @@ To remove one host from a pool, you can go to the "Advanced" tab of the host pag
|
||||
|
||||
|
||||
:::warning
|
||||
|
||||
- Detaching a host will remove all the VM disks stored on the Local Storage of this host, and reboot the host.
|
||||
- The host you want to remove must be a slave, not the master!
|
||||
:::
|
||||
:::
|
||||
|
||||
### Network bonding
|
||||
|
||||
@@ -725,19 +755,19 @@ The primary goals are to improve redundancy —so that a single cable or interfa
|
||||
|
||||
Xen Orchestra supports the following bond types:
|
||||
|
||||
- **Active / Active (Balance-SLB)**
|
||||
- **Active / Active (Balance-SLB)**
|
||||
- Does **not** require switch awareness (no LACP needed)
|
||||
- Balances traffic across all bond members
|
||||
- Automatically shifts traffic to remaining members if one link fails
|
||||
- May have compatibility issues with some low-end switches
|
||||
- Unlike LACP, Active/Active bonds can span ports on different switches
|
||||
|
||||
- **Active / Backup**
|
||||
- **Active / Backup**
|
||||
- Does **not** require switch awareness
|
||||
- Uses one primary interface for traffic, switching to backup only upon link failure
|
||||
- Unlike LACP, Active/Backup bonds can span ports on different switches
|
||||
|
||||
- **LACP**
|
||||
- **LACP**
|
||||
- Requires the switch to support and be configured for LACP
|
||||
- Offers rapid failover and can detect mid-span link failures even if the physical link stays up
|
||||
- Balances traffic across all bond members during normal operations
|
||||
@@ -745,33 +775,34 @@ Xen Orchestra supports the following bond types:
|
||||
|
||||
#### Creating a LACP bond
|
||||
|
||||
1. Navigate to **New → Network** to open the **Create Network** page.
|
||||
1. Navigate to **New → Network** to open the **Create Network** page.
|
||||
2. Choose the pool where you want the bond to be created.
|
||||
3. Enable the **Bonded network** toggle:
|
||||
|
||||
4. In the **Interface** dropdown, select two or more physical interfaces (PIFs) to include in the bond.
|
||||
|
||||
4. In the **Interface** dropdown, select two or more physical interfaces (PIFs) to include in the bond.
|
||||
5. Provide a **name** and a **description** for the bond.
|
||||
6. Set the **Bond mode** to **LACP**.
|
||||
7. Leave the **MTU** field blank. This will work as-is in most cases.
|
||||
8. Click **Create network**.
|
||||
XO will then create the bond on the host(s):
|
||||
|
||||
XO will then create the bond on the host(s):
|
||||
|
||||
|
||||
:::tip
|
||||
|
||||
- If applied to a pool, the bond is created across all pool members, so make sure all your pool members are cabled and configured on the switch side for LACP!
|
||||
- If the host’s management interface resides on one of the selected interfaces, XO will **automatically migrate the management interface** on top of the new bond (e.g., if management was on `eth0` and the bond consists of `eth0` + `eth1`).
|
||||
:::
|
||||
:::
|
||||
|
||||
#### Adding VLANs on top of bonds
|
||||
|
||||
Once a bond (e.g., `bond0`) is created, you can build VLANs over it:
|
||||
|
||||
1. Go again to **New → Network** screen.
|
||||
2. Select the new bond interface (e.g., `bond0`) in the **Interface** dropdown.
|
||||
1. Go again to **New → Network** screen.
|
||||
2. Select the new bond interface (e.g., `bond0`) in the **Interface** dropdown.
|
||||
3. Provide a **name**, **description**, and specify the **VLAN** ID (for instance, `20`):
|
||||
|
||||
|
||||
4. Validate the VLAN network creation.\
|
||||
This creates a virtual network on top of the bond, using the name and ID you've specified, available for VM attachment.
|
||||
This creates a virtual network on top of the bond, using the name and ID you've specified, available for VM attachment.
|
||||
|
||||
:::tip
|
||||
This process is the same whether you're adding a VLAN on top of a LACP network bond, or a bond in another mode (such as Active/Active or Active/Backup).
|
||||
@@ -895,21 +926,20 @@ To receive reports, you first need to enable the **usage-report** plugin:
|
||||
|
||||
1. Go to **Settings → Plugin**. A list of plugins will appear.
|
||||
2. From the list, look for the **usage-report** plugin (you can scroll through the list or search for it by name).
|
||||
3. Enable the plugin by switching on the toggle next to the plugin name.
|
||||
3. Enable the plugin by switching on the toggle next to the plugin name.
|
||||
4. Click the **+** icon to open the plugin settings:
|
||||
|
||||
|
||||
5. In the **emails** section, click **Add**. A new text field will appear.
|
||||
6. Enter the email address where you want to receive the reports. Repeat steps 5–6 to add more addresses.
|
||||
7. Turn on the **all** toggle to include stats for all resources in your report, including:
|
||||
- VMs
|
||||
- Hosts
|
||||
- Storage repositories
|
||||
6. Enter the email address where you want to receive the reports. Repeat steps 5–6 to add more addresses.
|
||||
7. Turn on the **all** toggle to include stats for all resources in your report, including:
|
||||
- VMs
|
||||
- Hosts
|
||||
- Storage repositories
|
||||
|
||||
:::tip
|
||||
To include these stats in your reports, the plugin must have saved them at least once.
|
||||
:::
|
||||
8. In the **periodicity** dropdown, choose whether you want to receive reports **daily**, **weekly**, or **monthly**.
|
||||
9. Click **Save configuration**. Reports will be sent to the email addresses you entered at the frequency you selected.
|
||||
::: 8. In the **periodicity** dropdown, choose whether you want to receive reports **daily**, **weekly**, or **monthly**.
|
||||
9. Click **Save configuration**. Reports will be sent to the email addresses you entered at the frequency you selected.
|
||||
|
||||
## Software RAID
|
||||
|
||||
|
||||
@@ -37,6 +37,22 @@ verboseApiLogsOnErrors = false
|
||||
# if no events could be fetched during this delay, the server will be marked as disconnected
|
||||
xapiMarkDisconnectedDelay = '5 minutes'
|
||||
|
||||
# Observability of rolling pool updates/reboots (RPU/RPR).
|
||||
#
|
||||
# Each run writes two files in the traces directory:
|
||||
# - a NDJSON trace of all the events of the task tree, crash-resistant
|
||||
# - a heartbeat file updated every few seconds while the operation is running
|
||||
#
|
||||
#[rpu]
|
||||
#
|
||||
# Directory where the traces are written
|
||||
#
|
||||
# Default: '<datadir>/rpu-traces'
|
||||
#tracesDir = '/var/lib/xo-server/data/rpu-traces'
|
||||
#
|
||||
# How long the traces are kept on disk
|
||||
#tracesRetention = '31 days'
|
||||
|
||||
# https:#github.com/websockets/ws#websocket-compression
|
||||
[apiWebSocketOptions]
|
||||
perMessageDeflate = { threshold = 524288 } # 512kiB
|
||||
|
||||
@@ -1,3 +1,65 @@
|
||||
## On-disk traces
|
||||
|
||||
Each RPU/RPR run writes two files in `<datadir>/rpu-traces` (`/var/lib/xo-server/data/rpu-traces` on XOA, `rpu.tracesDir` to override):
|
||||
|
||||
```
|
||||
rpu-<poolId>-<ISO timestamp>.ndjson # the trace
|
||||
rpu-<poolId>-<ISO timestamp>.heartbeat.json # the liveness file
|
||||
```
|
||||
|
||||
The trace is a tee of the task tree below: every `start`, `property` and `end` event, serialized one JSON object per line, linked by `id`/`parentId`. Writes are synchronous, so nothing is lost on SIGKILL and the file is parsable line by line after any crash (a truncated last line is possible, skip it). The `end` event of a failed task carries the full serialized error, XAPI traceback included.
|
||||
|
||||
The heartbeat is rewritten every 5 seconds while the run is alive:
|
||||
|
||||
| Heartbeat content | Meaning |
|
||||
| ------------------------------------------------- | ----------------------------------------------------------------------------- |
|
||||
| `{"lastUpdated": <fresh>, "status": "pending"}` | run in progress |
|
||||
| `{"lastUpdated": <stale>, "status": "pending"}` | run died mid-flight, xo-server not restarted yet |
|
||||
| `{"status": "interrupted", "lastAlive": ...}` | xo-server restarted and reconciled; `lastAlive` = last heartbeat before death |
|
||||
| `{"status": "success"}` / `{"status": "failure"}` | run finished, file frozen |
|
||||
|
||||
Old traces are garbage-collected on mtime (`rpu.tracesRetention`, 31 days by default), at startup and every 6 hours. Traces of running operations are never collected.
|
||||
|
||||
### Diagnosing a run (support)
|
||||
|
||||
1. Find the trace. Its path is logged at start and at reconciliation:
|
||||
|
||||
```
|
||||
journalctl -u xo-server | grep 'trace in'
|
||||
```
|
||||
|
||||
It's also the `traceFile` property on the root task.
|
||||
|
||||
2. Check the heartbeat first. It answers "is it dead or just slow?" in one read. During `Installing patches` the task tree can stay silent for 20 minutes, that's normal as long as the heartbeat moves.
|
||||
|
||||
3. Read the trace itself. The last line is the last thing that happened before the crash:
|
||||
|
||||
```bash
|
||||
tail -1 <trace>.ndjson
|
||||
```
|
||||
|
||||
To list the steps in the order they were reached:
|
||||
|
||||
```bash
|
||||
grep '"type":"start"' <trace>.ndjson | grep -o '"name":"[^"]*"'
|
||||
```
|
||||
|
||||
A step that shows up here but has no matching `"type":"end"` line was still running when the run died. The interrupted task in the XO tasks view shows the same thing per subtask.
|
||||
|
||||
4. For a `failure`, the root `end` event has the error. `journalctl` is not reliable for this: the UI polls `listMissingPatches` during the run and produces the exact same `updater plugin is busy` stack traces as a real failure.
|
||||
|
||||
### Failure signatures seen in the field
|
||||
|
||||
| Error | What it is |
|
||||
| ------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `The updater plugin is busy (current operation: update)` on `Listing missing patches` | Race between the LINSTOR update step and `check_update`: the updater lock on the host is not released yet. Not destructive, happens before anything is installed. Relaunch. |
|
||||
| `VM_LACKS_FEATURE` with OpaqueRefs, on `Updating and rebooting` | A running VM has no PV drivers. `assert_can_evacuate` runs on all hosts before anything starts, so the RPU refuses upfront. Shut the VM down or install guest tools. |
|
||||
| `MESSAGE_PARAMETER_COUNT_MISMATCH(host.evacuate, 1, 3)` (WARN) | Signature fallback on XAPI 8.2. Harmless. |
|
||||
| Timeout on `Waiting for host to be up` | Host takes too long to boot. `xapiOptions.restartHostTimeout` (default 20 minutes). |
|
||||
| Pool stays `disconnected` after the master rebooted, `EHOSTUNREACH` | Stale connection error, the retry did not kick in yet. `POST /rest/v0/servers/<id>/actions/connect` reconnects immediately. |
|
||||
|
||||
Note on granularity: `Evacuate` is a single `host.evacuate` XAPI call, there is no per-VM detail in the tree for that phase. Per-VM subtasks only exist in `Migrate VMs back`.
|
||||
|
||||
## Task logs
|
||||
|
||||
Rolling pool update and rolling pool reboot task logs have major parts in common.
|
||||
|
||||
19
packages/xo-server/src/_rpuGuard.mjs
Normal file
19
packages/xo-server/src/_rpuGuard.mjs
Normal file
@@ -0,0 +1,19 @@
|
||||
import { forbiddenOperation } from 'xo-common/api-errors.js'
|
||||
|
||||
const runningPools = new Set()
|
||||
|
||||
/**
|
||||
* Ensures a single rolling pool update/reboot at a time per pool.
|
||||
*
|
||||
* @param {string} poolId - Identifier of the pool to lock
|
||||
* @param {string} operation - Name of the calling operation, used in the error
|
||||
* @returns {() => void} Release function, to be called in a `finally`
|
||||
* @throws {Error} `forbiddenOperation` if an RPU/RPR is already running on this pool
|
||||
*/
|
||||
export function acquireRpuGuard(poolId, operation) {
|
||||
if (runningPools.has(poolId)) {
|
||||
throw forbiddenOperation(operation, `a rolling pool update or reboot is already running on pool ${poolId}`)
|
||||
}
|
||||
runningPools.add(poolId)
|
||||
return () => runningPools.delete(poolId)
|
||||
}
|
||||
22
packages/xo-server/src/_rpuGuard.test.mjs
Normal file
22
packages/xo-server/src/_rpuGuard.test.mjs
Normal file
@@ -0,0 +1,22 @@
|
||||
import assert from 'assert/strict'
|
||||
import test from 'node:test'
|
||||
|
||||
import { acquireRpuGuard } from './_rpuGuard.mjs'
|
||||
|
||||
const { describe, it } = test
|
||||
|
||||
describe('acquireRpuGuard', function () {
|
||||
it('allows a single RPU per pool and releases the lock', function () {
|
||||
const release = acquireRpuGuard('pool-1', 'rollingPoolUpdate')
|
||||
assert.throws(() => acquireRpuGuard('pool-1', 'rollingPoolReboot'), { code: 5 /* forbiddenOperation */ })
|
||||
release()
|
||||
acquireRpuGuard('pool-1', 'rollingPoolUpdate')()
|
||||
})
|
||||
|
||||
it('is independent between pools', function () {
|
||||
const release1 = acquireRpuGuard('pool-a', 'rollingPoolUpdate')
|
||||
const release2 = acquireRpuGuard('pool-b', 'rollingPoolUpdate')
|
||||
release1()
|
||||
release2()
|
||||
})
|
||||
})
|
||||
263
packages/xo-server/src/_rpuObservability.mjs
Normal file
263
packages/xo-server/src/_rpuObservability.mjs
Normal file
@@ -0,0 +1,263 @@
|
||||
import { closeSync, mkdirSync, openSync, renameSync, rmdirSync, unlinkSync, writeFileSync, writeSync } from 'node:fs'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
import { join } from 'node:path'
|
||||
import { readdir, readFile, stat, unlink, writeFile } from 'node:fs/promises'
|
||||
import { serializeError } from '@vates/task'
|
||||
import stringify from 'json-stringify-safe'
|
||||
|
||||
const log = createLogger('xo:rpu-observability')
|
||||
|
||||
const DEFAULT_TRACES_RETENTION = 31 * 24 * 60 * 60 * 1000
|
||||
const HEARTBEAT_INTERVAL = 5e3
|
||||
const SENSITIVE_KEY_RE = /password|token|secret|credential|authorization|session|api[-_]?key/i
|
||||
|
||||
const activeTraces = new Set()
|
||||
|
||||
/**
|
||||
* Resolves the traces directory and retention from the server configuration.
|
||||
*
|
||||
* @param {object} app - The xo-server instance (uses `app.config`)
|
||||
* @returns {{ dir: string, retention: number }} Traces directory and retention in milliseconds
|
||||
*/
|
||||
export function getRpuTracesConfig(app) {
|
||||
return {
|
||||
dir: app.config.getOptional('rpu.tracesDir') ?? join(app.config.get('datadir'), 'rpu-traces'),
|
||||
retention: app.config.getOptionalDuration('rpu.tracesRetention') ?? DEFAULT_TRACES_RETENTION,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `JSON.stringify` replacer: serializes errors, converts BigInt and scrubs
|
||||
* secret-looking keys since the trace contains debug-level data.
|
||||
*
|
||||
* Use with `json-stringify-safe` which breaks real cycles only: an object
|
||||
* referenced twice is serialized twice, not flagged as circular.
|
||||
*
|
||||
* @param {string} key
|
||||
* @param {any} value
|
||||
* @returns {any}
|
||||
*/
|
||||
export function replacer(key, value) {
|
||||
if (key !== '' && SENSITIVE_KEY_RE.test(key)) {
|
||||
return '[REDACTED]'
|
||||
}
|
||||
if (typeof value === 'bigint') {
|
||||
return value.toString()
|
||||
}
|
||||
if (value instanceof Error) {
|
||||
return serializeError(value)
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the trace and heartbeat files for one RPU/RPR run.
|
||||
*
|
||||
* @param {object} params
|
||||
* @param {string} params.dir - Directory where the files are created
|
||||
* @param {'rpu'|'rpr'} params.kind - Kind of operation, used as file name prefix
|
||||
* @param {string} params.poolId - Identifier of the target pool
|
||||
* @returns {{ traceFile: string, heartbeatFile: string, attach: (task: object) => void, stop: () => void } | undefined}
|
||||
* `undefined` if the files cannot be created (observability is optional)
|
||||
*/
|
||||
export function openRpuTrace({ dir, kind, poolId }) {
|
||||
let base, fd, heartbeatFile, traceFile
|
||||
try {
|
||||
const timestamp = new Date().toISOString().replace(/[-:.]/g, '')
|
||||
base = `${kind}-${String(poolId).replace(/[^a-zA-Z0-9-]/g, '')}-${timestamp}`
|
||||
mkdirSync(dir, { recursive: true, mode: 0o700 })
|
||||
traceFile = join(dir, base + '.ndjson')
|
||||
heartbeatFile = join(dir, base + '.heartbeat.json')
|
||||
fd = openSync(traceFile, 'ax', 0o600)
|
||||
} catch (error) {
|
||||
try {
|
||||
log.warn('could not create RPU trace files, continuing without observability', { error, dir })
|
||||
} catch {}
|
||||
// best effort cleanup of anything created before the failure
|
||||
if (fd !== undefined) {
|
||||
try {
|
||||
closeSync(fd)
|
||||
} catch {}
|
||||
try {
|
||||
unlinkSync(traceFile)
|
||||
} catch {}
|
||||
}
|
||||
try {
|
||||
rmdirSync(dir) // only succeeds if the dir is empty
|
||||
} catch {}
|
||||
return
|
||||
}
|
||||
|
||||
activeTraces.add(base)
|
||||
|
||||
let beat
|
||||
let interval
|
||||
let stopped = false
|
||||
let writeErrorLogged = false
|
||||
const onWriteError = error => {
|
||||
if (!writeErrorLogged) {
|
||||
writeErrorLogged = true
|
||||
try {
|
||||
log.warn('failed to write RPU trace, observability is degraded', { error, traceFile })
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
heartbeatFile,
|
||||
traceFile,
|
||||
|
||||
attach(task) {
|
||||
const inner = task._onProgress
|
||||
task._onProgress = event => {
|
||||
if (!stopped) {
|
||||
try {
|
||||
let sanitized = event
|
||||
if (event.type === 'property' && SENSITIVE_KEY_RE.test(event.name)) {
|
||||
sanitized = { ...event, value: '[REDACTED]' }
|
||||
}
|
||||
writeSync(fd, stringify(sanitized, replacer) + '\n')
|
||||
} catch (error) {
|
||||
onWriteError(error)
|
||||
}
|
||||
}
|
||||
inner(event)
|
||||
}
|
||||
|
||||
beat = () => {
|
||||
try {
|
||||
const tmp = heartbeatFile + '.tmp'
|
||||
writeFileSync(
|
||||
tmp,
|
||||
JSON.stringify({
|
||||
lastUpdated: new Date().toISOString(),
|
||||
status: task.status,
|
||||
...(writeErrorLogged && { degraded: true }),
|
||||
}) + '\n',
|
||||
{ mode: 0o600 }
|
||||
)
|
||||
renameSync(tmp, heartbeatFile)
|
||||
} catch (error) {
|
||||
onWriteError(error)
|
||||
}
|
||||
}
|
||||
beat()
|
||||
interval = setInterval(beat, HEARTBEAT_INTERVAL)
|
||||
interval.unref()
|
||||
|
||||
try {
|
||||
task.set('traceFile', traceFile)
|
||||
} catch {}
|
||||
},
|
||||
|
||||
stop() {
|
||||
if (!stopped) {
|
||||
stopped = true
|
||||
activeTraces.delete(base)
|
||||
clearInterval(interval)
|
||||
if (beat !== undefined) {
|
||||
beat()
|
||||
}
|
||||
try {
|
||||
closeSync(fd)
|
||||
} catch (error) {
|
||||
onWriteError(error)
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconciles the heartbeats of interrupted runs at boot: a heartbeat left
|
||||
* `pending` on disk cannot belong to a running operation anymore since
|
||||
* xo-server just started.
|
||||
*
|
||||
* Never throws: errors are logged and the remaining files are still processed.
|
||||
*
|
||||
* @param {string} dir - Traces directory
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
export async function reconcileRpuTraces(dir) {
|
||||
let names
|
||||
try {
|
||||
names = await readdir(dir)
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
try {
|
||||
log.warn('could not list RPU traces for reconciliation', { error, dir })
|
||||
} catch {}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
for (const name of names) {
|
||||
const isHeartbeat = /^rp[ru]-.*\.heartbeat\.json$/.test(name)
|
||||
const isActive = [...activeTraces].some(base => name.startsWith(base))
|
||||
if (isHeartbeat && !isActive) {
|
||||
const path = join(dir, name)
|
||||
try {
|
||||
const heartbeat = JSON.parse(await readFile(path, 'utf8'))
|
||||
if (heartbeat.status === 'pending') {
|
||||
await writeFile(
|
||||
path,
|
||||
JSON.stringify({
|
||||
lastUpdated: new Date().toISOString(),
|
||||
status: 'interrupted',
|
||||
lastAlive: heartbeat.lastUpdated,
|
||||
}) + '\n',
|
||||
{ mode: 0o600 }
|
||||
)
|
||||
const traceFile = join(dir, name.replace(/\.heartbeat\.json$/, '.ndjson'))
|
||||
log.info(`interrupted RPU/RPR detected (last alive ${heartbeat.lastUpdated}): trace in ${traceFile}`)
|
||||
}
|
||||
} catch (error) {
|
||||
try {
|
||||
log.warn('could not reconcile RPU heartbeat', { error, path })
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Deletes the trace/heartbeat files older than the retention, skipping the
|
||||
* traces still being written.
|
||||
*
|
||||
* Never throws: errors are logged and the remaining files are still processed.
|
||||
*
|
||||
* @param {string} dir - Traces directory
|
||||
* @param {number} retention - Max age in milliseconds
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
export async function gcRpuTraces(dir, retention) {
|
||||
let names
|
||||
try {
|
||||
names = await readdir(dir)
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
try {
|
||||
log.warn('could not list RPU traces for GC', { error, dir })
|
||||
} catch {}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
const limit = Date.now() - retention
|
||||
for (const name of names) {
|
||||
const isTraceFile = /^rp[ru]-.*\.(ndjson|heartbeat\.json(\.tmp)?)$/.test(name)
|
||||
const isActive = [...activeTraces].some(base => name.startsWith(base))
|
||||
if (isTraceFile && !isActive) {
|
||||
const path = join(dir, name)
|
||||
try {
|
||||
if ((await stat(path)).mtimeMs < limit) {
|
||||
await unlink(path)
|
||||
}
|
||||
} catch (error) {
|
||||
try {
|
||||
log.warn('could not GC RPU trace file', { error, path })
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
283
packages/xo-server/src/_rpuObservability.test.mjs
Normal file
283
packages/xo-server/src/_rpuObservability.test.mjs
Normal file
@@ -0,0 +1,283 @@
|
||||
import assert from 'assert/strict'
|
||||
import test from 'node:test'
|
||||
import stringify from 'json-stringify-safe'
|
||||
import { mkdtemp, readFile, stat, utimes, writeFile } from 'node:fs/promises'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
|
||||
import { gcRpuTraces, openRpuTrace, reconcileRpuTraces, replacer } from './_rpuObservability.mjs'
|
||||
|
||||
const { describe, it, beforeEach } = test
|
||||
|
||||
const makeFakeTask = () => {
|
||||
const events = []
|
||||
return { events, status: 'pending', _onProgress: event => events.push(event) }
|
||||
}
|
||||
|
||||
let dir
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), 'rpu-observability-'))
|
||||
})
|
||||
|
||||
describe('replacer', function () {
|
||||
const serialize = value => stringify(value, replacer)
|
||||
|
||||
it('serializes errors with message, name and stack', function () {
|
||||
const parsed = JSON.parse(serialize({ result: new Error('boom') }))
|
||||
assert.equal(parsed.result.message, 'boom')
|
||||
assert.equal(parsed.result.name, 'Error')
|
||||
assert.ok(parsed.result.stack.includes('boom'))
|
||||
})
|
||||
|
||||
it('converts BigInt to string', function () {
|
||||
assert.equal(serialize({ size: 42n }), '{"size":"42"}')
|
||||
})
|
||||
|
||||
it('breaks cycles', function () {
|
||||
const value = { name: 'a' }
|
||||
value.self = value
|
||||
assert.equal(serialize(value), '{"name":"a","self":"[Circular ~]"}')
|
||||
})
|
||||
|
||||
it('serializes an object referenced twice without flagging it as circular', function () {
|
||||
const shared = { name: 'a' }
|
||||
assert.equal(
|
||||
serialize({ blockedBy: shared, relatedTo: shared }),
|
||||
'{"blockedBy":{"name":"a"},"relatedTo":{"name":"a"}}'
|
||||
)
|
||||
})
|
||||
|
||||
it('scrubs sensitive keys', function () {
|
||||
const parsed = JSON.parse(
|
||||
serialize({
|
||||
apikey: 'S',
|
||||
api_key: 'S',
|
||||
name: 'a',
|
||||
password: 'hunter2',
|
||||
sessionId: 'abc',
|
||||
xsCredentials: { username: 'u' },
|
||||
})
|
||||
)
|
||||
assert.equal(parsed.name, 'a')
|
||||
assert.equal(parsed.apikey, '[REDACTED]')
|
||||
assert.equal(parsed.api_key, '[REDACTED]')
|
||||
assert.equal(parsed.password, '[REDACTED]')
|
||||
assert.equal(parsed.sessionId, '[REDACTED]')
|
||||
assert.equal(parsed.xsCredentials, '[REDACTED]')
|
||||
})
|
||||
})
|
||||
|
||||
describe('openRpuTrace', function () {
|
||||
it('tees events to a NDJSON trace and keeps calling the inner onProgress', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
task._onProgress({ type: 'start', id: 't1' })
|
||||
task._onProgress({ type: 'end', id: 't1', status: 'success' })
|
||||
trace.stop()
|
||||
|
||||
const lines = (await readFile(trace.traceFile, 'utf8')).trim().split('\n')
|
||||
assert.deepEqual(lines.map(JSON.parse), [
|
||||
{ type: 'start', id: 't1' },
|
||||
{ type: 'end', id: 't1', status: 'success' },
|
||||
])
|
||||
assert.equal(task.events.length, 2)
|
||||
})
|
||||
|
||||
it('scrubs sensitive property events without altering what the task machinery receives', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
const secret = { apikey: 'SECRET' }
|
||||
task._onProgress({ type: 'property', name: 'xsCredentials', value: secret, id: 't1' })
|
||||
trace.stop()
|
||||
|
||||
const content = await readFile(trace.traceFile, 'utf8')
|
||||
assert.ok(!content.includes('SECRET'))
|
||||
assert.ok(content.includes('[REDACTED]'))
|
||||
// the inner onProgress still receives the original, untouched event
|
||||
assert.equal(task.events[0].value, secret)
|
||||
})
|
||||
|
||||
it('writes a heartbeat with lastUpdated and status, updated on stop', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
let heartbeat = JSON.parse(await readFile(trace.heartbeatFile, 'utf8'))
|
||||
assert.equal(heartbeat.status, 'pending')
|
||||
assert.equal(heartbeat.degraded, undefined)
|
||||
assert.ok(!Number.isNaN(Date.parse(heartbeat.lastUpdated)))
|
||||
|
||||
task.status = 'success'
|
||||
trace.stop()
|
||||
heartbeat = JSON.parse(await readFile(trace.heartbeatFile, 'utf8'))
|
||||
assert.equal(heartbeat.status, 'success')
|
||||
})
|
||||
|
||||
it('updates the heartbeat even when a stale tmp file is left over', async function () {
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
await writeFile(trace.heartbeatFile + '.tmp', 'stale')
|
||||
|
||||
trace.attach(makeFakeTask())
|
||||
trace.stop()
|
||||
|
||||
const heartbeat = JSON.parse(await readFile(trace.heartbeatFile, 'utf8'))
|
||||
assert.equal(heartbeat.status, 'pending')
|
||||
assert.equal(heartbeat.degraded, undefined)
|
||||
})
|
||||
|
||||
it('flags the heartbeat as degraded when trace writes fail', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
// a throwing getter makes the serialization fail: the error must be
|
||||
// swallowed and surfaced through the heartbeat's degraded flag
|
||||
task._onProgress({
|
||||
type: 'info',
|
||||
id: 't1',
|
||||
get data() {
|
||||
throw new Error('boom')
|
||||
},
|
||||
})
|
||||
assert.equal(task.events.length, 1)
|
||||
|
||||
trace.stop()
|
||||
const heartbeat = JSON.parse(await readFile(trace.heartbeatFile, 'utf8'))
|
||||
assert.equal(heartbeat.degraded, true)
|
||||
})
|
||||
|
||||
it('sanitizes the pool id in file names', function () {
|
||||
const trace = openRpuTrace({ dir, kind: 'rpr', poolId: '../../etc/passwd' })
|
||||
assert.ok(trace.traceFile.startsWith(join(dir, 'rpr-etcpasswd-')))
|
||||
trace.stop()
|
||||
})
|
||||
|
||||
it('never throws into the task, even after stop', function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
trace.stop()
|
||||
trace.stop() // idempotent
|
||||
|
||||
// fd is closed: the write is skipped but the event still reaches the
|
||||
// inner onProgress and nothing is thrown
|
||||
task._onProgress({ type: 'info', id: 't1' })
|
||||
assert.equal(task.events.length, 1)
|
||||
})
|
||||
|
||||
it('returns undefined when the trace files cannot be created', async function () {
|
||||
// a path whose parent is a regular file fails with ENOTDIR
|
||||
const file = join(dir, 'not-a-dir')
|
||||
await writeFile(file, '')
|
||||
assert.equal(openRpuTrace({ dir: join(file, 'sub'), kind: 'rpu', poolId: 'pool-1' }), undefined)
|
||||
})
|
||||
|
||||
it('creates the traces directory and files with restrictive permissions', async function () {
|
||||
const nested = join(dir, 'nested')
|
||||
const trace = openRpuTrace({ dir: nested, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(makeFakeTask())
|
||||
trace.stop()
|
||||
assert.equal((await stat(nested)).mode & 0o777, 0o700)
|
||||
assert.equal((await stat(trace.traceFile)).mode & 0o777, 0o600)
|
||||
assert.equal((await stat(trace.heartbeatFile)).mode & 0o777, 0o600)
|
||||
})
|
||||
})
|
||||
|
||||
describe('gcRpuTraces', function () {
|
||||
it('deletes only expired RPU/RPR files', async function () {
|
||||
const old = new Date(Date.now() - 3600e3)
|
||||
const oldTrace = join(dir, 'rpu-pool-1-20260101T000000000Z.ndjson')
|
||||
const oldHeartbeat = join(dir, 'rpr-pool-1-20260101T000000000Z.heartbeat.json')
|
||||
const oldTmp = join(dir, 'rpr-pool-1-20260101T000000000Z.heartbeat.json.tmp')
|
||||
const oldWrongExtension = join(dir, 'rpu-pool-1-20260101T000000000Z.log')
|
||||
const recentTrace = join(dir, 'rpu-pool-2-20260706T000000000Z.ndjson')
|
||||
const unrelated = join(dir, 'unrelated.ndjson')
|
||||
for (const path of [oldTrace, oldHeartbeat, oldTmp, oldWrongExtension, recentTrace, unrelated]) {
|
||||
await writeFile(path, '')
|
||||
if (path !== recentTrace) {
|
||||
await utimes(path, old, old)
|
||||
}
|
||||
}
|
||||
|
||||
await gcRpuTraces(dir, 1800e3)
|
||||
|
||||
assert.equal(existsSync(oldTrace), false)
|
||||
assert.equal(existsSync(oldHeartbeat), false)
|
||||
assert.equal(existsSync(oldTmp), false)
|
||||
assert.equal(existsSync(oldWrongExtension), true)
|
||||
assert.equal(existsSync(recentTrace), true)
|
||||
assert.equal(existsSync(unrelated), true)
|
||||
})
|
||||
|
||||
it('never deletes the files of a trace still being written', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
const old = new Date(Date.now() - 3600e3)
|
||||
await utimes(trace.traceFile, old, old)
|
||||
await utimes(trace.heartbeatFile, old, old)
|
||||
|
||||
await gcRpuTraces(dir, 0)
|
||||
assert.equal(existsSync(trace.traceFile), true)
|
||||
assert.equal(existsSync(trace.heartbeatFile), true)
|
||||
|
||||
trace.stop()
|
||||
await utimes(trace.traceFile, old, old)
|
||||
await utimes(trace.heartbeatFile, old, old)
|
||||
await gcRpuTraces(dir, 1800e3)
|
||||
assert.equal(existsSync(trace.traceFile), false)
|
||||
assert.equal(existsSync(trace.heartbeatFile), false)
|
||||
})
|
||||
|
||||
it('does not throw on a missing directory', async function () {
|
||||
await gcRpuTraces(join(dir, 'does-not-exist'), 0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('reconcileRpuTraces', function () {
|
||||
it('stamps pending heartbeats as interrupted, preserving the time of death', async function () {
|
||||
const path = join(dir, 'rpu-pool-1-20260101T000000000Z.heartbeat.json')
|
||||
await writeFile(path, JSON.stringify({ lastUpdated: '2026-01-01T00:00:00.000Z', status: 'pending' }))
|
||||
|
||||
await reconcileRpuTraces(dir)
|
||||
|
||||
const heartbeat = JSON.parse(await readFile(path, 'utf8'))
|
||||
assert.equal(heartbeat.status, 'interrupted')
|
||||
assert.equal(heartbeat.lastAlive, '2026-01-01T00:00:00.000Z')
|
||||
assert.ok(!Number.isNaN(Date.parse(heartbeat.lastUpdated)))
|
||||
})
|
||||
|
||||
it('leaves finished heartbeats untouched', async function () {
|
||||
const path = join(dir, 'rpu-pool-1-20260101T000000000Z.heartbeat.json')
|
||||
const content = JSON.stringify({ lastUpdated: '2026-01-01T00:00:00.000Z', status: 'failure' })
|
||||
await writeFile(path, content)
|
||||
|
||||
await reconcileRpuTraces(dir)
|
||||
|
||||
assert.equal(await readFile(path, 'utf8'), content)
|
||||
})
|
||||
|
||||
it('skips the heartbeat of a run in progress', async function () {
|
||||
const task = makeFakeTask()
|
||||
const trace = openRpuTrace({ dir, kind: 'rpu', poolId: 'pool-1' })
|
||||
trace.attach(task)
|
||||
|
||||
await reconcileRpuTraces(dir)
|
||||
|
||||
const heartbeat = JSON.parse(await readFile(trace.heartbeatFile, 'utf8'))
|
||||
assert.equal(heartbeat.status, 'pending')
|
||||
trace.stop()
|
||||
})
|
||||
|
||||
it('does not throw on a corrupt heartbeat or a missing directory', async function () {
|
||||
await writeFile(join(dir, 'rpu-pool-1-20260101T000000000Z.heartbeat.json'), '{truncated')
|
||||
await reconcileRpuTraces(dir)
|
||||
await reconcileRpuTraces(join(dir, 'does-not-exist'))
|
||||
})
|
||||
})
|
||||
@@ -424,7 +424,9 @@ const methods = {
|
||||
throw new Error(result.stderr)
|
||||
}
|
||||
|
||||
log.debug(result.stdout)
|
||||
// `stdout` may be undefined depending on the XCP-ng version, always log
|
||||
// a meaningful end-of-install line (this step can last >15 minutes)
|
||||
log.debug(`patches installed on host ${host.uuid}`, { stdout: result.stdout })
|
||||
await host.update_other_config('rpm_patch_installation_time', String(Date.now() / 1000))
|
||||
}
|
||||
},
|
||||
|
||||
@@ -7,8 +7,15 @@ import some from 'lodash/some.js'
|
||||
import stubTrue from 'lodash/stubTrue.js'
|
||||
import uniq from 'lodash/uniq.js'
|
||||
import { asyncEach } from '@vates/async-each'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
import { decorateMethodsWith } from '@vates/decorate-with'
|
||||
import { defer } from 'golike-defer'
|
||||
import { Task } from '@vates/task'
|
||||
|
||||
import { acquireRpuGuard } from '../_rpuGuard.mjs'
|
||||
import { gcRpuTraces, getRpuTracesConfig, openRpuTrace, reconcileRpuTraces } from '../_rpuObservability.mjs'
|
||||
|
||||
const log = createLogger('xo:xo-mixins:pool')
|
||||
|
||||
async function enforceHostsHaveLicense($defer, app, productType, hostIds) {
|
||||
const now = Date.now()
|
||||
@@ -53,6 +60,19 @@ async function enforceHostsHaveLicense($defer, app, productType, hostIds) {
|
||||
export default class Pools {
|
||||
constructor(app) {
|
||||
this._app = app
|
||||
|
||||
const gc = () => {
|
||||
const { dir, retention } = getRpuTracesConfig(app)
|
||||
return gcRpuTraces(dir, retention)
|
||||
}
|
||||
app.hooks.on('clean', gc)
|
||||
// the clean hook only runs at startup and on manual xo.clean: also enforce
|
||||
// the retention periodically (same pattern as xo-mixins/logs)
|
||||
setInterval(gc, 6 * 60 * 60 * 1000).unref()
|
||||
|
||||
// a heartbeat left pending on disk after a restart belongs to an
|
||||
// interrupted run: stamp it so the disk alone is unambiguous
|
||||
app.hooks.on('start', () => reconcileRpuTraces(getRpuTracesConfig(app).dir))
|
||||
}
|
||||
|
||||
async mergeInto($defer, { sources: sourceIds, target, force }) {
|
||||
@@ -188,20 +208,28 @@ export default class Pools {
|
||||
async rollingPoolReboot(pool, { parentTask } = {}) {
|
||||
const { _app } = this
|
||||
await _app.checkFeatureAuthorization('ROLLING_POOL_REBOOT')
|
||||
const hasParentTask = parentTask !== undefined
|
||||
let task = parentTask
|
||||
const fn = async () => _app.getXapi(pool).rollingPoolReboot(task)
|
||||
const releaseGuard = acquireRpuGuard(pool.id, 'rollingPoolReboot')
|
||||
const trace = openRpuTrace({ dir: getRpuTracesConfig(_app).dir, kind: 'rpr', poolId: pool.id })
|
||||
try {
|
||||
if (trace !== undefined) {
|
||||
log.info(`rolling pool reboot of pool ${pool.id}: trace in ${trace.traceFile}`)
|
||||
}
|
||||
|
||||
if (!hasParentTask) {
|
||||
task = _app.tasks.create({
|
||||
name: `Rolling pool reboot`,
|
||||
const properties = {
|
||||
name: 'Rolling pool reboot',
|
||||
objectId: pool.id,
|
||||
poolId: pool.id,
|
||||
poolName: pool.name_label,
|
||||
progress: 0,
|
||||
})
|
||||
await task.run(fn)
|
||||
} else {
|
||||
await fn()
|
||||
type: 'pool.rolling_reboot',
|
||||
...(trace !== undefined && { traceFile: trace.traceFile }),
|
||||
}
|
||||
const task = parentTask === undefined ? _app.tasks.create(properties) : new Task({ properties })
|
||||
trace?.attach(task)
|
||||
await task.run(async () => _app.getXapi(pool).rollingPoolReboot(task))
|
||||
} finally {
|
||||
trace?.stop()
|
||||
releaseGuard()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,9 +13,12 @@ import { networkInterfaces } from 'os'
|
||||
import { noSuchObject, incorrectState } from 'xo-common/api-errors.js'
|
||||
import { parseDuration } from '@vates/parse-duration'
|
||||
import { pDelay, ignoreErrors } from 'promise-toolbox'
|
||||
import { Task } from '@vates/task'
|
||||
|
||||
import * as XenStore from '../_XenStore.mjs'
|
||||
import Xapi from '../xapi/index.mjs'
|
||||
import { acquireRpuGuard } from '../_rpuGuard.mjs'
|
||||
import { getRpuTracesConfig, openRpuTrace } from '../_rpuObservability.mjs'
|
||||
import xapiObjectToXo from '../xapi-object-to-xo.mjs'
|
||||
import XapiStats from '../xapi-stats.mjs'
|
||||
import { camelToSnakeCase, forEach, isEmpty, popProperty } from '../utils.mjs'
|
||||
@@ -753,6 +756,8 @@ export default class XenServers {
|
||||
|
||||
const poolId = pool.id
|
||||
|
||||
$defer(acquireRpuGuard(poolId, 'rollingPoolUpdate'))
|
||||
|
||||
const jobsOfthePool = []
|
||||
jobs.forEach(({ id: jobId, vms }) => {
|
||||
if (vms.id !== undefined) {
|
||||
@@ -799,25 +804,29 @@ export default class XenServers {
|
||||
$defer(() => xapi.call('pool.set_wlb_enabled', pool._xapiRef, true))
|
||||
}
|
||||
|
||||
const hasParentTask = parentTask !== undefined
|
||||
let task = parentTask
|
||||
const fn = async () =>
|
||||
const trace = openRpuTrace({ dir: getRpuTracesConfig(app).dir, kind: 'rpu', poolId })
|
||||
$defer(() => trace?.stop())
|
||||
if (trace !== undefined) {
|
||||
log.info(`rolling pool update of pool ${poolId}: trace in ${trace.traceFile}`)
|
||||
}
|
||||
|
||||
const properties = {
|
||||
name: 'Rolling pool update',
|
||||
objectId: poolId,
|
||||
poolId,
|
||||
poolName: pool.name_label,
|
||||
progress: 0,
|
||||
type: 'pool.rolling_update',
|
||||
...(trace !== undefined && { traceFile: trace.traceFile }),
|
||||
}
|
||||
const task = parentTask === undefined ? app.tasks.create(properties) : new Task({ properties })
|
||||
trace?.attach(task)
|
||||
await task.run(async () =>
|
||||
this.getXapi(pool).rollingPoolUpdate(task, {
|
||||
xsCredentials: app.apiContext.user.preferences.xsCredentials,
|
||||
rebootVm,
|
||||
})
|
||||
|
||||
if (!hasParentTask) {
|
||||
task = app.tasks.create({
|
||||
name: `Rolling pool update`,
|
||||
poolId,
|
||||
poolName: pool.name_label,
|
||||
progress: 0,
|
||||
})
|
||||
await task.run(fn)
|
||||
} else {
|
||||
await fn()
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user