mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
rcourtman
8fad031d95
Reloading configuration replaced the monitor's config pointer while API and authentication handlers retained the original object. Tokens created afterwards could authenticate successfully while Agent Doctor reported an unlisted token. Refresh the canonical runtime configuration in place before constructing the replacement manager. Cover repeated reloads, live token creation and revocation.
441 lines
13 KiB
Go
441 lines
13 KiB
Go
package monitoring
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/websocket"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// InstallSnapshotCounts holds install-wide resource and alert counts aggregated
|
|
// across tenant monitors.
|
|
type InstallSnapshotCounts struct {
|
|
PVENodes int
|
|
PBSInstances int
|
|
PMGInstances int
|
|
VMs int
|
|
Containers int
|
|
AgentHosts int
|
|
DockerHosts int
|
|
DockerContainers int
|
|
KubernetesClusters int
|
|
KubernetesNodes int
|
|
KubernetesPods int
|
|
KubernetesDeployments int
|
|
StoragePools int
|
|
PhysicalDisks int
|
|
CephClusters int
|
|
NetworkShares int
|
|
TrueNASSystems int
|
|
TrueNASVMs int
|
|
TrueNASApps int
|
|
VMwareHosts int
|
|
VMwareVMs int
|
|
VMwareDatastores int
|
|
AvailabilityTargets int
|
|
ActiveAlerts int
|
|
AlertsFired30d int
|
|
AlertsAcknowledged30d int
|
|
AlertsResolved30d int
|
|
NotificationAttempts7d int
|
|
NotificationDeliveries7d int
|
|
NotificationFailures7d int
|
|
NotificationFailuresAuthentication7d int
|
|
NotificationFailuresRateLimited7d int
|
|
NotificationFailuresConnectivity7d int
|
|
NotificationFailuresTLS7d int
|
|
NotificationFailuresConfiguration7d int
|
|
NotificationFailuresRejected7d int
|
|
NotificationFailuresServerError7d int
|
|
NotificationFailuresUnknown7d int
|
|
AlertQuality alerts.AlertQualitySnapshot
|
|
}
|
|
|
|
// ReloadableMonitor wraps a Monitor with reload capability
|
|
type ReloadableMonitor struct {
|
|
mu sync.RWMutex
|
|
mtMonitor *MultiTenantMonitor
|
|
persistence *config.MultiTenantPersistence
|
|
config *config.Config
|
|
wsHub *websocket.Hub
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
parentCtx context.Context
|
|
reloadChan chan chan error
|
|
}
|
|
|
|
// NewReloadableMonitor creates a new reloadable monitor
|
|
func NewReloadableMonitor(cfg *config.Config, persistence *config.MultiTenantPersistence, wsHub *websocket.Hub) (*ReloadableMonitor, error) {
|
|
mtMonitor := NewMultiTenantMonitor(cfg, persistence, wsHub)
|
|
// No error check needed for NewMultiTenantMonitor as it doesn't return error
|
|
|
|
rm := &ReloadableMonitor{
|
|
mtMonitor: mtMonitor,
|
|
config: cfg,
|
|
persistence: persistence,
|
|
wsHub: wsHub,
|
|
reloadChan: make(chan chan error, 1),
|
|
}
|
|
|
|
return rm, nil
|
|
}
|
|
|
|
// Start starts the monitor with reload capability
|
|
func (rm *ReloadableMonitor) Start(ctx context.Context) {
|
|
rm.mu.Lock()
|
|
rm.parentCtx = ctx
|
|
rm.ctx, rm.cancel = context.WithCancel(ctx)
|
|
rm.mu.Unlock()
|
|
|
|
// Start the multi-tenant monitor manager
|
|
// Note: It doesn't start individual monitors until requested via GetMonitor()
|
|
// But we might want to start "default" monitor if it exists?
|
|
// For now, lazy loading handles it.
|
|
|
|
// Watch for reload signals
|
|
go rm.watchReload(ctx)
|
|
}
|
|
|
|
// Reload triggers a monitor reload
|
|
func (rm *ReloadableMonitor) Reload() error {
|
|
rm.mu.RLock()
|
|
ctx := rm.parentCtx
|
|
rm.mu.RUnlock()
|
|
if ctx == nil {
|
|
return fmt.Errorf("reloadable monitor has not been started")
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// The watcher exits with the server context. Neither a full request
|
|
// channel nor an abandoned reply may keep a caller waiting after that.
|
|
// Keep the reply buffered so an in-flight reload can still finish.
|
|
done := make(chan error, 1)
|
|
select {
|
|
case rm.reloadChan <- done:
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
select {
|
|
case err := <-done:
|
|
return err
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
// watchReload watches for reload signals
|
|
func (rm *ReloadableMonitor) watchReload(ctx context.Context) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case done := <-rm.reloadChan:
|
|
log.Info().Msg("reloading monitor configuration")
|
|
if err := rm.doReload(); err != nil {
|
|
log.Error().Err(err).Msg("failed to reload monitor")
|
|
done <- err
|
|
} else {
|
|
log.Info().Msg("monitor reloaded successfully")
|
|
done <- nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// doReload performs the actual reload
|
|
func (rm *ReloadableMonitor) doReload() error {
|
|
rm.mu.Lock()
|
|
defer rm.mu.Unlock()
|
|
|
|
// Load fresh configuration
|
|
cfg, err := config.LoadWithoutLoggingInit()
|
|
if err != nil {
|
|
return fmt.Errorf("reloadable monitor reload config: %w", err)
|
|
}
|
|
|
|
// Polling interval changes and other settings require a full reload
|
|
log.Info().Msg("performing full monitor reload")
|
|
|
|
// Cancel current monitor
|
|
if rm.cancel != nil {
|
|
rm.cancel()
|
|
}
|
|
|
|
// Stop the underlying multi-tenant monitor to ensure its global context is canceled
|
|
// and all child monitors are stopped.
|
|
if rm.mtMonitor != nil {
|
|
rm.mtMonitor.Stop()
|
|
}
|
|
|
|
// Wait a moment for cleanup
|
|
time.Sleep(1 * time.Second)
|
|
|
|
// Authentication middleware and API handlers retain the original runtime
|
|
// configuration. Refresh that object so the replacement default monitor also
|
|
// sees later token creation, rotation and revocation through the same owner.
|
|
config.Mu.Lock()
|
|
if rm.config == nil {
|
|
rm.config = cfg
|
|
} else {
|
|
*rm.config = *cfg
|
|
}
|
|
config.Mu.Unlock()
|
|
|
|
// Create new multi-tenant monitor
|
|
// Note: We lose existing instances state here, which is expected on full reload.
|
|
newMTMonitor := NewMultiTenantMonitor(rm.config, rm.persistence, rm.wsHub)
|
|
|
|
// Replace monitor
|
|
rm.mtMonitor = newMTMonitor
|
|
|
|
// Start new monitor context (individual monitors are lazy loaded/started)
|
|
rm.ctx, rm.cancel = context.WithCancel(rm.parentCtx)
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetMultiTenantMonitor returns the current multi-tenant monitor instance
|
|
func (rm *ReloadableMonitor) GetMultiTenantMonitor() *MultiTenantMonitor {
|
|
rm.mu.RLock()
|
|
defer rm.mu.RUnlock()
|
|
return rm.mtMonitor
|
|
}
|
|
|
|
// GetMonitor returns the default monitor instance (compatibility shim)
|
|
// It ensures the "default" tenant is initialized.
|
|
func (rm *ReloadableMonitor) GetMonitor() *Monitor {
|
|
rm.mu.RLock()
|
|
defer rm.mu.RUnlock()
|
|
if rm.mtMonitor == nil {
|
|
return nil
|
|
}
|
|
monitor, err := rm.mtMonitor.GetMonitor("default")
|
|
if err != nil {
|
|
log.Debug().Err(err).Msg("Default monitor unavailable")
|
|
return nil
|
|
}
|
|
return monitor
|
|
}
|
|
|
|
// GetConfig returns the current configuration used by the monitor.
|
|
func (rm *ReloadableMonitor) GetConfig() *config.Config {
|
|
rm.mu.RLock()
|
|
defer rm.mu.RUnlock()
|
|
if rm.config == nil {
|
|
return nil
|
|
}
|
|
return rm.config
|
|
}
|
|
|
|
// ReadSnapshot returns the current state for a specific tenant.
|
|
func (rm *ReloadableMonitor) ReadSnapshot(orgID string) interface{} {
|
|
if orgID == "" {
|
|
orgID = "default"
|
|
}
|
|
mtMonitor := rm.GetMultiTenantMonitor()
|
|
if mtMonitor == nil {
|
|
log.Debug().Str("orgID", orgID).Msg("ReadSnapshot requested with no active multi-tenant monitor")
|
|
return nil
|
|
}
|
|
|
|
monitor, err := mtMonitor.GetMonitor(orgID)
|
|
if err != nil {
|
|
log.Debug().Err(err).Str("orgID", orgID).Msg("ReadSnapshot monitor unavailable")
|
|
return nil
|
|
}
|
|
return monitor.ReadSnapshot()
|
|
}
|
|
|
|
// AggregateInstallSnapshotCounts returns install-wide resource and alert counts
|
|
// across all provisioned organizations.
|
|
func (rm *ReloadableMonitor) AggregateInstallSnapshotCounts() InstallSnapshotCounts {
|
|
rm.mu.RLock()
|
|
mtMonitor := rm.mtMonitor
|
|
persistence := rm.persistence
|
|
rm.mu.RUnlock()
|
|
|
|
if mtMonitor == nil {
|
|
return InstallSnapshotCounts{}
|
|
}
|
|
|
|
orgIDs := []string{"default"}
|
|
if persistence != nil {
|
|
orgs, err := persistence.ListOrganizations()
|
|
if err != nil {
|
|
log.Warn().Err(err).Msg("Telemetry snapshot falling back to default organization after tenant listing failed")
|
|
} else {
|
|
seen := make(map[string]struct{}, len(orgs))
|
|
orgIDs = orgIDs[:0]
|
|
for _, org := range orgs {
|
|
if org == nil {
|
|
continue
|
|
}
|
|
orgID := strings.TrimSpace(org.ID)
|
|
if orgID == "" {
|
|
continue
|
|
}
|
|
if _, exists := seen[orgID]; exists {
|
|
continue
|
|
}
|
|
seen[orgID] = struct{}{}
|
|
orgIDs = append(orgIDs, orgID)
|
|
}
|
|
if len(orgIDs) == 0 {
|
|
orgIDs = []string{"default"}
|
|
}
|
|
}
|
|
}
|
|
|
|
var counts InstallSnapshotCounts
|
|
now := time.Now().UTC()
|
|
for _, orgID := range orgIDs {
|
|
monitor, err := mtMonitor.GetMonitor(orgID)
|
|
if err != nil || monitor == nil {
|
|
log.Debug().Err(err).Str("org_id", orgID).Msg("Telemetry snapshot could not load tenant monitor")
|
|
continue
|
|
}
|
|
accumulateInstallSnapshotCounts(&counts, monitor)
|
|
accumulateInstallOutcomeCounts(&counts, monitor, now)
|
|
}
|
|
return counts
|
|
}
|
|
|
|
func accumulateInstallOutcomeCounts(counts *InstallSnapshotCounts, monitor *Monitor, now time.Time) {
|
|
if counts == nil || monitor == nil {
|
|
return
|
|
}
|
|
if now.IsZero() {
|
|
now = time.Now().UTC()
|
|
}
|
|
if alertManager := monitor.GetAlertManager(); alertManager != nil {
|
|
quality := alertManager.AlertQualityTelemetrySnapshot(now)
|
|
counts.AlertQuality.Add(quality)
|
|
counts.AlertsFired30d += quality.Fired30d
|
|
counts.AlertsAcknowledged30d += quality.Acknowledged30d
|
|
counts.AlertsResolved30d += quality.Resolved30d
|
|
}
|
|
if notificationManager := monitor.GetNotificationManager(); notificationManager != nil {
|
|
stats, err := notificationManager.GetTelemetryStats(now.Add(-7 * 24 * time.Hour))
|
|
if err != nil {
|
|
log.Debug().Err(err).Msg("Telemetry snapshot could not read notification delivery aggregates")
|
|
return
|
|
}
|
|
counts.NotificationAttempts7d += stats.Attempts
|
|
counts.NotificationDeliveries7d += stats.Deliveries
|
|
counts.NotificationFailures7d += stats.Failures
|
|
counts.NotificationFailuresAuthentication7d += stats.FailureClasses.Authentication
|
|
counts.NotificationFailuresRateLimited7d += stats.FailureClasses.RateLimited
|
|
counts.NotificationFailuresConnectivity7d += stats.FailureClasses.Connectivity
|
|
counts.NotificationFailuresTLS7d += stats.FailureClasses.TLS
|
|
counts.NotificationFailuresConfiguration7d += stats.FailureClasses.Configuration
|
|
counts.NotificationFailuresRejected7d += stats.FailureClasses.Rejected
|
|
counts.NotificationFailuresServerError7d += stats.FailureClasses.ServerError
|
|
counts.NotificationFailuresUnknown7d += stats.FailureClasses.Unknown
|
|
}
|
|
}
|
|
|
|
// accumulateAlertOutcomeCounts remains the focused history-folding seam used
|
|
// by monitoring tests. Production collection uses the manager snapshot so the
|
|
// same calculation also includes current age, adoption, and persistence health.
|
|
func accumulateAlertOutcomeCounts(counts *InstallSnapshotCounts, history []alerts.Alert, cutoff time.Time) {
|
|
if counts == nil {
|
|
return
|
|
}
|
|
quality := alerts.CalculateAlertQualitySnapshot(history, nil, cutoff, cutoff.Add(30*24*time.Hour))
|
|
counts.AlertQuality.Add(quality)
|
|
counts.AlertsFired30d += quality.Fired30d
|
|
counts.AlertsAcknowledged30d += quality.Acknowledged30d
|
|
counts.AlertsResolved30d += quality.Resolved30d
|
|
}
|
|
|
|
func accumulateInstallSnapshotCounts(counts *InstallSnapshotCounts, monitor *Monitor) {
|
|
if counts == nil || monitor == nil {
|
|
return
|
|
}
|
|
|
|
readState := monitor.GetUnifiedReadStateOrSnapshot()
|
|
if readState != nil {
|
|
counts.PVENodes += len(readState.Nodes())
|
|
counts.PBSInstances += len(readState.PBSInstances())
|
|
counts.PMGInstances += len(readState.PMGInstances())
|
|
counts.VMs += len(readState.VMs())
|
|
counts.Containers += len(readState.Containers())
|
|
counts.AgentHosts += len(readState.Hosts())
|
|
counts.DockerHosts += len(readState.DockerHosts())
|
|
counts.DockerContainers += len(readState.DockerContainers())
|
|
counts.KubernetesClusters += len(readState.K8sClusters())
|
|
counts.KubernetesNodes += len(readState.K8sNodes())
|
|
counts.KubernetesPods += len(readState.Pods())
|
|
counts.KubernetesDeployments += len(readState.K8sDeployments())
|
|
counts.StoragePools += len(readState.StoragePools())
|
|
counts.PhysicalDisks += len(readState.PhysicalDisks())
|
|
}
|
|
resources, _ := monitor.UnifiedResourceSnapshot()
|
|
accumulateInstallSnapshotUnifiedResourceCounts(counts, resources)
|
|
counts.ActiveAlerts += len(monitor.ActiveAlertsSnapshot())
|
|
}
|
|
|
|
func accumulateInstallSnapshotUnifiedResourceCounts(counts *InstallSnapshotCounts, resources []unifiedresources.Resource) {
|
|
if counts == nil {
|
|
return
|
|
}
|
|
for _, resource := range resources {
|
|
resourceType := unifiedresources.CanonicalResourceType(resource.Type)
|
|
switch resourceType {
|
|
case unifiedresources.ResourceTypeCeph:
|
|
counts.CephClusters++
|
|
case unifiedresources.ResourceTypeNetworkShare:
|
|
counts.NetworkShares++
|
|
}
|
|
|
|
if resource.Availability != nil && resource.Availability.Enabled {
|
|
counts.AvailabilityTargets++
|
|
}
|
|
|
|
if resource.TrueNAS != nil {
|
|
switch resourceType {
|
|
case unifiedresources.ResourceTypeAgent:
|
|
counts.TrueNASSystems++
|
|
case unifiedresources.ResourceTypeVM:
|
|
counts.TrueNASVMs++
|
|
case unifiedresources.ResourceTypeAppContainer:
|
|
counts.TrueNASApps++
|
|
}
|
|
}
|
|
|
|
if resource.VMware != nil {
|
|
switch resourceType {
|
|
case unifiedresources.ResourceTypeAgent:
|
|
counts.VMwareHosts++
|
|
case unifiedresources.ResourceTypeVM:
|
|
counts.VMwareVMs++
|
|
case unifiedresources.ResourceTypeStorage:
|
|
counts.VMwareDatastores++
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Stop stops the monitor
|
|
func (rm *ReloadableMonitor) Stop() {
|
|
rm.mu.Lock()
|
|
defer rm.mu.Unlock()
|
|
|
|
if rm.cancel != nil {
|
|
rm.cancel()
|
|
}
|
|
|
|
if rm.mtMonitor != nil {
|
|
rm.mtMonitor.Stop()
|
|
}
|
|
}
|