Support extended pod logging options

Increase the supported controls on pod logging. Add validaiton to pod
log options. Ensure the Kubelet is using a consistent, structured way to
process pod log arguments.

Add ?sinceSeconds=<durationInSeconds>, &sinceTime=<RFC3339>, ?timestamps=<bool>,
?tailLines=<number>, and ?limitBytes=<number>

pkg/registry/pod/etcd/etcd.go

@@ -27,6 +27,7 @@ import (
 	etcderr "k8s.io/kubernetes/pkg/api/errors/etcd"
 	"k8s.io/kubernetes/pkg/api/rest"
 	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/api/validation"
 	"k8s.io/kubernetes/pkg/capabilities"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/fields"
@@ -211,6 +212,7 @@ func (r *StatusREST) Update(ctx api.Context, obj runtime.Object) (runtime.Object
 }
 
 // LogREST implements the log endpoint for a Pod
+// TODO: move me into pod/rest - I'm generic to store type via ResourceGetter
 type LogREST struct {
 	store       *etcdgeneric.Etcd
 	kubeletConn client.ConnectionInfoGetter
@@ -231,6 +233,9 @@ func (r *LogREST) Get(ctx api.Context, name string, opts runtime.Object) (runtim
 	if !ok {
 		return nil, fmt.Errorf("Invalid options object: %#v", opts)
 	}
+	if errs := validation.ValidatePodLogOptions(logOpts); len(errs) > 0 {
+		return nil, errors.NewInvalid("podlogs", name, errs)
+	}
 	location, transport, err := pod.LogLocation(r.store, r.kubeletConn, ctx, name, logOpts)
 	if err != nil {
 		return nil, err
@@ -249,6 +254,7 @@ func (r *LogREST) NewGetOptions() (runtime.Object, bool, string) {
 }
 
 // ProxyREST implements the proxy subresource for a Pod
+// TODO: move me into pod/rest - I'm generic to store type via ResourceGetter
 type ProxyREST struct {
 	store *etcdgeneric.Etcd
 }
@@ -291,6 +297,7 @@ func (r *ProxyREST) Connect(ctx api.Context, id string, opts runtime.Object) (re
 var upgradeableMethods = []string{"GET", "POST"}
 
 // AttachREST implements the attach subresource for a Pod
+// TODO: move me into pod/rest - I'm generic to store type via ResourceGetter
 type AttachREST struct {
 	store       *etcdgeneric.Etcd
 	kubeletConn client.ConnectionInfoGetter
@@ -328,6 +335,7 @@ func (r *AttachREST) ConnectMethods() []string {
 }
 
 // ExecREST implements the exec subresource for a Pod
+// TODO: move me into pod/rest - I'm generic to store type via ResourceGetter
 type ExecREST struct {
 	store       *etcdgeneric.Etcd
 	kubeletConn client.ConnectionInfoGetter
@@ -365,6 +373,7 @@ func (r *ExecREST) ConnectMethods() []string {
 }
 
 // PortForwardREST implements the portforward subresource for a Pod
+// TODO: move me into pod/rest - I'm generic to store type via ResourceGetter
 type PortForwardREST struct {
 	store       *etcdgeneric.Etcd
 	kubeletConn client.ConnectionInfoGetter

pkg/registry/pod/etcd/etcd_test.go

@@ -734,3 +734,21 @@ func TestEtcdUpdateStatus(t *testing.T) {
 		t.Errorf("unexpected object: %s", util.ObjectDiff(&expected, podOut))
 	}
 }
+
+func TestPodLogValidates(t *testing.T) {
+	etcdStorage, _ := registrytest.NewEtcdStorage(t, "")
+	storage := NewStorage(etcdStorage, false, nil)
+
+	negativeOne := int64(-1)
+	testCases := []*api.PodLogOptions{
+		{SinceSeconds: &negativeOne},
+		{TailLines: &negativeOne},
+	}
+
+	for _, tc := range testCases {
+		_, err := storage.Log.Get(api.NewDefaultContext(), "test", tc)
+		if !errors.IsInvalid(err) {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	}
+}

pkg/registry/pod/strategy.go

@@ -21,6 +21,8 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"strconv"
+	"time"
 
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/errors"
@@ -253,6 +255,21 @@ func LogLocation(getter ResourceGetter, connInfo client.ConnectionInfoGetter, ct
 	if opts.Previous {
 		params.Add("previous", "true")
 	}
+	if opts.Timestamps {
+		params.Add("timestamps", "true")
+	}
+	if opts.SinceSeconds != nil {
+		params.Add("sinceSeconds", strconv.FormatInt(*opts.SinceSeconds, 10))
+	}
+	if opts.SinceTime != nil {
+		params.Add("sinceTime", opts.SinceTime.Format(time.RFC3339))
+	}
+	if opts.TailLines != nil {
+		params.Add("tailLines", strconv.FormatInt(*opts.TailLines, 10))
+	}
+	if opts.LimitBytes != nil {
+		params.Add("limitBytes", strconv.FormatInt(*opts.LimitBytes, 10))
+	}
 	loc := &url.URL{
 		Scheme:   nodeScheme,
 		Host:     fmt.Sprintf("%s:%d", nodeHost, nodePort),