Documentation
¶
Index ¶
- Variables
- type AgentInfo
- type AgentLoad
- type BackgroundScheduler
- type Config
- type PostgresElector
- type ServiceScheduler
- func (s *ServiceScheduler) FindLeastLoadedAgent(ctx context.Context, provider string, az *string, excludeHost string) (string, error)
- func (s *ServiceScheduler) GetStaleAgents(ctx context.Context, provider string) ([]AgentInfo, error)
- func (s *ServiceScheduler) MigrateService(ctx context.Context, serviceID strfmt.UUID, currentHost, targetHost string) error
- func (s *ServiceScheduler) RebalanceServices(ctx context.Context, provider string, az *string) error
- func (s *ServiceScheduler) RescheduleStaleAgentServices(ctx context.Context, provider string) error
- func (s *ServiceScheduler) ShouldRebalance(ctx context.Context, provider string, az *string) (bool, error)
Constants ¶
This section is empty.
Variables ¶
var ErrNotLeader = errors.New("not the leader")
ErrNotLeader is returned when the elector determines this instance is not the leader.
Functions ¶
This section is empty.
Types ¶
type AgentInfo ¶
type AgentInfo struct {
Host string
AvailabilityZone *string
ServiceCount int
HeartbeatAt time.Time
}
AgentInfo contains information about an agent and its load.
type BackgroundScheduler ¶
type BackgroundScheduler struct {
// contains filtered or unexported fields
}
BackgroundScheduler runs periodic rescheduling and rebalancing jobs using gocron. It uses PostgreSQL advisory locks for distributed leader election to ensure only one instance runs jobs across multiple API server instances.
func NewBackgroundScheduler ¶
func NewBackgroundScheduler(serviceScheduler *ServiceScheduler, pool db.PgxIface, cfg Config) (*BackgroundScheduler, error)
NewBackgroundScheduler creates a new background scheduler with distributed leader election.
func (*BackgroundScheduler) Start ¶
func (b *BackgroundScheduler) Start(ctx context.Context) error
Start starts the background scheduler and registers all periodic jobs.
func (*BackgroundScheduler) Stop ¶
func (b *BackgroundScheduler) Stop() error
Stop stops the background scheduler gracefully.
type Config ¶
type Config struct {
StaleTimeout time.Duration
CheckInterval time.Duration
RebalanceDelay time.Duration
RebalanceThreshold float64
RebalanceMaxMigrations int
}
Config holds configuration for the service scheduler.
type PostgresElector ¶
type PostgresElector struct {
// contains filtered or unexported fields
}
PostgresElector implements gocron.Elector using PostgreSQL advisory locks. Only one instance across all API servers can hold the lock at a time. It holds a dedicated connection to ensure advisory lock ownership is maintained across calls (advisory locks are session-scoped in PostgreSQL).
func NewPostgresElector ¶
func NewPostgresElector(pool db.PgxIface, lockID int64, name string) *PostgresElector
NewPostgresElector creates a new PostgreSQL-based elector with the specified lock ID. Call Start() to acquire the dedicated connection before using IsLeader().
func (*PostgresElector) Close ¶
func (e *PostgresElector) Close()
Close releases the dedicated connection and any held advisory locks.
type ServiceScheduler ¶
type ServiceScheduler struct {
// contains filtered or unexported fields
}
ServiceScheduler handles service scheduling, rescheduling, and rebalancing.
func NewServiceScheduler ¶
func NewServiceScheduler(pool db.PgxIface, cfg Config, notifyFunc func(host string)) *ServiceScheduler
NewServiceScheduler creates a new service scheduler.
func (*ServiceScheduler) FindLeastLoadedAgent ¶
func (s *ServiceScheduler) FindLeastLoadedAgent(ctx context.Context, provider string, az *string, excludeHost string) (string, error)
FindLeastLoadedAgent returns the least-loaded healthy agent for a provider/AZ. excludeHost can be set to exclude a specific host (e.g., during migration).
func (*ServiceScheduler) GetStaleAgents ¶
func (s *ServiceScheduler) GetStaleAgents(ctx context.Context, provider string) ([]AgentInfo, error)
GetStaleAgents returns agents that haven't sent a heartbeat within the stale timeout.
func (*ServiceScheduler) MigrateService ¶
func (s *ServiceScheduler) MigrateService(ctx context.Context, serviceID strfmt.UUID, currentHost, targetHost string) error
MigrateService moves a service from one agent to another. If targetHost is empty, the least-loaded agent is selected.
func (*ServiceScheduler) RebalanceServices ¶
func (s *ServiceScheduler) RebalanceServices(ctx context.Context, provider string, az *string) error
RebalanceServices redistributes services across agents when imbalance exceeds threshold.
func (*ServiceScheduler) RescheduleStaleAgentServices ¶
func (s *ServiceScheduler) RescheduleStaleAgentServices(ctx context.Context, provider string) error
RescheduleStaleAgentServices migrates services from stale agents to healthy ones. Only disables agents after all their services have been successfully migrated.
func (*ServiceScheduler) ShouldRebalance ¶
func (s *ServiceScheduler) ShouldRebalance(ctx context.Context, provider string, az *string) (bool, error)
ShouldRebalance checks if the imbalance between agents exceeds the threshold. Returns true if rebalancing is needed.
Source Files