diff --git a/README.md b/README.md index f585672..7e588e2 100644 --- a/README.md +++ b/README.md @@ -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 +``` diff --git a/bcachefs_mqtt.py b/bcachefs_mqtt.py index 88c621f..326c444 100644 --- a/bcachefs_mqtt.py +++ b/bcachefs_mqtt.py @@ -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) diff --git a/homeassistant/bcachefs-reconcile-card.mjs b/homeassistant/bcachefs-reconcile-card.mjs new file mode 100644 index 0000000..ebb241d --- /dev/null +++ b/homeassistant/bcachefs-reconcile-card.mjs @@ -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 = `

Pending reconcile work

+
+
+
Remaining data—
+
Average net progress—
+
Estimated time remaining—
+

Loading Home Assistant history…

+

Speed = (first − last pending bytes) ÷ elapsed time in the selected window. ETA assumes that average continues. Categories can overlap; they are not added together.

+
`; + 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); diff --git a/reconcile_metrics.py b/reconcile_metrics.py new file mode 100644 index 0000000..f1f66a7 --- /dev/null +++ b/reconcile_metrics.py @@ -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) diff --git a/tests/test_mqtt_report.py b/tests/test_mqtt_report.py index 6cfa56d..149f59a 100644 --- a/tests/test_mqtt_report.py +++ b/tests/test_mqtt_report.py @@ -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']) diff --git a/tests/test_reconcile_history.mjs b/tests/test_reconcile_history.mjs new file mode 100644 index 0000000..a7ee0ac --- /dev/null +++ b/tests/test_reconcile_history.mjs @@ -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'); diff --git a/tests/test_reconcile_metrics.py b/tests/test_reconcile_metrics.py new file mode 100644 index 0000000..25c5251 --- /dev/null +++ b/tests/test_reconcile_metrics.py @@ -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()