implement econet exporter

pkg/econet/econetmetrics/builder.go

@@ -0,0 +1,41 @@
+package econetmetrics
+
+import "go.opentelemetry.io/otel/metric"
+
+// gaugeBuilder creates observable gauges, accumulating them for a single
+// RegisterCallback and short-circuiting on the first creation error.
+//
+// Units are intentionally baked into the metric names (Prometheus idiom)
+// rather than set via metric.WithUnit — the OTEL→Prometheus exporter would
+// otherwise append a second unit suffix (e.g. econet_setpoint_fahrenheit_F).
+type gaugeBuilder struct {
+	m   metric.Meter
+	obs []metric.Observable
+	err error
+}
+
+func (b *gaugeBuilder) i64(name, desc string) metric.Int64ObservableGauge {
+	if b.err != nil {
+		return nil
+	}
+	g, err := b.m.Int64ObservableGauge(name, metric.WithDescription(desc))
+	if err != nil {
+		b.err = err
+		return nil
+	}
+	b.obs = append(b.obs, g)
+	return g
+}
+
+func (b *gaugeBuilder) f64(name, desc string) metric.Float64ObservableGauge {
+	if b.err != nil {
+		return nil
+	}
+	g, err := b.m.Float64ObservableGauge(name, metric.WithDescription(desc))
+	if err != nil {
+		b.err = err
+		return nil
+	}
+	b.obs = append(b.obs, g)
+	return g
+}

pkg/econet/econetmetrics/gauges.go

