Payload Logger (kserve#503)

* Logger files, and test

* Remove sample and add modelId

* Updates from review

* Update from review comments

* Change println to log

* Remove some printlns

* Change test libsonnet to build logger not executor

Makefile

@@ -5,7 +5,7 @@ HAS_LINT := $(shell command -v golint;)
 IMG ?= kfserving-controller:latest
 EXECUTOR_IMG ?= kfserving-executor:latest
 
-all: test manager executor
+all: test manager logger
 
 # Run tests
 test: generate fmt vet lint manifests
@@ -16,8 +16,8 @@ manager: generate fmt vet lint
 	go build -o bin/manager ./cmd/manager
 
 # Build manager binary
-executor: fmt vet
-	go build -o bin/executor ./cmd/executor
+logger: fmt vet
+	go build -o bin/logger ./cmd/logger
 
 # Run against the configured Kubernetes cluster in ~/.kube/config
 run: generate fmt vet lint

cmd/logger/main.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"github.com/kubeflow/kfserving/pkg/apis/serving/v1alpha2"
+	"net/http"
+	"net/url"
+	"os"
+	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
+
+	"github.com/kubeflow/kfserving/pkg/logger"
+	"github.com/pkg/errors"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+)
+
+var (
+	logUrl        = flag.String("log-url", "", "The URL to send request/response logs to")
+	port          = flag.String("port", "8080", "Logger port")
+	componentHost = flag.String("component-host", "0.0.0.0", "Component host")
+	componentPort = flag.String("component-port", "8081", "Component port")
+	workers       = flag.Int("workers", 5, "Number of workers")
+	sourceUri     = flag.String("source-uri", "", "The source URI to use when publishing cloudevents")
+	logMode       = flag.String("log-mode", string(v1alpha2.LogAll), "Whether to log 'request', 'response' or 'all'")
+	modelId       = flag.String("model-id", "", "The model ID to add as header to log events")
+)
+
+func main() {
+	flag.Parse()
+
+	logf.SetLogger(logf.ZapLogger(false))
+	log := logf.Log.WithName("entrypoint")
+
+	if *logUrl == "" {
+		log.Info("log-url argument must not be empty.")
+		os.Exit(-1)
+	}
+
+	logUrlParsed, err := url.Parse(*logUrl)
+	if err != nil {
+		log.Info("Malformed log-url", "URL", *logUrl)
+		os.Exit(-1)
+	}
+	loggingMode := v1alpha2.LoggerMode(*logMode)
+	switch loggingMode {
+	case v1alpha2.LogAll, v1alpha2.LogRequest, v1alpha2.LogResponse:
+	default:
+		log.Info("Malformed log-mode", "mode", *logMode)
+		os.Exit(-1)
+	}
+
+	if *sourceUri == "" {
+		*sourceUri = fmt.Sprintf("http://localhost:%s/", *port)
+	}
+	sourceUriParsed, err := url.Parse(*sourceUri)
+	if err != nil {
+		log.Info("Malformed source_uri", "URL", *sourceUri)
+		os.Exit(-1)
+	}
+
+	stopCh := signals.SetupSignalHandler()
+
+	var eh http.Handler = logger.New(log, *componentHost, *componentPort, logUrlParsed, sourceUriParsed, loggingMode, *modelId)
+
+	h1s := &http.Server{
+		Addr:    ":" + *port,
+		Handler: h2c.NewHandler(eh, &http2.Server{}),
+	}
+
+	log.Info("Starting the log dispatcher")
+	logger.StartDispatcher(*workers, log)
+
+	log.Info("Starting", "port", *port)
+
+	errCh := make(chan error, 1)
+	go func(name string, s *http.Server) {
+		// Don't forward ErrServerClosed as that indicates we're already shutting down.
+		if err := s.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			errCh <- errors.Wrapf(err, "%s server failed", name)
+		}
+	}("default", h1s)
+
+	// Exit as soon as we see a shutdown signal or the server failed.
+	select {
+	case <-stopCh:
+	case err := <-errCh:
+		log.Error(err, "Failed to run HTTP server")
+	}
+
+	err = h1s.Shutdown(context.Background())
+	if err != nil {
+		log.Error(err, "Failed to shutdown HTTP server")
+	}
+
+}

docs/samples/logger/local/README.md

