Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions api/orchestration/v1alpha1/stormservice_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,14 @@ type StormServiceSpec struct {
// +optional
Paused bool `json:"paused,omitempty" protobuf:"varint,7,opt,name=paused"`

// The maximum time in seconds for a StormService to make progress before it
// is considered to be failed. The controller will continue to process failed
// StormServices and surface a condition with a ProgressDeadlineExceeded reason.
// Progress is not estimated while a StormService is paused. Defaults to 600s.
// +kubebuilder:default=600

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

progressDeadlineSeconds should be validated as a positive value. Right now the CRD/API only sets default: 600, but does not add a minimum, so users can create a StormService with progressDeadlineSeconds: 0 or a negative value. can we valid in webhook ?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks — addressed in 3b3b5f7. I added structural CRD validation with +kubebuilder:validation:Minimum=1, so both zero and negative values are rejected by the API server on create and update. This is enforced independently of the custom webhook availability or its failure policy. I also regenerated the module and Helm CRDs and added integration cases verifying that 0 and -1 return Kubernetes Invalid errors.

// +optional
ProgressDeadlineSeconds *int32 `json:"progressDeadlineSeconds,omitempty" protobuf:"varint,9,opt,name=progressDeadlineSeconds"`
Comment on lines +69 to +72

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

ProgressDeadlineSeconds currently lacks a minimum validation marker. A negative value would cause the controller to immediately time out the rollout. We should add +kubebuilder:validation:Minimum=0 to prevent invalid negative values.

Suggested change
// +kubebuilder:default=600
// +optional
ProgressDeadlineSeconds *int32 `json:"progressDeadlineSeconds,omitempty" protobuf:"varint,9,opt,name=progressDeadlineSeconds"`
// +kubebuilder:default=600
// +kubebuilder:validation:Minimum=0
// +optional
ProgressDeadlineSeconds *int32 `json:"progressDeadlineSeconds,omitempty" protobuf:"varint,9,opt,name=progressDeadlineSeconds"`


// DisruptionTolerance indicates how many roleSets can be unavailable during the preemption/eviction.
// +optional
DisruptionTolerance DisruptionTolerance `json:"disruptionTolerance,omitempty"`
Expand Down
5 changes: 5 additions & 0 deletions api/orchestration/v1alpha1/zz_generated.deepcopy.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,10 @@ spec:
type: object
paused:
type: boolean
progressDeadlineSeconds:
default: 600
format: int32
type: integer
replicas:
format: int32
type: integer
Expand Down
4 changes: 4 additions & 0 deletions dist/chart/crds/orchestration.aibrix.ai_stormservices.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,10 @@ spec:
type: object
paused:
type: boolean
progressDeadlineSeconds:
default: 600
format: int32
type: integer
replicas:
format: int32
type: integer
Expand Down

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

144 changes: 144 additions & 0 deletions pkg/controller/stormservice/progress.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
/*
Copyright 2026 The Aibrix Team.

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.
*/

package stormservice

import (
"fmt"
"time"

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

orchestrationv1alpha1 "github.com/vllm-project/aibrix/api/orchestration/v1alpha1"
utils "github.com/vllm-project/aibrix/pkg/controller/util/orchestration"
)

const (
defaultProgressDeadlineSeconds int32 = 600

ProgressingReason = "Processing"
ProgressDeadlineExceededReason = "ProgressDeadlineExceeded"
PausedReason = "DeploymentPaused"
ResumedReason = "DeploymentResumed"
Comment on lines +33 to +36
)

func newProgressingCondition(status corev1.ConditionStatus, reason, message string, now time.Time) orchestrationv1alpha1.Condition {
timestamp := metav1.NewTime(now)
return orchestrationv1alpha1.Condition{
Type: orchestrationv1alpha1.StormServiceProgressing,
Status: status,
LastUpdateTime: &timestamp,
LastTransitionTime: &timestamp,
Reason: reason,
Message: message,
}
}

func stormServiceProgressing(oldStatus, newStatus *orchestrationv1alpha1.StormServiceStatus) bool {
oldReplicas := oldStatus.Replicas - oldStatus.UpdatedReplicas
newReplicas := newStatus.Replicas - newStatus.UpdatedReplicas
return newStatus.UpdatedReplicas > oldStatus.UpdatedReplicas ||
newReplicas < oldReplicas ||
newStatus.ReadyReplicas > oldStatus.ReadyReplicas ||
newStatus.UpdatedReadyReplicas > oldStatus.UpdatedReadyReplicas
}

func progressDeadline(stormService *orchestrationv1alpha1.StormService) time.Duration {
seconds := defaultProgressDeadlineSeconds
if stormService.Spec.ProgressDeadlineSeconds != nil {
seconds = *stormService.Spec.ProgressDeadlineSeconds
}
return time.Duration(seconds) * time.Second
}

func syncStormServiceProgressingCondition(stormService *orchestrationv1alpha1.StormService, oldStatus *orchestrationv1alpha1.StormServiceStatus, now time.Time) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you help double-check whether replacing the whole Status.Conditions slice here can cause any issue?

In syncStormServiceProgressingCondition, we assign stormService.Status.Conditions = orchestrationv1alpha1.Conditions{condition}. I noticed StormService defines multiple
condition types (Ready, Progressing, ReplicaFailure), and there is also a SetStormServiceCondition helper that preserves other condition types while updating one condition.

Could you verify whether the current StormService status model intentionally keeps only one active condition at a time, or whether this path might accidentally drop another condition that should be preserved?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for flagging this. I traced every current StormService Status.Conditions writer. Although the Ready, Progressing, and ReplicaFailure paths currently happen to replace the slice, the API defines multiple condition types and does not document a single-active-condition contract, so replacing the whole slice here can discard an unrelated condition.

I reproduced that risk on the previous head: a focused regression test failed because updating Progressing removed an existing Ready condition. Commit 92687e7 fixes this by filtering out only existing Progressing entries and appending the newly computed Progressing condition, preserving all other condition types unchanged and collapsing malformed duplicate Progressing entries.

I did not call SetStormServiceCondition directly because its same-status/same-reason early return would suppress the LastUpdateTime refresh required when rollout progress is observed. After merging current main, the focused regression, StormService unit and race tests, controller envtest (47/47), webhook envtest (66/66), lint, codegen, manifests, and CRD sync all pass at ae4a360.

currentCondition := utils.GetCondition(oldStatus.Conditions, orchestrationv1alpha1.StormServiceProgressing)
var condition orchestrationv1alpha1.Condition

Comment on lines +68 to +71
switch {
case stormService.Spec.Paused:
if currentCondition != nil && currentCondition.Reason == ProgressDeadlineExceededReason {
condition = *currentCondition
} else if currentCondition != nil && currentCondition.Reason == PausedReason {
condition = *currentCondition
} else {
condition = newProgressingCondition(
corev1.ConditionUnknown,
PausedReason,
fmt.Sprintf("StormService %q is paused.", stormService.Name),
now,
)
}
Comment on lines +73 to +85

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

When stormService.Spec.Paused is true, the current implementation preserves the ProgressDeadlineExceededReason condition if it was already set. This deviates from standard Kubernetes behavior and prevents a timed-out rollout from being recovered via a pause/resume cycle.\n\nIn Kubernetes, pausing a timed-out deployment transitions its progressing condition to DeploymentPaused (ConditionUnknown). When resumed, it transitions to DeploymentResumed (ConditionTrue), which resets the progress deadline and allows the rollout to attempt recovery.\n\nWe should transition to PausedReason when paused, regardless of whether the previous state was timed out.

	case stormService.Spec.Paused:
		if currentCondition != nil && currentCondition.Reason == PausedReason {
			condition = *currentCondition
		} else {
			condition = newProgressingCondition(
				corev1.ConditionUnknown,
			PausedReason,
			fmt.Sprintf("StormService %q is paused.", stormService.Name),
			now,
		)
		}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz check this too

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for asking me to verify this as well. I checked Kubernetes v1.31.8 checkPausedConditions directly. It intentionally returns without replacing an existing ProgressDeadlineExceeded condition when a timed-out Deployment is paused. For a non-timed-out rollout, both DeploymentPaused and DeploymentResumed use ConditionUnknown; resume refreshes LastUpdateTime, while LastTransitionTime remains unchanged because the status stays Unknown.

The current StormService implementation follows those semantics, including excluding paused time from the deadline, so I do not recommend changing this path. Kubernetes reference: https://github.com/kubernetes/kubernetes/blob/v1.31.8/pkg/controller/deployment/sync.go#L75-L94

case currentCondition != nil && currentCondition.Reason == PausedReason:
condition = newProgressingCondition(
corev1.ConditionUnknown,
ResumedReason,
fmt.Sprintf("StormService %q is resumed.", stormService.Name),
now,
)
Comment on lines +87 to +92
condition.LastTransitionTime = currentCondition.LastTransitionTime
Comment on lines +86 to +93

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

When transitioning from PausedReason to ResumedReason, the condition status is set to ConditionUnknown and LastTransitionTime is overwritten with currentCondition.LastTransitionTime (the time when the service was paused).\n\nThis is incorrect for two reasons:\n1. Resuming a rollout means it is actively attempting to progress, so the status should be ConditionTrue (matching Kubernetes DeploymentResumed behavior).\n2. The transition to Resumed occurs at now (the resume time), so LastTransitionTime must be updated to now instead of keeping the old pause timestamp.

Suggested change
case currentCondition != nil && currentCondition.Reason == PausedReason:
condition = newProgressingCondition(
corev1.ConditionUnknown,
ResumedReason,
fmt.Sprintf("StormService %q is resumed.", stormService.Name),
now,
)
condition.LastTransitionTime = currentCondition.LastTransitionTime
case currentCondition != nil && currentCondition.Reason == PausedReason:
condition = newProgressingCondition(
corev1.ConditionTrue,
ResumedReason,
fmt.Sprintf("StormService %q is resumed.", stormService.Name),
now,
)

case stormServiceProgressing(oldStatus, &stormService.Status):
condition = newProgressingCondition(
corev1.ConditionTrue,
ProgressingReason,
fmt.Sprintf("StormService %q is progressing.", stormService.Name),
now,
)
if currentCondition != nil && currentCondition.Status == corev1.ConditionTrue {
condition.LastTransitionTime = currentCondition.LastTransitionTime
}
case currentCondition == nil || currentCondition.LastUpdateTime == nil:
condition = newProgressingCondition(
corev1.ConditionTrue,
ProgressingReason,
fmt.Sprintf("StormService %q is progressing.", stormService.Name),
now,
)
case currentCondition.Reason == ProgressDeadlineExceededReason:
condition = *currentCondition
case !currentCondition.LastUpdateTime.Add(progressDeadline(stormService)).After(now):
condition = newProgressingCondition(
corev1.ConditionFalse,
ProgressDeadlineExceededReason,
fmt.Sprintf("StormService %q has timed out progressing.", stormService.Name),
now,
)
default:
condition = *currentCondition
}

stormService.Status.Conditions = orchestrationv1alpha1.Conditions{condition}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The controller currently overwrites the entire stormService.Status.Conditions slice with a single condition. This violates Kubernetes API conventions where multiple conditions (such as Ready and Progressing) should coexist to provide a complete view of the resource's status.\n\nWe should update or append the progressing condition in-place within the slice instead of overwriting it entirely.

Suggested change
stormService.Status.Conditions = orchestrationv1alpha1.Conditions{condition}
found := false
for i, cond := range stormService.Status.Conditions {
if cond.Type == orchestrationv1alpha1.StormServiceProgressing {
stormService.Status.Conditions[i] = condition
found = true
break
}
}
if !found {
stormService.Status.Conditions = append(stormService.Status.Conditions, condition)
}

}

func progressDeadlineRequeueAfter(stormService *orchestrationv1alpha1.StormService, now time.Time) time.Duration {
if stormService.Spec.Paused {
return DefaultRequeueAfter
}
condition := utils.GetCondition(stormService.Status.Conditions, orchestrationv1alpha1.StormServiceProgressing)
if condition == nil || condition.LastUpdateTime == nil || condition.Reason == ProgressDeadlineExceededReason {
return DefaultRequeueAfter
}

requeueAfter := condition.LastUpdateTime.Add(progressDeadline(stormService)).Sub(now) + time.Second
if requeueAfter < time.Second {
return time.Second
}
if requeueAfter < DefaultRequeueAfter {
return requeueAfter
}
return DefaultRequeueAfter
}
Loading
Loading