@@ -0,0 +1,132 @@
+package econetmetrics
+
+import (
+	"context"
+
+	rheemcloud "github.com/kevinburke/rheemcloud-go"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// registerLive wires the per-device live-state gauges to a single callback
+// that snapshots client.Devices() once per scrape (cheap, in-memory).
+func (c *Collector) registerLive() error {
+	b := &gaugeBuilder{m: c.meter}
+	g := &liveGauges{
+		setpoint:    b.i64("econet_setpoint_fahrenheit", "Target water temperature (°F)"),
+		setpointMin: b.i64("econet_setpoint_min_fahrenheit", "Minimum allowed setpoint (°F)"),
+		setpointMax: b.i64("econet_setpoint_max_fahrenheit", "Maximum allowed setpoint (°F)"),
+		connected:   b.i64("econet_connected", "1 if the device is connected"),
+		running:     b.i64("econet_running", "1 if the device is actively heating"),
+		enabled:     b.i64("econet_enabled", "1 if the device is enabled"),
+		away:        b.i64("econet_away", "1 if away mode is on"),
+		hotWater:    b.i64("econet_hot_water_availability_percent", "Hot water availability (%)"),
+		alerts:      b.i64("econet_alert_count", "Number of active alerts"),
+		wifi:        b.i64("econet_wifi_signal_db", "WiFi signal strength (dB)"),
+		lastUpdated: b.i64("econet_last_updated_seconds", "Unix time of last MQTT state update"),
+		info:        b.i64("econet_device_info", "Device metadata (value is always 1)"),
+	}
+	if b.err != nil {
+		return b.err
+	}
+	_, err := c.meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+		for _, d := range c.client.Devices() {
+			g.observe(o, d)
+		}
+		return nil
+	}, b.obs...)
+	return err
+}
+
+// registerUsage wires the polled energy/water gauges, reading the cache the
+// poller refreshes (the underlying REST calls are too slow for a callback).
+func (c *Collector) registerUsage() error {
+	b := &gaugeBuilder{m: c.meter}
+	g := &usageGauges{
+		energy: b.f64("econet_energy_usage_kwh", "Energy used today (kWh)"),
+		cost:   b.f64("econet_energy_cost_dollars", "Energy cost today in US dollars (kWh * costPerKWH)"),
+		water:  b.f64("econet_water_usage_gallons", "Water used today (gallons)"),
+	}
+	if b.err != nil {
+		return b.err
+	}
+	_, err := c.meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+		c.mu.RLock()
+		defer c.mu.RUnlock()
+		for serial, u := range c.usage {
+			g.observe(o, serial, u, c.cfg.CostPerKWH)
+		}
+		return nil
+	}, b.obs...)
+	return err
+}
+
+type liveGauges struct {
+	setpoint, setpointMin, setpointMax  metric.Int64ObservableGauge
+	connected, running, enabled, away   metric.Int64ObservableGauge
+	hotWater, alerts, wifi, lastUpdated metric.Int64ObservableGauge
+	info                                metric.Int64ObservableGauge
+}
+
+func (g *liveGauges) observe(o metric.Observer, d *rheemcloud.Device) {
+	set := metric.WithAttributes(deviceAttrs(d)...)
+	min, max := d.SetpointLimits()
+	o.ObserveInt64(g.setpoint, int64(d.Setpoint()), set)
+	o.ObserveInt64(g.setpointMin, int64(min), set)
+	o.ObserveInt64(g.setpointMax, int64(max), set)
+	o.ObserveInt64(g.connected, b2i(d.Connected()), set)
+	o.ObserveInt64(g.running, b2i(d.Running()), set)
+	o.ObserveInt64(g.enabled, b2i(d.Enabled()), set)
+	o.ObserveInt64(g.away, b2i(d.Away()), set)
+	o.ObserveInt64(g.hotWater, int64(d.HotWaterAvailability()), set)
+	o.ObserveInt64(g.alerts, int64(d.AlertCount()), set)
+	o.ObserveInt64(g.wifi, int64(d.WiFiSignal()), set)
+	// LastUpdated is zero until the first MQTT update lands; skip it then
+	// rather than emit a nonsensical negative epoch.
+	if !d.LastUpdated.IsZero() {
+		o.ObserveInt64(g.lastUpdated, d.LastUpdated.Unix(), set)
+	}
+	o.ObserveInt64(g.info, 1, metric.WithAttributes(infoAttrs(d)...))
+}
+
+type usageGauges struct {
+	energy, cost, water metric.Float64ObservableGauge
+}
+
+func (g *usageGauges) observe(o metric.Observer, serial string, u usage, costPerKWH float64) {
+	set := metric.WithAttributes(attribute.String("econet.serial", serial))
+	o.ObserveFloat64(g.energy, u.energyKWH, metric.WithAttributes(
+		attribute.String("econet.serial", serial),
+		attribute.String("econet.energy_type", u.energyType),
+	))
+	o.ObserveFloat64(g.water, u.waterGallons, set)
+	if u.energyType == "KWH" {
+		o.ObserveFloat64(g.cost, u.energyKWH*costPerKWH, set)
+	}
+}
+
+// deviceAttrs are low-cardinality identity attributes safe for every series.
+// They follow OTEL semantic-convention style (namespaced, dotted keys) and
+// deliberately exclude anything that changes over time (e.g. timestamps).
+func deviceAttrs(d *rheemcloud.Device) []attribute.KeyValue {
+	return []attribute.KeyValue{
+		attribute.String("econet.serial", d.SerialNumber()),
+		attribute.String("econet.device_id", d.DeviceID()),
+		attribute.String("econet.friendly_name", d.FriendlyName()),
+	}
+}
+
+func infoAttrs(d *rheemcloud.Device) []attribute.KeyValue {
+	return append(deviceAttrs(d),
+		attribute.String("econet.type", d.Type().String()),
+		attribute.String("econet.generic_type", d.GenericType()),
+		attribute.String("econet.mode", d.Mode().String()),
+	)
+}
+
+func b2i(b bool) int64 {
+	if b {
+		return 1
+	}
+	return 0
+}

pkg/econet/econetmetrics/gauges_test.go

