Avoid downtime when updating functions (#699)

* Avoid downtime when updating functions

* Add deps checksum

pkg/langruntime/langruntime.go

@@ -177,8 +177,15 @@ func (l *Langruntimes) GetImageSecrets(runtime string) ([]v1.LocalObjectReferenc
 	return lors, nil
 }
 
+func appendToCommand(orig string, command ...string) string {
+	if len(orig) > 0 {
+		return fmt.Sprintf("%s && %s", orig, strings.Join(command, " && "))
+	}
+	return strings.Join(command, " && ")
+}
+
 // GetBuildContainer returns a Container definition based on a runtime
-func (l *Langruntimes) GetBuildContainer(runtime string, env []v1.EnvVar, installVolume v1.VolumeMount) (v1.Container, error) {
+func (l *Langruntimes) GetBuildContainer(runtime, depsChecksum string, env []v1.EnvVar, installVolume v1.VolumeMount) (v1.Container, error) {
 	runtimeInf, err := l.GetRuntimeInfo(runtime)
 	if err != nil {
 		return v1.Container{}, err
@@ -190,9 +197,16 @@ func (l *Langruntimes) GetBuildContainer(runtime string, env []v1.EnvVar, instal
 	}
 
 	var command string
+	// Validate deps checksum
+	shaFile := "/tmp/deps.sha256"
+	command = appendToCommand(command,
+		fmt.Sprintf("echo '%s  %s' > %s", depsChecksum, depsFile, shaFile),
+		fmt.Sprintf("sha256sum -c %s", shaFile))
+
 	switch {
 	case strings.Contains(runtime, "python"):
-		command = "pip install --prefix=" + installVolume.MountPath + " -r " + depsFile
+		command = appendToCommand(command,
+			"pip install --prefix="+installVolume.MountPath+" -r "+depsFile)
 	case strings.Contains(runtime, "nodejs"):
 		registry := "https://registry.npmjs.org"
 		scope := ""
@@ -204,15 +218,20 @@ func (l *Langruntimes) GetBuildContainer(runtime string, env []v1.EnvVar, instal
 				scope = v.Value + ":"
 			}
 		}
-		command = "npm config set " + scope + "registry " + registry +
-			" && npm install --production --prefix=" + installVolume.MountPath
+		command = appendToCommand(command,
+			"npm config set "+scope+"registry "+registry,
+			"npm install --production --prefix="+installVolume.MountPath)
 	case strings.Contains(runtime, "ruby"):
-		command = "bundle install --gemfile=" + depsFile + " --path=" + installVolume.MountPath
+		command = appendToCommand(command,
+			"bundle install --gemfile="+depsFile+" --path="+installVolume.MountPath)
 
 	case strings.Contains(runtime, "php"):
-		command = "composer install -d " + installVolume.MountPath
+		command = appendToCommand(command,
+			"composer install -d "+installVolume.MountPath)
 	case strings.Contains(runtime, "go"):
-		command = "cd $GOPATH/src/kubeless && dep ensure > /dev/termination-log 2>&1"
+		command = appendToCommand(command,
+			"cd $GOPATH/src/kubeless",
+			"dep ensure > /dev/termination-log 2>&1")
 	}
 
 	return v1.Container{

pkg/langruntime/langruntime_test.go

@@ -96,23 +96,23 @@ func TestGetBuildContainer(t *testing.T) {
 	lr.ReadConfigMap()
 
 	// It should throw an error if there is not an image available
-	_, err := lr.GetBuildContainer("notExists", []v1.EnvVar{}, v1.VolumeMount{})
+	_, err := lr.GetBuildContainer("notExists", "", []v1.EnvVar{}, v1.VolumeMount{})
 	if err == nil {
 		t.Error("Expected to throw an error")
 	}
 
 	// It should return the proper build image for python
 	env := []v1.EnvVar{}
 	vol1 := v1.VolumeMount{Name: "v1", MountPath: "/v1"}
-	c, err := lr.GetBuildContainer("python2.7", env, vol1)
+	c, err := lr.GetBuildContainer("python2.7", "abc123", env, vol1)
 	if err != nil {
 		t.Errorf("Unexpected error: %s", err)
 	}
 	expectedContainer := v1.Container{
 		Name:            "install",
 		Image:           "tuna/python-pillow:2.7.11-alpine",
 		Command:         []string{"sh", "-c"},
-		Args:            []string{"pip install --prefix=/v1 -r /v1/requirements.txt"},
+		Args:            []string{"echo 'abc123  /v1/requirements.txt' > /tmp/deps.sha256 && sha256sum -c /tmp/deps.sha256 && pip install --prefix=/v1 -r /v1/requirements.txt"},
 		VolumeMounts:    []v1.VolumeMount{vol1},
 		WorkingDir:      "/v1",
 		ImagePullPolicy: v1.PullIfNotPresent,
@@ -127,7 +127,7 @@ func TestGetBuildContainer(t *testing.T) {
 		{Name: "NPM_REGISTRY", Value: "http://reg.com"},
 		{Name: "NPM_SCOPE", Value: "myorg"},
 	}
-	c, err = lr.GetBuildContainer("nodejs6", nodeEnv, vol1)
+	c, err = lr.GetBuildContainer("nodejs6", "abc123", nodeEnv, vol1)
 	if err != nil {
 		t.Errorf("Unexpected error: %s", err)
 	}
@@ -139,7 +139,7 @@ func TestGetBuildContainer(t *testing.T) {
 	}
 
 	// It should return the proper build image for ruby
-	c, err = lr.GetBuildContainer("ruby2.4", env, vol1)
+	c, err = lr.GetBuildContainer("ruby2.4", "abc123", env, vol1)
 	if err != nil {
 		t.Errorf("Unexpected error: %s", err)
 	}

pkg/utils/k8sutil.go

@@ -18,7 +18,9 @@ package utils
 
 import (
 	"crypto/rand"
+	"crypto/sha256"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"net/url"
@@ -746,7 +748,6 @@ func EnsureFuncService(client kubernetes.Interface, funcObj *kubelessApi.Functio
 		newSvc.ObjectMeta.Labels = funcObj.ObjectMeta.Labels
 		newSvc.ObjectMeta.OwnerReferences = or
 		newSvc.Spec.Ports = svc.Spec.Ports
-		newSvc.Spec.Selector = svc.Spec.Selector
 		_, err = client.Core().Services(funcObj.ObjectMeta.Namespace).Update(newSvc)
 		if err != nil && k8sErrors.IsAlreadyExists(err) {
 			// The service may already exist and there is nothing to update
@@ -837,7 +838,13 @@ func populatePodSpec(funcObj *kubelessApi.Function, lr *langruntime.Langruntimes
 		if len(result.Containers) > 0 {
 			envVars = result.Containers[0].Env
 		}
-		depsInstallContainer, err := lr.GetBuildContainer(funcObj.Spec.Runtime, envVars, runtimeVolumeMount)
+		h := sha256.New()
+		_, err = h.Write([]byte(funcObj.Spec.Deps))
+		if err != nil {
+			return fmt.Errorf("Unable to obtain dependencies checksum: %v", err)
+		}
+		checksum := hex.EncodeToString(h.Sum(nil))
+		depsInstallContainer, err := lr.GetBuildContainer(funcObj.Spec.Runtime, checksum, envVars, runtimeVolumeMount)
 		if err != nil {
 			return err
 		}
@@ -1143,27 +1150,19 @@ func EnsureFuncDeployment(client kubernetes.Interface, funcObj *kubelessApi.Func
 		newDpm.ObjectMeta.Labels = funcObj.ObjectMeta.Labels
 		newDpm.ObjectMeta.Annotations = funcObj.Spec.Deployment.ObjectMeta.Annotations
 		newDpm.ObjectMeta.OwnerReferences = or
+		// We should maintain previous selector to avoid duplicated ReplicaSets
+		selector := newDpm.Spec.Selector
 		newDpm.Spec = dpm.Spec
-		_, err = client.ExtensionsV1beta1().Deployments(funcObj.ObjectMeta.Namespace).Update(newDpm)
+		newDpm.Spec.Selector = selector
+		data, err := json.Marshal(newDpm)
 		if err != nil {
 			return err
 		}
-
-		// kick existing function pods then it will be recreated
-		// with the new data mount from updated configmap.
-		// TODO: This is a workaround.  Do something better.
-		var pods *v1.PodList
-		pods, err = GetPodsByLabel(client, funcObj.ObjectMeta.Namespace, "function", funcObj.ObjectMeta.Name)
+		// Use `Patch` to do a rolling update
+		_, err = client.ExtensionsV1beta1().Deployments(funcObj.ObjectMeta.Namespace).Patch(newDpm.Name, types.MergePatchType, data)
 		if err != nil {
 			return err
 		}
-		for _, pod := range pods.Items {
-			err = client.Core().Pods(funcObj.ObjectMeta.Namespace).Delete(pod.Name, &metav1.DeleteOptions{})
-			if err != nil && !k8sErrors.IsNotFound(err) {
-				// non-fatal
-				logrus.Warnf("Unable to delete pod %s/%s, may be running stale version of function: %v", funcObj.ObjectMeta.Namespace, pod.Name, err)
-			}
-		}
 	}
 
 	return err

pkg/utils/k8sutil_test.go

@@ -242,6 +242,9 @@ func TestEnsureService(t *testing.T) {
 	if !reflect.DeepEqual(svc.ObjectMeta.Labels, newLabels) {
 		t.Error("Unable to update the service")
 	}
+	if reflect.DeepEqual(svc.Spec.Selector, newLabels) {
+		t.Error("It should not update the selector")
+	}
 }
 
 func TestEnsureImage(t *testing.T) {
@@ -531,16 +534,8 @@ func TestEnsureDeployment(t *testing.T) {
 	if err != nil {
 		t.Errorf("Unexpected error: %s", err)
 	}
-	dpm, err = clientset.ExtensionsV1beta1().Deployments(ns).Get(f1Name, metav1.GetOptions{})
-	if err != nil {
-		t.Errorf("Unexpected error: %s", err)
-	}
-	if getEnvValueFromList("FUNC_HANDLER", dpm.Spec.Template.Spec.Containers[0].Env) != "bar2" {
-		t.Error("Unable to update deployment")
-	}
-	if dpm.Annotations["new-key"] != "value" {
-		t.Errorf("Unable to update deployment %v", dpm.Annotations)
-	}
+	// Unable to ensure that the new deployment is patched since fake
+	// ignores PATCH actions: https://github.com/kubernetes/client-go/issues/364
 
 	// It should return an error if some dependencies are given but the runtime is not supported
 	f7 := getDefaultFunc("func7", ns)