controller

type Controller ¶

type Controller interface {
	Start(ctx context.Context) error
	Stop()
}

Controller processes entities of a specific kind

type ControllerEntity ¶

type ControllerEntity interface {
	Decode(getter entity.AttrGetter)
	Encode() []entity.Attr
}

type ControllerManager ¶

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

ControllerManager manages multiple controllers

func NewControllerManager() *ControllerManager

func (m *ControllerManager) AddController(controller Controller)

func (m *ControllerManager) Start(ctx context.Context) error

func (m *ControllerManager) Stop()

type DeletingReconcileController ¶

type DeletingReconcileController interface {
	Delete(ctx context.Context, e entity.Id) error
}

DeletingReconcileController is an optional interface that reconcile controllers can implement to handle deletion of their managed entities.

type EntityAccessClient ¶

type EntityAccessClient interface {
	// Get returns a single entity by its ID
	Get(ctx context.Context, id string) (*entityserver_v1alpha.EntityAccessClientGetResults, error)

	// List returns all entities matching the given index
	List(ctx context.Context, index entity.Attr) (*entityserver_v1alpha.EntityAccessClientListResults, error)

	// WatchIndex watches for changes to entities matching the given index
	// and sends updates through the provided sender
	WatchIndex(ctx context.Context, index entity.Attr, sender stream.SendStream[*entityserver_v1alpha.EntityOp]) error
}

EntityAccessClient defines the interface for interacting with the entity server

type Event ¶

type Event struct {
	Type EventType
	Id   entity.Id

	Entity *entity.Entity // The entity that was changed

	Rev, PrevRev int64 // Revision and previous revision for the entity
}

Event represents a change to an entity

type EventType ¶

type EventType string

EventType represents the type of event that occurred on an entity

const (
	EventAdded   EventType = "ADDED"
	EventUpdated EventType = "UPDATED"
	EventDeleted EventType = "DELETED"
)

type GenericController ¶

type GenericController[P ControllerEntity] interface {
	Init(context.Context) error
	Create(ctx context.Context, obj P, meta *entity.Meta) error
	// Delete is called when an entity is deleted. The obj parameter contains the
	// entity data from before deletion (available since delete watch events now
	// include the previous value). It may be nil if the entity data was unavailable
	// (e.g., watch reconnect after etcd compaction).
	Delete(ctx context.Context, id entity.Id, obj P) error
}

type HandlerFunc ¶

type HandlerFunc func(ctx context.Context, event Event) ([]entity.Attr, error)

HandlerFunc is a function that processes an entity

func AdaptController[
	T any,
	P interface {
		*T
		ControllerEntity
	},
	C GenericController[P],
](cont C) HandlerFunc

func AdaptReconcileController[
	T any,
	P interface {
		*T
		ControllerEntity
	},
	C ReconcileControllerI[P],
](cont C) HandlerFunc

type ReconcileController ¶

type ReconcileController struct {
	Log *slog.Logger
	// contains filtered or unexported fields
}

ReconcileController implements the Controller interface

func NewReconcileController(name string, log *slog.Logger, index entity.Attr, esc *entityserver_v1alpha.EntityAccessClient, handler HandlerFunc, resyncPeriod time.Duration, workers int) *ReconcileController

func (c *ReconcileController) Enqueue(event Event)

func (c *ReconcileController) ProcessEventForTest(ctx context.Context, event Event) error

func (c *ReconcileController) RecordWrite(revision int64)

func (c *ReconcileController) SetPeriodic(often time.Duration, fn func(ctx context.Context) error)

func (c *ReconcileController) Start(top context.Context) error

func (c *ReconcileController) Stop()

func (c *ReconcileController) WriteTracker() WriteTracker

type ReconcileControllerI ¶

type ReconcileControllerI[P ControllerEntity] interface {
	Init(context.Context) error
	Reconcile(ctx context.Context, obj P, meta *entity.Meta) error
}

ReconcileControllerI is for controllers that maintain aggregate state across multiple entities. Unlike GenericController which maps 1:1 between an entity and a resource, ReconcileControllerI handles controllers where one entity drives reconciliation of N resources.

type RingSet ¶

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

RingSet maintains a fixed-size set of recently seen int64 values using a ring buffer. Controllers use this to track recently written entity revisions, allowing them to skip watch events for their own writes and reduce unnecessary reconciliation noise.

Thread-safe for concurrent access by multiple workers and the watch goroutine.

func NewRingSet(capacity int) *RingSet

func (r *RingSet) Add(value int64)

func (r *RingSet) Contains(value int64) bool

type UpdatingController ¶

type UpdatingController[P ControllerEntity] interface {
	Update(ctx context.Context, obj P, meta *entity.Meta) error
}

UpdatingController is an optional interface that controllers can implement to handle updates differently from creates

type WriteTracker ¶

type WriteTracker interface {
	RecordWrite(revision int64)
}

WriteTracker provides a way to track entity write revisions to skip self-generated watch events. Controllers that make manual entity writes outside the reconciliation framework can use this to record their writes and avoid unnecessary re-reconciliation.