Documentation
¶
Overview ¶
Package controller implements the TMIComponent reconciler for the TMI Component Platform.
Index ¶
- func ConsumerConfigFor(c *platformv1alpha1.TMIComponent) *nats.ConsumerConfig
- func RenderDeployment(c *platformv1alpha1.TMIComponent) *appsv1.Deployment
- func RenderNetworkPolicy(c *platformv1alpha1.TMIComponent) *networkingv1.NetworkPolicy
- func RenderScaledObject(c *platformv1alpha1.TMIComponent) *unstructured.Unstructured
- func StreamConfigFor(c *platformv1alpha1.TMIComponent) *nats.StreamConfig
- func ValidateComponent(c *platformv1alpha1.TMIComponent) error
- type NATSProvisioner
- type StreamProvisioner
- type TMIComponentReconciler
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ConsumerConfigFor ¶
func ConsumerConfigFor(c *platformv1alpha1.TMIComponent) *nats.ConsumerConfig
ConsumerConfigFor returns the durable consumer the controller pre-creates on a component's stream. The name MUST equal consumerNameFor(c) because that is exactly the consumer the KEDA ScaledObject (render_scaledobject.go) watches for queue depth — KEDA cannot scale a worker from zero unless this consumer already exists under that name. The worker (cmd/extractor, cmd/chunkembed) binds this same durable rather than creating its own.
The consumer carries NO FilterSubject: each per-component stream is single-purpose, so the durable consumes every subject the stream binds. This keeps the config identical regardless of how many jobSubjects a component declares and avoids a filter-subject conflict when the worker binds it. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: build a durable JetStream consumer config for a component with configurable ack-wait (pure)
func RenderDeployment ¶
func RenderDeployment(c *platformv1alpha1.TMIComponent) *appsv1.Deployment
RenderDeployment builds the worker Deployment for a TMIComponent. Pod hardening (readOnlyRootFilesystem, runAsNonRoot, all caps dropped, RuntimeDefault seccomp) is applied unconditionally — it is a hard platform invariant, not a per-component option. SEM@6b7b976: build a hardened worker Deployment with NATS client-cert mount and Reloader annotation (pure)
func RenderNetworkPolicy ¶
func RenderNetworkPolicy(c *platformv1alpha1.TMIComponent) *networkingv1.NetworkPolicy
RenderNetworkPolicy builds the NetworkPolicy for a component from its egress posture. The Egress policy type is always set so a limited egress list is an enforced deny. Every egress rule carries an explicit To peer selector — a rule with ports but no To matches all destinations, so scoping the peers is what makes "egress: none" mean "NATS only". The controller always renders this as a cluster-layer backstop, independent of any in-code egress guarding. SEM@b606b1e53cb73ed3007ef11a16d7867e87b3f130: build a Kubernetes NetworkPolicy for a component enforcing egress posture (pure)
func RenderScaledObject ¶
func RenderScaledObject(c *platformv1alpha1.TMIComponent) *unstructured.Unstructured
RenderScaledObject builds the KEDA ScaledObject for a component as an unstructured object, avoiding a KEDA Go-module dependency. KEDA scales the worker Deployment on JetStream pending-message depth. SEM@f3ef9e21c1105bdf75b7cf13c31001603e713458: build a KEDA ScaledObject for a TMI component targeting a NATS JetStream consumer (pure)
func StreamConfigFor ¶
func StreamConfigFor(c *platformv1alpha1.TMIComponent) *nats.StreamConfig
StreamConfigFor returns the JetStream stream configuration that binds all of a component's job subjects. SEM@2bf9318487695f6e17a03631c7e5f410aedc5758: build a JetStream work-queue stream config for a component's job subjects (pure)
func ValidateComponent ¶
func ValidateComponent(c *platformv1alpha1.TMIComponent) error
ValidateComponent checks a TMIComponent spec for internal consistency beyond what the CRD OpenAPI schema can express. SEM@63e2aad01818e6abee8652287d395a8b4f205986: validate a TMIComponent spec for semantic constraints beyond CRD schema (pure)
Types ¶
type NATSProvisioner ¶
type NATSProvisioner struct {
// contains filtered or unexported fields
}
NATSProvisioner is the live StreamProvisioner backed by a NATS JetStream connection. In the current out-of-cluster e2e flow the controller reaches NATS through the host port-forward (nats://127.0.0.1:4222); once the controller ships as an in-cluster Deployment it will use the in-cluster service DNS (tls://nats.tmi-platform.svc:4222). Either way the URL is supplied via the TMI_NATS_URL env var. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: live StreamProvisioner backed by a NATS JetStream connection (pure)
func NewNATSProvisioner ¶
func NewNATSProvisioner(url string) (*NATSProvisioner, error)
NewNATSProvisioner dials NATS and opens a JetStream context. It uses RetryOnFailedConnect so the controller can start before the port-forward (or the NATS pod) is reachable: JetStream calls then fail until the connection establishes, which surfaces as a reconcile error and a requeue. SEM@946ec29: connect to NATS with retry and optional TLS, returning a JetStream provisioner
func (*NATSProvisioner) Close ¶
func (p *NATSProvisioner) Close()
Close closes the underlying NATS connection. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: close the underlying NATS connection (mutates shared state)
func (*NATSProvisioner) EnsureStreamAndConsumer ¶
func (p *NATSProvisioner) EnsureStreamAndConsumer(ctx context.Context, c *platformv1alpha1.TMIComponent) error
EnsureStreamAndConsumer creates the component's stream and durable consumer if they do not already exist. It is idempotent: an existing stream/consumer is left untouched (the config is stable across reconciles), so it is safe to call on every reconcile. Creation is gated on a NotFound lookup rather than blind AddStream/AddConsumer so a transient connectivity error is returned as an error (triggering requeue) instead of being mistaken for "already exists". SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: idempotently create a JetStream stream and durable consumer for a component if absent
type StreamProvisioner ¶
type StreamProvisioner interface {
EnsureStreamAndConsumer(ctx context.Context, c *platformv1alpha1.TMIComponent) error
}
StreamProvisioner ensures the JetStream stream and durable consumer a component needs exist before any worker pod runs. It is the missing half of reconciliation: rendering the KEDA ScaledObject is pointless unless the stream and consumer it watches already exist, because KEDA scales a worker from zero by reading that consumer's pending-message depth.
The reconciler holds this as an interface so envtest unit tests (which have no NATS) can leave it nil and skip provisioning. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: interface for idempotently ensuring a JetStream stream and consumer exist for a component (pure)
type TMIComponentReconciler ¶
type TMIComponentReconciler struct {
client.Client
Scheme *runtime.Scheme
// Streams provisions the JetStream stream and durable consumer for each
// component. It is optional: envtest unit tests have no NATS and leave it
// nil, which skips provisioning.
Streams StreamProvisioner
}
TMIComponentReconciler reconciles a TMIComponent into its child objects: a Deployment, a NetworkPolicy, a KEDA ScaledObject, and (when Streams is set) the JetStream stream + durable consumer the ScaledObject watches. Pre-creating the stream and consumer is required for KEDA scale-from-zero: KEDA reads the consumer's pending depth to decide when to scale the worker up from zero, so the consumer must exist before any worker pod runs. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: Kubernetes reconciler that converges a TMIComponent into its Deployment, NetworkPolicy, and ScaledObject children
func (*TMIComponentReconciler) Reconcile ¶
func (r *TMIComponentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error)
Reconcile is the controller-runtime entrypoint. SEM@50b942b21c528f6a4405c3ce2dccedfdd379012a: controller-runtime entrypoint; delegates to ReconcileComponent
func (*TMIComponentReconciler) ReconcileComponent ¶
func (r *TMIComponentReconciler) ReconcileComponent(ctx context.Context, key types.NamespacedName) (ctrl.Result, error)
ReconcileComponent renders and applies the child objects for one component. Split out from Reconcile so tests can drive it directly. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: render and apply all child objects for a TMIComponent, provisioning JetStream if configured
func (*TMIComponentReconciler) SetupWithManager ¶
func (r *TMIComponentReconciler) SetupWithManager(mgr ctrl.Manager) error
SetupWithManager registers the reconciler and its owned child types. SEM@50b942b21c528f6a4405c3ce2dccedfdd379012a: register the reconciler with the controller-runtime manager and declare owned child types
Source Files