ronantakizawa/codereview-bench huggingface dataset:
Repository: vllm-project/aibrix
File: pkg/utils/pod.go (Go)

Code before:
	}
	return podList, nil
}

func CountReadyPods(podList *v1.PodList) (int64, error) {
	if podList == nil || len(podList.Items) == 0 {
		return 0, nil
	}

	readyPodCount := 0
	for _, pod := range podList.Items {
		isReady := IsPodReady(&pod)
		if pod.Status.Phase == v1.PodRunning && isReady {
			readyPodCount++
		}
		klog.V(4).InfoS("CountReadyPods Pod status", "name", pod.Name, "phase", pod.Status.Phase, "ready", isReady)
	}

	return int64(readyPodCount), nil
}

// FilterReadyPods filters and returns a list of pods that have a valid PodIP.
func FilterReadyPods(pods map[string]*v1.Pod) []*v1.Pod {
	var readyPods []*v1.Pod
	for _, pod := range pods {
		if pod.Status.PodIP != "" && IsPodReady(pod) {
			readyPods = append(readyPods, pod)
		}
	}
	return readyPods
}


Review comment (performance):
is this method being used by other files? like some other controllers other than cache.go? I am a little bit concerned on the impact if we have other scenarios. 

the key difference of this change is `terminated` pod?


Code after:
	}
	return podList, nil
}

func CountReadyPods(podList *v1.PodList) (int64, error) {
	if podList == nil || len(podList.Items) == 0 {
		return 0, nil
	}

	readyPodCount := 0
	for _, pod := range podList.Items {
		isReady := IsPodReady(&pod)
		if pod.Status.Phase == v1.PodRunning && isReady {
			readyPodCount++
		}
		klog.V(4).InfoS("CountReadyPods Pod status", "name", pod.Name, "phase", pod.Status.Phase, "ready", isReady)
	}

	return int64(readyPodCount), nil
}

// FilterReadyPods filters and returns a list of pods that have a valid PodIP.
func FilterReadyPods(pods map[string]*v1.Pod) []*v1.Pod {
	var readyPods []*v1.Pod
	for _, pod := range pods {
		if pod.Status.PodIP == "" || IsPodTerminating(pod) || !IsPodReady(pod) {
			continue
		}

		readyPods = append(readyPods, pod)

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

Code before:
		if !ok {
			continue
		}
		if model["id"] == instance.Name {
			return true, nil
		}
	}

	return false, nil
}

