From a9dd4bf81a7a852261bad74e0ae2aa3e44b14b3b Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Sun, 14 Jul 2019 21:09:35 +0800 Subject: [PATCH] Fix fetcher of newdeploy pod error out when starting up (#1210) * Connect to k8s API server to ensure network connectivity * Fix readiness probe path --- cmd/fetcher/app/server.go | 27 ++++++++- cmd/fission-bundle/main.go | 16 +++++ pkg/error/network/error.go | 107 ++++++++++++++++++++++++++++++++++ pkg/fetcher/config/config.go | 2 +- pkg/fetcher/fetcher.go | 87 +++++++++++++++++++-------- pkg/router/functionHandler.go | 5 +- pkg/utils/utils.go | 16 ----- 7 files changed, 214 insertions(+), 46 deletions(-) create mode 100644 pkg/error/network/error.go diff --git a/cmd/fetcher/app/server.go b/cmd/fetcher/app/server.go index 24f669b2..566c8fbb 100644 --- a/cmd/fetcher/app/server.go +++ b/cmd/fetcher/app/server.go @@ -1,3 +1,19 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + package app import ( @@ -100,17 +116,24 @@ func Run(logger *zap.Logger) { mux.HandleFunc("/specialize", f.SpecializeHandler) mux.HandleFunc("/upload", f.UploadHandler) mux.HandleFunc("/version", f.VersionHandler) - mux.HandleFunc("/readniess-healthz", func(w http.ResponseWriter, r *http.Request) { + + readinessHandler := func(w http.ResponseWriter, r *http.Request) { if !*specializeOnStart || readyToServe { w.WriteHeader(http.StatusOK) } else { w.WriteHeader(http.StatusServiceUnavailable) } - }) + } + + mux.HandleFunc("/readiness-healthz", readinessHandler) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) + // For backward compatibility + // TODO: remove this path in future + mux.HandleFunc("/readniess-healthz", readinessHandler) + logger.Info("fetcher ready to receive requests") http.ListenAndServe(":8000", &ochttp.Handler{ Handler: mux, diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index eeaa2f38..b6722dab 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -1,3 +1,19 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + package main import ( diff --git a/pkg/error/network/error.go b/pkg/error/network/error.go new file mode 100644 index 00000000..718ece61 --- /dev/null +++ b/pkg/error/network/error.go @@ -0,0 +1,107 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package network + +import ( + "net" + "net/url" + "os" + "strings" + "syscall" +) + +type ( + Error struct { + err error + } +) + +// Adapter returns an Error if the pass-in error is a network error; +// otherwise, nil will be returned. +func Adapter(err error) *Error { + if err == nil { + return nil + } + + netErr, ok := err.(net.Error) + if !ok { + return nil + } + + return &Error{err: netErr} +} + +func (e Error) Error() string { + return e.err.Error() +} + +// IsDialError returns true if its a network dial error +func (e Error) IsDialError() bool { + netOpErr, ok := e.err.(*net.OpError) + if !ok { + return false + } + + if netOpErr.Op == "dial" { + return true + } + + return false +} + +// IsConnRefusedError returns true if an error is a "connection refused" error +func (e Error) IsConnRefusedError() bool { + urlErr, ok := e.err.(*url.Error) + if !ok { + return false + } + + if strings.Contains(urlErr.Error(), "connection refused") { + return true + } + + netOpErr, ok := e.err.(*net.OpError) + if !ok { + return false + } + + switch t := netOpErr.Err.(type) { + case *os.SyscallError: + if errno, ok := t.Err.(syscall.Errno); ok { + switch errno { + case syscall.ECONNREFUSED: + return true + } + } + } + + return false +} + +// IsUnsupportedProtoScheme returns true if an error is a "unsupported protocol scheme" error +func (e Error) IsUnsupportedProtoScheme() bool { + urlErr, ok := e.err.(*url.Error) + if !ok { + return false + } + + if strings.Contains(urlErr.Error(), "unsupported protocol scheme") { + return true + } + + return false +} diff --git a/pkg/fetcher/config/config.go b/pkg/fetcher/config/config.go index cbb073c6..15533cbc 100644 --- a/pkg/fetcher/config/config.go +++ b/pkg/fetcher/config/config.go @@ -232,7 +232,7 @@ func (cfg *Config) addFetcherToPodSpecWithCommand(podSpec *apiv1.PodSpec, mainCo FailureThreshold: 30, Handler: apiv1.Handler{ HTTPGet: &apiv1.HTTPGetAction{ - Path: "/readniess-healthz", + Path: "/readiness-healthz", Port: intstr.IntOrString{ Type: intstr.Int, IntVal: 8000, diff --git a/pkg/fetcher/fetcher.go b/pkg/fetcher/fetcher.go index d2755c99..cb5403fe 100644 --- a/pkg/fetcher/fetcher.go +++ b/pkg/fetcher/fetcher.go @@ -1,3 +1,19 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + package fetcher import ( @@ -9,9 +25,7 @@ import ( "fmt" "io" "io/ioutil" - "net" "net/http" - "net/url" "os" "path/filepath" "time" @@ -29,6 +43,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/error/network" "github.com/fission/fission/pkg/info" storageSvcClient "github.com/fission/fission/pkg/storagesvc/client" "github.com/fission/fission/pkg/types" @@ -188,7 +203,14 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { return } - code, err := fetcher.Fetch(r.Context(), req) + pkg, err := fetcher.getPkgInformation(req) + if err != nil { + fetcher.logger.Error("error getting package information", zap.Error(err)) + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + code, err := fetcher.Fetch(r.Context(), pkg, req) if err != nil { fetcher.logger.Error("error fetching", zap.Error(err)) http.Error(w, err.Error(), code) @@ -242,7 +264,7 @@ func (fetcher *Fetcher) SpecializeHandler(w http.ResponseWriter, r *http.Request // Fetch takes FetchRequest and makes the fetch call // It returns the HTTP code and error if any -func (fetcher *Fetcher) Fetch(ctx context.Context, req types.FunctionFetchRequest) (int, error) { +func (fetcher *Fetcher) Fetch(ctx context.Context, pkg *fv1.Package, req types.FunctionFetchRequest) (int, error) { // check that the requested filename is not an empty string and error out if so if len(req.Filename) == 0 { e := "fetch request received for an empty file name" @@ -270,16 +292,6 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, req types.FunctionFetchReques return http.StatusBadRequest, errors.Wrapf(err, "%s: %s", e, req.Url) } } else { - // get pkg - pkg, err := fetcher.fissionClient.Packages(req.Package.Namespace).Get(req.Package.Name) - if err != nil { - e := "failed to get package" - fetcher.logger.Error(e, - zap.String("package_name", req.Package.Name), - zap.String("package_namespace", req.Package.Namespace)) - return http.StatusInternalServerError, errors.Wrap(err, e) - } - var archive *fv1.Archive if req.FetchType == types.FETCH_SOURCE { archive = &pkg.Spec.Source @@ -301,7 +313,7 @@ func (fetcher *Fetcher) Fetch(ctx context.Context, req types.FunctionFetchReques // get package data as literal or by url if len(archive.Literal) > 0 { // write pkg.Literal into tmpPath - err = ioutil.WriteFile(tmpPath, archive.Literal, 0600) + err := ioutil.WriteFile(tmpPath, archive.Literal, 0600) if err != nil { e := "failed to write file" fetcher.logger.Error(e, zap.Error(err), zap.String("location", tmpPath)) @@ -586,6 +598,28 @@ func (fetcher *Fetcher) unarchive(src string, dst string) error { return nil } +// getPkgInformation gets package information from k8s api server. +func (fetcher *Fetcher) getPkgInformation(req types.FunctionFetchRequest) (pkg *fv1.Package, err error) { + maxRetries := 5 + for i := 0; i < maxRetries; i++ { + pkg, err = fetcher.fissionClient.Packages(req.Package.Namespace).Get(req.Package.Name) + if err == nil { + return pkg, nil + } + if i < maxRetries-1 { + // All outbound requests are blocked if istio is enabled at the first seconds. + // So if an error is a "connection refused" or "dial" error, wait for a while + // before retrying so that envoy proxy will start to serve requests. + // For details, see https://github.com/istio/istio/issues/12187 + netErr := network.Adapter(err) + if netErr != nil && (netErr.IsDialError() || netErr.IsConnRefusedError()) { + time.Sleep(500 * time.Duration(i+1) * time.Millisecond) + } + } + } + return nil, err +} + func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq types.FunctionFetchRequest, loadReq types.FunctionLoadRequest) error { startTime := time.Now() defer func() { @@ -593,7 +627,12 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq types.Functi fetcher.logger.Info("specialize request done", zap.Duration("elapsed_time", elapsed)) }() - _, err := fetcher.Fetch(ctx, fetchReq) + pkg, err := fetcher.getPkgInformation(fetchReq) + if err != nil { + return errors.Wrap(err, "error getting package information") + } + + _, err = fetcher.Fetch(ctx, pkg, fetchReq) if err != nil { return errors.Wrap(err, "error fetching deploy package") } @@ -638,19 +677,17 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq types.Functi return nil } + netErr := network.Adapter(err) // Only retry for the specific case of a connection error. - if urlErr, ok := err.(*url.Error); ok { - if netErr, ok := urlErr.Err.(*net.OpError); ok { - if netErr.Op == "dial" { - if i < maxRetries-1 { - time.Sleep(500 * time.Duration(2*i) * time.Millisecond) - fetcher.logger.Error("error connecting to function environment pod for specialization request, retrying", zap.Error(netErr)) - continue - } - } + if netErr != nil && (netErr.IsConnRefusedError() || netErr.IsDialError()) { + if i < maxRetries-1 { + time.Sleep(500 * time.Duration(2*i) * time.Millisecond) + fetcher.logger.Error("error connecting to function environment pod for specialization request, retrying", zap.Error(netErr)) + continue } } + // for 4xx, 5xx if err == nil { err = ferror.MakeErrorFromHTTP(resp) } diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 14826a7f..a236dc4d 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -31,7 +31,6 @@ import ( "time" "github.com/fission/fission/pkg/types" - "github.com/fission/fission/pkg/utils" "github.com/gorilla/mux" "github.com/pkg/errors" "github.com/satori/go.uuid" @@ -42,6 +41,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/error/network" executorClient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/redis" "github.com/fission/fission/pkg/throttler" @@ -310,7 +310,8 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt } // if transport.RoundTrip returns a non-network dial error, then relay it back to user - if !utils.IsNetworkDialError(err) { + netErr := network.Adapter(err) + if netErr != nil && !netErr.IsDialError() { err = errors.Wrapf(err, "error sending request to function %v", fnMeta.Name) return resp, err } diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go index d3150a61..d0114537 100644 --- a/pkg/utils/utils.go +++ b/pkg/utils/utils.go @@ -95,22 +95,6 @@ func LoggingMiddleware(logger *zap.Logger) func(next http.Handler) http.Handler } } -// IsNetworkDialError returns true if its a network dial error -func IsNetworkDialError(err error) bool { - netErr, ok := err.(net.Error) - if !ok { - return false - } - netOpErr, ok := netErr.(*net.OpError) - if !ok { - return false - } - if netOpErr.Op == "dial" { - return true - } - return false -} - // IsReadyPod checks both all containers in a pod are ready and whether // the .metadata.DeletionTimestamp is nil. func IsReadyPod(pod *apiv1.Pod) bool {