codit/backend/internal/monitor/manager.go

package monitor

import "context"
import "crypto/tls"
import "crypto/x509"
import "strings"
import "sync"
import "time"

import "codit/config"
import "codit/internal/db"
import "codit/internal/models"
import codit_logger "codit/logger"

const logIDMonitor string = "monitor"

const monitorTickInterval time.Duration = 5 * time.Second
const monitorDueBatch int = 32
const monitorMaxConcurrent int = 8
const monitorCheckGrace time.Duration = 5 * time.Second
const monitorCAPoolTTL time.Duration = 5 * time.Minute
const monitorRetentionInterval time.Duration = 24 * time.Hour

// Manager runs the monitor check loop. It mirrors rpm.MirrorManager's lifecycle
// (Start/Stop/Wait + a ticker loop that picks due work). State-transition events
// are written transactionally by the store (FinishMonitorCheck); external
// delivery is handled uniformly by the notify.Relay draining the events outbox,
// so the manager has no notifier of its own.
type Manager struct {
	store     *db.Store
	logger    codit_logger.Logger
	policy    EgressPolicy
	productID string
	summary   *SummaryCache
	sem       chan struct{}
	// auditRetentionDays comes from the config file (audit.retention-days); <=0
	// keeps audit records indefinitely. Swept alongside the other retention jobs.
	auditRetentionDays int
	stopCh    chan struct{}
	stopOnce  sync.Once
	wg        sync.WaitGroup
	cancel    context.CancelFunc
	// Cached pool of this deployment's PKI CA certs, used to flag server certs
	// issued by an internal CA. Rebuilt lazily on a TTL (CAs change rarely).
	caMu     sync.Mutex
	caPool   *x509.CertPool
	caPoolAt time.Time
}

// NewManager builds the monitor manager. serverId identifies this deployment and
// is used to compose the probe's User-Agent and ICMP payload (no hardcoded
// product name).
func NewManager(store *db.Store, logger codit_logger.Logger, policy EgressPolicy, serverId string) *Manager {
	return &Manager{
		store:     store,
		logger:    logger,
		policy:    policy,
		productID: config.NormalizeServerId(serverId),
		summary:   NewSummaryCache(store, summaryTTL),
		sem:       make(chan struct{}, monitorMaxConcurrent),
		stopCh:    make(chan struct{}),
	}
}

// Summary exposes the derived-summary cache so handlers can serve uptime/beats
// on a slower cadence than the list is polled.
func (m *Manager) Summary() *SummaryCache {
	return m.summary
}

// ourCAPool returns this deployment's active PKI CA certs as a pool, rebuilt on
// a TTL. On a load error the previous (possibly nil) pool is kept.
func (m *Manager) ourCAPool() *x509.CertPool {
	var cas []models.PKICA
	var ca models.PKICA
	var pool *x509.CertPool
	var err error

	m.caMu.Lock()
	defer m.caMu.Unlock()
	if m.caPool != nil && time.Since(m.caPoolAt) < monitorCAPoolTTL {
		return m.caPool
	}
	cas, err = m.store.ListPKICAs()
	if err != nil {
		return m.caPool
	}
	pool = x509.NewCertPool()
	for _, ca = range cas {
		if ca.Status != models.PKIStatusActive {
			continue
		}
		pool.AppendCertsFromPEM([]byte(ca.CertPEM))
	}
	m.caPool = pool
	m.caPoolAt = time.Now()
	return pool
}

// resolveClientCert loads the referenced PKI client cert (cert+key) as a TLS
// certificate for mTLS. Returns nil (logged) if the ref is missing or invalid.
func (m *Manager) resolveClientCert(ref string) *tls.Certificate {
	var cert models.PKICert
	var pair tls.Certificate
	var err error

	cert, err = m.store.GetPKICert(strings.TrimSpace(ref))
	if err != nil {
		m.log(codit_logger.LOG_WARN, "client cert %q not found: %v", ref, err)
		return nil
	}
	pair, err = tls.X509KeyPair([]byte(cert.CertPEM), []byte(cert.KeyPEM))
	if err != nil {
		m.log(codit_logger.LOG_ERROR, "client cert %q unusable: %v", ref, err)
		return nil
	}
	return &pair
}

// resolveTLSContext assembles the deployment TLS material for a check: our CA
// pool (for issuer-known) plus, for a url monitor configured with a client cert
// ref, the client certificate to present for mTLS.
func (m *Manager) resolveTLSContext(monitor models.Monitor) *TLSContext {
	var tc TLSContext
	tc.OurCAs = m.ourCAPool()
	if monitor.Type == models.MonitorTypeURL {
		var cfg urlCheckConfig
		cfg = parseURLConfig(monitor.Config)
		if strings.TrimSpace(cfg.ClientCertRef) != "" {
			tc.ClientCert = m.resolveClientCert(cfg.ClientCertRef)
		}
	}
	return &tc
}

func (m *Manager) log(level codit_logger.LogLevel, format string, args ...interface{}) {
	if m.logger != nil {
		m.logger.Write(logIDMonitor, level, format, args...)
	}
}

func (m *Manager) Start(ctx context.Context) error {
	var runCtx context.Context
	var err error

	err = m.store.ResetRunningMonitors(ctx)
	if err != nil {
		m.log(codit_logger.LOG_ERROR, "reset running monitors failed err=%v", err)
		return err
	}
	m.runRetention()
	_ = m.summary.Refresh(ctx) // warm the derived-summary cache

	runCtx, m.cancel = context.WithCancel(ctx)
	m.wg.Add(1)
	go m.loop(runCtx)
	return nil
}

