worker/internal/distworker/peer.go

package distworker

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"rocketgit.ru/rsmon/worker/internal/wire"
)

// peerStatusPath is the HTTP path the worker webapp serves at
// GET /api/peer/status. Kept on a single constant so the polling
// client and the handler can never drift.
const peerStatusPath = "/api/peer/status"

// peerStatusTimeout bounds how long a single peer probe is allowed
// to take. Short enough that a sluggish peer does not stall the
// selfcheck loop; long enough to absorb a single TCP retransmit.
const peerStatusTimeout = 5 * time.Second

// peerStatusMaxAge is how stale a peer's last observation is allowed
// to get before the consensus treats it as "unknown" (excluded from
// the vote). Equal to two poll intervals so a single dropped probe
// does not immediately disqualify a peer.
const peerStatusMaxAge = 2 * peerPollInterval

// peerPollInterval drives the peer-poller cadence. Kept equal to the
// local selfcheck interval so observations stay aligned.
const peerPollInterval = 30 * time.Second

// PeerStatus is the JSON a worker returns at GET /api/peer/status.
// The shape is stable so peer workers can pin against it without
// coordinating a schema bump. ObservedAt is the wall-clock time the
// local selfcheck produced this status; peers ignore entries older
// than peerStatusMaxAge.
type PeerStatus struct {
	WorkerID   string    `json:"worker_id"`
	Up         bool      `json:"up"`
	ObservedAt time.Time `json:"observed_at"`
}

// peerObservation is the per-peer entry the consensus uses. The
// selfcheck module does not reach into the wire type directly; it
// always works with the cached peerObservation so the JSON shape can
// evolve without forcing a selfcheck refactor.
type peerObservation struct {
	WorkerID   string
	Up         bool
	ObservedAt time.Time
	Err        error // transient fetch error; nil = observation is valid
}

// peerCache holds the latest observation per peer worker_id. The map
// is guarded by a single RWMutex because reads (consensus) vastly
// outnumber writes (one per peer per poll tick). Entries survive
// across applyInit refreshes; the peer poller overwrites the
// per-worker slot every tick.
type peerCache struct {
	mu    sync.RWMutex
	items map[string]peerObservation
}

func newPeerCache() *peerCache {
	return &peerCache{items: map[string]peerObservation{}}
}

