Skip completed jobs and pods when restoring

Completed jobs and pods may be useful in the backup for auditing
purposes, but don't recreate them when restoring.

Signed-off-by: Nolan Brubaker <nolan@heptio.com>

pkg/backup/item_backupper.go

@@ -162,9 +162,6 @@ func (ib *defaultItemBackupper) backupItem(logger logrus.FieldLogger, obj runtim
 
 	log.Info("Backing up resource")
 
-	// Never save status
-	delete(obj.UnstructuredContent(), "status")
-
 	log.Debug("Executing pre hooks")
 	if err := ib.itemHookHandler.handleHooks(log, groupResource, obj, ib.resourceHooks, hookPhasePre); err != nil {
 		return err

pkg/backup/item_backupper_test.go

@@ -163,13 +163,6 @@ func TestBackupItemNoSkips(t *testing.T) {
 			expectExcluded:        false,
 			expectedTarHeaderName: "resources/resource.group/cluster/bar.json",
 		},
-		{
-			name:                  "make sure status is deleted",
-			item:                  `{"metadata":{"name":"bar"},"spec":{"color":"green"},"status":{"foo":"bar"}}`,
-			expectError:           false,
-			expectExcluded:        false,
-			expectedTarHeaderName: "resources/resource.group/cluster/bar.json",
-		},
 		{
 			name:                "tar header write error",
 			item:                `{"metadata":{"name":"bar"},"spec":{"color":"green"},"status":{"foo":"bar"}}`,
@@ -376,17 +369,14 @@ func TestBackupItemNoSkips(t *testing.T) {
 				return
 			}
 
-			// we have to delete status as that's what backupItem does,
-			// and this ensures that we're verifying the right data
-			delete(item, "status")
-			itemWithoutStatus, err := json.Marshal(&item)
+			// Convert to JSON for comparing number of bytes to the tar header
+			itemJSON, err := json.Marshal(&item)
 			if err != nil {
 				t.Fatal(err)
 			}
-
 			require.Equal(t, 1, len(w.headers), "headers")
 			assert.Equal(t, test.expectedTarHeaderName, w.headers[0].Name, "header.name")
-			assert.Equal(t, int64(len(itemWithoutStatus)), w.headers[0].Size, "header.size")
+			assert.Equal(t, int64(len(itemJSON)), w.headers[0].Size, "header.size")
 			assert.Equal(t, byte(tar.TypeReg), w.headers[0].Typeflag, "header.typeflag")
 			assert.Equal(t, int64(0755), w.headers[0].Mode, "header.mode")
 			assert.False(t, w.headers[0].ModTime.IsZero(), "header.modTime set")

pkg/restore/restore.go

@@ -559,6 +559,16 @@ func (ctx *context) restoreResource(resource, namespace, resourcePath string) (a
 			continue
 		}
 
+		complete, err := isCompleted(obj, groupResource)
+		if err != nil {
+			addToResult(&errs, namespace, fmt.Errorf("error checking completion %q: %v", fullPath, err))
+			continue
+		}
+		if complete {
+			ctx.infof("%s is complete - skipping", kube.NamespaceAndName(obj))
+			continue
+		}
+
 		if resourceClient == nil {
 			// initialize client for this Resource. we need
 			// metadata from an object to do this.
@@ -782,8 +792,7 @@ func resetMetadataAndStatus(obj *unstructured.Unstructured) (*unstructured.Unstr
 		}
 	}
 
-	// this should never be backed up anyway, but remove it just
-	// in case.
+	// Never restore status
 	delete(obj.UnstructuredContent(), "status")
 
 	return obj, nil
@@ -814,6 +823,37 @@ func hasControllerOwner(refs []metav1.OwnerReference) bool {
 	return false
 }
 
+// TODO(0.9): These are copied from backup/item_backupper. Definititions should be moved
+// to a shared package and imported here and in the backup package.
+var podsGroupResource = schema.GroupResource{Group: "", Resource: "pods"}
+var jobsGroupResource = schema.GroupResource{Group: "batch", Resource: "jobs"}
+
+// isCompleted returns whether or not an object is considered completed.
+// Used to identify whether or not an object should be restored. Only Jobs or Pods are considered
+func isCompleted(obj *unstructured.Unstructured, groupResource schema.GroupResource) (bool, error) {
+	switch groupResource {
+	case podsGroupResource:
+		phase, _, err := unstructured.NestedString(obj.UnstructuredContent(), "status", "phase")
+		if err != nil {
+			return false, errors.WithStack(err)
+		}
+		if phase == string(v1.PodFailed) || phase == string(v1.PodSucceeded) {
+			return true, nil
+		}
+
+	case jobsGroupResource:
+		ct, found, err := unstructured.NestedString(obj.UnstructuredContent(), "status", "completionTime")
+		if err != nil {
+			return false, errors.WithStack(err)
+		}
+		if found && ct != "" {
+			return true, nil
+		}
+	}
+	// Assume any other resource isn't complete and can be restored
+	return false, nil
+}
+
 // unmarshal reads the specified file, unmarshals the JSON contained within it
 // and returns an Unstructured object.
 func (ctx *context) unmarshal(filePath string) (*unstructured.Unstructured, error) {

pkg/restore/restore_test.go

@@ -18,6 +18,7 @@ package restore
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"os"
 	"testing"
@@ -676,6 +677,75 @@ func TestResetMetadataAndStatus(t *testing.T) {
 	}
 }
 
+func TestIsCompleted(t *testing.T) {
+	tests := []struct {
+		name          string
+		expected      bool
+		content       string
+		groupResource schema.GroupResource
+		expectedErr   bool
+	}{
+		{
+			name:          "Failed pods are complete",
+			expected:      true,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"phase": "Failed"}}`,
+			groupResource: schema.GroupResource{Group: "", Resource: "pods"},
+		},
+		{
+			name:          "Succeeded pods are complete",
+			expected:      true,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"phase": "Succeeded"}}`,
+			groupResource: schema.GroupResource{Group: "", Resource: "pods"},
+		},
+		{
+			name:          "Pending pods aren't complete",
+			expected:      false,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"phase": "Pending"}}`,
+			groupResource: schema.GroupResource{Group: "", Resource: "pods"},
+		},
+		{
+			name:          "Running pods aren't complete",
+			expected:      false,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"phase": "Running"}}`,
+			groupResource: schema.GroupResource{Group: "", Resource: "pods"},
+		},
+		{
+			name:          "Jobs without a completion time aren't complete",
+			expected:      false,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}}`,
+			groupResource: schema.GroupResource{Group: "batch", Resource: "jobs"},
+		},
+		{
+			name:          "Jobs with a completion time are completed",
+			expected:      true,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"completionTime": "bar"}}`,
+			groupResource: schema.GroupResource{Group: "batch", Resource: "jobs"},
+		},
+		{
+			name:          "Jobs with an empty completion time are not completed",
+			expected:      false,
+			content:       `{"apiVersion":"v1","kind":"Pod","metadata":{"namespace":"ns","name":"pod1"}, "status": {"completionTime": ""}}`,
+			groupResource: schema.GroupResource{Group: "batch", Resource: "jobs"},
+		},
+		{
+			name:          "Something not a pod or a job may actually be complete, but we're not concerned with that",
+			expected:      false,
+			content:       `{"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "ns"}, "status": {"completionTime": "bar", "phase":"Completed"}}`,
+			groupResource: schema.GroupResource{Group: "", Resource: "namespaces"},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			u := unstructuredOrDie(fmt.Sprintf(test.content))
+			backup, err := isCompleted(u, test.groupResource)
+
+			if assert.Equal(t, test.expectedErr, err != nil) {
+				assert.Equal(t, test.expected, backup)
+			}
+		})
+	}
+}
+
 func TestObjectsAreEqual(t *testing.T) {
 	tests := []struct {
 		name        string

pkg/util/kube/groupresource.go

@@ -0,0 +1,25 @@
+/*
+Copyright 2018 the Heptio Ark 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 kube
+
+import (
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+var PodsGroupResource = schema.GroupResource{Group: "", Resource: "pods"}
+var JobsGroupResource = schema.GroupResource{Group: "batch", Resource: "jobs"}
+var NamespacesGroupResource = schema.GroupResource{Group: "", Resource: "namespaces"}