Add etcdRecovery maintenance type for admin update - ARO-1534

In the event that a master node changes IP addresses (or NIC's) the etcd
quorum will become degraded. The node with the change will then have
it's etcd pod in a crashloop. This is due to the hardcoded etcd spec.

This PR adds the remediation type EtcdRecovery maintenance task to
remediate this issue.

How it works:
  1. Verify this is the issue by comparing etcd's env variables to the
     node's IP address. a degradedEtcd object is returned with relevant
information.
  1. Create a batch job to backup etcd's data directory and move the
     etcd manifest to stop the pod from crash looping.
  1. A batch job is created to run a pod that ssh's into the peer etcd
     container's to remove the failing node from it's member list.
  1. Secret's for the failing pod are deleted
  1. Etcd is patched

Currently there is no endpoint to access this recovery task yet. An
endpoint will be added in a later PR.

Additional scenarios handled:

  - Sometimes the etcd deployement can remediate itself after an IP address change, but there is still data present from the previous IP address\'s member. This results in 4/5 containers running in the pod with the etcd container failing, but no IP address conflicts to use for remediation. Added code to find the failing member based on the conditions if no conflict is found
  - Check for multiple etcd pods with IP mismatches
  - Wait for jobs to reach a succeeded state, when the shell script
    exits with code 0. If this never happens the context is cancelled.
  - Return container log files to user from jobs

pkg/api/admin/openshiftcluster_validatestatic.go

@@ -29,7 +29,10 @@ func (sv openShiftClusterStaticValidator) validateDelta(oc, current *OpenShiftCl
 		return api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodePropertyChangeNotAllowed, err.Target, err.Message)
 	}
 
-	if !(oc.Properties.MaintenanceTask == "" || oc.Properties.MaintenanceTask == MaintenanceTaskEverything || oc.Properties.MaintenanceTask == MaintenanceTaskOperator || oc.Properties.MaintenanceTask == MaintenanceTaskRenewCerts) {
+	if !(oc.Properties.MaintenanceTask == "" ||
+		oc.Properties.MaintenanceTask == MaintenanceTaskEverything ||
+		oc.Properties.MaintenanceTask == MaintenanceTaskOperator ||
+		oc.Properties.MaintenanceTask == MaintenanceTaskRenewCerts) {
 		return api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidParameter, "properties.maintenanceTask", "Invalid enum parameter.")
 	}
 

pkg/frontend/admin_openshiftcluster_etcdrecovery.go

@@ -0,0 +1,69 @@
+package frontend
+
+// Copyright (c) Microsoft Corporation.
+// Licensed under the Apache License 2.0.
+
+import (
+	"context"
+	"net/http"
+	"path/filepath"
+	"strings"
+
+	"github.com/go-chi/chi/v5"
+	operatorclient "github.com/openshift/client-go/operator/clientset/versioned"
+	"github.com/sirupsen/logrus"
+
+	"github.com/Azure/ARO-RP/pkg/api"
+	"github.com/Azure/ARO-RP/pkg/database/cosmosdb"
+	"github.com/Azure/ARO-RP/pkg/frontend/middleware"
+	"github.com/Azure/ARO-RP/pkg/util/restconfig"
+)
+
+func (f *frontend) postAdminOpenShiftClusterEtcdRecovery(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	log := ctx.Value(middleware.ContextKeyLog).(*logrus.Entry)
+	r.URL.Path = filepath.Dir(r.URL.Path)
+
+	b, err := f._postAdminOpenShiftClusterEtcdRecovery(ctx, r, log)
+
+	adminReply(log, w, nil, b, err)
+}
+
+func (f *frontend) _postAdminOpenShiftClusterEtcdRecovery(ctx context.Context, r *http.Request, log *logrus.Entry) ([]byte, error) {
+	resType, resName, resGroupName := chi.URLParam(r, "resourceType"), chi.URLParam(r, "resourceName"), chi.URLParam(r, "resourceGroupName")
+	resourceID := strings.TrimPrefix(r.URL.Path, "/admin")
+
+	doc, err := f.dbOpenShiftClusters.Get(ctx, resourceID)
+	switch {
+	case cosmosdb.IsErrorStatusCode(err, http.StatusNotFound):
+		return []byte{}, api.NewCloudError(http.StatusNotFound, api.CloudErrorCodeResourceNotFound, "", "The Resource '%s/%s' under resource group '%s' was not found.", resType, resName, resGroupName)
+	case err != nil:
+		return []byte{}, err
+	}
+	kubeActions, err := f.kubeActionsFactory(log, f.env, doc.OpenShiftCluster)
+	if err != nil {
+		return []byte{}, err
+	}
+
+	gvr, err := kubeActions.ResolveGVR("Etcd")
+	if err != nil {
+		return []byte{}, err
+	}
+
+	err = validateAdminKubernetesObjects(r.Method, gvr, namespaceEtcds, "cluster")
+	if err != nil {
+		return []byte{}, err
+	}
+
+	restConfig, err := restconfig.RestConfig(f.env, doc.OpenShiftCluster)
+	if err != nil {
+		return []byte{}, err
+	}
+
+	operatorcli, err := operatorclient.NewForConfig(restConfig)
+	if err != nil {
+		return []byte{}, err
+	}
+
+	return f.fixEtcd(ctx, log, f.env, doc, kubeActions, operatorcli.OperatorV1().Etcds())
+}

