Skip to content

Commit

Permalink
Merge pull request #28 from janlauber/fix-multi-rollouts
Browse files Browse the repository at this point in the history 
Fix multi rollouts
  • Loading branch information
@janlauber
janlauber authored Apr 15, 2024
2 parents 4e4038f + a7431e4 commit 2c0c91f
Show file tree
Hide file tree
Showing 4 changed files with 92 additions and 131 deletions.
4 changes: 3 additions & 1 deletion .gitignore
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,6 @@ Dockerfile.cross
*.swp
*.swo
*~
vendor/
vendor/

test.yaml
50 changes: 31 additions & 19 deletions controllers/deployment.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,31 +22,43 @@ import (

func (r *RolloutReconciler) reconcileDeployment(ctx context.Context, f *oneclickiov1alpha1.Rollout) error {
log := log.FromContext(ctx)
deploymentName := f.Name
namespace := f.Namespace

desiredDeployment := r.deploymentForRollout(ctx, f)

currentDeployment := &appsv1.Deployment{}
err := r.Get(ctx, types.NamespacedName{Name: f.Name, Namespace: f.Namespace}, currentDeployment)
if err != nil && errors.IsNotFound(err) {
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Creating", "Creating Deployment %s", f.Name)
return r.Create(ctx, desiredDeployment)
} else if err != nil {
log.Error(err, "Failed to get Deployment")
return err
}

// Compare the current Deployment with the Rollout spec
if needsUpdate(currentDeployment, f) {
// Update the Deployment to align it with the Rollout spec
currentDeployment.Spec = desiredDeployment.Spec
err = r.Update(ctx, currentDeployment)
result := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentDeployment := &appsv1.Deployment{}
err := r.Get(ctx, types.NamespacedName{Name: deploymentName, Namespace: namespace}, currentDeployment)
if err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "UpdateFailed", "Failed to update Deployment %s", f.Name)
if errors.IsNotFound(err) {
// Deployment not found, create a new one
desiredDeployment := r.deploymentForRollout(ctx, f)
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Creating", "Creating Deployment %s", deploymentName)
return r.Create(ctx, desiredDeployment)
}
// Other error while fetching the Deployment
return err
}

// Deployment found, check if it needs updating
desiredDeployment := r.deploymentForRollout(ctx, f)
if needsUpdate(currentDeployment, f) {
// Update the Deployment to align it with the Rollout spec
currentDeployment.Spec = desiredDeployment.Spec
updateErr := r.Update(ctx, currentDeployment)
if updateErr != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "UpdateFailed", "Failed to update Deployment %s", deploymentName)
return updateErr
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Updated", "Updated Deployment %s", deploymentName)
}
return nil
})

if result != nil {
log.Error(result, "Failed to reconcile Deployment", "Deployment.Namespace", namespace, "Deployment.Name", deploymentName)
}

return nil
return result
}

