checkpointer: factor out checkpoint lib.

The monolithic main.go file was becoming too unwieldy to keep working
on. This is a straightforward refactor: I moved functions into logical
files and made the main() as lightweight as possible.

Author: diegs

cmd/checkpoint/main.go

@@ -0,0 +1,23 @@
+package checkpoint
+
+import (
+	"github.com/golang/glog"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/pkg/api"
+	"k8s.io/client-go/pkg/api/v1"
+)
+
+// getAPIParentPods will retrieve all pods from apiserver that are parents & should be checkpointed
+func (c *checkpointer) getAPIParentPods(nodeName string) map[string]*v1.Pod {
+	opts := metav1.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector(api.PodHostField, nodeName).String(),
+	}
+
+	podList, err := c.apiserver.CoreV1().Pods(api.NamespaceAll).List(opts)
+	if err != nil {
+		glog.Warningf("Unable to contact APIServer, skipping garbage collection: %v", err)
+		return nil
+	}
+	return podListToParentPods(podList)
+}

cmd/checkpoint/main_test.go

@@ -0,0 +1,125 @@
+// Package checkpoint provides libraries that are used by the pod-checkpointer utility to checkpoint
+// pods on a node. See cmd/checkpoint/README.md in this repository for more information.
+package checkpoint
+
+import (
+	"fmt"
+	"time"
+
+	"k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
+)
+
+const (
+	activeCheckpointPath    = "/etc/kubernetes/manifests"
+	inactiveCheckpointPath  = "/etc/kubernetes/inactive-manifests"
+	checkpointSecretPath    = "/etc/kubernetes/checkpoint-secrets"
+	checkpointConfigMapPath = "/etc/kubernetes/checkpoint-configmaps"
+
+	shouldCheckpointAnnotation = "checkpointer.alpha.coreos.com/checkpoint"    // = "true"
+	checkpointParentAnnotation = "checkpointer.alpha.coreos.com/checkpoint-of" // = "podName"
+	podSourceAnnotation        = "kubernetes.io/config.source"
+
+	shouldCheckpoint = "true"
+	podSourceFile    = "file"
+
+	defaultPollingFrequency  = 3 * time.Second
+	defaultCheckpointTimeout = 1 * time.Minute
+)
+
+var (
+	lastCheckpoint time.Time
+)
+
+// Options defines the parameters that are required to start the checkpointer.
+type Options struct {
+	// CheckpointerPod holds information about this checkpointer pod.
+	CheckpointerPod CheckpointerPod
+	// KubeConfig is a valid kubeconfig for communicating with the APIServer.
+	KubeConfig *restclient.Config
+	// RemoteRuntimeEndpoint is the location of the CRI GRPC endpoint.
+	RemoteRuntimeEndpoint string
+	// RuntimeRequestTimeout is the timeout that is used for requests to the RemoteRuntimeEndpoint.
+	RuntimeRequestTimeout time.Duration
+}
+
+// CheckpointerPod holds information about this checkpointer pod.
+type CheckpointerPod struct {
+	// The name of the node this checkpointer is running on.
+	NodeName string
+	// The name of the pod that is running this checkpointer.
+	PodName string
+	// The namespace of the pod that is running this checkpointer.
+	PodNamespace string
+}
+
+// checkpointer holds state used by the checkpointer to perform its duties.
+type checkpointer struct {
+	apiserver       kubernetes.Interface
+	kubelet         *kubeletClient
+	cri             *remoteRuntimeService
+	checkpointerPod CheckpointerPod
+}
+
+// Run instantiates and starts a new checkpointer. Returns error if there was a problem creating
+// the checkpointer, otherwise never returns.
+func Run(opts Options) error {
+	apiserver := kubernetes.NewForConfigOrDie(opts.KubeConfig)
+
+	kubelet, err := newKubeletClient(opts.KubeConfig)
+	if err != nil {
+		return fmt.Errorf("failed to load kubelet client: %v", err)
+	}
+
+	// Open a GRPC connection to the CRI shim
+	cri, err := newRemoteRuntimeService(opts.RemoteRuntimeEndpoint, opts.RuntimeRequestTimeout)
+	if err != nil {
+		return fmt.Errorf("failed to connect to CRI server: %v", err)
+	}
+
+	cp := &checkpointer{
+		apiserver:       apiserver,
+		kubelet:         kubelet,
+		cri:             cri,
+		checkpointerPod: opts.CheckpointerPod,
+	}
+	cp.run()
+
+	return nil
+}
+
+// run is the main checkpointing loop.
+func (c *checkpointer) run() {
+	for {
+		time.Sleep(defaultPollingFrequency)
+
+		// We must use both the :10255/pods endpoint and CRI shim, because /pods
+		// endpoint could have stale data. The /pods endpoint will only show the last cached
+		// status which has successfully been written to an apiserver. However, if there is
+		// no apiserver, we may get stale state (e.g. saying pod is running, when it really is
+		// not).
+		localParentPods := c.kubelet.localParentPods()
+		localRunningPods := c.cri.localRunningPods()
+
+		c.createCheckpointsForValidParents(localParentPods)
+
+		// Try to get scheduled pods from the apiserver.
+		// These will be used to GC checkpoints for parents no longer scheduled to this node.
+		// A return value of nil is assumed to be "could not contact apiserver"
+		// TODO(aaron): only check this every 30 seconds or so
+		apiParentPods := c.getAPIParentPods(c.checkpointerPod.NodeName)
+
+		// Get on disk copies of (in)active checkpoints
+		//TODO(aaron): Could be racy to load from disk each time, but much easier than trying to keep in-memory state in sync.
+		activeCheckpoints := getFileCheckpoints(activeCheckpointPath)
+		inactiveCheckpoints := getFileCheckpoints(inactiveCheckpointPath)
+
+		start, stop, remove := process(localRunningPods, localParentPods, apiParentPods, activeCheckpoints, inactiveCheckpoints, c.checkpointerPod)
+
+		// Handle remove at last because we may still have some work to do
+		// before removing the checkpointer itself.
+		handleStop(stop)
+		handleStart(start)
+		handleRemove(remove)
+	}
+}