pkg/frontend/admin_openshiftcluster_kubernetesobjects.go

@@ -129,7 +129,7 @@ func (f *frontend) _deleteAdminKubernetesObjects(ctx context.Context, r *http.Re
 		return err
 	}
 
-	return k.KubeDelete(ctx, groupKind, namespace, name, force)
+	return k.KubeDelete(ctx, groupKind, namespace, name, force, nil)
 }
 
 func (f *frontend) postAdminKubernetesObjects(w http.ResponseWriter, r *http.Request) {

pkg/frontend/admin_openshiftcluster_kubernetesobjects_test.go

@@ -107,7 +107,7 @@ func TestAdminKubernetesObjectsGetAndDelete(t *testing.T) {
 			objName:      "config",
 			mocks: func(tt *test, k *mock_adminactions.MockKubeActions) {
 				k.EXPECT().
-					KubeDelete(gomock.Any(), tt.objKind, tt.objNamespace, tt.objName, false).
+					KubeDelete(gomock.Any(), tt.objKind, tt.objNamespace, tt.objName, false, nil).
 					Return(nil)
 				k.EXPECT().ResolveGVR(tt.objKind).Return(&schema.GroupVersionResource{Resource: "configmaps"}, nil)
 			},
@@ -123,7 +123,7 @@ func TestAdminKubernetesObjectsGetAndDelete(t *testing.T) {
 			force:        "true",
 			mocks: func(tt *test, k *mock_adminactions.MockKubeActions) {
 				k.EXPECT().
-					KubeDelete(gomock.Any(), tt.objKind, tt.objNamespace, tt.objName, true).
+					KubeDelete(gomock.Any(), tt.objKind, tt.objNamespace, tt.objName, true, nil).
 					Return(nil)
 				k.EXPECT().ResolveGVR(tt.objKind).Return(&schema.GroupVersionResource{Resource: "pods"}, nil)
 			},

pkg/frontend/adminactions/kubeactions.go

@@ -15,6 +15,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 
@@ -29,14 +30,16 @@ type KubeActions interface {
 	KubeGet(ctx context.Context, groupKind, namespace, name string) ([]byte, error)
 	KubeList(ctx context.Context, groupKind, namespace string) ([]byte, error)
 	KubeCreateOrUpdate(ctx context.Context, obj *unstructured.Unstructured) error
-	KubeDelete(ctx context.Context, groupKind, namespace, name string, force bool) error
+	KubeDelete(ctx context.Context, groupKind, namespace, name string, force bool, propagationPolicy *metav1.DeletionPropagation) error
 	ResolveGVR(groupKind string) (*schema.GroupVersionResource, error)
 	CordonNode(ctx context.Context, nodeName string, unschedulable bool) error
 	DrainNode(ctx context.Context, nodeName string) error
 	ApproveCsr(ctx context.Context, csrName string) error
 	ApproveAllCsrs(ctx context.Context) error
 	Upgrade(ctx context.Context, upgradeY bool) error
 	KubeGetPodLogs(ctx context.Context, namespace, name, containerName string) ([]byte, error)
+	// kubeWatch returns a watch object for the provided label selector key
+	KubeWatch(ctx context.Context, o *unstructured.Unstructured, label string) (watch.Interface, error)
 }
 
 type kubeActions struct {
@@ -149,7 +152,26 @@ func (k *kubeActions) KubeCreateOrUpdate(ctx context.Context, o *unstructured.Un
 	return err
 }
 
-func (k *kubeActions) KubeDelete(ctx context.Context, groupKind, namespace, name string, force bool) error {
+func (k *kubeActions) KubeWatch(ctx context.Context, o *unstructured.Unstructured, labelKey string) (watch.Interface, error) {
+	gvr, err := k.gvrResolver.Resolve(o.GroupVersionKind().GroupKind().String(), o.GroupVersionKind().Version)
+	if err != nil {
+		return nil, err
+	}
+
+	listOpts := metav1.ListOptions{
+		Limit:         1000, // just in case
+		LabelSelector: o.GetLabels()[labelKey],
+	}
+
+	w, err := k.dyn.Resource(*gvr).Namespace(o.GetNamespace()).Watch(ctx, listOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	return w, nil
+}
+
+func (k *kubeActions) KubeDelete(ctx context.Context, groupKind, namespace, name string, force bool, propagationPolicy *metav1.DeletionPropagation) error {
 	gvr, err := k.gvrResolver.Resolve(groupKind, "")
 	if err != nil {
 		return err
@@ -160,5 +182,9 @@ func (k *kubeActions) KubeDelete(ctx context.Context, groupKind, namespace, name
 		resourceDeleteOptions.GracePeriodSeconds = to.Int64Ptr(0)
 	}
 
+	if propagationPolicy != nil {
+		resourceDeleteOptions.PropagationPolicy = propagationPolicy
+	}
+
 	return k.dyn.Resource(*gvr).Namespace(namespace).Delete(ctx, name, resourceDeleteOptions)
 }