func (r *RolloutReconciler) deploymentForRollout(ctx context.Context, f *oneclickiov1alpha1.Rollout) *appsv1.Deployment {
Expand Down
69 changes: 23 additions & 46 deletions controllers/service.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,65 +18,47 @@ import (
func (r *RolloutReconciler) reconcileService(ctx context.Context, f *oneclickiov1alpha1.Rollout) error {
log := log.FromContext(ctx)

// Keep track of services that should exist according to the Rollout spec
expectedServices := make(map[string]bool)
for _, intf := range f.Spec.Interfaces {
serviceName := f.Name + "-" + intf.Name + "-svc"
expectedServices[serviceName] = true

expectedServices[f.Name+"-"+intf.Name+"-svc"] = true
// Process each interface
service := r.serviceForRollout(f, intf)

foundService := &corev1.Service{}
err := r.Get(ctx, types.NamespacedName{Name: service.Name, Namespace: f.Namespace}, foundService)
err := r.Get(ctx, types.NamespacedName{Name: serviceName, Namespace: f.Namespace}, foundService)
if err != nil && errors.IsNotFound(err) {
// If the Service is not found, create a new one
err = r.Create(ctx, service)
if err != nil {
// Handle creation error
r.Recorder.Eventf(f, corev1.EventTypeWarning, "CreationFailed", "Failed to create Service %s", service.Name)
if err := r.Create(ctx, service); err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "CreationFailed", "Failed to create Service %s", serviceName)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Created", "Created Service %s", service.Name)
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Created", "Created Service %s", serviceName)
} else if err != nil {
// Handle other errors
r.Recorder.Eventf(f, corev1.EventTypeWarning, "GetFailed", "Failed to get Service %s", service.Name)
r.Recorder.Eventf(f, corev1.EventTypeWarning, "GetFailed", "Failed to get Service %s", serviceName)
return err
} else {
// If the Service exists, check if it needs to be updated
desiredPorts := getServicePorts(intf)
if !reflect.DeepEqual(foundService.Spec.Ports, desiredPorts) {
foundService.Spec.Ports = desiredPorts
err = r.Update(ctx, foundService)
if err != nil {
// Handle update error
r.Recorder.Eventf(f, corev1.EventTypeWarning, "UpdateFailed", "Failed to update Service %s", foundService.Name)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Updated", "Updated Service %s", foundService.Name)
} else if !reflect.DeepEqual(foundService.Spec.Ports, getServicePorts(intf)) {
foundService.Spec.Ports = getServicePorts(intf)
if err := r.Update(ctx, foundService); err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "UpdateFailed", "Failed to update Service %s", serviceName)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Updated", "Updated Service %s", serviceName)
}
}

// Delete services that are no longer specified in the Rollout spec
serviceList := &corev1.ServiceList{}
listOpts := []client.ListOption{client.InNamespace(f.Namespace)}
err := r.List(ctx, serviceList, listOpts...)
if err != nil {
if err := r.List(ctx, serviceList, listOpts...); err != nil {
log.Error(err, "Failed to list services", "Namespace", f.Namespace)
return err
}

for _, service := range serviceList.Items {
if _, exists := expectedServices[service.Name]; !exists {
// Service is no longer needed, delete it
if service.Labels["one-click.dev/projectId"] == f.Namespace && service.Labels["one-click.dev/deploymentId"] == f.Name {
err = r.Delete(ctx, &service)
if err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "DeletionFailed", "Failed to delete Service %s", service.Name)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Deleted", "Deleted Service %s", service.Name)
if _, exists := expectedServices[service.Name]; !exists && service.Labels["one-click.dev/projectId"] == f.Namespace && service.Labels["one-click.dev/deploymentId"] == f.Name {
if err := r.Delete(ctx, &service); err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "DeletionFailed", "Failed to delete Service %s", service.Name)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Deleted", "Deleted Service %s", service.Name)
}
}

Expand All @@ -88,24 +70,19 @@ func (r *RolloutReconciler) serviceForRollout(f *oneclickiov1alpha1.Rollout, int
"one-click.dev/projectId": f.Namespace,
"one-click.dev/deploymentId": f.Name,
}
selectorLabels := map[string]string{
"one-click.dev/projectId": f.Namespace,
"one-click.dev/deploymentId": f.Name,
}
svc := &corev1.Service{
ObjectMeta: metav1.ObjectMeta{
Name: f.Name + "-" + intf.Name + "-svc", // Create a unique name for the Service
Name: f.Name + "-" + intf.Name + "-svc",
Namespace: f.Namespace,
Labels: labels,
},
Spec: corev1.ServiceSpec{
Selector: selectorLabels,
Selector: labels,
Ports: getServicePorts(intf),
Type: corev1.ServiceTypeClusterIP, // Default to ClusterIP, modify if needed
Type: corev1.ServiceTypeClusterIP,
},
}

// Set Rollout instance as the owner and controller
ctrl.SetControllerReference(f, svc, r.Scheme)
return svc
}
Expand All @@ -116,7 +93,7 @@ func getServicePorts(intf oneclickiov1alpha1.InterfaceSpec) []corev1.ServicePort
Name: intf.Name,
Port: intf.Port,
TargetPort: intstr.FromInt(int(intf.Port)),
Protocol: corev1.ProtocolTCP, // Defaulting to TCP, change as needed
Protocol: corev1.ProtocolTCP,
},
}
}
100 changes: 35 additions & 65 deletions controllers/volume.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package controllers

import (
"context"
"fmt"

oneclickiov1alpha1 "github.com/janlauber/one-click-operator/api/v1alpha1"
corev1 "k8s.io/api/core/v1"
Expand All @@ -11,83 +10,60 @@ import (
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
)

func (r *RolloutReconciler) reconcilePVCs(ctx context.Context, f *oneclickiov1alpha1.Rollout) error {
log := log.FromContext(ctx)

// Check if the Rollout spec's volume list is empty
if len(f.Spec.Volumes) == 0 {
// If no volumes are defined, delete all PVCs associated with this Rollout
return r.deleteAllPVCsForRollout(ctx, f)
}

// Keep track of the PVCs that should exist according to the Rollout specification
expectedPVCs := make(map[string]struct{})
for _, volSpec := range f.Spec.Volumes {
// expectedPVCs[volSpec.Name] = struct{}{}
expectedPVCs[f.Name+"-"+volSpec.Name] = struct{}{}
desiredPvc := r.constructPVCForRollout(f, volSpec)
pvcName := f.Name + "-" + volSpec.Name
expectedPVCs[pvcName] = struct{}{}

foundPvc := &corev1.PersistentVolumeClaim{}
err := r.Get(ctx, types.NamespacedName{Name: desiredPvc.Name, Namespace: f.Namespace}, foundPvc)
desiredPVC := r.constructPVCForRollout(f, volSpec)
foundPVC := &corev1.PersistentVolumeClaim{}
err := r.Get(ctx, types.NamespacedName{Name: pvcName, Namespace: f.Namespace}, foundPVC)
if err != nil && errors.IsNotFound(err) {
err = r.Create(ctx, desiredPvc)
if err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "CreationFailed", "Failed to create PVC %s", desiredPvc.Name)
if err := r.Create(ctx, desiredPVC); err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "CreationFailed", "Failed to create PVC %s", pvcName)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Created", "Created PVC %s", desiredPvc.Name)
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Created", "Created PVC %s", pvcName)
} else if err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "GetFailed", "Failed to get PVC %s", desiredPvc.Name)
r.Recorder.Eventf(f, corev1.EventTypeWarning, "GetFailed", "Failed to get PVC %s", pvcName)
return err
} else {
// Existing PVC found, check for changes
} else if currentSize, desiredSize := foundPVC.Spec.Resources.Requests[corev1.ResourceStorage], resource.MustParse(volSpec.Size); desiredSize.Cmp(currentSize) > 0 {
if foundPVC.Spec.VolumeMode == nil || *foundPVC.Spec.VolumeMode != corev1.PersistentVolumeFilesystem {
log.Info("PVC resizing is only supported for filesystem volume mode")
return nil
}

storageClass := &storagev1.StorageClass{}
if err := r.Get(ctx, types.NamespacedName{Name: *foundPVC.Spec.StorageClassName}, storageClass); err != nil {
log.Error(err, "Failed to get the storage class of the PVC", "StorageClass", *foundPVC.Spec.StorageClassName)
return err
}

// Preventing name and storage class changes
if foundPvc.Name != desiredPvc.Name || (foundPvc.Spec.StorageClassName != nil && *foundPvc.Spec.StorageClassName != *desiredPvc.Spec.StorageClassName) {
log.Error(fmt.Errorf("name or storage class change not allowed"), "Invalid PVC update", "PVC.Name", foundPvc.Name)
return fmt.Errorf("name or storage class change not allowed for PVC %s", foundPvc.Name)
if !allowsVolumeExpansion(storageClass) {
log.Info("StorageClass does not allow volume expansion", "StorageClass", storageClass.Name)
return nil
}

// Handle size increase
currentSize := foundPvc.Spec.Resources.Requests[corev1.ResourceStorage]
desiredSize := resource.MustParse(volSpec.Size)
if desiredSize.Cmp(currentSize) > 0 {

// Check if PVC and its StorageClass allow resizing
if foundPvc.Spec.VolumeMode == nil || *foundPvc.Spec.VolumeMode != corev1.PersistentVolumeFilesystem {
log.Info("PVC resizing is only supported for filesystem volume mode")
return nil // or handle the error as per your application's logic
}

storageClass := &storagev1.StorageClass{}
if err := r.Get(ctx, types.NamespacedName{Name: *foundPvc.Spec.StorageClassName}, storageClass); err != nil {
log.Error(err, "Failed to get the storage class of the PVC", "StorageClass", *foundPvc.Spec.StorageClassName)
return err
}

if !allowsVolumeExpansion(storageClass) {
log.Info("StorageClass does not allow volume expansion", "StorageClass", storageClass.Name)
return nil // or handle the error as per your application's logic
}

// Update PVC size (considering Kubernetes limitations - PVCs can generally only be increased in size)
foundPvc.Spec.Resources.Requests[corev1.ResourceStorage] = desiredSize
err := r.Update(ctx, foundPvc)
if err != nil {
log.Error(err, "Failed to update PVC size", "PVC.Namespace", foundPvc.Namespace, "PVC.Name", foundPvc.Name)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Updated", "Updated PVC %s", foundPvc.Name)
foundPVC.Spec.Resources.Requests[corev1.ResourceStorage] = desiredSize
if err := r.Update(ctx, foundPVC); err != nil {
log.Error(err, "Failed to update PVC size", "PVC.Namespace", foundPVC.Namespace, "PVC.Name", pvcName)
return err
}
r.Recorder.Eventf(f, corev1.EventTypeNormal, "Updated", "Updated PVC %s", pvcName)
}
}

// List all PVCs in the namespace
pvcList := &corev1.PersistentVolumeClaimList{}
if err := r.List(ctx, pvcList, client.InNamespace(f.Namespace)); err != nil {
log.Error(err, "Failed to list PVCs", "Rollout.Namespace", f.Namespace)
Expand All @@ -96,7 +72,6 @@ func (r *RolloutReconciler) reconcilePVCs(ctx context.Context, f *oneclickiov1al

for _, pvc := range pvcList.Items {
if _, exists := expectedPVCs[pvc.Name]; !exists && isOwnedByRollout(&pvc, f) {
// Delete the PVC if it's not in the expected list and is owned by the Rollout
if err := r.Delete(ctx, &pvc); err != nil {
r.Recorder.Eventf(f, corev1.EventTypeWarning, "DeletionFailed", "Failed to delete PVC %s", pvc.Name)
return err
Expand All @@ -118,25 +93,21 @@ func (r *RolloutReconciler) constructPVCForRollout(f *oneclickiov1alpha1.Rollout
"one-click.dev/deploymentId": f.Name,
}

pvc := &corev1.PersistentVolumeClaim{
return &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Name: f.Name + "-" + volSpec.Name,
Namespace: f.Namespace,
Labels: labels,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(f, f.GroupVersionKind()),
},
},
Spec: corev1.PersistentVolumeClaimSpec{
AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
Resources: corev1.VolumeResourceRequirements{
Requests: corev1.ResourceList{
corev1.ResourceStorage: resource.MustParse(volSpec.Size),
},
},
AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
Resources: corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse(volSpec.Size)}},
StorageClassName: &volSpec.StorageClass,
},
}

ctrl.SetControllerReference(f, pvc, r.Scheme)
return pvc
}

func isOwnedByRollout(pvc *corev1.PersistentVolumeClaim, f *oneclickiov1alpha1.Rollout) bool {
Expand All @@ -148,13 +119,12 @@ func isOwnedByRollout(pvc *corev1.PersistentVolumeClaim, f *oneclickiov1alpha1.R
return false
}

// deleteAllPVCsForRollout deletes all PVCs associated with a given Rollout
func (r *RolloutReconciler) deleteAllPVCsForRollout(ctx context.Context, f *oneclickiov1alpha1.Rollout) error {
log := log.FromContext(ctx)

pvcList := &corev1.PersistentVolumeClaimList{}
if err := r.List(ctx, pvcList, client.InNamespace(f.Namespace)); err != nil {
log.Error(err, "Failed to list PVCs for Rollout", "Rollout.Namespace", f.Namespace, "Rollout.Name", f.Name)
log.Error(err, "Failed to list PVCs for Rollout", "Rollout.Namespace", f.Namespace)
return err
}

Expand Down

0 comments on commit 2c0c91f

Please sign in to comment.