controller

Documentation

Overview

Package controller implements the core Δ-controller controller runtime that processes declarative pipeline specifications and manages incremental view reconciliation.

This package provides the main controller abstraction that bridges declarative YAML specifications with imperative Go reconciliation logic. Controllers watch source Kubernetes resources, process them through declarative pipelines to generate view objects, and manage target resource updates.

Controllers support:

• multiple source resource types with optional label selectors,
• declarative pipeline processing with joins and aggregations,
• incremental reconciliation using DBSP,
• configurable error handling and status reporting, and
• integration with the composite client system.

Example usage:

ctrl, _ := controller.New("my-controller", mgr, controller.Config{
    Sources: []opv1a1.Source{{Kind: "Pod"}},
    Target: opv1a1.Target{Kind: "PodView"},
    Pipeline: opv1a1.Pipeline{...},
})

Index

• Constants
• func NewErrorReporter(c *Controller, errorChan chan error) *errorReporter
• type Controller
  • func New(mgr runtimeManager.Manager, operator string, config opv1a1.Controller, ...) (*Controller, error)
  • func (c *Controller) GetGVKs() []schema.GroupVersionKind
  • func (c *Controller) GetName() string
  • func (c *Controller) GetStatus(gen int64) opv1a1.ControllerStatus
  • func (c *Controller) GetWatcher() chan reconciler.Request
  • func (s Controller) HasCritical() bool
  • func (s Controller) IsEmpty() bool
  • func (s Controller) Pop()
  • func (s Controller) Push(err error) error
  • func (s Controller) PushCritical(err error) error
  • func (s Controller) PushCriticalError(msg string) error
  • func (s Controller) PushCriticalErrorf(format string, a ...any) error
  • func (s Controller) PushError(msg string) error
  • func (s Controller) PushErrorf(format string, a ...any) error
  • func (s Controller) Report() []string
  • func (c *Controller) ReportErrors() []string
  • func (c *Controller) SetPipeline(pipeline pipeline.Evaluator)
  • func (s Controller) Size() int
  • func (c *Controller) Start(ctx context.Context) error
  • func (s Controller) String() string
  • func (s Controller) Top() error
• type ControllerReconciler
  • func NewControllerReconciler(mgr runtimeManager.Manager, c *Controller) *ControllerReconciler
  • func (r *ControllerReconciler) Reconcile(ctx context.Context, req reconciler.Request) (reconcile.Result, error)
• type Error
• type Options
• type ProcessorFunc

Constants

const (
	// ErrorReporterStackSize controls the depth of the LIFO error buffer.
	ErrorReporterStackSize int = 10

	// TrimPrefixSuffixLen contols the number of characters to retain at the prefix and the
	// suffix of long strings.
	TrimPrefixSuffixLen = 120
)

const WatcherBufferSize int = 1024

WatcherBufferSize is the default buffer size for the watch event handler.

Functions

func NewErrorReporter

func NewErrorReporter(c *Controller, errorChan chan error) *errorReporter

NewErrorReporter creates a new error reporter.

Types

type Controller

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

Controller is a dcontroller reconciler.

func New

func New(mgr runtimeManager.Manager, operator string, config opv1a1.Controller, opts Options) (*Controller, error)

New registers a new controller for an operator, given by the source resource(s) the controller watches, a target resource the controller sends its output, and a processing pipeline to process the base resources into target resources.

func (*Controller) GetGVKs

func (c *Controller) GetGVKs() []schema.GroupVersionKind

GetGVKs returns the GVKs of the views registered with the controller.

func (*Controller) GetName

func (c *Controller) GetName() string

GetName returns the name of the controller.

func (*Controller) GetStatus

func (c *Controller) GetStatus(gen int64) opv1a1.ControllerStatus

GetStatus returns the status of the controller.

func (*Controller) GetWatcher

func (c *Controller) GetWatcher() chan reconciler.Request

GetWatcher returns the channel that multiplexes the requests coming from the base resources.

func (Controller) HasCritical

func (s Controller) HasCritical() bool

HasCritical returns true if the stack contains a critical error.

func (Controller) IsEmpty

func (s Controller) IsEmpty() bool

IsEmpty returns true if the stack is empty.

func (Controller) Pop

func (s Controller) Pop()

Pop pops the extra error from the stack.

func (Controller) Push

func (s Controller) Push(err error) error

Push pushes a noncritical error to the error stack.

func (Controller) PushCritical

func (s Controller) PushCritical(err error) error

PushCritical pushes a critical error to the error stack.

func (Controller) PushCriticalError

func (s Controller) PushCriticalError(msg string) error

PushCriticalError pushes a critical error to the error stack.

func (Controller) PushCriticalErrorf

func (s Controller) PushCriticalErrorf(format string, a ...any) error

PushCriticalErrorf pushes a formatted critical error to the error stack.

func (Controller) PushError

func (s Controller) PushError(msg string) error

PushError pushes a simple noncritical error to the error stack.

func (Controller) PushErrorf

func (s Controller) PushErrorf(format string, a ...any) error

PushErrorf pushes a formatted noncritical error to the error stack.

func (Controller) Report

func (s Controller) Report() []string

Report returns the error messages on the stack.

func (*Controller) ReportErrors

func (c *Controller) ReportErrors() []string

ReportErrors returns a short report on the error stack of the controller.

func (*Controller) SetPipeline

func (c *Controller) SetPipeline(pipeline pipeline.Evaluator)

SetPipeline overrides the pipeline of the controller. Useful for adding a custom pipeline to a controller.

func (Controller) Size

func (s Controller) Size() int

Size returns the number of errors on the stack.

func (*Controller) Start

func (c *Controller) Start(ctx context.Context) error

Start starts running the controller. The Start function blocks until the context is closed or an error occurs, and it will stop running when the context is closed.

func (Controller) String

func (s Controller) String() string

String stringifies the error stack.

func (Controller) Top

func (s Controller) Top() error

Top returns the last errors from the stack.

type ControllerReconciler

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

ControllerReconciler is a generic reconciler that feeds the requests received from any of the sources into the controller processor pipeline.

func NewControllerReconciler

func NewControllerReconciler(mgr runtimeManager.Manager, c *Controller) *ControllerReconciler

NewControllerReconciler creates a new generic reconciler.

func (*ControllerReconciler) Reconcile

func (r *ControllerReconciler) Reconcile(ctx context.Context, req reconciler.Request) (reconcile.Result, error)

Reconcile implements the reconciler.

type Error

type Error struct {
	Operator, Controller string
	// contains filtered or unexported fields
}

Error is a error wrapped with an operator and a controller name.

type Options

type Options struct {
	// Processor allows to override the default request processor of the controller.
	Processor ProcessorFunc

	// ErrorChannel is a channel to receive errors from the controller.
	ErrorChan chan error
}

Options defines the controller configuration.

type ProcessorFunc

type ProcessorFunc func(ctx context.Context, c *Controller, req reconciler.Request) error

ProcessorFunc is the request processor type for the controller.