// Separate method to load the LoRA adapter
func (r *ModelAdapterReconciler) loadModelAdapter(host string, instance *modelv1alpha1.ModelAdapter) error {
	artifactURL := instance.Spec.ArtifactURL
	if strings.HasPrefix(instance.Spec.ArtifactURL, "huggingface://") {
		artifactURL, err := extractHuggingFacePath(instance.Spec.ArtifactURL)
		if err != nil {
			// Handle error, e.g., log it and return
			klog.ErrorS(err, "Invalid artifact URL", "artifactURL", artifactURL)
			return err
		}
	}
	// TODO: extend to other artifacts

	payload := map[string]string{
		"lora_name": instance.Name,
		"lora_path": artifactURL,
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	url := fmt.Sprintf("%s/v1/load_lora_adapter", host)
	req, err := http.NewRequest("POST", url, bytes.NewBuffer(payloadBytes))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	// Check if "api-key" exists in the map and set the Authorization header accordingly
	if token, ok := instance.Spec.AdditionalConfig["api-key"]; ok {
		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
	}

	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer func() {
		if err := resp.Body.Close(); err != nil {

Review comment (suggestion):
We can communicate later on how to also support the model on TOS. At present, the `/v1/model/download` interface provided on the runtime should come in handy. Let's create an issue to discuss about it

Code after:
	var response map[string]interface{}
	if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
		return false, err
	}

	data, ok := response["data"].([]interface{})
	if !ok {
		return false, errors.New("invalid data format")
	}

	for _, item := range data {
		model, ok := item.(map[string]interface{})
		if !ok {
			continue
		}
		if model["id"] == instance.Name {
			return true, nil
		}
	}

	return false, nil
}

// Separate method to load the LoRA adapter
func (r *ModelAdapterReconciler) loadModelAdapter(url string, instance *modelv1alpha1.ModelAdapter) error {
	artifactURL := instance.Spec.ArtifactURL
	if strings.HasPrefix(instance.Spec.ArtifactURL, "huggingface://") {
		var err error
		artifactURL, err = extractHuggingFacePath(instance.Spec.ArtifactURL)
		if err != nil {
			// Handle error, e.g., log it and return
			klog.ErrorS(err, "Invalid artifact URL", "artifactURL", artifactURL)
			return err
		}
	}
	// TODO: extend to other artifacts

	payload := map[string]string{
		"lora_name": instance.Name,
		"lora_path": artifactURL,
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", url, bytes.NewBuffer(payloadBytes))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
Repository: vllm-project/aibrix
File: test/e2e/util.go (Go)

Code before:
	)
}

func validateInference(t *testing.T, modelName string) {
	client := createOpenAIClient(gatewayURL, apiKey)
	validateInferenceWithClient(t, client, modelName)
}

func validateInferenceWithClient(t *testing.T, client *openai.Client, modelName string) {
	chatCompletion, err := client.Chat.Completions.New(context.TODO(), openai.ChatCompletionNewParams{
		Messages: openai.F([]openai.ChatCompletionMessageParamUnion{
			openai.UserMessage("Say this is a test"),
		}),
		Model: openai.F(modelName),
	})
	if err != nil {
		t.Fatalf("chat completions failed : %v", err)
	}
	assert.Equal(t, modelName, chatCompletion.Model)
	assert.NotEmpty(t, chatCompletion.Choices, "chat completion has no choices returned")
	assert.NotNil(t, chatCompletion.Choices[0].Message.Content, "chat completion has no message returned")
}

func validateAllPodsAreReady(t *testing.T, client *kubernetes.Clientset, expectedPodCount int) {
	allPodsReady := false
	for i := 0; i < expectedPodCount; i++ {
		podlist, err := client.CoreV1().Pods("default").List(context.Background(), v1.ListOptions{})
		if err != nil {
			continue
		}
		assert.Equal(t, expectedPodCount, len(utils.FilterActivePods(podlist.Items)))
		fmt.Println("all pods are ready", len(utils.FilterActivePods(podlist.Items)))
		allPodsReady = true
		break
	}
	assert.True(t, allPodsReady, "ensure all pods are ready")
}


Review comment (suggestion):
 We set `expectedPodCount` to be the actual expected number, and using it for traversal seems to be OK, but I'm not sure if it will be confusing? 🤔 

Code after:
			r.URL.Path = "/v1" + r.URL.Path
			return mn(r)
		}),
		option.WithHeader("routing-strategy", routingStrategy),
		option.WithMaxRetries(0),
		respOpt,
	)
}

func validateInference(t *testing.T, modelName string) {
	client := createOpenAIClient(gatewayURL, apiKey)
	validateInferenceWithClient(t, client, modelName)
}

func validateInferenceWithClient(t *testing.T, client *openai.Client, modelName string) {
	chatCompletion, err := client.Chat.Completions.New(context.TODO(), openai.ChatCompletionNewParams{
		Messages: openai.F([]openai.ChatCompletionMessageParamUnion{
			openai.UserMessage("Say this is a test"),
		}),
		Model: openai.F(modelName),
	})
	if err != nil {
		t.Fatalf("chat completions failed : %v", err)
	}
	assert.Equal(t, modelName, chatCompletion.Model)
	assert.NotEmpty(t, chatCompletion.Choices, "chat completion has no choices returned")
	assert.NotNil(t, chatCompletion.Choices[0].Message.Content, "chat completion has no message returned")
}

