Merge pull request #2637 from erictune/service_env_var_2

cmd/e2e/e2e.go

@@ -24,6 +24,7 @@ import (
 	"path/filepath"
 	goruntime "runtime"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
@@ -58,6 +59,39 @@ func waitForPodRunning(c *client.Client, id string) {
 	}
 }
 
+// waitForPodSuccess returns true if the pod reached state success, or false if it reached failure or ran too long.
+func waitForPodSuccess(c *client.Client, podName string, contName string) bool {
+	for i := 0; i < 10; i++ {
+		if i > 0 {
+			time.Sleep(5 * time.Second)
+		}
+		pod, err := c.Pods(api.NamespaceDefault).Get(podName)
+		if err != nil {
+			glog.Warningf("Get pod failed: %v", err)
+			continue
+		}
+		// Cannot use pod.Status.Phase == api.PodSucceeded/api.PodFailed due to #2632
+		ci, ok := pod.Status.Info[contName]
+		if !ok {
+			glog.Infof("No Status.Info for container %s in pod %s yet", contName, podName)
+		} else {
+			if ci.State.Termination != nil {
+				if ci.State.Termination.ExitCode == 0 {
+					glog.Infof("Saw pod success")
+					return true
+				} else {
+					glog.Infof("Saw pod failure: %+v", ci.State.Termination)
+				}
+				glog.Infof("Waiting for pod %q status to be success or failure", podName)
+			} else {
+				glog.Infof("Nil State.Termination for container %s in pod %s so far", contName, podName)
+			}
+		}
+	}
+	glog.Warningf("Gave up waiting for pod %q status to be success or failure", podName)
+	return false
+}
+
 // assetPath returns a path to the requested file; safe on all
 // OSes. NOTE: If you use an asset in this test, you MUST add it to
 // the KUBE_TEST_PORTABLE array in hack/lib/golang.sh.
