go-ecoflow-exporter/prometheus.go

package main

import (
	"context"
	"errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"log/slog"
	"net/http"
	"strings"
	"sync"
)

// check that PrometheusExporter implements MetricHandler
var _ MetricHandler = (*PrometheusExporter)(nil)
var _ Shutdownable = (*PrometheusExporter)(nil)

// PrometheusConfig represents the configuration for recording Prometheus metrics.
type PrometheusConfig struct {
	// Prefix represents the metric prefix to be used when recording Prometheus metrics. It's a good idea to use value "ecoflow"
	Prefix     string
	ServerPort string
}

type PrometheusExporter struct {
	Config  *PrometheusConfig
	metrics map[string]prometheus.Gauge
	mu      sync.RWMutex
	Server  *http.Server
}

func NewPrometheusExporter(config *PrometheusConfig) *PrometheusExporter {
	slog.Debug("Creating prometheus exporter")

	// Set up HTTP server for Prometheus metrics
	mux := http.NewServeMux()
	mux.Handle("/metrics", promhttp.Handler())

	server := &http.Server{
		Addr:    ":" + config.ServerPort,
		Handler: mux,
	}

	go func() {
		// Start the HTTP server
		slog.Debug("Starting HTTP server", "port", config.ServerPort)
		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			slog.Error("HTTP server ListenAndServe error", "error", err)
		}
	}()
	return &PrometheusExporter{
		Config:  config,
		Server:  server,
		mu:      sync.RWMutex{},
		metrics: make(map[string]prometheus.Gauge),
	}
}

func (p *PrometheusExporter) Handle(_ context.Context, device EcoflowDevice, rawParameters map[string]interface{}) {
	slog.Debug("Handling prometheus metrics for device", "sn", device.SN)
	if device.Online == 0 {
		slog.Info("Device is offline. Setting all metrics to 0", "SN", device.SN, "device name", device.Name)
		p.handleOfflineDevice(device)
		return
	}
	rawParameters["online"] = float64(device.Online)
	p.handleMetrics(device, rawParameters)
}

func (p *PrometheusExporter) handleOfflineDevice(device EcoflowDevice) {
	for k, v := range p.metrics {
		if strings.Contains(k, device.SN) {
			v.Set(0)
		}
	}
}

func (p *PrometheusExporter) handleMetrics(device EcoflowDevice, parameters map[string]interface{}) {
	for field, val := range parameters {
		p.handleOneMetric(device, field, val)
	}
}

func (p *PrometheusExporter) handleOneMetric(device EcoflowDevice, field string, val interface{}) {
	metricName, deviceMetricName, err := generateMetricName(field, p.Config.Prefix, device.SN)
	if err != nil {
		slog.Error("Unable to generate metric name", "metric", field)
		return
	}
	p.mu.Lock()
	gauge, ok := p.metrics[deviceMetricName]
	p.mu.Unlock()
	if !ok {
		slog.Debug("Adding new metric", "metric", metricName, "device", device.SN, "device_name", device.Name)
		gauge = prometheus.NewGauge(prometheus.GaugeOpts{
			Name: metricName,
			ConstLabels: map[string]string{
				"device":        device.Name,
				"serial_number": device.SN,
			},
		})
		prometheus.MustRegister(gauge)
		p.mu.Lock()
		p.metrics[deviceMetricName] = gauge
		p.mu.Unlock()
	} else {
		slog.Debug("Updating metric", "metric", metricName, "value", val, "device", device.SN, "device_name", device.Name)
	}
	_, ok = val.([]interface{})
	if ok {
		slog.Debug("The value is an array, skipping it", "metric", metricName)
		return
	}
	floatVal, ok := val.(float64)
	if ok {
		gauge.Set(floatVal)
	} else {
		slog.Error("Unable to convert value to float, skipping metric", "value", val, "metric", metricName)
	}
}

func (p *PrometheusExporter) Close(ctx context.Context) {
	// Shutdown HTTP server
	slog.Debug("Shutting down HTTP server...")
	if err := p.Server.Shutdown(ctx); err != nil {
		slog.Error("HTTP server shutdown error", "error", err)
	} else {
		slog.Debug("HTTP server gracefully stopped")
	}
}