Merge pull request #75 from fission/cleanup-deleted-pools

Delete generic pools when environments are deleted

poolmgr/gp.go

@@ -44,19 +44,19 @@ const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId"
 type (
 	GenericPool struct {
 		env              *fission.Environment
-		replicas         int32               // num containers
+		replicas         int32               // num idle pods
 		deployment       *v1beta1.Deployment // kubernetes deployment
 		namespace        string              // namespace to keep our resources
 		podReadyTimeout  time.Duration       // timeout for generic pods to become ready
 		controllerUrl    string
 		idlePodReapTime  time.Duration         // pods unused for idlePodReapTime are deleted
 		fsCache          *functionServiceCache // cache funcSvc's by function, address and podname
-		useSvc           bool                  // create service
+		useSvc           bool                  // create k8s service for specialized pods
 		poolInstanceId   string                // small random string to uniquify pod names
 		kubernetesClient *kubernetes.Clientset
-		instanceId       string
+		instanceId       string // poolmgr instance id
-
+		labelsForPool    map[string]string
-		requestChannel chan *choosePodRequest
+		requestChannel   chan *choosePodRequest
 	}
 
 	// serialize the choosing of pods so that choices don't conflict
@@ -95,7 +95,14 @@ func MakeGenericPool(
 		poolInstanceId:   uniuri.NewLen(8),
 		instanceId:       instanceId,
 
-		useSvc: false,
+		useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start
+	}
+
+	// Labels for generic deployment/RS/pods.
+	gp.labelsForPool = map[string]string{
+		"environmentName":        gp.env.Metadata.Name,
+		"environmentUid":         gp.env.Metadata.Uid,
+		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
 	}
 
 	// create the pool
@@ -206,7 +213,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
 	}
 }
 
