Improve Model Endpoint and Batch Job's Logging UX

.github/workflows/ci.yml

@@ -17,9 +17,9 @@ jobs:
     steps:
       - name: Checkout to the target branch
         uses: actions/checkout@v2
-      - uses: actions/setup-node@v1
+      - uses: actions/setup-node@v2
         with:
-          node-version: 12
+          node-version: 14
       - name: Install dependencies
         run: make init-dep-ui
       - name: Lint UI files
@@ -64,9 +64,9 @@ jobs:
     steps:
       - name: Checkout to the target branch
         uses: actions/checkout@v2
-      - uses: actions/setup-node@v1
+      - uses: actions/setup-node@v2
         with:
-          node-version: 12
+          node-version: 14
       - name: Install dependencies
         run: make init-dep-ui
       - name: Test UI files
@@ -86,9 +86,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - uses: actions/setup-node@v1
+      - uses: actions/setup-node@v2
         with:
-          node-version: 12
+          node-version: 14
       - name: Install dependencies
         run: make init-dep-ui
       - name: Build UI static files

Dockerfile

@@ -30,10 +30,10 @@ RUN go build -o bin/merlin_api ./cmd/api
 # ============================================================
 # Build stage 2: Build UI
 # ============================================================
-FROM node:14-alpine as node-builder
+FROM node:14 as node-builder
 WORKDIR /src/ui
 COPY ui .
-RUN yarn
+RUN yarn install --network-concurrency 1
 RUN yarn run build
 
 # ============================================================

Makefile

@@ -15,6 +15,7 @@ all: setup init-dep lint test clean build run
 .PHONY: setup
 setup:
 	@echo "> Setting up tools ..."
+	@test -x ${GOPATH}/bin/goimports || go get -u golang.org/x/tools/cmd/goimports
 	@test -x ${GOPATH}/bin/golint || go get -u golang.org/x/lint/golint
 	@test -x ${GOPATH}/bin/gotest || go get -u github.com/rakyll/gotest
 
@@ -24,7 +25,7 @@ init-dep: init-dep-ui init-dep-api
 .PHONY: init-dep-ui
 init-dep-ui:
 	@echo "> Initializing UI dependencies ..."
-	@cd ${UI_PATH} && yarn
+	@cd ${UI_PATH} && yarn install --network-concurrency 1
 
 .PHONY: init-dep-api
 init-dep-api:

api/api/log_api.go

