Merge pull request #7715 from csrwng/exec_portfw_client

pkg/client/remotecommand/remotecommand.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/conversion/queryparams"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/httpstream"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/httpstream/spdy"
 	"github.com/golang/glog"
@@ -39,7 +40,8 @@ func (u *defaultUpgrader) upgrade(req *client.Request, config *client.Config) (h
 	return req.Upgrade(config, spdy.NewRoundTripper)
 }
 
-type RemoteCommandExecutor struct {
+// Executor executes a command on a pod container
+type Executor struct {
 	req     *client.Request
 	config  *client.Config
 	command []string
@@ -51,8 +53,9 @@ type RemoteCommandExecutor struct {
 	upgrader upgrader
 }
 
-func New(req *client.Request, config *client.Config, command []string, stdin io.Reader, stdout, stderr io.Writer, tty bool) *RemoteCommandExecutor {
-	return &RemoteCommandExecutor{
+// New creates a new RemoteCommandExecutor
+func New(req *client.Request, config *client.Config, command []string, stdin io.Reader, stdout, stderr io.Writer, tty bool) *Executor {
+	return &Executor{
 		req:     req,
 		config:  config,
 		command: command,
@@ -66,26 +69,27 @@ func New(req *client.Request, config *client.Config, command []string, stdin io.
 // Execute sends a remote command execution request, upgrading the
 // connection and creating streams to represent stdin/stdout/stderr. Data is
 // copied between these streams and the supplied stdin/stdout/stderr parameters.
-func (e *RemoteCommandExecutor) Execute() error {
-	doStdin := (e.stdin != nil)
-	doStdout := (e.stdout != nil)
-	doStderr := (!e.tty && e.stderr != nil)
-
-	if doStdin {
-		e.req.Param(api.ExecStdinParam, "1")
-	}
-	if doStdout {
-		e.req.Param(api.ExecStdoutParam, "1")
+func (e *Executor) Execute() error {
+	opts := api.PodExecOptions{
+		Stdin:   (e.stdin != nil),
+		Stdout:  (e.stdout != nil),
+		Stderr:  (!e.tty && e.stderr != nil),
+		TTY:     e.tty,
+		Command: e.command,
 	}
-	if doStderr {
-		e.req.Param(api.ExecStderrParam, "1")
+
+	versioned, err := api.Scheme.ConvertToVersion(&opts, e.config.Version)
+	if err != nil {
+		return err
 	}
-	if e.tty {
-		e.req.Param(api.ExecTTYParam, "1")
+	params, err := queryparams.Convert(versioned)
+	if err != nil {
+		return err
 	}
-
-	for _, s := range e.command {
-		e.req.Param(api.ExecCommandParamm, s)
+	for k, v := range params {
+		for _, vv := range v {
+			e.req.Param(k, vv)
+		}
 	}
 
 	if e.upgrader == nil {
@@ -130,7 +134,7 @@ func (e *RemoteCommandExecutor) Execute() error {
 	}()
 	defer errorStream.Reset()
 
-	if doStdin {
+	if opts.Stdin {
 		headers.Set(api.StreamType, api.StreamTypeStdin)
 		remoteStdin, err := conn.CreateStream(headers)
 		if err != nil {
@@ -147,7 +151,7 @@ func (e *RemoteCommandExecutor) Execute() error {
 	waitCount := 0
 	completedStreams := 0
 
-	if doStdout {
+	if opts.Stdout {
 		waitCount++
 		headers.Set(api.StreamType, api.StreamTypeStdout)
 		remoteStdout, err := conn.CreateStream(headers)
@@ -158,7 +162,7 @@ func (e *RemoteCommandExecutor) Execute() error {
 		go cp(api.StreamTypeStdout, e.stdout, remoteStdout)
 	}
 
-	if doStderr && !e.tty {
+	if opts.Stderr && !e.tty {
 		waitCount++
 		headers.Set(api.StreamType, api.StreamTypeStderr)
 		remoteStderr, err := conn.CreateStream(headers)

pkg/conversion/queryparams/convert.go

+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 queryparams
+
+import (
+	"fmt"
+	"net/url"
+	"reflect"
+	"strings"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+)
+
+func jsonTag(field reflect.StructField) (string, bool) {
+	structTag := field.Tag.Get("json")
+	if len(structTag) == 0 {
+		return "", false
+	}
+	parts := strings.Split(structTag, ",")
+	tag := parts[0]
+	if tag == "-" {
+		tag = ""
+	}
+	omitempty := false
+	parts = parts[1:]
+	for _, part := range parts {
+		if part == "omitempty" {
+			omitempty = true
+			break
+		}
+	}
+	return tag, omitempty
+}
+
+func formatValue(value interface{}) string {
+	return fmt.Sprintf("%v", value)
+}
+
+func isValueKind(kind reflect.Kind) bool {
+	switch kind {
+	case reflect.String, reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16,
+		reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8,
+		reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Float32,
+		reflect.Float64, reflect.Complex64, reflect.Complex128:
+		return true
+	default:
+		return false
+	}
+}
+
+func zeroValue(value reflect.Value) bool {
+	return reflect.DeepEqual(reflect.Zero(value.Type()).Interface(), value.Interface())
+}
+
+func addParam(values url.Values, tag string, omitempty bool, value reflect.Value) {
+	if omitempty && zeroValue(value) {
+		return
+	}
+	values.Add(tag, fmt.Sprintf("%v", value.Interface()))
+}
+
+func addListOfParams(values url.Values, tag string, omitempty bool, list reflect.Value) {
+	for i := 0; i < list.Len(); i++ {
+		addParam(values, tag, omitempty, list.Index(i))
+	}
+}
+
+// Convert takes a versioned runtime.Object and serializes it to a url.Values object
+// using JSON tags as parameter names. Only top-level simple values, arrays, and slices
+// are serialized. Embedded structs, maps, etc. will not be serialized.
+func Convert(obj runtime.Object) (url.Values, error) {
+	result := url.Values{}
+	if obj == nil {
+		return result, nil
+	}
+	var sv reflect.Value
+	switch reflect.TypeOf(obj).Kind() {
+	case reflect.Ptr, reflect.Interface:
+		sv = reflect.ValueOf(obj).Elem()
+	default:
+		return nil, fmt.Errorf("Expecting a pointer or interface")
+	}
+	st := sv.Type()
+	if st.Kind() != reflect.Struct {
+		return nil, fmt.Errorf("Expecting a pointer to a struct")
+	}
+	for i := 0; i < st.NumField(); i++ {
+		field := sv.Field(i)
+		tag, omitempty := jsonTag(st.Field(i))
+		if len(tag) == 0 {
+			continue
+		}
+		ft := field.Type()
+		switch {
+		case isValueKind(ft.Kind()):
+			addParam(result, tag, omitempty, field)
+		case ft.Kind() == reflect.Array || ft.Kind() == reflect.Slice:
+			if isValueKind(ft.Elem().Kind()) {
+				addListOfParams(result, tag, omitempty, field)
+			}
+		}
+	}
+	return result, nil
+}

pkg/conversion/queryparams/convert_test.go

+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 queryparams
+
+import (
+	"net/url"
+	"reflect"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+)
+
+type namedString string
+type namedBool bool
+
+type bar struct {
+	Float1   float32 `json:"float1"`
+	Float2   float64 `json:"float2"`
+	Int1     int64   `json:"int1,omitempty"`
+	Int2     int32   `json:"int2,omitempty"`
+	Int3     int16   `json:"int3,omitempty"`
+	Str1     string  `json:"str1,omitempty"`
+	Ignored  int
+	Ignored2 string
+}
+
+func (*bar) IsAnAPIObject() {}
+
+type foo struct {
+	Str       string            `json:"str"`
+	Integer   int               `json:"integer,omitempty"`
+	Slice     []string          `json:"slice,omitempty"`
+	Boolean   bool              `json:"boolean,omitempty"`
+	NamedStr  namedString       `json:"namedStr,omitempty"`
+	NamedBool namedBool         `json:"namedBool,omitempty"`
+	Foobar    bar               `json:"foobar,omitempty"`
+	Testmap   map[string]string `json:"testmap,omitempty"`
+}
+
+func (*foo) IsAnAPIObject() {}
+
+func validateResult(t *testing.T, input interface{}, actual, expected url.Values) {
+	local := url.Values{}
+	for k, v := range expected {
+		local[k] = v
+	}
+	for k, v := range actual {
+		if ev, ok := local[k]; !ok || !reflect.DeepEqual(ev, v) {
+			if !ok {
+				t.Errorf("%#v: actual value key %s not found in expected map", input, k)
+			} else {
+				t.Errorf("%#v: values don't match: actual: %#v, expected: %#v", input, v, ev)
+			}
+			break
+		}
+		delete(local, k)
+	}
+	if len(local) > 0 {
+		t.Errorf("%#v: expected map has keys that were not found in actual map: %#v", input, local)
+	}
+}
+
+func TestConvert(t *testing.T) {
+	tests := []struct {
+		input    runtime.Object
+		expected url.Values
+	}{
+		{
+			input: &foo{
+				Str: "hello",
+			},
+			expected: url.Values{"str": {"hello"}},
+		},
+		{
+			input: &foo{
+				Str:     "test string",
+				Slice:   []string{"one", "two", "three"},
+				Integer: 234,
+				Boolean: true,
+			},
+			expected: url.Values{"str": {"test string"}, "slice": {"one", "two", "three"}, "integer": {"234"}, "boolean": {"true"}},
+		},
+		{
+			input: &foo{
+				Str:       "named types",
+				NamedStr:  "value1",
+				NamedBool: true,
+			},
+			expected: url.Values{"str": {"named types"}, "namedStr": {"value1"}, "namedBool": {"true"}},
+		},
+		{
+			input: &foo{
+				Str: "ignore embedded struct",
+				Foobar: bar{
+					Float1: 5.0,
+				},
+			},
+			expected: url.Values{"str": {"ignore embedded struct"}},
+		},
+		{
+			// Ignore untagged fields
+			input: &bar{
+				Float1:   23.5,
+				Float2:   100.7,
+				Int1:     1,
+				Int2:     2,
+				Int3:     3,
+				Ignored:  1,
+				Ignored2: "ignored",
+			},
+			expected: url.Values{"float1": {"23.5"}, "float2": {"100.7"}, "int1": {"1"}, "int2": {"2"}, "int3": {"3"}},
+		},
+		{
+			// include fields that are not tagged omitempty
+			input: &foo{
+				NamedStr: "named str",
+			},
+			expected: url.Values{"str": {""}, "namedStr": {"named str"}},
+		},
+	}
+
+	for _, test := range tests {
+		result, err := Convert(test.input)
+		if err != nil {
+			t.Errorf("Unexpected error while converting %#v: %v", test.input, err)
+		}
+		validateResult(t, test.input, result, test.expected)
+	}
+}

pkg/conversion/queryparams/doc.go

+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 queryparams provides conversion from versioned
+// runtime objects to URL query values
+package queryparams

pkg/kubectl/cmd/exec.go

@@ -23,6 +23,7 @@ import (
 	"syscall"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/remotecommand"
 	cmdutil "github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/cmd/util"
 	"github.com/docker/docker/pkg/term"
@@ -39,36 +40,53 @@ $ kubectl exec -p 123456-7890 -c ruby-container -i -t -- bash -il`
 )
 
 func NewCmdExec(f *cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer) *cobra.Command {
+	params := &execParams{}
 	cmd := &cobra.Command{
 		Use:     "exec -p POD -c CONTAINER -- COMMAND [args...]",
 		Short:   "Execute a command in a container.",
 		Long:    "Execute a command in a container.",
 		Example: exec_example,
 		Run: func(cmd *cobra.Command, args []string) {
-			err := RunExec(f, cmdIn, cmdOut, cmdErr, cmd, args)
+			err := RunExec(f, cmd, cmdIn, cmdOut, cmdErr, params, args, &defaultRemoteExecutor{})
 			cmdutil.CheckErr(err)
 		},
 	}
-	cmd.Flags().StringP("pod", "p", "", "Pod name")
+	cmd.Flags().StringVarP(&params.podName, "pod", "p", "", "Pod name")
 	cmd.MarkFlagRequired("pod")
 	// TODO support UID
-	cmd.Flags().StringP("container", "c", "", "Container name")
+	cmd.Flags().StringVarP(&params.containerName, "container", "c", "", "Container name")
 	cmd.MarkFlagRequired("container")
-	cmd.Flags().BoolP("stdin", "i", false, "Pass stdin to the container")
-	cmd.Flags().BoolP("tty", "t", false, "Stdin is a TTY")
+	cmd.Flags().BoolVarP(&params.stdin, "stdin", "i", false, "Pass stdin to the container")
+	cmd.Flags().BoolVarP(&params.tty, "tty", "t", false, "Stdin is a TTY")
 	return cmd
 }
 
-func RunExec(f *cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd *cobra.Command, args []string) error {
-	podName := cmdutil.GetFlagString(cmd, "pod")
-	if len(podName) == 0 {
+type remoteExecutor interface {
+	Execute(req *client.Request, config *client.Config, command []string, stdin io.Reader, stdout, stderr io.Writer, tty bool) error
+}
+
+type defaultRemoteExecutor struct{}
+
+func (*defaultRemoteExecutor) Execute(req *client.Request, config *client.Config, command []string, stdin io.Reader, stdout, stderr io.Writer, tty bool) error {
+	executor := remotecommand.New(req, config, command, stdin, stdout, stderr, tty)
+	return executor.Execute()
+}
+
+type execParams struct {
+	podName       string
+	containerName string
+	stdin         bool
+	tty           bool
+}
+
+func RunExec(f *cmdutil.Factory, cmd *cobra.Command, cmdIn io.Reader, cmdOut, cmdErr io.Writer, p *execParams, args []string, re remoteExecutor) error {
+	if len(p.podName) == 0 {
 		return cmdutil.UsageError(cmd, "POD is required for exec")
 	}
 
 	if len(args) < 1 {
 		return cmdutil.UsageError(cmd, "COMMAND is required for exec")
 	}
-
 	namespace, err := f.DefaultNamespace()
 	if err != nil {
 		return err
@@ -79,7 +97,7 @@ func RunExec(f *cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd
 		return err
 	}
 
-	pod, err := client.Pods(namespace).Get(podName)
+	pod, err := client.Pods(namespace).Get(p.podName)
 	if err != nil {
 		return err
 	}
@@ -88,14 +106,14 @@ func RunExec(f *cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd
 		glog.Fatalf("Unable to execute command because pod is not running. Current status=%v", pod.Status.Phase)
 	}
 
-	containerName := cmdutil.GetFlagString(cmd, "container")
+	containerName := p.containerName
 	if len(containerName) == 0 {
 		containerName = pod.Spec.Containers[0].Name
 	}
 
 	var stdin io.Reader
-	tty := cmdutil.GetFlagBool(cmd, "tty")
-	if cmdutil.GetFlagBool(cmd, "stdin") {
+	tty := p.tty
+	if p.stdin {
 		stdin = cmdIn
 		if tty {
 			if file, ok := cmdIn.(*os.File); ok {
@@ -135,11 +153,11 @@ func RunExec(f *cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd
 	}
 
 	req := client.RESTClient.Get().
-		Prefix("proxy").
-		Resource("nodes").
-		Name(pod.Spec.Host).
-		Suffix("exec", namespace, podName, containerName)
+		Resource("pods").
+		Name(pod.Name).
+		Namespace(namespace).
+		SubResource("exec").
+		Param("container", containerName)
 
-	e := remotecommand.New(req, config, args, stdin, cmdOut, cmdErr, tty)
-	return e.Execute()
+	return re.Execute(req, config, args, stdin, cmdOut, cmdErr, tty)
 }

pkg/kubectl/cmd/exec_test.go

+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 cmd
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+
+	"github.com/spf13/cobra"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+)
+
+type fakeRemoteExecutor struct {
+	req     *client.Request
+	execErr error
+}
+
+func (f *fakeRemoteExecutor) Execute(req *client.Request, config *client.Config, command []string, stdin io.Reader, stdout, stderr io.Writer, tty bool) error {
+	f.req = req
+	return f.execErr
+}
+
+func TestExec(t *testing.T) {
+
+	tests := []struct {
+		name, version, podPath, execPath, container string
+		nsInQuery                                   bool
+		pod                                         *api.Pod
+		execErr                                     bool
+	}{
+		{
+			name:      "v1beta1 - pod exec",
+			version:   "v1beta1",
+			podPath:   "/api/v1beta1/pods/foo",
+			execPath:  "/api/v1beta1/pods/foo/exec",
+			nsInQuery: true,
+			pod:       execPod(),
+		},
+		{
+			name:      "v1beta3 - pod exec",
+			version:   "v1beta3",
+			podPath:   "/api/v1beta3/namespaces/test/pods/foo",
+			execPath:  "/api/v1beta3/namespaces/test/pods/foo/exec",
+			nsInQuery: false,
+			pod:       execPod(),
+		},
+		{
+			name:      "v1beta3 - pod exec error",
+			version:   "v1beta3",
+			podPath:   "/api/v1beta3/namespaces/test/pods/foo",
+			execPath:  "/api/v1beta3/namespaces/test/pods/foo/exec",
+			nsInQuery: false,
+			pod:       execPod(),
+			execErr:   true,
+		},
+	}
+	for _, test := range tests {
+		f, tf, codec := NewAPIFactory()
+		tf.Client = &client.FakeRESTClient{
+			Codec: codec,
+			Client: client.HTTPClientFunc(func(req *http.Request) (*http.Response, error) {
+				switch p, m := req.URL.Path, req.Method; {
+				case p == test.podPath && m == "GET":
+					if test.nsInQuery {
+						if ns := req.URL.Query().Get("namespace"); ns != "test" {
+							t.Errorf("%s: did not get expected namespace: %s\n", test.name, ns)
+						}
+					}
+					body := objBody(codec, test.pod)
+					return &http.Response{StatusCode: 200, Body: body}, nil
+				default:
+					// Ensures no GET is performed when deleting by name
+					t.Errorf("%s: unexpected request: %#v\n%#v", test.name, req.URL, req)
+					return nil, nil
+				}
+			}),
+		}
+		tf.Namespace = "test"
+		tf.ClientConfig = &client.Config{Version: test.version}
+		bufOut := bytes.NewBuffer([]byte{})
+		bufErr := bytes.NewBuffer([]byte{})
+		bufIn := bytes.NewBuffer([]byte{})
+		ex := &fakeRemoteExecutor{}
+		if test.execErr {
+			ex.execErr = fmt.Errorf("exec error")
+		}
+		params := &execParams{
+			podName:       "foo",
+			containerName: "bar",
+		}
+		cmd := &cobra.Command{}
+		err := RunExec(f, cmd, bufIn, bufOut, bufErr, params, []string{"test", "command"}, ex)
+		if test.execErr && err != ex.execErr {
+			t.Errorf("%s: Unexpected exec error: %v", test.name, err)
+		}
+		if !test.execErr && ex.req.URL().Path != test.execPath {
+			t.Errorf("%s: Did not get expected path for exec request", test.name)
+		}
+		if !test.execErr && err != nil {
+			t.Errorf("%s: Unexpected error: %v", test.name, err)
+		}
+	}
+}
+
+func execPod() *api.Pod {
+	return &api.Pod{
+		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "10"},
+		Spec: api.PodSpec{
+			RestartPolicy: api.RestartPolicyAlways,
+			DNSPolicy:     api.DNSClusterFirst,
+			Containers: []api.Container{
+				{
+					Name: "bar",
+				},
+			},
+		},
+		Status: api.PodStatus{
+			Phase: api.PodRunning,
+		},
+	}
+}

pkg/kubectl/cmd/portforward.go

@@ -21,6 +21,7 @@ import (
 	"os/signal"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/portforward"
 	cmdutil "github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/cmd/util"
 	"github.com/golang/glog"
@@ -49,7 +50,7 @@ func NewCmdPortForward(f *cmdutil.Factory) *cobra.Command {
 		Long:    "Forward one or more local ports to a pod.",
 		Example: portforward_example,
 		Run: func(cmd *cobra.Command, args []string) {
-			err := RunPortForward(f, cmd, args)
+			err := RunPortForward(f, cmd, args, &defaultPortForwarder{})
 			cmdutil.CheckErr(err)
 		},
 	}
@@ -59,12 +60,25 @@ func NewCmdPortForward(f *cmdutil.Factory) *cobra.Command {
 	return cmd
 }
 
-func RunPortForward(f *cmdutil.Factory, cmd *cobra.Command, args []string) error {
+type portForwarder interface {
+	ForwardPorts(req *client.Request, config *client.Config, ports []string, stopChan <-chan struct{}) error
+}
+
+type defaultPortForwarder struct{}
+
+func (*defaultPortForwarder) ForwardPorts(req *client.Request, config *client.Config, ports []string, stopChan <-chan struct{}) error {
+	fw, err := portforward.New(req, config, ports, stopChan)
+	if err != nil {
+		return err
+	}
+	return fw.ForwardPorts()
+}
+
+func RunPortForward(f *cmdutil.Factory, cmd *cobra.Command, args []string, fw portForwarder) error {
 	podName := cmdutil.GetFlagString(cmd, "pod")
 	if len(podName) == 0 {
 		return cmdutil.UsageError(cmd, "POD is required for exec")
 	}
-
 	if len(args) < 1 {
 		return cmdutil.UsageError(cmd, "at least 1 PORT is required for port-forward")
 	}
@@ -104,15 +118,10 @@ func RunPortForward(f *cmdutil.Factory, cmd *cobra.Command, args []string) error
 	}()
 
 	req := client.RESTClient.Get().
-		Prefix("proxy").
-		Resource("nodes").
-		Name(pod.Spec.Host).
-		Suffix("portForward", namespace, podName)
-
-	pf, err := portforward.New(req, config, args, stopCh)
-	if err != nil {
-		return err
-	}
+		Resource("pods").
+		Namespace(namespace).
+		Name(pod.Name).
+		SubResource("portforward")
 
-	return pf.ForwardPorts()
+	return fw.ForwardPorts(req, config, args, stopCh)
 }

pkg/kubectl/cmd/portforward_test.go

+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 cmd
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/spf13/cobra"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+)
+
+type fakePortForwarder struct {
+	req   *client.Request
+	pfErr error
+}
+
+func (f *fakePortForwarder) ForwardPorts(req *client.Request, config *client.Config, ports []string, stopChan <-chan struct{}) error {
+	f.req = req
+	return f.pfErr
+}
+
+func TestPortForward(t *testing.T) {
+
+	tests := []struct {
+		name, version, podPath, pfPath, container string
+		nsInQuery                                 bool
+		pod                                       *api.Pod
+		pfErr                                     bool
+	}{
+		{
+			name:      "v1beta1 - pod portforward",
+			version:   "v1beta1",
+			podPath:   "/api/v1beta1/pods/foo",
+			pfPath:    "/api/v1beta1/pods/foo/portforward",
+			nsInQuery: true,
+			pod:       execPod(),
+		},
+		{
+			name:      "v1beta3 - pod portforward",
+			version:   "v1beta3",
+			podPath:   "/api/v1beta3/namespaces/test/pods/foo",
+			pfPath:    "/api/v1beta3/namespaces/test/pods/foo/portforward",
+			nsInQuery: false,
+			pod:       execPod(),
+		},
+		{
+			name:      "v1beta3 - pod portforward error",
+			version:   "v1beta3",
+			podPath:   "/api/v1beta3/namespaces/test/pods/foo",
+			pfPath:    "/api/v1beta3/namespaces/test/pods/foo/portforward",
+			nsInQuery: false,
+			pod:       execPod(),
+			pfErr:     true,
+		},
+	}
+	for _, test := range tests {
+		f, tf, codec := NewAPIFactory()
+		tf.Client = &client.FakeRESTClient{
+			Codec: codec,
+			Client: client.HTTPClientFunc(func(req *http.Request) (*http.Response, error) {
+				switch p, m := req.URL.Path, req.Method; {
+				case p == test.podPath && m == "GET":
+					if test.nsInQuery {
+						if ns := req.URL.Query().Get("namespace"); ns != "test" {
+							t.Errorf("%s: did not get expected namespace: %s\n", test.name, ns)
+						}
+					}
+					body := objBody(codec, test.pod)
+					return &http.Response{StatusCode: 200, Body: body}, nil
+				default:
+					// Ensures no GET is performed when deleting by name
+					t.Errorf("%s: unexpected request: %#v\n%#v", test.name, req.URL, req)
+					return nil, nil
+				}
+			}),
+		}
+		tf.Namespace = "test"
+		tf.ClientConfig = &client.Config{Version: test.version}
+		ff := &fakePortForwarder{}
+		if test.pfErr {
+			ff.pfErr = fmt.Errorf("pf error")
+		}
+		cmd := &cobra.Command{}
+		podPtr := cmd.Flags().StringP("pod", "p", "", "Pod name")
+		*podPtr = "foo"
+		err := RunPortForward(f, cmd, []string{":5000", ":1000"}, ff)
+		if test.pfErr && err != ff.pfErr {
+			t.Errorf("%s: Unexpected exec error: %v", test.name, err)
+		}
+		if !test.pfErr && ff.req.URL().Path != test.pfPath {
+			t.Errorf("%s: Did not get expected path for portforward request", test.name)
+		}
+		if !test.pfErr && err != nil {
+			t.Errorf("%s: Unexpected error: %v", test.name, err)
+		}
+	}
+}