seaweedfs/telemetry/server/storage/prometheus.go

package storage

import (
	"sync"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/seaweedfs/seaweedfs/telemetry/proto"
)

type PrometheusStorage struct {
	// Prometheus metrics
	totalClusters     prometheus.Gauge
	activeClusters    prometheus.Gauge
	confirmedClusters prometheus.Gauge
	volumeServerCount *prometheus.GaugeVec
	totalDiskBytes    *prometheus.GaugeVec
	totalVolumeCount  *prometheus.GaugeVec
	filerCount        *prometheus.GaugeVec
	brokerCount       *prometheus.GaugeVec
	clusterInfo       *prometheus.GaugeVec
	telemetryReceived prometheus.Counter

	// In-memory storage for API endpoints (if needed)
	mu        sync.RWMutex
	instances map[string]*telemetryData
	histories map[string][]HistorySample
	stats     map[string]interface{}
	dirty     bool // instances changed since the last successful state save
}

// telemetryData is an internal struct that includes the received timestamp
type telemetryData struct {
	*proto.TelemetryData
	ReceivedAt time.Time `json:"received_at"`
}

func NewPrometheusStorage() *PrometheusStorage {
	return newPrometheusStorage(prometheus.DefaultRegisterer)
}

func newPrometheusStorage(reg prometheus.Registerer) *PrometheusStorage {
	promauto := promauto.With(reg)
	return &PrometheusStorage{
		totalClusters: promauto.NewGauge(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_total_clusters",
			Help: "Total number of unique SeaweedFS clusters (last 30 days)",
		}),
		activeClusters: promauto.NewGauge(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_active_clusters",
			Help: "Number of active SeaweedFS clusters (last 7 days)",
		}),
		confirmedClusters: promauto.NewGauge(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_confirmed_clusters",
			Help: "Active clusters seen on at least 7 distinct days (last 7 days)",
		}),
		volumeServerCount: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_volume_servers",
			Help: "Number of volume servers per cluster",
		}, []string{"cluster_id"}),
		totalDiskBytes: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_disk_bytes",
			Help: "Total disk usage in bytes per cluster",
		}, []string{"cluster_id"}),
		totalVolumeCount: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_volume_count",
			Help: "Total number of volumes per cluster",
		}, []string{"cluster_id"}),
		filerCount: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_filer_count",
			Help: "Number of filer servers per cluster",
		}, []string{"cluster_id"}),
		brokerCount: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_broker_count",
			Help: "Number of broker servers per cluster",
		}, []string{"cluster_id"}),
		clusterInfo: promauto.NewGaugeVec(prometheus.GaugeOpts{
			Name: "seaweedfs_telemetry_cluster_info",
			Help: "Cluster information (always 1, labels contain metadata)",
		}, []string{"cluster_id", "version", "os"}),
		telemetryReceived: promauto.NewCounter(prometheus.CounterOpts{
			Name: "seaweedfs_telemetry_reports_received_total",
			Help: "Total number of telemetry reports received",
		}),
		instances: make(map[string]*telemetryData),
		histories: make(map[string][]HistorySample),
		stats:     make(map[string]interface{}),
	}
}

func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Drop the cluster_info series recorded under the previous label set when
	// a cluster reports back with a different version or OS, so it is not
	// counted under two versions at once.
	if prev, ok := s.instances[data.TopologyId]; ok &&
		(prev.TelemetryData.Version != data.Version || prev.TelemetryData.Os != data.Os) {
		s.clusterInfo.Delete(infoLabels(prev.TelemetryData))
	}
	s.setClusterMetrics(data)

	s.telemetryReceived.Inc()

	// Store in memory for API endpoints
	receivedAt := time.Now().UTC()
	s.instances[data.TopologyId] = &telemetryData{
		TelemetryData: data,
		ReceivedAt:    receivedAt,
	}
	s.appendHistory(data, receivedAt)
	s.dirty = true

	// Update aggregated stats
	s.updateStats()

	return nil
}

// setClusterMetrics records a report's values on the Prometheus gauges.
// Value gauges are keyed by cluster_id only so a cluster's series continues
// across upgrades; version/os metadata lives on cluster_info (join with
// `* on(cluster_id) group_left(version, os)`). Callers must hold s.mu.
func (s *PrometheusStorage) setClusterMetrics(data *proto.TelemetryData) {
	labels := prometheus.Labels{
		"cluster_id": data.TopologyId,
	}
	s.volumeServerCount.With(labels).Set(float64(data.VolumeServerCount))
	s.totalDiskBytes.With(labels).Set(float64(data.TotalDiskBytes))
	s.totalVolumeCount.With(labels).Set(float64(data.TotalVolumeCount))
	s.filerCount.With(labels).Set(float64(data.FilerCount))
	s.brokerCount.With(labels).Set(float64(data.BrokerCount))
	s.clusterInfo.With(infoLabels(data)).Set(1)
}

func (s *PrometheusStorage) GetStats() (map[string]interface{}, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// Return cached stats
	result := make(map[string]interface{})
	for k, v := range s.stats {
		result[k] = v
	}
	return result, nil
}

