Add Function BackupDataAllUsingKopiaServer

docs/functions.rst

@@ -1758,6 +1758,64 @@ For this phase, we will use the ``backupIdentifier`` Artifact provided by backup
       backupID: "{{ .ArtifactsIn.backupIdentifier.KeyValue.id }}"
       image: ghcr.io/kanisterio/kanister-tools:0.89.0
 
+.. _backupdataallusingkopiaserver:
+
+BackupDataAllUsingKopiaServer
+-----------------------------
+
+This function concurrently backs up data from one or more pods into an
+object store supported by Kanister using Kopia Repository Server as a data mover.
+
+.. note::
+   It is important that the application includes a ``kanister-tools``
+   sidecar container. This sidecar is necessary to run the
+   tools that backup the volume and store it on the object store.
+
+   Additionally, in order to use this function, a RepositoryServer CR is needed
+   while creating the :ref:`actionsets`
+
+Arguments:
+
+.. csv-table::
+   :header: "Argument", "Required", "Type", "Description"
+   :align: left
+   :widths: 5,5,5,15
+
+   `namespace`, Yes, `string`, namespace of the container that you want to backup the data of
+   `pods`, No, `string`, pods in which you want to backup the data of (by default runs on all the pods)
+   `container`, Yes, `string`, name of the kanister sidecar container
+   `includePath`, Yes, `string`, path of the data to be backed up
+   `repositoryServerUserHostname`, No, `string`, user's hostname to access the kopia repository server. Hostname would be available in the user access credential secret
+
+Outputs:
+
+.. csv-table::
+   :header: "Output", "Type", "Description"
+   :align: left
+   :widths: 5,5,15
+
+   `BackupAllUsingKopiaServerInfo`,`string`, info about snapshot id required for restore
+
+Example:
+
+.. code-block:: yaml
+  :linenos:
+
+  actions:
+    backup-all:
+    outputArtifacts:
+      backupIdentifier:
+        keyValue:
+          backupInfo: "{{ .Phases.backupToS3All.Output.BackupAllUsingKopiaServerInfo }}"
+    phases:
+      - func: BackupDataAllUsingKopiaServer
+        name: backupToS3All
+        args:
+          namespace: "{{ .Deployment.Namespace }}"
+          pods: "{{ index .Deployment.Pods 0 }}"
+          container: kanister-sidecar
+          includePath: /var/log
+
 Registering Functions
 ---------------------
 

pkg/function/backup_data_all_using_kopia_server.go

