87 lines
2.8 KiB
Go
87 lines
2.8 KiB
Go
package health
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/domain/activitypool"
|
|
)
|
|
|
|
var ErrInvalidUnhealthyReaper = errors.New("invalid unhealthy reaper")
|
|
|
|
type UnhealthyReaperOptions struct {
|
|
PollInterval time.Duration
|
|
BatchSize int
|
|
Now func() time.Time
|
|
}
|
|
|
|
// 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
|
|
}
|
|
return reaper.store.SweepUnhealthy(ctx, activitypool.UnhealthySweepCommand{
|
|
Now: now.UTC(), Limit: reaper.options.BatchSize, RemoveAfterByUpstream: policies,
|
|
})
|
|
}
|
|
|
|
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
|
|
}
|