[FVR-182] Restarting Pods Through Deletion

service/k8s/statefulset.go

@@ -29,6 +29,7 @@ type StatefulSet interface {
 	CreateOrUpdateStatefulSet(namespace string, statefulSet *appsv1.StatefulSet) error
 	DeleteStatefulSet(namespace string, name string) error
 	ListStatefulSets(namespace string) (*appsv1.StatefulSetList, error)
+	DeleteStatefulSetPods(namespace string, name string) error
 }
 
 // StatefulSetService is the service account service implementation using API calls to kubernetes.
@@ -161,7 +162,17 @@ func (s *StatefulSetService) CreateOrUpdateStatefulSet(namespace string, statefu
 				annotations["storageCapacity"] = fmt.Sprintf("%d", stateCapacity)
 				storedStatefulSet.Annotations = annotations
 				if realUpdate {
+
 					s.logger.WithField("namespace", namespace).WithField("statefulSet", statefulSet.Name).Infof("resize statefulset pvcs from %d to %d Success", storedCapacity, stateCapacity)
+
+					s.logger.WithField("namespace", namespace).WithField("statefulSet", statefulSet.Name).Infof("deleting statefulset pods in order to update pvc mount")
+					err := s.DeleteStatefulSetPods(namespace, storedStatefulSet.Name)
+
+					if err != nil {
+						s.logger.WithField("namespace", namespace).WithField("statefulSet", statefulSet.Name).Warningf("deletion of sts pods failed:%s", err.Error())
+						return err
+					}
+
 				} else {
 					s.logger.WithField("namespace", namespace).WithField("pvc", rfName).Warningf("set annotations,resize nothing")
 				}
@@ -188,3 +199,27 @@ func (s *StatefulSetService) ListStatefulSets(namespace string) (*appsv1.Statefu
 	recordMetrics(namespace, "StatefulSet", metrics.NOT_APPLICABLE, "LIST", err, s.metricsRecorder)
 	return stsList, err
 }
+
+func (s *StatefulSetService) DeleteStatefulSetPods(namespace, name string) error {
+
+	rps, err := s.GetStatefulSetPods(namespace, name)
+
+	if err != nil {
+		return err
+	}
+
+	var deleteErrors []string
+	for _, rp := range rps.Items {
+		err := s.kubeClient.CoreV1().Pods(namespace).Delete(context.TODO(), rp.Name, metav1.DeleteOptions{})
+		recordMetrics(namespace, "Pod", name, "DELETE", err, s.metricsRecorder)
+		if err != nil {
+			deleteErrors = append(deleteErrors, err.Error())
+		}
+	}
+
+	if len(deleteErrors) > 0 {
+		return fmt.Errorf("failed to delete some pods: %s", strings.Join(deleteErrors, "; "))
+	}
+
+	return nil
+}

service/k8s/statefulset_test.go

@@ -20,10 +20,14 @@ import (
 	"github.com/spotahome/redis-operator/log"
 	"github.com/spotahome/redis-operator/metrics"
 	"github.com/spotahome/redis-operator/service/k8s"
+
+	corev1 "k8s.io/api/core/v1"
 )
 
 var (
-	statefulSetsGroup = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "statefulsets"}
+	statefulSetsGroup          = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "statefulsets"}
+	persistentVolumeClaimGroup = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "persistentvolumeclaims"}
+	podGroup                   = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}
 )
 
 func newStatefulSetUpdateAction(ns string, statefulSet *appsv1.StatefulSet) kubetesting.UpdateActionImpl {
@@ -38,6 +42,22 @@ func newStatefulSetCreateAction(ns string, statefulSet *appsv1.StatefulSet) kube
 	return kubetesting.NewCreateAction(statefulSetsGroup, ns, statefulSet)
 }
 
