Record reconcile work in Home Assistant and calculate dashboard estimates from history

This commit is contained in:
feedc0de committed 2026-11-02 11:43:13 +01:00
1 parent c27f64dc3d
commit a515828879
7 files changed
+368 -1

No files matched your search

+68
View File
@@ -192,3 +192,71 @@ make
The collector listens on localhost:8000 and launches `bcachefs fs top /mnt`
and `sudo perf trace`. Building does not deploy or restart it.
## Reconcile history dashboard
The MQTT bridge also discovers 16 pending-work sensors:
`sensor.bcachefs_reconcile_{replicas,checksum,erasure_code,compression,target,high_priority,pending,stripes}_{data,metadata}`.
It runs the existing `bcachefs_reconcile.sh` collector every 30 seconds, reusing
Telegraf's parser and publishing byte counts only. No speed or ETA sensors are
created. Missing fields are unknown rather than fabricated zeros. Collection
failures/unmounts make measurements unavailable, and stale sensors expire after
90 seconds. The existing switch controls are unchanged. Set `FSPATH` in the
service environment for a mount other than `/mnt`.
These sensors use `state_class: measurement` and `force_update: true`: Recorder
stores sampling timestamps even when bcachefs's rounded values stay unchanged.
This matters for flat periods and pauses. Keep at least seven days of Recorder
history and do not exclude these entities if using the seven-day calculation.
History begins when the sensors are installed; no InfluxDB data is imported.
`homeassistant/bcachefs-reconcile-card.mjs` is a custom dashboard card using the
already-installed ApexCharts Card for graphs and Home Assistant's authenticated
history WebSocket API for estimates. The host's existing
`/dashboard-bcachefs/0` dashboard includes this card alongside its switches.
- One selector applies 3h, 6h, 12h, 24h, 2d or 7d to both graphs and calculations.
- Another selector chooses `target_data` or `erasure_code_data` for the estimate.
- Average net progress = (first pending bytes − last pending bytes) / elapsed
seconds between those actual samples in the selected window.
- ETA = last pending bytes / positive average progress. Flat or growing work
has no completion estimate; zero pending bytes is complete.
- Estimates require at least one minute of valid history, show actual coverage
for partial windows, and flag unavailable periods. Wall-clock elapsed time
includes pauses. This is an extrapolation, not a prediction of future load.
- Charts aggregate for display; estimates use raw recorded endpoints, not graph
averages. No artificial state from before the window is used as an endpoint.
- Both charts and estimates refresh every minute. Selecting a range recalculates
immediately, and stale asynchronous results cannot replace a newer selection.
To install or update the card on another Home Assistant instance:
1. Install ApexCharts Card (the original host already has it through HACS).
2. Copy `homeassistant/bcachefs-reconcile-card.mjs` into
`/config/www/bcachefs/bcachefs-reconcile-card.mjs`.
3. Register `/local/bcachefs/bcachefs-reconcile-card.mjs?v=VERSION` as a JavaScript
module resource. Change `VERSION` after edits to invalidate browser caches.
4. Add this card in the dashboard's YAML editor (sensor IDs can be adjusted with
`entity_prefix`):
```yaml
type: custom:bcachefs-reconcile-card
entity_prefix: sensor.bcachefs_reconcile_
hours: 6
grid_options:
columns: full
rows: auto
```
The deployed dashboard was updated through Home Assistant's WebSocket API,
preserving its existing cards. Before each update its configuration was backed
up under `/config/bcachefs-dashboard-backups/`. No Home Assistant restart or
Grafana changes were required. The source for the card and collector lives in
this repository; Kubernetes continues to own application deployment only.
Additional checks:
```sh
node tests/test_reconcile_history.mjs
.venv/bin/python -m unittest discover -s tests -p 'test_reconcile_metrics.py' -v
```
+45
View File
@@ -7,6 +7,10 @@ from pathlib import Path
import signal
import socket
import threading
import time
import subprocess
from reconcile_metrics import FIELDS, read_metrics
import uuid as uuid_module
import paho.mqtt.client as mqtt
@@ -23,6 +27,7 @@ class Bridge:
self.discovery = discovery
self.lock = threading.RLock()
self.last = {}
self.reconcile_refresh = threading.Event()
client.on_connect = self.on_connect
client.on_message = self.on_message
client.will_set(f'{self.base}/availability', 'offline', qos=1, retain=True)
@@ -48,8 +53,40 @@ class Bridge:
}
self.publish(f'{self.discovery}/switch/bcachefs_{self.uuid}/{option}/config', json.dumps(config))
for field in FIELDS:
config = {
'name': 'Reconcile ' + field.replace('_', ' '),
'unique_id': f'bcachefs_{self.uuid}_reconcile_{field}',
'default_entity_id': f'sensor.bcachefs_reconcile_{field}',
'state_topic': f'{self.base}/reconcile/state',
'value_template': '{{ value_json.' + field + ' | default(none) }}',
'device_class': 'data_size', 'state_class': 'measurement',
'unit_of_measurement': 'B', 'suggested_display_precision': 0,
# Keep sampling timestamps even while rounded byte counts are flat.
'force_update': True, 'expire_after': 90,
'availability_mode': 'all',
'availability': [{'topic': f'{self.base}/availability'},
{'topic': f'{self.base}/reconcile/availability'}],
'device': {'identifiers': [f'bcachefs_{self.uuid}']},
}
self.publish(f'{self.discovery}/sensor/bcachefs_{self.uuid}/reconcile_{field}/config', json.dumps(config))
def sample_reconcile(self, mount):
try:
if not self.options.is_dir():
raise OSError('Filesystem not mounted')
values = read_metrics(self.uuid, mount)
except (OSError, ValueError, subprocess.SubprocessError) as error:
logging.warning('Reconcile measurements unavailable: %s', error)
self.publish(f'{self.base}/reconcile/availability', 'offline')
return
self.publish(f'{self.base}/reconcile/state', json.dumps(values))
self.publish(f'{self.base}/reconcile/availability', 'online')
def sample(self, force=False):
with self.lock:
if not self.options.is_dir():
self.publish(f'{self.base}/reconcile/availability', 'offline')
for option in OPTIONS:
topic = f'{self.base}/{option}'
try:
@@ -70,6 +107,8 @@ class Bridge:
return
with self.lock:
self.publish(f'{self.base}/availability', 'offline')
self.publish(f'{self.base}/reconcile/availability', 'offline')
self.reconcile_refresh.set()
self.announce()
for option in OPTIONS:
client.subscribe(f'{self.base}/{option}/set', qos=1)
@@ -84,6 +123,7 @@ class Bridge:
if message.payload == b'online':
self.announce()
self.sample(force=True)
self.reconcile_refresh.set()
return
option = next((name for name in OPTIONS if message.topic == f'{self.base}/{name}/set'), None)
if option is None or message.retain or message.payload not in (b'0', b'1'):
@@ -118,10 +158,15 @@ def main():
signal.signal(sig, lambda *_: stopped.set())
client.connect_async(os.environ['MQTT_HOST'], int(os.environ.get('MQTT_PORT', '1883')), keepalive=15)
client.loop_start()
next_reconcile = 0.0
try:
while not stopped.wait(interval):
if client.is_connected():
bridge.sample()
if bridge.reconcile_refresh.is_set() or time.monotonic() >= next_reconcile:
bridge.reconcile_refresh.clear()
bridge.sample_reconcile(os.environ.get('FSPATH', '/mnt'))
next_reconcile = time.monotonic() + 30
finally:
if client.is_connected():
bridge.publish(f'{bridge.base}/availability', 'offline').wait_for_publish(timeout=3)
+161
View File
@@ -0,0 +1,161 @@
const RANGES = [[3, 'Last 3 hours'], [6, 'Last 6 hours'], [12, 'Last 12 hours'], [24, 'Last 24 hours'], [48, 'Last 2 days'], [168, 'Last 7 days']];
const TYPES = ['replicas', 'checksum', 'erasure_code', 'compression', 'target', 'high_priority', 'pending', 'stripes'];
const LABELS = {replicas: 'Replicas', checksum: 'Checksum', erasure_code: 'Erasure code', compression: 'Compression', target: 'Target', high_priority: 'High priority', pending: 'Pending', stripes: 'Stripes'};
const UNITS = {B: 1, kB: 1e3, MB: 1e6, GB: 1e9, TB: 1e12, KiB: 1024, MiB: 1024 ** 2, GiB: 1024 ** 3, TiB: 1024 ** 4};
// Endpoints are real Recorder samples within the selected window, not a fixed rate sensor.
export function estimate(history, start, end, unit = 'B') {
const factor = UNITS[unit];
if (!factor) return {status: 'Unsupported sensor unit'};
const points = history.map(row => {
const raw = row.s ?? row.state;
const stamp = row.lu ?? row.last_updated ?? row.lc ?? row.last_changed;
const time = typeof stamp === 'number' ? stamp * 1000 : Date.parse(stamp);
const value = raw !== null && raw !== undefined && String(raw).trim() !== '' ? Number(raw) * factor : NaN;
return {time, value};
}).filter(p => Number.isFinite(p.time) && p.time >= start && p.time <= end).sort((a, b) => a.time - b.time);
if (!points.length) return {status: 'Waiting for recorded history'};
const last = points.at(-1);
if (!Number.isFinite(last.value) || last.value < 0 || end - last.time > 120000) return {status: 'Measurements unavailable or stale'};
const valid = points.filter(p => Number.isFinite(p.value) && p.value >= 0);
const first = valid[0];
const seconds = (last.time - first.time) / 1000;
if (valid.length < 2 || seconds < 60) return {status: 'Collecting history (at least 1 minute)', remaining: last.value};
const rate = (first.value - last.value) / seconds;
return {status: 'ok', remaining: last.value, rate, eta: last.value === 0 ? 0 : rate > 0 ? last.value / rate : null,
seconds, samples: valid.length, gap: points.length !== valid.length,
coverage: seconds / ((end - start) / 1000), first: first.time, last: last.time};
}
function bytes(value) {
if (!Number.isFinite(value)) return '—';
const abs = Math.abs(value);
const index = Math.min(4, Math.max(0, Math.floor(Math.log2(abs || 1) / 10)));
return `${(value / 1024 ** index).toLocaleString(undefined, {maximumFractionDigits: 2})} ${['B', 'KiB', 'MiB', 'GiB', 'TiB'][index]}`;
}
function duration(seconds) {
if (!Number.isFinite(seconds)) return 'No estimate';
if (seconds === 0) return 'Complete';
if (seconds < 60) return '< 1 min';
const minutes = Math.round(seconds / 60);
return minutes >= 1440 ? `${Math.floor(minutes / 1440)}d ${Math.floor(minutes % 1440 / 60)}h` : minutes >= 60 ? `${Math.floor(minutes / 60)}h ${minutes % 60}m` : `${minutes}m`;
}
class BcachefsReconcileCard extends (globalThis.HTMLElement || class {}) {
constructor() {
super();
this.attachShadow({mode: 'open'});
this._hours = 6;
this._field = 'target_data';
this._request = 0;
this._graphs = [];
}
setConfig(config) {
this._config = {entity_prefix: 'sensor.bcachefs_reconcile_', ...config};
this._hours = RANGES.some(([hours]) => hours === config.hours) ? config.hours : this._hours;
this.shadowRoot.innerHTML = `<style>
:host{display:block}ha-card{padding:16px}.controls{display:flex;gap:16px;flex-wrap:wrap}
h2{margin:0 0 16px;font-size:20px;font-weight:500}label{display:flex;flex-direction:column;gap:6px;font-size:13px;color:var(--secondary-text-color)}
select{font:inherit;color:var(--primary-text-color);background:var(--card-background-color);border:1px solid var(--divider-color);border-radius:8px;padding:10px;min-width:160px}
.summary{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:12px;margin:20px 0 12px}.metric{background:var(--secondary-background-color);padding:12px;border-radius:8px}
.metric span{display:block;font-size:12px;color:var(--secondary-text-color)}.metric strong{display:block;margin-top:8px;font-size:22px;font-weight:500;overflow-wrap:anywhere}
.note{font-size:13px;color:var(--secondary-text-color);line-height:1.5;margin:6px 0}.graphs{display:grid;gap:16px;margin-top:16px}
@media(max-width:450px){.summary{grid-template-columns:1fr}.metric strong{font-size:20px}}
</style><ha-card><h2>Pending reconcile work</h2>
<div class="controls"><label>History and calculation window<select id="range"></select></label>
<label>Estimate for<select id="field"><option value="target_data">Target data</option><option value="erasure_code_data">Erasure-code data</option></select></label></div>
<div class="summary"><div class="metric"><span>Remaining data</span><strong id="remaining">—</strong></div>
<div class="metric"><span>Average net progress</span><strong id="rate">—</strong></div>
<div class="metric"><span>Estimated time remaining</span><strong id="eta">—</strong></div></div>
<p id="status" class="note" role="status">Loading Home Assistant history…</p>
<p class="note">Speed = (first − last pending bytes) ÷ elapsed time in the selected window. ETA assumes that average continues. Categories can overlap; they are not added together.</p>
<div class="graphs"></div></ha-card>`;
const select = this.shadowRoot.querySelector('#range');
for (const [value, text] of RANGES) select.add(new Option(text, value));
select.value = String(this._hours);
this.shadowRoot.querySelector('#field').value = this._field;
select.addEventListener('change', () => {this._hours = Number(select.value); this._buildGraphs(); this._load();});
this.shadowRoot.querySelector('#field').addEventListener('change', event => {this._field = event.target.value; this._load();});
this._buildGraphs();
if (this._hass) this._load();
}
set hass(hass) {
const first = !this._hass;
this._hass = hass;
this._graphs.forEach(graph => {graph.hass = hass;});
const entity = hass.states[this._entity()];
if (entity && !Number.isFinite(Number(entity.state))) {
++this._request;
this._show({status: 'Measurements unavailable'});
} else if (first && this._config) this._load();
}
connectedCallback() {
clearInterval(this._timer);
this._timer = setInterval(() => this._load(), 60000);
if (this._hass && this._config) this._load();
}
disconnectedCallback() { clearInterval(this._timer); ++this._request; }
getCardSize() {return 14;}
getGridOptions() {return {columns: 'full', rows: 'auto'};}
_entity() {return `${this._config?.entity_prefix || 'sensor.bcachefs_reconcile_'}${this._field}`;}
async _buildGraphs() {
const generation = this._graphGeneration = (this._graphGeneration || 0) + 1;
try {
const helpers = await window.loadCardHelpers();
if (generation !== this._graphGeneration) return;
const graphs = ['data', 'metadata'].map(part => helpers.createCardElement({
type: 'custom:apexcharts-card', graph_span: `${this._hours}h`, update_interval: '60s',
header: {show: true, title: part === 'data' ? 'Pending data (GiB)' : 'Pending metadata (MiB)'},
apex_config: {chart: {height: 300}, legend: {show: true}, yaxis: {decimalsInFloat: 2}},
all_series_config: {stroke_width: 2, extend_to: false, float_precision: 2,
group_by: {func: 'avg', fill: 'null', duration: `${Math.max(1, Math.ceil(this._hours * 60 / 360))}min`}},
series: TYPES.map(type => ({entity: `${this._config.entity_prefix}${type}_${part}`, name: LABELS[type],
unit: part === 'data' ? 'GiB' : 'MiB', transform: `return x == null || !Number.isFinite(Number(x)) ? null : Number(x) / ${part === 'data' ? 1024 ** 3 : 1024 ** 2};`})),
}));
graphs.forEach(graph => {if (this._hass) graph.hass = this._hass;});
this._graphs = graphs;
this.shadowRoot.querySelector('.graphs').replaceChildren(...graphs);
} catch (error) {
this.shadowRoot.querySelector('.graphs').textContent = `Could not load charts: ${error.message}. ApexCharts Card must be installed.`;
}
}
async _load() {
if (!this._hass || !this._config) return;
const id = ++this._request;
const entityId = this._entity();
const entity = this._hass.states[entityId];
if (!entity || !Number.isFinite(Number(entity.state))) {this._show({status: 'Measurements unavailable'}); return;}
this._show({status: 'Loading selected history…'});
const end = Date.now();
const start = end - this._hours * 3600000;
try {
const history = await this._hass.callWS({type: 'history/history_during_period',
start_time: new Date(start).toISOString(), end_time: new Date(end).toISOString(),
entity_ids: [entityId], include_start_time_state: false, significant_changes_only: false,
minimal_response: false, no_attributes: true});
if (id !== this._request) return;
this._show(estimate(history[entityId] || [], start, end, entity.attributes.unit_of_measurement));
} catch (error) {
if (id === this._request) this._show({status: `History request failed: ${error.message || error.code || error}`});
}
}
_show(result) {
if (!this._config) return;
const root = this.shadowRoot;
root.querySelector('#remaining').textContent = bytes(result.remaining);
root.querySelector('#rate').textContent = Number.isFinite(result.rate) ? `${bytes(result.rate)}/s` : '—';
root.querySelector('#eta').textContent = result.status === 'ok' ? duration(result.eta) : '—';
let note = result.status;
if (result.status === 'ok') {
note = `${result.samples.toLocaleString()} samples over ${duration(result.seconds)} of the selected ${this._hours}h. `;
if (result.coverage < .95) note += 'Partial history: the full selected window is not available yet. ';
if (result.rate < 0) note += 'Pending work increased; no completion estimate. ';
else if (result.rate === 0 && result.remaining > 0) note += 'No net progress during this window; no completion estimate. ';
if (result.gap) note += 'This window includes unavailable readings. ';
if (this._hass?.states['switch.bcachefs_reconcile_enabled']?.state === 'off') note += 'Reconcile is currently paused. ';
note += `Last sample: ${new Date(result.last).toLocaleTimeString()}.`;
}
root.querySelector('#status').textContent = note;
}
}
if (globalThis.customElements && !customElements.get('bcachefs-reconcile-card')) customElements.define('bcachefs-reconcile-card', BcachefsReconcileCard);
+32
View File
@@ -0,0 +1,32 @@
"""Reuse the Telegraf collector's parsing; MQTT transports measurements only."""
import os
from pathlib import Path
import subprocess
TYPES = ('replicas', 'checksum', 'erasure_code', 'compression', 'target', 'high_priority', 'pending', 'stripes')
FIELDS = tuple(f'{kind}_{part}' for part in ('data', 'metadata') for kind in TYPES)
def parse_metrics(line):
lines = line.strip().splitlines()
if len(lines) != 1 or not lines[0].startswith('bcachefs_reconcile,'):
raise ValueError('Expected one reconcile measurement')
values = {}
for field in lines[0].rsplit(' ', 1)[1].split(','):
key, value = field.split('=', 1)
if key in FIELDS:
if not value.isdecimal():
raise ValueError(f'Invalid byte count for {key}')
values[key] = int(value)
if not values:
raise ValueError('No reconcile byte measurements')
return values
def read_metrics(uuid, mount):
result = subprocess.run(
[str(Path(__file__).with_name('bcachefs_reconcile.sh')), mount],
env=dict(os.environ, BCACHEFS_UUID=uuid), capture_output=True, text=True,
check=True, timeout=15,
)
return parse_metrics(result.stdout)
+1 -1
View File
@@ -48,7 +48,7 @@ class BridgeTest(unittest.TestCase):
def test_connect_discovery_and_state_before_online(self):
self.bridge.on_connect(self.client, None, None, SimpleNamespace(is_failure=False), None)
configs = [json.loads(value) for topic, value in self.client.messages if topic.endswith('/config')]
configs = [json.loads(value) for topic, value in self.client.messages if '/switch/' in topic and topic.endswith('/config')]
self.assertEqual(len(configs), 2)
self.assertFalse(configs[0]['optimistic'])
self.assertFalse(configs[0]['retain'])
+28
View File
@@ -0,0 +1,28 @@
import assert from 'node:assert/strict';
import {estimate} from '../homeassistant/bcachefs-reconcile-card.mjs';
const hour = 3600000;
const end = 1800000000000;
const history = [
{lu:(end-6*hour)/1000,s:'1000'},
{lu:(end-3*hour)/1000,s:'900'},
{lu:end/1000,s:'600'},
];
const six = estimate(history, end-6*hour, end);
const three = estimate(history, end-3*hour, end);
assert.equal(six.rate,400/21600);
assert.equal(three.rate,300/10800);
assert.equal(six.eta,32400);
assert.equal(three.eta,21600);
assert.notEqual(six.rate,three.rate);
assert.notEqual(six.eta,three.eta);
const flat = [{lu:(end-hour)/1000,s:'5'},{lu:end/1000,s:'5'}];
assert.equal(estimate(flat,end-6*hour,end).eta,null);
assert.equal(estimate(flat,end-6*hour,end).coverage,1/6);
assert.equal(estimate([{lu:(end-hour)/1000,s:'0'},{lu:end/1000,s:'0'}],end-hour,end).eta,0);
assert.equal(estimate([{lu:(end-hour)/1000,s:'5'},{lu:end/1000,s:'10'}],end-hour,end).eta,null);
assert.match(estimate([{lu:end/1000,s:'unavailable'}],end-hour,end).status,/unavailable/);
assert.match(estimate([{lu:(end-hour)/1000,s:'5'}],end-6*hour,end).status,/stale/);
assert.match(estimate([],end-hour,end).status,/Waiting/);
assert.equal(estimate(history,end-6*hour,end,'GiB').rate,six.rate*1024**3);
assert.equal(estimate(history,end-6*hour,end,'GiB').eta,six.eta);
console.log('PASS: selected-window speed/ETA, flat/increasing/completed work, partial history, missing/stale data, unit conversion');
+33
View File
@@ -0,0 +1,33 @@
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
from reconcile_metrics import parse_metrics, FIELDS
from bcachefs_mqtt import Bridge
from test_mqtt_report import Client, UUID
class MetricsTest(unittest.TestCase):
def test_byte_values_and_missing_fields(self):
self.assertEqual(parse_metrics('bcachefs_reconcile,uuid=test scan_pending=0,target_data=123,erasure_code_data=456\n'),{'target_data':123,'erasure_code_data':456})
for data in ('', 'invalid', 'bcachefs_reconcile,uuid=test target_data=-1', 'bcachefs_reconcile,uuid=test target_data=NaN', 'bcachefs_reconcile,uuid=test scan_pending=0'):
with self.subTest(data=data), self.assertRaises(ValueError):parse_metrics(data)
def test_discovery_and_failed_collection(self):
with tempfile.TemporaryDirectory() as root:
options=Path(root)/UUID/'options'; options.mkdir(parents=True)
client=Client(); bridge=Bridge(client,UUID,root)
bridge.announce()
configs=[json.loads(payload) for topic,payload in client.messages if '/sensor/' in topic]
self.assertEqual(len(configs),16)
self.assertTrue(all(c['force_update'] and c['unit_of_measurement']=='B' for c in configs))
self.assertEqual({c['unique_id'] for c in configs}, {f'bcachefs_{UUID}_reconcile_{field}' for field in FIELDS})
with patch('bcachefs_mqtt.read_metrics',return_value={'target_data':123}):bridge.sample_reconcile('/mnt')
self.assertIn((bridge.base+'/reconcile/state','{"target_data": 123}'),client.messages)
with patch('bcachefs_mqtt.read_metrics',side_effect=subprocess.TimeoutExpired('collector',15)):bridge.sample_reconcile('/mnt')
self.assertEqual(client.messages[-1],(bridge.base+'/reconcile/availability','offline'))
if __name__=='__main__':unittest.main()