EvoBGP/internal/store/telemetry.go

package store

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"strconv"
	"strings"
	"time"

	"evobgp/internal/birdfmt"
)

// ErrTelemetryLeaseLost means a collector result belongs to an expired or replaced lease.
var ErrTelemetryLeaseLost = errors.New("store: telemetry lease lost")

// TelemetryLocalTargetID identifies the synthetic local BIRD target of a tenant.
func TelemetryLocalTargetID(tenantID, authority string) string {
	return "cp-local/" + tenantID + "/" + authority
}

// TelemetryObservation is the most recent attempt and last successful reading of one source.
type TelemetryObservation struct {
	ObservedAt    time.Time `json:"observed_at"`
	ReceivedAt    time.Time `json:"received_at"`
	LastSuccessAt time.Time `json:"last_success_at"`
	Source        string    `json:"source"`
	Error         string    `json:"error,omitempty"`
}

// TelemetryBGP is one BIRD protocols measurement.
type TelemetryBGP struct {
	TelemetryObservation
	Sessions []birdfmt.BGPSession `json:"sessions,omitempty"`
}

// TelemetryHealth is one legacy agent health measurement.
type TelemetryHealth struct {
	TelemetryObservation
	OK                    bool   `json:"ok"`
	LastSyncAt            string `json:"last_sync_at,omitempty"`
	LastAppliedRevisionID string `json:"last_applied_revision_id,omitempty"`
}

// TelemetrySnapshot contains independent BGP and agent health readings for one target.
type TelemetrySnapshot struct {
	TenantID, TargetID, SpeakerID, Kind string
	LocalAuthority                      string
	ConfigFingerprint                   string
	BGP                                 TelemetryBGP
	Health                              TelemetryHealth
	NextPollAt                          time.Time
	FailureCount                        int
	DiscoveryCursor                     int
}

// TelemetrySpeakerFingerprint binds a snapshot to the configuration it observed.
func TelemetrySpeakerFingerprint(sp *Speaker) string {
	if sp == nil {
		return ""
	}
	if strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
		sum := sha256.Sum256([]byte("master\x00" + sp.Endpoint))
		return hex.EncodeToString(sum[:])
	}
	meta := ParseSpeakerMeta(sp.MetaJSON)
	sum := sha256.Sum256([]byte(sp.Role + "\x00" + sp.Endpoint + "\x00" + meta.AgentDomain + "\x00" + meta.AgentSecret + "\x00" + strconv.Itoa(meta.AgentPort)))
	return hex.EncodeToString(sum[:])
}

// TelemetryClaim fences a bounded collector attempt. Speaker is the current config.
type TelemetryClaim struct {
	TelemetrySnapshot
	Speaker      *Speaker
	OwnerID      string
	FencingToken int64
}

// TelemetryResult is the atomic pair of results of one collector attempt.
type TelemetryResult struct {
	BGP    TelemetryBGP
	Health TelemetryHealth
}

// TelemetryDiscoveryBatch rotates over long protocol lists without starving later entries.
func TelemetryDiscoveryBatch(sessions []birdfmt.BGPSession, cursor, limit int) ([]birdfmt.BGPSession, int) {
	if len(sessions) == 0 || limit <= 0 {
		return nil, 0
	}
	if cursor < 0 || cursor >= len(sessions) {
		cursor = 0
	}
	out := make([]birdfmt.BGPSession, 0, limit)
	for visited := 0; visited < len(sessions); visited++ {
		idx := (cursor + visited) % len(sessions)
		if birdfmt.IsDynamicDiscoverySession(sessions[idx].Name) {
			out = append(out, sessions[idx])
			if len(out) == limit {
				return out, (idx + 1) % len(sessions)
			}
		}
	}
	return out, 0
}

// TelemetryStore coordinates polling and read-only snapshots across API replicas.
type TelemetryStore interface {
	ClaimTelemetry(context.Context, string, int, time.Duration, string) ([]TelemetryClaim, error)
	CompleteTelemetry(context.Context, TelemetryClaim, TelemetryResult, time.Time) error
	ListTelemetry(context.Context, string) ([]TelemetrySnapshot, error)
	RequestTelemetryRefresh(context.Context, string, time.Duration) (bool, error)
}