Skip to content

Commit 12ef6a3

Browse files
j-mok-devyizhaodev
j-mok-dev
authored and
yizhaodev
committed
feat: expose heartbeat and reconciler config in CR
Signed-off-by: Jooyeon Mok <jmok@redhat.com>
1 parent 3b125d3 commit 12ef6a3

6 files changed

Lines changed: 170 additions & 0 deletions

File tree

‎api/v1alpha1/llmbatchgateway_types.go‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ type LLMBatchGatewayList struct {
4040

4141
// LLMBatchGatewaySpec defines the desired state of the batch gateway deployment.
4242
// +kubebuilder:validation:XValidation:rule="size(self.secretRef.name) > 0",message="spec.secretRef.name is required"
43+
// +kubebuilder:validation:XValidation:rule="!has(self.processor.config) || !has(self.processor.config.heartbeatInterval) || !has(self.gc.config) || !has(self.gc.config.reconciler) || !has(self.gc.config.reconciler.interval) || duration(self.processor.config.heartbeatInterval) < duration(self.gc.config.reconciler.interval)",message="processor.config.heartbeatInterval must be shorter than gc.config.reconciler.interval"
4344
type LLMBatchGatewaySpec struct {
4445
// SecretRef references the Kubernetes Secret that holds runtime credentials
4546
// (database URL, S3 keys, inference API key, etc.).
@@ -450,6 +451,14 @@ type ProcessorConfigSpec struct {
450451
// EnablePprof enables the Go pprof profiling HTTP endpoints.
451452
EnablePprof bool `json:"enablePprof,omitempty"`
452453

454+
// HeartbeatInterval is how often the processor refreshes in-flight job entries
455+
// to signal liveness. Must be shorter than gc.config.reconciler.interval to
456+
// prevent active jobs from being misidentified as orphans.
457+
// +kubebuilder:default="5m"
458+
// +kubebuilder:validation:MaxLength=32
459+
// +kubebuilder:validation:Pattern=`^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$`
460+
HeartbeatInterval string `json:"heartbeatInterval,omitempty"`
461+
453462
// Logging configures log verbosity for the processor.
454463
Logging *LoggingConfig `json:"logging,omitempty"`
455464
}
@@ -491,6 +500,23 @@ type GCConfigSpec struct {
491500

492501
// Logging configures log verbosity for the GC.
493502
Logging *LoggingConfig `json:"logging,omitempty"`
503+
504+
// Reconciler configures the orphan reconciler that detects and cleans up
505+
// stale jobs missed by the normal collector sweep.
506+
Reconciler *ReconcilerSpec `json:"reconciler,omitempty"`
507+
}
508+
509+
// ReconcilerSpec configures the GC orphan reconciler.
510+
type ReconcilerSpec struct {
511+
// Enabled controls whether the orphan reconciler runs.
512+
// +kubebuilder:default=true
513+
Enabled bool `json:"enabled,omitempty"`
514+
515+
// Interval is both the scan frequency and the staleness threshold (e.g. "60m").
516+
// +kubebuilder:default="60m"
517+
// +kubebuilder:validation:MaxLength=32
518+
// +kubebuilder:validation:Pattern=`^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$`
519+
Interval string `json:"interval,omitempty"`
494520
}
495521

496522
// --- Observability ---

‎api/v1alpha1/zz_generated.deepcopy.go‎

Lines changed: 20 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎config/crd/bases/batch.llm-d.ai_llmbatchgateways.yaml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -365,6 +365,24 @@ spec:
365365
operations performed in parallel.
366366
format: int32
367367
type: integer
368+
reconciler:
369+
description: |-
370+
Reconciler configures the orphan reconciler that detects and cleans up
371+
stale jobs missed by the normal collector sweep.
372+
properties:
373+
enabled:
374+
default: true
375+
description: Enabled controls whether the orphan reconciler
376+
runs.
377+
type: boolean
378+
interval:
379+
default: 60m
380+
description: Interval is both the scan frequency and the
381+
staleness threshold (e.g. "60m").
382+
maxLength: 32
383+
pattern: ^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$
384+
type: string
385+
type: object
368386
type: object
369387
imagePullPolicy:
370388
description: |-
@@ -994,6 +1012,15 @@ spec:
9941012
description: EnablePprof enables the Go pprof profiling HTTP
9951013
endpoints.
9961014
type: boolean
1015+
heartbeatInterval:
1016+
default: 5m
1017+
description: |-
1018+
HeartbeatInterval is how often the processor refreshes in-flight job entries
1019+
to signal liveness. Must be shorter than gc.config.reconciler.interval to
1020+
prevent active jobs from being misidentified as orphans.
1021+
maxLength: 32
1022+
pattern: ^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$
1023+
type: string
9971024
logging:
9981025
description: Logging configures log verbosity for the processor.
9991026
properties:
@@ -1375,6 +1402,10 @@ spec:
13751402
x-kubernetes-validations:
13761403
- message: spec.secretRef.name is required
13771404
rule: size(self.secretRef.name) > 0
1405+
- message: processor.config.heartbeatInterval must be shorter than gc.config.reconciler.interval
1406+
rule: '!has(self.processor.config) || !has(self.processor.config.heartbeatInterval)
1407+
|| !has(self.gc.config) || !has(self.gc.config.reconciler) || !has(self.gc.config.reconciler.interval)
1408+
|| duration(self.processor.config.heartbeatInterval) < duration(self.gc.config.reconciler.interval)'
13781409
status:
13791410
description: Status defines the observed state of LLMBatchGateway.
13801411
properties:

‎internal/controller/helm_batch.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -282,6 +282,13 @@ func specToBatchHelmValues(gw *batchv1alpha1.LLMBatchGateway, secretName string,
282282
"verbosity": int64(gw.Spec.GC.Config.Logging.Verbosity),
283283
}
284284
}
285+
if gw.Spec.GC.Config.Reconciler != nil {
286+
reconciler := map[string]any{
287+
"enabled": gw.Spec.GC.Config.Reconciler.Enabled,
288+
}
289+
setIfNotEmpty(reconciler, "interval", gw.Spec.GC.Config.Reconciler.Interval)
290+
gcConfig["reconciler"] = reconciler
291+
}
285292
}
286293
if len(gcCollector) > 0 {
287294
gcConfig["collector"] = gcCollector
@@ -462,6 +469,7 @@ func mergeProcessorConfig(m map[string]interface{}, cfg *batchv1alpha1.Processor
462469
if cfg.EnablePprof {
463470
m["enablePprof"] = true
464471
}
472+
setIfNotEmpty(m, "heartbeatInterval", cfg.HeartbeatInterval)
465473
}
466474

467475
func resourceRequirementsToMap(r *corev1.ResourceRequirements) map[string]interface{} {

‎internal/controller/helm_test.go‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -872,6 +872,83 @@ func TestSpecToHelmValues_GCConfig(t *testing.T) {
872872
}
873873
}
874874

875+
func TestSpecToHelmValues_ProcessorHeartbeatInterval(t *testing.T) {
876+
gw := minimalGateway()
877+
gw.Spec.Processor.Config = &batchv1alpha1.ProcessorConfigSpec{
878+
HeartbeatInterval: "5m",
879+
}
880+
881+
vals := specToBatchHelmValues(gw, testSecretName(gw), testImages())
882+
883+
processor := vals["processor"].(map[string]interface{})
884+
config := processor["config"].(map[string]interface{})
885+
if got := config["heartbeatInterval"]; got != "5m" {
886+
t.Errorf("heartbeatInterval = %v, want 5m", got)
887+
}
888+
}
889+
890+
func TestSpecToHelmValues_GCReconciler(t *testing.T) {
891+
gw := minimalGateway()
892+
gw.Spec.GC.Config = &batchv1alpha1.GCConfigSpec{
893+
Reconciler: &batchv1alpha1.ReconcilerSpec{
894+
Enabled: true,
895+
Interval: "60m",
896+
},
897+
}
898+
899+
vals := specToBatchHelmValues(gw, testSecretName(gw), testImages())
900+
901+
gc := vals["gc"].(map[string]interface{})
902+
config := gc["config"].(map[string]interface{})
903+
reconciler, ok := config["reconciler"].(map[string]interface{})
904+
if !ok {
905+
t.Fatalf("reconciler not found or wrong type: %T", config["reconciler"])
906+
}
907+
if got := reconciler["enabled"]; got != true {
908+
t.Errorf("reconciler.enabled = %v, want true", got)
909+
}
910+
if got := reconciler["interval"]; got != "60m" {
911+
t.Errorf("reconciler.interval = %v, want 60m", got)
912+
}
913+
}
914+
915+
func TestSpecToHelmValues_GCReconcilerDisabled(t *testing.T) {
916+
gw := minimalGateway()
917+
gw.Spec.GC.Config = &batchv1alpha1.GCConfigSpec{
918+
Reconciler: &batchv1alpha1.ReconcilerSpec{},
919+
}
920+
921+
vals := specToBatchHelmValues(gw, testSecretName(gw), testImages())
922+
923+
gc := vals["gc"].(map[string]interface{})
924+
config := gc["config"].(map[string]interface{})
925+
reconciler, ok := config["reconciler"].(map[string]interface{})
926+
if !ok {
927+
t.Fatalf("reconciler not found or wrong type: %T", config["reconciler"])
928+
}
929+
if got := reconciler["enabled"]; got != false {
930+
t.Errorf("reconciler.enabled = %v, want false", got)
931+
}
932+
if _, hasInterval := reconciler["interval"]; hasInterval {
933+
t.Errorf("reconciler.interval should be absent when not set, got %v", reconciler["interval"])
934+
}
935+
}
936+
937+
func TestSpecToHelmValues_GCReconcilerNil(t *testing.T) {
938+
gw := minimalGateway()
939+
gw.Spec.GC.Config = &batchv1alpha1.GCConfigSpec{}
940+
941+
vals := specToBatchHelmValues(gw, testSecretName(gw), testImages())
942+
943+
gc := vals["gc"].(map[string]interface{})
944+
if _, hasConfig := gc["config"]; hasConfig {
945+
config := gc["config"].(map[string]interface{})
946+
if _, hasReconciler := config["reconciler"]; hasReconciler {
947+
t.Error("reconciler should not be present when Reconciler is nil")
948+
}
949+
}
950+
}
951+
875952
func TestSpecToHelmValues_InferenceGatewayMaxRetries(t *testing.T) {
876953
maxRetries := int32(3)
877954
gw := minimalGateway()

‎internal/controller/llmbatchgateway_controller_test.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -758,6 +758,7 @@ func TestReconcile(t *testing.T) {
758758
},
759759
DefaultOutputExpirationSeconds: 7200,
760760
EnablePprof: true,
761+
HeartbeatInterval: "10m",
761762
}
762763
if err := k8sClient.Update(ctx, gw); err != nil {
763764
t.Fatalf("updating CR: %v", err)
@@ -774,6 +775,7 @@ func TestReconcile(t *testing.T) {
774775
assertConfigMapContains(t, cm, "per_endpoint: 25")
775776
assertConfigMapContains(t, cm, "default_output_expiration_seconds: 7200")
776777
assertConfigMapContains(t, cm, "enable_pprof: true")
778+
assertConfigMapContains(t, cm, `heartbeat_interval: "10m"`)
777779

778780
procAfter := findOwnedDeployment(ctx, t, gw, "processor")
779781
checksumAfter := procAfter.Spec.Template.Annotations["checksum/config"]
@@ -806,6 +808,10 @@ func TestReconcile(t *testing.T) {
806808
gw.Spec.GC.Config = &batchv1alpha1.GCConfigSpec{
807809
DryRun: true,
808810
MaxConcurrency: 10,
811+
Reconciler: &batchv1alpha1.ReconcilerSpec{
812+
Enabled: true,
813+
Interval: "90m",
814+
},
809815
}
810816
if err := k8sClient.Update(ctx, gw); err != nil {
811817
t.Fatalf("updating CR: %v", err)
@@ -820,6 +826,8 @@ func TestReconcile(t *testing.T) {
820826
assertConfigMapContains(t, cm, `interval: "1h"`)
821827
assertConfigMapContains(t, cm, "dry_run: true")
822828
assertConfigMapContains(t, cm, "max_concurrency: 10")
829+
assertConfigMapContains(t, cm, "enabled: true")
830+
assertConfigMapContains(t, cm, `interval: "90m"`)
823831

824832
gcAfter := findOwnedDeployment(ctx, t, gw, "gc")
825833
checksumAfter := gcAfter.Spec.Template.Annotations["checksum/config"]

0 commit comments

Comments
 (0)