@@ -0,0 +1,167 @@
+// Copyright 2023 The Kanister Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+
+	"github.com/pkg/errors"
+	"k8s.io/client-go/kubernetes"
+
+	kanister "github.com/kanisterio/kanister/pkg"
+	"github.com/kanisterio/kanister/pkg/consts"
+	"github.com/kanisterio/kanister/pkg/field"
+	kankopia "github.com/kanisterio/kanister/pkg/kopia"
+	"github.com/kanisterio/kanister/pkg/kube"
+	"github.com/kanisterio/kanister/pkg/param"
+)
+
+const (
+	// BackupDataAllUsingKopiaServerFuncName gives the name of the function
+	BackupDataAllUsingKopiaServerFuncName = "BackupDataAllUsingKopiaServer"
+	BackupDataAllUsingKopiaServerOutput   = "BackupAllUsingKopiaServerInfo"
+)
+
+type backupDataAllUsingKopiaServerFunc struct{}
+
+func init() {
+	_ = kanister.Register(&backupDataAllUsingKopiaServerFunc{})
+}
+
+func (*backupDataAllUsingKopiaServerFunc) Name() string {
+	return BackupDataAllUsingKopiaServerFuncName
+}
+
+func (*backupDataAllUsingKopiaServerFunc) RequiredArgs() []string {
+	return []string{
+		BackupDataAllNamespaceArg,
+		BackupDataAllContainerArg,
+		BackupDataAllIncludePathArg,
+	}
+}
+
+func (*backupDataAllUsingKopiaServerFunc) Arguments() []string {
+	return []string{
+		BackupDataAllNamespaceArg,
+		BackupDataAllContainerArg,
+		BackupDataAllIncludePathArg,
+		BackupDataAllPodsArg,
+		KopiaRepositoryServerUserHostname,
+	}
+}
+
+func (*backupDataAllUsingKopiaServerFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
+	var namespace, pods, container, includePath, userHostname string
+	var err error
+	if err = Arg(args, BackupDataAllNamespaceArg, &namespace); err != nil {
+		return nil, err
+	}
+	if err = Arg(args, BackupDataAllContainerArg, &container); err != nil {
+		return nil, err
+	}
+	if err = Arg(args, BackupDataAllIncludePathArg, &includePath); err != nil {
+		return nil, err
+	}
+	if err = OptArg(args, BackupDataAllPodsArg, &pods, ""); err != nil {
+		return nil, err
+	}
+	if err = OptArg(args, KopiaRepositoryServerUserHostname, &userHostname, ""); err != nil {
+		return nil, err
+	}
+
+	cli, err := kube.NewClient()
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to create Kubernetes client")
+	}
+	var ps []string
+	if pods == "" {
+		switch {
+		case tp.Deployment != nil:
+			ps = tp.Deployment.Pods
+		case tp.StatefulSet != nil:
+			ps = tp.StatefulSet.Pods
+		default:
+			return nil, errors.New("Failed to get pods")
+		}
+	} else {
+		ps = strings.Fields(pods)
+	}
+	ctx = field.Context(ctx, consts.ContainerNameKey, container)
+	userPassphrase, cert, err := userCredentialsAndServerTLS(&tp)
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to fetch User Credentials/Certificate Data from Template Params")
+	}
+
+	fingerprint, err := kankopia.ExtractFingerprintFromCertificateJSON(cert)
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to fetch Kopia API Server Certificate Secret Data from Certificate")
+	}
+
+	hostname, userAccessPassphrase, err := hostNameAndUserPassPhraseFromRepoServer(userPassphrase, userHostname)
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to fetch Hostname/User Passphrase from Secret")
+	}
+	return backupDataAllUsingKopiaServer(cli, ctx, namespace, container, includePath, hostname, tp.RepositoryServer.Address, fingerprint, tp.RepositoryServer.Username, userAccessPassphrase, ps)
+}
+
+func backupDataAllUsingKopiaServer(
+	cli kubernetes.Interface,
+	ctx context.Context,
+	namespace,
+	container,
+	includePath,
+	hostname,
+	serverAddress,
+	fingerprint,
+	username,
+	userPassphrase string,
+	ps []string,
+) (map[string]interface{}, error) {
+	errChan := make(chan error, len(ps))
+	outChan := make(chan BackupInfo, len(ps))
+	Output := make(map[string]BackupInfo)
+
+	for _, pod := range ps {
+		go func(pod string, container string) {
+			ctx = field.Context(ctx, consts.PodNameKey, pod)
+			backupOutputs, err := backupDataUsingKopiaServer(cli, container, hostname, includePath, namespace, pod, serverAddress, fingerprint, username, userPassphrase, nil)
+			errChan <- errors.Wrapf(err, "Failed to backup data for pod %s", pod)
+			outChan <- BackupInfo{PodName: pod, BackupID: backupOutputs.SnapshotID}
+		}(pod, container)
+	}
+	errs := make([]string, 0, len(ps))
+	for i := 0; i < len(ps); i++ {
+		err := <-errChan
+		output := <-outChan
+		if err != nil {
+			errs = append(errs, err.Error())
+		} else {
+			Output[output.PodName] = output
+		}
+	}
+	if len(errs) != 0 {
+		return nil, errors.New(strings.Join(errs, "\n"))
+	}
+	manifestData, err := json.Marshal(Output)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to encode JSON data")
+	}
+	return map[string]interface{}{
+		BackupDataAllUsingKopiaServerOutput: string(manifestData),
+		FunctionOutputVersion:               kanister.DefaultVersion,
+	}, nil
+}