diff --git a/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelroutes.yaml b/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelroutes.yaml
index ed84f4132..d644a7454 100644
--- a/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelroutes.yaml
+++ b/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelroutes.yaml
@@ -355,6 +355,73 @@ spec:
rule: self.modelName != "" || size(self.loraAdapters) > 0
status:
description: ModelRouteStatus defines the observed state of ModelRoute.
+ properties:
+ conditions:
+ description: |-
+ Conditions track the lifecycle of this ModelRoute.
+ Types:
+ - "Ready": true when the route has been registered with the router store
+ and is available for request matching.
+ items:
+ description: Condition contains details for one aspect of the current
+ state of this API Resource.
+ properties:
+ lastTransitionTime:
+ description: |-
+ lastTransitionTime is the last time the condition transitioned from one status to another.
+ This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable.
+ format: date-time
+ type: string
+ message:
+ description: |-
+ message is a human readable message indicating details about the transition.
+ This may be an empty string.
+ maxLength: 32768
+ type: string
+ observedGeneration:
+ description: |-
+ observedGeneration represents the .metadata.generation that the condition was set based upon.
+ For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date
+ with respect to the current state of the instance.
+ format: int64
+ minimum: 0
+ type: integer
+ reason:
+ description: |-
+ reason contains a programmatic identifier indicating the reason for the condition's last transition.
+ Producers of specific condition types may define expected values and meanings for this field,
+ and whether the values are considered a guaranteed API.
+ The value should be a CamelCase string.
+ This field may not be empty.
+ maxLength: 1024
+ minLength: 1
+ pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+ type: string
+ status:
+ description: status of the condition, one of True, False, Unknown.
+ enum:
+ - "True"
+ - "False"
+ - Unknown
+ type: string
+ type:
+ description: type of condition in CamelCase or in foo.example.com/CamelCase.
+ maxLength: 316
+ pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+ type: string
+ required:
+ - lastTransitionTime
+ - message
+ - reason
+ - status
+ - type
+ type: object
+ type: array
+ observedGeneration:
+ description: observedGeneration is the most recent generation observed
+ for this ModelRoute.
+ format: int64
+ type: integer
type: object
required:
- spec
diff --git a/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelservers.yaml b/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelservers.yaml
index f9f32bd31..de538f23b 100644
--- a/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelservers.yaml
+++ b/charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelservers.yaml
@@ -159,6 +159,87 @@ spec:
type: object
status:
description: ModelServerStatus defines the observed state of ModelServer.
+ properties:
+ conditions:
+ description: |-
+ Conditions track the lifecycle of this ModelServer.
+ Types:
+ - "Ready": true when at least one ready pod is matched and the ModelServer has been
+ registered with the router store. Signals that the ModelServer can serve traffic.
+ - "AllPodsReady": true when at least one pod is matched and every matched pod is ready
+ and registered with the router store.
+ items:
+ description: Condition contains details for one aspect of the current
+ state of this API Resource.
+ properties:
+ lastTransitionTime:
+ description: |-
+ lastTransitionTime is the last time the condition transitioned from one status to another.
+ This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable.
+ format: date-time
+ type: string
+ message:
+ description: |-
+ message is a human readable message indicating details about the transition.
+ This may be an empty string.
+ maxLength: 32768
+ type: string
+ observedGeneration:
+ description: |-
+ observedGeneration represents the .metadata.generation that the condition was set based upon.
+ For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date
+ with respect to the current state of the instance.
+ format: int64
+ minimum: 0
+ type: integer
+ reason:
+ description: |-
+ reason contains a programmatic identifier indicating the reason for the condition's last transition.
+ Producers of specific condition types may define expected values and meanings for this field,
+ and whether the values are considered a guaranteed API.
+ The value should be a CamelCase string.
+ This field may not be empty.
+ maxLength: 1024
+ minLength: 1
+ pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+ type: string
+ status:
+ description: status of the condition, one of True, False, Unknown.
+ enum:
+ - "True"
+ - "False"
+ - Unknown
+ type: string
+ type:
+ description: type of condition in CamelCase or in foo.example.com/CamelCase.
+ maxLength: 316
+ pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+ type: string
+ required:
+ - lastTransitionTime
+ - message
+ - reason
+ - status
+ - type
+ type: object
+ type: array
+ matchedReplicas:
+ description: |-
+ matchedReplicas is the total number of pods (ready or not) that match the
+ workloadSelector.
+ format: int32
+ type: integer
+ observedGeneration:
+ description: observedGeneration is the most recent generation observed
+ for this ModelServer.
+ format: int64
+ type: integer
+ readyReplicas:
+ description: |-
+ readyReplicas is the number of ready pods that are currently matched by the
+ workloadSelector and have been successfully registered in the router's store.
+ format: int32
+ type: integer
type: object
type: object
served: true
diff --git a/client-go/applyconfiguration/networking/v1alpha1/modelroute.go b/client-go/applyconfiguration/networking/v1alpha1/modelroute.go
index b082e08c4..42357e0dd 100644
--- a/client-go/applyconfiguration/networking/v1alpha1/modelroute.go
+++ b/client-go/applyconfiguration/networking/v1alpha1/modelroute.go
@@ -19,7 +19,6 @@ limitations under the License.
package v1alpha1
import (
- networkingv1alpha1 "github.com/volcano-sh/kthena/pkg/apis/networking/v1alpha1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
types "k8s.io/apimachinery/pkg/types"
v1 "k8s.io/client-go/applyconfigurations/meta/v1"
@@ -30,8 +29,8 @@ import (
type ModelRouteApplyConfiguration struct {
v1.TypeMetaApplyConfiguration `json:",inline"`
*v1.ObjectMetaApplyConfiguration `json:"metadata,omitempty"`
- Spec *ModelRouteSpecApplyConfiguration `json:"spec,omitempty"`
- Status *networkingv1alpha1.ModelRouteStatus `json:"status,omitempty"`
+ Spec *ModelRouteSpecApplyConfiguration `json:"spec,omitempty"`
+ Status *ModelRouteStatusApplyConfiguration `json:"status,omitempty"`
}
// ModelRoute constructs a declarative configuration of the ModelRoute type for use with
@@ -215,8 +214,8 @@ func (b *ModelRouteApplyConfiguration) WithSpec(value *ModelRouteSpecApplyConfig
// WithStatus sets the Status field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Status field is set to the value of the last call.
-func (b *ModelRouteApplyConfiguration) WithStatus(value networkingv1alpha1.ModelRouteStatus) *ModelRouteApplyConfiguration {
- b.Status = &value
+func (b *ModelRouteApplyConfiguration) WithStatus(value *ModelRouteStatusApplyConfiguration) *ModelRouteApplyConfiguration {
+ b.Status = value
return b
}
diff --git a/client-go/applyconfiguration/networking/v1alpha1/modelroutestatus.go b/client-go/applyconfiguration/networking/v1alpha1/modelroutestatus.go
new file mode 100644
index 000000000..3615fa569
--- /dev/null
+++ b/client-go/applyconfiguration/networking/v1alpha1/modelroutestatus.go
@@ -0,0 +1,57 @@
+/*
+Copyright The Volcano Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by applyconfiguration-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+ v1 "k8s.io/client-go/applyconfigurations/meta/v1"
+)
+
+// ModelRouteStatusApplyConfiguration represents a declarative configuration of the ModelRouteStatus type for use
+// with apply.
+type ModelRouteStatusApplyConfiguration struct {
+ ObservedGeneration *int64 `json:"observedGeneration,omitempty"`
+ Conditions []v1.ConditionApplyConfiguration `json:"conditions,omitempty"`
+}
+
+// ModelRouteStatusApplyConfiguration constructs a declarative configuration of the ModelRouteStatus type for use with
+// apply.
+func ModelRouteStatus() *ModelRouteStatusApplyConfiguration {
+ return &ModelRouteStatusApplyConfiguration{}
+}
+
+// WithObservedGeneration sets the ObservedGeneration field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ObservedGeneration field is set to the value of the last call.
+func (b *ModelRouteStatusApplyConfiguration) WithObservedGeneration(value int64) *ModelRouteStatusApplyConfiguration {
+ b.ObservedGeneration = &value
+ return b
+}
+
+// WithConditions adds the given value to the Conditions field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, values provided by each call will be appended to the Conditions field.
+func (b *ModelRouteStatusApplyConfiguration) WithConditions(values ...*v1.ConditionApplyConfiguration) *ModelRouteStatusApplyConfiguration {
+ for i := range values {
+ if values[i] == nil {
+ panic("nil value passed to WithConditions")
+ }
+ b.Conditions = append(b.Conditions, *values[i])
+ }
+ return b
+}
diff --git a/client-go/applyconfiguration/networking/v1alpha1/modelserver.go b/client-go/applyconfiguration/networking/v1alpha1/modelserver.go
index b52d80d60..e743bbba4 100644
--- a/client-go/applyconfiguration/networking/v1alpha1/modelserver.go
+++ b/client-go/applyconfiguration/networking/v1alpha1/modelserver.go
@@ -19,7 +19,6 @@ limitations under the License.
package v1alpha1
import (
- networkingv1alpha1 "github.com/volcano-sh/kthena/pkg/apis/networking/v1alpha1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
types "k8s.io/apimachinery/pkg/types"
v1 "k8s.io/client-go/applyconfigurations/meta/v1"
@@ -30,8 +29,8 @@ import (
type ModelServerApplyConfiguration struct {
v1.TypeMetaApplyConfiguration `json:",inline"`
*v1.ObjectMetaApplyConfiguration `json:"metadata,omitempty"`
- Spec *ModelServerSpecApplyConfiguration `json:"spec,omitempty"`
- Status *networkingv1alpha1.ModelServerStatus `json:"status,omitempty"`
+ Spec *ModelServerSpecApplyConfiguration `json:"spec,omitempty"`
+ Status *ModelServerStatusApplyConfiguration `json:"status,omitempty"`
}
// ModelServer constructs a declarative configuration of the ModelServer type for use with
@@ -215,8 +214,8 @@ func (b *ModelServerApplyConfiguration) WithSpec(value *ModelServerSpecApplyConf
// WithStatus sets the Status field in the declarative configuration to the given value
// and returns the receiver, so that objects can be built by chaining "With" function invocations.
// If called multiple times, the Status field is set to the value of the last call.
-func (b *ModelServerApplyConfiguration) WithStatus(value networkingv1alpha1.ModelServerStatus) *ModelServerApplyConfiguration {
- b.Status = &value
+func (b *ModelServerApplyConfiguration) WithStatus(value *ModelServerStatusApplyConfiguration) *ModelServerApplyConfiguration {
+ b.Status = value
return b
}
diff --git a/client-go/applyconfiguration/networking/v1alpha1/modelserverstatus.go b/client-go/applyconfiguration/networking/v1alpha1/modelserverstatus.go
new file mode 100644
index 000000000..b4ee96c0e
--- /dev/null
+++ b/client-go/applyconfiguration/networking/v1alpha1/modelserverstatus.go
@@ -0,0 +1,75 @@
+/*
+Copyright The Volcano Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by applyconfiguration-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+ v1 "k8s.io/client-go/applyconfigurations/meta/v1"
+)
+
+// ModelServerStatusApplyConfiguration represents a declarative configuration of the ModelServerStatus type for use
+// with apply.
+type ModelServerStatusApplyConfiguration struct {
+ ObservedGeneration *int64 `json:"observedGeneration,omitempty"`
+ ReadyReplicas *int32 `json:"readyReplicas,omitempty"`
+ MatchedReplicas *int32 `json:"matchedReplicas,omitempty"`
+ Conditions []v1.ConditionApplyConfiguration `json:"conditions,omitempty"`
+}
+
+// ModelServerStatusApplyConfiguration constructs a declarative configuration of the ModelServerStatus type for use with
+// apply.
+func ModelServerStatus() *ModelServerStatusApplyConfiguration {
+ return &ModelServerStatusApplyConfiguration{}
+}
+
+// WithObservedGeneration sets the ObservedGeneration field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ObservedGeneration field is set to the value of the last call.
+func (b *ModelServerStatusApplyConfiguration) WithObservedGeneration(value int64) *ModelServerStatusApplyConfiguration {
+ b.ObservedGeneration = &value
+ return b
+}
+
+// WithReadyReplicas sets the ReadyReplicas field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ReadyReplicas field is set to the value of the last call.
+func (b *ModelServerStatusApplyConfiguration) WithReadyReplicas(value int32) *ModelServerStatusApplyConfiguration {
+ b.ReadyReplicas = &value
+ return b
+}
+
+// WithMatchedReplicas sets the MatchedReplicas field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the MatchedReplicas field is set to the value of the last call.
+func (b *ModelServerStatusApplyConfiguration) WithMatchedReplicas(value int32) *ModelServerStatusApplyConfiguration {
+ b.MatchedReplicas = &value
+ return b
+}
+
+// WithConditions adds the given value to the Conditions field in the declarative configuration
+// and returns the receiver, so that objects can be build by chaining "With" function invocations.
+// If called multiple times, values provided by each call will be appended to the Conditions field.
+func (b *ModelServerStatusApplyConfiguration) WithConditions(values ...*v1.ConditionApplyConfiguration) *ModelServerStatusApplyConfiguration {
+ for i := range values {
+ if values[i] == nil {
+ panic("nil value passed to WithConditions")
+ }
+ b.Conditions = append(b.Conditions, *values[i])
+ }
+ return b
+}
diff --git a/client-go/applyconfiguration/utils.go b/client-go/applyconfiguration/utils.go
index 9247efca7..7b266f57c 100644
--- a/client-go/applyconfiguration/utils.go
+++ b/client-go/applyconfiguration/utils.go
@@ -46,10 +46,14 @@ func ForKind(kind schema.GroupVersionKind) interface{} {
return &networkingv1alpha1.ModelRouteApplyConfiguration{}
case v1alpha1.SchemeGroupVersion.WithKind("ModelRouteSpec"):
return &networkingv1alpha1.ModelRouteSpecApplyConfiguration{}
+ case v1alpha1.SchemeGroupVersion.WithKind("ModelRouteStatus"):
+ return &networkingv1alpha1.ModelRouteStatusApplyConfiguration{}
case v1alpha1.SchemeGroupVersion.WithKind("ModelServer"):
return &networkingv1alpha1.ModelServerApplyConfiguration{}
case v1alpha1.SchemeGroupVersion.WithKind("ModelServerSpec"):
return &networkingv1alpha1.ModelServerSpecApplyConfiguration{}
+ case v1alpha1.SchemeGroupVersion.WithKind("ModelServerStatus"):
+ return &networkingv1alpha1.ModelServerStatusApplyConfiguration{}
case v1alpha1.SchemeGroupVersion.WithKind("PDGroup"):
return &networkingv1alpha1.PDGroupApplyConfiguration{}
case v1alpha1.SchemeGroupVersion.WithKind("RateLimit"):
diff --git a/cmd/kthena-router/app/controller.go b/cmd/kthena-router/app/controller.go
index e70e2fac0..2ece44348 100644
--- a/cmd/kthena-router/app/controller.go
+++ b/cmd/kthena-router/app/controller.go
@@ -77,11 +77,11 @@ func startControllers(store datastore.Store, stop <-chan struct{}, enableGateway
kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
kthenaInformerFactory := kthenaInformers.NewSharedInformerFactory(kthenaClient, 0)
- modelRouteController, err := controller.NewModelRouteController(kthenaInformerFactory, store)
+ modelRouteController, err := controller.NewModelRouteController(kthenaClient, kthenaInformerFactory, store)
if err != nil {
klog.Fatalf("Error creating model route controller: %s", err.Error())
}
- modelServerController, err := controller.NewModelServerController(kthenaInformerFactory, kubeInformerFactory, store)
+ modelServerController, err := controller.NewModelServerController(kthenaClient, kthenaInformerFactory, kubeInformerFactory, store)
if err != nil {
klog.Fatalf("Error creating model server controller: %s", err.Error())
}
diff --git a/docs/kthena/docs/reference/crd/networking.serving.volcano.sh.md b/docs/kthena/docs/reference/crd/networking.serving.volcano.sh.md
index fa7bc47a6..0f6f45e9d 100644
--- a/docs/kthena/docs/reference/crd/networking.serving.volcano.sh.md
+++ b/docs/kthena/docs/reference/crd/networking.serving.volcano.sh.md
@@ -140,6 +140,8 @@ _Appears in:_
| `status` _[ModelRouteStatus](#modelroutestatus)_ | | | |
+
+
#### ModelRouteList
@@ -188,6 +190,9 @@ ModelRouteStatus defines the observed state of ModelRoute.
_Appears in:_
- [ModelRoute](#modelroute)
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `observedGeneration` _integer_ | observedGeneration is the most recent generation observed for this ModelRoute. | | |
#### ModelServer
@@ -209,6 +214,8 @@ _Appears in:_
| `status` _[ModelServerStatus](#modelserverstatus)_ | | | |
+
+
#### ModelServerList
@@ -258,6 +265,11 @@ ModelServerStatus defines the observed state of ModelServer.
_Appears in:_
- [ModelServer](#modelserver)
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `observedGeneration` _integer_ | observedGeneration is the most recent generation observed for this ModelServer. | | |
+| `readyReplicas` _integer_ | readyReplicas is the number of ready pods that are currently matched by the
workloadSelector and have been successfully registered in the router's store. | | |
+| `matchedReplicas` _integer_ | matchedReplicas is the total number of pods (ready or not) that match the
workloadSelector. | | |
#### PDGroup
diff --git a/pkg/apis/networking/v1alpha1/modelroute_types.go b/pkg/apis/networking/v1alpha1/modelroute_types.go
index cc6062a05..2b0202834 100644
--- a/pkg/apis/networking/v1alpha1/modelroute_types.go
+++ b/pkg/apis/networking/v1alpha1/modelroute_types.go
@@ -165,8 +165,32 @@ const (
Month RateLimitUnit = "month"
)
+// ModelRouteConditionType is the type of a status condition on a ModelRoute.
+type ModelRouteConditionType string
+
+const (
+ // ModelRouteConditionReady indicates the route has been registered in the router store.
+ ModelRouteConditionReady ModelRouteConditionType = "Ready"
+
+ // ReasonRouteRegistered is the reason used when a ModelRoute has been successfully
+ // registered in the router store.
+ ReasonRouteRegistered = "RouteRegistered"
+)
+
// ModelRouteStatus defines the observed state of ModelRoute.
type ModelRouteStatus struct {
+ // observedGeneration is the most recent generation observed for this ModelRoute.
+ // +optional
+ ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+
+ // Conditions track the lifecycle of this ModelRoute.
+ // Types:
+ // - "Ready": true when the route has been registered with the router store
+ // and is available for request matching.
+ // +optional
+ // +patchMergeKey=type
+ // +patchStrategy=merge
+ Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`
}
// +kubebuilder:object:root=true
diff --git a/pkg/apis/networking/v1alpha1/modelserver_types.go b/pkg/apis/networking/v1alpha1/modelserver_types.go
index 04622d9f8..73e326f05 100644
--- a/pkg/apis/networking/v1alpha1/modelserver_types.go
+++ b/pkg/apis/networking/v1alpha1/modelserver_types.go
@@ -110,6 +110,17 @@ const (
ConnectorTypeMoonCake KVConnectorType = "mooncake" // Indicates `MoonCakeConnector` in vllm-ascend
)
+// ModelServerConditionType is the type of a status condition on a ModelServer.
+type ModelServerConditionType string
+
+const (
+ // ModelServerConditionReady indicates at least one ready pod is registered in the router store.
+ ModelServerConditionReady ModelServerConditionType = "Ready"
+
+ // ModelServerConditionAllPodsReady indicates every matched pod is ready and registered in the router store.
+ ModelServerConditionAllPodsReady ModelServerConditionType = "AllPodsReady"
+)
+
// KVConnectorSpec defines KV connector configuration for PD disaggregated routing
type KVConnectorSpec struct {
// Type specifies the connector type.
@@ -143,8 +154,30 @@ type Retry struct {
// ModelServerStatus defines the observed state of ModelServer.
type ModelServerStatus struct {
- // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
- // Important: Run "make" to regenerate code after modifying this file
+ // observedGeneration is the most recent generation observed for this ModelServer.
+ // +optional
+ ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+
+ // readyReplicas is the number of ready pods that are currently matched by the
+ // workloadSelector and have been successfully registered in the router's store.
+ // +optional
+ ReadyReplicas int32 `json:"readyReplicas,omitempty"`
+
+ // matchedReplicas is the total number of pods (ready or not) that match the
+ // workloadSelector.
+ // +optional
+ MatchedReplicas int32 `json:"matchedReplicas,omitempty"`
+
+ // Conditions track the lifecycle of this ModelServer.
+ // Types:
+ // - "Ready": true when at least one ready pod is matched and the ModelServer has been
+ // registered with the router store. Signals that the ModelServer can serve traffic.
+ // - "AllPodsReady": true when at least one pod is matched and every matched pod is ready
+ // and registered with the router store.
+ // +optional
+ // +patchMergeKey=type
+ // +patchStrategy=merge
+ Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`
}
// +kubebuilder:object:root=true
diff --git a/pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go
index 26f1f0a88..ab7b4f698 100644
--- a/pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go
@@ -128,7 +128,7 @@ func (in *ModelRoute) DeepCopyInto(out *ModelRoute) {
out.TypeMeta = in.TypeMeta
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
in.Spec.DeepCopyInto(&out.Spec)
- out.Status = in.Status
+ in.Status.DeepCopyInto(&out.Status)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelRoute.
@@ -227,6 +227,13 @@ func (in *ModelRouteSpec) DeepCopy() *ModelRouteSpec {
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ModelRouteStatus) DeepCopyInto(out *ModelRouteStatus) {
*out = *in
+ if in.Conditions != nil {
+ in, out := &in.Conditions, &out.Conditions
+ *out = make([]metav1.Condition, len(*in))
+ for i := range *in {
+ (*in)[i].DeepCopyInto(&(*out)[i])
+ }
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelRouteStatus.
@@ -245,7 +252,7 @@ func (in *ModelServer) DeepCopyInto(out *ModelServer) {
out.TypeMeta = in.TypeMeta
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
in.Spec.DeepCopyInto(&out.Spec)
- out.Status = in.Status
+ in.Status.DeepCopyInto(&out.Status)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelServer.
@@ -337,6 +344,13 @@ func (in *ModelServerSpec) DeepCopy() *ModelServerSpec {
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ModelServerStatus) DeepCopyInto(out *ModelServerStatus) {
*out = *in
+ if in.Conditions != nil {
+ in, out := &in.Conditions, &out.Conditions
+ *out = make([]metav1.Condition, len(*in))
+ for i := range *in {
+ (*in)[i].DeepCopyInto(&(*out)[i])
+ }
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelServerStatus.
diff --git a/pkg/kthena-router/controller/modelroute_controller.go b/pkg/kthena-router/controller/modelroute_controller.go
index c4c6be823..e29f534cc 100644
--- a/pkg/kthena-router/controller/modelroute_controller.go
+++ b/pkg/kthena-router/controller/modelroute_controller.go
@@ -17,23 +17,30 @@ limitations under the License.
package controller
import (
+ "context"
"fmt"
"sync/atomic"
"time"
"k8s.io/apimachinery/pkg/api/errors"
+ "k8s.io/apimachinery/pkg/api/meta"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
+ "k8s.io/client-go/util/retry"
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"
+ kthenaclient "github.com/volcano-sh/kthena/client-go/clientset/versioned"
informersv1alpha1 "github.com/volcano-sh/kthena/client-go/informers/externalversions"
listerv1alpha1 "github.com/volcano-sh/kthena/client-go/listers/networking/v1alpha1"
+ aiv1alpha1 "github.com/volcano-sh/kthena/pkg/apis/networking/v1alpha1"
"github.com/volcano-sh/kthena/pkg/kthena-router/datastore"
)
type ModelRouteController struct {
+ kthenaClient kthenaclient.Interface
modelRouteLister listerv1alpha1.ModelRouteLister
modelRouteSynced cache.InformerSynced
registration cache.ResourceEventHandlerRegistration
@@ -44,12 +51,14 @@ type ModelRouteController struct {
}
func NewModelRouteController(
+ kthenaClient kthenaclient.Interface,
kthenaInformerFactory informersv1alpha1.SharedInformerFactory,
store datastore.Store,
) (*ModelRouteController, error) {
modelRouteInformer := kthenaInformerFactory.Networking().V1alpha1().ModelRoutes()
controller := &ModelRouteController{
+ kthenaClient: kthenaClient,
modelRouteLister: modelRouteInformer.Lister(),
modelRouteSynced: modelRouteInformer.Informer().HasSynced,
workqueue: workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[any]()),
@@ -151,7 +160,45 @@ func (c *ModelRouteController) syncHandler(key string) error {
return err
}
- return nil
+ return c.updateModelRouteStatus(mr)
+}
+
+func (c *ModelRouteController) updateModelRouteStatus(mr *aiv1alpha1.ModelRoute) error {
+ // If the Ready condition is already set with the expected values,
+ // no update is needed. We check the condition content rather than
+ // Status.ObservedGeneration to avoid unnecessary status writes on
+ // spec-only changes (e.g. rate limit updates). Unnecessary writes
+ // increment the resourceVersion and cause conflicts with clients
+ // that are simultaneously updating the same ModelRoute.
+ readyCond := meta.FindStatusCondition(mr.Status.Conditions, string(aiv1alpha1.ModelRouteConditionReady))
+ if readyCond != nil &&
+ readyCond.Status == metav1.ConditionTrue &&
+ readyCond.Reason == aiv1alpha1.ReasonRouteRegistered &&
+ readyCond.ObservedGeneration == mr.Generation {
+ return nil
+ }
+
+ return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+ // Get latest version from API server to avoid conflicts
+ ctx := context.Background()
+ latest, err := c.kthenaClient.NetworkingV1alpha1().ModelRoutes(mr.Namespace).Get(ctx, mr.Name, metav1.GetOptions{})
+ if err != nil {
+ return err
+ }
+
+ mrCopy := latest.DeepCopy()
+ mrCopy.Status.ObservedGeneration = latest.Generation
+
+ meta.SetStatusCondition(&mrCopy.Status.Conditions, metav1.Condition{
+ Type: string(aiv1alpha1.ModelRouteConditionReady),
+ Status: metav1.ConditionTrue,
+ Reason: aiv1alpha1.ReasonRouteRegistered,
+ Message: "ModelRoute registered in router store",
+ })
+
+ _, err = c.kthenaClient.NetworkingV1alpha1().ModelRoutes(mr.Namespace).UpdateStatus(ctx, mrCopy, metav1.UpdateOptions{})
+ return err
+ })
}
func (c *ModelRouteController) enqueueModelRoute(obj interface{}) {
diff --git a/pkg/kthena-router/controller/modelroute_controller_test.go b/pkg/kthena-router/controller/modelroute_controller_test.go
index 9fe0a0217..f866ef09d 100644
--- a/pkg/kthena-router/controller/modelroute_controller_test.go
+++ b/pkg/kthena-router/controller/modelroute_controller_test.go
@@ -23,6 +23,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
kthenafake "github.com/volcano-sh/kthena/client-go/clientset/versioned/fake"
@@ -36,7 +37,7 @@ func TestModelRouteController_Lifecycle(t *testing.T) {
kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
store := datastore.New()
- controller, err := NewModelRouteController(kthenaInformerFactory, store)
+ controller, err := NewModelRouteController(kthenaClient, kthenaInformerFactory, store)
require.NoError(t, err)
stop := make(chan struct{})
@@ -148,7 +149,7 @@ func TestModelRouteController_ErrorHandling(t *testing.T) {
kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
store := datastore.New()
- controller, err := NewModelRouteController(kthenaInformerFactory, store)
+ controller, err := NewModelRouteController(kthenaClient, kthenaInformerFactory, store)
require.NoError(t, err)
stop := make(chan struct{})
@@ -174,7 +175,7 @@ func TestModelRouteController_WorkQueueProcessing(t *testing.T) {
kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
store := datastore.New()
- controller, err := NewModelRouteController(kthenaInformerFactory, store)
+ controller, err := NewModelRouteController(kthenaClient, kthenaInformerFactory, store)
require.NoError(t, err)
stop := make(chan struct{})
@@ -197,3 +198,101 @@ func TestModelRouteController_WorkQueueProcessing(t *testing.T) {
assert.True(t, result)
})
}
+
+// TestModelRouteController_StatusUpdate verifies that syncHandler sets
+// the Ready condition on the ModelRoute status.
+func TestModelRouteController_StatusUpdate(t *testing.T) {
+ kthenaClient := kthenafake.NewSimpleClientset()
+ kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
+ store := datastore.New()
+
+ controller, err := NewModelRouteController(kthenaClient, kthenaInformerFactory, store)
+ require.NoError(t, err)
+
+ stop := make(chan struct{})
+ defer close(stop)
+ kthenaInformerFactory.Start(stop)
+
+ // sync of a newly created ModelRoute writes the Ready condition
+ t.Run("SetsReadyConditionAfterSync", func(t *testing.T) {
+ mr := &aiv1alpha1.ModelRoute{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "test-status-route",
+ },
+ Spec: aiv1alpha1.ModelRouteSpec{
+ ModelName: "status-test-model",
+ Rules: []*aiv1alpha1.Rule{
+ {
+ Name: "rule-1",
+ TargetModels: []*aiv1alpha1.TargetModel{
+ {ModelServerName: "test-server"},
+ },
+ },
+ },
+ },
+ }
+
+ _, err := kthenaClient.NetworkingV1alpha1().ModelRoutes("default").Create(
+ context.Background(), mr, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ waitForCacheSync(t, 5*time.Second, controller.modelRouteSynced)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelRouteLister.ModelRoutes("default").Get("test-status-route")
+ return err == nil
+ })
+
+ err = controller.syncHandler("default/test-status-route")
+ assert.NoError(t, err)
+
+ // verify status was written to the fake API server
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelRoutes("default").Get(
+ context.Background(), "test-status-route", metav1.GetOptions{})
+ assert.NoError(t, err)
+
+ cond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelRouteConditionReady))
+ require.NotNil(t, cond)
+ assert.Equal(t, metav1.ConditionTrue, cond.Status)
+ assert.Equal(t, aiv1alpha1.ReasonRouteRegistered, cond.Reason)
+ })
+
+ // sync skips status update when already up-to-date
+ t.Run("SkipsUpdateWhenAlreadyReady", func(t *testing.T) {
+ mr := &aiv1alpha1.ModelRoute{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "test-status-skip",
+ },
+ Spec: aiv1alpha1.ModelRouteSpec{
+ ModelName: "skip-test-model",
+ Rules: []*aiv1alpha1.Rule{
+ {
+ Name: "rule-1",
+ TargetModels: []*aiv1alpha1.TargetModel{
+ {ModelServerName: "test-server"},
+ },
+ },
+ },
+ },
+ }
+
+ _, err := kthenaClient.NetworkingV1alpha1().ModelRoutes("default").Create(
+ context.Background(), mr, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ waitForCacheSync(t, 5*time.Second, controller.modelRouteSynced)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelRouteLister.ModelRoutes("default").Get("test-status-skip")
+ return err == nil
+ })
+
+ // first sync: sets status
+ err = controller.syncHandler("default/test-status-skip")
+ assert.NoError(t, err)
+
+ // second sync: should skip (already ready) and not error
+ err = controller.syncHandler("default/test-status-skip")
+ assert.NoError(t, err)
+ })
+}
diff --git a/pkg/kthena-router/controller/modelserver_controller.go b/pkg/kthena-router/controller/modelserver_controller.go
index a39aff20c..8c5eb0ce9 100644
--- a/pkg/kthena-router/controller/modelserver_controller.go
+++ b/pkg/kthena-router/controller/modelserver_controller.go
@@ -17,6 +17,7 @@ limitations under the License.
package controller
import (
+ "context"
"fmt"
"sync/atomic"
"time"
@@ -24,6 +25,7 @@ import (
"istio.io/istio/pkg/util/sets"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
+ "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
@@ -32,9 +34,11 @@ import (
"k8s.io/client-go/informers"
corelisters "k8s.io/client-go/listers/core/v1"
"k8s.io/client-go/tools/cache"
+ "k8s.io/client-go/util/retry"
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"
+ kthenaclient "github.com/volcano-sh/kthena/client-go/clientset/versioned"
informersv1alpha1 "github.com/volcano-sh/kthena/client-go/informers/externalversions"
listerv1alpha1 "github.com/volcano-sh/kthena/client-go/listers/networking/v1alpha1"
aiv1alpha1 "github.com/volcano-sh/kthena/pkg/apis/networking/v1alpha1"
@@ -50,6 +54,15 @@ const (
ResourceTypePod ResourceType = "Pod"
)
+// Condition reasons for ModelServer status.
+const (
+ ReasonNoReadyPods = "NoReadyPods"
+ ReasonPodsReady = "PodsReady"
+ ReasonNoMatchedPods = "NoMatchedPods"
+ ReasonPodsNotFullyReady = "PodsNotFullyReady"
+ ReasonAllPodsReady = "AllPodsReady"
+)
+
// QueueItem represents an item in the work queue
type QueueItem struct {
ResourceType ResourceType
@@ -57,6 +70,7 @@ type QueueItem struct {
}
type ModelServerController struct {
+ kthenaClient kthenaclient.Interface
modelServerLister listerv1alpha1.ModelServerLister
podLister corelisters.PodLister
@@ -73,6 +87,7 @@ type ModelServerController struct {
}
func NewModelServerController(
+ kthenaClient kthenaclient.Interface,
kthenaInformerFactory informersv1alpha1.SharedInformerFactory,
kubeInformerFactory informers.SharedInformerFactory,
store datastore.Store,
@@ -81,6 +96,7 @@ func NewModelServerController(
podInformer := kubeInformerFactory.Core().V1().Pods()
controller := &ModelServerController{
+ kthenaClient: kthenaClient,
modelServerLister: modelServerInformer.Lister(),
podLister: podInformer.Lister(),
modelServerSynced: modelServerInformer.Informer().HasSynced,
@@ -214,7 +230,14 @@ func (c *ModelServerController) syncModelServerHandler(key string) error {
}
}
- _ = c.store.AddOrUpdateModelServer(ms, pods)
+ if err := c.store.AddOrUpdateModelServer(ms, pods); err != nil {
+ return err
+ }
+
+ if err := c.updateModelServerStatus(ms, podList, pods); err != nil {
+ klog.Errorf("Failed to update ModelServer status for %s/%s: %v", ms.Namespace, ms.Name, err)
+ return err
+ }
// Bind every ready pod selected by this ModelServer. Pods that already have
// an entry in the store get the binding appended so their runtime metrics and
@@ -245,6 +268,78 @@ func (c *ModelServerController) syncModelServerHandler(key string) error {
return nil
}
+func (c *ModelServerController) updateModelServerStatus(ms *aiv1alpha1.ModelServer, podList []*corev1.Pod, readyPods sets.Set[types.NamespacedName]) error {
+ matchedReplicas := int32(len(podList))
+ readyReplicas := int32(readyPods.Len())
+
+ // Ready condition (at least one pod ready)
+ readyStatus := metav1.ConditionFalse
+ readyReason := ReasonNoReadyPods
+ readyMessage := "No ready pods match workloadSelector"
+ if readyReplicas > 0 {
+ readyStatus = metav1.ConditionTrue
+ readyReason = ReasonPodsReady
+ readyMessage = fmt.Sprintf("%d ready pods matched", readyReplicas)
+ }
+
+ // AllPodsReady condition (every matched pod ready)
+ allPodsReadyStatus := metav1.ConditionFalse
+ allPodsReadyReason := ReasonNoMatchedPods
+ allPodsReadyMessage := "No pods match workloadSelector"
+ if matchedReplicas > 0 {
+ allPodsReadyReason = ReasonPodsNotFullyReady
+ allPodsReadyMessage = fmt.Sprintf("%d of %d matched pods are ready", readyReplicas, matchedReplicas)
+ if readyReplicas == matchedReplicas {
+ allPodsReadyStatus = metav1.ConditionTrue
+ allPodsReadyReason = ReasonAllPodsReady
+ allPodsReadyMessage = fmt.Sprintf("All %d matched pods are ready", matchedReplicas)
+ }
+ }
+
+ // Check if status is already up-to-date
+ if ms.Status.ObservedGeneration == ms.Generation &&
+ ms.Status.MatchedReplicas == matchedReplicas &&
+ ms.Status.ReadyReplicas == readyReplicas {
+ readyCond := meta.FindStatusCondition(ms.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ allReadyCond := meta.FindStatusCondition(ms.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ if readyCond != nil && readyCond.Status == readyStatus && readyCond.Reason == readyReason &&
+ allReadyCond != nil && allReadyCond.Status == allPodsReadyStatus && allReadyCond.Reason == allPodsReadyReason {
+ return nil
+ }
+ }
+
+ return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+ // Get latest version from API server to avoid conflicts
+ ctx := context.Background()
+ latest, err := c.kthenaClient.NetworkingV1alpha1().ModelServers(ms.Namespace).Get(ctx, ms.Name, metav1.GetOptions{})
+ if err != nil {
+ return err
+ }
+
+ msCopy := latest.DeepCopy()
+ msCopy.Status.ObservedGeneration = latest.Generation
+ msCopy.Status.MatchedReplicas = matchedReplicas
+ msCopy.Status.ReadyReplicas = readyReplicas
+
+ meta.SetStatusCondition(&msCopy.Status.Conditions, metav1.Condition{
+ Type: string(aiv1alpha1.ModelServerConditionReady),
+ Status: readyStatus,
+ Reason: readyReason,
+ Message: readyMessage,
+ })
+
+ meta.SetStatusCondition(&msCopy.Status.Conditions, metav1.Condition{
+ Type: string(aiv1alpha1.ModelServerConditionAllPodsReady),
+ Status: allPodsReadyStatus,
+ Reason: allPodsReadyReason,
+ Message: allPodsReadyMessage,
+ })
+
+ _, err = c.kthenaClient.NetworkingV1alpha1().ModelServers(ms.Namespace).UpdateStatus(ctx, msCopy, metav1.UpdateOptions{})
+ return err
+ })
+}
+
func (c *ModelServerController) syncPodHandler(key string) error {
namespace, name, err := cache.SplitMetaNamespaceKey(key)
if err != nil {
@@ -255,6 +350,9 @@ func (c *ModelServerController) syncPodHandler(key string) error {
pod, err := c.podLister.Pods(namespace).Get(name)
if errors.IsNotFound(err) {
_ = c.store.DeletePod(types.NamespacedName{Namespace: namespace, Name: name})
+ // When a pod is deleted, enqueue all ModelServers in the namespace
+ // to refresh their status (we no longer have the pod to check labels).
+ c.enqueueModelServersInNamespace(namespace)
return nil
}
if err != nil {
@@ -263,36 +361,79 @@ func (c *ModelServerController) syncPodHandler(key string) error {
if !isPodReady(pod) {
_ = c.store.DeletePod(types.NamespacedName{Namespace: namespace, Name: name})
+ // Pod became not ready: enqueue matching ModelServers to refresh status
+ c.enqueueMatchingModelServers(pod)
return nil
}
- return c.addOrUpdatePod(pod)
+ if err := c.addOrUpdatePod(pod); err != nil {
+ return err
+ }
+ // Pod became ready: enqueue matching ModelServers to refresh status
+ c.enqueueMatchingModelServers(pod)
+ return nil
}
// addOrUpdatePod finds all ModelServers that match the given pod
// and adds or updates the pod-server binding in the data store
func (c *ModelServerController) addOrUpdatePod(pod *corev1.Pod) error {
+ servers, err := c.getMatchingModelServers(pod)
+ if err != nil {
+ return err
+ }
+
+ if len(servers) > 0 {
+ if err := c.store.AddOrUpdatePod(pod, servers); err != nil {
+ return fmt.Errorf("failed to add or update pod %s/%s in data store: %v", pod.Namespace, pod.Name, err)
+ }
+ }
+
+ return nil
+}
+
+// getMatchingModelServers returns all ModelServers in the same namespace whose
+// workloadSelector matches the pod's labels.
+func (c *ModelServerController) getMatchingModelServers(pod *corev1.Pod) ([]*aiv1alpha1.ModelServer, error) {
modelServers, err := c.modelServerLister.ModelServers(pod.Namespace).List(labels.Everything())
if err != nil {
- return fmt.Errorf("failed to list ModelServers for pod %s/%s: %v", pod.Namespace, pod.Name, err)
+ return nil, fmt.Errorf("failed to list ModelServers for pod %s/%s: %v", pod.Namespace, pod.Name, err)
}
- servers := []*aiv1alpha1.ModelServer{}
+ var matching []*aiv1alpha1.ModelServer
for _, item := range modelServers {
selector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{MatchLabels: item.Spec.WorkloadSelector.MatchLabels})
if err != nil || !selector.Matches(labels.Set(pod.Labels)) {
continue
}
- servers = append(servers, item)
+ matching = append(matching, item)
}
+ return matching, nil
+}
- if len(servers) > 0 {
- if err := c.store.AddOrUpdatePod(pod, servers); err != nil {
- return fmt.Errorf("failed to add or update pod %s/%s in data store: %v", pod.Namespace, pod.Name, err)
- }
+// enqueueMatchingModelServers enqueues all ModelServers whose workloadSelector
+// matches the pod, triggering a status re-evaluation.
+func (c *ModelServerController) enqueueMatchingModelServers(pod *corev1.Pod) {
+ matching, err := c.getMatchingModelServers(pod)
+ if err != nil {
+ klog.V(4).Infof("failed to find matching ModelServers for pod %s/%s: %v", pod.Namespace, pod.Name, err)
+ return
+ }
+ for _, ms := range matching {
+ c.enqueueModelServer(ms)
}
+}
- return nil
+// enqueueModelServersInNamespace enqueues all ModelServers in the given namespace.
+// Used when a pod is deleted and we no longer have pod labels to narrow the search.
+func (c *ModelServerController) enqueueModelServersInNamespace(namespace string) {
+ msList, err := c.modelServerLister.ModelServers(namespace).List(labels.Everything())
+ if err != nil {
+ klog.V(4).Infof("failed to list ModelServers in namespace %s: %v", namespace, err)
+ return
+ }
+ for _, ms := range msList {
+ c.enqueueModelServer(ms)
+ }
}
func (c *ModelServerController) enqueueModelServer(obj interface{}) {
diff --git a/pkg/kthena-router/controller/modelserver_controller_test.go b/pkg/kthena-router/controller/modelserver_controller_test.go
index 972dc878b..f305a11ac 100644
--- a/pkg/kthena-router/controller/modelserver_controller_test.go
+++ b/pkg/kthena-router/controller/modelserver_controller_test.go
@@ -25,6 +25,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
corev1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
@@ -71,12 +72,13 @@ func TestModelServerController_ModelServerLifecycle(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
)
require.NoError(t, err)
- modelServerIndexer := kthenaInformerFactory.Networking().V1alpha1().ModelServers().Informer().GetIndexer()
+ _ = kthenaInformerFactory.Networking().V1alpha1().ModelServers().Informer().GetIndexer()
stop := make(chan struct{})
defer close(stop)
@@ -106,13 +108,15 @@ func TestModelServerController_ModelServerLifecycle(t *testing.T) {
t.Fatal("Failed to sync caches within timeout")
}
- require.NoError(t, modelServerIndexer.Add(ms.DeepCopy()))
- _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver")
+ // write through fake API so status update can Get
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
require.NoError(t, err)
- // Simulate controller receiving the event
- controller.enqueueModelServer(ms)
- assert.Equal(t, 1, controller.workqueue.Len())
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver")
+ return err == nil
+ })
// Process the queue item
err = controller.syncModelServerHandler("default/test-modelserver")
@@ -147,35 +151,31 @@ func TestModelServerController_ModelServerLifecycle(t *testing.T) {
},
}
- require.NoError(t, modelServerIndexer.Add(ms.DeepCopy()))
- _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-update")
+ // update via fake API, verify store reflects changes
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
require.NoError(t, err)
- // Process initial creation
- controller.enqueueModelServer(ms)
- err = controller.syncModelServerHandler("default/test-modelserver-update")
- assert.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-update")
+ return err == nil
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/test-modelserver-update"))
// Update ModelServer
updatedMS := ms.DeepCopy()
updatedMS.Labels["version"] = "v2"
updatedMS.Spec.WorkloadSelector.MatchLabels["environment"] = "production"
- require.NoError(t, modelServerIndexer.Update(updatedMS.DeepCopy()))
- cachedMS, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-update")
+ _, err = kthenaClient.NetworkingV1alpha1().ModelServers("default").Update(
+ context.Background(), updatedMS, metav1.UpdateOptions{})
require.NoError(t, err)
- assert.Equal(t, "v2", cachedMS.Labels["version"])
- // Simulate controller receiving update event
- controller.enqueueModelServer(updatedMS)
- // Clear any previous items from queue
- for controller.workqueue.Len() > 0 {
- item, _ := controller.workqueue.Get()
- controller.workqueue.Done(item)
- controller.workqueue.Forget(item)
- }
- controller.enqueueModelServer(updatedMS)
- assert.Equal(t, 1, controller.workqueue.Len())
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ cached, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-update")
+ return err == nil && cached.Labels["version"] == "v2"
+ })
// Process the update
err = controller.syncModelServerHandler("default/test-modelserver-update")
@@ -208,10 +208,16 @@ func TestModelServerController_ModelServerLifecycle(t *testing.T) {
},
}
- require.NoError(t, modelServerIndexer.Add(ms.DeepCopy()))
- _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-delete")
+ // delete via fake API, verify removed from store
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-delete")
+ return err == nil
+ })
+
// Process creation
err = controller.syncModelServerHandler("default/test-modelserver-delete")
assert.NoError(t, err)
@@ -223,9 +229,14 @@ func TestModelServerController_ModelServerLifecycle(t *testing.T) {
})
require.NotNil(t, storedMS, "ModelServer should be found in store before deletion")
- require.NoError(t, modelServerIndexer.Delete(ms.DeepCopy()))
- _, err = controller.modelServerLister.ModelServers("default").Get("test-modelserver-delete")
- assert.Error(t, err)
+ err = kthenaClient.NetworkingV1alpha1().ModelServers("default").Delete(
+ context.Background(), "test-modelserver-delete", metav1.DeleteOptions{})
+ assert.NoError(t, err)
+
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelServerLister.ModelServers("default").Get("test-modelserver-delete")
+ return err != nil
+ })
// Process the deletion - this should handle the NotFound error gracefully
err = controller.syncModelServerHandler("default/test-modelserver-delete")
@@ -273,6 +284,7 @@ func TestModelServerController_PodLifecycle(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -490,6 +502,7 @@ func TestModelServerController_ErrorHandling(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -535,6 +548,7 @@ func TestModelServerController_WorkQueueProcessing(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -611,6 +625,7 @@ func TestModelServerController_PodSelectionLogic(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -800,6 +815,7 @@ func TestModelServerController_ComprehensiveLifecycleTest(t *testing.T) {
// Create controller and store
store := newStoreWithMockBackend()
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -977,6 +993,7 @@ func TestModelServerController_SharedPods(t *testing.T) {
// Create controller
controller, err := NewModelServerController(
+ kthenaClient,
kthenaInformerFactory,
kubeInformerFactory,
store,
@@ -1103,3 +1120,528 @@ func waitForObjectInCache(t *testing.T, timeout time.Duration, checkFunc func()
}
}
}
+
+// drainModelServerKeys drains the workqueue and returns keys of enqueued ModelServer items.
+func drainModelServerKeys(controller *ModelServerController) []string {
+ var keys []string
+ for controller.workqueue.Len() > 0 {
+ item, _ := controller.workqueue.Get()
+ controller.workqueue.Done(item)
+ controller.workqueue.Forget(item)
+ if item.ResourceType == ResourceTypeModelServer {
+ keys = append(keys, item.Key)
+ }
+ }
+ return keys
+}
+
+// TestModelServerController_PodEnqueuesModelServer verifies that syncPodHandler
+// enqueues matching ModelServers after pod state changes.
+func TestModelServerController_PodEnqueuesModelServer(t *testing.T) {
+ kubeClient := kubefake.NewSimpleClientset()
+ kthenaClient := kthenafake.NewSimpleClientset()
+
+ ms := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "test-ms-enq"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: map[string]string{"app": "enq-test"},
+ },
+ },
+ }
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
+ kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
+ store := newStoreWithMockBackend()
+ controller, err := NewModelServerController(
+ kthenaClient, kthenaInformerFactory, kubeInformerFactory, store,
+ )
+ require.NoError(t, err)
+
+ stop := make(chan struct{})
+ defer close(stop)
+ kthenaInformerFactory.Start(stop)
+ kubeInformerFactory.Start(stop)
+ waitForCacheSync(t, 5*time.Second, controller.modelServerSynced, controller.podSynced)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, err := controller.modelServerLister.ModelServers("default").Get("test-ms-enq")
+ return err == nil
+ })
+
+ // ready pod sync enqueues its matching ModelServer
+ t.Run("ReadyPodEnqueuesMatchingModelServer", func(t *testing.T) {
+ pod := &corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "ready-pod-enq",
+ Labels: map[string]string{"app": "enq-test"},
+ },
+ Status: corev1.PodStatus{
+ Phase: corev1.PodRunning,
+ Conditions: []corev1.PodCondition{
+ {Type: corev1.PodReady, Status: corev1.ConditionTrue},
+ },
+ },
+ }
+ _, err := kubeClient.CoreV1().Pods("default").Create(
+ context.Background(), pod, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("ready-pod-enq")
+ return e == nil && p.Name == "ready-pod-enq"
+ })
+ // drain any pending items before sync
+ drainModelServerKeys(controller)
+
+ err = controller.syncPodHandler("default/ready-pod-enq")
+ assert.NoError(t, err)
+
+ keys := drainModelServerKeys(controller)
+ assert.Contains(t, keys, "default/test-ms-enq")
+ })
+
+ // not-ready pod sync deletes from store and enqueues matching ModelServer
+ t.Run("NotReadyPodEnqueuesMatchingModelServer", func(t *testing.T) {
+ pod := &corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "notready-pod-enq",
+ Labels: map[string]string{"app": "enq-test"},
+ },
+ Status: corev1.PodStatus{
+ Phase: corev1.PodPending,
+ Conditions: []corev1.PodCondition{
+ {Type: corev1.PodReady, Status: corev1.ConditionFalse},
+ },
+ },
+ }
+ _, err := kubeClient.CoreV1().Pods("default").Create(
+ context.Background(), pod, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("notready-pod-enq")
+ return e == nil && p.Name == "notready-pod-enq"
+ })
+ drainModelServerKeys(controller)
+
+ err = controller.syncPodHandler("default/notready-pod-enq")
+ assert.NoError(t, err)
+
+ keys := drainModelServerKeys(controller)
+ assert.Contains(t, keys, "default/test-ms-enq")
+ })
+
+ // deleted pod (not found in lister) enqueues all namespace ModelServers
+ t.Run("DeletedPodEnqueuesAllNamespaceModelServers", func(t *testing.T) {
+ drainModelServerKeys(controller)
+
+ err := controller.syncPodHandler("default/non-existent-pod-enq")
+ assert.NoError(t, err)
+
+ keys := drainModelServerKeys(controller)
+ assert.Contains(t, keys, "default/test-ms-enq")
+ })
+}
+
+// TestModelServerController_GetMatchingModelServers verifies the helper
+// returns correct ModelServers based on pod label matching.
+func TestModelServerController_GetMatchingModelServers(t *testing.T) {
+ kthenaClient := kthenafake.NewSimpleClientset()
+
+ ms1 := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-match"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: map[string]string{"app": "myapp"},
+ },
+ },
+ }
+ ms2 := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-nomatch"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: map[string]string{"app": "other"},
+ },
+ },
+ }
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms1, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ _, err = kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms2, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ kubeInformerFactory := informers.NewSharedInformerFactory(kubefake.NewSimpleClientset(), 0)
+ kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
+ controller, err := NewModelServerController(
+ kthenaClient, kthenaInformerFactory, kubeInformerFactory, newStoreWithMockBackend(),
+ )
+ require.NoError(t, err)
+
+ stop := make(chan struct{})
+ defer close(stop)
+ kthenaInformerFactory.Start(stop)
+ kubeInformerFactory.Start(stop)
+ waitForCacheSync(t, 5*time.Second, controller.modelServerSynced)
+
+ // pod labels match one ModelServer's workloadSelector
+ t.Run("MatchingLabelsReturnsModelServer", func(t *testing.T) {
+ pod := &corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "matching-pod",
+ Labels: map[string]string{"app": "myapp"},
+ },
+ }
+ result, err := controller.getMatchingModelServers(pod)
+ assert.NoError(t, err)
+ assert.Len(t, result, 1)
+ assert.Equal(t, "ms-match", result[0].Name)
+ })
+
+ // pod labels match no ModelServer
+ t.Run("NonMatchingLabelsReturnsEmpty", func(t *testing.T) {
+ pod := &corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: "other-pod",
+ Labels: map[string]string{"app": "nonexistent"},
+ },
+ }
+ result, err := controller.getMatchingModelServers(pod)
+ assert.NoError(t, err)
+ assert.Empty(t, result)
+ })
+}
+
+func helperPod(name string, labels map[string]string, ready bool) *corev1.Pod {
+ phase := corev1.PodPending
+ condStatus := corev1.ConditionFalse
+ if ready {
+ phase = corev1.PodRunning
+ condStatus = corev1.ConditionTrue
+ }
+ return &corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: "default",
+ Name: name,
+ Labels: labels,
+ },
+ Status: corev1.PodStatus{
+ Phase: phase,
+ Conditions: []corev1.PodCondition{
+ {Type: corev1.PodReady, Status: condStatus},
+ },
+ },
+ }
+}
+
+// TestModelServerController_ReadinessConditions verifies that the new AllPodsReady
+// condition is set alongside Ready for empty / partial / full pod readiness.
+func TestModelServerController_ReadinessConditions(t *testing.T) {
+ kthenaClient := kthenafake.NewSimpleClientset()
+ kubeClient := kubefake.NewSimpleClientset()
+ kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
+ kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
+ store := newStoreWithMockBackend()
+ controller, err := NewModelServerController(kthenaClient, kthenaInformerFactory, kubeInformerFactory, store)
+ require.NoError(t, err)
+
+ stop := make(chan struct{})
+ defer close(stop)
+ kthenaInformerFactory.Start(stop)
+ kubeInformerFactory.Start(stop)
+ waitForCacheSync(t, 5*time.Second, controller.modelServerSynced, controller.podSynced)
+
+ // zero matched pods: Ready=False, AllPodsReady=False
+ t.Run(ReasonNoMatchedPods, func(t *testing.T) {
+ noMatchLabels := map[string]string{"app": "readiness-no-match"}
+ ms := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-no-pods"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: noMatchLabels,
+ },
+ },
+ }
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, e := controller.modelServerLister.ModelServers("default").Get("ms-no-pods")
+ return e == nil
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-no-pods"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-no-pods", metav1.GetOptions{})
+ require.NoError(t, err)
+ assert.Equal(t, int32(0), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(0), updated.Status.ReadyReplicas)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionFalse, readyCond.Status)
+ assert.Equal(t, ReasonNoReadyPods, readyCond.Reason)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionFalse, allReadyCond.Status)
+ assert.Equal(t, ReasonNoMatchedPods, allReadyCond.Reason)
+ })
+
+ // 1/2 pods ready: Ready=True, AllPodsReady=False
+ t.Run("PartialReadiness", func(t *testing.T) {
+ partialLabels := map[string]string{"app": "readiness-partial"}
+ ms := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-partial"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: partialLabels,
+ },
+ },
+ }
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, e := controller.modelServerLister.ModelServers("default").Get("ms-partial")
+ return e == nil
+ })
+
+ // Create 1 ready + 1 not-ready pod
+ readyPod := helperPod("ready-a", partialLabels, true)
+ notReadyPod := helperPod("notready-a", partialLabels, false)
+ _, _ = kubeClient.CoreV1().Pods("default").Create(context.Background(), readyPod, metav1.CreateOptions{})
+ _, _ = kubeClient.CoreV1().Pods("default").Create(context.Background(), notReadyPod, metav1.CreateOptions{})
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("notready-a")
+ return e == nil && p.Name == "notready-a"
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-partial"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-partial", metav1.GetOptions{})
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(1), updated.Status.ReadyReplicas)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionTrue, readyCond.Status)
+ assert.Equal(t, ReasonPodsReady, readyCond.Reason)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionFalse, allReadyCond.Status)
+ assert.Equal(t, ReasonPodsNotFullyReady, allReadyCond.Reason)
+ })
+
+ // 2/2 pods ready: Ready=True, AllPodsReady=True
+ t.Run("FullReadiness", func(t *testing.T) {
+ fullLabels := map[string]string{"app": "readiness-full"}
+ ms := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-full"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: fullLabels,
+ },
+ },
+ }
+ _, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, e := controller.modelServerLister.ModelServers("default").Get("ms-full")
+ return e == nil
+ })
+
+ pod1 := helperPod("ready-full-1", fullLabels, true)
+ pod2 := helperPod("ready-full-2", fullLabels, true)
+ _, _ = kubeClient.CoreV1().Pods("default").Create(context.Background(), pod1, metav1.CreateOptions{})
+ _, _ = kubeClient.CoreV1().Pods("default").Create(context.Background(), pod2, metav1.CreateOptions{})
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("ready-full-2")
+ return e == nil && p.Name == "ready-full-2"
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-full"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-full", metav1.GetOptions{})
+ require.NoError(t, err)
+ assert.Equal(t, int32(2), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(2), updated.Status.ReadyReplicas)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionTrue, readyCond.Status)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionTrue, allReadyCond.Status)
+ assert.Equal(t, ReasonAllPodsReady, allReadyCond.Reason)
+ })
+}
+
+// TestModelServerController_ReadinessConditionTransitions verifies condition
+// values across a lifecycle: 0 pods → 1/1 ready → 1/2 ready → 2/2 ready.
+func TestModelServerController_ReadinessConditionTransitions(t *testing.T) {
+ kthenaClient := kthenafake.NewSimpleClientset()
+ kubeClient := kubefake.NewSimpleClientset()
+ kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
+ kthenaInformerFactory := informersv1alpha1.NewSharedInformerFactory(kthenaClient, 0)
+ store := newStoreWithMockBackend()
+ controller, err := NewModelServerController(kthenaClient, kthenaInformerFactory, kubeInformerFactory, store)
+ require.NoError(t, err)
+
+ stop := make(chan struct{})
+ defer close(stop)
+ kthenaInformerFactory.Start(stop)
+ kubeInformerFactory.Start(stop)
+ waitForCacheSync(t, 5*time.Second, controller.modelServerSynced, controller.podSynced)
+
+ tLabels := map[string]string{"app": "readiness-trans"}
+
+ ms := &aiv1alpha1.ModelServer{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ms-trans"},
+ Spec: aiv1alpha1.ModelServerSpec{
+ InferenceEngine: aiv1alpha1.VLLM,
+ WorkloadSelector: &aiv1alpha1.WorkloadSelector{
+ MatchLabels: tLabels,
+ },
+ },
+ }
+ _, err = kthenaClient.NetworkingV1alpha1().ModelServers("default").Create(
+ context.Background(), ms, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ _, e := controller.modelServerLister.ModelServers("default").Get("ms-trans")
+ return e == nil
+ })
+
+ // Step 1: no matching pods
+ t.Run("Step1_NoMatchingPods", func(t *testing.T) {
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-trans"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-trans", metav1.GetOptions{})
+ require.NoError(t, err)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionFalse, readyCond.Status)
+ assert.Equal(t, ReasonNoReadyPods, readyCond.Reason)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionFalse, allReadyCond.Status)
+ assert.Equal(t, ReasonNoMatchedPods, allReadyCond.Reason)
+ assert.Equal(t, int32(0), updated.Status.MatchedReplicas)
+ })
+
+ // Step 2: add one ready pod → Ready=True, AllPodsReady=True (only 1 matched)
+ t.Run("Step2_AllPodsReadyWithOnePod", func(t *testing.T) {
+ readyPod := helperPod("trans-pod-1", tLabels, true)
+ _, err := kubeClient.CoreV1().Pods("default").Create(context.Background(), readyPod, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("trans-pod-1")
+ return e == nil && p.Name == "trans-pod-1"
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-trans"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-trans", metav1.GetOptions{})
+ require.NoError(t, err)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionTrue, readyCond.Status)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionTrue, allReadyCond.Status)
+ assert.Equal(t, ReasonAllPodsReady, allReadyCond.Reason)
+
+ assert.Equal(t, int32(1), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(1), updated.Status.ReadyReplicas)
+ })
+
+ // Step 3: add one not-ready pod → Ready=True, AllPodsReady=False (1/2)
+ t.Run("Step3_PartialAfterNotReady", func(t *testing.T) {
+ notReadyPod := helperPod("trans-pod-2", tLabels, false)
+ _, err := kubeClient.CoreV1().Pods("default").Create(context.Background(), notReadyPod, metav1.CreateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("trans-pod-2")
+ return e == nil && p.Name == "trans-pod-2"
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-trans"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-trans", metav1.GetOptions{})
+ require.NoError(t, err)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionTrue, readyCond.Status)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionFalse, allReadyCond.Status)
+ assert.Equal(t, ReasonPodsNotFullyReady, allReadyCond.Reason)
+
+ assert.Equal(t, int32(2), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(1), updated.Status.ReadyReplicas)
+ })
+
+ // Step 4: make the second pod ready → Ready=True, AllPodsReady=True (2/2)
+ t.Run("Step4_FullReadyAfterTransition", func(t *testing.T) {
+ // Update not-ready pod to ready
+ notReadyPod, err := kubeClient.CoreV1().Pods("default").Get(context.Background(), "trans-pod-2", metav1.GetOptions{})
+ require.NoError(t, err)
+ updatedPod := notReadyPod.DeepCopy()
+ updatedPod.Status.Phase = corev1.PodRunning
+ updatedPod.Status.Conditions[0].Status = corev1.ConditionTrue
+ _, err = kubeClient.CoreV1().Pods("default").Update(context.Background(), updatedPod, metav1.UpdateOptions{})
+ require.NoError(t, err)
+ waitForObjectInCache(t, 2*time.Second, func() bool {
+ p, e := controller.podLister.Pods("default").Get("trans-pod-2")
+ return e == nil && isPodReady(p)
+ })
+
+ assert.NoError(t, controller.syncModelServerHandler("default/ms-trans"))
+
+ updated, err := kthenaClient.NetworkingV1alpha1().ModelServers("default").Get(
+ context.Background(), "ms-trans", metav1.GetOptions{})
+ require.NoError(t, err)
+
+ readyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionReady))
+ require.NotNil(t, readyCond)
+ assert.Equal(t, metav1.ConditionTrue, readyCond.Status)
+
+ allReadyCond := meta.FindStatusCondition(updated.Status.Conditions, string(aiv1alpha1.ModelServerConditionAllPodsReady))
+ require.NotNil(t, allReadyCond)
+ assert.Equal(t, metav1.ConditionTrue, allReadyCond.Status)
+ assert.Equal(t, ReasonAllPodsReady, allReadyCond.Reason)
+
+ assert.Equal(t, int32(2), updated.Status.MatchedReplicas)
+ assert.Equal(t, int32(2), updated.Status.ReadyReplicas)
+ })
+}