EvoBGP/internal/telemetry/collector.go

// Package telemetry polls legacy agent and BIRD endpoints into shared snapshots.
package telemetry

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"encoding/hex"
	"errors"
	"math/rand"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"evobgp/internal/birdfmt"
	"evobgp/internal/logging"
	"evobgp/internal/nodedispatch"
	"evobgp/internal/store"

	"github.com/google/uuid"
)

// Config bounds collector concurrency, freshness and external request duration.
type Config struct {
	Interval, StaleAfter, Timeout, Lease, BackoffMax time.Duration
	Workers                                          int
	BirdSocket, BirdBin, LocalAuthority              string
	HTTPClient                                       *http.Client
}

// ConfigFromEnv returns safe defaults and derives a stable local BIRD authority.
func ConfigFromEnv() Config {
	cfg := Config{Interval: 20 * time.Second, StaleAfter: time.Minute, Timeout: 10 * time.Second, Lease: 30 * time.Second, BackoffMax: 2 * time.Minute, Workers: 4}
	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_TELEMETRY_INTERVAL_SEC"))); err == nil && n >= 15 && n <= 300 {
		cfg.Interval = time.Duration(n) * time.Second
	}
	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_TELEMETRY_WORKERS"))); err == nil && n >= 1 && n <= 32 {
		cfg.Workers = n
	}
	cfg.StaleAfter = 3 * cfg.Interval
	if cfg.BackoffMax < 4*cfg.Interval {
		cfg.BackoffMax = 4 * cfg.Interval
	}
	cfg.BirdSocket = strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
	cfg.BirdBin = strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN"))
	if cfg.BirdSocket != "" {
		identity := strings.TrimSpace(os.Getenv("EVOBGP_TELEMETRY_LOCAL_ID"))
		if identity == "" {
			host, err := os.Hostname()
			if err == nil && host != "" {
				identity = host + "\x00" + cfg.BirdSocket + "\x00" + cfg.BirdBin
			}
		}
		if identity != "" {
			sum := sha256.Sum256([]byte(identity))
			cfg.LocalAuthority = hex.EncodeToString(sum[:12])
		}
	}
	return cfg
}

// Collector owns a fixed number of lease-claiming workers.
type Collector struct {
	cancel context.CancelFunc
	done   chan struct{}
	client *http.Client
}

// Start launches bounded polling until ctx cancellation or Close.
func Start(ctx context.Context, st store.TelemetryStore, cfg Config) *Collector {
	workerCtx, cancel := context.WithCancel(ctx)
	c := &Collector{cancel: cancel, done: make(chan struct{})}
	if st == nil {
		close(c.done)
		return c
	}
	if cfg.Interval <= 0 {
		cfg.Interval = 20 * time.Second
	}
	if cfg.Timeout <= 0 {
		cfg.Timeout = 10 * time.Second
	}
	if cfg.Lease <= cfg.Timeout+5*time.Second {
		cfg.Lease = cfg.Timeout + 20*time.Second
	}
	if cfg.BackoffMax < 4*cfg.Interval {
		cfg.BackoffMax = 4 * cfg.Interval
	}
	if cfg.Workers <= 0 {
		cfg.Workers = 4
	}
	if cfg.Workers > 32 {
		cfg.Workers = 32
	}
	if cfg.HTTPClient != nil {
		c.client = cfg.HTTPClient
	} else {
		c.client = newPollHTTPClient(cfg)
	}
	var wg sync.WaitGroup
	for i := 0; i < cfg.Workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			worker(workerCtx, st, cfg, c.client, uuid.NewString())
		}()
	}
	go func() { wg.Wait(); close(c.done) }()
	return c
}

