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

@@ -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,

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
 	}

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
 }