From b0959eb0f3e79bab4bf60d9c9dd43072dc7f724a 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 14ac138cb..845c41369 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 252252981..a1fb5ea23 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 117e88797..888d2b9a6 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
 }