pkg/checkpoint/apiserver.go

@@ -0,0 +1,58 @@
+package checkpoint
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/pkg/api/v1"
+)
+
+// checkpointConfigMapVolumes ensures that all pod configMaps are checkpointed locally, then converts the configMap volume to a hostpath.
+func (c *checkpointer) checkpointConfigMapVolumes(pod *v1.Pod) (*v1.Pod, error) {
+	for i := range pod.Spec.Volumes {
+		v := &pod.Spec.Volumes[i]
+		if v.ConfigMap == nil {
+			continue
+		}
+
+		_, err := c.checkpointConfigMap(pod.Namespace, pod.Name, v.ConfigMap.Name)
+		if err != nil {
+			return nil, fmt.Errorf("failed to checkpoint configMap for pod %s/%s: %v", pod.Namespace, pod.Name, err)
+		}
+	}
+	return pod, nil
+}
+
+// checkpointConfigMap will locally store configMap data.
+// The path to the configMap data becomes: checkpointConfigMapPath/namespace/podname/configMapName/configMap.file
+// Where each "configMap.file" is a key from the configMap.Data field.
+func (c *checkpointer) checkpointConfigMap(namespace, podName, configMapName string) (string, error) {
+	configMap, err := c.apiserver.Core().ConfigMaps(namespace).Get(configMapName, metav1.GetOptions{})
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve configMap %s/%s: %v", namespace, configMapName, err)
+	}
+
+	basePath := configMapPath(namespace, podName, configMapName)
+	if err := os.MkdirAll(basePath, 0700); err != nil {
+		return "", fmt.Errorf("failed to create configMap checkpoint path %s: %v", basePath, err)
+	}
+	// TODO(aaron): No need to store if already exists
+	for f, d := range configMap.Data {
+		if err := writeAndAtomicRename(filepath.Join(basePath, f), []byte(d), 0600); err != nil {
+			return "", fmt.Errorf("failed to write configMap %s: %v", configMap.Name, err)
+		}
+	}
+	return basePath, nil
+}
+
+func configMapPath(namespace, podName, configMapName string) string {
+	return filepath.Join(checkpointConfigMapPath, namespace, podName, configMapName)
+}
+
+func podFullNameToConfigMapPath(id string) string {
+	namespace, podname := path.Split(id)
+	return filepath.Join(checkpointConfigMapPath, namespace, podname)
+}

pkg/checkpoint/checkpoint.go