// Close stops claims and waits for in-flight polls to finish or honor cancellation.
func (c *Collector) Close(ctx context.Context) error {
	if c == nil {
		return nil
	}
	c.cancel()
	select {
	case <-c.done:
		if c.client != nil {
			c.client.CloseIdleConnections()
		}
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func newPollHTTPClient(cfg Config) *http.Client {
	transport := http.DefaultTransport.(*http.Transport).Clone()
	transport.MaxIdleConns = cfg.Workers * 4
	transport.MaxIdleConnsPerHost = cfg.Workers * 2
	if strings.TrimSpace(os.Getenv("EVOBGP_NODE_DISPATCH_INSECURE_TLS")) == "1" {
		transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec // lab-only legacy dispatch setting
	}
	return &http.Client{Transport: transport, Timeout: cfg.Timeout}
}

func worker(ctx context.Context, st store.TelemetryStore, cfg Config, client *http.Client, owner string) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		claimCtx, claimCancel := context.WithTimeout(ctx, 3*time.Second)
		claims, err := st.ClaimTelemetry(claimCtx, owner, 1, cfg.Lease, cfg.LocalAuthority)
		claimCancel()
		if err != nil {
			logging.With("component", "telemetry.collector").Warn("claim failed", "err", err)
			continue
		}
		if len(claims) == 0 {
			continue
		}
		claim := claims[0]
		pollCtx, cancel := context.WithTimeout(ctx, cfg.Timeout)
		result := poll(pollCtx, claim, cfg, client)
		cancel()
		failed := result.BGP.Error != "" || result.Health.Source != "" && !result.Health.OK
		next := time.Now().Add(nextDelay(cfg, claim.FailureCount, failed))
		persistCtx, persistCancel := context.WithTimeout(ctx, 3*time.Second)
		err = st.CompleteTelemetry(persistCtx, claim, result, next)
		persistCancel()
		if err != nil && !errors.Is(err, store.ErrTelemetryLeaseLost) && ctx.Err() == nil {
			logging.With("component", "telemetry.collector").Warn("completion failed", "err", err)
		}
	}
}

func nextDelay(cfg Config, priorFailures int, failed bool) time.Duration {
	delay := cfg.Interval
	if failed {
		for i := 0; i <= priorFailures && delay < cfg.BackoffMax; i++ {
			delay *= 2
		}
		if delay > cfg.BackoffMax {
			delay = cfg.BackoffMax
		}
	}
	return delay + time.Duration(rand.Int63n(int64(delay/5+1)))
}

func poll(ctx context.Context, claim store.TelemetryClaim, cfg Config, client *http.Client) store.TelemetryResult {
	if claim.Kind == "local_bird" {
		out, err := birdfmt.ShowProtocols(ctx, cfg.BirdSocket, cfg.BirdBin)
		received := time.Now().UTC()
		bgp := store.TelemetryBGP{TelemetryObservation: store.TelemetryObservation{ObservedAt: received, ReceivedAt: received, Source: "cp_birdc"}}
		if err != nil {
			bgp.Error = err.Error()
		} else {
			bgp.Sessions = birdfmt.ParseBGPSessions(out)
		}
		return store.TelemetryResult{BGP: bgp}
	}
	if claim.Speaker == nil {
		return store.TelemetryResult{BGP: store.TelemetryBGP{TelemetryObservation: store.TelemetryObservation{ObservedAt: time.Now().UTC(), ReceivedAt: time.Now().UTC(), Source: "agent_bird_protocols", Error: "speaker missing"}}}
	}
	var result store.TelemetryResult
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		res := nodedispatch.FetchBirdProtocols(ctx, claim.Speaker, nodedispatch.Options{Timeout: cfg.Timeout, HTTPClient: client})
		received := time.Now().UTC()
		result.BGP = store.TelemetryBGP{TelemetryObservation: store.TelemetryObservation{ObservedAt: received, ReceivedAt: received, Source: "agent_bird_protocols", Error: res.Error}, Sessions: res.Sessions}
	}()
	go func() {
		defer wg.Done()
		res := nodedispatch.FetchAgentHealth(ctx, claim.Speaker, nodedispatch.Options{Timeout: cfg.Timeout, HTTPClient: client})
		received := time.Now().UTC()
		result.Health = store.TelemetryHealth{TelemetryObservation: store.TelemetryObservation{ObservedAt: received, ReceivedAt: received, Source: "agent_health", Error: res.Error}, OK: res.OK, LastSyncAt: res.LastSyncAt, LastAppliedRevisionID: res.LastAppliedRevisionID}
	}()
	wg.Wait()
	return result
}