@@ -0,0 +1,12 @@
+package econetmetrics
+
+import "testing"
+
+func TestB2I(t *testing.T) {
+	if got := b2i(true); got != 1 {
+		t.Errorf("b2i(true) = %d, want 1", got)
+	}
+	if got := b2i(false); got != 0 {
+		t.Errorf("b2i(false) = %d, want 0", got)
+	}
+}

pkg/econet/econetmetrics/metrics.go

@@ -0,0 +1,123 @@
+// Package econetmetrics registers OTEL observable gauges for Rheem EcoNet
+// device state and periodically polls energy/water usage history so the
+// values are available at Prometheus scrape time.
+package econetmetrics
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	rheemcloud "github.com/kevinburke/rheemcloud-go"
+	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/metric"
+
+	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
+
+	"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/config"
+)
+
+// usage holds the most recent polled energy/water totals for a device.
+type usage struct {
+	energyKWH    float64
+	energyType   string
+	waterGallons float64
+}
+
+type Collector struct {
+	ctx    context.Context
+	cfg    *config.ServiceConfig
+	log    *zerolog.Logger
+	client *rheemcloud.Client
+	meter  metric.Meter
+
+	mu    sync.RWMutex
+	usage map[string]usage // keyed by serial number
+
+	stop chan struct{}
+}
+
+func NewCollector(ctx context.Context, cfg *config.ServiceConfig, client *rheemcloud.Client) *Collector {
+	return &Collector{
+		ctx:    ctx,
+		cfg:    cfg,
+		log:    zerolog.Ctx(ctx),
+		client: client,
+		meter:  otel.GetMeter(ctx, "econet"),
+		usage:  make(map[string]usage),
+		stop:   make(chan struct{}),
+	}
+}
+
+// Start registers the gauge callbacks and launches the usage poller and
+// event drain goroutines. The rheemcloud client keeps live state fresh over
+// MQTT; draining its event channel keeps the cap-64 buffer from filling.
+func (c *Collector) Start() error {
+	if err := c.registerLive(); err != nil {
+		return err
+	}
+	if err := c.registerUsage(); err != nil {
+		return err
+	}
+	go c.drainEvents()
+	go c.pollUsage()
+	return nil
+}
+
+func (c *Collector) Stop() { close(c.stop) }
+
+func (c *Collector) drainEvents() {
+	events := c.client.Subscribe()
+	for {
+		select {
+		case <-c.stop:
+			return
+		case <-c.ctx.Done():
+			return
+		case ev, ok := <-events:
+			if !ok {
+				return
+			}
+			c.log.Debug().Str("event", ev.Kind.String()).Msg("econet event")
+		}
+	}
+}
+
+func (c *Collector) pollUsage() {
+	c.refreshUsage()
+	t := time.NewTicker(c.cfg.GetUsageInterval())
+	defer t.Stop()
+	for {
+		select {
+		case <-c.stop:
+			return
+		case <-c.ctx.Done():
+			return
+		case <-t.C:
+			c.refreshUsage()
+		}
+	}
+}
+
+// refreshUsage polls each device's energy/water usage for the current day
+// and caches the totals for the usage gauge callback to read.
+func (c *Collector) refreshUsage() {
+	now := time.Now()
+	start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
+	for serial, d := range c.client.Devices() {
+		u := usage{}
+		if e, err := c.client.EnergyUsage(c.ctx, d, start, now); err == nil {
+			u.energyKWH, u.energyType = e.Total, e.EnergyType
+		} else {
+			c.log.Debug().Err(err).Str("serial", serial).Msg("energy usage poll failed")
+		}
+		if w, err := c.client.WaterUsage(c.ctx, d, start, now); err == nil {
+			u.waterGallons = w.Total
+		} else {
+			c.log.Debug().Err(err).Str("serial", serial).Msg("water usage poll failed")
+		}
+		c.mu.Lock()
+		c.usage[serial] = u
+		c.mu.Unlock()
+	}
+}