From 67c38f2f29a161d5fa394c4ade1e18b7c9503575 Mon Sep 17 00:00:00 2001
From: Soam Vasani <soamvasani@platform9.com>
Date: Mon, 18 Sep 2017 15:57:06 -0700
Subject: [PATCH] Function service cache partial support for multiple
 specialization (#332)

This adds partial support for multiple specialization to the
function service cache. It allows Add() to succeed if the pod and
address maps already contain entries.

But it doesn't deal with DeletePod and TouchByAddress. That's ok for
now, because the workflow engine never deletes the environment pod
(that pod is the workflow engine).

However, if/when we want full support for multiple specialization,
that will require updating both DeletePod and TouchByAddress.
---
 cache/cache.go                  |  3 +--
 poolmgr/functionServiceCache.go | 14 ++++++++++++++
 poolmgr/gp.go                   | 19 ++++++++++++-------
 3 files changed, 27 insertions(+), 9 deletions(-)

diff --git a/cache/cache.go b/cache/cache.go
index 14ac138c..845c4136 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -17,7 +17,6 @@ limitations under the License.
 package cache
 
 import (
-	"errors"
 	"fmt"
 	"time"
 
@@ -114,7 +113,7 @@ func (c *Cache) service() {
 				val := c.cache[req.key]
 				val.atime = time.Now()
 				resp.existingValue = val.value
-				resp.error = errors.New("value already exists")
+				resp.error = fission.MakeError(fission.ErrorNameExists, "key already exists")
 			} else {
 				c.cache[req.key] = &Value{
 					value: req.value,
diff --git a/poolmgr/functionServiceCache.go b/poolmgr/functionServiceCache.go
index 25225298..a1fb5ea2 100644
--- a/poolmgr/functionServiceCache.go
+++ b/poolmgr/functionServiceCache.go
@@ -22,6 +22,7 @@ import (
 
 	"k8s.io/client-go/1.5/pkg/api"
 
+	"github.com/fission/fission"
 	"github.com/fission/fission/cache"
 	"github.com/fission/fission/tpr"
 )
@@ -134,6 +135,7 @@ func (fsc *functionServiceCache) GetByFunction(m *api.ObjectMeta) (*funcSvc, err
 	return &fsvcCopy, nil
 }
 
+// TODO: error should be second return
 func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) {
 	err, existing := fsc.byFunction.Set(tpr.CacheKey(fsvc.function), &fsvc)
 	if err != nil {
@@ -152,13 +154,25 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) {
 	fsvc.ctime = now
 	fsvc.atime = now
 
+	// Add to byAddress and byPod caches. Ignore NameExists errors
+	// because of multiple-specialization. See issue #331.
 	err, _ = fsc.byAddress.Set(fsvc.address, *fsvc.function)
 	if err != nil {
+		if fe, ok := err.(fission.Error); ok {
+			if fe.Code == fission.ErrorNameExists {
+				err = nil
+			}
+		}
 		log.Printf("error caching fsvc: %v", err)
 		return err, nil
 	}
 	err, _ = fsc.byPod.Set(fsvc.podName, *fsvc.function)
 	if err != nil {
+		if fe, ok := err.(fission.Error); ok {
+			if fe.Code == fission.ErrorNameExists {
+				err = nil
+			}
+		}
 		log.Printf("error caching fsvc: %v", err)
 		return err, nil
 	}
diff --git a/poolmgr/gp.go b/poolmgr/gp.go
index 117e8879..888d2b9a 100644
--- a/poolmgr/gp.go
+++ b/poolmgr/gp.go
@@ -511,13 +511,18 @@ func (gp *GenericPool) GetFuncSvc(m *api.ObjectMeta) (*funcSvc, error) {
 
 	err, existingFsvc := gp.fsCache.Add(*fsvc)
 	if err != nil {
-		// Some other thread beat us to it -- return the other thread's fsvc and clean up
-		// our own.
-		log.Printf("func svc already exists: %v", existingFsvc.podName)
-		go func() {
-			gp.kubernetesClient.Core().Pods(gp.namespace).Delete(fsvc.podName, nil)
-		}()
-		return existingFsvc, nil
+		if fe, ok := err.(fission.Error); ok {
+			if fe.Code == fission.ErrorNameExists {
+				// Some other thread beat us to it -- return the other thread's fsvc and clean up
+				// our own.
+				log.Printf("func svc already exists: %v", existingFsvc.podName)
+				go func() {
+					gp.kubernetesClient.Core().Pods(gp.namespace).Delete(fsvc.podName, nil)
+				}()
+				return existingFsvc, nil
+			}
+		}
+		return nil, err
 	}
 	return fsvc, nil
 }