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"`