Merge pull request #97 from kanisterio/sync

Add DeleteData Kanister function

docs/functions.rst

@@ -344,6 +344,36 @@ Example:
       backupArtifact: s3://bucket/path/artifact
       restorePath: /mnt/data
 
+DeleteData
+----------
+
+This function uses a Kubernetes Job to delete the specified artifact
+from an S3 compatible object store.
+
+.. csv-table::
+   :header: "Argument", "Required", "Type", "Description"
+   :align: left
+   :widths: 5,5,5,15
+
+   `namespace`, No, `string`, namespace in which to execute
+   `artifact`, Yes, `string`, artifact to be deleted from the object store
+
+.. note::
+   The Kubernetes job uses the `kanisterio/kanister-tools` image,
+   since it includes all the tools required to delete the artifact
+   from an S3 compatible object store.
+
+Example:
+
+.. code-block:: yaml
+  :linenos:
+
+  - func: DeleteData
+    name: DeleteFromObjectStore
+    args:
+      namespace: "{{ .Deployment.Namespace }}"
+      artifact: s3://bucket/path/artifact
+
 Registering Functions
 ---------------------
 

examples/time-log/blueprint.yaml

@@ -51,16 +51,8 @@ actions:
     inputArtifactNames:
     - timeLog
     phases:
-    - func: KubeExec
+    - func: DeleteData
       name: deleteFromS3
       args:
         namespace: "{{ .Deployment.Namespace }}"
-        pod: "{{ index .Deployment.Pods 0 }}"
-        container: test-container
-        command:
-          - bash
-          - -c
-          - |
-            AWS_ACCESS_KEY_ID={{ .Profile.Credential.KeyPair.ID | toString }} \
-            AWS_SECRET_ACCESS_KEY={{ .Profile.Credential.KeyPair.Secret | toString }} \
-            aws s3 rm {{ .ArtifactsIn.timeLog.KeyValue.path | quote }}
+        artifact: "{{ .ArtifactsIn.timeLog.KeyValue.path }}"

pkg/function/delete_data.go

@@ -0,0 +1,71 @@
+package function
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	kanister "github.com/kanisterio/kanister/pkg"
+	"github.com/kanisterio/kanister/pkg/param"
+)
+
+const (
+	// DeleteDataNamespaceArg provides the namespace
+	DeleteDataNamespaceArg = "namespace"
+	// DeleteDataArtifactArg provides the path to the artifacts on the object store
+	DeleteDataArtifactArg = "artifact"
+)
+
+func init() {
+	kanister.Register(&deleteDataFunc{})
+}
+
+var _ kanister.Func = (*deleteDataFunc)(nil)
+
+type deleteDataFunc struct{}
+
+func (*deleteDataFunc) Name() string {
+	return "DeleteData"
+}
+
+func generateDeleteCommand(artifact string, profile *param.Profile) []string {
+	// Command to export credentials
+	cmd := []string{"export", fmt.Sprintf("AWS_SECRET_ACCESS_KEY=%s\n", profile.Credential.KeyPair.Secret)}
+	cmd = append(cmd, "export", fmt.Sprintf("AWS_ACCESS_KEY_ID=%s\n", profile.Credential.KeyPair.ID))
+	// Command to delete from the object store
+	cmd = append(cmd, "aws")
+	if profile.Location.S3Compliant.Endpoint != "" {
+		cmd = append(cmd, "--endpoint", profile.Location.S3Compliant.Endpoint)
+	}
+	if profile.SkipSSLVerify {
+		cmd = append(cmd, "--no-verify-ssl")
+	}
+	cmd = append(cmd, "s3", "rm", artifact)
+	command := strings.Join(cmd, " ")
+	return []string{"bash", "-o", "errexit", "-o", "pipefail", "-c", command}
+}
+
+func (*deleteDataFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) error {
+	var artifact, namespace string
+	var err error
+	if err = Arg(args, DeleteDataArtifactArg, &artifact); err != nil {
+		return err
+	}
+	if err = OptArg(args, DeleteDataNamespaceArg, &namespace, ""); err != nil {
+		return err
+	}
+	// Validate the Profile
+	if err = validateProfile(tp.Profile); err != nil {
+		return errors.Wrapf(err, "Failed to validate Profile")
+	}
+	// Generate delete command
+	cmd := generateDeleteCommand(artifact, tp.Profile)
+	// Use KubeTask to delete the artifact
+	return kubeTask(ctx, namespace, "kanisterio/kanister-tools:0.0.1", cmd)
+}
+
+func (*deleteDataFunc) RequiredArgs() []string {
+	return []string{DeleteDataArtifactArg}
+}

