Add pod dependencies

Pod dependencies check for at least one pod which satisfies all of:

* On the same host as the kubernetes-entrypoint container
* In the specified namespace
* Matches the specified labels
* In ready state

It uses JSON for the the env var encoding to avoid complexity of parsing
labels.

README.md

@@ -11,10 +11,10 @@ Kubernetes-entrypoint enables complex deployments on top of Kubernetes.
 
 ## Overview
 
-Kubernetes-entrypoint is meant to be used as a container entrypoint, which means it has to bundled in the container. 
+Kubernetes-entrypoint is meant to be used as a container entrypoint, which means it has to bundled in the container.
 Before launching the desired application, the entrypoint verifies and waits for all specified dependencies to be met.
 
-The Kubernetes-entrypoint queries directly the Kubernetes API and each container is self-aware of its dependencies and their states. 
+The Kubernetes-entrypoint queries directly the Kubernetes API and each container is self-aware of its dependencies and their states.
 Therefore, no centralized orchestration layer is required to manage deployments and scenarios such as failure recovery or pod migration become easy.
 
 ## Usage
@@ -28,7 +28,7 @@ Kubernetes-entrypoint introduces a wide variety of dependencies which can be use
 
 ## Latest features
 
-Extending functionality of kubernetes-entrypoint by adding an ability to specify dependencies in different namespaces. The new format for writing dependencies is `namespace:name`. To ensure backward compatibility if the dependency name is without colon, it behaves just like in previous versions so it assumes that dependecies are running at the same namespace as kubernetes-entrypoint. This feature is not implemented for container, config and socket dependency because in such cases the different namespace is irrelevant.
+Extending functionality of kubernetes-entrypoint by adding an ability to specify dependencies in different namespaces. The new format for writing dependencies is `namespace:name`, with the exception of pod dependencies which us json. To ensure backward compatibility if the `namespace:` is omitted, it behaves just like in previous versions so it assumes that dependecies are running at the same namespace as kubernetes-entrypoint. This feature is not implemented for container, config and socket dependency because in such cases the different namespace is irrelevant.
 
 For instance:
 `
@@ -71,14 +71,14 @@ Example:
 `DEPENDENCY_JOBS=nova-init,neutron-init`
 
 ### Config
-This dependency performs a container level templating of configuration files. It can template an ip address `{{ .IP }}` and hostname `{{ .HOSTNAME }}`. 
+This dependency performs a container level templating of configuration files. It can template an ip address `{{ .IP }}` and hostname `{{ .HOSTNAME }}`.
 Templated config has to be stored in an arbitrary directory `/configmaps/<name_of_file>/<name_of_file>`.
-This dependency requires `INTERFACE_NAME` environment variable to know which interface to use for obtain ip address. 
+This dependency requires `INTERFACE_NAME` environment variable to know which interface to use for obtain ip address.
 Example:
 
 `DEPENDENCY_CONFIG=/etc/nova/nova.conf`
 
-The Kubernetes-entrypoint will look for the configuration file `/configmaps/nova.conf/nova.conf`, template 
+The Kubernetes-entrypoint will look for the configuration file `/configmaps/nova.conf/nova.conf`, template
 `{{ .IP }} and {{ .HOSTNAME }}` tags and save the file as `/etc/nova/nova.conf`.
 
 ### Socket
@@ -87,6 +87,16 @@ Example:
 
 `DEPENDENCY_SOCKET=/var/run/openvswitch/ovs.socket`
 
+### Pod
+Checks if at least one pod matching the specified labels is already running on the same host.
+In contrast to other dependencies, the syntax uses json in order to avoid inventing a new
+format to specify labels and the parsing complexities that would come with that.
+This dependency requires a `POD_NAME` env which can be easily passed through the
+[downward api](http://kubernetes.io/docs/user-guide/downward-api/). The `POD_NAME` variable is mandatory and is used to resolve dependencies.
+Example:
+
+`DEPENDENCY_POD="[{\"namespace\": \"foo\", \"labels\": {\"k1\": \"v1\", \"k2\": \"v2\"}}, {\"labels\": {\"k1\": \"v1\", \"k2\": \"v2\"}}]"`
+
 ## Image
 
 Build process for image is trigged after each commit.

dependencies/daemonset/daemonset_test.go

@@ -62,7 +62,7 @@ var _ = Describe("Daemonset", func() {
 	})
 
 	It("checks resolution failure of a daemonset with incorrect match labels", func() {
-		daemonset, _ := NewDaemonset(mocks.IncorrectMatchLabelsDaemonsetName, daemonsetNamespace)
+		daemonset, _ := NewDaemonset(mocks.FailingMatchLabelsDaemonsetName, daemonsetNamespace)
 
 		isResolved, err := daemonset.IsResolved(testEntrypoint)
 
@@ -73,7 +73,7 @@ var _ = Describe("Daemonset", func() {
 	It(fmt.Sprintf("checks resolution failure of a daemonset with incorrect %s value", PodNameEnvVar), func() {
 		// Set POD_NAME to value not present in the mocks
 		os.Setenv(PodNameEnvVar, mocks.PodNotPresent)
-		daemonset, _ := NewDaemonset(mocks.IncorrectMatchLabelsDaemonsetName, daemonsetNamespace)
+		daemonset, _ := NewDaemonset(mocks.FailingMatchLabelsDaemonsetName, daemonsetNamespace)
 
 		isResolved, err := daemonset.IsResolved(testEntrypoint)
 

dependencies/pod/pod.go

@@ -0,0 +1,106 @@
+package pod
+
+import (
+	"fmt"
+	"os"
+
+	entry "github.com/stackanetes/kubernetes-entrypoint/entrypoint"
+	"github.com/stackanetes/kubernetes-entrypoint/logger"
+	"github.com/stackanetes/kubernetes-entrypoint/util/env"
+	api "k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/api/v1"
+	"k8s.io/client-go/1.5/pkg/labels"
+)
+
+const (
+	PodNameEnvVar            = "POD_NAME"
+	PodNameNotSetErrorFormat = "Env POD_NAME not set. Pod dependency in namespace %s will be ignored!"
+)
+
+type Pod struct {
+	namespace string
+	labels    map[string]string
+	podName   string
+}
+
+func init() {
+	podEnv := fmt.Sprintf("%sPOD", entry.DependencyPrefix)
+	if podDeps := env.SplitPodEnvToDeps(podEnv); podDeps != nil {
+		for _, dep := range podDeps {
+			pod, err := NewPod(dep.Labels, dep.Namespace)
+			if err != nil {
+				logger.Error.Printf("Cannot initialize pod: %v", err)
+				continue
+			}
+			entry.Register(pod)
+		}
+	}
+}
+
+func NewPod(labels map[string]string, namespace string) (*Pod, error) {
+	if os.Getenv(PodNameEnvVar) == "" {
+		return nil, fmt.Errorf(PodNameNotSetErrorFormat, namespace)
+	}
+	return &Pod{
+		namespace: namespace,
+		labels:    labels,
+		podName:   os.Getenv(PodNameEnvVar),
+	}, nil
+}
+
+func (p Pod) IsResolved(entrypoint entry.EntrypointInterface) (bool, error) {
+	myPod, err := entrypoint.Client().Pods(env.GetBaseNamespace()).Get(p.podName)
+	if err != nil {
+		return false, fmt.Errorf("Getting POD: %v failed : %v", p.podName, err)
+	}
+	myHost := myPod.Status.HostIP
+
+	label := labels.SelectorFromSet(p.labels)
+	opts := api.ListOptions{LabelSelector: label}
+
+	matchingPodList, err := entrypoint.Client().Pods(p.namespace).List(opts)
+	if err != nil {
+		return false, err
+	}
+
+	matchingPods := matchingPodList.Items
+	if len(matchingPods) == 0 {
+		return false, fmt.Errorf("No pods found matching labels: %v", p.labels)
+	}
+
+	hostPodCount := 0
+	for _, pod := range matchingPods {
+		if !isPodOnHost(&pod, myHost) {
+			continue
+		}
+		hostPodCount++
+		if isPodReady(pod) {
+			return true, nil
+		}
+	}
+	if hostPodCount == 0 {
+		return false, fmt.Errorf("Found no pods on host matching labels: %v", p.labels)
+	} else {
+		return false, fmt.Errorf("Found %v pods on host, but none ready, matching labels: %v", hostPodCount, p.labels)
+	}
+}
+
+func isPodOnHost(pod *v1.Pod, hostIP string) bool {
+	if pod.Status.HostIP == hostIP {
+		return true
+	}
+	return false
+}
+
+func isPodReady(pod v1.Pod) bool {
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == v1.PodReady && condition.Status == "True" {
+			return true
+		}
+	}
+	return false
+}
+
+func (p Pod) String() string {
+	return fmt.Sprintf("Pod on same host with labels %v in namespace %s", p.labels, p.namespace)
+}

dependencies/pod/pod_suite_test.go

@@ -0,0 +1,13 @@
+package pod_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestPod(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Pod Suite")
+}

dependencies/pod/pod_test.go

@@ -0,0 +1,119 @@
+package pod
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/stackanetes/kubernetes-entrypoint/entrypoint"
+	"github.com/stackanetes/kubernetes-entrypoint/mocks"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+const (
+	podEnvVariableValue = "podlist"
+	podNamespace        = "test"
+)
+
+var testEntrypoint entrypoint.EntrypointInterface
+var testLabels = map[string]string{"foo": "bar"}
+
+var _ = Describe("Pod", func() {
+
+	BeforeEach(func() {
+		err := os.Setenv(PodNameEnvVar, podEnvVariableValue)
+		Expect(err).NotTo(HaveOccurred())
+
+		testEntrypoint = mocks.NewEntrypoint()
+	})
+
+	It(fmt.Sprintf("checks failure of new pod creation without %s set", PodNameEnvVar), func() {
+		os.Unsetenv(PodNameEnvVar)
+		pod, err := NewPod(testLabels, podNamespace)
+
+		Expect(pod).To(BeNil())
+		Expect(err.Error()).To(Equal(fmt.Sprintf(PodNameNotSetErrorFormat, podNamespace)))
+	})
+
+	It(fmt.Sprintf("creates new pod with %s set and checks its name", PodNameEnvVar), func() {
+		pod, err := NewPod(testLabels, podNamespace)
+		Expect(pod).NotTo(BeNil())
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(pod.labels).To(Equal(testLabels))
+	})
+
+	It("is resolved via all pods matching labels ready on same host", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.SameHostReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeTrue())
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("is resolved via some pods matching labels ready on same host", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.SameHostSomeReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeTrue())
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("is not resolved via a pod matching labels not ready on same host", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.SameHostNotReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("is not resolved via pod matching labels ready on different host", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.DifferentHostReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("is not resolved via pod matching labels not ready on different host", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.DifferentHostNotReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("is not resolved via no pods matching labels", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.NoPodsMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("is not resolved when getting pods matching labels from api fails", func() {
+		pod, _ := NewPod(map[string]string{"name": mocks.FailingMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+
+	It(fmt.Sprintf("is not resolved when getting current pod via %s value fails", PodNameEnvVar), func() {
+		os.Setenv(PodNameEnvVar, mocks.PodNotPresent)
+		pod, _ := NewPod(map[string]string{"name": mocks.SameHostReadyMatchLabel}, podNamespace)
+
+		isResolved, err := pod.IsResolved(testEntrypoint)
+
+		Expect(isResolved).To(BeFalse())
+		Expect(err).To(HaveOccurred())
+	})
+})

kubernetes-entrypoint.go

@@ -9,6 +9,7 @@ import (
 	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/container"
 	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/daemonset"
 	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/job"
+	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/pod"
 	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/service"
 	_ "github.com/stackanetes/kubernetes-entrypoint/dependencies/socket"
 	"github.com/stackanetes/kubernetes-entrypoint/logger"

mocks/daemonset.go

@@ -20,10 +20,8 @@ const (
 	IncorrectNamespaceDaemonsetName = "INCORRECT_DAEMONSET_NAMESPACE"
 	CorrectDaemonsetNamespace       = "CORRECT_DAEMONSET"
 
-	IncorrectMatchLabelsDaemonsetName = "DAEMONSET_INCORRECT_MATCH_LABELS"
-	NotReadyMatchLabelsDaemonsetName  = "DAEMONSET_NOT_READY_MATCH_LABELS"
-	IncorrectMatchLabel               = "INCORRECT"
-	NotReadyMatchLabel                = "INCORRECT"
+	FailingMatchLabelsDaemonsetName  = "DAEMONSET_INCORRECT_MATCH_LABELS"
+	NotReadyMatchLabelsDaemonsetName = "DAEMONSET_NOT_READY_MATCH_LABELS"
 )
 
 func (d dClient) Get(name string) (*extensions.DaemonSet, error) {
@@ -31,10 +29,10 @@ func (d dClient) Get(name string) (*extensions.DaemonSet, error) {
 
 	if name == FailingDaemonsetName {
 		return nil, fmt.Errorf("Mock daemonset didnt work")
-	} else if name == IncorrectMatchLabelsDaemonsetName {
-		matchLabelName = IncorrectMatchLabel
+	} else if name == FailingMatchLabelsDaemonsetName {
+		matchLabelName = FailingMatchLabel
 	} else if name == NotReadyMatchLabelsDaemonsetName {
-		matchLabelName = NotReadyMatchLabel
+		matchLabelName = SameHostNotReadyMatchLabel
 	}
 
 	ds := &extensions.DaemonSet{

mocks/pod.go

@@ -17,8 +17,15 @@ type pClient struct {
 }
 
 const (
-	PodNotPresent       = "NOT_PRESENT"
-	PodEnvVariableValue = "podlist"
+	PodNotPresent                   = "NOT_PRESENT"
+	PodEnvVariableValue             = "podlist"
+	FailingMatchLabel               = "INCORRECT"
+	SameHostNotReadyMatchLabel      = "SAME_HOST_NOT_READY"
+	SameHostReadyMatchLabel         = "SAME_HOST_READY"
+	SameHostSomeReadyMatchLabel     = "SAME_HOST_SOME_READY"
+	DifferentHostReadyMatchLabel    = "DIFFERENT_HOST_READY"
+	DifferentHostNotReadyMatchLabel = "DIFFERENT_HOST_NOT_READY"
+	NoPodsMatchLabel                = "NO_PODS"
 )
 
 func (p pClient) Get(name string) (*v1.Pod, error) {
@@ -53,37 +60,38 @@ func (p pClient) DeleteCollection(options *api.DeleteOptions, listOptions api.Li
 }
 
 func (p pClient) List(options api.ListOptions) (*v1.PodList, error) {
-	if options.LabelSelector.String() == "name=INCORRECT" {
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", FailingMatchLabel) {
 		return nil, fmt.Errorf("Client received incorrect pod label names")
 	}
 
-	readyStatus := true
+	readyPodSameHost := NewPod(true, "127.0.0.1")
+	notReadyPodSameHost := NewPod(false, "127.0.0.1")
+	readyPodDifferentHost := NewPod(true, "10.0.0.1")
+	notReadyPodDifferentHost := NewPod(false, "10.0.0.1")
 
-	if options.LabelSelector.String() == "name=NOT_READY" {
-		readyStatus = false
+	var pods []v1.Pod
+
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", SameHostNotReadyMatchLabel) {
+		pods = []v1.Pod{notReadyPodSameHost}
+	}
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", SameHostReadyMatchLabel) {
+		pods = []v1.Pod{readyPodSameHost, notReadyPodDifferentHost}
+	}
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", SameHostSomeReadyMatchLabel) {
+		pods = []v1.Pod{readyPodSameHost, notReadyPodSameHost}
+	}
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", DifferentHostReadyMatchLabel) {
+		pods = []v1.Pod{notReadyPodSameHost, readyPodDifferentHost}
+	}
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", DifferentHostNotReadyMatchLabel) {
+		pods = []v1.Pod{notReadyPodDifferentHost}
+	}
+	if options.LabelSelector.String() == fmt.Sprintf("name=%s", NoPodsMatchLabel) {
+		pods = []v1.Pod{}
 	}
 
 	return &v1.PodList{
-		Items: []v1.Pod{
-			{
-				ObjectMeta: v1.ObjectMeta{Name: PodEnvVariableValue},
-				Status: v1.PodStatus{
-					HostIP: "127.0.01",
-					Conditions: []v1.PodCondition{
-						{
-							Type:   v1.PodReady,
-							Status: "True",
-						},
-					},
-					ContainerStatuses: []v1.ContainerStatus{
-						{
-							Name:  MockContainerName,
-							Ready: readyStatus,
-						},
-					},
-				},
-			},
-		},
+		Items: pods,
 	}, nil
 }
 
@@ -117,3 +125,29 @@ func (p pClient) Patch(name string, pt api.PatchType, data []byte, subresources
 func NewPClient() v1core.PodInterface {
 	return pClient{}
 }
+
+func NewPod(ready bool, hostIP string) v1.Pod {
+	podReadyStatus := v1.ConditionTrue
+	if !ready {
+		podReadyStatus = v1.ConditionFalse
+	}
+
+	return v1.Pod{
+		ObjectMeta: v1.ObjectMeta{Name: PodEnvVariableValue},
+		Status: v1.PodStatus{
+			HostIP: hostIP,
+			Conditions: []v1.PodCondition{
+				{
+					Type:   v1.PodReady,
+					Status: podReadyStatus,
+				},
+			},
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					Name:  MockContainerName,
+					Ready: ready,
+				},
+			},
+		},
+	}
+}

util/env/env.go

@@ -1,6 +1,7 @@
 package env
 
 import (
+	"encoding/json"
 	"os"
 	"strings"
 
@@ -16,6 +17,11 @@ type Dependency struct {
 	Namespace string
 }
 
+type PodDependency struct {
+	Labels    map[string]string
+	Namespace string
+}
+
 func SplitCommand() []string {
 	command := os.Getenv("COMMAND")
 	if command == "" {
@@ -62,6 +68,33 @@ func SplitEnvToDeps(env string) (envList []Dependency) {
 	return envList
 }
 
+//SplitPodEnvToDeps returns list of PodDependency
+func SplitPodEnvToDeps(env string) []PodDependency {
+	deps := []PodDependency{}
+
+	namespace := GetBaseNamespace()
+
+	e := os.Getenv(env)
+	if e == "" {
+		return deps
+	}
+
+	err := json.Unmarshal([]byte(e), &deps)
+	if err != nil {
+		logger.Warning.Printf("Invalid format: ", e)
+		return []PodDependency{}
+	}
+
+	for i, dep := range deps {
+		if dep.Namespace == "" {
+			dep.Namespace = namespace
+			deps[i] = dep
+		}
+	}
+
+	return deps
+}
+
 //GetBaseNamespace returns default namespace when user set empty one
 func GetBaseNamespace() string {
 	namespace := os.Getenv("NAMESPACE")

util/env/env_test.go

@@ -2,6 +2,7 @@ package env
 
 import (
 	"os"
+	"reflect"
 	"testing"
 )
 
@@ -91,6 +92,46 @@ func TestSplitEmptyEnvWithColon(t *testing.T) {
 	}
 }
 
+func TestSplitPodEnvToDepsSuccess(t *testing.T) {
+	defer os.Unsetenv("NAMESPACE")
+	os.Setenv("NAMESPACE", `TEST_NAMESPACE`)
+	defer os.Unsetenv("TEST_LIST")
+	os.Setenv("TEST_LIST", `[{"namespace": "foo", "labels": {"k1": "v1", "k2": "v2"}}, {"labels": {"k1": "v1", "k2": "v2"}}]`)
+	actual := SplitPodEnvToDeps("TEST_LIST")
+	expected := []PodDependency{
+		PodDependency{Namespace: "foo", Labels: map[string]string{
+			"k1": "v1",
+			"k2": "v2",
+		}},
+		PodDependency{Namespace: "TEST_NAMESPACE", Labels: map[string]string{
+			"k1": "v1",
+			"k2": "v2",
+		}},
+	}
+
+	if !reflect.DeepEqual(expected, actual) {
+		t.Errorf("Expected: %v Got: %v", expected, actual)
+	}
+}
+
+func TestSplitPodEnvToDepsUnset(t *testing.T) {
+	defer os.Unsetenv("TEST_LIST")
+	os.Setenv("TEST_LIST", "")
+	actual := SplitPodEnvToDeps("TEST_LIST")
+	if len(actual) != 0 {
+		t.Errorf("Expected: no dependencies Got: %v", actual)
+	}
+}
+
+func TestSplitPodEnvToDepsIgnoreInvalid(t *testing.T) {
+	defer os.Unsetenv("TEST_LIST")
+	os.Setenv("TEST_LIST", `[{"invalid": json}`)
+	actual := SplitPodEnvToDeps("TEST_LIST")
+	if len(actual) != 0 {
+		t.Errorf("Expected: ignore invalid dependencies Got: %v", actual)
+	}
+}
+
 func TestSplitCommand(t *testing.T) {
 	defer os.Unsetenv("COMMAND")
 	list2 := SplitCommand()