func validateAllPodsAreReady(t *testing.T, client *kubernetes.Clientset, expectedPodCount int) {
	allPodsReady := false
	for i := 0; i <= 3; i++ {
		podlist, err := client.CoreV1().Pods("default").List(context.Background(), v1.ListOptions{})
		if err == nil && expectedPodCount == len(utils.FilterActivePods(podlist.Items)) {
			allPodsReady = true
			t.Log("all pods are ready", len(utils.FilterActivePods(podlist.Items)))
			break
		}
		time.Sleep(1 * time.Second)
	}
	assert.True(t, allPodsReady, "ensure all pods are ready")
}
Repository: vllm-project/aibrix
File: pkg/cache/informers.go (Go)

Code before:
	nodeType        = "ray.io/node-type"
	nodeWorker      = "worker"
)

func initCacheInformers(instance *Store, config *rest.Config, stopCh <-chan struct{}) error {
	if err := v1alpha1scheme.AddToScheme(scheme.Scheme); err != nil {
		return err
	}

	k8sClientSet, err := kubernetes.NewForConfig(config)
	if err != nil {
		return err
	}

	crdClientSet, err := v1alpha1.NewForConfig(config)
	if err != nil {
		return err
	}

	factory := informers.NewSharedInformerFactoryWithOptions(k8sClientSet, 0)
	crdFactory := crdinformers.NewSharedInformerFactoryWithOptions(crdClientSet, 0)

	podInformer := factory.Core().V1().Pods().Informer()
	modelInformer := crdFactory.Model().V1alpha1().ModelAdapters().Informer()

	defer runtime.HandleCrash()
	factory.Start(stopCh)
	crdFactory.Start(stopCh)

	if !cache.WaitForCacheSync(stopCh, podInformer.HasSynced, modelInformer.HasSynced) {
		return errors.New("timed out waiting for caches to sync")
	}

	if _, err := podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    instance.addPod,
		UpdateFunc: instance.updatePod,
		DeleteFunc: instance.deletePod,
	}); err != nil {
		return err
	}

	if _, err = modelInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    instance.addModelAdapter,
		UpdateFunc: instance.updateModelAdapter,
		DeleteFunc: instance.deleteModelAdapter,
	}); err != nil {
		return err
	}

	return nil
}

Review comment (bug):
HandleCrash is usually used in goroutinue or in functions that contain functions that are not sure whether a panic will occur or not.
it doesn't seem necessary in here.


Code after:
	nodeWorker      = "worker"
)

func initCacheInformers(instance *Store, config *rest.Config, stopCh <-chan struct{}) error {
	if err := v1alpha1scheme.AddToScheme(scheme.Scheme); err != nil {
		return err
	}

	k8sClientSet, err := kubernetes.NewForConfig(config)
	if err != nil {
		return err
	}

	crdClientSet, err := v1alpha1.NewForConfig(config)
	if err != nil {
		return err
	}

	factory := informers.NewSharedInformerFactoryWithOptions(k8sClientSet, 0)
	crdFactory := crdinformers.NewSharedInformerFactoryWithOptions(crdClientSet, 0)

	podInformer := factory.Core().V1().Pods().Informer()
	if _, err := podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    instance.addPod,
		UpdateFunc: instance.updatePod,
		DeleteFunc: instance.deletePod,
	}); err != nil {
		return err
	}

	modelInformer := crdFactory.Model().V1alpha1().ModelAdapters().Informer()
	if _, err = modelInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    instance.addModelAdapter,
		UpdateFunc: instance.updateModelAdapter,
		DeleteFunc: instance.deleteModelAdapter,
	}); err != nil {
		return err
	}

	factory.Start(stopCh)
	crdFactory.Start(stopCh)

	if !cache.WaitForCacheSync(stopCh, podInformer.HasSynced, modelInformer.HasSynced) {
		return errors.New("timed out waiting for caches to sync")
	}
	return nil
}

