Repository: vllm-project/aibrix
File: pkg/controller/stormservice/sync.go (Go)

Code before:
	expectedService := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      service.Name,
			Namespace: service.Namespace,
			Labels:    service.Labels,
		},
		Spec: corev1.ServiceSpec{
			Type:      corev1.ServiceTypeClusterIP,
			ClusterIP: corev1.ClusterIPNone,
			Selector:  map[string]string{constants.StormServiceNameLabelKey: service.Name},
		},
	}

	headlessService := &corev1.Service{}
	err := r.Client.Get(ctx, client.ObjectKey{Name: service.Name, Namespace: service.Namespace}, headlessService)
	if err != nil {
		if errors.IsNotFound(err) {
			// service doesn't exist, create it
			if createErr := r.Client.Create(ctx, expectedService); createErr != nil {
				return fmt.Errorf("failed to create headless service: %w", createErr)
			}
			r.EventRecorder.Eventf(service, corev1.EventTypeNormal, HeadlessServiceEventType, "Headless Service(discovery) %s created", service.Name)
			return nil
		}
		return err // Return other errors immediately
	}
	if headlessService.Spec.Type == expectedService.Spec.Type &&
		apiequality.Semantic.DeepEqual(headlessService.Spec.Selector, expectedService.Spec.Selector) &&
		headlessService.Spec.ClusterIP == expectedService.Spec.ClusterIP {
		return nil
	}
	// service existed but needs an update.
	headlessService.Spec.Type = expectedService.Spec.Type
	headlessService.Spec.Selector = expectedService.Spec.Selector
	headlessService.Spec.ClusterIP = expectedService.Spec.ClusterIP

	if err := r.Client.Update(ctx, headlessService); err != nil {
		return err
	}
	r.EventRecorder.Eventf(service, corev1.EventTypeNormal, HeadlessServiceEventType, "Headless Service(discovery) %s updated", service.Name)

	return nil
}

func calculateReplicas(desiredReplica, current, updated int32) (desiredCurrent int32, desiredUpdated int32) {
	if desiredReplica == current+updated {
		return current, updated
	}
	currentTotal := current + updated
	if currentTotal != 0 {
		desiredCurrent = integer.RoundToInt32(float64(current*desiredReplica) / float64(currentTotal))

Review comment (bug):
Can we use something like this? (This code seems more concise)

```go
if !isServiceEqual(headlessService, expectedService) {
    headlessService.Spec = expectedService.Spec
    if err := r.Client.Update(ctx, headlessService); err != nil {
	    return fmt.Errorf("failed to update headless service: %w", err)
    }
    r.EventRecorder.Eventf(service, corev1.EventTypeNormal, "Updated", "Headless Service updated")
}
```


```go
func isServiceEqual(a, b *corev1.Service) bool {
	return a.Spec.Type == b.Spec.Type &&
		apiequality.Semantic.DeepEqual(a.Spec.Selector, b.Spec.Selector) &&
		a.Spec.ClusterIP == b.Spec.ClusterIP
}
```

Also, I'm not sure why we need a sync update. The headless service seems to be injected internally at the beginning and will not involve modifications? Right? 🤔 

Code after:
	expectedService := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      service.Name,
			Namespace: service.Namespace,
			Labels:    service.Labels,
		},
		Spec: corev1.ServiceSpec{
			Type:      corev1.ServiceTypeClusterIP,
			ClusterIP: corev1.ClusterIPNone,
			Selector:  map[string]string{constants.StormServiceNameLabelKey: service.Name},
		},
	}

	headlessService := &corev1.Service{}
	err := r.Client.Get(ctx, client.ObjectKey{Name: service.Name, Namespace: service.Namespace}, headlessService)
	if err != nil {
		if errors.IsNotFound(err) {
			// service doesn't exist, create it
			if createErr := r.Client.Create(ctx, expectedService); createErr != nil {
				return fmt.Errorf("failed to create headless service: %w", createErr)
			}
			r.EventRecorder.Eventf(service, corev1.EventTypeNormal, HeadlessServiceEventType, "Headless Service(discovery) %s created", service.Name)
			return nil
		}
		return err // Return other errors immediately
	}

	if !isServiceEqual(headlessService, expectedService) {
		headlessService.Spec = expectedService.Spec
		if err := r.Client.Update(ctx, headlessService); err != nil {
			return fmt.Errorf("failed to update headless service: %w", err)
		}
		r.EventRecorder.Eventf(service, corev1.EventTypeNormal, "Updated", "Headless Service updated")
	}

	r.EventRecorder.Eventf(service, corev1.EventTypeNormal, HeadlessServiceEventType, "Headless Service(discovery) %s updated", service.Name)

	return nil
}

func calculateReplicas(desiredReplica, current, updated int32) (desiredCurrent int32, desiredUpdated int32) {
	if desiredReplica == current+updated {
		return current, updated
	}
	currentTotal := current + updated
	if currentTotal != 0 {
		desiredCurrent = integer.RoundToInt32(float64(current*desiredReplica) / float64(currentTotal))
		desiredUpdated = integer.RoundToInt32(float64(updated*desiredReplica) / float64(currentTotal))
	}
	// 1. if current == updated == 0, scaling out updated
	// 2. overcome the rounding error