114 lines
3.7 KiB
Go
114 lines
3.7 KiB
Go
package health
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/domain/activitypool"
|
|
healthDomain "proxy-pool/internal/domain/health"
|
|
)
|
|
|
|
var ErrInvalidUnhealthyReaper = errors.New("invalid unhealthy reaper")
|
|
|
|
type UnhealthyReaperOptions struct {
|
|
PollInterval time.Duration
|
|
BatchSize int
|
|
Now func() time.Time
|
|
Metrics healthDomain.DrainMetricsObserver
|
|
}
|
|
|
|
// ConfiguredUnhealthyReaper applies the current upstream grace policies to a
|
|
// bounded activity-pool sweep. It performs no work when every policy disables
|
|
// removal, making the default behavior recovery-only.
|
|
type ConfiguredUnhealthyReaper struct {
|
|
configuration ConfigurationSource
|
|
store activitypool.UnhealthyRemover
|
|
options UnhealthyReaperOptions
|
|
}
|
|
|
|
func NewConfiguredUnhealthyReaper(
|
|
configuration ConfigurationSource,
|
|
store activitypool.UnhealthyRemover,
|
|
options UnhealthyReaperOptions,
|
|
) (*ConfiguredUnhealthyReaper, error) {
|
|
if nilInterface(configuration) || nilInterface(store) || options.PollInterval <= 0 || options.BatchSize <= 0 || options.Now == nil {
|
|
return nil, ErrInvalidUnhealthyReaper
|
|
}
|
|
return &ConfiguredUnhealthyReaper{configuration: configuration, store: store, options: options}, nil
|
|
}
|
|
|
|
func (reaper *ConfiguredUnhealthyReaper) Tick(ctx context.Context) (activitypool.UnhealthySweepResult, error) {
|
|
if ctx == nil || reaper == nil || nilInterface(reaper.configuration) || nilInterface(reaper.store) ||
|
|
reaper.options.PollInterval <= 0 || reaper.options.BatchSize <= 0 || reaper.options.Now == nil {
|
|
return activitypool.UnhealthySweepResult{}, ErrInvalidUnhealthyReaper
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return activitypool.UnhealthySweepResult{}, err
|
|
}
|
|
now := reaper.options.Now()
|
|
if now.IsZero() {
|
|
return activitypool.UnhealthySweepResult{}, ErrInvalidUnhealthyReaper
|
|
}
|
|
policies := unhealthyRemovalPolicies(reaper.configuration.Current())
|
|
if len(policies) == 0 {
|
|
return activitypool.UnhealthySweepResult{}, nil
|
|
}
|
|
utcNow := now.UTC()
|
|
result, err := reaper.store.SweepUnhealthy(ctx, activitypool.UnhealthySweepCommand{
|
|
Now: utcNow, Limit: reaper.options.BatchSize, RemoveAfterByUpstream: policies,
|
|
})
|
|
if err != nil {
|
|
return activitypool.UnhealthySweepResult{}, err
|
|
}
|
|
drainer, supported := reaper.store.(activitypool.UnhealthyDrainStarter)
|
|
if !supported || nilInterface(drainer) {
|
|
if reaper.options.Metrics != nil {
|
|
reaper.options.Metrics.ObserveDrain(healthDomain.DrainReasonUnhealthy, len(result.DrainCandidates), 0)
|
|
}
|
|
return result, nil
|
|
}
|
|
started := 0
|
|
for _, candidate := range result.DrainCandidates {
|
|
didStart, err := drainer.BeginUnhealthyDrain(ctx, utcNow, candidate)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
if didStart {
|
|
started++
|
|
}
|
|
}
|
|
if reaper.options.Metrics != nil {
|
|
reaper.options.Metrics.ObserveDrain(healthDomain.DrainReasonUnhealthy, len(result.DrainCandidates), started)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (reaper *ConfiguredUnhealthyReaper) Run(ctx context.Context) error {
|
|
if reaper == nil {
|
|
return ErrInvalidUnhealthyReaper
|
|
}
|
|
return runScheduler(ctx, reaper.options.PollInterval, func(tickCtx context.Context) (TickResult, error) {
|
|
_, err := reaper.Tick(tickCtx)
|
|
return TickResult{}, err
|
|
})
|
|
}
|
|
|
|
func unhealthyRemovalPolicies(configuration *config.Config) map[string]time.Duration {
|
|
if configuration == nil {
|
|
return nil
|
|
}
|
|
policies := make(map[string]time.Duration)
|
|
for upstreamID, upstream := range configuration.Upstreams {
|
|
if !upstream.Enabled {
|
|
continue
|
|
}
|
|
check := config.EffectiveCheck(configuration.Defaults.Check, upstream.Check)
|
|
if removeAfter := check.UnhealthyRemoveAfter.Value(); removeAfter > 0 {
|
|
policies[upstreamID] = removeAfter
|
|
}
|
|
}
|
|
return policies
|
|
}
|