Documentation
¶
Overview ¶
Package k8s provides a Kubernetes-native cluster.Store implementation.
Worker discovery uses Pod annotations with a configurable label selector. Leader election uses the coordination/v1 Lease API (standard K8s pattern).
Example:
client := kubernetes.NewForConfigOrDie(rest.InClusterConfig()) provider := k8s.New(client, "my-namespace") // Use provider as a cluster.Store
Index ¶
- type Option
- type Provider
- func (p *Provider) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)
- func (p *Provider) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error)
- func (p *Provider) DeregisterWorker(ctx context.Context, workerID id.WorkerID) error
- func (p *Provider) GetLeader(ctx context.Context) (*cluster.Worker, error)
- func (p *Provider) GetWorker(ctx context.Context, workerID id.WorkerID) (*cluster.Worker, error)
- func (p *Provider) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error
- func (p *Provider) ListWorkers(ctx context.Context) ([]*cluster.Worker, error)
- func (p *Provider) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error)
- func (p *Provider) RegisterWorker(ctx context.Context, w *cluster.Worker) error
- func (p *Provider) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Provider)
Option configures a Provider.
func WithAnnotationPrefix ¶
WithAnnotationPrefix sets the prefix for worker-data annotations on Pods. Default: "dispatch.xraph.com/".
func WithLabelSelector ¶
WithLabelSelector overrides the label selector used to discover worker Pods. Default: "app.kubernetes.io/component=dispatch-worker".
func WithLeaseName ¶
WithLeaseName sets the Lease object name used for leader election. Default: "dispatch-leader".
type Provider ¶
type Provider struct {
// contains filtered or unexported fields
}
Provider implements cluster.Store using Kubernetes primitives:
- Worker discovery via Pod annotations and label selectors
- Leader election via the coordination/v1 Lease API
func New ¶
func New(client kubernetes.Interface, namespace string, opts ...Option) *Provider
New creates a Kubernetes cluster provider. The clientset and namespace are required. Use functional options to customise the lease name, label selector, annotation prefix, or logger.
func (*Provider) AcquireLeadership ¶
func (p *Provider) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)
AcquireLeadership attempts to become the cluster leader using the coordination/v1 Lease API.
func (*Provider) DeleteStaleWorkers ¶
DeleteStaleWorkers removes dispatch annotations from pods whose last-seen is older than threshold. The pods themselves are owned by Kubernetes and are not deleted here — only the dispatch metadata is cleared, so a fresh instance can re-register without colliding with stale leadership state.
func (*Provider) DeregisterWorker ¶
DeregisterWorker removes dispatch annotations from the worker's Pod.
func (*Provider) GetLeader ¶
GetLeader returns the current cluster leader from the Lease, or nil if there is no active leader.
func (*Provider) GetWorker ¶
GetWorker returns the worker whose Pod carries the given worker-id annotation.
func (*Provider) HeartbeatWorker ¶
HeartbeatWorker updates the last-seen annotation on the worker's Pod.
func (*Provider) ListWorkers ¶
ListWorkers returns all registered dispatch workers by scanning Pod annotations.
func (*Provider) ReapDeadWorkers ¶
func (p *Provider) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error)
ReapDeadWorkers returns workers whose last-seen annotation is older than the given threshold.
func (*Provider) RegisterWorker ¶
RegisterWorker stores worker metadata as annotations on the worker's Pod. The Pod is located by matching the worker's Hostname to the Pod name.
Source Files