@@ -0,0 +1,57 @@
+package checkpoint
+
+import (
+	"fmt"
+
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/pkg/api/v1"
+	"k8s.io/client-go/rest"
+)
+
+// A minimal kubelet client. It assumes the kubelet can be reached the kubelet's insecure API at
+// 127.0.0.1:10255 and the secure API at 127.0.0.1:10250.
+type kubeletClient struct {
+	insecureClient *rest.RESTClient
+	secureClient   *rest.RESTClient
+}
+
+func newKubeletClient(config *rest.Config) (*kubeletClient, error) {
+	// Use the core API group serializer. Same logic as client-go.
+	// https://github.com/kubernetes/client-go/blob/v3.0.0/kubernetes/typed/core/v1/core_client.go#L147
+	config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: scheme.Codecs}
+
+	// Kubelet is using a self-signed cert.
+	config.TLSClientConfig.Insecure = true
+	config.TLSClientConfig.CAFile = ""
+	config.TLSClientConfig.CAData = nil
+
+	// Shallow copy.
+	insecureConfig := *config
+	secureConfig := *config
+
+	insecureConfig.Host = "http://127.0.0.1:10255"
+	secureConfig.Host = "https://127.0.0.1:10250"
+
+	client := new(kubeletClient)
+	var err error
+	if client.insecureClient, err = rest.UnversionedRESTClientFor(&insecureConfig); err != nil {
+		return nil, fmt.Errorf("failed creating kubelet client for debug endpoints: %v", err)
+	}
+	if client.secureClient, err = rest.UnversionedRESTClientFor(&secureConfig); err != nil {
+		return nil, fmt.Errorf("failed creating kubelet client: %v", err)
+	}
+
+	return client, nil
+}
+
+// localParentPods will retrieve all pods from kubelet api that are parents & should be checkpointed
+func (k *kubeletClient) localParentPods() map[string]*v1.Pod {
+	podList := new(v1.PodList)
+	if err := k.insecureClient.Get().AbsPath("/pods/").Do().Into(podList); err != nil {
+		// Assume there are no local parent pods.
+		glog.Errorf("failed to list local parent pods, assuming none are running: %v", err)
+	}
+	return podListToParentPods(podList)
+}

pkg/checkpoint/config_map.go

@@ -0,0 +1,75 @@
+package checkpoint
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/pkg/api"
+	"k8s.io/client-go/pkg/api/v1"
+)
+
+// getFileCheckpoints will retrieve all checkpoint manifests from a given filepath.
+func getFileCheckpoints(path string) map[string]*v1.Pod {
+	checkpoints := make(map[string]*v1.Pod)
+
+	fi, err := ioutil.ReadDir(path)
+	if err != nil {
+		glog.Fatalf("Failed to read checkpoint manifest path: %v", err)
+	}
+
+	for _, f := range fi {
+		manifest := filepath.Join(path, f.Name())
+		b, err := ioutil.ReadFile(manifest)
+		if err != nil {
+			glog.Errorf("Error reading manifest: %v", err)
+			continue
+		}
+
+		cp := &v1.Pod{}
+		if err := runtime.DecodeInto(api.Codecs.UniversalDecoder(), b, cp); err != nil {
+			glog.Errorf("Error unmarshalling manifest from %s: %v", filepath.Join(path, f.Name()), err)
+			continue
+		}
+
+		if isCheckpoint(cp) {
+			if _, ok := checkpoints[podFullName(cp)]; ok { // sanity check
+				glog.Warningf("Found multiple checkpoint pods in %s with same id: %s", path, podFullName(cp))
+			}
+			checkpoints[podFullName(cp)] = cp
+		}
+	}
+	return checkpoints
+}
+
+// writeCheckpointManifest will save the pod to the inactive checkpoint location if it doesn't already exist.
+func writeCheckpointManifest(pod *v1.Pod) (bool, error) {
+	buff := &bytes.Buffer{}
+	if err := podSerializer.Encode(pod, buff); err != nil {
+		return false, err
+	}
+	path := filepath.Join(inactiveCheckpointPath, pod.Namespace+"-"+pod.Name+".json")
+	// Make sure the inactive checkpoint path exists.
+	if err := os.MkdirAll(filepath.Dir(path), 0600); err != nil {
+		return false, err
+	}
+	return writeManifestIfDifferent(path, podFullName(pod), buff.Bytes())
+}
+
+// writeManifestIfDifferent writes `data` to `path` if data is different from the existing content.
+// The `name` parameter is used for debug output.
+func writeManifestIfDifferent(path, name string, data []byte) (bool, error) {
+	existing, err := ioutil.ReadFile(path)
+	if err != nil && !os.IsNotExist(err) {
+		return false, err
+	}
+	if bytes.Equal(existing, data) {
+		glog.V(4).Infof("Checkpoint manifest for %q already exists. Skipping", name)
+		return false, nil
+	}
+	glog.Infof("Writing manifest for %q to %q", name, path)
+	return true, writeAndAtomicRename(path, data, 0644)
+}