// runRetention prunes heartbeat history (per monitor) and the shared events feed.
// It runs once at startup and about daily from the loop.
func (m *Manager) runRetention() {
	var monitors []models.MonitorWithStatus
	var i int
	var err error

	var keepMax int
	var keepDays int
	keepMax = int(m.store.SettingInt(db.SettingRetentionMonitorCheckMax))
	keepDays = int(m.store.SettingInt(db.SettingRetentionMonitorCheckDays))

	// we don't use the same transaction as this operation is on a best-effort basis.
	monitors, err = m.store.ListMonitors()
	if err == nil {
		for i = 0; i < len(monitors); i++ {
			_ = m.store.CleanupMonitorChecksRetention(monitors[i].ID, keepMax, keepDays)
		}
	}

	// [TODO]
	//  The events table is generic and contains events from other background jobs like rpm mirroring.
	//  We have to migrate this event sweep to somewhere more neutral. For now, let's keep it here.
	//  As long as the manager job runs, the event from other jobs are cleaned up properly.
	_ = m.store.CleanupEventsRetention(int(m.store.SettingInt(db.SettingRetentionEventDays)))

	// Admin audit log retention is config-file driven (audit.retention-days).
	_ = m.store.CleanupAuditLogRetention(m.auditRetentionDays)
}

// SetAuditRetentionDays sets the audit-log retention window (days) from config.
// A value <= 0 disables purging (keep indefinitely).
func (m *Manager) SetAuditRetentionDays(days int) {
	m.auditRetentionDays = days
}

func (m *Manager) Stop() {
	m.stopOnce.Do(func() {
		if m.cancel != nil {
			m.cancel()
		}
		close(m.stopCh)
	})
}

func (m *Manager) Wait() {
	m.wg.Wait()
}

func (m *Manager) loop(ctx context.Context) {
	var ticker *time.Ticker
	var lastRetention time.Time
	var lastSummary time.Time
	defer m.wg.Done()
	ticker = time.NewTicker(monitorTickInterval)
	defer ticker.Stop()

	lastRetention = time.Now() // Start() already ran an immediate sweep
	lastSummary = time.Now()   // Start() already warmed the summary cache
	m.runDue(ctx)
	for {
		select {
		case <-ctx.Done():
			return
		case <-m.stopCh:
			return
		case <-ticker.C:
			m.runDue(ctx)
			if time.Since(lastSummary) >= summaryRefreshInterval {
				_ = m.summary.Refresh(ctx)
				lastSummary = time.Now()
			}
			if time.Since(lastRetention) >= monitorRetentionInterval {
				m.runRetention()
				lastRetention = time.Now()
			}
		}
	}
}

func (m *Manager) runDue(ctx context.Context) {
	var due []models.Monitor
	var now int64
	var checkID int64
	var started bool
	var i int
	var err error
	var writeCtx context.Context
	var writeCancel context.CancelFunc

	now = time.Now().UTC().Unix()
	due, err = m.store.ListDueMonitors(now, monitorDueBatch)
	if err != nil {
		m.log(codit_logger.LOG_ERROR, "list due monitors failed err=%v", err)
		return
	}

	for i = 0; i < len(due); i++ {
		select {
		case <-ctx.Done():
			return
		case <-m.stopCh:
			return
		default:
		}
		// Independent bounded write context so claiming the check does not get
		// skipped by loop cancellation mid-tick.
		writeCtx, writeCancel = context.WithTimeout(context.Background(), db.IndependentDBOperationTimeout)
		checkID, started, err = m.store.TryStartMonitorCheck(writeCtx, due[i].ID, now)
		writeCancel()
		if err != nil {
			m.log(codit_logger.LOG_ERROR, "try start check failed monitor=%s err=%v", due[i].ID, err)
			continue
		}
		if !started {
			continue
		}
		m.sem <- struct{}{}
		m.wg.Add(1)
		go m.runOne(ctx, due[i], checkID)
	}
}

func (m *Manager) runOne(ctx context.Context, monitor models.Monitor, checkID int64) {
	var checkCtx context.Context
	var checkCancel context.CancelFunc
	var result models.MonitorCheckResult
	var transitioned bool
	var newState string
	var now int64
	var writeCtx context.Context
	var writeCancel context.CancelFunc
	var err error

	defer func() {
		<-m.sem
		m.wg.Done()
	}()

	checkCtx, checkCancel = context.WithTimeout(ctx, monitorCheckGrace + (time.Duration(monitor.TimeoutSec) * time.Second))
	result = runCheck(checkCtx, m.policy, monitor, m.productID, m.resolveTLSContext(monitor))
	checkCancel()

	now = time.Now().UTC().Unix()
	writeCtx, writeCancel = context.WithTimeout(context.Background(), db.IndependentDBOperationTimeout)
	// The transition event is written to the events outbox in the same tx; the
	// notify.Relay drains it for external delivery, so no fan-out here.
	transitioned, newState, _, _, err = m.store.FinishMonitorCheck(writeCtx, monitor.ID, checkID, result, now, monitor.FailureThreshold, monitor.IntervalSec)
	writeCancel()
	if err != nil {
		m.log(codit_logger.LOG_ERROR, "finish check failed monitor=%s err=%v", monitor.ID, err)
		return
	}
	if transitioned {
		m.log(codit_logger.LOG_INFO, "monitor=%s state=%s ok=%t msg=%q", monitor.ID, newState, result.OK, result.Message)
	}
}