Be able to specify the timeout to wait for pod for kubectl logs/attach

hack/verify-flags/known-flags.txt

@@ -516,6 +516,7 @@ pod-infra-container-image
 pod-manifest-path
 pod-network-cidr
 pod-running
+pod-running-timeout
 pods-per-core
 policy-config-file
 poll-interval

pkg/kubectl/cmd/attach.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"net/url"
+	"time"
 
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
@@ -55,6 +56,11 @@ var (
 		`)
 )
 
+const (
+	defaultPodAttachTimeout = 60 * time.Second
+	defaultPodLogsTimeout   = 20 * time.Second
+)
+
 func NewCmdAttach(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer) *cobra.Command {
 	options := &AttachOptions{
 		StreamOptions: StreamOptions{
@@ -76,7 +82,7 @@ func NewCmdAttach(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer)
 			cmdutil.CheckErr(options.Run())
 		},
 	}
-	// TODO support UID
+	cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodAttachTimeout)
 	cmd.Flags().StringVarP(&options.ContainerName, "container", "c", "", "Container name. If omitted, the first container in the pod will be chosen")
 	cmd.Flags().BoolVarP(&options.Stdin, "stdin", "i", false, "Pass stdin to the container")
 	cmd.Flags().BoolVarP(&options.TTY, "tty", "t", false, "Stdin is a TTY")
@@ -114,9 +120,10 @@ type AttachOptions struct {
 
 	Pod *api.Pod
 
-	Attach    RemoteAttach
+	Attach        RemoteAttach
-	PodClient coreclient.PodsGetter
+	PodClient     coreclient.PodsGetter
-	Config    *restclient.Config
+	GetPodTimeout time.Duration
+	Config        *restclient.Config
 }
 
 // Complete verifies command line arguments and loads data from the command environment
@@ -133,6 +140,11 @@ func (p *AttachOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, argsIn [
 		return err
 	}
 
+	p.GetPodTimeout, err = cmdutil.GetPodRunningTimeoutFlag(cmd)
+	if err != nil {
+		return cmdutil.UsageError(cmd, err.Error())
+	}
+
 	mapper, typer := f.Object()
 	builder := resource.NewBuilder(mapper, typer, resource.ClientMapperFunc(f.ClientForMapping), f.Decoder(true)).
 		NamespaceParam(namespace).DefaultNamespace()
@@ -149,7 +161,7 @@ func (p *AttachOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, argsIn [
 		return err
 	}
 
-	attachablePod, err := f.AttachablePodForObject(obj)
+	attachablePod, err := f.AttachablePodForObject(obj, p.GetPodTimeout)
 	if err != nil {
 		return err
 	}

pkg/kubectl/cmd/attach_test.go

@@ -24,6 +24,7 @@ import (
 	"net/url"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -34,6 +35,7 @@ import (
 	"k8s.io/client-go/rest/fake"
 	"k8s.io/kubernetes/pkg/api"
 	cmdtesting "k8s.io/kubernetes/pkg/kubectl/cmd/testing"
+	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/util/term"
 )
 
@@ -57,18 +59,21 @@ func TestPodAndContainerAttach(t *testing.T) {
 		expectError       bool
 		expectedPod       string
 		expectedContainer string
+		timeout           time.Duration
 		obj               runtime.Object
 	}{
 		{
 			p:           &AttachOptions{},
 			expectError: true,
 			name:        "empty",
+			timeout:     1,
 		},
 		{
 			p:           &AttachOptions{},
 			args:        []string{"one", "two", "three"},
 			expectError: true,
 			name:        "too many args",
+			timeout:     2,
 		},
 		{
 			p:           &AttachOptions{},
@@ -76,6 +81,7 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectedPod: "foo",
 			name:        "no container, no flags",
 			obj:         attachPod(),
+			timeout:     defaultPodLogsTimeout,
 		},
 		{
 			p:                 &AttachOptions{StreamOptions: StreamOptions{ContainerName: "bar"}},
@@ -84,6 +90,7 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectedContainer: "bar",
 			name:              "container in flag",
 			obj:               attachPod(),
+			timeout:           10000000,
 		},
 		{
 			p:                 &AttachOptions{StreamOptions: StreamOptions{ContainerName: "initfoo"}},
@@ -92,6 +99,7 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectedContainer: "initfoo",
 			name:              "init container in flag",
 			obj:               attachPod(),
+			timeout:           30,
 		},
 		{
 			p:           &AttachOptions{StreamOptions: StreamOptions{ContainerName: "bar"}},
@@ -99,6 +107,7 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectError: true,
 			name:        "non-existing container in flag",
 			obj:         attachPod(),
+			timeout:     10,
 		},
 		{
 			p:           &AttachOptions{},
@@ -106,6 +115,7 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectedPod: "foo",
 			name:        "no container, no flags, pods and name",
 			obj:         attachPod(),
+			timeout:     10000,
 		},
 		{
 			p:           &AttachOptions{},
@@ -113,6 +123,16 @@ func TestPodAndContainerAttach(t *testing.T) {
 			expectedPod: "foo",
 			name:        "no container, no flags, pod/name",
 			obj:         attachPod(),
+			timeout:     1,
+		},
+		{
+			p:           &AttachOptions{},
+			args:        []string{"pod/foo"},
+			expectedPod: "foo",
+			name:        "invalid get pod timeout value",
+			obj:         attachPod(),
+			expectError: true,
+			timeout:     0,
 		},
 	}
 
@@ -133,6 +153,8 @@ func TestPodAndContainerAttach(t *testing.T) {
 
 		cmd := &cobra.Command{}
 		options := test.p
+		cmdutil.AddPodRunningTimeoutFlag(cmd, test.timeout)
+
 		err := options.Complete(f, cmd, test.args)
 		if test.expectError && err == nil {
 			t.Errorf("unexpected non-error (%s)", test.name)
@@ -227,9 +249,11 @@ func TestAttach(t *testing.T) {
 				Out:           bufOut,
 				Err:           bufErr,
 			},
-			Attach: remoteAttach,
+			Attach:        remoteAttach,
+			GetPodTimeout: 1000,
 		}
 		cmd := &cobra.Command{}
+		cmdutil.AddPodRunningTimeoutFlag(cmd, 1000)
 		if err := params.Complete(f, cmd, []string{"foo"}); err != nil {
 			t.Fatal(err)
 		}
@@ -310,9 +334,11 @@ func TestAttachWarnings(t *testing.T) {
 				Stdin:         test.stdin,
 				TTY:           test.tty,
 			},
-			Attach: ex,
+			Attach:        ex,
+			GetPodTimeout: 1000,
 		}
 		cmd := &cobra.Command{}
+		cmdutil.AddPodRunningTimeoutFlag(cmd, 1000)
 		if err := params.Complete(f, cmd, []string{"foo"}); err != nil {
 			t.Fatal(err)
 		}

pkg/kubectl/cmd/clusterinfo_dump.go

@@ -46,6 +46,7 @@ func NewCmdClusterInfoDump(f cmdutil.Factory, cmdOut io.Writer) *cobra.Command {
 	cmd.Flags().String("output-directory", "", i18n.T("Where to output the files.  If empty or '-' uses stdout, otherwise creates a directory hierarchy in that directory"))
 	cmd.Flags().StringSlice("namespaces", []string{}, "A comma separated list of namespaces to dump.")
 	cmd.Flags().Bool("all-namespaces", false, "If true, dump all namespaces.  If true, --namespaces is ignored.")
+	cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodLogsTimeout)
 	return cmd
 }
 
@@ -88,6 +89,11 @@ func setupOutputWriter(cmd *cobra.Command, defaultWriter io.Writer, filename str
 }
 
 func dumpClusterInfo(f cmdutil.Factory, cmd *cobra.Command, args []string, out io.Writer) error {
+	timeout, err := cmdutil.GetPodRunningTimeoutFlag(cmd)
+	if err != nil {
+		return cmdutil.UsageError(cmd, err.Error())
+	}
+
 	clientset, err := f.ClientSet()
 	if err != nil {
 		return err
@@ -190,7 +196,7 @@ func dumpClusterInfo(f cmdutil.Factory, cmd *cobra.Command, args []string, out i
 			pod := &pods.Items[ix]
 			writer := setupOutputWriter(cmd, out, path.Join(namespace, pod.Name, "logs.txt"))
 			writer.Write([]byte(fmt.Sprintf("==== START logs for %s/%s ====\n", pod.Namespace, pod.Name)))
-			request, err := f.LogsForObject(pod, &api.PodLogOptions{})
+			request, err := f.LogsForObject(pod, &api.PodLogOptions{}, timeout)
 			if err != nil {
 				return err
 			}

pkg/kubectl/cmd/logs.go

@@ -80,7 +80,8 @@ type LogsOptions struct {
 	Decoder      runtime.Decoder
 
 	Object        runtime.Object
-	LogsForObject func(object, options runtime.Object) (*restclient.Request, error)
+	GetPodTimeout time.Duration
+	LogsForObject func(object, options runtime.Object, timeout time.Duration) (*restclient.Request, error)
 
 	Out io.Writer
 }
@@ -113,10 +114,10 @@ func NewCmdLogs(f cmdutil.Factory, out io.Writer) *cobra.Command {
 	cmd.Flags().String("since-time", "", i18n.T("Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used."))
 	cmd.Flags().Duration("since", 0, "Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.")
 	cmd.Flags().StringP("container", "c", "", "Print the logs of this container")
-
 	cmd.Flags().Bool("interactive", false, "If true, prompt the user for input when required.")
 	cmd.Flags().MarkDeprecated("interactive", "This flag is no longer respected and there is no replacement.")
 	cmdutil.AddInclude3rdPartyFlags(cmd)
+	cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodLogsTimeout)
 	cmd.Flags().StringP("selector", "l", "", "Selector (label query) to filter on.")
 	return cmd
 }
@@ -173,6 +174,10 @@ func (o *LogsOptions) Complete(f cmdutil.Factory, out io.Writer, cmd *cobra.Comm
 		sec := int64(math.Ceil(float64(sinceSeconds) / float64(time.Second)))
 		logOptions.SinceSeconds = &sec
 	}
+	o.GetPodTimeout, err = cmdutil.GetPodRunningTimeoutFlag(cmd)
+	if err != nil {
+		return err
+	}
 	o.Options = logOptions
 	o.LogsForObject = f.LogsForObject
 	o.ClientMapper = resource.ClientMapperFunc(f.ClientForMapping)
@@ -243,7 +248,7 @@ func (o LogsOptions) RunLogs() error {
 }
 
 func (o LogsOptions) getLogs(obj runtime.Object) error {
-	req, err := o.LogsForObject(obj, o.Options)
+	req, err := o.LogsForObject(obj, o.Options, o.GetPodTimeout)
 	if err != nil {
 		return err
 	}

pkg/kubectl/cmd/run.go

@@ -21,9 +21,8 @@ import (
 	"io"
 	"os"
 
-	"github.com/spf13/cobra"
-
 	"github.com/docker/distribution/reference"
+	"github.com/spf13/cobra"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -108,6 +107,7 @@ func NewCmdRun(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer) *co
 	cmdutil.AddApplyAnnotationFlags(cmd)
 	cmdutil.AddRecordFlag(cmd)
 	cmdutil.AddInclude3rdPartyFlags(cmd)
+	cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodAttachTimeout)
 	return cmd
 }
 
@@ -149,6 +149,11 @@ func Run(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd *cobr
 		return cmdutil.UsageError(cmd, "NAME is required for run")
 	}
 
+	timeout, err := cmdutil.GetPodRunningTimeoutFlag(cmd)
+	if err != nil {
+		return cmdutil.UsageError(cmd, err.Error())
+	}
+
 	// validate image name
 	imageName := cmdutil.GetFlagString(cmd, "image")
 	validImageRef := reference.ReferenceRegexp.MatchString(imageName)
@@ -287,8 +292,8 @@ func Run(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd *cobr
 				TTY:   tty,
 				Quiet: quiet,
 			},
-
+			GetPodTimeout: timeout,
-			CommandName: cmd.Parent().CommandPath() + " attach",
+			CommandName:   cmd.Parent().CommandPath() + " attach",
 
 			Attach: &DefaultRemoteAttach{},
 		}
@@ -304,7 +309,7 @@ func Run(f cmdutil.Factory, cmdIn io.Reader, cmdOut, cmdErr io.Writer, cmd *cobr
 		}
 		opts.PodClient = clientset.Core()
 
-		attachablePod, err := f.AttachablePodForObject(obj)
+		attachablePod, err := f.AttachablePodForObject(obj, opts.GetPodTimeout)
 		if err != nil {
 			return err
 		}
@@ -446,7 +451,7 @@ func handleAttachPod(f cmdutil.Factory, podClient coreclient.PodsGetter, ns, nam
 		return err
 	}
 	if pod.Status.Phase == api.PodSucceeded || pod.Status.Phase == api.PodFailed {
-		req, err := f.LogsForObject(pod, &api.PodLogOptions{Container: ctrName})
+		req, err := f.LogsForObject(pod, &api.PodLogOptions{Container: ctrName}, opts.GetPodTimeout)
 		if err != nil {
 			return err
 		}
@@ -467,7 +472,7 @@ func handleAttachPod(f cmdutil.Factory, podClient coreclient.PodsGetter, ns, nam
 	stderr := opts.Err
 	if err := opts.Run(); err != nil {
 		fmt.Fprintf(stderr, "Error attaching, falling back to logs: %v\n", err)
-		req, err := f.LogsForObject(pod, &api.PodLogOptions{Container: ctrName})
+		req, err := f.LogsForObject(pod, &api.PodLogOptions{Container: ctrName}, opts.GetPodTimeout)
 		if err != nil {
 			return err
 		}

pkg/kubectl/cmd/testing/fake.go

@@ -376,7 +376,7 @@ func (f *FakeFactory) LabelsForObject(runtime.Object) (map[string]string, error)
 	return nil, nil
 }
 
-func (f *FakeFactory) LogsForObject(object, options runtime.Object) (*restclient.Request, error) {
+func (f *FakeFactory) LogsForObject(object, options runtime.Object, timeout time.Duration) (*restclient.Request, error) {
 	return nil, nil
 }
 
@@ -423,7 +423,7 @@ func (f *FakeFactory) CanBeAutoscaled(schema.GroupKind) error {
 	return nil
 }
 
-func (f *FakeFactory) AttachablePodForObject(ob runtime.Object) (*api.Pod, error) {
+func (f *FakeFactory) AttachablePodForObject(ob runtime.Object, timeout time.Duration) (*api.Pod, error) {
 	return nil, nil
 }
 
@@ -613,7 +613,7 @@ func (f *fakeAPIFactory) Printer(mapping *meta.RESTMapping, options printers.Pri
 	return f.tf.Printer, f.tf.Err
 }
 
-func (f *fakeAPIFactory) LogsForObject(object, options runtime.Object) (*restclient.Request, error) {
+func (f *fakeAPIFactory) LogsForObject(object, options runtime.Object, timeout time.Duration) (*restclient.Request, error) {
 	c, err := f.ClientSet()
 	if err != nil {
 		panic(err)
@@ -635,7 +635,7 @@ func (f *fakeAPIFactory) LogsForObject(object, options runtime.Object) (*restcli
 	}
 }
 
-func (f *fakeAPIFactory) AttachablePodForObject(object runtime.Object) (*api.Pod, error) {
+func (f *fakeAPIFactory) AttachablePodForObject(object runtime.Object, timeout time.Duration) (*api.Pod, error) {
 	switch t := object.(type) {
 	case *api.Pod:
 		return t, nil

pkg/kubectl/cmd/util/factory.go

@@ -198,7 +198,7 @@ type ObjectMappingFactory interface {
 	Describer(mapping *meta.RESTMapping) (printers.Describer, error)
 
 	// LogsForObject returns a request for the logs associated with the provided object
-	LogsForObject(object, options runtime.Object) (*restclient.Request, error)
+	LogsForObject(object, options runtime.Object, timeout time.Duration) (*restclient.Request, error)
 	// Returns a Scaler for changing the size of the specified RESTMapping type or an error
 	Scaler(mapping *meta.RESTMapping) (kubectl.Scaler, error)
 	// Returns a Reaper for gracefully shutting down resources.
@@ -211,7 +211,7 @@ type ObjectMappingFactory interface {
 	StatusViewer(mapping *meta.RESTMapping) (kubectl.StatusViewer, error)
 
 	// AttachablePodForObject returns the pod to which to attach given an object.
-	AttachablePodForObject(object runtime.Object) (*api.Pod, error)
+	AttachablePodForObject(object runtime.Object, timeout time.Duration) (*api.Pod, error)
 
 	// Returns a schema that can validate objects stored on disk.
 	Validator(validate bool, cacheDir string) (validation.Schema, error)

pkg/kubectl/cmd/util/factory_object_mapping.go

@@ -208,7 +208,7 @@ func genericDescriber(clientAccessFactory ClientAccessFactory, mapping *meta.RES
 	return printersinternal.GenericDescriberFor(mapping, dynamicClient, eventsClient), nil
 }
 
-func (f *ring1Factory) LogsForObject(object, options runtime.Object) (*restclient.Request, error) {
+func (f *ring1Factory) LogsForObject(object, options runtime.Object, timeout time.Duration) (*restclient.Request, error) {
 	clientset, err := f.clientAccessFactory.ClientSetForVersion(nil)
 	if err != nil {
 		return nil, err
@@ -265,7 +265,7 @@ func (f *ring1Factory) LogsForObject(object, options runtime.Object) (*restclien
 	}
 
 	sortBy := func(pods []*v1.Pod) sort.Interface { return controller.ByLogging(pods) }
-	pod, numPods, err := GetFirstPod(clientset.Core(), namespace, selector, 20*time.Second, sortBy)
+	pod, numPods, err := GetFirstPod(clientset.Core(), namespace, selector, timeout, sortBy)
 	if err != nil {
 		return nil, err
 	}
@@ -325,7 +325,7 @@ func (f *ring1Factory) StatusViewer(mapping *meta.RESTMapping) (kubectl.StatusVi
 	return kubectl.StatusViewerFor(mapping.GroupVersionKind.GroupKind(), clientset)
 }
 
-func (f *ring1Factory) AttachablePodForObject(object runtime.Object) (*api.Pod, error) {
+func (f *ring1Factory) AttachablePodForObject(object runtime.Object, timeout time.Duration) (*api.Pod, error) {
 	clientset, err := f.clientAccessFactory.ClientSetForVersion(nil)
 	if err != nil {
 		return nil, err
@@ -374,7 +374,7 @@ func (f *ring1Factory) AttachablePodForObject(object runtime.Object) (*api.Pod,
 	}
 
 	sortBy := func(pods []*v1.Pod) sort.Interface { return sort.Reverse(controller.ActivePods(pods)) }
-	pod, _, err := GetFirstPod(clientset.Core(), namespace, selector, 1*time.Minute, sortBy)
+	pod, _, err := GetFirstPod(clientset.Core(), namespace, selector, timeout, sortBy)
 	return pod, err
 }
 

pkg/kubectl/cmd/util/factory_object_mapping_test.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
+	"time"
 )
 
 type fakeClientAccessFactory struct {
@@ -146,7 +147,7 @@ func TestLogsForObject(t *testing.T) {
 	for _, test := range tests {
 		caf := newFakeClientAccessFactory(test.pods)
 		omf := NewObjectMappingFactory(caf)
-		_, err := omf.LogsForObject(test.obj, test.opts)
+		_, err := omf.LogsForObject(test.obj, test.opts, 20*time.Second)
 		if err != nil {
 			t.Errorf("%s: unexpected error: %v", test.name, err)
 			continue

pkg/kubectl/cmd/util/helpers.go

@@ -387,6 +387,14 @@ func GetFlagDuration(cmd *cobra.Command, flag string) time.Duration {
 	return d
 }
 
+func GetPodRunningTimeoutFlag(cmd *cobra.Command) (time.Duration, error) {
+	timeout := GetFlagDuration(cmd, "pod-running-timeout")
+	if timeout <= 0 {
+		return timeout, fmt.Errorf("--pod-running-timeout must be higher than zero")
+	}
+	return timeout, nil
+}
+
 func AddValidateFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool("validate", true, "If true, use a schema to validate the input before sending it")
 	cmd.Flags().String("schema-cache-dir", fmt.Sprintf("~/%s/%s", clientcmd.RecommendedHomeDir, clientcmd.RecommendedSchemaName), fmt.Sprintf("If non-empty, load/store cached API schemas in this directory, default is '$HOME/%s/%s'", clientcmd.RecommendedHomeDir, clientcmd.RecommendedSchemaName))
@@ -403,6 +411,10 @@ func AddDryRunFlag(cmd *cobra.Command) {
 	cmd.Flags().Bool("dry-run", false, "If true, only print the object that would be sent, without sending it.")
 }
 
+func AddPodRunningTimeoutFlag(cmd *cobra.Command, defaultTimeout time.Duration) {
+	cmd.Flags().Duration("pod-running-timeout", defaultTimeout, "The length of time (like 5s, 2m, or 3h, higher than zero) to wait until at least one pod is running")
+}
+
 func AddApplyAnnotationFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(ApplyAnnotationsFlag, false, "If true, the configuration of current object will be saved in its annotation. Otherwise, the annotation will be unchanged. This flag is useful when you want to perform kubectl apply on this object in the future.")
 }