dcm

Documentation

Overview

Package dcm implements DCM registration and heartbeat lifecycle management.

Index

• Variables
• func ParseRetryAfter(value string, now time.Time) (time.Duration, bool)
• type ConsumerLagProvider
• type RateLimitError
  • func (e *RateLimitError) Error() string
  • func (e *RateLimitError) Unwrap() error
• type Registrar
  • func NewRegistrar(cfg RegistrarConfig, lister ServiceTypeLister, lagProvider ConsumerLagProvider, ...) (*Registrar, error)
  • func (r *Registrar) AgentID() (string, bool)
  • func (r *Registrar) Done() <-chan struct{}
  • func (r *Registrar) NotifyServiceTypeChange()
  • func (r *Registrar) Start(ctx context.Context)
• type RegistrarConfig
• type ResourceCapacityProvider
• type ServiceTypeLister

Variables

var ErrNonRetryable = errors.New("non-retryable DCM error")

var ErrRateLimited = errors.New("dcm rate limited")

Functions

func ParseRetryAfter

func ParseRetryAfter(value string, now time.Time) (time.Duration, bool)

ParseRetryAfter parses the Retry-After header value. Supports seconds (integer) and HTTP-date (RFC1123) formats per RFC 7231 §7.1.3. Returns the duration to wait and whether parsing succeeded.

Types

type ConsumerLagProvider

type ConsumerLagProvider interface {
	ConsumerLag() int64
}

ConsumerLagProvider returns the current consumer lag for heartbeat payloads.

type RateLimitError

type RateLimitError struct {
	RetryAfter    time.Duration
	HasRetryAfter bool
}

RateLimitError is returned on HTTP 429 responses.

func (*RateLimitError) Error

func (e *RateLimitError) Error() string

func (*RateLimitError) Unwrap

func (e *RateLimitError) Unwrap() error

type Registrar

type Registrar struct {
	// contains filtered or unexported fields
}

Registrar handles DCM registration and heartbeat lifecycle.

func NewRegistrar

func NewRegistrar(
	cfg RegistrarConfig,
	lister ServiceTypeLister,
	lagProvider ConsumerLagProvider,
	resourceProvider ResourceCapacityProvider,
	logger *slog.Logger,
) (*Registrar, error)

NewRegistrar creates a Registrar. Returns error if config is invalid.

func (*Registrar) AgentID

func (r *Registrar) AgentID() (string, bool)

AgentID returns the DCM-assigned agent ID, or ("", false) if not yet registered.

func (*Registrar) Done

func (r *Registrar) Done() <-chan struct{}

Done returns a channel closed when the registrar goroutine exits.

func (*Registrar) NotifyServiceTypeChange

func (r *Registrar) NotifyServiceTypeChange()

NotifyServiceTypeChange signals that the advertisable service types may have changed.

func (*Registrar) Start

func (r *Registrar) Start(ctx context.Context)

Start begins the async registration + heartbeat loop. Non-blocking, idempotent.

type RegistrarConfig

type RegistrarConfig struct {
	AgentName                 string
	Environment               string
	Cost                      string
	TopicName                 string
	RegistrationURL           string
	InitialBackoff            time.Duration
	MaxBackoff                time.Duration
	HeartbeatInterval         time.Duration
	PrerequisiteRetryInterval time.Duration
}

RegistrarConfig holds the configuration for DCM registration.

type ResourceCapacityProvider

type ResourceCapacityProvider interface {
	ResourceCapacity() *v1alpha1.ResourceCapacity
}

ResourceCapacityProvider optionally returns resource availability for registration. Returns nil when not available (REQ-DCM-030 is SHOULD, not MUST).

type ServiceTypeLister

type ServiceTypeLister interface {
	AdvertisableServiceTypes() []string
}

ServiceTypeLister returns the set of currently advertisable service types (backed by SPs in Ready or Unhealthy state — NOT Unavailable).