Repository: vllm-project/aibrix
File: pkg/controller/modeladapter/modeladapter_controller.go (Go)

Code before:

	// Step 2: Reconcile Loading
	if err := r.reconcileLoading(ctx, instance, selectedPod); err != nil {
		// retry any of the failure.
		instance.Status.Phase = modelv1alpha1.ModelAdapterFailed
		if err := r.Status().Update(ctx, instance); err != nil {
			klog.InfoS("Got error when updating status", "cluster name", req.Name, "error", err, "ModelAdapter", instance)
			return ctrl.Result{RequeueAfter: defaultRequeueDuration}, err
		}

		return ctrl.Result{RequeueAfter: defaultRequeueDuration}, err
	}

	// Step 3: Reconcile Service
	if ctrlResult, err := r.reconcileService(ctx, instance); err != nil {
		if updateErr := r.updateModelAdapterState(ctx, instance, modelv1alpha1.ModelAdapterFailed); updateErr != nil {
			klog.ErrorS(updateErr, "ModelAdapter update state error", "cluster name", req.Name)
		}
		return ctrlResult, err
	}

	// Step 4: Reconcile EndpointSlice
	if ctrlResult, err := r.reconcileEndpointSlice(ctx, instance, selectedPod); err != nil {
		if updateErr := r.updateModelAdapterState(ctx, instance, modelv1alpha1.ModelAdapterFailed); updateErr != nil {
			klog.ErrorS(updateErr, "ModelAdapter update state error", "cluster name", req.Name)
		}
		return ctrlResult, err
	}

	// Check if need to update the status.
	if r.inconsistentModelAdapterStatus(oldInstance.Status, instance.Status) {
		klog.InfoS("model adapter reconcile", "Update CR status", req.Name, "status", instance.Status)
		instance.Status.Phase = modelv1alpha1.ModelAdapterRunning
		if err = r.updateStatus(ctx, instance); err != nil {
			return reconcile.Result{}, fmt.Errorf("update modelAdapter status error: %v", err)
		}
	}

	return ctrl.Result{}, nil
}

func (r *ModelAdapterReconciler) updateStatus(ctx context.Context, instance *modelv1alpha1.ModelAdapter) error {
	meta.SetStatusCondition(&instance.Status.Conditions, metav1.Condition{
		Type:               string(modelv1alpha1.ModelAdapterConditionReady),
		Status:             metav1.ConditionTrue,
		Reason:             "Reconciling",
		Message:            fmt.Sprintf("ModelAdapter %s is ready", klog.KObj(instance)),
		LastTransitionTime: metav1.Now(),
	})

	return r.Status().Update(ctx, instance)

Review comment (bug):
should we have wait period or retry count. Because right now if pod going thru rolling restart, then adapters will crash loop with them.

For example, in rolling restart for pod A, all adapters are scheduled on pod B, then pod B restarts and adapters move to pod C. 

Code after:
		// check the scheduled pod first, verify the mapping is still valid.
		// TODO: this needs to be changed once we support multiple lora adapters
		selectedPodName := instance.Status.Instances[0]
		if err := r.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: selectedPodName}, selectedPod); err != nil && apierrors.IsNotFound(err) {
			klog.ErrorS(err, "Failed to get selected pod but pod is still in ModelAdapter instance list", "modelAdapter", klog.KObj(instance))

			// instance.Status.Instances has been outdated, and we need to clear the pod list
			// after the pod list is cleaned up, let's reconcile the instance object again in the next loop
			return ctrl.Result{}, r.clearModelAdapterInstanceList(ctx, instance)
		} else if err != nil {
			// failed to fetch the pod, let's requeue
			return ctrl.Result{RequeueAfter: defaultRequeueDuration}, err
		} else {
			// compare instance and model adapter labels.
			selector, err := metav1.LabelSelectorAsSelector(instance.Spec.PodSelector)
			if err != nil {
				// TODO: this should barely happen, let's move this logic to earlier validation logics.
				return ctrl.Result{}, fmt.Errorf("failed to convert pod selector: %v", err)
			}

			if !selector.Matches(labels.Set(selectedPod.Labels)) {
				klog.Warning("current assigned pod selector doesn't match model adapter selector")
				return ctrl.Result{}, r.clearModelAdapterInstanceList(ctx, instance)
			}

			if !utils.IsPodReady(selectedPod) || utils.IsPodTerminating(selectedPod) {
				klog.Warning(fmt.Sprintf("current assigned pod %s/%s is not ready, let's clean it up and reschedule the adapter", selectedPod.Namespace, selectedPod.Name))
				// continue to requeue the object to remove endpoint etc in current loop.
				if err = r.clearModelAdapterInstanceList(ctx, instance); err != nil {
					return ctrl.Result{}, err
				}
			}

			existPods = true
		}
	}

	if !existPods {
		// TODO: as we plan to support lora replicas, it needs some corresponding changes.
		// it should return a list of pods in future, otherwise, it should be invoked by N times.
		selectedPod, err = r.schedulePod(ctx, instance)
		if err != nil {
			klog.ErrorS(err, "Failed to schedule Pod for ModelAdapter", "modelAdapter", instance.Name)
			return ctrl.Result{}, err
		}
		if selectedPod != nil {
			instance.Status.Phase = modelv1alpha1.ModelAdapterScheduling
			instance.Status.Instances = []string{selectedPod.Name}
			meta.SetStatusCondition(&instance.Status.Conditions, metav1.Condition{
				Type:               string(modelv1alpha1.ModelAdapterConditionTypeSelectorMatched),
				Status:             metav1.ConditionTrue,