backup_manager.go

// Copyright 2016 The etcd-operator 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 cluster

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"path"
	"time"

	"github.com/coreos/etcd-operator/client/experimentalclient"
	api "github.com/coreos/etcd-operator/pkg/apis/etcd/v1beta2"
	"github.com/coreos/etcd-operator/pkg/backup/backupapi"
	"github.com/coreos/etcd-operator/pkg/cluster/backupstorage"
	"github.com/coreos/etcd-operator/pkg/util/k8sutil"

	"github.com/coreos/etcd-operator/pkg/util/constants"
	"github.com/sirupsen/logrus"
	appsv1beta1 "k8s.io/api/apps/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	defaultBackupHTTPTimeout     = 5 * time.Second
	defaultBackupCreatingTimeout = 1 * time.Minute
)

var (
	errNoPVForBackup       = errors.New("no backup could be created due to PVProvisioner (none) is set")
	errNoS3ConfigForBackup = errors.New("no backup could be created due to S3 configuration not set")
	errNoABSCredsForBackup = errors.New("no backup could be created due to ABS credentials not set")
)

type backupManager struct {
	logger *logrus.Entry

	config  Config
	cluster *api.EtcdCluster
	s       backupstorage.Storage

	bc experimentalclient.Backup
}

func newBackupManager(c Config, cl *api.EtcdCluster, l *logrus.Entry) (*backupManager, error) {
	bm := &backupManager{
		config:  c,
		cluster: cl,
		logger:  l,
		bc:      experimentalclient.NewBackup(&http.Client{}, "http", cl.GetName()),
	}
	var err error
	bm.s, err = bm.setupStorage()
	if err != nil {
		return nil, err
	}
	return bm, nil
}

// setupStorage will only set up the necessary structs in order for backup manager to
// use the storage. It doesn't creates the actual storage here.
func (bm *backupManager) setupStorage() (s backupstorage.Storage, err error) {
	cl, c := bm.cluster, bm.config

	b := cl.Spec.Backup
	switch b.StorageType {
	case api.BackupStorageTypePersistentVolume, api.BackupStorageTypeDefault:
		storageClass := b.PV.StorageClass
		if len(storageClass) == 0 {
			if c.PVProvisioner == constants.PVProvisionerNone {
				return nil, errNoPVForBackup
			}
			storageClass = k8sutil.StorageClassPrefix + "-" + path.Base(c.PVProvisioner)
		}
		s, err = backupstorage.NewPVStorage(c.KubeCli, cl.Name, cl.Namespace, storageClass, *b)
	case api.BackupStorageTypeS3:
		if b.S3 == nil {
			return nil, errNoS3ConfigForBackup
		}
		s, err = backupstorage.NewS3Storage(c.KubeCli, cl.Name, cl.Namespace, *b)
	case api.BackupStorageTypeABS:
		if b.ABS == nil {
			return nil, errNoABSCredsForBackup
		}
		s, err = backupstorage.NewABSStorage(c.KubeCli, cl.Name, cl.Namespace, *b)
	}
	return s, err
}

func (bm *backupManager) setup() error {
	r := bm.cluster.Spec.Restore
	restoreSameNameCluster := r != nil && r.BackupClusterName == bm.cluster.Name

	// There is only one case that we don't need to create underlying storage.
	// That is, the storage already exists and we are restoring cluster from it.
	if !restoreSameNameCluster {
		if err := bm.s.Create(); err != nil {
			return err
		}
	}

	if r != nil {
		bm.logger.Infof("restoring cluster from existing backup (%s)", r.BackupClusterName)
		if bm.cluster.Name != r.BackupClusterName {
			if err := bm.s.Clone(r.BackupClusterName); err != nil {
				return err
			}
		}
	}

	return bm.runSidecar()
}

func (bm *backupManager) runSidecar() error {
	if err := bm.createSidecarDeployment(); err != nil {
		return fmt.Errorf("failed to create backup sidecar Deployment: %v", err)
	}
	if err := bm.createBackupService(); err != nil {
		return fmt.Errorf("failed to create backup sidecar service: %v", err)
	}
	bm.logger.Info("backup sidecar deployment and service created")
	return nil
}

func (bm *backupManager) createSidecarDeployment() error {
	d := bm.makeSidecarDeployment()
	_, err := bm.config.KubeCli.AppsV1beta1().Deployments(bm.cluster.Namespace).Create(d)
	return err
}

func (bm *backupManager) updateSidecar(cl *api.EtcdCluster) error {
	// change local structs
	bm.cluster = cl
	var err error
	bm.s, err = bm.setupStorage()
	if err != nil {
		return err
	}
	ns, n := cl.Namespace, k8sutil.BackupSidecarName(cl.Name)
	// change k8s objects
	uf := func(d *appsv1beta1.Deployment) {
		d.Spec = bm.makeSidecarDeployment().Spec
	}
	return k8sutil.PatchDeployment(bm.config.KubeCli, ns, n, uf)
}