+func newPVCUpdateAction(pvc *corev1.PersistentVolumeClaim) kubetesting.UpdateActionImpl {
+	return kubetesting.NewUpdateAction(persistentVolumeClaimGroup, "", pvc)
+}
+
+func newPVCListAction(opts metav1.ListOptions) kubetesting.ListActionImpl {
+	return kubetesting.NewListAction(persistentVolumeClaimGroup, schema.GroupVersionKind{Group: "", Version: "v1", Kind: "PersistentVolumeClaim"}, "", opts)
+}
+
+func newPodListAction(ns string, opts metav1.ListOptions) kubetesting.ListActionImpl {
+	return kubetesting.NewListAction(podGroup, schema.GroupVersionKind{Group: "", Version: "v1", Kind: "Pod"}, ns, opts)
+}
+
+func newPodDeleteAction(ns string, name string) kubetesting.DeleteActionImpl {
+	return kubetesting.NewDeleteAction(podGroup, ns, name)
+}
+
 func TestStatefulSetServiceGetCreateOrUpdate(t *testing.T) {
 	testStatefulSet := &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -130,6 +150,7 @@ func TestStatefulSetServiceGetCreateOrUpdate(t *testing.T) {
 					ResourceVersion: "10",
 				},
 				Spec: appsv1.StatefulSetSpec{
+					Selector: &metav1.LabelSelector{},
 					VolumeClaimTemplates: []v1.PersistentVolumeClaim{
 						{
 							Spec: v1.PersistentVolumeClaimSpec{
@@ -162,6 +183,17 @@ func TestStatefulSetServiceGetCreateOrUpdate(t *testing.T) {
 					},
 				},
 			}
+
+			podList := &v1.PodList{
+				Items: []v1.Pod{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "pod-1",
+							Labels: map[string]string{},
+						},
+					},
+				},
+			}
 			pvcList := &v1.PersistentVolumeClaimList{
 				Items: []v1.PersistentVolumeClaim{
 					{
@@ -195,6 +227,18 @@ func TestStatefulSetServiceGetCreateOrUpdate(t *testing.T) {
 				},
 			}
 			// Mock.
+			opts := metav1.ListOptions{
+				LabelSelector: "app.kubernetes.io/component=redis,app.kubernetes.io/name=teststatefulSet1,app.kubernetes.io/part-of=redis-failover",
+			}
+			expActions := []kubetesting.Action{
+				newStatefulSetGetAction(testns, beforeSts.ObjectMeta.Name),
+				newPVCListAction(opts),
+				newPVCUpdateAction(&pvcList.Items[0]),
+				newStatefulSetGetAction(testns, afterSts.ObjectMeta.Name),
+				newPodListAction(testns, metav1.ListOptions{}),
+				newPodDeleteAction(testns, podList.Items[0].ObjectMeta.Name),
+				newStatefulSetUpdateAction(testns, afterSts),
+			}
 			mcli := &kubernetes.Clientset{}
 			mcli.AddReactor("get", "statefulsets", func(action kubetesting.Action) (bool, runtime.Object, error) {
 				return true, beforeSts, nil
@@ -207,17 +251,41 @@ func TestStatefulSetServiceGetCreateOrUpdate(t *testing.T) {
 				pvcList.Items[0] = *action.(kubetesting.UpdateActionImpl).Object.(*v1.PersistentVolumeClaim)
 				return true, action.(kubetesting.UpdateActionImpl).Object, nil
 			})
+
+			mcli.AddReactor("list", "pods", func(action kubetesting.Action) (handled bool, ret runtime.Object, err error) {
+				return true, podList, nil
+			})
+
+			mcli.AddReactor("delete", "pods", func(action kubetesting.Action) (handled bool, ret runtime.Object, err error) {
+				return true, podList, nil
+			})
+
 			service := k8s.NewStatefulSetService(mcli, log.Dummy, metrics.Dummy)
 			err := service.CreateOrUpdateStatefulSet(testns, afterSts)
 			assert.NoError(err)
 			assert.Equal(pvcList.Items[0].Spec.Resources, pvcList.Items[1].Spec.Resources)
+			assert.Equal(expActions, mcli.Actions())
+
 			// should not call update
+			mcli = &kubernetes.Clientset{}
+			mcli.AddReactor("get", "statefulsets", func(action kubetesting.Action) (bool, runtime.Object, error) {
+				return true, afterSts, nil
+			})
+
+			// no deletion pods anymore, as pvc is already resized
+			expActions = []kubetesting.Action{
+				newStatefulSetGetAction(testns, afterSts.ObjectMeta.Name),
+				newPVCListAction(opts),
+				newStatefulSetUpdateAction(testns, afterSts),
+			}
+
 			mcli.AddReactor("update", "persistentvolumeclaims", func(action kubetesting.Action) (handled bool, ret runtime.Object, err error) {
 				panic("shouldn't call update")
 			})
 			service = k8s.NewStatefulSetService(mcli, log.Dummy, metrics.Dummy)
 			err = service.CreateOrUpdateStatefulSet(testns, afterSts)
 			assert.NoError(err)
+			assert.Equal(expActions, mcli.Actions())
 		})
 	}
 }