Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Istio integration (#421)

Browse Source 
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/
This commit is contained in:
Ta-Ching Chen
2018-02-28 04:09:32 +08:00
committed by GitHub
parent b9a559af2f
commit 23942fdf7d
29 changed files with 731 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
executor/fscache/functionServiceCache.go
+3 -1
View File
@@ -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
executor/newdeploy/newdeploy.go
+55 -7
View File
@@ -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,
},
},
},
executor/newdeploy/newdeploymgr.go
+24 -7
View File
@@ -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 {
executor/poolmgr/gp.go
+137 -14
View File
@@ -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)
executor/poolmgr/gpm.go
+6 -2
View File
@@ -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)
}
}