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

Prevent deployment from rolling update due to different instance-id (#1454)

Browse Source 
The pod template is embedded inside the deployment. So if
the pod annotation contains instance-id, the deployment
will get updated and thus triggers a rolling update whenever
a new executor starts which is unwanted.

After this PR, poolmanager will patches instance-id when a
pod is chosen for a function.

For newdeploy, unlike poolmanager manages the lifecycle
of function pod directly, newdeploy is only responsible
to create the deployment so we append instance-id to top-
level controller (deployment) only.
This commit is contained in:
Ta-Ching Chen
2019-12-02 17:28:19 +08:00
committed by GitHub
parent 7f8cb69326
commit 3f3b11ffbf
6 changed files with 40 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/cms/cmscontroller.go
+3 -3
View File
@@ -85,7 +85,7 @@ func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClien
if err != nil {
logger.Error("Failed to get functions related to configmap", zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace))
}
recyclePods(logger, funcs, types)
refreshPods(logger, funcs, types)
}
},
})
@@ -131,7 +131,7 @@ func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
if err != nil {
logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace))
}
recyclePods(logger, funcs, types)
refreshPods(logger, funcs, types)
}
},
})
@@ -157,7 +157,7 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie
return relatedFunctions, nil
}
func recyclePods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) {
func refreshPods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) {
for _, f := range funcs {
var err error
pkg/executor/executortype/newdeploy/newdeploy.go
+4 -3
View File
@@ -192,12 +192,13 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen
if podAnnotations == nil {
podAnnotations = make(map[string]string)
}
// Here, we don't append deployAnnotations to podAnnotations
// since newdeploy doesn't manager pod lifecycle directly.
if deploy.useIstio && env.Spec.AllowAccessToExternalNetwork {
podAnnotations["sidecar.istio.io/inject"] = "false"
}
for k, v := range deployAnnotations {
podAnnotations[k] = v
}
resources := deploy.getResources(env, fn)
// Set maxUnavailable and maxSurge to 20% is because we want
pkg/executor/executortype/newdeploy/newdeploymgr.go
+5 -41
View File
@@ -19,7 +19,6 @@ package newdeploy
import (
"context"
"fmt"
"math/rand"
"os"
"strconv"
"strings"
@@ -245,44 +244,6 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err
}
func (deploy *NewDeploy) AdoptExistingResources() {
l := map[string]string{
types.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy),
}
podList, err := deploy.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(metav1.ListOptions{
LabelSelector: labels.Set(l).AsSelector().String(),
})
if err != nil {
deploy.logger.Error("error getting pod list", zap.Error(err))
return
}
// Unlike poolmanager manages the lifecycle of function pod directly,
// newdeploy is only responsible to create the deployment and Kubernetes
// will handle the rest. Hence we don't need to wait for the pod patching
// process to finish.
for i := range podList.Items {
pod := &podList.Items[i]
if !utils.IsReadyPod(pod) {
continue
}
go func() {
// avoid too many requests arrive Kubernetes API server at the same time.
time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond)
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, types.EXECUTOR_INSTANCEID_LABEL, deploy.instanceID)
pod, err = deploy.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch))
if err != nil {
// just log the error since it won't affect the function serving
deploy.logger.Warn("error patching executor instance ID of pod", zap.Error(err),
zap.String("pod", pod.Name), zap.String("ns", pod.Namespace))
return
}
deploy.logger.Info("adopt newdeploy function pod",
zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations))
}()
}
fnList, err := deploy.fissionClient.Functions(metav1.NamespaceAll).List(metav1.ListOptions{})
if err != nil {
deploy.logger.Error("error getting function list", zap.Error(err))
@@ -689,9 +650,12 @@ func (deploy *NewDeploy) updateFuncDeployment(fn *fv1.Function, env *fv1.Environ
return err
}
// use current replicas instead of minscale in the ExecutionStrategy.
// the resource version inside function packageRef is changed,
// so the content of fetchRequest in deployment cmd is different.
// Therefore, the deployment update will trigger a rolling update.
newDeployment, err := deploy.getDeploymentSpec(fn, env,
existingDepl.Spec.Replicas, fnObjName, ns, deployLabels, deploy.getDeployAnnotations(fn.Metadata))
existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy.
fnObjName, ns, deployLabels, deploy.getDeployAnnotations(fn.Metadata))
if err != nil {
deploy.updateStatus(fn, err, "failed to get new deployment spec while updating function")
return err
pkg/executor/executortype/poolmgr/gp.go
+19 -15
View File
@@ -146,7 +146,7 @@ func MakeGenericPool(
return gp, nil
}
func (gp *GenericPool) getDeployLabels() map[string]string {
func (gp *GenericPool) getEnvironmentPoolLabels() map[string]string {
return map[string]string{
types.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr),
types.ENVIRONMENT_NAME: gp.env.Metadata.Name,
@@ -246,6 +246,11 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro
// modified, this should fail; in that case just
// retry.
chosenPod.ObjectMeta.Labels = newLabels
// Append executor instance id to pod annotations to
// indicate this pod is managed by this executor.
chosenPod.ObjectMeta.Annotations = gp.getDeployAnnotations()
_, err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Update(chosenPod)
if err != nil {
gp.logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.ObjectMeta.Name))
@@ -259,7 +264,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro
}
func (gp *GenericPool) labelsForFunction(metadata *metav1.ObjectMeta) map[string]string {
label := gp.getDeployLabels()
label := gp.getEnvironmentPoolLabels()
label[types.FUNCTION_NAME] = metadata.Name
label[types.FUNCTION_UID] = string(metadata.UID)
label[types.FUNCTION_NAMESPACE] = metadata.Namespace // function CRD must stay within same namespace of environment CRD
@@ -274,11 +279,6 @@ func (gp *GenericPool) scheduleDeletePod(name string) {
// cleaned up. (We need a better solutions for both those things; log
// aggregation and storage will help.)
gp.logger.Error("error in pod - scheduling cleanup", zap.String("pod", name))
// 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)
}()
}
@@ -351,7 +351,7 @@ func (gp *GenericPool) getPoolName() string {
// A pool is a deployment of generic containers for an env. This
// creates the pool but doesn't wait for any pods to be ready.
func (gp *GenericPool) createPool() error {
deployLabels := gp.getDeployLabels()
deployLabels := gp.getEnvironmentPoolLabels()
deployAnnotations := gp.getDeployAnnotations()
// Use long terminationGracePeriodSeconds for connection draining in case that
@@ -365,12 +365,15 @@ func (gp *GenericPool) createPool() error {
if podAnnotations == nil {
podAnnotations = make(map[string]string)
}
// Here, we don't append executor instance-id to pod annotations
// to prevent unwanted rolling updates occur. Pool manager will
// append executor instance-id to pod annotations when a pod is chosen
// for function specialization.
if gp.useIstio && gp.env.Spec.AllowAccessToExternalNetwork {
podAnnotations["sidecar.istio.io/inject"] = "false"
}
for k, v := range deployAnnotations {
podAnnotations[k] = v
}
container, err := util.MergeContainer(&apiv1.Container{
Name: gp.env.Metadata.Name,
@@ -526,7 +529,8 @@ func (gp *GenericPool) waitForReadyPod() error {
func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.Service, error) {
service := apiv1.Service{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Name: name,
Labels: labels,
},
Spec: apiv1.ServiceSpec{
Type: apiv1.ServiceTypeClusterIP,
@@ -546,7 +550,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.
func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
gp.logger.Info("choosing pod from pool", zap.Any("function", fn.Metadata))
newLabels := gp.labelsForFunction(&fn.Metadata)
funcLabels := gp.labelsForFunction(&fn.Metadata)
if gp.useIstio {
// Istio only allows accessing pod through k8s service, and requests come to
@@ -589,7 +593,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
}
}
pod, err := gp.choosePod(newLabels)
pod, err := gp.choosePod(funcLabels)
if err != nil {
return nil, err
}
@@ -608,7 +612,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
svcName = fmt.Sprintf("%s-%v", svcName, fn.Metadata.UID)
}
svc, err := gp.createSvc(svcName, newLabels)
svc, err := gp.createSvc(svcName, funcLabels)
if err != nil {
gp.scheduleDeletePod(pod.ObjectMeta.Name)
return nil, err
pkg/executor/executortype/poolmgr/gpm.go
+6 -2
View File
@@ -183,7 +183,7 @@ func (gpm *GenericPoolManager) TapService(svcHost string) error {
// containers in it are reporting a ready status for the healthCheck.
func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
for _, obj := range fsvc.KubernetesObjects {
if obj.Kind == "pod" {
if strings.ToLower(obj.Kind) == "pod" {
pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{})
if err == nil && utils.IsReadyPod(pod) {
// Normally, the address format is http://[pod-ip]:[port], however, if the
@@ -192,7 +192,11 @@ func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
// Otherwise, we need to ensure that the address contains pod ip.
if gpm.enableIstio ||
(!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) {
gpm.logger.Debug("valid address", zap.String("address", fsvc.Address))
gpm.logger.Debug("valid address",
zap.String("address", fsvc.Address),
zap.Any("function", fsvc.Function),
zap.String("executor", string(fsvc.Executor)),
)
return true
}
}
pkg/executor/util/util.go
+3 -1
View File
@@ -31,7 +31,9 @@ import (
// when creating the environment deployment since kubelet will retry to
// pull image until successes.
func ApplyImagePullSecret(secret string, podspec apiv1.PodSpec) *apiv1.PodSpec {
podspec.ImagePullSecrets = []apiv1.LocalObjectReference{{Name: secret}}
if len(secret) > 0 {
podspec.ImagePullSecrets = []apiv1.LocalObjectReference{{Name: secret}}
}
return &podspec
}