Merge pull request #1318 from jhadvig/k8s_log_retrieval

Adding endpoint for log retrieval on the minion

pkg/kubelet/dockertools/docker.go

@@ -25,6 +25,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/fsouza/go-dockerclient"
@@ -46,6 +47,7 @@ type DockerInterface interface {
 	StartContainer(id string, hostConfig *docker.HostConfig) error
 	StopContainer(id string, timeout uint) error
 	PullImage(opts docker.PullImageOptions, auth docker.AuthConfiguration) error
+	Logs(opts docker.LogsOptions) error
 }
 
 // DockerID is an ID of docker container. It is a type to make it clear when we're working with docker container Ids
@@ -202,6 +204,30 @@ func GetRecentDockerContainersWithNameAndUUID(client DockerInterface, podFullNam
 	return result, nil
 }
 
+// GetKubeletDockerContainerLogs returns logs of specific container
+// By default the function will return snapshot of the container log
+// Log streaming is possible if 'follow' param is set to true
+// Log tailing is possible when number of tailed lines are set and only if 'follow' is false
+func GetKubeletDockerContainerLogs(client DockerInterface, containerID, tail string, follow bool, writer io.Writer) (err error) {
+	opts := docker.LogsOptions{
+		Container:    containerID,
+		Stdout:       true,
+		Stderr:       true,
+		OutputStream: writer,
+		ErrorStream:  writer,
+		Timestamps:   true,
+		RawTerminal:  true,
+		Follow:       follow,
+	}
+
+	if !follow {
+		opts.Tail = tail
+	}
+
+	err = client.Logs(opts)
+	return
+}
+
 // ErrNoContainersInPod is returned when there are no containers for a given pod
 var ErrNoContainersInPod = errors.New("no containers exist for this pod")
 

pkg/kubelet/dockertools/fake_docker_client.go

@@ -111,6 +111,15 @@ func (f *FakeDockerClient) StopContainer(id string, timeout uint) error {
 	return f.Err
 }
 
