validate pod volumes hostpath mount on restic server startup (#1616)

Signed-off-by: Adnan Abdulhussein <aadnan@vmware.com>

changelogs/unreleased/1616-prydonius

@@ -0,0 +1 @@
+adds validation for pod volumes hostPath mount on restic server startup

pkg/cmd/cli/restic/server.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2018 the Velero contributors.
+Copyright 2019 the Velero contributors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -25,7 +25,9 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	kubeinformers "k8s.io/client-go/informers"
 	corev1informers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -39,6 +41,7 @@ import (
 	clientset "github.com/heptio/velero/pkg/generated/clientset/versioned"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions"
 	"github.com/heptio/velero/pkg/restic"
+	"github.com/heptio/velero/pkg/util/filesystem"
 	"github.com/heptio/velero/pkg/util/logging"
 )
 
@@ -79,6 +82,7 @@ type resticServer struct {
 	logger                logrus.FieldLogger
 	ctx                   context.Context
 	cancelFunc            context.CancelFunc
+	fileSystem            filesystem.Interface
 }
 
 func newResticServer(logger logrus.FieldLogger, baseName string) (*resticServer, error) {
@@ -127,7 +131,7 @@ func newResticServer(logger logrus.FieldLogger, baseName string) (*resticServer,
 
 	ctx, cancelFunc := context.WithCancel(context.Background())
 
-	return &resticServer{
+	s := &resticServer{
 		kubeClient:            kubeClient,
 		veleroClient:          veleroClient,
 		veleroInformerFactory: informers.NewFilteredSharedInformerFactory(veleroClient, 0, os.Getenv("VELERO_NAMESPACE"), nil),
@@ -137,7 +141,14 @@ func newResticServer(logger logrus.FieldLogger, baseName string) (*resticServer,
 		logger:                logger,
 		ctx:                   ctx,
 		cancelFunc:            cancelFunc,
-	}, nil
+		fileSystem:            filesystem.NewFileSystem(),
+	}
+
+	if err := s.validatePodVolumesHostPath(); err != nil {
+		return nil, err
+	}
+
+	return s, nil
 }
 
 func (s *resticServer) run() {
@@ -191,3 +202,50 @@ func (s *resticServer) run() {
 	s.logger.Info("Waiting for all controllers to shut down gracefully")
 	wg.Wait()
 }
+
+// validatePodVolumesHostPath validates that the pod volumes path contains a
+// directory for each Pod running on this node
+func (s *resticServer) validatePodVolumesHostPath() error {
+	files, err := s.fileSystem.ReadDir("/host_pods/")
+	if err != nil {
+		return errors.Wrap(err, "could not read pod volumes host path")
+	}
+
+	// create a map of directory names inside the pod volumes path
+	dirs := sets.NewString()
+	for _, f := range files {
+		if f.IsDir() {
+			dirs.Insert(f.Name())
+		}
+	}
+
+	pods, err := s.kubeClient.CoreV1().Pods("").List(metav1.ListOptions{FieldSelector: fmt.Sprintf("spec.nodeName=%s,status.phase=Running", os.Getenv("NODE_NAME"))})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	valid := true
+	for _, pod := range pods.Items {
+		dirName := string(pod.GetUID())
+
+		// if the pod is a mirror pod, the directory name is the hash value of the
+		// mirror pod annotation
+		if hash, ok := pod.GetAnnotations()[v1.MirrorPodAnnotationKey]; ok {
+			dirName = hash
+		}
+
+		if !dirs.Has(dirName) {
+			valid = false
+			s.logger.WithFields(logrus.Fields{
+				"pod":  fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName()),
+				"path": "/host_pods/" + dirName,
+			}).Debug("could not find volumes for pod in host path")
+		}
+	}
+
+	if !valid {
+		return errors.New("unexpected directory structure for host-pods volume, ensure that the host-pods volume corresponds to the pods subdirectory of the kubelet root directory")
+	}
+
+	return nil
+}

pkg/cmd/cli/restic/server_test.go

@@ -0,0 +1,109 @@
+/*
+Copyright 2019 the Velero contributors.
+
+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 restic
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	"github.com/heptio/velero/pkg/test"
+	testutil "github.com/heptio/velero/pkg/util/test"
+)
+
+func Test_validatePodVolumesHostPath(t *testing.T) {
+	tests := []struct {
+		name    string
+		pods    []*corev1.Pod
+		dirs    []string
+		wantErr bool
+	}{
+		{
+			name: "no error when pod volumes are present",
+			pods: []*corev1.Pod{
+				test.NewPod("foo", "bar", test.WithUID("foo")),
+				test.NewPod("zoo", "raz", test.WithUID("zoo")),
+			},
+			dirs:    []string{"foo", "zoo"},
+			wantErr: false,
+		},
+		{
+			name: "no error when pod volumes are present and there are mirror pods",
+			pods: []*corev1.Pod{
+				test.NewPod("foo", "bar", test.WithUID("foo")),
+				test.NewPod("zoo", "raz", test.WithUID("zoo"), test.WithAnnotations(v1.MirrorPodAnnotationKey, "baz")),
+			},
+			dirs:    []string{"foo", "baz"},
+			wantErr: false,
+		},
+		{
+			name: "error when all pod volumes missing",
+			pods: []*corev1.Pod{
+				test.NewPod("foo", "bar", test.WithUID("foo")),
+				test.NewPod("zoo", "raz", test.WithUID("zoo")),
+			},
+			dirs:    []string{"unexpected-dir"},
+			wantErr: true,
+		},
+		{
+			name: "error when some pod volumes missing",
+			pods: []*corev1.Pod{
+				test.NewPod("foo", "bar", test.WithUID("foo")),
+				test.NewPod("zoo", "raz", test.WithUID("zoo")),
+			},
+			dirs:    []string{"foo"},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fs := testutil.NewFakeFileSystem()
+
+			for _, dir := range tt.dirs {
+				err := fs.MkdirAll(filepath.Join("/host_pods/", dir), os.ModePerm)
+				if err != nil {
+					t.Error(err)
+				}
+			}
+
+			kubeClient := fake.NewSimpleClientset()
+			for _, pod := range tt.pods {
+				_, err := kubeClient.CoreV1().Pods(pod.GetNamespace()).Create(pod)
+				if err != nil {
+					t.Error(err)
+				}
+			}
+
+			s := &resticServer{
+				kubeClient: kubeClient,
+				logger:     testutil.NewLogger(),
+				fileSystem: fs,
+			}
+
+			err := s.validatePodVolumesHostPath()
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

pkg/test/resources.go

@@ -23,6 +23,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 // APIResource stores information about a specific Kubernetes API
@@ -321,6 +322,13 @@ func WithDeletionTimestamp(val time.Time) func(obj metav1.Object) {
 	}
 }
 
+// WithUID is a functional option that applies the specified UID to an object.
+func WithUID(val string) func(obj metav1.Object) {
+	return func(obj metav1.Object) {
+		obj.SetUID(types.UID(val))
+	}
+}
+
 // WithReclaimPolicy is a functional option for persistent volumes that sets
 // the specified reclaim policy. It panics if the object is not a persistent
 // volume.