Go wrapper for incomplete runtimes

docker/runtime/golang/Dockerfile.init

@@ -11,6 +11,9 @@ RUN ln -s /kubeless $GOPATH/src/kubeless
 # Install controller
 RUN mkdir -p $GOPATH/src/github.com/kubeless/kubeless/pkg/functions
 ADD pkg/functions/* $GOPATH/src/github.com/kubeless/kubeless/pkg/functions
+ADD pkg/function-proxy $GOPATH/src/github.com/kubeless/kubeless/pkg/function-proxy
+WORKDIR $GOPATH/src/github.com/kubeless/kubeless/pkg/function-proxy
+RUN dep ensure
 RUN mkdir -p $GOPATH/src/controller
 ADD docker/runtime/golang/Gopkg.toml $GOPATH/src/controller/
 WORKDIR $GOPATH/src/controller/

docker/runtime/golang/Gopkg.toml

@@ -1,5 +1,5 @@
 
-ignored = ["github.com/kubeless/kubeless/pkg/functions"]
+ignored = ["github.com/kubeless/kubeless/pkg/functions", "github.com/kubeless/kubeless/pkg/function-proxy/utils"]
 
 [[constraint]]
   name = "github.com/prometheus/client_golang"

docker/runtime/golang/kubeless.tpl.go

@@ -17,156 +17,67 @@ limitations under the License.
 package main
 
 import (
-	"fmt"
 	"golang.org/x/net/context"
 	"io/ioutil"
-	"log"
 	"net/http"
 	"os"
-	"strconv"
-	"time"
 
 	"kubeless"
 
+	proxyUtils "github.com/kubeless/kubeless/pkg/function-proxy/utils"
 	"github.com/kubeless/kubeless/pkg/functions"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 var (
-	funcHandler   = os.Getenv("FUNC_HANDLER")
-	timeout       = os.Getenv("FUNC_TIMEOUT")
-	funcPort      = os.Getenv("FUNC_PORT")
-	runtime       = os.Getenv("FUNC_RUNTIME")
-	memoryLimit   = os.Getenv("FUNC_MEMORY_LIMIT")
-	intTimeout    int
-	funcContext   functions.Context
-	funcHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Name: "function_duration_seconds",
-		Help: "Duration of user function in seconds",
-	}, []string{"method"})
-	funcCalls = prometheus.NewCounterVec(prometheus.CounterOpts{
-		Name: "function_calls_total",
-		Help: "Number of calls to user function",
-	}, []string{"method"})
-	funcErrors = prometheus.NewCounterVec(prometheus.CounterOpts{
-		Name: "function_failures_total",
-		Help: "Number of exceptions in user function",
-	}, []string{"method"})
+	funcContext functions.Context
 )
 
 func init() {
+	timeout := os.Getenv("FUNC_TIMEOUT")
 	if timeout == "" {
 		timeout = "180"
 	}
-	if funcPort == "" {
-		funcPort = "8080"
-	}
 	funcContext = functions.Context{
-		FunctionName: funcHandler,
+		FunctionName: os.Getenv("FUNC_HANDLER"),
 		Timeout:      timeout,
-		Runtime:      runtime,
-		MemoryLimit:  memoryLimit,
-	}
-	var err error
-	intTimeout, err = strconv.Atoi(timeout)
-	if err != nil {
-		panic(err)
+		Runtime:      os.Getenv("FUNC_RUNTIME"),
+		MemoryLimit:  os.Getenv("FUNC_MEMORY_LIMIT"),
 	}
-	prometheus.MustRegister(funcHistogram, funcCalls, funcErrors)
-}
-
-// Logging Functions, required to expose statusCode property
-type loggingResponseWriter struct {
-	http.ResponseWriter
-	statusCode int
-}
-
-func newLoggingResponseWriter(w http.ResponseWriter) *loggingResponseWriter {
-	return &loggingResponseWriter{w, http.StatusOK}
-}
-
-func (lrw *loggingResponseWriter) WriteHeader(code int) {
-	lrw.statusCode = code
-	lrw.ResponseWriter.WriteHeader(code)
 }
 
-func logReq(handler http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		lrw := newLoggingResponseWriter(w)
-		handler.ServeHTTP(lrw, r)
-		log.Printf("%s \"%s %s %s\" %d %s", r.RemoteAddr, r.Method, r.RequestURI, r.Proto, lrw.statusCode, r.UserAgent())
-		if lrw.statusCode == 408 {
-			go func() {
-				// Give time to return timeout response
-				time.Sleep(time.Second)
-				log.Fatal("Request timeout. Forcing exit")
-			}()
-		}
-	})
-}
-
-// Handling Functions
 func health(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte("OK"))
 }
 
-func handler(w http.ResponseWriter, r *http.Request) {
+func handle(ctx context.Context, w http.ResponseWriter, r *http.Request) ([]byte, error) {
 	data, err := ioutil.ReadAll(r.Body)
 	if err != nil {
-		panic(err)
+		return []byte{}, err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(intTimeout)*time.Second)
-	defer cancel()
 	event := functions.Event{
 		Data:           string(data),
 		EventID:        r.Header.Get("event-id"),
 		EventType:      r.Header.Get("event-type"),
 		EventTime:      r.Header.Get("event-time"),
 		EventNamespace: r.Header.Get("event-namespace"),
 		Extensions: functions.Extension{
-			Request: r,
+			Request:  r,
 			Response: w,
-			Context: ctx,
+			Context:  ctx,
 		},
 	}
-	funcChannel := make(chan struct {
-		res string
-		err error
-	}, 1)
-	go func() {
-		funcCalls.With(prometheus.Labels{"method": r.Method}).Inc()
-		start := time.Now()
-		res, err := kubeless.<<FUNCTION>>(event, funcContext)
-		funcHistogram.With(prometheus.Labels{"method": r.Method}).Observe(time.Since(start).Seconds())
-		pack := struct {
-			res string
-			err error
-		}{res, err}
-		funcChannel <- pack
-	}()
-	select {
-	case respPack := <-funcChannel:
-		if respPack.err != nil {
-			funcErrors.With(prometheus.Labels{"method": r.Method}).Inc()
-			w.WriteHeader(http.StatusInternalServerError)
-			w.Write([]byte(fmt.Sprintf("Error: %v", respPack.err)))
-		} else {
-			w.Write([]byte(respPack.res))
-		}
-	// Send Timeout response
-	case <-ctx.Done():
-		funcErrors.With(prometheus.Labels{"method": r.Method}).Inc()
-		w.WriteHeader(http.StatusRequestTimeout)
-		w.Write([]byte("Timeout exceeded"))
-	}
+	res, err := kubeless.<<FUNCTION>>(event, funcContext)
+	return []byte(res), err
+}
+
+func handler(w http.ResponseWriter, r *http.Request) {
+	proxyUtils.Handler(w, r, handle)
 }
 
 func main() {
 	http.HandleFunc("/", handler)
 	http.HandleFunc("/healthz", health)
 	http.Handle("/metrics", promhttp.Handler())
-	if err := http.ListenAndServe(fmt.Sprintf(":%s", funcPort), logReq(http.DefaultServeMux)); err != nil {
-		panic(err)
-	}
+	proxyUtils.ListenAndServe()
 }

docker/runtime/php/.gitignore

@@ -1 +1,2 @@
-vendor
+vendor
+proxy

docker/runtime/php/Dockerfile

@@ -4,7 +4,7 @@ FROM php:7.2-fpm-stretch
 ENV COMPOSER_HOME /composer
 ENV PATH="/root/.composer/vendor/bin:${PATH}"
 ENV COMPOSER_ALLOW_SUPERUSER=1
-ENV FUNC_PORT 8080
+ENV FUNC_PROCESS="/usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf"
 
 RUN apt-get update \
     && apt-get install -y \
@@ -51,4 +51,5 @@ RUN rm /etc/nginx/sites-available/default && \
     chmod -R a+w /run/nginx.pid /app /var/lib/nginx /var/log/nginx
 USER 1000
 
-CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]
+ADD proxy /
+CMD [ "/proxy" ]

docker/runtime/php/Makefile

@@ -1,6 +1,7 @@
 
 
 build:
+	env GOOS=linux GOARCH=amd64 go build $$GOPATH/src/github.com/kubeless/kubeless/pkg/function-proxy/proxy.go
 	docker build -t kubeless/php:7.2$$RUNTIME_TAG_MODIFIER .
 
 push:

docker/runtime/php/config/default.conf

@@ -1,5 +1,5 @@
 server {
-    listen 8080;
+    listen 8090;
     server_name _;
     index Controller.php;
     error_log /dev/stderr;

docker/runtime/ruby/.gitignore

@@ -0,0 +1 @@
+proxy

docker/runtime/ruby/Dockerfile

@@ -3,11 +3,12 @@ FROM bitnami/ruby:2.4
 LABEL maintainer "Bitnami <containers@bitnami.com>"
 
 ENV RACK_ENV="production"
-
+ENV FUNC_PROCESS="ruby /kubeless.rb"
 RUN gem install sinatra --no-rdoc --no-ri
 
 ADD kubeless.rb /
 
 USER 1000
 
-CMD ["ruby", "/kubeless.rb"]
+ADD proxy /
+CMD [ "/proxy" ]

docker/runtime/ruby/Makefile

@@ -1,5 +1,6 @@
 
 build2.4:
+	env GOOS=linux GOARCH=amd64 go build $$GOPATH/src/github.com/kubeless/kubeless/pkg/function-proxy/proxy.go
 	docker build -t kubeless/ruby:2.4$$RUNTIME_TAG_MODIFIER -f Dockerfile .
 
 push2.4:

docker/runtime/ruby/kubeless.rb

@@ -44,7 +44,7 @@
 end
 
 set :server, 'webrick'
-set :port, ENV['FUNC_PORT'] || 8080
+set :port, 8090
 
 def funcWrapper(mod, t)
     status = Timeout::timeout(t) {

docs/implementing-new-runtime.md

@@ -41,19 +41,26 @@ The HTTP server should satisfy the following requirements:
 
  - The file to load can be specified using an environment variable `MOD_NAME`.
  - The function to load can be specified using an environment variable `FUNC_HANDLER`.
- - The port used to expose the service can be modified using an environment variable `FUNC_PORT`.
  - The server should return `200 - OK` to requests at `/healthz`.
- - Functions should run `FUNC_TIMEOUT` as maximum. If, due to language limitations, it is not possible not stop the user function, at least a `408 - Timeout` response should be returned to the HTTP request.
  - Functions should receive two parameters: `event` and `context` and should return the value that will be used as HTTP response. See [the functions standard signature](/docs/runtimes#runtimes-interface) for more information. The information that will be available in `event` parameter will be received as HTTP headers.
  - Requests should be served in parallel.
- - Requests should be logged to stdout including date, HTTP method, requested path and status code of the response.
  - Exceptions in the function should be caught. The server should not exit due to a function error.
- - [Optional] The function should expose Prometheus statistics in the path `/metrics`. At least it should expose:
+
+See an example of an runtime image for [Python](https://github.com/kubeless/kubeless/blob/master/docker/runtime/python/Dockerfile.2.7).
+
+## 2.1 Additional features
+
+Apart from the requirements above, the runtime should satisfy:
+
+ - The port used to expose the service can be modified using an environment variable `FUNC_PORT`.
+ - Functions should run `FUNC_TIMEOUT` as maximum. If, due to language limitations, it is not possible not stop the user function, at least a `408 - Timeout` response should be returned to the HTTP request.
+ - Requests should be logged to stdout including date, HTTP method, requested path and status code of the response.
+ - The function should expose Prometheus statistics in the path `/metrics`. At least it should expose:
    - Calls per HTTP method
    - Errors per HTTP method
    - Histogram with the execution time per HTTP method
 
-See an example of an runtime image for [Python](https://github.com/kubeless/kubeless/blob/master/docker/runtime/python/Dockerfile.2.7).
+In any case, it is not necessary that the native runtime fulfill the above. Those features can be obtained adding a Go proxy that already implente those features and redirect the request to the new runtime. For adding it is only necessary to add the proxy binary to the image and run it as the `CMD`. See the [ruby example](https://github.com/kubeless/kubeless/blob/master/docker/runtime/ruby/).
 
 ## 3. Update the kubeless-config configmap
 

kubeless-non-rbac.jsonnet

@@ -126,7 +126,7 @@ local runtime_images ='[
       {
         "name": "ruby24",
         "version": "2.4",
-        "runtimeImage": "kubeless/ruby@sha256:0dce29c0eb2a246f7d825b6644eeae7957b26f2bfad2b7987f2134cc7b350f2f",
+        "runtimeImage": "kubeless/ruby@sha256:01665f1a32fe4fab4195af048627857aa7b100e392ae7f3e25a44bd296d6f105",
         "initImage": "bitnami/ruby:2.4"
       }
     ],
@@ -140,7 +140,7 @@ local runtime_images ='[
       {
         "name": "php72",
         "version": "7.2",
-        "runtimeImage": "kubeless/php@sha256:b605bb6b5ae3b1a2a93570939296618904259d7767a14002fa9733e66d59849b",
+        "runtimeImage": "kubeless/php@sha256:9b86066b2640bedcd88acb27f43dfaa2b338f0d74d9d91131ea781402f7ec8ec",
         "initImage": "composer:1.6"
       }
     ],
@@ -155,7 +155,7 @@ local runtime_images ='[
         "name": "go1.10",
         "version": "1.10",
         "runtimeImage": "kubeless/go@sha256:e2fd49f09b6ff8c9bac6f1592b3119ea74237c47e2955a003983e08524cb3ae5",
-        "initImage": "kubeless/go-init@sha256:d0812c4e8351bfd95d0574efd23613cff2664d6a57af4ed0a20ebc651382d476"
+        "initImage": "kubeless/go-init@sha256:983b3f06452321a2299588966817e724d1a9c24be76cf1b12c14843efcdff502"
       }
     ],
     "depName": "Gopkg.toml",

pkg/function-proxy/Gopkg.toml

@@ -0,0 +1,3 @@
+[[constraint]]
+  name = "github.com/prometheus/client_golang"
+  revision = "f504d69affe11ec1ccb2e5948127f86878c9fd57"