// snapshot returns a defensive copy of the current observations in
// no particular order. Callers must not mutate the returned slice.
func (c *peerCache) snapshot() []peerObservation {
	if c == nil {
		return nil
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	out := make([]peerObservation, 0, len(c.items))
	for _, v := range c.items {
		out = append(out, v)
	}
	return out
}

// put stores a fresh observation. The poller calls this every tick;
// the consensus reads via snapshot().
func (c *peerCache) put(obs peerObservation) {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.items[obs.WorkerID] = obs
	c.mu.Unlock()
}

// resetFor replaces the cache contents with one fresh observation per
// peer in the supplied list. Used when applyInit refreshes the peer
// set so a removed peer's stale observation is dropped immediately
// rather than lingering until peerStatusMaxAge expires.
func (c *peerCache) resetFor(peers []wire.PeerInfo) {
	if c == nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.items = make(map[string]peerObservation, len(peers))
	for i := range peers {
		p := peers[i]
		if p.WorkerID == "" {
			continue
		}
		c.items[p.WorkerID] = peerObservation{WorkerID: p.WorkerID}
	}
}

// fetchPeerStatus issues a single GET /api/peer/status to the given
// peer and decodes the response. Network errors and non-2xx codes
// are returned as a populated peerObservation with Err set so the
// caller can decide whether to update the cache (we do, to mark the
// peer as "we tried"). Caller is responsible for not calling this
// concurrently for the same peer in a way that would race the
// peerCache lock; the per-peer update is serialized via the cache.
//
// peer is taken by pointer to keep the wire.PeerInfo copy off the
// hot path; the function is called once per peer per poll interval.
func fetchPeerStatus(ctx context.Context, peer *wire.PeerInfo) peerObservation {
	obs := peerObservation{WorkerID: peer.WorkerID}
	if peer.URL == "" {
		obs.Err = fmt.Errorf("peer %s: empty url", peer.WorkerID)
		return obs
	}
	u, err := url.Parse(peer.URL)
	if err != nil || (u.Scheme != schemeHTTP && u.Scheme != schemeHTTPS) {
		obs.Err = fmt.Errorf("peer %s: bad url %q", peer.WorkerID, peer.URL)
		return obs
	}
	endpoint := strings.TrimRight(peer.URL, "/") + peerStatusPath
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, http.NoBody)
	if err != nil {
		obs.Err = fmt.Errorf("peer %s: build request: %w", peer.WorkerID, err)
		return obs
	}
	if peer.Login != "" {
		req.SetBasicAuth(peer.Login, peer.Password)
	}
	client := &http.Client{Timeout: peerStatusTimeout}
	resp, err := client.Do(req)
	if err != nil {
		obs.Err = fmt.Errorf("peer %s: get: %w", peer.WorkerID, err)
		return obs
	}
	defer resp.Body.Close() //nolint:errcheck
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		obs.Err = fmt.Errorf("peer %s: status %d", peer.WorkerID, resp.StatusCode)
		return obs
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 8*1024))
	if err != nil {
		obs.Err = fmt.Errorf("peer %s: read body: %w", peer.WorkerID, err)
		return obs
	}
	var status PeerStatus
	if err := json.Unmarshal(body, &status); err != nil {
		obs.Err = fmt.Errorf("peer %s: decode: %w", peer.WorkerID, err)
		return obs
	}
	obs.Up = status.Up
	obs.ObservedAt = status.ObservedAt
	return obs
}

// peerPollerLoop runs once per peerPollInterval and refreshes the
// cache. It exits when ctx is canceled. The loop runs each peer
// sequentially so a slow / unreachable peer does not spawn N
// concurrent goroutines that would overwhelm the local listener.
func (r *Runner) peerPollerLoop(ctx context.Context) {
	ticker := time.NewTicker(peerPollInterval)
	defer ticker.Stop()
	// Run once shortly after startup so a worker that boots while
	// the master is already down does not wait a full interval
	// before the first peer observation lands.
	first := time.NewTimer(2 * time.Second)
	defer first.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-first.C:
		case <-ticker.C:
		}
		r.pollPeersOnce(ctx)
	}
}

// pollPeersOnce fetches every cached peer's status and stores the
// results. Safe to call directly from tests to drive a deterministic
// poll cycle.
func (r *Runner) pollPeersOnce(ctx context.Context) {
	peers := r.Peers()
	if len(peers) == 0 {
		return
	}
	for i := range peers {
		p := peers[i]
		obs := fetchPeerStatus(ctx, &p)
		r.peerCache.put(obs)
		if obs.Err != nil {
			log.Printf("worker peer poll: %s: %v", p.WorkerID, obs.Err)
		}
	}
}

// peerObservationsForConsensus returns the peer observations the
// consensus should consider. A peer whose last observation is older
// than peerStatusMaxAge is dropped from the vote (treated as
// "unknown") so a worker that lost connectivity to one peer cannot
// single-handedly decide the cluster is healthy.
func (r *Runner) peerObservationsForConsensus(now time.Time) []peerObservation {
	all := r.peerCache.snapshot()
	out := make([]peerObservation, 0, len(all))
	for _, obs := range all {
		if obs.WorkerID == "" {
			continue
		}
		if obs.Err != nil {
			continue
		}
		if obs.ObservedAt.IsZero() {
			continue
		}
		if now.Sub(obs.ObservedAt) > peerStatusMaxAge {
			continue
		}
		out = append(out, obs)
	}
	return out
}