Merge pull request #69 from fission/poolmgr-gc

Poolmgr: ensure orphaned resources are cleaned up

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)
+	}
+}

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,
+		"functionName":           metadata.Name,
-		"functionUid":  metadata.Uid,
+		"functionUid":            metadata.Uid,
-		"unmanaged":    "true", // this allows us to easily find pods not managed by the deployment
+		"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,
+				"environmentName":        gp.env.Metadata.Name,
-				"environmentUid":  gp.env.Metadata.Uid,
+				"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

poolmgr/gpm.go

@@ -34,8 +34,8 @@ type (
 		controllerUrl    string
 		controllerClient *client.Client
 		fsCache          *functionServiceCache
-
+		instanceId       string
-		requestChannel chan *request
+		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

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