Merge pull request #9646 from rajatchopra/status_hook

pkg/kubelet/dockertools/manager.go

@@ -262,7 +262,7 @@ type containerStatusResult struct {
 	err    error
 }
 
-func (dm *DockerManager) inspectContainer(dockerID, containerName, tPath string) *containerStatusResult {
+func (dm *DockerManager) inspectContainer(dockerID, containerName, tPath string, pod *api.Pod) *containerStatusResult {
 	result := containerStatusResult{api.ContainerStatus{}, "", nil}
 
 	inspectResult, err := dm.client.InspectContainer(dockerID)
@@ -288,8 +288,19 @@ func (dm *DockerManager) inspectContainer(dockerID, containerName, tPath string)
 		result.status.State.Running = &api.ContainerStateRunning{
 			StartedAt: util.NewTime(inspectResult.State.StartedAt),
 		}
-		if containerName == PodInfraContainerName && inspectResult.NetworkSettings != nil {
-			result.ip = inspectResult.NetworkSettings.IPAddress
+		if containerName == PodInfraContainerName {
+			if inspectResult.NetworkSettings != nil {
+				result.ip = inspectResult.NetworkSettings.IPAddress
+			}
+			// override the above if a network plugin exists
+			if dm.networkPlugin.Name() != network.DefaultPluginName {
+				netStatus, err := dm.networkPlugin.Status(pod.Namespace, pod.Name, kubeletTypes.DockerID(dockerID))
+				if err != nil {
+					glog.Errorf("NetworkPlugin %s failed on the status hook for pod '%s' - %v", dm.networkPlugin.Name(), pod.Name, err)
+				} else if netStatus != nil {
+					result.ip = netStatus.IP.String()
+				}
+			}
 		}
 	} else if !inspectResult.State.FinishedAt.IsZero() {
 		reason := ""
@@ -389,7 +400,7 @@ func (dm *DockerManager) GetPodStatus(pod *api.Pod) (*api.PodStatus, error) {
 
 		var terminationState *api.ContainerState = nil
 		// Inspect the container.
-		result := dm.inspectContainer(value.ID, dockerContainerName, terminationMessagePath)
+		result := dm.inspectContainer(value.ID, dockerContainerName, terminationMessagePath, pod)
 		if result.err != nil {
 			return nil, result.err
 		} else if result.status.State.Terminated != nil {

pkg/kubelet/network/exec/exec.go

@@ -29,6 +29,15 @@ limitations under the License.
 //         - setup, called after the infra container of a pod is
 //                created, but before other containers of the pod are created
 //         - teardown, called before the pod infra container is killed
+//         - status, called at regular intervals and is supposed to return a json
+//                formatted output indicating the pod's IPAddress(v4/v6). An empty string value or an erroneous output
+//                will mean the container runtime (docker) will be asked for the PodIP
+//                e.g. {
+//                         "apiVersion" : "v1beta1",
+//                         "kind" : "PodNetworkStatus",
+//                         "ip" : "10.20.30.40"
+//                     }
+//                The fields "apiVersion" and "kind" are optional in version v1beta1
 // As the executables are called, the file-descriptors stdin, stdout, stderr
 // remain open. The combined output of stdout/stderr is captured and logged.
 //
@@ -48,6 +57,7 @@ limitations under the License.
 package exec
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -55,6 +65,7 @@ import (
 	"strings"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
 	utilexec "k8s.io/kubernetes/pkg/util/exec"
@@ -70,6 +81,7 @@ const (
 	initCmd     = "init"
 	setUpCmd    = "setup"
 	tearDownCmd = "teardown"
+	statusCmd   = "status"
 )
 
 func ProbeNetworkPlugins(pluginDir string) []network.NetworkPlugin {
@@ -131,3 +143,37 @@ func (plugin *execNetworkPlugin) TearDownPod(namespace string, name string, id k
 	glog.V(5).Infof("TearDownPod 'exec' network plugin output: %s, %v", string(out), err)
 	return err
 }
+
+func (plugin *execNetworkPlugin) Status(namespace string, name string, id kubeletTypes.DockerID) (*network.PodNetworkStatus, error) {
+	out, err := utilexec.New().Command(plugin.getExecutable(), statusCmd, namespace, name, string(id)).CombinedOutput()
+	glog.V(5).Infof("Status 'exec' network plugin output: %s, %v", string(out), err)
+	if err != nil {
+		return nil, err
+	}
+	if string(out) == "" {
+		return nil, nil
+	}
+	findVersion := struct {
+		api.TypeMeta `json:",inline"`
+	}{}
+	err = json.Unmarshal(out, &findVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	// check kind and version
+	if findVersion.Kind != "" && findVersion.Kind != "PodNetworkStatus" {
+		errStr := fmt.Sprintf("Invalid 'kind' returned in network status for pod '%s'. Valid value is 'PodNetworkStatus', got '%s'.", name, findVersion.Kind)
+		return nil, errors.New(errStr)
+	}
+	switch findVersion.APIVersion {
+	case "":
+		fallthrough
+	case "v1beta1":
+		networkStatus := &network.PodNetworkStatus{}
+		err = json.Unmarshal(out, networkStatus)
+		return networkStatus, err
+	}
+	errStr := fmt.Sprintf("Unknown version '%s' in network status for pod '%s'.", findVersion.APIVersion, name)
+	return nil, errors.New(errStr)
+}

pkg/kubelet/network/exec/exec_test.go

@@ -19,12 +19,14 @@ limitations under the License.
 package exec
 
 import (
+	"bytes"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
 	"os"
 	"path"
 	"testing"
+	"text/template"
 
 	"k8s.io/kubernetes/pkg/kubelet/network"
 )
@@ -32,7 +34,7 @@ import (
 // The temp dir where test plugins will be stored.
 const testPluginPath = "/tmp/fake/plugins/net"
 
-func installPluginUnderTest(t *testing.T, vendorName string, plugName string) {
+func installPluginUnderTest(t *testing.T, vendorName string, plugName string, execTemplateData *map[string]interface{}) {
 	vendoredName := plugName
 	if vendorName != "" {
 		vendoredName = fmt.Sprintf("%s~%s", vendorName, plugName)
@@ -51,8 +53,32 @@ func installPluginUnderTest(t *testing.T, vendorName string, plugName string) {
 	if err != nil {
 		t.Errorf("Failed to set exec perms on plugin")
 	}
-	writeStr := fmt.Sprintf("#!/bin/bash\necho -n $@ &> %s", path.Join(pluginDir, plugName+".out"))
-	_, err = f.WriteString(writeStr)
+	const execScriptTempl = `#!/bin/bash
+
+# If status hook is called print the expected json to stdout
+if [ "$1" == "status" ]; then
+  echo -n '{
+	"ip" : "{{.IPAddress}}"
+}'
+fi
+
+# Direct the arguments to a file to be tested against later
+echo -n $@ &> {{.OutputFile}}
+`
+	if execTemplateData == nil {
+		execTemplateData = &map[string]interface{}{
+			"IPAddress":  "10.20.30.40",
+			"OutputFile": path.Join(pluginDir, plugName+".out"),
+		}
+	}
+
+	tObj := template.Must(template.New("test").Parse(execScriptTempl))
+	buf := &bytes.Buffer{}
+	if err := tObj.Execute(buf, *execTemplateData); err != nil {
+		t.Errorf("Error in executing script template - %v", err)
+	}
+	execScript := buf.String()
+	_, err = f.WriteString(execScript)
 	if err != nil {
 		t.Errorf("Failed to write plugin exec")
 	}
@@ -70,7 +96,7 @@ func TestSelectPlugin(t *testing.T) {
 	// install some random plugin under testPluginPath
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
-	installPluginUnderTest(t, "", pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
 
 	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), pluginName, network.NewFakeHost(nil))
 	if err != nil {
@@ -86,7 +112,7 @@ func TestSelectVendoredPlugin(t *testing.T) {
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
 	vendor := "mycompany"
-	installPluginUnderTest(t, vendor, pluginName)
+	installPluginUnderTest(t, vendor, pluginName, nil)
 
 	vendoredPluginName := fmt.Sprintf("%s/%s", vendor, pluginName)
 	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), vendoredPluginName, network.NewFakeHost(nil))
@@ -102,7 +128,7 @@ func TestSelectWrongPlugin(t *testing.T) {
 	// install some random plugin under testPluginPath
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
-	installPluginUnderTest(t, "", pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
 
 	wrongPlugin := "abcd"
 	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), wrongPlugin, network.NewFakeHost(nil))
@@ -114,7 +140,7 @@ func TestSelectWrongPlugin(t *testing.T) {
 func TestPluginValidation(t *testing.T) {
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
-	installPluginUnderTest(t, "", pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
 
 	// modify the perms of the pluginExecutable
 	f, err := os.Open(path.Join(testPluginPath, pluginName, pluginName))
@@ -137,7 +163,7 @@ func TestPluginValidation(t *testing.T) {
 func TestPluginSetupHook(t *testing.T) {
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
-	installPluginUnderTest(t, "", pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
 
 	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), pluginName, network.NewFakeHost(nil))
 
@@ -159,7 +185,7 @@ func TestPluginSetupHook(t *testing.T) {
 func TestPluginTearDownHook(t *testing.T) {
 	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
 	defer tearDownPlugin(pluginName)
-	installPluginUnderTest(t, "", pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
 
 	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), pluginName, network.NewFakeHost(nil))
 
@@ -177,3 +203,58 @@ func TestPluginTearDownHook(t *testing.T) {
 		t.Errorf("Mismatch in expected output for teardown hook. Expected '%s', got '%s'", expectedOutput, string(output))
 	}
 }
+
+func TestPluginStatusHook(t *testing.T) {
+	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
+	defer tearDownPlugin(pluginName)
+	installPluginUnderTest(t, "", pluginName, nil)
+
+	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), pluginName, network.NewFakeHost(nil))
+
+	ip, err := plug.Status("namespace", "name", "dockerid2345")
+	if err != nil {
+		t.Errorf("Expected nil got %v", err)
+	}
+	// check output of status hook
+	output, err := ioutil.ReadFile(path.Join(testPluginPath, pluginName, pluginName+".out"))
+	if err != nil {
+		t.Errorf("Expected nil")
+	}
+	expectedOutput := "status namespace name dockerid2345"
+	if string(output) != expectedOutput {
+		t.Errorf("Mismatch in expected output for status hook. Expected '%s', got '%s'", expectedOutput, string(output))
+	}
+	if ip.IP.String() != "10.20.30.40" {
+		t.Errorf("Mismatch in expected output for status hook. Expected '10.20.30.40', got '%s'", ip.IP.String())
+	}
+}
+
+func TestPluginStatusHookIPv6(t *testing.T) {
+	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
+	defer tearDownPlugin(pluginName)
+	pluginDir := path.Join(testPluginPath, pluginName)
+	execTemplate := &map[string]interface{}{
+		"IPAddress":  "fe80::e2cb:4eff:fef9:6710",
+		"OutputFile": path.Join(pluginDir, pluginName+".out"),
+	}
+	installPluginUnderTest(t, "", pluginName, execTemplate)
+
+	plug, err := network.InitNetworkPlugin(ProbeNetworkPlugins(testPluginPath), pluginName, network.NewFakeHost(nil))
+
+	ip, err := plug.Status("namespace", "name", "dockerid2345")
+	if err != nil {
+		t.Errorf("Expected nil got %v", err)
+	}
+	// check output of status hook
+	output, err := ioutil.ReadFile(path.Join(testPluginPath, pluginName, pluginName+".out"))
+	if err != nil {
+		t.Errorf("Expected nil")
+	}
+	expectedOutput := "status namespace name dockerid2345"
+	if string(output) != expectedOutput {
+		t.Errorf("Mismatch in expected output for status hook. Expected '%s', got '%s'", expectedOutput, string(output))
+	}
+	if ip.IP.String() != "fe80::e2cb:4eff:fef9:6710" {
+		t.Errorf("Mismatch in expected output for status hook. Expected 'fe80::e2cb:4eff:fef9:6710', got '%s'", ip.IP.String())
+	}
+}

pkg/kubelet/network/plugins.go

@@ -18,6 +18,7 @@ package network
 
 import (
 	"fmt"
+	"net"
 	"strings"
 
 	"github.com/golang/glog"
@@ -47,6 +48,21 @@ type NetworkPlugin interface {
 
 	// TearDownPod is the method called before a pod's infra container will be deleted
 	TearDownPod(namespace string, name string, podInfraContainerID kubeletTypes.DockerID) error
+
+	// Status is the method called to obtain the ipv4 or ipv6 addresses of the container
+	Status(namespace string, name string, podInfraContainerID kubeletTypes.DockerID) (*PodNetworkStatus, error)
+}
+
+// PodNetworkStatus stores the network status of a pod (currently just the primary IP address)
+// This struct represents version "v1"
+type PodNetworkStatus struct {
+	api.TypeMeta `json:",inline"`
+
+	// IP is the primary ipv4/ipv6 address of the pod. Among other things it is the address that -
+	//   - kube expects to be reachable across the cluster
+	//   - service endpoints are constructed with
+	//   - will be reported in the PodStatus.PodIP field (will override the IP reported by docker)
+	IP net.IP `json:"ip" description:"Primary IP address of the pod"`
 }
 
 // Host is an interface that plugins can use to access the kubelet.
@@ -120,3 +136,7 @@ func (plugin *noopNetworkPlugin) SetUpPod(namespace string, name string, id kube
 func (plugin *noopNetworkPlugin) TearDownPod(namespace string, name string, id kubeletTypes.DockerID) error {
 	return nil
 }
+
+func (plugin *noopNetworkPlugin) Status(namespace string, name string, id kubeletTypes.DockerID) (*PodNetworkStatus, error) {
+	return nil, nil
+}