kfctl delete refactor (#309)

* kustomize: Refactor deletion algorithm

Refactor the kfctl deletion algorithm as described in issue
kubectl/kfctl#293:

1. Build kustomize folder if necessary
2. For every application in reverse KfDef order, do an equivalent of
   `kustomize build | kubectl delete -f -`

Signed-off-by: Yannis Zarkadas <yanniszark@arrikto.com>

* Don't fail early, attempt to delete everything

Signed-off-by: Yannis Zarkadas <yanniszark@arrikto.com>

cmd/manager/main.go

@@ -21,18 +21,18 @@ import (
 	"github.com/operator-framework/operator-sdk/pkg/metrics"
 	"github.com/operator-framework/operator-sdk/pkg/restmapper"
 	sdkVersion "github.com/operator-framework/operator-sdk/version"
+	log "github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
-	log "github.com/sirupsen/logrus"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
 )
 
 // Kubeflow operator version
 var (
-	Version         string = "1.0.0"
+	Version string = "1.0.0"
 )
 
 // Change below variables to serve metrics on different host or port.

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/aws/aws-sdk-go v1.27.1
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/chai2010/gettext-go v0.0.0-20170215093142-bf70f2a70fb1 // indirect
+	github.com/davecgh/go-spew v1.1.1
 	github.com/deckarep/golang-set v1.7.1
 	github.com/docker/docker v1.13.1 // indirect
 	github.com/docker/spdystream v0.0.0-20181023171402-6480d4af844c // indirect
@@ -17,6 +18,7 @@ require (
 	github.com/fatih/color v1.7.0
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
 	github.com/go-openapi/jsonpointer v0.19.2 // indirect
+	github.com/go-yaml/yaml v2.1.0+incompatible
 	github.com/gogo/protobuf v1.3.1
 	github.com/golangplus/testing v0.0.0-20180327235837-af21d9c3145e
 	github.com/google/go-cmp v0.3.1

go.sum

@@ -300,6 +300,8 @@ github.com/go-ozzo/ozzo-validation v3.5.0+incompatible/go.mod h1:gsEKFIVnabGBt6m
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/go-yaml/yaml v2.1.0+incompatible h1:RYi2hDdss1u4YE7GwixGzWwVo47T8UQwnTLB6vQiq+o=
+github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
 github.com/gobuffalo/envy v1.6.5/go.mod h1:N+GkhhZ/93bGZc6ZKhJLP6+m+tCNPKwgSpH9kaifseQ=
 github.com/gobuffalo/envy v1.7.0/go.mod h1:n7DRkBerg/aorDM8kbduw5dN3oXGswK5liaSCx4T5NI=
 github.com/gobuffalo/envy v1.7.1/go.mod h1:FurDp9+EDPE4aIUS3ZLyD+7/9fpx7YRt/ukY6jIHf0w=
@@ -342,7 +344,9 @@ github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golangplus/bytes v0.0.0-20160111154220-45c989fe5450 h1:7xqw01UYS+KCI25bMrPxwNYkSns2Db1ziQPpVq99FpE=
 github.com/golangplus/bytes v0.0.0-20160111154220-45c989fe5450/go.mod h1:Bk6SMAONeMXrxql8uvOKuAZSu8aM5RUGv+1C6IJaEho=
+github.com/golangplus/fmt v0.0.0-20150411045040-2a5d6d7d2995 h1:f5gsjBiF9tRRVomCvrkGMMWI8W1f2OBFar2c5oakAP0=
 github.com/golangplus/fmt v0.0.0-20150411045040-2a5d6d7d2995/go.mod h1:lJgMEyOkYFkPcDKwRXegd+iM6E7matEszMG5HhwytU8=
 github.com/golangplus/testing v0.0.0-20180327235837-af21d9c3145e h1:KhcknUwkWHKZPbFy2P7jH5LKJ3La+0ZeknkkmrSgqb0=
 github.com/golangplus/testing v0.0.0-20180327235837-af21d9c3145e/go.mod h1:0AA//k/eakGydO4jKRoRL2j92ZKSzTgj9tclaCrvXHk=

pkg/kfapp/kustomize/kustomize.go

@@ -22,10 +22,12 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io/ioutil"
+	errutil "k8s.io/apimachinery/pkg/util/errors"
 	"math/rand"
 	"os"
 	"path"
 	"path/filepath"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"strconv"
 	"strings"
 	"time"
@@ -324,12 +326,6 @@ func (kustomize *kustomize) deleteGlobalResources() error {
 
 // Delete is called from 'kfctl delete ...'. Will delete all resources deployed from the Apply method
 func (kustomize *kustomize) Delete(resources kftypesv3.ResourceEnum) error {
-	if err := kustomize.initK8sClients(); err != nil {
-		return &kfapisv3.KfError{
-			Code:    int(kfapisv3.INVALID_ARGUMENT),
-			Message: fmt.Sprintf("Error: kustomize plugin couldn't initialize a K8s client %v", err),
-		}
-	}
 	annotations := kustomize.kfDef.GetAnnotations()
 	forceDelete := false
 	if forceDel, ok := annotations[strings.Join([]string{utils.KfDefAnnotation, utils.ForceDelete}, "/")]; ok {
@@ -340,46 +336,92 @@ func (kustomize *kustomize) Delete(resources kftypesv3.ResourceEnum) error {
 	if forceDelete {
 		log.Warnf("running force deletion.")
 	}
-	if kustomize.kfDef.ClusterName == "" {
-		msg := "cannot find ClusterName within KfDef, this may cause error deletion to clusters."
+
+	// Get kubeconfig for cluster and initialize clients
+	msg := ""
+	kubeconfig := kftypesv3.GetKubeConfig()
+	if kubeconfig == nil {
+		msg = "unable to load .kubeconfig."
+	} else {
+		currentCtx := kubeconfig.CurrentContext
+		if ctx, ok := kubeconfig.Contexts[currentCtx]; !ok || ctx == nil {
+			msg = "cannot find current-context in kubeconfig."
+		} else {
+			if kustomize.kfDef.ClusterName != ctx.Cluster {
+				msg = fmt.Sprintf("cluster name doesn't match: KfDef(%v) v.s. current-context(%v)",
+					kustomize.kfDef.ClusterName, ctx.Cluster)
+			}
+		}
+	}
+	if msg != "" {
 		if forceDelete {
-			log.Warnf(msg + " ;running kfctl delete because force-deletion is set.")
+			log.Warnf(msg)
 		} else {
 			return &kfapisv3.KfError{
 				Code:    int(kfapisv3.INVALID_ARGUMENT),
 				Message: msg,
 			}
 		}
-	} else {
-		msg := ""
-		kubeconfig := kftypesv3.GetKubeConfig()
-		if kubeconfig == nil {
-			msg = "unable to load .kubeconfig."
-		} else {
-			currentCtx := kubeconfig.CurrentContext
-			if ctx, ok := kubeconfig.Contexts[currentCtx]; !ok || ctx == nil {
-				msg = "cannot find current-context in kubeconfig."
-			} else {
-				if kustomize.kfDef.ClusterName != ctx.Cluster {
-					msg = fmt.Sprintf("cluster name doesn't match: KfDef(%v) v.s. current-context(%v)",
-						kustomize.kfDef.ClusterName, ctx.Cluster)
-				}
+	}
+	kustomize.initK8sClients()
+	kubeclient, err := client.New(kustomize.restConfig, client.Options{})
+	if err != nil {
+		return &kfapisv3.KfError{
+			Code:    int(kfapisv3.INTERNAL_ERROR),
+			Message: fmt.Sprintf("error initializing k8s client Error %v", err),
+		}
+	}
+
+	// Delete in reverse application order
+	kustomizeDir := path.Join(kustomize.kfDef.Spec.AppDir, outputDir)
+	errList := []error{}
+	for idx := range kustomize.kfDef.Spec.Applications {
+		app := &kustomize.kfDef.Spec.Applications[len(kustomize.kfDef.Spec.Applications)-1-idx]
+		log.Infof("Deleting application %v", app.Name)
+		resMap, err := EvaluateKustomizeManifest(path.Join(kustomizeDir, app.Name))
+		if err != nil {
+			log.Errorf("error evaluating kustomization manifest for %v Error %v", app.Name, err)
+			return &kfapisv3.KfError{
+				Code:    int(kfapisv3.INTERNAL_ERROR),
+				Message: fmt.Sprintf("error evaluating kustomization manifest for %v Error %v", app.Name, err),
 			}
 		}
-		if msg != "" {
-			if forceDelete {
-				log.Warnf(msg)
-			} else {
-				return &kfapisv3.KfError{
-					Code:    int(kfapisv3.INVALID_ARGUMENT),
-					Message: msg,
-				}
+		yamlBytes, err := resMap.AsYaml()
+		if err != nil {
+			return &kfapisv3.KfError{
+				Code:    int(kfapisv3.INTERNAL_ERROR),
+				Message: fmt.Sprintf("error evaluating kustomization manifest for %v Error %v", app.Name, err),
+			}
+		}
+		resources, err := utils.SplitYAML(yamlBytes)
+		if err != nil {
+			return &kfapisv3.KfError{
+				Code:    int(kfapisv3.INTERNAL_ERROR),
+				Message: fmt.Sprintf("error splitting yaml: %v", err),
+			}
+		}
+		for _, r := range resources {
+
+			err := utils.DeleteResource(r, kubeclient, 5*time.Minute)
+			if err != nil {
+				msg := fmt.Sprintf("error evaluating kustomization manifest for %v Error %v", app.Name, err)
+				errList = append(errList, errors.New(msg))
+				log.Warn(msg)
 			}
 		}
 	}
-	if err := kustomize.deleteGlobalResources(); err != nil {
-		return err
+
+	aggrError := errutil.NewAggregate(errList)
+	if aggrError != nil {
+		return &kfapisv3.KfError{
+			Code:    int(kfapisv3.INTERNAL_ERROR),
+			Message: fmt.Sprintf("error deleting kustomize manifests... Error %v", aggrError),
+		}
 	}
+
+	// Finally, delete the kubeflow namespace
+	// TODO(yanniszark): Remove this once the Kubeflow namespace is created by kustomize manifests
+
 	corev1client, err := corev1.NewForConfig(kustomize.restConfig)
 	if err != nil {
 		return &kfapisv3.KfError{
@@ -399,6 +441,7 @@ func (kustomize *kustomize) Delete(resources kftypesv3.ResourceEnum) error {
 			}
 		}
 	}
+
 	return nil
 }
 

pkg/kfapp/kustomize/kustomize_test.go

@@ -76,12 +76,12 @@ func TestGenerateKustomizationFile(t *testing.T) {
 // TestGenerateYamlWithOwnerReferences
 func TestGenerateYamlWithOwnerReferences(t *testing.T) {
 	type testCase struct {
-		appDir		string
-		expected 	string
+		appDir   string
+		expected string
 	}
-	testCases := []testCase {
+	testCases := []testCase{
 		{
-			appDir: "testdata/operator",
+			appDir:   "testdata/operator",
 			expected: "testdata/operator/expected/service.yaml",
 		},
 	}
@@ -192,7 +192,7 @@ func TestCreateStackAppKustomization(t *testing.T) {
 
 		expectedStr := strings.TrimSpace(string(expected))
 		dataStr := strings.TrimSpace(string(data))
-		
+
 		if diff := cmp.Diff(expectedStr, dataStr); diff != "" {
 			t.Fatalf("kustomization.yaml is different from expected. (-want, +got):\n%s", diff)
 		}

pkg/utils/k8utils.go

@@ -20,15 +20,18 @@ import (
 	"fmt"
 	"github.com/cenkalti/backoff"
 	"github.com/ghodss/yaml"
+	goyaml "github.com/go-yaml/yaml"
 	gogetter "github.com/hashicorp/go-getter"
 	configtypes "github.com/kubeflow/kfctl/v3/config"
 	kfapis "github.com/kubeflow/kfctl/v3/pkg/apis"
 	kftypes "github.com/kubeflow/kfctl/v3/pkg/apis/apps"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	"io"
 	"io/ioutil"
 	"k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	k8stypes "k8s.io/apimachinery/pkg/types"
@@ -520,3 +523,81 @@ func (a *Apply) deleteFlags(usage string) *kubectldelete.DeleteFlags {
 		Output:         &output,
 	}
 }
+
+func DeleteResource(resourceBytes []byte, kubeclient client.Client, timeout time.Duration) error {
+
+	// Convert to unstructured in order to access object metadata
+	resourceMap := map[string]interface{}{}
+	err := yaml.Unmarshal(resourceBytes, &resourceMap)
+	if err != nil {
+		return err
+	}
+	unstructuredObject := &unstructured.Unstructured{
+		Object: resourceMap,
+	}
+	name, namespace := unstructuredObject.GetName(), unstructuredObject.GetNamespace()
+
+	log.Infof("Deleting Kind '%s' in APIVersion '%s' with name '%s' in namespace '%s'",
+		unstructuredObject.GetKind(), unstructuredObject.GetAPIVersion(), name, namespace)
+
+	// Check if resource exists
+	err = kubeclient.Get(context.TODO(), k8stypes.NamespacedName{Name: name, Namespace: namespace}, unstructuredObject)
+	if k8serrors.IsNotFound(err) {
+		log.Warnf("Resource %s/%s not found", namespace, name)
+		return nil
+	}
+	if _, ok := err.(*meta.NoKindMatchError); ok {
+		log.Warnf("No matches for Kind %s in Group %s", unstructuredObject.GetKind(), unstructuredObject.GetAPIVersion())
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	// Resource exists, try to delete
+	if unstructuredObject.GetDeletionTimestamp().IsZero() {
+		err = kubeclient.Delete(context.TODO(), unstructuredObject)
+		if err != nil {
+			return errors.Wrapf(err, "Failed to delete resource %s/%s", namespace, name)
+		}
+	}
+
+	// Delete succeeded, poll until the delete is completed
+	interval := 5 * time.Second
+	b := backoff.WithMaxRetries(backoff.NewConstantBackOff(interval), uint64(timeout/interval+1))
+	err = backoff.Retry(func() error {
+		err := kubeclient.Get(context.TODO(), k8stypes.NamespacedName{Name: name, Namespace: namespace}, unstructuredObject.DeepCopy())
+		if !k8serrors.IsNotFound(err) {
+			return errors.New("deleted resource is not cleaned up yet")
+		}
+		return nil
+	}, b)
+	if err != nil {
+		return errors.New(fmt.Sprintf("Timed out waiting for resource %s/%s to be deleted. Error %v", namespace, name, err))
+	}
+
+	return nil
+}
+
+func SplitYAML(resources []byte) ([][]byte, error) {
+
+	dec := goyaml.NewDecoder(bytes.NewReader(resources))
+
+	var res [][]byte
+	for {
+		var value interface{}
+		err := dec.Decode(&value)
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, err
+		}
+		valueBytes, err := goyaml.Marshal(value)
+		if err != nil {
+			return nil, err
+		}
+		res = append(res, valueBytes)
+	}
+	return res, nil
+}

pkg/utils/k8utils_test.go

@@ -39,3 +39,31 @@ func Test_IsRemoteFile(t *testing.T) {
 		}
 	}
 }
+
+func TestSplitYAML(t *testing.T) {
+	tests := []struct {
+		name     string
+		yaml     []byte
+		expected [][]byte
+	}{
+		{
+			name:     "simple",
+			yaml:     []byte("a: b\n---\nc: d"),
+			expected: [][]byte{[]byte("a: b\n"), []byte("c: d\n")},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resources, err := SplitYAML(test.yaml)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			for idx := range resources {
+				if string(resources[idx]) != string(test.expected[idx]) {
+					t.Fatalf("Resource in place %v. Got '%s', Want '%s'.", idx, resources[idx], test.expected[idx])
+				}
+			}
+		})
+	}
+}