From eccde4fe7440b5cf6d002896d988f46307aaed04 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 6 Jan 2017 17:01:21 -0800 Subject: [PATCH 1/2] 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 From 37a00b611684d4a721f07341d2fb5e76f8843b96 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 6 Jan 2017 17:14:54 -0800 Subject: [PATCH 2/2] Add cleanup.go --- poolmgr/cleanup.go | 147 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 poolmgr/cleanup.go diff --git a/poolmgr/cleanup.go b/poolmgr/cleanup.go new file mode 100644 index 00000000..d715aac5 --- /dev/null +++ b/poolmgr/cleanup.go @@ -0,0 +1,147 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package poolmgr + +import ( + "log" + "time" + + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/pkg/api" +) + +// cleanupOldPoolmgrResources looks for resources created by an old +// poolmgr instance and cleans them up. +func cleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) { + go func() { + err := cleanup(client, namespace, instanceId) + if err != nil { + // TODO retry cleanup; logged and ignored for now + log.Printf("Failed to cleanup: %v", err) + } + }() +} + +func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) error { + // Deployments are used for idle pools and can be cleaned up + // immediately. (We should "adopt" these instead of creating + // a new pool.) + err := cleanupDeployments(client, namespace, instanceId) + if err != nil { + return err + } + // See K8s #33845 and related bugs: deleting a deployment + // through the API doesn't cause the associated ReplicaSet to + // be deleted. (Fixed recently, but we may be running a + // version before the fix.) + err = cleanupReplicaSets(client, namespace, instanceId) + if err != nil { + return err + } + + // Pods might still be running user functions, so we give them + // a few minutes before terminating them. This time is the + // maximum function runtime, plus the time a router might + // still route to an old instance, i.e. router cache expiry + // time. + time.Sleep(6 * time.Minute) + + err = cleanupPods(client, namespace, instanceId) + if err != nil { + return err + } + + err = cleanupServices(client, namespace, instanceId) + if err != nil { + return err + } + + return nil +} + +func cleanupDeployments(client *kubernetes.Clientset, namespace string, instanceId string) error { + deploymentList, err := client.Extensions().Deployments(namespace).List(api.ListOptions{}) + if err != nil { + return err + } + for _, dep := range deploymentList.Items { + id, ok := dep.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up deployment %v", dep.ObjectMeta.Name) + err := client.Extensions().Deployments(namespace).Delete(dep.ObjectMeta.Name, nil) + logErr("cleaning up deployment", err) + // ignore err + } + } + return nil +} + +func cleanupReplicaSets(client *kubernetes.Clientset, namespace string, instanceId string) error { + rsList, err := client.Extensions().ReplicaSets(namespace).List(api.ListOptions{}) + if err != nil { + return err + } + for _, rs := range rsList.Items { + id, ok := rs.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up replicaset %v", rs.ObjectMeta.Name) + err := client.Extensions().ReplicaSets(namespace).Delete(rs.ObjectMeta.Name, nil) + logErr("cleaning up replicaset", err) + } + } + return nil +} + +func cleanupPods(client *kubernetes.Clientset, namespace string, instanceId string) error { + podList, err := client.Core().Pods(namespace).List(api.ListOptions{}) + if err != nil { + return err + } + for _, pod := range podList.Items { + id, ok := pod.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name) + err := client.Core().Pods(namespace).Delete(pod.ObjectMeta.Name, nil) + logErr("cleaning up pod", err) + // ignore err + } + } + return nil +} + +func cleanupServices(client *kubernetes.Clientset, namespace string, instanceId string) error { + svcList, err := client.Core().Services(namespace).List(api.ListOptions{}) + if err != nil { + return err + } + for _, svc := range svcList.Items { + id, ok := svc.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL] + if ok && id != instanceId { + log.Printf("Cleaning up svc %v", svc.ObjectMeta.Name) + err := client.Core().Services(namespace).Delete(svc.ObjectMeta.Name, nil) + logErr("cleaning up svc", err) + // ignore err + } + } + return nil +} + +func logErr(msg string, err error) { + if err != nil { + log.Printf("Error %v: %v", msg, err) + } +}