Initial commit

metrics/pve_metrics_manager.go

@@ -0,0 +1,126 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	log "github.com/sirupsen/logrus"
+	"lostak.dev/pve-exporter/configuration"
+	"lostak.dev/pve-exporter/proxmox"
+	"lostak.dev/pve-exporter/utils"
+)
+
+// PVE metrics collector interface.
+type PveMetricsCollector interface {
+	CollectMetrics() error // Called by manager on metrics colletion.
+	GetName() string       // Used for logging purposes if error happens in collector.
+}
+
+// PVE metrics manager
+type PveMetricsManager struct {
+	apiClient  *proxmox.PveApiClient // Proxmox virtual environment API client instance.
+	collectors []PveMetricsCollector // Metrics collector instances.
+
+	latencySummary *prometheus.SummaryVec // Collection latency summary.
+	interval       int                    // Collection interval.
+
+	stop chan struct{} // Stop channel which is used in ticker.
+}
+
+// Create new instance of PVE metrics collector.
+func NewPveMetricsManager(apiClient *proxmox.PveApiClient, conf *configuration.PveConfiguration) *PveMetricsManager {
+	c := PveMetricsManager{apiClient: apiClient, interval: conf.Interval}
+	metricsCf := conf.Metrics
+
+	// Cluster state metrics collector.
+	if metricsCf.ClusterState {
+		c.RegisterCollector(NewPveClusterStateCollector(apiClient))
+	}
+
+	// Node state metrics collector.
+	if metricsCf.NodeStatus {
+		c.RegisterCollector(NewPveNodeStatusCollector(apiClient))
+	}
+
+	// Node subscription state collector.
+	if metricsCf.Subscription {
+		c.RegisterCollector(NewPveSubscriptionCollector(apiClient))
+	}
+
+	// Node disk collector.
+	if metricsCf.Disk {
+		c.RegisterCollector(NewPveNodeDiskCollector(apiClient))
+	}
+
+	// Node SDN collector.
+	if metricsCf.SDN {
+		c.RegisterCollector(NewPveSdnCollector(apiClient))
+	}
+
+	// Node storage collector.
+	if metricsCf.Storage {
+		c.RegisterCollector(NewPveStorageCollector(apiClient))
+	}
+
+	// Node container collector.
+	if metricsCf.LXC {
+		c.RegisterCollector(NewPveContainerCollector(apiClient))
+	}
+
+	// Node virtual machine collector.
+	if metricsCf.QEMU {
+		c.RegisterCollector(NewPveVirtualMachineCollector(apiClient))
+	}
+
+	// Metrics collection latency summary.
+	c.latencySummary = promauto.NewSummaryVec(prometheus.SummaryOpts{
+		Name: "pve_metrics_collection_latency_ms",
+		Help: "Summary of metrics collection latency milliseconds from PVE API.",
+	}, []string{"collector"})
+
+	return &c
+}
+
+// Called periodically by ticker and invokes collection on all registered collectors.
+func (c *PveMetricsManager) collectMetrics() {
+	for _, collector := range c.collectors {
+		start := time.Now()
+		log.Tracef("Collecting %s metrics...", collector.GetName())
+		err := collector.CollectMetrics()
+		if err != nil {
+			log.Errorf("Failed to collect '%s' metrics. Error: %s.", collector.GetName(), err)
+		} else {
+			latency := time.Since(start)
+			log.Tracef("Finished collecting '%s' metrics after %s.", collector.GetName(), utils.HumanDuration(latency))
+			c.latencySummary.With(prometheus.Labels{"collector": collector.GetName()}).Observe(float64(latency.Milliseconds()))
+		}
+	}
+}
+
+// Registers collector instance for metrics collection.
+func (c *PveMetricsManager) RegisterCollector(collector PveMetricsCollector) {
+	c.collectors = append(c.collectors, collector)
+	log.Infof("Metrics collector '%s' registered successfully.", collector.GetName())
+}
+
+// Start metrics collector
+func (c *PveMetricsManager) Start() {
+	ticker := time.NewTicker(time.Second * time.Duration(c.interval))
+	c.collectMetrics()
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				c.collectMetrics()
+			case <-c.stop:
+				return
+			}
+		}
+	}()
+}
+
+// Close stop metrics collector.
+func (c *PveMetricsManager) Stop() {
+	close(c.stop)
+}