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

Changed podName to a generic objectReference in fscache (#391)

Browse Source 
Changed podName to a generic objectReference in function service cache implementation.
This commit is contained in:
Vishal
2017-11-15 13:57:19 -08:00
committed by Soam Vasani
parent 8a097abf9b
commit 7bb397dfee
3 changed files with 86 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
poolmgr/functionServiceCache.go
+47 -39
View File
@@ -21,6 +21,7 @@ import (
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/pkg/api"
"github.com/fission/fission"
"github.com/fission/fission/cache"
@@ -28,43 +29,50 @@ import (
)
type fscRequestType int
type backendType int
const (
TOUCH fscRequestType = iota
LISTOLD
LOG
DELETE_BY_POD
DELETE_BY_OBJECT
)
const (
POOLMGR backendType = iota
NEWDEPLOY
)
type (
funcSvc struct {
function *metav1.ObjectMeta // function this pod/service is for
environment *crd.Environment // function's environment
address string // Host:Port or IP:Port that the function's service can be reached at.
podName string // pod name (within the function namespace)
function *metav1.ObjectMeta // function this pod/service is for
environment *crd.Environment // function's environment
address string // Host:Port or IP:Port that the function's service can be reached at.
kubernetesObject api.ObjectReference // Kubernetes Object (within the function namespace)
backend backendType
ctime time.Time
atime time.Time
}
functionServiceCache struct {
byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc
byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta
byPod *cache.Cache // podname -> function : map[string]metav1.ObjectMeta
byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc
byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta
byKubeObject *cache.Cache // obj -> function : map[api.ObjectReference]metav1.ObjectMeta
requestChannel chan *fscRequest
}
fscRequest struct {
requestType fscRequestType
address string
podName string
age time.Duration
env *metav1.ObjectMeta // used for ListOld
responseChannel chan *fscResponse
requestType fscRequestType
address string
kubernetesObject api.ObjectReference
age time.Duration
env *metav1.ObjectMeta // used for ListOld
responseChannel chan *fscResponse
}
fscResponse struct {
podNames []string
deleted bool
objects []api.ObjectReference
deleted bool
error
}
)
@@ -73,7 +81,7 @@ func MakeFunctionServiceCache() *functionServiceCache {
fsc := &functionServiceCache{
byFunction: cache.MakeCache(0, 0),
byAddress: cache.MakeCache(0, 0),
byPod: cache.MakeCache(0, 0),
byKubeObject: cache.MakeCache(0, 0),
requestChannel: make(chan *fscRequest),
}
go fsc.service()
@@ -90,9 +98,9 @@ func (fsc *functionServiceCache) service() {
resp.error = fsc._touchByAddress(req.address)
case LISTOLD:
// get svcs idle for > req.age
byPodCopy := fsc.byPod.Copy()
pods := make([]string, 0)
for podNameI, mI := range byPodCopy {
byKubeObjectCopy := fsc.byKubeObject.Copy()
kubeObjects := make([]api.ObjectReference, 0)
for objI, mI := range byKubeObjectCopy {
m := mI.(metav1.ObjectMeta)
fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
if err != nil {
@@ -102,21 +110,21 @@ func (fsc *functionServiceCache) service() {
if fsvc.environment.Metadata.UID == req.env.UID &&
time.Now().Sub(fsvc.atime) > req.age {
podName := podNameI.(string)
pods = append(pods, podName)
obj := objI.(api.ObjectReference)
kubeObjects = append(kubeObjects, obj)
}
}
}
resp.podNames = pods
resp.objects = kubeObjects
case LOG:
funcCopy := fsc.byFunction.Copy()
log.Printf("Cache has %v entries", len(funcCopy))
for key, fsvcI := range funcCopy {
fsvc := fsvcI.(*funcSvc)
log.Printf("%v\t%v", key, fsvc.podName)
log.Printf("%v\t%v\t%v", key, fsvc.kubernetesObject.Kind, fsvc.kubernetesObject.Name)
}
case DELETE_BY_POD:
resp.deleted, resp.error = fsc._deleteByPod(req.podName, req.age)
case DELETE_BY_OBJECT:
resp.deleted, resp.error = fsc._deleteByKubeObject(req.kubernetesObject, req.age)
}
req.responseChannel <- resp
}
@@ -157,7 +165,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) {
fsvc.ctime = now
fsvc.atime = now
// Add to byAddress and byPod caches. Ignore NameExists errors
// Add to byAddress and byKubernetesObject caches. Ignore NameExists errors
// because of multiple-specialization. See issue #331.
err, _ = fsc.byAddress.Set(fsvc.address, *fsvc.function)
if err != nil {
@@ -169,7 +177,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) {
log.Printf("error caching fsvc: %v", err)
return err, nil
}
err, _ = fsc.byPod.Set(fsvc.podName, *fsvc.function)
err, _ = fsc.byKubeObject.Set(fsvc.kubernetesObject, *fsvc.function)
if err != nil {
if fe, ok := err.(fission.Error); ok {
if fe.Code == fission.ErrorNameExists {
@@ -208,22 +216,22 @@ func (fsc *functionServiceCache) _touchByAddress(address string) error {
return nil
}
func (fsc *functionServiceCache) DeleteByPod(podName string, minAge time.Duration) (bool, error) {
func (fsc *functionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) {
responseChannel := make(chan *fscResponse)
fsc.requestChannel <- &fscRequest{
requestType: DELETE_BY_POD,
podName: podName,
age: minAge,
responseChannel: responseChannel,
requestType: DELETE_BY_OBJECT,
kubernetesObject: obj,
age: minAge,
responseChannel: responseChannel,
}
resp := <-responseChannel
return resp.deleted, resp.error
}
// _deleteByPod deletes the entry keyed by podName, but only if it is
// _deleteByKubeObject deletes the entry keyed by Kubernetes Object, but only if it is
// at least minAge old.
func (fsc *functionServiceCache) _deleteByPod(podName string, minAge time.Duration) (bool, error) {
mI, err := fsc.byPod.Get(podName)
func (fsc *functionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) {
mI, err := fsc.byKubeObject.Get(obj)
if err != nil {
return false, err
}
@@ -240,11 +248,11 @@ func (fsc *functionServiceCache) _deleteByPod(podName string, minAge time.Durati
fsc.byFunction.Delete(crd.CacheKey(&m))
fsc.byAddress.Delete(fsvc.address)
fsc.byPod.Delete(podName)
fsc.byKubeObject.Delete(obj)
return true, nil
}
func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]string, error) {
func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) {
responseChannel := make(chan *fscResponse)
fsc.requestChannel <- &fscRequest{
requestType: LISTOLD,
@@ -253,7 +261,7 @@ func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Durati
responseChannel: responseChannel,
}
resp := <-responseChannel
return resp.podNames, resp.error
return resp.objects, resp.error
}
func (fsc *functionServiceCache) Log() {
poolmgr/functionServiceCache_test.go
+10 -4
View File
@@ -6,6 +6,7 @@ import (
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/pkg/api"
"github.com/fission/fission"
"github.com/fission/fission/crd"
@@ -39,9 +40,14 @@ func TestFunctionServiceCache(t *testing.T) {
},
},
address: "xxx",
podName: "yyy",
ctime: now,
atime: now,
kubernetesObject: api.ObjectReference{
Kind: "pod",
Name: "xxx",
APIVersion: "v1",
Namespace: "fission-function",
},
ctime: now,
atime: now,
}
err, _ := fsc.Add(*fsvc)
if err != nil {
@@ -67,7 +73,7 @@ func TestFunctionServiceCache(t *testing.T) {
log.Panicf("Failed to touch fsvc: %v", err)
}
deleted, err := fsc.DeleteByPod(fsvc.podName, 0)
deleted, err := fsc.DeleteByKubeObject(fsvc.kubernetesObject, 0)
if err != nil {
fsc.Log()
log.Panicf("Failed to delete fsvc: %v", err)
poolmgr/gp.go
+29 -18
View File
@@ -36,6 +36,7 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/pkg/api"
apiv1 "k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/extensions/v1beta1"
@@ -556,13 +557,23 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) {
svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP)
}
kubeObjRef := api.ObjectReference{
Kind: pod.TypeMeta.Kind,
Name: pod.ObjectMeta.Name,
APIVersion: pod.TypeMeta.APIVersion,
Namespace: pod.ObjectMeta.Namespace,
ResourceVersion: pod.ObjectMeta.ResourceVersion,
UID: pod.ObjectMeta.UID,
}
fsvc := &funcSvc{
function: m,
environment: gp.env,
address: svcHost,
podName: pod.ObjectMeta.Name,
ctime: time.Now(),
atime: time.Now(),
function: m,
environment: gp.env,
address: svcHost,
kubernetesObject: kubeObjRef,
backend: POOLMGR,
ctime: time.Now(),
atime: time.Now(),
}
err, existingFsvc := gp.fsCache.Add(*fsvc)
@@ -571,9 +582,9 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) {
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)
log.Printf("func svc already exists: %v", existingFsvc.kubernetesObject.Name)
go func() {
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(fsvc.podName, nil)
gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(fsvc.kubernetesObject.Name, nil)
}()
return existingFsvc, nil
}
@@ -583,19 +594,19 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) {
return fsvc, nil
}
func (gp *GenericPool) CleanupFunctionService(podName string) error {
func (gp *GenericPool) CleanupFunctionService(obj api.ObjectReference) error {
// remove ourselves from fsCache (only if we're still old)
deleted, err := gp.fsCache.DeleteByPod(podName, gp.idlePodReapTime)
deleted, err := gp.fsCache.DeleteByKubeObject(obj, gp.idlePodReapTime)
if err != nil {
return err
}
if !deleted {
log.Printf("Not deleting %v, in use", podName)
log.Printf("Not deleting %v, in use", obj.Name)
return nil
}
pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(podName, metav1.GetOptions{})
pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(obj.Name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -613,7 +624,7 @@ func (gp *GenericPool) CleanupFunctionService(podName string) error {
}
// delete pod
err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(podName, nil)
err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(obj.Name, nil)
if err != nil {
return err
}
@@ -624,16 +635,16 @@ func (gp *GenericPool) CleanupFunctionService(podName string) error {
func (gp *GenericPool) idlePodReaper() {
for {
time.Sleep(time.Minute)
podNames, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime)
objects, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime)
if err != nil {
log.Printf("Error reaping idle pods: %v", err)
continue
}
for _, podName := range podNames {
log.Printf("Reaping idle pod '%v'", podName)
err := gp.CleanupFunctionService(podName)
for _, obj := range objects {
log.Printf("Reaping idle pod '%v'", obj.Name)
err := gp.CleanupFunctionService(obj)
if err != nil {
log.Printf("Error deleting idle pod '%v': %v", podName, err)
log.Printf("Error deleting idle pod '%v': %v", obj.Name, err)
}
}
}