func (c *Store) addPod(obj interface{}) {
	pod := obj.(*v1.Pod)
	// only track pods with model deployments
Repository: vllm-project/aibrix
File: pkg/plugins/gateway/algorithms/util.go (Go)

Code before:
// mean calculates the mean of a slice of float64 numbers.
func mean(numbers []float64) float64 {
	sum := 0.0
	for _, number := range numbers {
		sum += number
	}
	return sum / float64(len(numbers))
}

// standardDeviation calculates the standard deviation of a slice of float64 numbers.
func standardDeviation(numbers []float64) float64 {
	avg := mean(numbers)
	sumOfSquares := 0.0
	for _, number := range numbers {
		sumOfSquares += math.Pow(number-avg, 2)
	}
	variance := sumOfSquares / float64(len(numbers)-1)
	return math.Sqrt(variance)
}

// SelectRandomPodAsFallback selects a pod randomly as a fallback.
// This method should only be used when all other selection mechanisms have failed.
// For example, if no pods meet the required criteria (e.g., valid metrics or specific conditions),
// this method can be called to randomly select a pod from the provided list.
func SelectRandomPodAsFallback(ctx *types.RoutingContext, pods []*v1.Pod, randomFunc func(int) int) (*v1.Pod, error) {
	klog.V(4).InfoS("No suitable pods found; selecting a pod randomly as fallback", "requestID", ctx.RequestID)
	targetPod, err := utils.SelectRandomPod(pods, randomFunc)
	if err != nil {
		klog.ErrorS(err, "Random fallback selection failed", "requestID", ctx.RequestID)
		return nil, fmt.Errorf("random fallback selection failed: %w", err)
	}
	return targetPod, nil
}


Review comment (suggestion):
You can do a warn log message here. And remove the warn message individually from caller just after if targetPod == nil  check, this warn message does not have requestID so it is not useful eitherwise.

Code after:
// mean calculates the mean of a slice of float64 numbers.
func mean(numbers []float64) float64 {
	sum := 0.0
	for _, number := range numbers {
		sum += number
	}
	return sum / float64(len(numbers))
}

// standardDeviation calculates the standard deviation of a slice of float64 numbers.
func standardDeviation(numbers []float64) float64 {
	avg := mean(numbers)
	sumOfSquares := 0.0
	for _, number := range numbers {
		sumOfSquares += math.Pow(number-avg, 2)
	}
	variance := sumOfSquares / float64(len(numbers)-1)
	return math.Sqrt(variance)
}

// SelectRandomPodAsFallback selects a pod randomly as a fallback.
// This method should only be used when all other selection mechanisms have failed.
// For example, if no pods meet the required criteria (e.g., valid metrics or specific conditions),
// this method can be called to randomly select a pod from the provided list.
func SelectRandomPodAsFallback(ctx *types.RoutingContext, pods []*v1.Pod, randomFunc func(int) int) (*v1.Pod, error) {
	klog.Warningf("No suitable pods found; selecting a pod randomly as fallback, requestID: %s", ctx.RequestID)
	targetPod, err := utils.SelectRandomPod(pods, randomFunc)
	if err != nil {
		klog.ErrorS(err, "Random fallback selection failed", "requestID", ctx.RequestID)
		return nil, fmt.Errorf("random fallback selection failed: %w", err)
	}
	return targetPod, nil
}
Repository: vllm-project/aibrix
File: pkg/cache/cache_impl.go (Go)

Code before:
*/

package cache

import (
	"fmt"
	"sync/atomic"

	"github.com/vllm-project/aibrix/pkg/metrics"
	"github.com/vllm-project/aibrix/pkg/types"
	v1 "k8s.io/api/core/v1"
	"k8s.io/klog/v2"
)

// GetPod retrieves a Pod object by name from the cache
// Parameters:
//
//		podName: Name of the pod to retrieve
//	 podNamespace: Namespace of the pod to retrieve
//
// Returns:
//
//	*v1.Pod: The found Pod object
//	error: Error if pod doesn't exist
func (c *Store) GetPod(podName, podNamespace string) (*v1.Pod, error) {
	key := fmt.Sprintf("%s/%s", podNamespace, podName)
	metaPod, ok := c.metaPods.Load(key)
	if !ok {
		return nil, fmt.Errorf("pod does not exist in the cache: %s", podName)
	}

	return metaPod.Pod, nil
}

// ListPods returns all cached Pod objects
// Do not call this directly, for debug purpose and less efficient.
// Returns:
//
//	[]*v1.Pod: Slice of Pod objects
func (c *Store) ListPods() []*v1.Pod {
	pods := make([]*v1.Pod, 0, c.metaPods.Len())
	c.metaPods.Range(func(_ string, metaPod *Pod) bool {
		pods = append(pods, metaPod.Pod)
		return true
	})
	return pods
}

// ListPodsByModel gets Pods associated with a specific model
// Parameters:
//

Review comment (performance):
- Better to make a util method here. 

- Update error messages as well, for example `fmt.Errorf("pod does not exist in the cache: %s", podName)` -> `fmt.Errorf("key does not exist in the cache: %s", key)`

Code after:
*/

package cache

import (
	"fmt"
	"sync/atomic"

	"github.com/vllm-project/aibrix/pkg/metrics"
	"github.com/vllm-project/aibrix/pkg/types"
	"github.com/vllm-project/aibrix/pkg/utils"

	v1 "k8s.io/api/core/v1"
	"k8s.io/klog/v2"
)

// GetPod retrieves a Pod object by name from the cache
// Parameters:
//
//	podName: Name of the pod to retrieve
//	podNamespace: Namespace of the pod to retrieve
//
// Returns:
//
//	*v1.Pod: The found Pod object
//	error: Error if pod doesn't exist
func (c *Store) GetPod(podName, podNamespace string) (*v1.Pod, error) {
	key := utils.GeneratePodKey(podNamespace, podName)
	metaPod, ok := c.metaPods.Load(key)
	if !ok {
		return nil, fmt.Errorf("key does not exist in the cache: %s", key)
	}

	return metaPod.Pod, nil
}

// ListPods returns all cached Pod objects
// Do not call this directly, for debug purpose and less efficient.
// Returns:
//
//	[]*v1.Pod: Slice of Pod objects
func (c *Store) ListPods() []*v1.Pod {
	pods := make([]*v1.Pod, 0, c.metaPods.Len())
	c.metaPods.Range(func(_ string, metaPod *Pod) bool {
		pods = append(pods, metaPod.Pod)
		return true
	})
	return pods
}

// ListPodsByModel gets Pods associated with a specific model
Repository: vllm-project/aibrix
File: test/e2e/vtc_routing_test.go (Go)

Code before:
	ExcellentDistribution: "EXCELLENT",
	GoodDistribution:      "GOOD",
	FairDistribution:      "FAIR",
	PoorDistribution:      "POOR",
}

