From b229d6e1b07b566a86e2ebb554ad85f00478034a Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Tue, 2 Jul 2019 15:43:55 +0800 Subject: [PATCH] Add experimental environment: tensorflow-serving (#1212) --- environments/tensorflow-serving/Dockerfile | 23 ++ environments/tensorflow-serving/README.md | 29 +++ environments/tensorflow-serving/server.go | 205 ++++++++++++++++++ examples/tensorflow-serving/README.md | 23 ++ .../half_plus_two/00000123/saved_model.pb | Bin 0 -> 9337 bytes .../variables/variables.data-00000-of-00001 | Bin 0 -> 12 bytes .../00000123/variables/variables.index | Bin 0 -> 151 bytes hack/release-build.sh | 27 +-- hack/release.sh | 27 +-- test/build_and_test.sh | 1 + test/test_utils.sh | 8 +- .../test_tensorflow_serving_env.sh | 61 ++++++ test/utils.sh | 34 ++- 13 files changed, 408 insertions(+), 30 deletions(-) create mode 100644 environments/tensorflow-serving/Dockerfile create mode 100644 environments/tensorflow-serving/README.md create mode 100644 environments/tensorflow-serving/server.go create mode 100644 examples/tensorflow-serving/README.md create mode 100644 examples/tensorflow-serving/half_plus_two/00000123/saved_model.pb create mode 100644 examples/tensorflow-serving/half_plus_two/00000123/variables/variables.data-00000-of-00001 create mode 100644 examples/tensorflow-serving/half_plus_two/00000123/variables/variables.index create mode 100755 test/tests/test_environments/test_tensorflow_serving_env.sh diff --git a/environments/tensorflow-serving/Dockerfile b/environments/tensorflow-serving/Dockerfile new file mode 100644 index 00000000..ade52016 --- /dev/null +++ b/environments/tensorflow-serving/Dockerfile @@ -0,0 +1,23 @@ +ARG GO_VERSION=1.9.2 + +FROM tensorflow/serving as serving +RUN apt update && apt install -y ca-certificates && rm -rf /var/lib/apt/lists/* + +FROM golang:${GO_VERSION} AS builder + +ENV GOPATH /usr +ENV APP ${GOPATH}/src/github.com/fission/fission/environments/tensorflow-serving + +WORKDIR ${APP} + +ADD server.go ${APP} + +RUN go get +RUN go build -a -o /server server.go + +FROM serving +WORKDIR / +COPY --from=builder /server / + +ENTRYPOINT ["/server"] +EXPOSE 8888 diff --git a/environments/tensorflow-serving/README.md b/environments/tensorflow-serving/README.md new file mode 100644 index 00000000..a1c4ea1c --- /dev/null +++ b/environments/tensorflow-serving/README.md @@ -0,0 +1,29 @@ +# Fission: Tensorflow Serving Environment + +This is the Tensorflow Serving environment for Fission. + +It's a Docker image containing a Go runtime, along with a tensorflow serving service. + +## How it works + +Tensorflow Serving is an serving service that supports both RESTful API and gRPC endpoints. In current implementation, +Go server launches `tensorflow_model_server` to load in model during specialization. As long as the Go server receives +requests from router it creates a reverse proxy that connects to RESTful API endpoint exposed by tensorflow_model_server +and get response from the upstream server for user. + +## Build this image + +``` +docker build -t USER/tensorflow-serving . && docker push USER/tensorflow-serving +``` + +## Using the image in fission + +You can add this customized image to fission with "fission env create": + +``` +fission env create --name tensorflow --image USER/tensorflow-serving --version 2 +``` + +After this, fission functions that have the env parameter set to the +same environment name as this command will use this environment. diff --git a/environments/tensorflow-serving/server.go b/environments/tensorflow-serving/server.go new file mode 100644 index 00000000..d935476f --- /dev/null +++ b/environments/tensorflow-serving/server.go @@ -0,0 +1,205 @@ +package main + +import ( + "encoding/json" + "fmt" + "io/ioutil" + "log" + "net" + "net/http" + "net/http/httputil" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/pkg/errors" + "go.uber.org/zap" +) + +const ( + PortgRPC = 8500 + PortRestAPI = 8501 +) + +var ( + specialized = false + + // for tensorflow serving to use + MODEL_NAME = "" +) + +type ( + FunctionLoadRequest struct { + // FilePath is an absolute filesystem path to the + // function. What exactly is stored here is + // env-specific. Optional. + FilePath string `json:"filepath"` + + // FunctionName has an environment-specific meaning; + // usually, it defines a function within a module + // containing multiple functions. Optional; default is + // environment-specific. + FunctionName string `json:"functionName"` + + // URL to expose this function at. Optional; defaults + // to "/". + URL string `json:"url"` + } +) + +func specializeHandler(logger *zap.Logger) func(http.ResponseWriter, *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + logger.Error("v1 interface is not implemented") + w.WriteHeader(http.StatusNotImplemented) + } +} + +func specializeHandlerV2(logger *zap.Logger) func(http.ResponseWriter, *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + if specialized { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("Not a generic container")) + return + } + + body, err := ioutil.ReadAll(r.Body) + if err != nil { + logger.Error("error reading request body", zap.Error(err)) + w.WriteHeader(http.StatusInternalServerError) + return + } + var loadreq FunctionLoadRequest + err = json.Unmarshal(body, &loadreq) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + + // To ensure we load model from the expected path + basePath := fmt.Sprintf("%v/%v", loadreq.FilePath, loadreq.FunctionName) + basePath, err = filepath.Abs(basePath) + if err != nil { + msg := "error getting absolute path of model" + logger.Error(msg, zap.Error(err)) + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } else if !strings.HasPrefix(basePath, loadreq.FilePath) { + msg := "incorrect model base path" + logger.Error(msg, zap.String("model_base_path", basePath)) + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(msg)) + return + } + + _, err = os.Stat(basePath) + if err != nil { + msg := "error checking model status" + logger.Error(msg, zap.Error(err)) + w.WriteHeader(http.StatusInternalServerError) + return + } + + // get directory name that holds model + MODEL_NAME = filepath.Base(basePath) + + argModelBasePath := fmt.Sprintf("--model_base_path=%v", basePath) + argModelName := fmt.Sprintf("--model_name=%v", MODEL_NAME) + argPortgRPC := fmt.Sprintf("--port=%v", PortgRPC) + argPortREST := fmt.Sprintf("--rest_api_port=%v", PortRestAPI) + + logger.Info(fmt.Sprintf("specializing: %v %v", loadreq.FunctionName, loadreq.FilePath)) + + // Future: could be improved by keeping subprocess open while environment is specialized + cmd := exec.Command("tensorflow_model_server", + argPortgRPC, argPortREST, argModelName, argModelBasePath) + + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + + err = cmd.Start() + if err != nil { + msg := "error starting tensorflow serving" + logger.Error(msg, zap.Error(err)) + err = errors.Wrap(err, msg) + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + + go func() { + err = cmd.Wait() + if err != nil { + logger.Fatal("error running tensorflow serving", zap.Error(err)) + } + }() + + t := time.Now() + retryInterval := 50 * time.Millisecond + + // tensorflow serving takes some time to load model + // into memory, keep retrying until it starts REST api server. + for { + if time.Since(t) > 30*time.Second { + w.WriteHeader(http.StatusGatewayTimeout) + return + } + conn, err := net.Dial("tcp", "localhost:8501") + if err == nil { + conn.Close() + break + } else { + logger.Info(fmt.Sprintf("waiting for tensorflow serving to be ready: %v", err.Error())) + time.Sleep(retryInterval) + retryInterval = retryInterval * 2 + } + } + + specialized = true + + logger.Info("done") + } +} + +func readinessProbeHandler(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) +} + +func main() { + logger, err := zap.NewProduction() + if err != nil { + log.Fatalf("can't initialize zap logger: %v", err) + } + defer logger.Sync() + + http.HandleFunc("/healthz", readinessProbeHandler) + http.HandleFunc("/specialize", specializeHandler(logger.Named("specialize_handler"))) + http.HandleFunc("/v2/specialize", specializeHandlerV2(logger.Named("specialize_v2_handler"))) + + // Generic route -- all http requests go to the user function. + http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if !specialized { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("Generic container: no requests supported")) + return + } + + // TODO: replace it with gRPC (https://gist.github.com/mauri870/1f953a183ee6c186e70a0a72e78b088c) + // set up proxy server director + director := func(req *http.Request) { + req.URL.Scheme = "http" + req.URL.Host = "localhost:8501" + req.URL.Path = fmt.Sprintf("/v1/models/%v:predict", MODEL_NAME) + } + + proxy := &httputil.ReverseProxy{ + Director: director, + } + proxy.ServeHTTP(w, r) + }) + + logger.Info("listening on 8888 ...") + http.ListenAndServe(":8888", nil) +} diff --git a/examples/tensorflow-serving/README.md b/examples/tensorflow-serving/README.md new file mode 100644 index 00000000..591829f8 --- /dev/null +++ b/examples/tensorflow-serving/README.md @@ -0,0 +1,23 @@ +# Tensorflow Serving Environment Example + +## Create Environment + +```bash +$ fission env create --name tensorflow --image fission/tensorflow-serving --version 2 +``` + +## Create Package + +```bash +$ zip -r half_plus_two.zip ./half_plus_two +$ fission pkg create --env tensorflow --deploy half_plus_two.zip +``` + +## Create Function + +Here, the `--entrypoint` represents the name of top directory contains trained model. + +```bash +$ fission fn create --name t1 --pkg --env tensorflow --entrypoint "half_plus_two" +$ fission fn test --name t1 --body '{"instances": [1.0, 2.0, 0.0]}' --method POST +``` diff --git a/examples/tensorflow-serving/half_plus_two/00000123/saved_model.pb b/examples/tensorflow-serving/half_plus_two/00000123/saved_model.pb new file mode 100644 index 0000000000000000000000000000000000000000..c79ebe01c240863890d2048ebee5da046c93a90a GIT binary patch literal 9337 zcmd5?-H+Q?70=Cg&ZL>loqnWu+U*k2cJN9~CPSAtD$q^EYC-ByXeq3a#&T>=YfUE3 z#-4Uk9@qySKtgB*LLeaoJRFNP?*|B_`#=yP=L}aV{~eBI(Xxr`jb;%kKVUy5zRJ(1}=3h!?$(sr7^K- zap<}umft1qE3k3u+4{&ekDc+0`HSBZ0{5Ww$Q^q=+k~>D-!55P%Jyc!oR)oi?77qS z1(0u~7H|iu-?r?r@Axw=s53^{xS$&mA=G*ri-nI$F~W zTfz^Y;pb&29=P9~z+)&rG0bDyfU&9=oJVV!ER3ipH7aiFYjD z%E3SLP=8`fJ^Nc{#>r%4BVF+9se>+iV_T?h4N7Cc{pBO`%l|A$F9SL;H6ZtR}XHv|2@W{ z*CL=2pI(N#8rMKgL26>XMZn zhQvi^3O>3qwsdbK99dyoP?77<-y&zdb_;^5q}95Fr^X1%PmNb^$&Ied?IYNH`qG$M zw)LSivbiR-4%MOIiEA11sMOSI6?is`PfvK^np$3~z;NMF@KLJ*2LMm~DVDD9V9lTx zpe$+)&kRgRG@PtkSR0c0cm-;njfrHSiD}sr6j+|-zyo-0VD1e^XbX%CQE_E$!>ZeE zBtt=|dst0traSfxXKW|Pwy$E|q5`%h+!p7E#&4G}-Cx;SY46VcJKbL8bbLH^Uytb@ z_rZisqvMPnAFF{b@`=it+`iDU>{saQ{-Y;9=y$b^umI<~M7sgK7cBrIjxw><-Q3%5QE0lR%vg>wL&9tW`Zu|L|5XxmDHbG`nLqCMkm48492uU6X;YiEr!$_FdWur5-4U0y5 z7>cz-ibedgh@(G;EBSMws~)aoZ0^tD|Hhxgls|_|Rd|UdRhW|H&*7T>9H#s^Tp~r< zpTo>xL4TSn`E#W#S>*JG54k_hW&T`=b~*#KrFkv#(@go(WU9idB^M>2xrW2Ej-?E zQPQ`g~3L(AO-9rWcf9Li;K9@5SOt${6p|@Ydtpu0EjYQ)7u%LG{I42@G zD?@1Lgxa7|tIJ)|Kcmg=a_&12j1AIqK`1ft-k{RV~874_fkns8G{s>Aas=i0hkvLRPTUB)U81l286j`!h!A__Oy*}v? zU|Pi-x)V>RPW-07r*d0W{`oHZu{06t5_c*}@I{c$&-PSftzeW@Qnk)OnCtim?3OH5{B>~oy7xOP+54MzVA~}PvRIg)m5ayo*9GZvoM}jk>>Jxnrz6AC8 zvWU+{Y$ZIw4j6VL67VQ~_Nin6@+4T``&8+~#hxJij6h3}oOcTlw_{t$)e)WEL|NhYG+QYkh_R-hwcaM&Y&cvAdD~SiE zu&GEvWBm&j)0$<9%k$`f@;Z+Ie%{64_%LJlhVKPz=Z%z5=h)_7$od@jgaCo@&A&i% z9DN-FlbR&^I2`g#t@Ahhb(rPgmk9M1aieM*iK*sgOQZ}I&yAwwY?eI`aRts19kWin zA$Pv!l_mJGpj``1y*=2%807G}Fd)W&3R~q)u~NQR`CnfHD65OyG(Th&SA( z?ueH=dyH_`oW=IB#p2ZuW2&f5za2G@uEP$!DjG;~c{0+oq9feqD)4s#?};zdvyIMz z!3Gsmg)ECok<-em;%7UNXb_AZIARPi0!|QQH6c97q(#%>(gougwvKM`%BD(uDLu8p z#t;00;EaJsTr^3V_u~?EQd%Z0G;6JDvy8GC*$Cdiju0j4$*+?0OH+CE%=-!4PcPR}?Dl#HG<{2>fwDH$S;sNX$g&7^(P_&P%Kt2eyYYLKeZ_2$x< zDd|!WeUa_H$adckfeHgnrc{V9%$ zIC#g|jvP99^f!clMsR{&Fpy+4Az^sMhZ6m6Z%dg7^}F}P`XOH*okZ{|T(3+V!u2L! zA0Ag2UWMzGjy7N4{tJQH|GSL0VVnEI(goz#PtdlyJHe4D7LWLeLwki?dp;s^j<7GP z`>sBtCcG2Q4(Ualu|`ecez#d+Sb6A|q3oG1I+HRe%5dB9kznv>_A{0(7vVK*PVE<{g%@bvqt>~q z--3!9Cwi1~&JvKZ1G>iTBL~N literal 0 HcmV?d00001 diff --git a/examples/tensorflow-serving/half_plus_two/00000123/variables/variables.data-00000-of-00001 b/examples/tensorflow-serving/half_plus_two/00000123/variables/variables.data-00000-of-00001 new file mode 100644 index 0000000000000000000000000000000000000000..15b75d6ef6bffc336d138d923badb3928b8c4c13 GIT binary patch literal 12 RcmZQzV6bOkU~m8;2LJ>^0RR91 literal 0 HcmV?d00001 diff --git a/examples/tensorflow-serving/half_plus_two/00000123/variables/variables.index b/examples/tensorflow-serving/half_plus_two/00000123/variables/variables.index new file mode 100644 index 0000000000000000000000000000000000000000..7ec9fb4fe2dd21d0a6c324aecd7658fc37cf2326 GIT binary patch literal 151 zcmZQzVB=tvV&Y(AVB}8ZU=(7|U@>L0P?u+5 /dev/null); then + log "test_post_route: resp = '$resp' expect = '$expect'" + log "test_post_route: expected string not found. Retrying ..." + sleep 1 + continue + fi + break + done + set -e +} +export -f test_post_route + wait_for_builder() { env=$1 JSONPATH='{range .items[*]}{@.metadata.name}:{range @.status.conditions[*]}{@.type}={@.status};{end}{end}' @@ -108,7 +137,9 @@ export -f waitBuild export FISSION_NAMESPACE=${FISSION_NAMESPACE:-fission} export FUNCTION_NAMESPACE=${FUNCTION_NAMESPACE:-fission-function} -export FISSION_ROUTER=$(kubectl -n $FISSION_NAMESPACE get svc router -o jsonpath='{...ip}') +router=$(kubectl -n $FISSION_NAMESPACE get svc router -o jsonpath='{...ip}') + +export FISSION_ROUTER=${FISSION_ROUTER:-$router} export FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(kubectl -n $FISSION_NAMESPACE get svc nats-streaming -o jsonpath='{...ip}:{.spec.ports[0].port}')" ## Parameters used by some specific test cases @@ -119,4 +150,5 @@ export GO_BUILDER_IMAGE=${GO_BUILDER_IMAGE:-fission/go-builder} export JVM_RUNTIME_IMAGE=${JVM_RUNTIME_IMAGE:-fission/jvm-env} export JVM_BUILDER_IMAGE=${JVM_BUILDER_IMAGE:-fission/jvm-builder} export NODE_RUNTIME_IMAGE=${NODE_RUNTIME_IMAGE:-fission/node-env} +export TS_RUNTIME_IMAGE=${TS_RUNTIME_IMAGE:-fission/tensorflow-serving-env}