From eccde4fe7440b5cf6d002896d988f46307aaed04 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 6 Jan 2017 17:01:21 -0800 Subject: [PATCH] Poolmgr: clean up orphaned resources on restart Label all poolmgr-created resources with an id that's unique to a running poolmgr instance. On poolmgr start up, clean up resources created by old instances. Resources that are idle are killed immediately; resources that could be running a user function are killed after the maximum function timeout. --- poolmgr/gp.go | 29 +++++++++++++++++++---------- poolmgr/gpm.go | 18 ++++++++++++++---- poolmgr/poolmgr.go | 7 +++++-- 3 files changed, 38 insertions(+), 16 deletions(-) diff --git a/poolmgr/gp.go b/poolmgr/gp.go index a61c9fe0..23a91051 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -39,6 +39,8 @@ import ( "github.com/fission/fission" ) +const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId" + type ( GenericPool struct { env *fission.Environment @@ -52,7 +54,9 @@ type ( useSvc bool // create service poolInstanceId string // small random string to uniquify pod names kubernetesClient *kubernetes.Clientset - requestChannel chan *choosePodRequest + instanceId string + + requestChannel chan *choosePodRequest } // serialize the choosing of pods so that choices don't conflict @@ -72,7 +76,8 @@ func MakeGenericPool( env *fission.Environment, initialReplicas int32, namespace string, - fsCache *functionServiceCache) (*GenericPool, error) { + fsCache *functionServiceCache, + instanceId string) (*GenericPool, error) { log.Printf("Creating pool for environment %v", env.Metadata) // TODO: in general we need to provide the user a way to configure pools. Initial @@ -88,6 +93,7 @@ func MakeGenericPool( idlePodReapTime: 3 * time.Minute, // TODO make this configurable fsCache: fsCache, poolInstanceId: uniuri.NewLen(8), + instanceId: instanceId, useSvc: false, } @@ -200,11 +206,12 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error) } } -func labelsForMetadata(metadata *fission.Metadata) map[string]string { +func (gp *GenericPool) labelsForMetadata(metadata *fission.Metadata) map[string]string { return map[string]string{ - "functionName": metadata.Name, - "functionUid": metadata.Uid, - "unmanaged": "true", // this allows us to easily find pods not managed by the deployment + "functionName": metadata.Name, + "functionUid": metadata.Uid, + "unmanaged": "true", // this allows us to easily find pods not managed by the deployment + POOLMGR_INSTANCEID_LABEL: gp.instanceId, } } @@ -212,7 +219,7 @@ func labelsForMetadata(metadata *fission.Metadata) map[string]string { // (via fetcher), and calls the function-run container to load it, resulting in a // specialized pod. func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error) { - newLabels := labelsForMetadata(metadata) + newLabels := gp.labelsForMetadata(metadata) log.Printf("[%v] Choosing pod from pool", metadata) pod, err := gp.choosePod(newLabels) @@ -287,6 +294,7 @@ func (gp *GenericPool) createPool() error { podLabels := map[string]string{ "pool": poolDeploymentName, + POOLMGR_INSTANCEID_LABEL: gp.instanceId, } sharedMountPath := "/userfunc" @@ -294,8 +302,9 @@ func (gp *GenericPool) createPool() error { ObjectMeta: v1.ObjectMeta{ Name: poolDeploymentName, Labels: map[string]string{ - "environmentName": gp.env.Metadata.Name, - "environmentUid": gp.env.Metadata.Uid, + "environmentName": gp.env.Metadata.Name, + "environmentUid": gp.env.Metadata.Uid, + POOLMGR_INSTANCEID_LABEL: gp.instanceId, }, }, Spec: v1beta1.DeploymentSpec{ @@ -411,7 +420,7 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { svcName += ("-" + m.Uid) } - labels := labelsForMetadata(m) + labels := gp.labelsForMetadata(m) svc, err := gp.createSvc(svcName, labels) if err != nil { return nil, err diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 0faf59d3..1c8b9ac0 100644 --- a/poolmgr/gpm.go +++ b/poolmgr/gpm.go @@ -34,8 +34,8 @@ type ( controllerUrl string controllerClient *client.Client fsCache *functionServiceCache - - requestChannel chan *request + instanceId string + requestChannel chan *request } request struct { env *fission.Environment @@ -47,7 +47,13 @@ type ( } ) -func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.Clientset, namespace string, fsCache *functionServiceCache) *GenericPoolManager { +func MakeGenericPoolManager( + controllerUrl string, + kubernetesClient *kubernetes.Clientset, + namespace string, + fsCache *functionServiceCache, + instanceId string) *GenericPoolManager { + gpm := &GenericPoolManager{ pools: make(map[fission.Environment]*GenericPool), kubernetesClient: kubernetesClient, @@ -55,6 +61,7 @@ func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.C controllerUrl: controllerUrl, controllerClient: client.MakeClient(controllerUrl), fsCache: fsCache, + instanceId: instanceId, requestChannel: make(chan *request), } go gpm.service() @@ -70,7 +77,10 @@ func (gpm *GenericPoolManager) service() { var err error pool, ok := gpm.pools[*req.env] if !ok { - pool, err = MakeGenericPool(gpm.controllerUrl, gpm.kubernetesClient, req.env, 3, gpm.namespace, gpm.fsCache) + pool, err = MakeGenericPool( + gpm.controllerUrl, gpm.kubernetesClient, req.env, + 3, // TODO configurable/autoscalable + gpm.namespace, gpm.fsCache, gpm.instanceId) if err != nil { req.responseChannel <- &response{error: err} continue diff --git a/poolmgr/poolmgr.go b/poolmgr/poolmgr.go index e5cc38a5..896cac96 100644 --- a/poolmgr/poolmgr.go +++ b/poolmgr/poolmgr.go @@ -20,6 +20,7 @@ import ( "log" "strings" + "github.com/dchest/uniuri" "k8s.io/client-go/1.5/kubernetes" "k8s.io/client-go/1.5/rest" @@ -56,11 +57,13 @@ func StartPoolmgr(controllerUrl string, namespace string, port int) error { return err } + instanceId := uniuri.NewLen(8) + cleanupOldPoolmgrResources(kubernetesClient, namespace, instanceId) + fsCache := MakeFunctionServiceCache() + gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace, fsCache, instanceId) - gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace, fsCache) api := MakeAPI(gpm, controllerClient, fsCache) - go api.Serve(port) return nil