func (bm *backupManager) makeSidecarDeployment() *appsv1beta1.Deployment {
	cl := bm.cluster
	podTemplate := k8sutil.NewBackupPodTemplate(cl.Name, bm.config.ServiceAccount, cl.Spec)
	switch cl.Spec.Backup.StorageType {
	case api.BackupStorageTypeDefault, api.BackupStorageTypePersistentVolume:
		k8sutil.PodSpecWithPV(&podTemplate.Spec, cl.Name)
	case api.BackupStorageTypeS3:
		if ss := cl.Spec.Backup.S3; ss != nil {
			k8sutil.AttachS3ToPodSpec(&podTemplate.Spec, *ss)
		}
	case api.BackupStorageTypeABS:
		if ws := cl.Spec.Backup.ABS; ws != nil {
			k8sutil.AttachABSToPodSpec(&podTemplate.Spec, *ws)
		}
	}
	name := k8sutil.BackupSidecarName(cl.Name)
	dplSel := k8sutil.LabelsForCluster(cl.Name)
	return k8sutil.NewBackupDeploymentManifest(name, dplSel, podTemplate, bm.cluster.AsOwner())
}

func (bm *backupManager) createBackupService() error {
	svc := k8sutil.NewBackupServiceManifest(bm.cluster.Name, bm.cluster.AsOwner())
	_, err := bm.config.KubeCli.CoreV1().Services(bm.cluster.Namespace).Create(svc)
	if err != nil {
		if !k8sutil.IsKubernetesResourceAlreadyExistError(err) {
			return err
		}
	}
	return nil
}

func (bm *backupManager) cleanup() error {
	// Only need to delete storage, Kubernetes related resources will be deleted by the GC.
	err := bm.s.Delete()
	if err != nil {
		return fmt.Errorf("fail to delete backup storage: %v", err)
	}
	return nil
}

func (bm *backupManager) requestBackup() error {
	ctx, cancel := context.WithTimeout(context.Background(), defaultBackupHTTPTimeout+defaultBackupCreatingTimeout)
	defer cancel()
	return bm.bc.Request(ctx)
}

func (bm *backupManager) checkBackupExist(ver string) (bool, error) {
	ctx, cancel := context.WithTimeout(context.Background(), defaultBackupHTTPTimeout)
	defer cancel()
	return bm.bc.Exist(ctx, ver)
}

func (bm *backupManager) getStatus() (*backupapi.ServiceStatus, error) {
	ctx, cancel := context.WithTimeout(context.Background(), defaultBackupHTTPTimeout)
	defer cancel()
	return bm.bc.ServiceStatus(ctx)
}

func backupServiceStatusToTPRBackupServiceStatu(s *backupapi.ServiceStatus) *api.BackupServiceStatus {
	b, err := json.Marshal(s)
	if err != nil {
		panic("unexpected json error")
	}

	var bs api.BackupServiceStatus
	err = json.Unmarshal(b, &bs)
	if err != nil {
		panic("unexpected json error")
	}
	return &bs
}

func (bm *backupManager) upgradeIfNeeded() error {
	ns, n := bm.cluster.Namespace, k8sutil.BackupSidecarName(bm.cluster.Name)

	d, err := bm.config.KubeCli.AppsV1beta1().Deployments(ns).Get(n, metav1.GetOptions{})
	if err != nil {
		return err
	}
	if d.Spec.Template.Spec.Containers[0].Image == k8sutil.BackupImage {
		return nil
	}

	bm.logger.Infof("upgrading backup sidecar from (%v) to (%v)",
		d.Spec.Template.Spec.Containers[0].Image, k8sutil.BackupImage)

	uf := func(d *appsv1beta1.Deployment) {
		d.Spec.Template.Spec.Containers[0].Image = k8sutil.BackupImage
		// TODO: backward compatibility for v0.2.6 . Remove this after v0.2.7 .
		d.Spec.Strategy = appsv1beta1.DeploymentStrategy{
			Type: appsv1beta1.RecreateDeploymentStrategyType,
		}
	}
	return k8sutil.PatchDeployment(bm.config.KubeCli, ns, n, uf)
}

func (bm *backupManager) deleteBackupSidecar() error {
	name, ns := k8sutil.BackupSidecarName(bm.cluster.Name), bm.cluster.Namespace
	err := bm.config.KubeCli.CoreV1().Services(bm.cluster.Namespace).Delete(name, nil)
	if err != nil && !k8sutil.IsKubernetesResourceNotFoundError(err) {
		return fmt.Errorf("backup manager deletion: failed to delete backup service: %v", err)
	}

	err = bm.config.KubeCli.AppsV1beta1().Deployments(ns).Delete(name, k8sutil.CascadeDeleteOptions(0))
	if err != nil && !k8sutil.IsKubernetesResourceNotFoundError(err) {
		return fmt.Errorf("backup manager deletion: failed to delete backup sidecar deployment: %v", err)
	}

	if err := bm.cleanup(); err != nil {
		return fmt.Errorf("backup manager deletion: %v", err)
	}
	return nil
}