@@ -68,11 +102,15 @@ func assetPath(pathElements ...string) string {
 func loadObjectOrDie(filePath string) runtime.Object {
 	data, err := ioutil.ReadFile(filePath)
 	if err != nil {
-		glog.Fatalf("Failed to read pod: %v", err)
+		glog.Fatalf("Failed to read object: %v", err)
 	}
+	return decodeObjectOrDie(data)
+}
+
+func decodeObjectOrDie(data []byte) runtime.Object {
 	obj, err := latest.Codec.Decode(data)
 	if err != nil {
-		glog.Fatalf("Failed to decode pod: %v", err)
+		glog.Fatalf("Failed to decode object: %v", err)
 	}
 	return obj
 }
@@ -112,6 +150,24 @@ func loadClientOrDie() *client.Client {
 	return c
 }
 
+func parsePodOrDie(json string) *api.Pod {
+	obj := decodeObjectOrDie([]byte(json))
+	pod, ok := obj.(*api.Pod)
+	if !ok {
+		glog.Fatalf("Failed to cast pod: %v", obj)
+	}
+	return pod
+}
+
+func parseServiceOrDie(json string) *api.Service {
+	obj := decodeObjectOrDie([]byte(json))
+	service, ok := obj.(*api.Service)
+	if !ok {
+		glog.Fatalf("Failed to cast service: %v", obj)
+	}
+	return service
+}
+
 func TestKubernetesROService(c *client.Client) bool {
 	svc := api.ServiceList{}
 	err := c.Get().
@@ -498,6 +554,140 @@ func TestClusterDNS(c *client.Client) bool {
 	return true
 }
 
+// TestPodHasServiceEnvVars checks that kubelets and scheduler send events about pods scheduling and running.
+func TestPodHasServiceEnvVars(c *client.Client) bool {
+	// Make a pod that will be a service.
+	// This pod serves its hostname via HTTP.
+	serverPod := parsePodOrDie(`{
+	  "kind": "Pod",
+	  "apiVersion": "v1beta1",
+	  "id": "srv",
+	  "desiredState": {
+		"manifest": {
+		  "version": "v1beta1",
+		  "id": "srv",
+		  "containers": [{
+			"name": "srv",
+			"image": "kubernetes/serve_hostname",
+			"ports": [{
+			  "containerPort": 80,
+			  "hostPort": 8080
+			}]
+		  }]
+		}
+	  },
+	  "labels": {
+		"name": "srv"
+	  }
+	}`)
+	_, err := c.Pods(api.NamespaceDefault).Create(serverPod)
+	if err != nil {
+		glog.Errorf("Failed to create serverPod: %v", err)
+		return false
+	}
+	defer c.Pods(api.NamespaceDefault).Delete(serverPod.Name)
+	waitForPodRunning(c, serverPod.Name)
+
+	// This service exposes pod p's port 8080 as a service on port 8765
+	svc := parseServiceOrDie(`{
+	  "id": "fooservice",
+	  "kind": "Service",
+	  "apiVersion": "v1beta1",
+	  "port": 8765,
+	  "containerPort": 8080,
+	  "selector": {
+		"name": "p"
+	  }
+	}`)
+	if err != nil {
+		glog.Errorf("Failed to delete service: %v", err)
+		return false
+	}
+	time.Sleep(2)
+	_, err = c.Services(api.NamespaceDefault).Create(svc)
+	if err != nil {
+		glog.Errorf("Failed to create service: %v", err)
+		return false
+	}
+	defer c.Services(api.NamespaceDefault).Delete(svc.Name)
+	// TODO: we don't have a way to wait for a service to be "running".
+	// If this proves flaky, then we will need to retry the clientPod or insert a sleep.
+
+	// Make a client pod that verifies that it has the service environment variables.
+	clientPod := parsePodOrDie(`{
+	  "apiVersion": "v1beta1",
+	  "kind": "Pod",
+	  "id": "env3",
+	  "desiredState": {
+		"manifest": {
+		  "version": "v1beta1",
+		  "id": "env3",
+		  "restartPolicy": { "never": {} },
+		  "containers": [{
+			"name": "env3cont",
+			"image": "busybox",
+			"command": ["sh", "-c", "env"]
+		  }]
+		}
+	  },
+	  "labels": { "name": "env3" }
+	}`)
+	_, err = c.Pods(api.NamespaceDefault).Create(clientPod)
+	if err != nil {
+		glog.Errorf("Failed to create pod: %v", err)
+		return false
+	}
+	defer c.Pods(api.NamespaceDefault).Delete(clientPod.Name)
+
+	// Wait for client pod to complete.
+	success := waitForPodSuccess(c, clientPod.Name, clientPod.Spec.Containers[0].Name)
+	if !success {
+		glog.Errorf("Failed to run client pod to detect service env vars.")
+	}
+
+	// Grab its logs.  Get host first.
+	clientPodStatus, err := c.Pods(api.NamespaceDefault).Get(clientPod.Name)
+	if err != nil {
+		glog.Errorf("Failed to get clientPod to know host: %v", err)
+		return false
+	}
+	glog.Infof("Trying to get logs from host %s pod %s container %s: %v",
+		clientPodStatus.Status.Host, clientPodStatus.Name, clientPodStatus.Spec.Containers[0].Name, err)
+	logs, err := c.Get().
+		Prefix("proxy").
+		Resource("minions").
+		Name(clientPodStatus.Status.Host).
+		Suffix("containerLogs", api.NamespaceDefault, clientPodStatus.Name, clientPodStatus.Spec.Containers[0].Name).
+		Do().
+		Raw()
+	if err != nil {
+		glog.Errorf("Failed to get logs from host %s pod %s container %s: %v",
+			clientPodStatus.Status.Host, clientPodStatus.Name, clientPodStatus.Spec.Containers[0].Name, err)
+		return false
+	}
+	glog.Info("clientPod logs:", string(logs))
+
+	toFind := []string{
+		"FOOSERVICE_SERVICE_HOST=",
+		"FOOSERVICE_SERVICE_PORT=",
+		"FOOSERVICE_PORT=",
+		"FOOSERVICE_PORT_8765_TCP_PORT=",
+		"FOOSERVICE_PORT_8765_TCP_PROTO=",
+		"FOOSERVICE_PORT_8765_TCP=",
+		"FOOSERVICE_PORT_8765_TCP_ADDR=",
+	}
+
+	for _, m := range toFind {
+		if !strings.Contains(string(logs), m) {
+			glog.Errorf("Unable to find env var %q in client env vars.", m)
+			success = false
+		}
+	}
+
+	// We could try a wget the service from the client pod.  But services.sh e2e test covers that pretty well.
+	return success
+}
+
 func main() {
 	flag.Parse()
 	goruntime.GOMAXPROCS(goruntime.NumCPU())
@@ -521,6 +711,7 @@ func main() {
 		{TestPodUpdate, "TestPodUpdate", 4},
 		{TestNetwork, "TestNetwork", 5},
 		{TestClusterDNS, "TestClusterDNS", 6},
+		{TestPodHasServiceEnvVars, "TestPodHasServiceEnvVars", 7},
 	}
 
 	info := []TestInfo{}