func (s *PrometheusStorage) GetInstances(limit int) ([]*telemetryData, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	var instances []*telemetryData
	count := 0
	for _, instance := range s.instances {
		if count >= limit {
			break
		}
		instances = append(instances, instance)
		count++
	}

	return instances, nil
}

// GetMetrics returns fleet-wide daily totals across confirmed clusters for the
// last `days` days, in the parallel-array shape the dashboard charts expect.
// Totals come from the daily histories rather than from s.instances, which holds
// only each cluster's most recent report and so would credit every cluster to
// the single day it last reported on.
func (s *PrometheusStorage) GetMetrics(days int) (map[string]interface{}, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	histories := s.seriesHistories()
	axis := newDailySeries(days, histories)
	activeSince := time.Now().UTC().AddDate(0, 0, -activeDays).Unix()

	diskUsage := make([]uint64, len(axis.dates))
	serverCounts := make([]int64, len(axis.dates))
	for _, history := range histories {
		if disk, ok := align(axis, history, activeSince, diskBytes); ok {
			for i, v := range disk {
				diskUsage[i] += v
			}
		}
		if servers, ok := align(axis, history, activeSince, serverCount); ok {
			for i, v := range servers {
				serverCounts[i] += int64(v)
			}
		}
	}

	return map[string]interface{}{
		"dates":         axis.dates,
		"server_counts": serverCounts,
		"disk_usage":    diskUsage,
	}, nil
}

func (s *PrometheusStorage) updateStats() {
	now := time.Now()
	last7Days := now.AddDate(0, 0, -activeDays)
	last30Days := now.AddDate(0, 0, -30)

	totalInstances := 0
	activeInstances := 0
	confirmedInstances := 0
	versionsAll := make(map[string]int)
	osAll := make(map[string]int)
	versionsConfirmed := make(map[string]int)
	osConfirmed := make(map[string]int)

	for _, instance := range s.instances {
		if instance.ReceivedAt.After(last30Days) {
			totalInstances++
		}
		if instance.ReceivedAt.After(last7Days) {
			activeInstances++
			versionsAll[instance.TelemetryData.Version]++
			osAll[instance.TelemetryData.Os]++
			// A cluster is confirmed once seen on confirmDays distinct UTC
			// days (histories hold one sample per day), so short-lived
			// clusters can't skew the distributions below.
			if len(s.histories[instance.TelemetryData.TopologyId]) >= confirmDays {
				confirmedInstances++
				versionsConfirmed[instance.TelemetryData.Version]++
				osConfirmed[instance.TelemetryData.Os]++
			}
		}
	}

	// Before any cluster has a week of history (fresh server with no
	// prior state), fall back to all active clusters so the dashboard
	// distributions aren't empty.
	versions, osDistribution := versionsConfirmed, osConfirmed
	if confirmedInstances == 0 {
		versions, osDistribution = versionsAll, osAll
	}

	// Update Prometheus gauges
	s.totalClusters.Set(float64(totalInstances))
	s.activeClusters.Set(float64(activeInstances))
	s.confirmedClusters.Set(float64(confirmedInstances))

	// Update cached stats for API
	s.stats = map[string]interface{}{
		"total_instances":     totalInstances,
		"active_instances":    activeInstances,
		"confirmed_instances": confirmedInstances,
		"versions":            versions,
		"os_distribution":     osDistribution,
	}
}

// CleanupOldInstances removes instances older than the specified duration
func (s *PrometheusStorage) CleanupOldInstances(maxAge time.Duration) {
	s.mu.Lock()
	defer s.mu.Unlock()

	cutoff := time.Now().Add(-maxAge)
	for instanceID, instance := range s.instances {
		if instance.ReceivedAt.Before(cutoff) {
			delete(s.instances, instanceID)
			s.deleteClusterMetrics(instance.TelemetryData)
			s.dirty = true
		}
	}

	for id, h := range s.histories {
		if _, ok := s.instances[id]; !ok {
			delete(s.histories, id)
			s.dirty = true
			continue
		}
		i := 0
		for i < len(h) && time.Unix(h[i].Ts, 0).Before(cutoff) {
			i++
		}
		if i > 0 {
			s.histories[id] = append([]HistorySample(nil), h[i:]...)
			s.dirty = true
		}
	}

	s.updateStats()
}

// deleteClusterMetrics removes all gauges stored for the given report's
// cluster. Callers must hold s.mu.
func (s *PrometheusStorage) deleteClusterMetrics(data *proto.TelemetryData) {
	labels := prometheus.Labels{
		"cluster_id": data.TopologyId,
	}
	s.volumeServerCount.Delete(labels)
	s.totalDiskBytes.Delete(labels)
	s.totalVolumeCount.Delete(labels)
	s.filerCount.Delete(labels)
	s.brokerCount.Delete(labels)
	s.clusterInfo.Delete(infoLabels(data))
}

// infoLabels is the full label set used by the cluster_info metric.
func infoLabels(data *proto.TelemetryData) prometheus.Labels {
	return prometheus.Labels{
		"cluster_id": data.TopologyId,
		"version":    data.Version,
		"os":         data.Os,
	}
}