feature: etcd backup

Signed-off-by: Maxim Vasilenko <[email protected]>

cmd/backup.go

@@ -0,0 +1,25 @@
+/*
+Copyright 2024 Flant JSC
+
+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 cmd
+
+import (
+ backup "github.com/deckhouse/deckhouse-cli/internal/backup/cmd"
+)
+
+func init() {
+ rootCmd.AddCommand(backup.NewCommand())
+}

internal/backup/cmd/backup.go

@@ -0,0 +1,43 @@
+/*
+Copyright 2024 Flant JSC
+
+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 backup
+
+import (
+ "github.com/spf13/cobra"
+ "k8s.io/kubectl/pkg/util/templates"
+
+ "github.com/deckhouse/deckhouse-cli/internal/backup/cmd/etcd"
+)
+
+var backupLong = templates.LongDesc(`
+Backup various parts of Deckhouse Kubernetes Platform
+
+© Flant JSC 2024`)
+
+func NewCommand() *cobra.Command {
+ backupCmd := &cobra.Command{
+ Use: "backup",
+ Short: "Backup various parts of Deckhouse Kubernetes Platform",
+ Long: backupLong,
+ }
+
+ backupCmd.AddCommand(
+ etcd.NewCommand(),
+ )
+
+ return backupCmd
+}

internal/backup/cmd/etcd/etcd.go

@@ -0,0 +1,312 @@
+/*
+Copyright 2024 Flant JSC
+
+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 etcd
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "time"
+
+ "github.com/samber/lo"
+ "github.com/spf13/cobra"
+ v1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ "k8s.io/client-go/kubernetes"
+ _ "k8s.io/client-go/plugin/pkg/client/auth"
+ "k8s.io/client-go/rest"
+ "k8s.io/client-go/tools/clientcmd"
+ "k8s.io/client-go/tools/remotecommand"
+ "k8s.io/kubectl/pkg/util/templates"
+)
+
+var etcdLong = templates.LongDesc(`
+Take a snapshot of ETCD state.
+ 
+This command creates a snapshot of the Kubernetes underlying key-value database ETCD.
+
+© Flant JSC 2024`)
+
+func NewCommand() *cobra.Command {
+ etcdCmd := &cobra.Command{
+ Use: "etcd <snapshot-path>",
+ Short: "Take a snapshot of ETCD state",
+ Long: etcdLong,
+ ValidArgs: []string{"snapshot-path"},
+ SilenceErrors: true,
+ SilenceUsage: true,
+ PreRunE: func(cmd *cobra.Command, args []string) error {
+ return validateFlags()
+ },
+ RunE: etcd,
+ }
+
+ addFlags(etcdCmd.Flags())
+ return etcdCmd
+}
+
+const (
+ etcdPodNamespace = "kube-system"
+ etcdPodsLabelSelector = "component=etcd"
+
+ bufferSize16MB = 16 * 1024 * 1024
+)
+
+var (
+ kubeconfigPath string
+ requestedEtcdPodName string
+
+ verboseLog bool
+)
+
+func etcd(_ *cobra.Command, args []string) error {
+ log.SetFlags(log.LstdFlags)
+ if len(args) != 1 {
+ return fmt.Errorf("This command requires exactly 1 argument")
+ }
+
+ config, kubeCl, err := setupK8sClientset(kubeconfigPath)
+ if err != nil {
+ return fmt.Errorf("Failed to setup Kubernetes client: %w", err)
+ }
+
+ etcdPods, err := findETCDPods(kubeCl)
+ if err != nil {
+ return fmt.Errorf("Looking up etcd pods failed: %w", err)
+ }
+
+ pipeExecOpts := &v1.PodExecOptions{
+ Stdout: true,
+ Stderr: true,
+ Container: "etcd",
+ Command: []string{
+ "/usr/bin/etcdctl",
+ "--endpoints", "https://127.0.0.1:2379/",
+ "--key", "/etc/kubernetes/pki/etcd/ca.key",
+ "--cert", "/etc/kubernetes/pki/etcd/ca.crt",
+ "--cacert", "/etc/kubernetes/pki/etcd/ca.crt",
+ "snapshot", "pipe",
+ },
+ }
+
+ if len(etcdPods) > 1 {
+ log.Println(
+ "Will try to snapshot these instances sequentially until one of them succeeds or all of them fail",
+ etcdPods)
+ }
+
+ for _, etcdPodName := range etcdPods {
+ log.Println("Trying to snapshot", etcdPodName)
+
+ snapshotFile, err := os.CreateTemp(".", ".*.snapshotPart")
+ if err != nil {
+ return fmt.Errorf("Failed to prepare temporary etcd snapshot file: %w", err)
+ }
+ defer func(fileName string) {
+ _ = os.Remove(fileName)
+ }(snapshotFile.Name())
+
+ stdout := bufio.NewWriterSize(snapshotFile, bufferSize16MB)
+ stderr := &bytes.Buffer{}
+
+ if err = checkEtcdPodExistsAndReady(kubeCl, etcdPodName); err != nil {
+ log.Printf("%s: Fail, %v\n", etcdPodName, err)
+ continue
+ }
+
+ snapshotStreamingSupported, err := checkEtcdInstanceSupportsSnapshotStreaming(kubeCl, config, etcdPodName)
+ if err != nil {
+ log.Printf("%s: Fail, %v\n", etcdPodName, err)
+ continue
+ }
+ if !snapshotStreamingSupported {
+ log.Printf("%s: etcd instance does not support snapshot streaming\n", etcdPodName)
+ continue
+ }
+
+ if err = streamCommand(kubeCl, config, pipeExecOpts, etcdPodName, etcdPodNamespace, stdout, stderr); err != nil {
+ log.Printf("%s: Fail, %v\n", etcdPodName, err)
+ if verboseLog {
+ log.Println("STDERR:", stderr.String())
+ }
+ continue
+ }
+
+ if err = stdout.Flush(); err != nil {
+ return fmt.Errorf("Flushing snapshot data to disk: %w", err)
+ }
+
+ if err = os.Rename(snapshotFile.Name(), args[0]); err != nil {
+ return fmt.Errorf("Failed to move snapshot file: %w", err)
+ }
+
+ log.Println("Snapshot successfully taken from", etcdPodName)
+ return nil
+ }
+
+ return fmt.Errorf("All known etcd replicas are unavailable to snapshot")
+}
+
+func checkEtcdInstanceSupportsSnapshotStreaming(
+ kubeCl *kubernetes.Clientset,
+ config *rest.Config,
+ etcdPodName string,
+) (bool, error) {
+ helpExecOpts := &v1.PodExecOptions{
+ Stdout: true,
+ Stderr: true,
+ Container: "etcd",
+ Command: []string{
+ "/usr/bin/etcdctl", "help",
+ },
+ }
+
+ stdout, stderr := &bytes.Buffer{}, &bytes.Buffer{}
+ if err := streamCommand(kubeCl, config, helpExecOpts, etcdPodName, etcdPodNamespace, stdout, stderr); err != nil {
+ if verboseLog {
+ log.Println("HELP STDERR:", stderr.String())
+ }
+ return false, fmt.Errorf("streamCommand: %w", err)
+ }
+
+ if bytes.Contains(stdout.Bytes(), []byte("snapshot pipe")) {
+ return true, nil
+ }
+
+ return false, nil
+}
+
+func streamCommand(
+ kubeCl kubernetes.Interface,
+ restConfig *rest.Config,
+ execOpts *v1.PodExecOptions,
+ podName, podNamespace string,
+ stdout, stderr io.Writer,
+) error {
+ scheme := runtime.NewScheme()
+ parameterCodec := runtime.NewParameterCodec(scheme)
+ if err := v1.AddToScheme(scheme); err != nil {
+ return fmt.Errorf("Failed to create parameter codec: %w", err)
+ }
+
+ request := kubeCl.CoreV1().
+ RESTClient().
+ Post().
+ Resource("pods").
+ SubResource("exec").
+ VersionedParams(execOpts, parameterCodec).
+ Namespace(podNamespace).
+ Name(podName)
+
+ executor, err := remotecommand.NewSPDYExecutor(restConfig, "POST", request.URL())
+ if err != nil {
+ log.Printf("Creating SPDY executor for Pod %s: %v", podName, err)
+ }
+
+ if err = executor.StreamWithContext(
+ context.Background(),
+ remotecommand.StreamOptions{
+ Stdout: stdout,
+ Stderr: stderr,
+ }); err != nil {
+ return err
+ }
+
+ return nil
+}
+
+func setupK8sClientset(kubeconfigPath string) (*rest.Config, *kubernetes.Clientset, error) {
+ config, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
+ &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfigPath}, nil).ClientConfig()
+ if err != nil {
+ return nil, nil, fmt.Errorf("Reading kubeconfig file: %w", err)
+ }
+
+ kubeCl, err := kubernetes.NewForConfig(config)
+ if err != nil {
+ return nil, nil, fmt.Errorf("Constructing Kubernetes clientset: %w", err)
+ }
+
+ return config, kubeCl, nil
+}
+
+func findETCDPods(kubeCl kubernetes.Interface) ([]string, error) {
+ if requestedEtcdPodName != "" {
+ if err := checkEtcdPodExistsAndReady(kubeCl, requestedEtcdPodName); err != nil {
+ return nil, err
+ }
+
+ return []string{requestedEtcdPodName}, nil
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ pods, err := kubeCl.CoreV1().Pods(etcdPodNamespace).List(ctx, metav1.ListOptions{
+ LabelSelector: etcdPodsLabelSelector,
+ })
+ if err != nil {
+ return nil, fmt.Errorf("listing etcd Pods: %w", err)
+ }
+
+ pods.Items = lo.Filter(pods.Items, func(pod v1.Pod, _ int) bool {
+ podIsReady := lo.FindOrElse(
+ pod.Status.Conditions, v1.PodCondition{},
+ func(condition v1.PodCondition) bool {
+ return condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue
+ }).Status == v1.ConditionTrue
+
+ _, foundEtcdContainer := lo.Find(pod.Spec.Containers, func(container v1.Container) bool {
+ return container.Name == "etcd"
+ })
+
+ return podIsReady && foundEtcdContainer
+ })
+
+ if len(pods.Items) == 0 {
+ return nil, fmt.Errorf("no valid etcd Pods found")
+ }
+
+ return lo.Map(pods.Items, func(pod v1.Pod, _ int) string {
+ return pod.Name
+ }), nil
+}
+
+func checkEtcdPodExistsAndReady(kubeCl kubernetes.Interface, podName string) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ pod, err := kubeCl.CoreV1().Pods(etcdPodNamespace).Get(ctx, podName, metav1.GetOptions{})
+ if err != nil {
+ return fmt.Errorf("Query Pod %s: %w", podName, err)
+ }
+
+ podReady := lo.FindOrElse(pod.Status.Conditions, v1.PodCondition{}, func(condition v1.PodCondition) bool {
+ return condition.Type == v1.PodReady
+ }).Status == v1.ConditionTrue
+
+ if !podReady {
+ return fmt.Errorf("Pod %s is not yet ready, cannot snapshot it now", podName)
+ }
+
+ return nil
+}