407 lines
13 KiB
Go
407 lines
13 KiB
Go
// Package routing coordinates Controller-side automatic Sequential switching.
|
|
// It consumes bounded Provider result notifications and keeps the actual state
|
|
// transition in the existing management-store compare-and-swap boundary.
|
|
package routing
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"errors"
|
|
"reflect"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/controller/provider"
|
|
"proxy-pool/internal/domain/adminstate"
|
|
"proxy-pool/internal/domain/upstream"
|
|
)
|
|
|
|
var ErrInvalidSequentialCoordinator = errors.New("invalid sequential routing coordinator")
|
|
|
|
const (
|
|
defaultPollInterval = time.Second
|
|
autoSwitchActor = "proxy-controller"
|
|
autoSwitchReason = "consecutive empty provider fetches reached routing threshold"
|
|
autoStopReason = "consecutive empty provider fetches reached terminal sequential end"
|
|
disabledSwitchReason = "current sequential upstream is disabled"
|
|
disabledStopReason = "current sequential upstream is disabled and no eligible successor remains"
|
|
)
|
|
|
|
// ConfigurationSource supplies one immutable configuration and its matching
|
|
// authoritative management revision.
|
|
type ConfigurationSource interface {
|
|
Snapshot() (*config.Config, uint64)
|
|
}
|
|
|
|
// StateStore contains the only mutation boundary for routing current-upstream
|
|
// state. Implementations enforce ExpectedCurrent as a compare-and-swap fence.
|
|
type StateStore interface {
|
|
Snapshot(context.Context) (adminstate.Snapshot, error)
|
|
SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error)
|
|
DisableRouting(context.Context, adminstate.DisableRoutingCommand) (adminstate.MutationResult, error)
|
|
}
|
|
|
|
type SnapshotRefreshNotifier interface {
|
|
NotifySnapshotRefresh()
|
|
}
|
|
|
|
type Options struct {
|
|
Now func() time.Time
|
|
PollInterval time.Duration
|
|
}
|
|
|
|
type TickResult struct {
|
|
Switched int
|
|
Stopped int
|
|
}
|
|
|
|
// SequentialCoordinator observes Provider fetch outcomes without blocking the
|
|
// fetch path. Tick serializes bounded scans and only records a generation as
|
|
// processed after it has made, or intentionally declined, a decision.
|
|
type SequentialCoordinator struct {
|
|
configuration ConfigurationSource
|
|
state StateStore
|
|
stats provider.StatsReader
|
|
refresh SnapshotRefreshNotifier
|
|
now func() time.Time
|
|
pollInterval time.Duration
|
|
notify chan struct{}
|
|
|
|
mu sync.Mutex
|
|
processed map[string]uint64
|
|
}
|
|
|
|
var _ provider.ResultObserver = (*SequentialCoordinator)(nil)
|
|
|
|
func NewSequentialCoordinator(
|
|
configuration ConfigurationSource,
|
|
state StateStore,
|
|
stats provider.StatsReader,
|
|
refresh SnapshotRefreshNotifier,
|
|
options Options,
|
|
) (*SequentialCoordinator, error) {
|
|
if nilInterface(configuration) || nilInterface(state) || nilInterface(stats) || options.Now == nil {
|
|
return nil, ErrInvalidSequentialCoordinator
|
|
}
|
|
if options.PollInterval == 0 {
|
|
options.PollInterval = defaultPollInterval
|
|
}
|
|
if options.PollInterval <= 0 {
|
|
return nil, ErrInvalidSequentialCoordinator
|
|
}
|
|
return &SequentialCoordinator{
|
|
configuration: configuration,
|
|
state: state,
|
|
stats: stats,
|
|
refresh: refresh,
|
|
now: options.Now,
|
|
pollInterval: options.PollInterval,
|
|
notify: make(chan struct{}, 1),
|
|
processed: make(map[string]uint64),
|
|
}, nil
|
|
}
|
|
|
|
// ObserveProviderResult deliberately performs no store calls. Every validated
|
|
// result can affect an empty episode, but one buffered notification is enough
|
|
// because Tick reads the latest bounded Stats snapshot.
|
|
func (coordinator *SequentialCoordinator) ObserveProviderResult(result provider.Result) {
|
|
if coordinator == nil || !validResultClass(result.Class) || result.UpstreamID == "" {
|
|
return
|
|
}
|
|
select {
|
|
case coordinator.notify <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (coordinator *SequentialCoordinator) Run(ctx context.Context) error {
|
|
if coordinator == nil || ctx == nil || coordinator.pollInterval <= 0 || coordinator.now == nil || coordinator.notify == nil {
|
|
return ErrInvalidSequentialCoordinator
|
|
}
|
|
ticker := time.NewTicker(coordinator.pollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
// A temporary management-store failure must not terminate Provider
|
|
// reconciliation. The retained empty generation is evaluated on retry.
|
|
_, _ = coordinator.Tick(ctx)
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-coordinator.notify:
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, error) {
|
|
if coordinator == nil || ctx == nil || nilInterface(coordinator.configuration) || nilInterface(coordinator.state) ||
|
|
nilInterface(coordinator.stats) || coordinator.now == nil {
|
|
return TickResult{}, ErrInvalidSequentialCoordinator
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return TickResult{}, err
|
|
}
|
|
coordinator.mu.Lock()
|
|
defer coordinator.mu.Unlock()
|
|
|
|
configuration, revision := coordinator.configuration.Snapshot()
|
|
if configuration == nil || revision == 0 {
|
|
return TickResult{}, ErrInvalidSequentialCoordinator
|
|
}
|
|
snapshot, err := coordinator.state.Snapshot(ctx)
|
|
if err != nil {
|
|
return TickResult{}, err
|
|
}
|
|
if snapshot.Config == nil || snapshot.Config.Revision != revision {
|
|
return TickResult{}, ErrInvalidSequentialCoordinator
|
|
}
|
|
|
|
upstreams := upstreamStates(snapshot.Upstreams)
|
|
routings := routingStates(snapshot.Routings)
|
|
result := TickResult{}
|
|
for _, route := range configuration.Routing {
|
|
if !route.Enabled || route.Strategy.Type != "sequential" || route.Strategy.SwitchAfterEmptyFetch <= 0 {
|
|
continue
|
|
}
|
|
state, exists := routings[route.Name]
|
|
if !exists || !state.Enabled {
|
|
continue
|
|
}
|
|
if !upstreamEnabled(configuration, upstreams, state.CurrentUpstream) {
|
|
transition := disabledCurrentTransition(route, state.CurrentUpstream, upstreams, configuration)
|
|
mutated, mutateErr := coordinator.applyTransition(
|
|
ctx, route.Name, state.CurrentUpstream, transition, "disabled", state.Revision, disabledSwitchReason, disabledStopReason,
|
|
)
|
|
if mutateErr != nil {
|
|
if errors.Is(mutateErr, adminstate.ErrConflict) {
|
|
continue
|
|
}
|
|
return result, mutateErr
|
|
}
|
|
result.Switched += mutated.Switched
|
|
result.Stopped += mutated.Stopped
|
|
continue
|
|
}
|
|
read := coordinator.stats.ReadProviderStats([]string{state.CurrentUpstream})
|
|
if len(read) != 1 {
|
|
return result, ErrInvalidSequentialCoordinator
|
|
}
|
|
stats := read[0]
|
|
if stats.UpstreamID != state.CurrentUpstream || stats.ConsecutiveEmptyFetch < int64(route.Strategy.SwitchAfterEmptyFetch) ||
|
|
stats.EmptyGeneration == 0 {
|
|
coordinator.clearProcessed(route.Name, state.CurrentUpstream)
|
|
continue
|
|
}
|
|
key := processedKey(route.Name, state.CurrentUpstream)
|
|
if coordinator.processed[key] == stats.EmptyGeneration {
|
|
continue
|
|
}
|
|
transition := nextTransition(route, state.CurrentUpstream, upstreams, configuration)
|
|
mutated, mutateErr := coordinator.applyTransition(
|
|
ctx, route.Name, state.CurrentUpstream, transition, "empty", stats.EmptyGeneration, autoSwitchReason, autoStopReason,
|
|
)
|
|
if mutateErr != nil {
|
|
if errors.Is(mutateErr, adminstate.ErrConflict) {
|
|
coordinator.processed[key] = stats.EmptyGeneration
|
|
continue
|
|
}
|
|
return result, mutateErr
|
|
}
|
|
coordinator.processed[key] = stats.EmptyGeneration
|
|
result.Switched += mutated.Switched
|
|
result.Stopped += mutated.Stopped
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
// applyTransition centralizes the mutation/refresh side effect for both the
|
|
// Provider-empty and management-disabled paths. ExpectedCurrent remains the
|
|
// only mutation fence, so concurrent Controller replicas cannot skip states.
|
|
func (coordinator *SequentialCoordinator) applyTransition(
|
|
ctx context.Context,
|
|
routeName string,
|
|
current string,
|
|
transition sequentialTransition,
|
|
origin string,
|
|
version uint64,
|
|
switchReason string,
|
|
stopReason string,
|
|
) (TickResult, error) {
|
|
if transition.stop {
|
|
mutation, err := coordinator.state.DisableRouting(ctx, adminstate.DisableRoutingCommand{
|
|
RequestID: requestID(origin+"-stop", routeName, current, version),
|
|
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
|
|
Name: routeName, ExpectedCurrent: current, Reason: stopReason,
|
|
})
|
|
if err != nil {
|
|
return TickResult{}, err
|
|
}
|
|
if !mutation.Changed {
|
|
return TickResult{}, nil
|
|
}
|
|
if coordinator.refresh != nil {
|
|
coordinator.refresh.NotifySnapshotRefresh()
|
|
}
|
|
return TickResult{Stopped: 1}, nil
|
|
}
|
|
if transition.target == "" {
|
|
return TickResult{}, nil
|
|
}
|
|
mutation, err := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
|
|
RequestID: requestID(origin+"-switch", routeName, current, version),
|
|
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
|
|
Name: routeName, ExpectedCurrent: current, Target: transition.target, Reason: switchReason,
|
|
})
|
|
if err != nil {
|
|
return TickResult{}, err
|
|
}
|
|
if !mutation.Changed {
|
|
return TickResult{}, nil
|
|
}
|
|
if coordinator.refresh != nil {
|
|
coordinator.refresh.NotifySnapshotRefresh()
|
|
}
|
|
return TickResult{Switched: 1}, nil
|
|
}
|
|
|
|
type sequentialTransition struct {
|
|
target string
|
|
stop bool
|
|
}
|
|
|
|
func nextTransition(
|
|
route config.Routing,
|
|
current string,
|
|
states map[string]adminstate.UpstreamState,
|
|
configuration *config.Config,
|
|
) sequentialTransition {
|
|
eligible := make([]string, 0, len(route.Upstreams))
|
|
currentIndex := -1
|
|
for _, upstream := range route.Upstreams {
|
|
if !upstreamEnabled(configuration, states, upstream) {
|
|
continue
|
|
}
|
|
if upstream == current {
|
|
currentIndex = len(eligible)
|
|
}
|
|
eligible = append(eligible, upstream)
|
|
}
|
|
if currentIndex < 0 {
|
|
return sequentialTransition{}
|
|
}
|
|
if currentIndex+1 < len(eligible) {
|
|
return sequentialTransition{target: eligible[currentIndex+1]}
|
|
}
|
|
switch route.Strategy.EndBehavior {
|
|
case "loop":
|
|
if eligible[0] != current {
|
|
return sequentialTransition{target: eligible[0]}
|
|
}
|
|
case "stop", "":
|
|
return sequentialTransition{stop: true}
|
|
case "stayLast":
|
|
return sequentialTransition{}
|
|
default:
|
|
return sequentialTransition{}
|
|
}
|
|
return sequentialTransition{}
|
|
}
|
|
|
|
// disabledCurrentTransition treats an administratively disabled current
|
|
// upstream as a permanent advance. stayLast cannot retain a disabled current,
|
|
// so only loop may wrap; every other terminal condition stops the Routing.
|
|
func disabledCurrentTransition(
|
|
route config.Routing,
|
|
current string,
|
|
states map[string]adminstate.UpstreamState,
|
|
configuration *config.Config,
|
|
) sequentialTransition {
|
|
currentIndex := -1
|
|
for index, upstream := range route.Upstreams {
|
|
if upstream == current {
|
|
currentIndex = index
|
|
break
|
|
}
|
|
}
|
|
if currentIndex < 0 {
|
|
return sequentialTransition{stop: true}
|
|
}
|
|
for index := currentIndex + 1; index < len(route.Upstreams); index++ {
|
|
if upstreamEnabled(configuration, states, route.Upstreams[index]) {
|
|
return sequentialTransition{target: route.Upstreams[index]}
|
|
}
|
|
}
|
|
if route.Strategy.EndBehavior == "loop" {
|
|
for index := 0; index < currentIndex; index++ {
|
|
if upstreamEnabled(configuration, states, route.Upstreams[index]) {
|
|
return sequentialTransition{target: route.Upstreams[index]}
|
|
}
|
|
}
|
|
}
|
|
return sequentialTransition{stop: true}
|
|
}
|
|
|
|
func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState {
|
|
result := make(map[string]adminstate.UpstreamState, len(values))
|
|
for _, value := range values {
|
|
result[value.Name] = value
|
|
}
|
|
return result
|
|
}
|
|
|
|
func routingStates(values []adminstate.RoutingState) map[string]adminstate.RoutingState {
|
|
result := make(map[string]adminstate.RoutingState, len(values))
|
|
for _, value := range values {
|
|
result[value.Name] = value
|
|
}
|
|
return result
|
|
}
|
|
|
|
func upstreamEnabled(configuration *config.Config, states map[string]adminstate.UpstreamState, name string) bool {
|
|
configured, exists := configuration.Upstreams[name]
|
|
if !exists || !configured.Enabled {
|
|
return false
|
|
}
|
|
state, exists := states[name]
|
|
return exists && state.Enabled
|
|
}
|
|
|
|
func (coordinator *SequentialCoordinator) clearProcessed(routeName, upstream string) {
|
|
delete(coordinator.processed, processedKey(routeName, upstream))
|
|
}
|
|
|
|
func processedKey(routeName, upstream string) string {
|
|
return routeName + "\x00" + upstream
|
|
}
|
|
|
|
func requestID(action, routeName, upstream string, generation uint64) string {
|
|
payload := action + "\x00" + routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10)
|
|
digest := sha256.Sum256([]byte(payload))
|
|
return "auto-sequential-" + hex.EncodeToString(digest[:16])
|
|
}
|
|
|
|
func validResultClass(class upstream.FetchClass) bool {
|
|
switch class {
|
|
case upstream.FetchValid, upstream.FetchEmpty, upstream.FetchDuplicateOnly, upstream.FetchError:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func nilInterface(value any) bool {
|
|
if value == nil {
|
|
return true
|
|
}
|
|
reflected := reflect.ValueOf(value)
|
|
switch reflected.Kind() {
|
|
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
|
|
return reflected.IsNil()
|
|
default:
|
|
return false
|
|
}
|
|
}
|