Documentation
¶
Index ¶
- func CreateConfigResolver(handler pkg.EventHandlerConfig, branch base.Branch) pkg.ConfigResolver
- func CreateConsumer(saramaClient sarama.Client, branch base.Branch, topicPrefix base.TopicPrefix, ...) (libkafka.Consumer, handler.TaskEventHandler)
- func CreateEventHandlerConfig() pkg.EventHandlerConfig
- func CreateHealthcheckCron(expression libcron.Expression, runner probe.HealthcheckRunner) pkg.CronScheduler
- func CreateHealthcheckRunner(configProvider pkg.EventHandlerConfig, syncProducer libkafka.SyncProducer, ...) probe.HealthcheckRunner
- func CreateJobWatcher(kubeClient kubernetes.Interface, namespace libk8s.Namespace, ...) pkg.JobWatcher
- func CreateK8sConnector(config *rest.Config) pkg.K8sConnector
- func CreateResourceEventHandlerConfig(ctx context.Context, handler pkg.EventHandlerConfig) cache.ResourceEventHandler
- func CreateZombieSweeper(jobWatcher pkg.JobWatcher, namespace libk8s.Namespace, ...) pkg.ZombieSweeper
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CreateConfigResolver ¶
func CreateConfigResolver( handler pkg.EventHandlerConfig, branch base.Branch, ) pkg.ConfigResolver
CreateConfigResolver returns a ConfigResolver backed by the given store.
func CreateConsumer ¶
func CreateConsumer( saramaClient sarama.Client, branch base.Branch, topicPrefix base.TopicPrefix, kubeClient kubernetes.Interface, namespace libk8s.Namespace, kafkaBrokers string, resolver pkg.ConfigResolver, logSamplerFactory log.SamplerFactory, currentDateTimeGetter libtime.CurrentDateTimeGetter, resultPublisher pkg.ResultPublisher, taskStore *pkg.TaskStore, jobTTLSecondsAfterFinished int32, ) (libkafka.Consumer, handler.TaskEventHandler)
CreateConsumer wires together all components and returns a Kafka Consumer that reads task events and spawns K8s Jobs for qualifying tasks, along with the TaskEventHandler so callers can wire RunDeferredRespawnLoop.
func CreateEventHandlerConfig ¶
func CreateEventHandlerConfig() pkg.EventHandlerConfig
CreateEventHandlerConfig returns an empty in-memory event handler for Config resources.
func CreateHealthcheckCron ¶
func CreateHealthcheckCron( expression libcron.Expression, runner probe.HealthcheckRunner, ) pkg.CronScheduler
CreateHealthcheckCron wraps the given runner in a cron scheduler. Pass the runner returned by CreateHealthcheckRunner so the cron and the HTTP handler share the same instance.
func CreateHealthcheckRunner ¶
func CreateHealthcheckRunner( configProvider pkg.EventHandlerConfig, syncProducer libkafka.SyncProducer, topicPrefix base.TopicPrefix, branch base.Branch, ) probe.HealthcheckRunner
CreateHealthcheckRunner creates the healthcheck runner shared between the cron path and the HTTP trigger path. Callers must pass the same instance to both CreateHealthcheckCron and the HTTP handler so probe behavior is identical regardless of invocation path.
func CreateJobWatcher ¶
func CreateJobWatcher( kubeClient kubernetes.Interface, namespace libk8s.Namespace, taskStore *pkg.TaskStore, publisher pkg.ResultPublisher, ) pkg.JobWatcher
CreateJobWatcher creates a JobWatcher that reacts to terminal batch/v1 Job states.
func CreateK8sConnector ¶
func CreateK8sConnector(config *rest.Config) pkg.K8sConnector
CreateK8sConnector returns a K8sConnector wired to the given rest.Config.
func CreateResourceEventHandlerConfig ¶
func CreateResourceEventHandlerConfig( ctx context.Context, handler pkg.EventHandlerConfig, ) cache.ResourceEventHandler
CreateResourceEventHandlerConfig adapts an EventHandlerConfig to cache.ResourceEventHandler.
func CreateZombieSweeper ¶
func CreateZombieSweeper( jobWatcher pkg.JobWatcher, namespace libk8s.Namespace, taskStore *pkg.TaskStore, publisher pkg.ResultPublisher, configProvider pkg.EventHandlerConfig, currentDateTime libtime.CurrentDateTimeGetter, ) pkg.ZombieSweeper
CreateZombieSweeper creates a deadline sweeper that classifies stuck tasks as zombies and emits failure events via the publisher. Interval and per-task deadline are sourced from the AgentConfig CRD knobs (see ConfigSpec). The sweeper receives the JobWatcher (not its lister) because the lister is populated only after JobWatcher.Run completes its informer cache sync; passing the watcher lets the sweeper resolve the lister lazily on each tick and skip the tick if cache sync has not yet happened (avoids a nil-deref panic at the first tick when service.Run starts all components concurrently).
Types ¶
This section is empty.
Source Files