-func (gp *GenericPool) labelsForMetadata(metadata *fission.Metadata) map[string]string {
+func (gp *GenericPool) labelsForFunction(metadata *fission.Metadata) map[string]string {
 	return map[string]string{
 		"functionName":           metadata.Name,
 		"functionUid":            metadata.Uid,
@@ -296,29 +303,20 @@ func (gp *GenericPool) createPool() error {
 	poolDeploymentName := fmt.Sprintf("%v-%v-%v",
 		gp.env.Metadata.Name, gp.env.Metadata.Uid, strings.ToLower(gp.poolInstanceId))
 
-	podLabels := map[string]string{
-		"pool": poolDeploymentName,
-		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
-	}
-
 	sharedMountPath := "/userfunc"
 	deployment := &v1beta1.Deployment{
 		ObjectMeta: v1.ObjectMeta{
-			Name: poolDeploymentName,
+			Name:   poolDeploymentName,
-			Labels: map[string]string{
+			Labels: gp.labelsForPool,
-				"environmentName":        gp.env.Metadata.Name,
-				"environmentUid":         gp.env.Metadata.Uid,
-				POOLMGR_INSTANCEID_LABEL: gp.instanceId,
-			},
 		},
 		Spec: v1beta1.DeploymentSpec{
 			Replicas: &gp.replicas,
 			Selector: &v1beta1.LabelSelector{
-				MatchLabels: podLabels,
+				MatchLabels: gp.labelsForPool,
 			},
 			Template: v1.PodTemplateSpec{
 				ObjectMeta: v1.ObjectMeta{
-					Labels: podLabels,
+					Labels: gp.labelsForPool,
 				},
 				Spec: v1.PodSpec{
 					Volumes: []v1.Volume{
@@ -413,7 +411,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Ser
 func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 
 	log.Printf("[%v] Choosing pod from pool", m)
-	newLabels := gp.labelsForMetadata(m)
+	newLabels := gp.labelsForFunction(m)
 	pod, err := gp.choosePod(newLabels)
 	if err != nil {
 		return nil, err
@@ -433,7 +431,7 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 			svcName += ("-" + m.Uid)
 		}
 
-		labels := gp.labelsForMetadata(m)
+		labels := gp.labelsForFunction(m)
 		svc, err := gp.createSvc(svcName, labels)
 		if err != nil {
 			gp.scheduleDeletePod(pod.ObjectMeta.Name)
@@ -513,3 +511,43 @@ func (gp *GenericPool) idlePodReaper() {
 		}
 	}
 }
+
+// destroys the pool -- the deployment, replicaset and pods
+func (gp *GenericPool) destroy() error {
+	// Destroy deployment
+	err := gp.kubernetesClient.Extensions().Deployments(gp.namespace).Delete(gp.deployment.ObjectMeta.Name, nil)
+	if err != nil {
+		log.Printf("Error destroying deployment: %v", err)
+		return err
+	}
+
+	// Destroy ReplicaSet.  Pre-1.6 K8s versions don't do this
+	// automatically but post-1.6 K8s will, and may beat us to it,
+	// so don't error out if we fail.
+	rsList, err := gp.kubernetesClient.Extensions().ReplicaSets(gp.namespace).List(api.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector(),
+	})
+	if len(rsList.Items) >= 0 {
+		for _, rs := range rsList.Items {
+			err = gp.kubernetesClient.Extensions().ReplicaSets(gp.namespace).Delete(rs.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting replicaset, ignoring: %v", err)
+			}
+		}
+	}
+
+	// Destroy Pods.  See note above.
+	podList, err := gp.kubernetesClient.Core().Pods(gp.namespace).List(api.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector(),
+	})
+	if len(podList.Items) >= 0 {
+		for _, pod := range podList.Items {
+			err = gp.kubernetesClient.Core().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting pod, ignoring: %v", err)
+			}
+		}
+	}
+
+	return nil
+}

poolmgr/gpm.go

@@ -26,6 +26,13 @@ import (
 	"github.com/fission/fission/controller/client"
 )
 
+type requestType int
+
+const (
+	GET_POOL requestType = iota
+	CLEANUP_POOLS
+)
+
 type (
 	GenericPoolManager struct {
 		pools            map[fission.Environment]*GenericPool
@@ -38,7 +45,9 @@ type (
 		requestChannel   chan *request
 	}
 	request struct {
+		requestType
 		env             *fission.Environment
+		envList         []fission.Environment
 		responseChannel chan *response
 	}
 	response struct {
@@ -72,8 +81,9 @@ func MakeGenericPoolManager(
 
 func (gpm *GenericPoolManager) service() {
 	for {
-		select {
+		req := <-gpm.requestChannel
-		case req := <-gpm.requestChannel:
+		switch req.requestType {
+		case GET_POOL:
 			var err error
 			pool, ok := gpm.pools[*req.env]
 			if !ok {
@@ -88,17 +98,45 @@ func (gpm *GenericPoolManager) service() {
 				gpm.pools[*req.env] = pool
 			}
 			req.responseChannel <- &response{pool: pool}
+		case CLEANUP_POOLS:
+			uids := make(map[string]bool)
+			for _, env := range req.envList {
+				uids[env.Metadata.Uid] = true
+			}
+			for env, pool := range gpm.pools {
+				_, ok := uids[env.Metadata.Uid]
+				if !ok {
+					// Env no longer exists -- remove our cache
+					log.Printf("Destroying generic pool for environment [%v]", env)
+					delete(gpm.pools, env)
+
+					// and delete the pool asynchronously.
+					go pool.destroy()
+				}
+			}
+			// no response, caller doesn't wait
 		}
 	}
 }
 
 func (gpm *GenericPoolManager) GetPool(env *fission.Environment) (*GenericPool, error) {
 	c := make(chan *response)
-	gpm.requestChannel <- &request{env: env, responseChannel: c}
+	gpm.requestChannel <- &request{
+		requestType:     GET_POOL,
+		env:             env,
+		responseChannel: c,
+	}
 	resp := <-c
 	return resp.pool, resp.error
 }
 
+func (gpm *GenericPoolManager) CleanupPools(envs []fission.Environment) {
+	gpm.requestChannel <- &request{
+		requestType: CLEANUP_POOLS,
+		envList:     envs,
+	}
+}
+
 func (gpm *GenericPoolManager) eagerPoolCreator() {
 	failureCount := 0
 	maxFailures := 5
@@ -125,5 +163,8 @@ func (gpm *GenericPoolManager) eagerPoolCreator() {
 				log.Printf("eager-create pool failed: %v", err)
 			}
 		}
+
+		// Clean up pools whose env was deleted
+		gpm.CleanupPools(envs)
 	}
 }