pkg/function/kube_task.go

@@ -35,34 +35,24 @@ func generateJobName(jobPrefix string) string {
 	return jobPrefix + jobNameSuffix
 }
 
-func (ktf *kubeTaskFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) error {
-	var namespace, image string
-	var command []string
+func kubeTask(ctx context.Context, namespace, image string, command []string) error {
+	var serviceAccount string
 	var err error
-	if err = Arg(args, KubeTaskNamespaceArg, &namespace); err != nil {
-		return err
-	}
-	if err = Arg(args, KubeTaskImageArg, &image); err != nil {
-		return err
-	}
-	if err = Arg(args, KubeTaskCommandArg, &command); err != nil {
-		return err
-	}
-
-	namespace, err = kube.GetControllerNamespace()
-	if err != nil {
-		return errors.Wrapf(err, "Failed to get controller namespace")
-	}
-
-	jobName := generateJobName(jobPrefix)
 	clientset, err := kube.NewClient()
 	if err != nil {
 		return errors.Wrapf(err, "Failed to create Kubernetes client")
 	}
-	serviceAccount, err := kube.GetControllerServiceAccount(clientset)
-	if err != nil {
-		return errors.Wrap(err, "Failed to get Controller Service Account")
+	if namespace == "" {
+		namespace, err = kube.GetControllerNamespace()
+		if err != nil {
+			return errors.Wrapf(err, "Failed to get controller namespace")
+		}
+		serviceAccount, err = kube.GetControllerServiceAccount(clientset)
+		if err != nil {
+			return errors.Wrap(err, "Failed to get Controller Service Account")
+		}
 	}
+	jobName := generateJobName(jobPrefix)
 	job, err := kube.NewJob(clientset, jobName, namespace, serviceAccount, image, nil, command...)
 	if err != nil {
 		return errors.Wrap(err, "Failed to create job")
@@ -77,6 +67,22 @@ func (ktf *kubeTaskFunc) Exec(ctx context.Context, tp param.TemplateParams, args
 	return nil
 }
 
+func (ktf *kubeTaskFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) error {
+	var namespace, image string
+	var command []string
+	var err error
+	if err = Arg(args, KubeTaskImageArg, &image); err != nil {
+		return err
+	}
+	if err = Arg(args, KubeTaskCommandArg, &command); err != nil {
+		return err
+	}
+	if err = OptArg(args, KubeTaskNamespaceArg, &namespace, ""); err != nil {
+		return err
+	}
+	return kubeTask(ctx, namespace, image, command)
+}
+
 func (*kubeTaskFunc) RequiredArgs() []string {
 	return []string{KubeTaskNamespaceArg, KubeTaskImageArg, KubeTaskCommandArg}
 }

pkg/function/kube_task_test.go

@@ -46,7 +46,7 @@ func (s *KubeTaskSuite) TearDownSuite(c *C) {
 	}
 }
 
-func newTaskBlueprint() *crv1alpha1.Blueprint {
+func newTaskBlueprint(namespace string) *crv1alpha1.Blueprint {
 	return &crv1alpha1.Blueprint{
 		Actions: map[string]*crv1alpha1.BlueprintAction{
 			"test": {
@@ -56,7 +56,7 @@ func newTaskBlueprint() *crv1alpha1.Blueprint {
 						Name: "test",
 						Func: "KubeTask",
 						Args: map[string]interface{}{
-							KubeTaskNamespaceArg: "namespace",
+							KubeTaskNamespaceArg: namespace,
 							KubeTaskImageArg:     "busybox",
 							KubeTaskCommandArg: []string{
 								"sleep",
@@ -68,7 +68,7 @@ func newTaskBlueprint() *crv1alpha1.Blueprint {
 						Name: "test2",
 						Func: "KubeTask",
 						Args: map[string]interface{}{
-							KubeTaskNamespaceArg: "test-namespace",
+							KubeTaskNamespaceArg: namespace,
 							KubeTaskImageArg:     "ubuntu:latest",
 							KubeTaskCommandArg: []string{
 								"sleep",
@@ -91,7 +91,8 @@ func (s *KubeTaskSuite) TestKubeTask(c *C) {
 	}
 
 	action := "test"
-	phases, err := kanister.GetPhases(*newTaskBlueprint(), action, tp)
+	bp := newTaskBlueprint(s.namespace)
+	phases, err := kanister.GetPhases(*bp, action, tp)
 	c.Assert(err, IsNil)
 	for _, p := range phases {
 		err := p.Exec(ctx, tp)