@@ -15,7 +15,6 @@
 package api
 
 import (
-	"bufio"
 	"fmt"
 	"net/http"
 
@@ -34,6 +33,13 @@ type LogController struct {
 
 // ReadLog parses log requests and fetches logs.
 func (l *LogController) ReadLog(w http.ResponseWriter, r *http.Request) {
+	// Make sure that the writer supports flushing.
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		InternalServerError("Streaming unsupported!").WriteTo(w)
+		return
+	}
+
 	var query service.LogQuery
 	err := decoder.Decode(&query, r.URL.Query())
 	if err != nil {
@@ -42,38 +48,36 @@ func (l *LogController) ReadLog(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	res, err := l.LogService.ReadLog(&query)
-	if err != nil {
-		log.Errorf("Error while retrieving log %v", err)
-		InternalServerError(fmt.Sprintf("Error while retrieving log for container %s: %s", query.Name, err)).WriteTo(w)
-		return
-	}
+	logLineCh := make(chan string, 1000)
+	stopCh := make(chan struct{})
 
-	// send status code and content-type
-	w.Header().Set("Content-Type", "plain/text; charset=UTF-8")
-	w.WriteHeader(http.StatusOK)
+	// Listen to the closing of the http connection via the CloseNotifier
+	notify := r.Context().Done()
+	go func() {
+		<-notify
+		close(stopCh)
+	}()
 
-	// stream the response body
-	defer res.Close()
-	buff := bufio.NewReader(res)
-	for {
-		line, readErr := buff.ReadString('\n')
-		_, writeErr := w.Write([]byte(line))
-		if writeErr != nil {
-			// connection from caller is closed
-			return
-		}
+	// Set the headers related to event streaming.
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("Transfer-Encoding", "chunked")
+
+	go func() {
+		for {
+			logLine := <-logLineCh
+
+			fmt.Fprint(w, logLine)
 
-		// send the response over network
-		// although it's not guaranteed to reach client if it sits behind proxy
-		flusher, ok := w.(http.Flusher)
-		if ok {
+			// Send the response over network
+			// although it's not guaranteed to reach client if it sits behind proxy
 			flusher.Flush()
 		}
+	}()
 
-		if readErr != nil {
-			// unable to read log from container anymore most likely EOF
-			return
-		}
+	if err := l.LogService.StreamLogs(logLineCh, stopCh, &query); err != nil {
+		InternalServerError(err.Error()).WriteTo(w)
+		return
 	}
 }

api/api/router.go

@@ -202,9 +202,7 @@ func NewRouter(appCtx AppContext) *mux.Router {
 	}
 
 	rawRoutes := []RawRoutes{
-		{
-			http.MethodGet, "/logs", http.HandlerFunc(logController.ReadLog), "ReadLogs",
-		},
+		{http.MethodGet, "/logs", http.HandlerFunc(logController.ReadLog), "ReadLogs"},
 	}
 
 	var authzMiddleware *middleware.Authorizer

api/cluster/container.go

@@ -55,25 +55,25 @@ func (cf *containerFetcher) GetContainers(namespace string, labelSelector string
 	containers := make([]*models.Container, 0)
 	for _, pod := range podList.Items {
 		for _, c := range pod.Spec.Containers {
-			container := &models.Container{
-				Name:       c.Name,
-				PodName:    pod.Name,
-				Namespace:  pod.Namespace,
-				Cluster:    cf.metadata.ClusterName,
-				GcpProject: cf.metadata.GcpProject,
-			}
+			container := models.NewContainer(
+				c.Name,
+				pod.Name,
+				pod.Namespace,
+				cf.metadata.ClusterName,
+				cf.metadata.GcpProject,
+			)
 
 			containers = append(containers, container)
 		}
 
 		for _, ic := range pod.Spec.InitContainers {
-			container := &models.Container{
-				Name:       ic.Name,
-				PodName:    pod.Name,
-				Namespace:  pod.Namespace,
-				Cluster:    cf.metadata.ClusterName,
-				GcpProject: cf.metadata.GcpProject,
-			}
+			container := models.NewContainer(
+				ic.Name,
+				pod.Name,
+				pod.Namespace,
+				cf.metadata.ClusterName,
+				cf.metadata.GcpProject,
+			)
 
 			containers = append(containers, container)
 		}

api/cluster/controller.go

@@ -15,12 +15,14 @@
 package cluster
 
 import (
+	"io"
 	"time"
 
 	kfsv1alpha2 "github.com/kubeflow/kfserving/pkg/apis/serving/v1alpha2"
 	kfservice "github.com/kubeflow/kfserving/pkg/client/clientset/versioned/typed/serving/v1alpha2"
 	"github.com/kubeflow/kfserving/pkg/constants"
 	"github.com/pkg/errors"
+	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -35,6 +37,10 @@ import (
 type Controller interface {
 	Deploy(modelService *models.Service) (*models.Service, error)
 	Delete(modelService *models.Service) (*models.Service, error)
+
+	ListPods(namespace, labelSelector string) (*v1.PodList, error)
+	StreamPodLogs(namespace, podName string, opts *v1.PodLogOptions) (io.ReadCloser, error)
+
 	ContainerFetcher
 }
 
@@ -219,3 +225,11 @@ func (k *controller) waitInferenceServiceReady(service *kfsv1alpha2.InferenceSer
 		}
 	}
 }
+
+func (c *controller) ListPods(namespace, labelSelector string) (*v1.PodList, error) {
+	return c.clusterClient.Pods(namespace).List(metav1.ListOptions{LabelSelector: labelSelector})
+}
+
+func (c *controller) StreamPodLogs(namespace, podName string, opts *v1.PodLogOptions) (io.ReadCloser, error) {
+	return c.clusterClient.Pods(namespace).GetLogs(podName, opts).Stream()
+}

api/cluster/controller_test.go

@@ -698,3 +698,61 @@ func isIn(container v1.Container, containers []*models.Container, podName string
 	}
 	return false
 }
+
+func Test_controller_ListPods(t *testing.T) {
+	namespace := "test-namespace"
+
+	v1Client := fake.NewSimpleClientset()
+	v1Client.PrependReactor(listMethod, podResource, func(action ktesting.Action) (bool, runtime.Object, error) {
+		return true, &v1.PodList{
+			Items: []v1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test-model-1-predictor-default-a",
+						Labels: map[string]string{
+							"serving.knative.dev/service": "test-model-1-predictor-default",
+						},
+					},
+					Spec: v1.PodSpec{
+						InitContainers: []v1.Container{
+							{Name: "storage-initializer"},
+						},
+						Containers: []v1.Container{
+							{Name: "kfserving-container"},
+							{Name: "queue-proxy"},
+							{Name: "inferenceservice-logger"},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test-model-1-predictor-default-b",
+						Labels: map[string]string{
+							"serving.knative.dev/service": "test-model-1-predictor-default",
+						},
+					},
+					Spec: v1.PodSpec{
+						InitContainers: []v1.Container{
+							{Name: "storage-initializer"},
+						},
+						Containers: []v1.Container{
+							{Name: "kfserving-container"},
+							{Name: "queue-proxy"},
+							{Name: "inferenceservice-logger"},
+						},
+					},
+				},
+			}}, nil
+	})
+
+	ctl := &controller{
+		clusterClient: v1Client.CoreV1(),
+	}
+
+	podList, err := ctl.ListPods(namespace, "serving.knative.dev/service=test-model-1-predictor-default")
+
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(podList.Items))
+	assert.Equal(t, "test-model-1-predictor-default-a", podList.Items[0].ObjectMeta.Name)
+	assert.Equal(t, "test-model-1-predictor-default-b", podList.Items[1].ObjectMeta.Name)
+}