+// Logs is a test-spy implementation of DockerInterface.Logs.
+// It adds an entry "logs" to the internal method call record.
+func (f *FakeDockerClient) Logs(opts docker.LogsOptions) error {
+	f.Lock()
+	defer f.Unlock()
+	f.called = append(f.called, "logs")
+	return f.Err
+}
+
 // PullImage is a test-spy implementation of DockerInterface.StopContainer.
 // It adds an entry "pull" to the internal method call record.
 func (f *FakeDockerClient) PullImage(opts docker.PullImageOptions, auth docker.AuthConfiguration) error {

pkg/kubelet/handlers.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"net"
 	"strconv"
+	"net/http"
+	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
@@ -95,3 +97,21 @@ func (h *httpActionHandler) Run(podFullName, uuid string, container *api.Contain
 	_, err := h.client.Get(url)
 	return err
 }
+
+// FlushWriter provides wrapper for responseWriter with HTTP streaming capabilities
+type FlushWriter struct {
+	flusher http.Flusher
+	writer io.Writer
+}
+
+// Write is a FlushWriter implementation of the io.Writer that sends any buffered data to the client.
+func (fw *FlushWriter) Write(p []byte) (n int, err error) {
+	n, err = fw.writer.Write(p)
+	if err != nil {
+		return
+	}
+	if fw.flusher != nil {
+		fw.flusher.Flush()
+	}
+	return
+}

pkg/kubelet/kubelet.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
@@ -746,6 +747,20 @@ func (kl *Kubelet) statsFromContainerPath(containerPath string, req *info.Contai
 	return cinfo, nil
 }
 
+// GetKubeletContainerLogs returns logs from the container
+func (kl *Kubelet) GetKubeletContainerLogs(podFullName, containerName, tail string, follow bool, writer io.Writer) error {
+	dockerContainers, err := dockertools.GetKubeletDockerContainers(kl.dockerClient)
+	if err != nil {
+		return err
+	}
+	var uuid string
+	dockerContainer, found, _ := dockerContainers.FindPodContainer(podFullName, uuid, containerName)
+	if !found {
+		return fmt.Errorf("container not found (%s)\n", containerName)
+	}
+	return dockertools.GetKubeletDockerContainerLogs(kl.dockerClient, dockerContainer.ID, tail , follow, writer)
+}
+
 // GetPodInfo returns information from Docker about the containers in a pod
 func (kl *Kubelet) GetPodInfo(podFullName, uuid string) (api.PodInfo, error) {
 	return dockertools.GetDockerPodInfo(kl.dockerClient, podFullName, uuid)

pkg/kubelet/server.go

@@ -68,6 +68,7 @@ type HostInterface interface {
 	GetMachineInfo() (*info.MachineInfo, error)
 	GetPodInfo(name, uuid string) (api.PodInfo, error)
 	RunInContainer(name, uuid, container string, cmd []string) ([]byte, error)
+	GetKubeletContainerLogs(podFullName, containerName, tail string, follow bool, writer io.Writer) error
 	ServeLogs(w http.ResponseWriter, req *http.Request)
 }
 
@@ -92,6 +93,7 @@ func (s *Server) InstallDefaultHandlers() {
 	s.mux.HandleFunc("/logs/", s.handleLogs)
 	s.mux.HandleFunc("/spec/", s.handleSpec)
 	s.mux.HandleFunc("/run/", s.handleRun)
+	s.mux.HandleFunc("/containerLogs/", s.handleContainerLogs)
 }
 
 // error serializes an error object into an HTTP response.
@@ -143,7 +145,54 @@ func (s *Server) handleContainers(w http.ResponseWriter, req *http.Request) {
 
 }
 
-// handlePodInfo handles podInfo requests against the Kubelet.
+// handleContainerLogs handles containerLogs request against the Kubelet
+func (s *Server) handleContainerLogs(w http.ResponseWriter, req *http.Request) {
+	defer req.Body.Close()
+	u, err := url.ParseRequestURI(req.RequestURI)
+	if err != nil {
+		s.error(w, err)
+		return
+	}
+	parts := strings.Split(u.Path, "/")
+
+	var podID, containerName string
+	if len(parts) == 4 {
+		podID = parts[2]
+		containerName = parts[3]
+	} else {
+		http.Error(w, "Unexpected path for command running", http.StatusBadRequest)
+		return
+	}
+
+	if len(podID) == 0 {
+		http.Error(w, `{"message": "Missing podID."}`, http.StatusBadRequest)
+		return
+	}
+	if len(containerName) == 0 {
+		http.Error(w, `{"message": "Missing container name."}`, http.StatusBadRequest)
+		return
+	}
+	
+	uriValues := u.Query()
+	follow, _ := strconv.ParseBool(uriValues.Get("follow"))
+	tail := uriValues.Get("tail")
+
+	podFullName := GetPodFullName(&Pod{Name: podID, Namespace: "etcd"})
+
+	fw := FlushWriter{writer: w}
+	if flusher, ok := w.(http.Flusher); ok {
+		fw.flusher = flusher
+	}
+	w.Header().Set("Transfer-Encoding", "chunked")
+	w.WriteHeader(http.StatusOK)
+	err = s.host.GetKubeletContainerLogs(podFullName, containerName, tail, follow, &fw)
+	if err != nil {
+		s.error(w, err)
+		return
+	}
+}
+
+// handlePodInfo handles podInfo requests against the Kubelet
 func (s *Server) handlePodInfo(w http.ResponseWriter, req *http.Request) {
 	u, err := url.ParseRequestURI(req.RequestURI)
 	if err != nil {

pkg/kubelet/server_test.go

@@ -27,6 +27,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
@@ -41,6 +42,7 @@ type fakeKubelet struct {
 	machineInfoFunc    func() (*info.MachineInfo, error)
 	logFunc            func(w http.ResponseWriter, req *http.Request)
 	runFunc            func(podFullName, uuid, containerName string, cmd []string) ([]byte, error)
+	containerLogsFunc  func(podFullName, containerName, tail string, follow bool, writer io.Writer)  error
 }
 
 func (fk *fakeKubelet) GetPodInfo(name, uuid string) (api.PodInfo, error) {
@@ -63,6 +65,10 @@ func (fk *fakeKubelet) ServeLogs(w http.ResponseWriter, req *http.Request) {
 	fk.logFunc(w, req)
 }
 
+func (fk *fakeKubelet) GetKubeletContainerLogs(podFullName, containerName, tail string, follow bool, writer io.Writer) error {
+	return fk.containerLogsFunc(podFullName, containerName, tail, follow, writer)
+}
+
 func (fk *fakeKubelet) RunInContainer(podFullName, uuid, containerName string, cmd []string) ([]byte, error) {
 	return fk.runFunc(podFullName, uuid, containerName, cmd)
 }
@@ -347,3 +353,121 @@ func TestServeRunInContainerWithUUID(t *testing.T) {
 		t.Errorf("expected %s, got %s", output, result)
 	}
 }
+
+func TestContainerLogs(t *testing.T) {
+    fw := newServerTest()
+    output := "foo bar"
+    podName := "foo"
+    expectedPodName := podName + ".etcd"
+    expectedContainerName := "baz"
+    expectedTail := ""
+    expectedFollow := false
+    // expected := api.Container{"goodpod": docker.Container{ID: "myContainerID"}}
+    fw.fakeKubelet.containerLogsFunc = func(podFullName, containerName, tail string, follow bool, writer io.Writer) error {
+            if podFullName != expectedPodName {
+                    t.Errorf("expected %s, got %s", expectedPodName, podFullName)
+            }
+			if containerName != expectedContainerName {
+				t.Errorf("expected %s, got %s", expectedContainerName, containerName)
+			}
+            if tail != expectedTail {
+                    t.Errorf("expected %s, got %s", expectedTail, tail)
+            }
+            if follow != expectedFollow {
+                    t.Errorf("expected %t, got %t", expectedFollow, follow)
+            }
+            return nil
+    }
+    resp, err := http.Get(fw.testHTTPServer.URL+"/containerLogs/" + podName + "/" + expectedContainerName)
+    if err != nil {
+            t.Errorf("Got error GETing: %v", err)
+    }
+    defer resp.Body.Close()
+
+    body, err := ioutil.ReadAll(resp.Body)
+    if err != nil {
+            t.Errorf("Error reading container logs: %v", err)
+    }
+    result := string(body)
+    if result != string(body) {
+            t.Errorf("Expected: '%v', got: '%v'", output, result)
+    }
+}
+
+func TestContainerLogsWithTail(t *testing.T) {
+    fw := newServerTest()
+    output := "foo bar"
+    podName := "foo"
+    expectedPodName := podName + ".etcd"
+    expectedContainerName := "baz"
+    expectedTail := "5"
+    expectedFollow := false
+    fw.fakeKubelet.containerLogsFunc = func(podFullName, containerName, tail string, follow bool, writer io.Writer) error {
+            if podFullName != expectedPodName {
+                    t.Errorf("expected %s, got %s", expectedPodName, podFullName)
+            }
+			if containerName != expectedContainerName {
+				t.Errorf("expected %s, got %s", expectedContainerName, containerName)
+			}
+            if tail != expectedTail {
+                    t.Errorf("expected %s, got %s", expectedTail, tail)
+            }
+            if follow != expectedFollow {
+                    t.Errorf("expected %t, got %t", expectedFollow, follow)
+            }
+            return nil
+    }
+    resp, err := http.Get(fw.testHTTPServer.URL+"/containerLogs/" + podName + "/" + expectedContainerName + "?tail=5")
+    if err != nil {
+            t.Errorf("Got error GETing: %v", err)
+    }
+    defer resp.Body.Close()
+
+    body, err := ioutil.ReadAll(resp.Body)
+    if err != nil {
+            t.Errorf("Error reading container logs: %v", err)
+    }
+    result := string(body)
+    if result != string(body) {
+            t.Errorf("Expected: '%v', got: '%v'", output, result)
+    }
+}
+
+func TestContainerLogsWithFollow(t *testing.T) {
+    fw := newServerTest()
+    output := "foo bar"
+    podName := "foo"
+    expectedPodName := podName + ".etcd"
+    expectedContainerName := "baz"
+    expectedTail := ""
+    expectedFollow := true
+    fw.fakeKubelet.containerLogsFunc = func(podFullName, containerName, tail string, follow bool, writer io.Writer) error {
+            if podFullName != expectedPodName {
+                    t.Errorf("expected %s, got %s", expectedPodName, podFullName)
+            }
+			if containerName != expectedContainerName {
+				t.Errorf("expected %s, got %s", expectedContainerName, containerName)
+			}
+            if tail != expectedTail {
+                    t.Errorf("expected %s, got %s", expectedTail, tail)
+            }
+            if follow != expectedFollow {
+                    t.Errorf("expected %t, got %t", expectedFollow, follow)
+            }
+            return nil
+    }
+    resp, err := http.Get(fw.testHTTPServer.URL+"/containerLogs/" + podName + "/" + expectedContainerName + "?follow=1")
+    if err != nil {
+            t.Errorf("Got error GETing: %v", err)
+    }
+    defer resp.Body.Close()
+
+    body, err := ioutil.ReadAll(resp.Body)
+    if err != nil {
+            t.Errorf("Error reading container logs: %v", err)
+    }
+    result := string(body)
+    if result != string(body) {
+            t.Errorf("Expected: '%v', got: '%v'", output, result)
+    }
+}