325 lines
11 KiB
Go
325 lines
11 KiB
Go
package admin
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/domain/adminstate"
|
|
)
|
|
|
|
var ErrInvalidApplicationService = errors.New("invalid admin application service")
|
|
|
|
type StateRepository interface {
|
|
adminstate.Mutator
|
|
adminstate.SnapshotReader
|
|
}
|
|
|
|
type OperationalStatusReader interface {
|
|
ReadOperationalStatus(context.Context) (OperationalStatus, error)
|
|
}
|
|
|
|
type ConfigurationLoader interface {
|
|
LoadConfiguration(context.Context) (LoadedConfiguration, error)
|
|
}
|
|
|
|
// ConfigurationPublisher publishes only a newer authoritative configuration revision.
|
|
type ConfigurationPublisher interface {
|
|
PublishRevision(*config.Config, uint64) bool
|
|
}
|
|
|
|
type RuntimeController interface {
|
|
Notify()
|
|
ValidateConfiguration(context.Context, *config.Config) error
|
|
ValidateUpstream(context.Context, string) error
|
|
}
|
|
|
|
var _ ConfigurationPublisher = (*config.Store)(nil)
|
|
|
|
type ApplicationDependencies struct {
|
|
State StateRepository
|
|
Operations OperationalStatusReader
|
|
Configuration ConfigurationLoader
|
|
Publisher ConfigurationPublisher
|
|
Runtime RuntimeController
|
|
}
|
|
|
|
type ApplicationOptions struct {
|
|
Now func() time.Time
|
|
FingerprintKey []byte
|
|
}
|
|
|
|
type OperationalStatus struct {
|
|
SnapshotVersion uint64
|
|
Upstreams []UpstreamActivity
|
|
Workers []WorkerStatus
|
|
}
|
|
|
|
type UpstreamActivity struct {
|
|
Name string
|
|
Available int64
|
|
Checking int64
|
|
Suspect int64
|
|
Draining int64
|
|
Extracted int64
|
|
ConsecutiveEmptyFetch int64
|
|
FetchErrorCount int64
|
|
}
|
|
|
|
type LoadedConfiguration struct {
|
|
Value *config.Config
|
|
Source string
|
|
}
|
|
|
|
type ApplicationService struct {
|
|
state StateRepository
|
|
operations OperationalStatusReader
|
|
configuration ConfigurationLoader
|
|
publisher ConfigurationPublisher
|
|
runtime RuntimeController
|
|
now func() time.Time
|
|
fingerprintKey []byte
|
|
}
|
|
|
|
var _ Service = (*ApplicationService)(nil)
|
|
|
|
func NewApplicationService(dependencies ApplicationDependencies, options ApplicationOptions) (*ApplicationService, error) {
|
|
if nilInterface(dependencies.State) || nilInterface(dependencies.Operations) || nilInterface(dependencies.Configuration) ||
|
|
nilInterface(dependencies.Publisher) || options.Now == nil ||
|
|
len(options.FingerprintKey) < config.MinimumFingerprintKeyBytes {
|
|
return nil, ErrInvalidApplicationService
|
|
}
|
|
return &ApplicationService{
|
|
state: dependencies.State,
|
|
operations: dependencies.Operations,
|
|
configuration: dependencies.Configuration,
|
|
publisher: dependencies.Publisher,
|
|
runtime: dependencies.Runtime,
|
|
now: options.Now,
|
|
fingerprintKey: append([]byte(nil), options.FingerprintKey...),
|
|
}, nil
|
|
}
|
|
|
|
func (service *ApplicationService) SetUpstreamEnabled(ctx context.Context, command SetUpstreamCommand) (MutationResult, error) {
|
|
if command.Enabled && service.runtime != nil {
|
|
if err := service.runtime.ValidateUpstream(ctx, command.Name); err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
|
|
}
|
|
}
|
|
result, err := service.state.SetUpstreamEnabled(ctx, adminstate.SetUpstreamCommand{
|
|
RequestID: command.RequestID,
|
|
Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
|
|
OccurredAt: service.now().UTC(),
|
|
Name: command.Name,
|
|
Enabled: command.Enabled,
|
|
})
|
|
if err == nil && service.runtime != nil {
|
|
service.runtime.Notify()
|
|
}
|
|
return mutationResult(result), mapAdminStateError(err)
|
|
}
|
|
|
|
func (service *ApplicationService) SwitchRouting(ctx context.Context, command SwitchCommand) (MutationResult, error) {
|
|
result, err := service.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
|
|
RequestID: command.RequestID,
|
|
Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
|
|
OccurredAt: service.now().UTC(),
|
|
Name: command.Name,
|
|
ExpectedCurrent: command.ExpectedCurrent,
|
|
Target: command.Target,
|
|
Reason: command.Reason,
|
|
})
|
|
return mutationResult(result), mapAdminStateError(err)
|
|
}
|
|
|
|
func (service *ApplicationService) Status(ctx context.Context) (Status, error) {
|
|
snapshot, err := service.state.Snapshot(ctx)
|
|
if err != nil {
|
|
return Status{}, mapAdminStateError(err)
|
|
}
|
|
operations, err := service.operations.ReadOperationalStatus(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return Status{}, err
|
|
}
|
|
return Status{}, errors.Join(ErrUnavailable, err)
|
|
}
|
|
|
|
activityByName := make(map[string]UpstreamActivity, len(operations.Upstreams))
|
|
for _, activity := range operations.Upstreams {
|
|
activityByName[activity.Name] = activity
|
|
}
|
|
status := Status{SnapshotVersion: operations.SnapshotVersion}
|
|
if snapshot.Config != nil {
|
|
status.ConfigVersion = snapshot.Config.ConfigVersion
|
|
}
|
|
status.Upstreams = make([]UpstreamStatus, 0, len(snapshot.Upstreams))
|
|
for _, authoritative := range snapshot.Upstreams {
|
|
activity := activityByName[authoritative.Name]
|
|
status.Upstreams = append(status.Upstreams, UpstreamStatus{
|
|
Name: authoritative.Name,
|
|
Enabled: authoritative.Enabled,
|
|
Available: activity.Available,
|
|
Checking: activity.Checking,
|
|
Suspect: activity.Suspect,
|
|
Draining: activity.Draining,
|
|
Extracted: activity.Extracted,
|
|
ConsecutiveEmptyFetch: activity.ConsecutiveEmptyFetch,
|
|
FetchErrorCount: activity.FetchErrorCount,
|
|
})
|
|
}
|
|
sort.Slice(status.Upstreams, func(left, right int) bool {
|
|
return status.Upstreams[left].Name < status.Upstreams[right].Name
|
|
})
|
|
status.Workers = append([]WorkerStatus(nil), operations.Workers...)
|
|
sort.Slice(status.Workers, func(left, right int) bool {
|
|
if status.Workers[left].ID == status.Workers[right].ID {
|
|
return status.Workers[left].Zone < status.Workers[right].Zone
|
|
}
|
|
return status.Workers[left].ID < status.Workers[right].ID
|
|
})
|
|
return status, nil
|
|
}
|
|
|
|
func (service *ApplicationService) ReloadConfiguration(ctx context.Context, command ReloadCommand) (MutationResult, error) {
|
|
loaded, err := service.configuration.LoadConfiguration(ctx)
|
|
if err != nil {
|
|
switch {
|
|
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
|
|
return MutationResult{RequestID: command.RequestID}, err
|
|
case errors.Is(err, ErrUnavailable), errors.Is(err, ErrInvalidConfiguration):
|
|
return MutationResult{RequestID: command.RequestID}, err
|
|
default:
|
|
return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
|
|
}
|
|
}
|
|
return service.ApplyConfiguration(ctx, command, loaded)
|
|
}
|
|
|
|
// ApplyConfiguration persists and publishes one already loaded configuration snapshot.
|
|
// Startup and runtime reload paths share this method so storage connections and the
|
|
// committed management view cannot be built from different reads of the source file.
|
|
func (service *ApplicationService) ApplyConfiguration(
|
|
ctx context.Context,
|
|
command ReloadCommand,
|
|
loaded LoadedConfiguration,
|
|
) (MutationResult, error) {
|
|
if ctx == nil {
|
|
return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, err
|
|
}
|
|
if loaded.Value == nil || strings.TrimSpace(loaded.Source) != loaded.Source || loaded.Source == "" ||
|
|
len(loaded.Source) > adminstate.MaxSourceBytes {
|
|
return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration
|
|
}
|
|
if err := config.Validate(loaded.Value); err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
|
|
}
|
|
if service.runtime != nil {
|
|
if err := service.runtime.ValidateConfiguration(ctx, loaded.Value); err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
|
|
}
|
|
}
|
|
|
|
checksum, err := config.Fingerprint(loaded.Value, service.fingerprintKey)
|
|
if err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
|
|
}
|
|
|
|
current, err := service.state.Snapshot(ctx)
|
|
if err != nil {
|
|
return MutationResult{RequestID: command.RequestID}, mapAdminStateError(err)
|
|
}
|
|
upstreams, routings := managementDefinitions(loaded.Value, current)
|
|
result, err := service.state.CommitConfig(ctx, adminstate.CommitConfigCommand{
|
|
RequestID: command.RequestID,
|
|
Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
|
|
OccurredAt: service.now().UTC(),
|
|
ConfigVersion: "cfg-" + checksum,
|
|
Checksum: checksum,
|
|
Source: loaded.Source,
|
|
Upstreams: upstreams,
|
|
Routings: routings,
|
|
})
|
|
if err != nil {
|
|
return mutationResult(result), mapAdminStateError(err)
|
|
}
|
|
if result.Revision == 0 {
|
|
return mutationResult(result), errors.Join(ErrUnavailable, ErrInvalidApplicationService)
|
|
}
|
|
published := service.publisher.PublishRevision(loaded.Value, result.Revision)
|
|
if published && service.runtime != nil {
|
|
service.runtime.Notify()
|
|
}
|
|
return mutationResult(result), nil
|
|
}
|
|
|
|
func managementDefinitions(configuration *config.Config, current adminstate.Snapshot) ([]adminstate.UpstreamDefinition, []adminstate.RoutingDefinition) {
|
|
upstreams := make([]adminstate.UpstreamDefinition, 0, len(configuration.Upstreams))
|
|
for name, upstream := range configuration.Upstreams {
|
|
upstreams = append(upstreams, adminstate.UpstreamDefinition{Name: name, Enabled: upstream.Enabled})
|
|
}
|
|
sort.Slice(upstreams, func(left, right int) bool { return upstreams[left].Name < upstreams[right].Name })
|
|
|
|
currentByName := make(map[string]string, len(current.Routings))
|
|
for _, routing := range current.Routings {
|
|
currentByName[routing.Name] = routing.CurrentUpstream
|
|
}
|
|
routings := make([]adminstate.RoutingDefinition, 0, len(configuration.Routing))
|
|
for _, routing := range configuration.Routing {
|
|
selected := ""
|
|
if existing := currentByName[routing.Name]; containsString(routing.Upstreams, existing) {
|
|
selected = existing
|
|
} else if len(routing.Upstreams) > 0 {
|
|
selected = routing.Upstreams[0]
|
|
}
|
|
routings = append(routings, adminstate.RoutingDefinition{
|
|
Name: routing.Name, Enabled: routing.Enabled,
|
|
Upstreams: append([]string(nil), routing.Upstreams...), CurrentUpstream: selected,
|
|
})
|
|
}
|
|
sort.Slice(routings, func(left, right int) bool { return routings[left].Name < routings[right].Name })
|
|
return upstreams, routings
|
|
}
|
|
|
|
func containsString(values []string, target string) bool {
|
|
for _, value := range values {
|
|
if value == target {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func mutationResult(result adminstate.MutationResult) MutationResult {
|
|
return MutationResult{
|
|
RequestID: result.RequestID,
|
|
Changed: result.Changed,
|
|
Version: result.Revision,
|
|
Message: result.Message,
|
|
}
|
|
}
|
|
|
|
func mapAdminStateError(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
switch {
|
|
case errors.Is(err, adminstate.ErrNotFound):
|
|
return errors.Join(ErrNotFound, err)
|
|
case errors.Is(err, adminstate.ErrConflict):
|
|
return errors.Join(ErrConflict, err)
|
|
case errors.Is(err, adminstate.ErrInvalidCommand):
|
|
return errors.Join(ErrInvalidConfiguration, err)
|
|
case errors.Is(err, adminstate.ErrUnavailable):
|
|
return errors.Join(ErrUnavailable, err)
|
|
default:
|
|
return err
|
|
}
|
|
}
|