var redisClient *redis.Client

var (
	availablePods []string
)

func (d DistributionQuality) String() string {
	if str, ok := distributionQualityStrings[d]; ok {
		return str
	}
	return "UNKNOWN"
}

func setupVTCUsers(t *testing.T) {
	if redisClient == nil {
		redisClient = utils.GetRedisClient()
		if redisClient == nil {
			t.Fatal("Failed to connect to Redis")
		}
		defer redisClient.Close()
	}

	getAvailablePods(t)

	ctx := context.Background()
	for _, user := range testUsers {
		err := utils.SetUser(ctx, user, redisClient)
		if err != nil {
			t.Fatalf("Failed to create test user %s: %v", user.Name, err)
		}
		t.Logf("Created test user: %s", user.Name)
	}

	t.Cleanup(func() {
		cleanupVTCUsers(t)
	})
}

func cleanupVTCUsers(t *testing.T) {
	if redisClient == nil {
		return
	}

	ctx := context.Background()
	for _, user := range testUsers {

Review comment (bug):
if we use `defer` here, the scope is associated to `setupVTCUsers`, but `setupVTCUsers` is invoked within the test and test might not be done, I think the idea is to close the connection no matter it succeed or failed. 

Technically, I feel replace with following code makes more sense. binding the lifecycle to test itself

```
		t.Cleanup(func() {
			redisClient.Close()
		})
```

Code after:
	ExcellentDistribution: "EXCELLENT",
	GoodDistribution:      "GOOD",
	FairDistribution:      "FAIR",
	PoorDistribution:      "POOR",
}

var redisClient *redis.Client

var (
	availablePods []string
)

func (d DistributionQuality) String() string {
	if str, ok := distributionQualityStrings[d]; ok {
		return str
	}
	return "UNKNOWN"
}

func setupVTCUsers(t *testing.T) {
	if redisClient == nil {
		redisClient = utils.GetRedisClient()
		if redisClient == nil {
			t.Fatal("Failed to connect to Redis")
		}
	}

	getAvailablePods(t)

	ctx := context.Background()
	for _, user := range testUsers {
		err := utils.SetUser(ctx, user, redisClient)
		if err != nil {
			t.Fatalf("Failed to create test user %s: %v", user.Name, err)
		}
		t.Logf("Created test user: %s", user.Name)
	}

	t.Cleanup(func() {
		cleanupVTCUsers(t)
		redisClient.Close()
	})
}

func cleanupVTCUsers(t *testing.T) {
	if redisClient == nil {
		return
	}

	ctx := context.Background()
	for _, user := range testUsers {
Repository: vllm-project/aibrix
File: hack/rdma/detect-gid-in-container.sh (Shell)

Code before:
#!/bin/bash
set -euo pipefail

DEVICE="mlx5_2"
PORT=1
BASE_PATH="/sys/class/infiniband/${DEVICE}/ports/${PORT}/gid_attrs/types"

echo "Scanning GID types in: $BASE_PATH"
if [ ! -d "$BASE_PATH" ]; then
    echo "❌ Directory not found: $BASE_PATH"
    echo "Check if the RDMA device and port are correct and visible in this container."
    exit 1
fi

for GID_FILE in "$BASE_PATH"/*; do
    IDX=$(basename "$GID_FILE")

    if [ ! -f "$GID_FILE" ]; then
        echo "⚠️  Skipping non-regular file: $GID_FILE"
        continue
    fi

    GID_TYPE=$(cat "$GID_FILE" 2> /tmp/gid_err || true)
    if [[ -z "$GID_TYPE" ]]; then
        ERR=$(</tmp/gid_err)
        if echo "$ERR" | grep -q "Invalid argument"; then
            echo "IDX $IDX: ❌ Invalid argument (likely VF or unsupported GID)"
        else
            echo "IDX $IDX: ❌ Error: $ERR"
        fi

Review comment (suggestion):
```
DEVICE=${DEVICE:-"mlx5_2"}
PORT=${PORT:-1}
```
would be better for users to overwrite these parameter without modifying the script.

Code after:
#!/bin/bash
set -euo pipefail

DEVICE=${DEVICE:-"mlx5_2"}
PORT=${PORT:-1}
BASE_PATH="/sys/class/infiniband/${DEVICE}/ports/${PORT}/gid_attrs/types"

echo "Scanning GID types in: $BASE_PATH"
if [ ! -d "$BASE_PATH" ]; then
    echo "❌ Directory not found: $BASE_PATH"
    echo "Check if the RDMA device and port are correct and visible in this container."
    exit 1
fi

for GID_FILE in "$BASE_PATH"/*; do
    IDX=$(basename "$GID_FILE")

    if [ ! -f "$GID_FILE" ]; then
        echo "⚠️  Skipping non-regular file: $GID_FILE"
        continue
    fi

    GID_TYPE=$(cat "$GID_FILE" 2> /tmp/gid_err || true)
    if [[ -z "$GID_TYPE" ]]; then
        ERR=$(</tmp/gid_err)
        if echo "$ERR" | grep -q "Invalid argument"; then
            echo "IDX $IDX: ❌ Invalid argument (likely VF or unsupported GID)"
        else
            echo "IDX $IDX: ❌ Error: $ERR"
        fi