@@ -0,0 +1,109 @@
+# Local Test of Kfserving Logger
+
+In one terminal start an echo http server on port 8000
+
+```
+docker run -it -p 8000:80 --rm -t mendhak/http-https-echo
+```
+
+Start an SKLearn Iris model on port 8081. You will need to have pip installed the sklearnserver. See `/python/sklearnserver`.
+
+```
+python -m sklearnserver --model_dir gs://kfserving-samples/models/sklearn/iris --model_name sklearn-iris --http_port 8081
+```
+
+Start the Kfserving logger from Kfserving root folder:
+
+```
+
+bin/logger --log-url http://0.0.0.0:8000 --component-port 8081 --log-mode all
+```
+
+Send a request:
+
+```
+curl -v -d @./input.json http://0.0.0.0:8080/v1/models/sklearn-iris:predict
+```
+
+You should see output like:
+
+```
+"
+*   Trying 0.0.0.0...
+* Connected to 0.0.0.0 (127.0.0.1) port 8080 (#0)
+> POST /v1/models/sklearn-iris:predict HTTP/1.1
+> Host: 0.0.0.0:8080
+> User-Agent: curl/7.47.0
+> Accept: */*
+> Content-Type: application/json
+> Content-Length: 76
+> 
+* upload completely sent off: 76 out of 76 bytes
+< HTTP/1.1 200 OK
+< Content-Type: application/json; charset=UTF-8
+< Date: Thu, 31 Oct 2019 15:20:50 GMT
+< Content-Length: 23
+< 
+* Connection #0 to host 0.0.0.0 left intact
+{"predictions": [1, 1]}
+```
+
+This shows the prediction has worked. In the output of the http-https-echo server you should see the request and response payloads echoed.
+
+
+```
+{ path: '/',
+  headers: 
+   { host: '0.0.0.0:8000',
+     'user-agent': 'Go-http-client/1.1',
+     'content-length': '76',
+     'ce-cloudeventsversion': '0.1',
+     'ce-eventid': '29232038-6c2a-44b3-a542-95c499732ec0',
+     'ce-eventtime': '2019-10-31T15:20:50.435513493Z',
+     'ce-eventtype': 'org.kubeflow.serving.inference.request',
+     'ce-source': 'http://localhost:8080/',
+     'content-type': 'application/json',
+     'kf-model-uri': '',
+     'accept-encoding': 'gzip' },
+  method: 'POST',
+  body: '{  "instances": [    [6.8,  2.8,  4.8,  1.4],    [6.0,  3.4,  4.5,  1.6]  ]}',
+  cookies: undefined,
+  fresh: false,
+  hostname: '0.0.0.0',
+  ip: '::ffff:172.17.0.1',
+  ips: [],
+  protocol: 'http',
+  query: {},
+  subdomains: [],
+  xhr: false,
+  os: { hostname: 'a167987d8875' } }
+::ffff:172.17.0.1 - - [31/Oct/2019:15:20:50 +0000] "POST / HTTP/1.1" 200 796 "-" "Go-http-client/1.1"
+-----------------
+{ path: '/',
+  headers: 
+   { host: '0.0.0.0:8000',
+     'user-agent': 'Go-http-client/1.1',
+     'content-length': '23',
+     'ce-cloudeventsversion': '0.1',
+     'ce-eventid': '29232038-6c2a-44b3-a542-95c499732ec0',
+     'ce-eventtime': '2019-10-31T15:20:50.438892641Z',
+     'ce-eventtype': 'org.kubeflow.serving.inference.response',
+     'ce-source': 'http://localhost:8080/',
+     'content-type': 'application/json; charset=UTF-8',
+     'kf-model-uri': '',
+     'accept-encoding': 'gzip' },
+  method: 'POST',
+  body: '{"predictions": [1, 1]}',
+  cookies: undefined,
+  fresh: false,
+  hostname: '0.0.0.0',
+  ip: '::ffff:172.17.0.1',
+  ips: [],
+  protocol: 'http',
+  query: {},
+  subdomains: [],
+  xhr: false,
+  os: { hostname: 'a167987d8875' } }
+::ffff:172.17.0.1 - - [31/Oct/2019:15:20:50 +0000] "POST / HTTP/1.1" 200 759 "-" "Go-http-client/1.1"
+```
+

docs/samples/logger/local/input.json

@@ -0,0 +1,6 @@
+{
+  "instances": [
+    [6.8,  2.8,  4.8,  1.4],
+    [6.0,  3.4,  4.5,  1.6]
+  ]
+}

logger.Dockerfile

@@ -0,0 +1,16 @@
+# Build the inference-logger binary
+FROM golang:1.10.3 as builder
+
+# Copy in the go src
+WORKDIR /go/src/github.com/kubeflow/kfserving
+COPY pkg/    pkg/
+COPY cmd/    cmd/
+COPY vendor/ vendor/
+# Build
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o logger ./cmd/logger
+
+# Copy the inference-logger into a thin image
+FROM alpine:latest
+WORKDIR /
+COPY --from=builder /go/src/github.com/kubeflow/kfserving/logger .
+ENTRYPOINT ["/logger"]