From 23942fdf7db917eae3052e8c4b7e356c0c189b63 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Wed, 28 Feb 2018 04:09:32 +0800 Subject: [PATCH] Istio integration (#421) This the very first step for fission to integrate with Istio, which is an open platform to connect, manage, and secure microservices. With Istio, users are able to monitor functions usage and trace requests latency through dashboards. For more information, please visit http://fission.io/docs/ --- .../tutorial/enabling-istio-on-fission.md | 203 ++++++++++++++++++ builder/client/client.go | 24 ++- buildermgr/buildermgr.go | 4 +- buildermgr/common.go | 2 +- buildermgr/envwatcher.go | 71 ++++-- buildermgr/pkgwatcher.go | 3 +- charts/fission-all/templates/deployment.yaml | 15 ++ .../templates/post-install-job.yaml | 4 + charts/fission-all/values.yaml | 3 + charts/fission-core/templates/deployment.yaml | 15 ++ .../templates/post-install-job.yaml | 4 + charts/fission-core/values.yaml | 3 + common.go | 12 ++ controller/api.go | 18 ++ controller/functionApi.go | 86 +++++++- crd/crd.go | 29 ++- environments/fetcher/client/client.go | 36 ++-- environments/fetcher/cmd/main.go | 9 - executor/fscache/functionServiceCache.go | 4 +- executor/newdeploy/newdeploy.go | 62 +++++- executor/newdeploy/newdeploymgr.go | 31 ++- executor/poolmgr/gp.go | 151 +++++++++++-- executor/poolmgr/gpm.go | 8 +- fission/environment.go | 9 +- fission/main.go | 5 +- mqtrigger/messageQueue/messageQueue.go | 6 + router/mutablemux.go | 3 +- timer/timerSync.go | 6 + types.go | 3 + 29 files changed, 731 insertions(+), 98 deletions(-) create mode 100644 Documentation/docs-site/content/tutorial/enabling-istio-on-fission.md diff --git a/Documentation/docs-site/content/tutorial/enabling-istio-on-fission.md b/Documentation/docs-site/content/tutorial/enabling-istio-on-fission.md new file mode 100644 index 00000000..3b498da7 --- /dev/null +++ b/Documentation/docs-site/content/tutorial/enabling-istio-on-fission.md @@ -0,0 +1,203 @@ +--- +title: "Enabling Istio on Fission" +draft: false +weight: 42 +--- + +This is the very first step for fission to integrate with [Istio](https://istio.io/). For those interested in trying to integrate fission with istio, following is the set up tutorial. + +## Test Environment +* Google Kubernetes Engine: 1.9.2-gke.1 + +## Set Up +### Create Kubernetes v1.9+ cluster + +Enable both RBAC & initializer features on kubernetes cluster. + +``` bash +$ export ZONE= +$ gcloud container clusters create istio-demo-1 \ + --machine-type=n1-standard-2 \ + --num-nodes=1 \ + --no-enable-legacy-authorization \ + --zone=$ZONE \ + --cluster-version=1.9.2-gke.1 +``` + +### Grant cluster admin permissions + +Grant admin permission for `system:serviceaccount:kube-system:default` and current user. + +``` bash +# for system:serviceaccount:kube-system:default +$ kubectl create clusterrolebinding --user system:serviceaccount:kube-system:default kube-system-cluster-admin --clusterrole cluster-admin + +# for current user +$ kubectl create clusterrolebinding cluster-admin-binding --clusterrole=cluster-admin --user=$(gcloud config get-value core/account) +``` + +### Set up Istio environment + +For Istio 0.5.1 you can follow the installation tutorial below. Also, you can follow the latest installation guides on Istio official site: [Quick Start](https://istio.io/docs/setup/kubernetes/quick-start.html) and [Sidecar Injection](https://istio.io/docs/setup/kubernetes/sidecar-injection.html). + + +Download Istio 0.5.1 + +``` bash +$ export ISTIO_VERSION=0.5.1 +$ curl -L https://git.io/getLatestIstio | sh - +$ cd istio-0.5.1 +``` + +Apply istio related YAML files + +``` bash +$ kubectl apply -f install/kubernetes/istio.yaml +``` + +Automatic sidecar injection + +``` bash +$ kubectl api-versions | grep admissionregistration +admissionregistration.k8s.io/v1beta1 +``` + +Installing the webhook + +Download the missing files in istio release 0.5.1 + +``` bash +$ wget https://raw.githubusercontent.com/istio/istio/master/install/kubernetes/webhook-create-signed-cert.sh -P install/kubernetes/ +$ wget https://raw.githubusercontent.com/istio/istio/master/install/kubernetes/webhook-patch-ca-bundle.sh -P install/kubernetes/ +$ chmod +x install/kubernetes/webhook-create-signed-cert.sh install/kubernetes/webhook-patch-ca-bundle.sh +``` + +Install the sidecar injection configmap. + +``` bash +$ ./install/kubernetes/webhook-create-signed-cert.sh \ + --service istio-sidecar-injector \ + --namespace istio-system \ + --secret sidecar-injector-certs + +$ kubectl apply -f install/kubernetes/istio-sidecar-injector-configmap-release.yaml +``` + +Install the sidecar injector + +``` bash +$ cat install/kubernetes/istio-sidecar-injector.yaml | \ + ./install/kubernetes/webhook-patch-ca-bundle.sh > \ + install/kubernetes/istio-sidecar-injector-with-ca-bundle.yaml + +$ kubectl apply -f install/kubernetes/istio-sidecar-injector-with-ca-bundle.yaml + +# Check sidecar injector status +$ kubectl -n istio-system get deployment -listio=sidecar-injector +NAME DESIRED CURRENT UP-TO-DATE AVAILABLE AGE +istio-sidecar-injector 1 1 1 1 26s +``` + +### Install fission + +Set default namespace for helm installation, here we use `fission` as example namespace. +``` bash +$ export FISSION_NAMESPACE=fission +``` + +Create namespace & add label for Istio sidecar injection. + +``` bash +$ kubectl create namespace $FISSION_NAMESPACE +$ kubectl label namespace $FISSION_NAMESPACE istio-injection=enabled +$ kubectl config set-context $(kubectl config current-context) --namespace=$FISSION_NAMESPACE +``` + +Follow the [installation guide](../../installation/) to install fission with flag `enableIstio` true. + +``` bash +$ helm install --namespace $FISSION_NAMESPACE --set enableIstio=true --name istio-demo +``` + +### Create a function + +Set environment + +``` bash +$ export FISSION_URL=http://$(kubectl --namespace fission get svc controller -o=jsonpath='{..ip}') +$ export FISSION_ROUTER=$(kubectl --namespace fission get svc router -o=jsonpath='{..ip}') +``` + +Let's create a simple function with Node.js. + +``` js +# hello.js +module.exports = async function(context) { + console.log(context.request.headers); + return { + status: 200, + body: "Hello, World!\n" + }; +} +``` + +Create environment + +``` bash +$ fission env create --name nodejs --image fission/node-env:latest +``` + +Create function + +``` bash +$ fission fn create --name h1 --env nodejs --code hello.js --method GET +``` + +Create route + +``` bash +$ fission route create --method GET --url /h1 --function h1 +``` + +Access function + +``` bash +$ curl http://$FISSION_ROUTER/h1 +Hello, World! +``` + + +### Install Istio Add-ons + +* Prometheus + +``` bash +$ kubectl apply -f istio-0.5.1/install/kubernetes/addons/prometheus.yaml +$ kubectl -n istio-system port-forward $(kubectl -n istio-system get pod -l app=prometheus -o jsonpath='{.items[0].metadata.name}') 9090:9090 +``` + +Web Link: [http://127.0.0.1:9090/graph](http://127.0.0.1:9090/graph) + +* Grafana + +Please install Prometheus first. + +![grafana min](https://user-images.githubusercontent.com/202578/33528556-639493e2-d89d-11e7-9768-976fb9208646.png) + +``` bash +$ kubectl apply -f istio-0.5.1/install/kubernetes/addons/grafana.yaml +$ kubectl -n istio-system port-forward $(kubectl -n istio-system get pod -l app=grafana -o jsonpath='{.items[0].metadata.name}') 3000:3000 +``` + +Web Link: [http://127.0.0.1:3000/dashboard/db/istio-dashboard](http://127.0.0.1:3000/dashboard/db/istio-dashboard) + +* Jaegar + +![jaeger min](https://user-images.githubusercontent.com/202578/33528554-572c4f28-d89d-11e7-8a01-1543fc2aa064.png) + +``` bash +$ kubectl apply -n istio-system -f https://raw.githubusercontent.com/jaegertracing/jaeger-kubernetes/master/all-in-one/jaeger-all-in-one-template.yml +$ kubectl port-forward -n istio-system $(kubectl get pod -n istio-system -l app=jaeger -o jsonpath='{.items[0].metadata.name}') 16686:16686 +``` + +Web Link: [http://localhost:16686](http://localhost:16686) diff --git a/builder/client/client.go b/builder/client/client.go index 4f2cdf5a..38525eec 100644 --- a/builder/client/client.go +++ b/builder/client/client.go @@ -23,6 +23,7 @@ import ( "log" "net/http" "strings" + "time" "github.com/fission/fission" builder "github.com/fission/fission/builder" @@ -45,10 +46,29 @@ func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildR if err != nil { return nil, err } - resp, err := http.Post(c.url, "application/json", bytes.NewReader(body)) - if err != nil { + + maxRetries := 20 + var resp *http.Response + + for i := 0; i < maxRetries; i++ { + resp, err = http.Post(c.url, "application/json", bytes.NewReader(body)) + + if err == nil { + if resp.StatusCode == 200 { + break + } + err = fission.MakeErrorFromHTTP(resp) + } + + if i < maxRetries-1 { + time.Sleep(50 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error building package (%v), retrying", err) + continue + } + return nil, err } + defer resp.Body.Close() rBody, err := ioutil.ReadAll(resp.Body) diff --git a/buildermgr/buildermgr.go b/buildermgr/buildermgr.go index 0a26bad0..9dac1495 100644 --- a/buildermgr/buildermgr.go +++ b/buildermgr/buildermgr.go @@ -24,6 +24,7 @@ import ( // Start the buildermgr service. func Start(storageSvcUrl string, envBuilderNamespace string) error { + fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() if err != nil { log.Printf("Failed to get kubernetes client: %v", err) @@ -33,7 +34,8 @@ func Start(storageSvcUrl string, envBuilderNamespace string) error { envWatcher := makeEnvironmentWatcher(fissionClient, kubernetesClient, envBuilderNamespace) go envWatcher.watchEnvironments() - pkgWatcher := makePackageWatcher(fissionClient, kubernetesClient.CoreV1().RESTClient(), envBuilderNamespace, storageSvcUrl) + pkgWatcher := makePackageWatcher(fissionClient, + kubernetesClient.CoreV1().RESTClient(), envBuilderNamespace, storageSvcUrl) go pkgWatcher.watchPackages() select {} diff --git a/buildermgr/common.go b/buildermgr/common.go index f0bffe58..6e3629f3 100644 --- a/buildermgr/common.go +++ b/buildermgr/common.go @@ -90,7 +90,7 @@ func buildPackage(fissionClient *crd.FissionClient, builderNamespace string, buildLogs = buildResp.BuildLogs } buildLogs += fmt.Sprintf("%v\n", e) - return nil, buildResp.BuildLogs, fission.MakeError(http.StatusInternalServerError, e) + return nil, buildLogs, fission.MakeError(http.StatusInternalServerError, e) } log.Printf("Build succeed, source package: %v, deployment package: %v", srcPkgFilename, buildResp.ArtifactFilename) diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go index ed6f694b..4a8df74a 100644 --- a/buildermgr/envwatcher.go +++ b/buildermgr/envwatcher.go @@ -20,6 +20,7 @@ import ( "fmt" "log" "os" + "strconv" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -30,6 +31,7 @@ import ( apiv1 "k8s.io/client-go/pkg/api/v1" "k8s.io/client-go/pkg/apis/extensions/v1beta1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) @@ -70,12 +72,23 @@ type ( kubernetesClient *kubernetes.Clientset fetcherImage string fetcherImagePullPolicy apiv1.PullPolicy + useIstio bool } ) func makeEnvironmentWatcher(fissionClient *crd.FissionClient, kubernetesClient *kubernetes.Clientset, builderNamespace string) *environmentWatcher { + useIstio := false + enableIstio := os.Getenv("ENABLE_ISTIO") + if len(enableIstio) > 0 { + istio, err := strconv.ParseBool(enableIstio) + if err != nil { + log.Println("Failed to parse ENABLE_ISTIO, defaults to false") + } + useIstio = istio + } + fetcherImage := os.Getenv("FETCHER_IMAGE") if len(fetcherImage) == 0 { fetcherImage = "fission/fetcher" @@ -104,6 +117,7 @@ func makeEnvironmentWatcher(fissionClient *crd.FissionClient, kubernetesClient: kubernetesClient, fetcherImage: fetcherImage, fetcherImagePullPolicy: pullPolicy, + useIstio: useIstio, } go envWatcher.service() @@ -129,6 +143,11 @@ func (envw *environmentWatcher) watchEnvironments() { ResourceVersion: rv, }) if err != nil { + if fission.IsNetworkError(err) { + log.Printf("Encounter network error, retrying later: %v", err) + time.Sleep(5 * time.Second) + continue + } log.Fatalf("Error watching environment list: %v", err) } @@ -152,25 +171,36 @@ func (envw *environmentWatcher) watchEnvironments() { } func (envw *environmentWatcher) sync() { - envList, err := envw.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) - if err != nil { - log.Fatalf("Error syncing environment CRD resources: %v", err) - } - - // Create environment builders for all environments - for i := range envList.Items { - env := envList.Items[i] - - if env.Spec.Version == 1 || // builder is not supported with v1 interface - len(env.Spec.Builder.Image) == 0 { // ignore env without builder image - continue - } - _, err := envw.getEnvBuilder(&env) + maxRetries := 10 + for i := 0; i < maxRetries; i++ { + envList, err := envw.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { - log.Printf("Error creating builder for %v: %v", env.Metadata.Name, err) + if fission.IsNetworkError(err) { + log.Printf("Error syncing environment CRD resources due to network error, retrying later: %v", err) + time.Sleep(50 * time.Duration(2*i) * time.Millisecond) + continue + } + log.Fatalf("Error syncing environment CRD resources: %v", err) } + + // Create environment builders for all environments + for i := range envList.Items { + env := envList.Items[i] + + if env.Spec.Version == 1 || // builder is not supported with v1 interface + len(env.Spec.Builder.Image) == 0 { // ignore env without builder image + continue + } + _, err := envw.getEnvBuilder(&env) + if err != nil { + log.Printf("Error creating builder for %v: %v", env.Metadata.Name, err) + } + } + + // Remove environment builders no longer needed + envw.cleanupEnvBuilders(envList.Items) + break } - envw.cleanupEnvBuilders(envList.Items) } func (envw *environmentWatcher) service() { @@ -423,6 +453,12 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment) (* name := envw.getCacheKey(env.Metadata.Name, env.Metadata.ResourceVersion) sel := envw.getLabels(env.Metadata.Name, env.Metadata.ResourceVersion) var replicas int32 = 1 + + podAnnotation := make(map[string]string) + if envw.useIstio && env.Spec.AllowAccessToExternalNetwork { + podAnnotation["sidecar.istio.io/inject"] = "false" + } + deployment := &v1beta1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Namespace: envw.builderNamespace, @@ -436,7 +472,8 @@ func (envw *environmentWatcher) createBuilderDeployment(env *crd.Environment) (* }, Template: apiv1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ - Labels: sel, + Labels: sel, + Annotations: podAnnotation, }, Spec: apiv1.PodSpec{ Volumes: []apiv1.Volume{ diff --git a/buildermgr/pkgwatcher.go b/buildermgr/pkgwatcher.go index e7714041..30ab11ea 100644 --- a/buildermgr/pkgwatcher.go +++ b/buildermgr/pkgwatcher.go @@ -138,7 +138,8 @@ func (pkgw *packageWatcher) build(buildCache *cache.Cache, pkg *crd.Package) { break } - uploadResp, buildLogs, err := buildPackage(pkgw.fissionClient, pkgw.builderNamespace, pkgw.storageSvcUrl, pkg) + uploadResp, buildLogs, err := buildPackage(pkgw.fissionClient, + pkgw.builderNamespace, pkgw.storageSvcUrl, pkg) if err != nil { log.Printf("Error building package %v: %v", pkg.Metadata.Name, err) updatePackage(pkgw.fissionClient, pkg, fission.BuildStatusFailed, buildLogs, nil) diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index e2832777..b42bfeea 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -6,6 +6,9 @@ metadata: labels: name: fission-function chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + {{- if .Values.enableIstio }} + istio-injection: enabled + {{- end }} --- apiVersion: v1 @@ -15,6 +18,9 @@ metadata: labels: name: fission-builder chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + {{- if .Values.enableIstio }} + istio-injection: enabled + {{- end }} --- apiVersion: v1 @@ -130,6 +136,11 @@ spec: imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] args: ["--controllerPort", "8888"] + env: + - name: FISSION_FUNCTION_NAMESPACE + value: "{{ .Values.functionNamespace }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" @@ -223,6 +234,8 @@ spec: value: "{{ .Values.pullPolicy }}" - name: RUNTIME_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" @@ -263,6 +276,8 @@ spec: value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" - name: FETCHER_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" serviceAccount: fission-svc --- diff --git a/charts/fission-all/templates/post-install-job.yaml b/charts/fission-all/templates/post-install-job.yaml index dd4e2d9d..7f660114 100644 --- a/charts/fission-all/templates/post-install-job.yaml +++ b/charts/fission-all/templates/post-install-job.yaml @@ -22,6 +22,10 @@ spec: labels: release: {{ .Release.Name }} app: {{ template "name" . }} + annotations: + {{- if .Values.enableIstio }} + "sidecar.istio.io/inject": "false" + {{- end }} spec: restartPolicy: Never containers: diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index de0ecea4..3b1812a9 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -41,6 +41,9 @@ functionNamespace: fission-function ## the release namespace) builderNamespace: fission-builder +## Enable istio integration +enableIstio: false + ## Logger config logger: influxdbAdmin: "admin" diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index a98098b2..d241f216 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -6,6 +6,9 @@ metadata: labels: name: fission-function chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + {{- if .Values.enableIstio }} + istio-injection: enabled + {{- end }} --- apiVersion: v1 @@ -15,6 +18,9 @@ metadata: labels: name: fission-builder chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + {{- if .Values.enableIstio }} + istio-injection: enabled + {{- end }} --- apiVersion: v1 @@ -130,6 +136,11 @@ spec: imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] args: ["--controllerPort", "8888"] + env: + - name: FISSION_FUNCTION_NAMESPACE + value: "{{ .Values.functionNamespace }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" @@ -221,6 +232,8 @@ spec: value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" - name: FETCHER_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" readinessProbe: httpGet: path: "/healthz" @@ -261,6 +274,8 @@ spec: value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" - name: FETCHER_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" + - name: ENABLE_ISTIO + value: "{{ .Values.enableIstio }}" serviceAccount: fission-svc --- diff --git a/charts/fission-core/templates/post-install-job.yaml b/charts/fission-core/templates/post-install-job.yaml index dd4e2d9d..7f660114 100644 --- a/charts/fission-core/templates/post-install-job.yaml +++ b/charts/fission-core/templates/post-install-job.yaml @@ -22,6 +22,10 @@ spec: labels: release: {{ .Release.Name }} app: {{ template "name" . }} + annotations: + {{- if .Values.enableIstio }} + "sidecar.istio.io/inject": "false" + {{- end }} spec: restartPolicy: Never containers: diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index c99fd577..f9b5c792 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -38,6 +38,9 @@ functionNamespace: fission-function ## the release namespace) builderNamespace: fission-builder +## Enable istio integration +enableIstio: false + ## Persist data to a persistent volume. persistence: enabled: true diff --git a/common.go b/common.go index 5c7d444b..f51bf2b6 100644 --- a/common.go +++ b/common.go @@ -18,6 +18,7 @@ package fission import ( "fmt" + "net" "os" "os/signal" "runtime/debug" @@ -40,3 +41,14 @@ func SetupStackTraceHandler() { os.Exit(1) }() } + +// IsNetworkError returns true if an error is a network error, and false otherwise. +func IsNetworkError(err error) bool { + _, ok := err.(net.Error) + return ok +} + +// GetFunctionIstioServiceName return service name of function for istio feature +func GetFunctionIstioServiceName(fnName, fnNamespace string) string { + return fmt.Sprintf("istio-%v-%v", fnName, fnNamespace) +} diff --git a/controller/api.go b/controller/api.go index 00bdd150..1b828493 100644 --- a/controller/api.go +++ b/controller/api.go @@ -21,6 +21,7 @@ import ( "net/http" "os" "runtime/debug" + "strconv" "strings" "github.com/gorilla/handlers" @@ -41,6 +42,8 @@ type ( storageServiceUrl string builderManagerUrl string workflowApiUrl string + functionNamespace string + useIstio bool } logDBConfig struct { @@ -74,6 +77,21 @@ func MakeAPI() (*API, error) { api.workflowApiUrl = "http://workflows-apiserver" } + fnNs := os.Getenv("FISSION_FUNCTION_NAMESPACE") + if len(fnNs) > 0 { + api.functionNamespace = fnNs + } else { + api.functionNamespace = "fission-function" + } + + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + log.Println("Failed to parse ENABLE_ISTIO") + } + api.useIstio = istio + } + return api, err } diff --git a/controller/functionApi.go b/controller/functionApi.go index e6b77b03..b13adb93 100644 --- a/controller/functionApi.go +++ b/controller/functionApi.go @@ -29,14 +29,22 @@ import ( "github.com/gorilla/mux" log "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/pkg/api" - "k8s.io/client-go/pkg/api/v1" + apiv1 "k8s.io/client-go/pkg/api/v1" restclient "k8s.io/client-go/rest" "github.com/fission/fission" "github.com/fission/fission/crd" ) +func (a *API) getIstioServiceLabels(fnName string) map[string]string { + return map[string]string{ + "functionName": fnName, + } +} + func (a *API) FunctionApiList(w http.ResponseWriter, r *http.Request) { funcs, err := a.fissionClient.Functions(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { @@ -85,6 +93,62 @@ func (a *API) FunctionApiCreate(w http.ResponseWriter, r *http.Request) { return } + // Since istio only allows accessing pod through k8s service, + // for the functions with executor type "poolmgr" we need to + // create a service for sending requests to pod in pool. + // Functions with executor type "Newdeploy" is specialized at + // pod starts. In this case, just ignore such functions. + fnExecutorType := f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + + if a.useIstio && fnExecutorType == fission.ExecutorTypePoolmgr { + // create a same name service for function + // since istio only allows the traffic to service + + sel := map[string]string{ + "functionName": fnew.Metadata.Name, + "functionUid": string(fnew.Metadata.UID), + } + + // service for accepting user traffic + svc := apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: a.functionNamespace, + Name: fission.GetFunctionIstioServiceName(f.Metadata.Name, f.Metadata.Namespace), + Labels: a.getIstioServiceLabels(f.Metadata.Name), + }, + Spec: apiv1.ServiceSpec{ + Type: apiv1.ServiceTypeClusterIP, + Ports: []apiv1.ServicePort{ + // Service port name should begin with a recognized prefix, or the traffic will be + // treated as TCP traffic. (https://istio.io/docs/setup/kubernetes/sidecar-injection.html) + // Originally the ports' name are similar to "http-fetch" and "http-specialize". + // But for istio 0.5.1, istio-proxy return unexpected 431 error with such naming. + // https://github.com/istio/istio/issues/928 + // Workaround: remove prefix + // TODO: prepend prefix once the bug fixed + { + Name: "fetch", + Protocol: apiv1.ProtocolTCP, + Port: 8000, + TargetPort: intstr.FromInt(8000), + }, + { + Name: "specialize", + Protocol: apiv1.ProtocolTCP, + Port: 8888, + TargetPort: intstr.FromInt(8888), + }, + }, + Selector: sel, + }, + } + _, err = a.kubernetesClient.CoreV1().Services(a.functionNamespace).Create(&svc) + if err != nil { + a.respondWithError(w, err) + return + } + } + w.WriteHeader(http.StatusCreated) a.respondWithSuccess(w, resp) } @@ -162,6 +226,20 @@ func (a *API) FunctionApiDelete(w http.ResponseWriter, r *http.Request) { return } + if a.useIstio { + // delete all istio services belong to the function + sel := a.getIstioServiceLabels(name) + svcList, err := a.kubernetesClient.CoreV1().Services(a.functionNamespace).List(metav1.ListOptions{ + LabelSelector: labels.Set(sel).AsSelector().String(), + }) + for _, svc := range svcList.Items { + err = a.kubernetesClient.CoreV1().Services(a.functionNamespace).Delete(svc.ObjectMeta.Name, &metav1.DeleteOptions{}) + // log error and continue + log.Printf("Failed to delete service %v: %v", svc.ObjectMeta.Name, err) + continue + } + } + a.respondWithSuccess(w, []byte("")) } @@ -219,7 +297,7 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) { // Get function Pods first selector := "functionName=" + fnName - podList, err := a.kubernetesClient.Core().Pods(ns).List(metav1.ListOptions{LabelSelector: selector}) + podList, err := a.kubernetesClient.CoreV1().Pods(ns).List(metav1.ListOptions{LabelSelector: selector}) if err != nil { a.respondWithError(w, err) return @@ -233,10 +311,10 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) { return itime.After(jtime) }) - podLogOpts := v1.PodLogOptions{Container: envName} // Only the env container, not fetcher + podLogOpts := apiv1.PodLogOptions{Container: envName} // Only the env container, not fetcher var podLogsReq *restclient.Request if len(pods) > 0 { - podLogsReq = a.kubernetesClient.Core().Pods(ns).GetLogs(pods[0].ObjectMeta.Name, &podLogOpts) + podLogsReq = a.kubernetesClient.CoreV1().Pods(ns).GetLogs(pods[0].ObjectMeta.Name, &podLogOpts) } else { a.respondWithError(w, errors.New("No active pods found")) return diff --git a/crd/crd.go b/crd/crd.go index 6373b634..0e2c54be 100644 --- a/crd/crd.go +++ b/crd/crd.go @@ -17,6 +17,9 @@ limitations under the License. package crd import ( + "log" + "time" + apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" "k8s.io/apimachinery/pkg/api/errors" @@ -32,14 +35,28 @@ const ( // needed. (Note that this creates the CRD type; it doesn't create any // _instances_ of that type.) func ensureCRD(clientset *apiextensionsclient.Clientset, crd *apiextensionsv1beta1.CustomResourceDefinition) error { - _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Get(crd.ObjectMeta.Name, metav1.GetOptions{}) - if err != nil { - if errors.IsNotFound(err) { - _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) - if err != nil { - return err + maxRetries := 5 + for i := 0; i < maxRetries; i++ { + _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Get(crd.ObjectMeta.Name, metav1.GetOptions{}) + + if err != nil { + if errors.IsNotFound(err) { + // crd resource not found error + _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) + if err != nil { + return err + } + } else { + // The requests fail to connect to k8s api server before + // istio-prxoy is ready to serve traffic. Retry again. + log.Printf("Error connecting to kubernetes api service (%v), retrying", err) + time.Sleep(500 * time.Duration(2*i) * time.Millisecond) + continue } } + + // resource already exists + break } return nil } diff --git a/environments/fetcher/client/client.go b/environments/fetcher/client/client.go index 8708da6b..76bd1964 100644 --- a/environments/fetcher/client/client.go +++ b/environments/fetcher/client/client.go @@ -5,9 +5,7 @@ import ( "encoding/json" "io/ioutil" "log" - "net" "net/http" - "net/url" "time" "github.com/fission/fission" @@ -38,29 +36,25 @@ func (c *Client) Fetch(fr *fetcher.FetchRequest) error { for i := 0; i < maxRetries; i++ { resp, err = http.Post(c.url, "application/json", bytes.NewReader(body)) - if err == nil && resp.StatusCode == 200 { - defer resp.Body.Close() - return nil - } - - // 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(50 * time.Duration(2*i) * time.Millisecond) - continue - } - } + if err == nil { + if resp.StatusCode == 200 { + resp.Body.Close() + return nil } + err = fission.MakeErrorFromHTTP(resp) } + + if i < maxRetries-1 { + time.Sleep(50 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error fetching package (%v), retrying", err) + continue + } + + log.Printf("Failed to fetch: %v", err) + return err } - if err == nil { - err = fission.MakeErrorFromHTTP(resp) - } - log.Printf("Failed to fetch: %v", err) - return err + return nil } func (c *Client) Upload(fr *fetcher.UploadRequest) (*fetcher.UploadResponse, error) { diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 099e9d64..1d930deb 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -128,15 +128,6 @@ func specializePod(f *fetcher.Fetcher, fetchPayload *string, loadPayload *string if err == nil && resp.StatusCode < 300 { // Success resp.Body.Close() - //On Success creates a file which is used as a readiness probe by Kubernetes for this container/pod - file, err := os.OpenFile("/tmp/ready", os.O_RDONLY|os.O_CREATE, 0666) - if err != nil { - log.Fatalf("Error creating readiness file: %v", err) - } - err = file.Close() - if err != nil { - log.Fatalf("Error closing readiness file: %v", err) - } break } diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 34219b7b..207b7229 100644 --- a/executor/fscache/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -163,7 +163,9 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { err = nil } } - log.Printf("error caching fsvc: %v", err) + if err != nil { + log.Printf("error caching fsvc: %v", err) + } return nil, err } return nil, nil diff --git a/executor/newdeploy/newdeploy.go b/executor/newdeploy/newdeploy.go index ed619399..bb794358 100644 --- a/executor/newdeploy/newdeploy.go +++ b/executor/newdeploy/newdeploy.go @@ -19,6 +19,7 @@ package newdeploy import ( "encoding/json" "errors" + "fmt" "log" "path/filepath" "strconv" @@ -55,6 +56,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro } targetFilename := "user" userfunc := "userfunc" + var gracePeriodSeconds int64 = 6 * 60 existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{}) if err == nil && existingDepl.Status.ReadyReplicas >= replicas { @@ -83,6 +85,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro if err != nil { return nil, err } + loadPayload, err := json.Marshal(loadReq) if err != nil { return nil, err @@ -94,6 +97,11 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro return nil, err } + podAnnotation := make(map[string]string) + if deploy.useIstio && env.Spec.AllowAccessToExternalNetwork { + podAnnotation["sidecar.istio.io/inject"] = "false" + } + deployment := &v1beta1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Labels: deployLabels, @@ -106,7 +114,8 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro }, Template: apiv1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ - Labels: deployLabels, + Labels: deployLabels, + Annotations: podAnnotation, }, Spec: apiv1.PodSpec{ Volumes: []apiv1.Volume{ @@ -130,6 +139,16 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro }, }, Resources: env.Spec.Resources, + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, }, { Name: "fetcher", @@ -148,6 +167,16 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro "-secret-dir", deploy.sharedSecretPath, "-cfgmap-dir", deploy.sharedCfgMapPath, deploy.sharedMountPath}, + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, Env: []apiv1.EnvVar{ { Name: envVersion, @@ -157,17 +186,36 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro // TBD Use smaller default resources, for now needed to make HPA work Resources: fetcherResources, ReadinessProbe: &apiv1.Probe{ - Handler: apiv1.Handler{ - Exec: &apiv1.ExecAction{ - Command: []string{"cat", "/tmp/ready"}, - }, - }, InitialDelaySeconds: 1, PeriodSeconds: 1, + FailureThreshold: 30, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, + }, + LivenessProbe: &apiv1.Probe{ + InitialDelaySeconds: 35, + PeriodSeconds: 5, + Handler: apiv1.Handler{ + HTTPGet: &apiv1.HTTPGetAction{ + Path: "/healthz", + Port: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + }, }, }, }, - ServiceAccountName: "fission-fetcher", + ServiceAccountName: "fission-fetcher", + TerminationGracePeriodSeconds: &gracePeriodSeconds, }, }, }, diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index f16616e8..baa9c34a 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -21,6 +21,7 @@ import ( "fmt" "log" "os" + "strconv" "time" "github.com/fission/fission" @@ -50,6 +51,7 @@ type ( sharedMountPath string sharedSecretPath string sharedCfgMapPath string + useIstio bool fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname requestChannel chan *fnRequest @@ -97,6 +99,15 @@ func MakeNewDeploy( fetcherImagePullPolicy = "IfNotPresent" } + enableIstio := false + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + log.Println("Failed to parse ENABLE_ISTIO") + } + enableIstio = istio + } + nd := &NewDeploy{ fissionClient: fissionClient, kubernetesClient: kubernetesClient, @@ -111,6 +122,7 @@ func MakeNewDeploy( sharedMountPath: "/userfunc", sharedSecretPath: "/secrets", sharedCfgMapPath: "/configs", + useIstio: enableIstio, requestChannel: make(chan *fnRequest), } @@ -249,18 +261,23 @@ func (deploy *NewDeploy) fnCreate(fn *crd.Function) (*fscache.FuncSvc, error) { "executorType": fission.ExecutorTypeNewdeploy, } - depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels) - if err != nil { - log.Printf("Error creating the deployment %v: %v", objName, err) - return fsvc, err - } - + // Envoy(istio-proxy) returns 404 directly before istio pilot + // propagates latest Envoy-specific configuration. + // Since newdeploy waits for pods of deployment to be ready, + // change the order of kubeObject creation (create service first, + // then deployment) to take advantage of waiting time. svc, err := deploy.createOrGetSvc(deployLabels, objName) if err != nil { log.Printf("Error creating the service %v: %v", objName, err) return fsvc, err } - svcAddress := svc.Spec.ClusterIP + svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) + + depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels) + if err != nil { + log.Printf("Error creating the deployment %v: %v", objName, err) + return fsvc, err + } hpa, err := deploy.createOrGetHpa(objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl) if err != nil { diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index feb1b121..93e68512 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -28,6 +28,7 @@ import ( "net/url" "os" "path/filepath" + "strconv" "strings" "time" @@ -60,7 +61,8 @@ type ( idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname useSvc bool // create k8s service for specialized pods - poolInstanceId string // small random string to uniquify pod names + useIstio bool + poolInstanceId string // small random string to uniquify pod names fetcherImage string fetcherImagePullPolicy apiv1.PullPolicy runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment @@ -120,6 +122,16 @@ func MakeGenericPool( runtimeImagePullPolicy = "IfNotPresent" } + enableIstio := false + + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + log.Println("Failed to parse ENABLE_ISTIO") + } + enableIstio = istio + } + // TODO: in general we need to provide the user a way to configure pools. Initial // replicas, autoscaling params, various timeouts, etc. gp := &GenericPool{ @@ -136,6 +148,7 @@ func MakeGenericPool( instanceId: instanceId, fetcherImage: fetcherImage, useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start + useIstio: enableIstio, // defaults off -- istio integration requires pod relabeling and it takes a second or more to become routable, slowing cold start sharedMountPath: "/userfunc", // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path sharedSecretPath: "/secrets", sharedCfgMapPath: "/configs", @@ -280,7 +293,11 @@ func (gp *GenericPool) scheduleDeletePod(name string) { // cleaned up. (We need a better solutions for both those things; log // aggregation and storage will help.) log.Printf("Error in pod '%v', scheduling cleanup", name) - time.Sleep(5 * time.Minute) + // Ignore sleep here if istio feature is enabled, function pod + // will be deleted after 6 mins (terminationGracePeriodSeconds). + if !gp.useIstio { + time.Sleep(5 * time.Minute) + } gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil) }() } @@ -339,10 +356,15 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta if len(podIP) == 0 { return errors.New("Pod has no IP") } + // specialize pod with service + if gp.useIstio { + svc := fission.GetFunctionIstioServiceName(metadata.Name, metadata.Namespace) + podIP = fmt.Sprintf("%v.%v", svc, gp.namespace) + } // tell fetcher to get the function. fetcherUrl := gp.getFetcherUrl(podIP) - log.Printf("[%v] calling fetcher to copy function", metadata.Name) + log.Printf("[%v] calling fetcher to copy function with fetcher url: %v", metadata.Name, fetcherUrl) fn, err := gp.fissionClient. Functions(metadata.Namespace). @@ -394,33 +416,48 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta var resp2 *http.Response if gp.env.Spec.Version == 2 { specializeUrl := gp.getSpecializeUrl(podIP, 2) + log.Printf("specialize url: %v", specializeUrl) resp2, err = http.Post(specializeUrl, "application/json", bytes.NewReader(body)) } else { specializeUrl := gp.getSpecializeUrl(podIP, 1) resp2, err = http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{})) } + if err == nil && resp2.StatusCode < 300 { // Success resp2.Body.Close() return nil } + retry := false + // 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) - log.Printf("Error connecting to pod (%v), retrying", netErr) - continue + retry = true } } } } + // Receive response with non-200 http code if err == nil { err = fission.MakeErrorFromHTTP(resp2) + // The istio-proxy block all http requests until it's ready + // to serve traffic. Retry if istio feature is enabled. + if gp.useIstio { + retry = true + } } + + if retry { + time.Sleep(500 * time.Duration(2*i) * time.Millisecond) + log.Printf("Error connecting to pod (%v), retrying", err) + continue + } + log.Printf("Failed to specialize pod: %v", err) return err } @@ -439,6 +476,16 @@ func (gp *GenericPool) createPool() error { return err } + // Use long terminationGracePeriodSeconds for connection draining in case that + // pod still runs user functions. + var gracePeriodSeconds int64 = 6 * 60 + + podAnnotation := make(map[string]string) + + if gp.useIstio && gp.env.Spec.AllowAccessToExternalNetwork { + podAnnotation["sidecar.istio.io/inject"] = "false" + } + deployment := &v1beta1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Name: poolDeploymentName, @@ -451,7 +498,8 @@ func (gp *GenericPool) createPool() error { }, Template: apiv1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ - Labels: gp.labelsForPool, + Labels: gp.labelsForPool, + Annotations: podAnnotation, }, Spec: apiv1.PodSpec{ Volumes: []apiv1.Volume{ @@ -461,14 +509,12 @@ func (gp *GenericPool) createPool() error { EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, - { Name: "secrets", VolumeSource: apiv1.VolumeSource{ EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, - { Name: "config", VolumeSource: apiv1.VolumeSource{ @@ -487,18 +533,32 @@ func (gp *GenericPool) createPool() error { Name: "userfunc", MountPath: gp.sharedMountPath, }, - { Name: "secrets", MountPath: gp.sharedSecretPath, }, - { Name: "config", MountPath: gp.sharedCfgMapPath, }, }, Resources: gp.env.Spec.Resources, + // Pod is removed from endpoints list for service when it's + // state became "Termination". We used preStop hook as the + // workaround for connection draining since pod maybe shutdown + // before grace period expires. + // https://kubernetes.io/docs/concepts/workloads/pods/pod/#termination-of-pods + // https://github.com/kubernetes/kubernetes/issues/47576#issuecomment-308900172 + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, }, { Name: "fetcher", @@ -510,12 +570,10 @@ func (gp *GenericPool) createPool() error { Name: "userfunc", MountPath: gp.sharedMountPath, }, - { Name: "secrets", MountPath: gp.sharedSecretPath, }, - { Name: "config", MountPath: gp.sharedCfgMapPath, @@ -526,6 +584,22 @@ func (gp *GenericPool) createPool() error { "-secret-dir", gp.sharedSecretPath, "-cfgmap-dir", gp.sharedCfgMapPath, gp.sharedMountPath}, + // Pod is removed from endpoints list for service when it's + // state became "Termination". We used preStop hook as the + // workaround for connection draining since pod maybe shutdown + // before grace period expires. + // https://kubernetes.io/docs/concepts/workloads/pods/pod/#termination-of-pods + // https://github.com/kubernetes/kubernetes/issues/47576#issuecomment-308900172 + Lifecycle: &apiv1.Lifecycle{ + PreStop: &apiv1.Handler{ + Exec: &apiv1.ExecAction{ + Command: []string{ + "sleep", + fmt.Sprintf("%v", gracePeriodSeconds), + }, + }, + }, + }, ReadinessProbe: &apiv1.Probe{ InitialDelaySeconds: 1, PeriodSeconds: 1, @@ -556,10 +630,15 @@ func (gp *GenericPool) createPool() error { }, }, ServiceAccountName: "fission-fetcher", + // TerminationGracePeriodSeconds should be equal to the + // sleep time of preStop to make sure that SIGTERM is sent + // to pod after 6 mins. + TerminationGracePeriodSeconds: &gracePeriodSeconds, }, }, }, } + depl, err := gp.kubernetesClient.ExtensionsV1beta1().Deployments(gp.namespace).Create(deployment) if err != nil { log.Printf("Error creating deployment for %s in kubernetes, err: %v", deployment.Name, err) @@ -613,9 +692,50 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1. } func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) { - log.Printf("[%v] Choosing pod from pool", m.Name) newLabels := gp.labelsForFunction(m) + + if gp.useIstio { + // Istio only allows accessing pod through k8s service, and requests come to + // service are not always being routed to the same pod. For example: + + // If there is only one pod (podA) behind the service svcX. + + // svcX -> podA + + // All requests (specialize request & function access requests) + // will be routed to podA without any problem. + + // If podA and podB are behind svcX. + + // svcX -> podA (specialized) + // -> podB (non-specialized) + + // The specialize request may be routed to podA and the function access + // requests may go to podB. In this case, the function cannot be served + // properly. + + // To prevent such problem, we need to delete old versions function pods + // and make sure that there is only one pod behind the service + + sel := map[string]string{ + "functionName": m.Name, + "functionUid": string(m.UID), + } + podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(metav1.ListOptions{ + LabelSelector: labels.Set(sel).AsSelector().String(), + }) + if err != nil { + return nil, err + } + + // Remove old versions function pods + for _, pod := range podList.Items { + // Delete pod no matter what status it is + gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil) + } + } + pod, err := gp.choosePod(newLabels) if err != nil { return nil, err @@ -649,6 +769,9 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error // the fission router isn't in the same namespace, so return a // namespace-qualified hostname svcHost = fmt.Sprintf("%v.%v", svcName, gp.namespace) + } else if gp.useIstio { + svc := fission.GetFunctionIstioServiceName(m.Name, m.Namespace) + svcHost = fmt.Sprintf("%v.%v:8888", svc, gp.namespace) } else { log.Printf("Using pod IP for specialized pod") svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP) diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index 9bbc62d1..e0cff1e3 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -151,11 +151,14 @@ func (gpm *GenericPoolManager) CleanupPools(envs []crd.Environment) { func (gpm *GenericPoolManager) eagerPoolCreator() { pollSleep := time.Duration(2 * time.Second) for { - time.Sleep(pollSleep) - // get list of envs from controller envs, err := gpm.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { + if fission.IsNetworkError(err) { + log.Printf("Encountered network error, retrying: %v", err) + time.Sleep(5 * time.Second) + continue + } log.Fatalf("Failed to get environment list: %v", err) } @@ -176,6 +179,7 @@ func (gpm *GenericPoolManager) eagerPoolCreator() { // Clean up pools whose env was deleted gpm.CleanupPools(envs.Items) + time.Sleep(pollSleep) } } diff --git a/fission/environment.go b/fission/environment.go index 7aba0d0b..bff734d7 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -54,6 +54,7 @@ func envCreate(c *cli.Context) error { envVersion := c.Int("version") envBuilderImg := c.String("builder") envBuildCmd := c.String("buildcmd") + envExternalNetwork := c.Bool("externalnetwork") if len(envBuilderImg) > 0 { envVersion = 2 @@ -84,8 +85,9 @@ func envCreate(c *cli.Context) error { Image: envBuilderImg, Command: envBuildCmd, }, - Poolsize: poolsize, - Resources: resourceReq, + Poolsize: poolsize, + Resources: resourceReq, + AllowAccessToExternalNetwork: envExternalNetwork, }, } @@ -130,6 +132,7 @@ func envUpdate(c *cli.Context) error { envImg := c.String("image") envBuilderImg := c.String("builder") envBuildCmd := c.String("buildcmd") + envExternalNetwork := c.Bool("externalnetwork") if len(envImg) == 0 && len(envBuilderImg) == 0 && len(envBuildCmd) == 0 { fatal("Need --image to specify env image, or use --builder to specify env builder, or use --buildcmd to specify new build command.") @@ -160,6 +163,8 @@ func envUpdate(c *cli.Context) error { env.Spec.Poolsize = c.Int("poolsize") } + env.Spec.AllowAccessToExternalNetwork = envExternalNetwork + _, err = client.EnvironmentUpdate(env) checkErr(err, "update environment") diff --git a/fission/main.go b/fission/main.go index 4dccb52c..8a716faf 100644 --- a/fission/main.go +++ b/fission/main.go @@ -142,12 +142,13 @@ func main() { envImageFlag := cli.StringFlag{Name: "image", Usage: "Environment image URL"} envBuilderImageFlag := cli.StringFlag{Name: "builder", Usage: "Environment builder image URL (optional)"} envBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "Build command for environment builder to build source package (optional)"} + envExternalNetworkFlag := cli.BoolFlag{Name: "externalnetwork", Usage: "Allow environment access external network when istio feature enabled (optional, defaults to false)"} envVersionFlag := cli.IntFlag{Name: "version", Usage: "Environment API version: defaults to 1 (means v1 interface)"} envSubcommands := []cli.Command{ - {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag}, Action: envCreate}, + {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envVersionFlag, envExternalNetworkFlag}, Action: envCreate}, {Name: "get", Usage: "Get environment details", Flags: []cli.Flag{envNameFlag}, Action: envGet}, - {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem}, Action: envUpdate}, + {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envPoolsizeFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, minCpu, maxCpu, minMem, maxMem, envExternalNetworkFlag}, Action: envUpdate}, {Name: "delete", Usage: "Delete environment", Flags: []cli.Flag{envNameFlag}, Action: envDelete}, {Name: "list", Usage: "List all environments", Flags: []cli.Flag{}, Action: envList}, } diff --git a/mqtrigger/messageQueue/messageQueue.go b/mqtrigger/messageQueue/messageQueue.go index c2ff3bd2..b299c232 100644 --- a/mqtrigger/messageQueue/messageQueue.go +++ b/mqtrigger/messageQueue/messageQueue.go @@ -24,6 +24,7 @@ import ( log "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) @@ -169,6 +170,11 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() { // get new set of triggers newTriggers, err := mqt.fissionClient.MessageQueueTriggers(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { + if fission.IsNetworkError(err) { + log.Printf("Encounter network error, retry again: %v", err) + time.Sleep(5 * time.Second) + continue + } log.Fatalf("Failed to read message queue trigger list: %v", err) } newTriggerMap := make(map[string]*crd.MessageQueueTrigger) diff --git a/router/mutablemux.go b/router/mutablemux.go index f556fc09..525ebd50 100644 --- a/router/mutablemux.go +++ b/router/mutablemux.go @@ -17,10 +17,11 @@ limitations under the License. package router import ( - "github.com/gorilla/mux" "log" "net/http" "sync/atomic" + + "github.com/gorilla/mux" ) // diff --git a/timer/timerSync.go b/timer/timerSync.go index 4509be08..c75ac9ea 100644 --- a/timer/timerSync.go +++ b/timer/timerSync.go @@ -22,6 +22,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" "github.com/fission/fission/crd" ) @@ -45,6 +46,11 @@ func (ws *TimerSync) syncSvc() { for { triggers, err := ws.fissionClient.TimeTriggers(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { + if fission.IsNetworkError(err) { + log.Printf("Encounter network error, retry again: %v", err) + time.Sleep(5 * time.Second) + continue + } log.Fatalf("Failed get time trigger list: %v", err) } ws.timer.Sync(triggers.Items) diff --git a/types.go b/types.go index e627d363..cb520834 100644 --- a/types.go +++ b/types.go @@ -233,6 +233,9 @@ type ( // Optional, defaults to 'AllowedFunctionsPerContainerSingle' AllowedFunctionsPerContainer AllowedFunctionsPerContainer `json:"allowedFunctionsPerContainer,omitempty"` + // Optional, defaults to 'false' + AllowAccessToExternalNetwork bool `json:"allowAccessToExternalNetwork,omitempty"` + // Request and limit resources for the environment Resources v1.ResourceRequirements `json:"resources"`