Reap idle pods

Idle pod reaper wakes up once a minute, looks in the functionServiceCache for pods that
haven't been accessed for more than idlePodReapTime, and deletes all such pods.

(This doesn't yet delete pods that may be leaked from previously terminated poolmgrs; it
only works with pods created by the current poolmgr instance.)

poolmgr/api.go

@@ -127,8 +127,6 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) {
 		return fsvc.address, nil
 	}
 
-	api.fsCache.Log()
-
 	// None exists, so create a new funcSvc:
 	log.Printf("[%v] No cached function service found, creating one", m.Name)
 

poolmgr/functionServiceCache.go

@@ -30,6 +30,7 @@ const (
 	TOUCH fscRequestType = iota
 	LISTOLD
 	LOG
+	DELETE_BY_POD
 )
 
 type (
@@ -43,11 +44,13 @@ type (
 	fscRequest struct {
 		requestType     fscRequestType
 		address         string
+		podName         string
 		age             time.Duration
 		responseChannel chan *fscResponse
 	}
 	fscResponse struct {
 		podNames []string
+		deleted  bool
 		error
 	}
 )
@@ -75,11 +78,17 @@ func (fsc *functionServiceCache) service() {
 			// get svcs idle for > req.age
 			byPodCopy := fsc.byPod.Copy()
 			pods := make([]string, 0)
-			for podNameI, fsvcI := range byPodCopy {
+			for podNameI, mI := range byPodCopy {
-				fsvc := fsvcI.(*funcSvc)
+				m := mI.(fission.Metadata)
-				if time.Now().Sub(fsvc.atime) > req.age {
+				fsvcI, err := fsc.byFunction.Get(m)
-					podName := podNameI.(string)
+				if err != nil {
-					pods = append(pods, podName)
+					resp.error = err
+				} else {
+					fsvc := fsvcI.(*funcSvc)
+					if time.Now().Sub(fsvc.atime) > req.age {
+						podName := podNameI.(string)
+						pods = append(pods, podName)
+					}
 				}
 			}
 			resp.podNames = pods
@@ -91,6 +100,8 @@ func (fsc *functionServiceCache) service() {
 				fsvc := fsvcI.(*funcSvc)
 				log.Printf("%v:%v\t%v", m.Name, m.Uid, fsvc.podName)
 			}
+		case DELETE_BY_POD:
+			resp.deleted, resp.error = fsc._deleteByPod(req.podName, req.age)
 		}
 		req.responseChannel <- resp
 	}
@@ -166,22 +177,51 @@ func (fsc *functionServiceCache) _touchByAddress(address string) error {
 	return nil
 }
 
-func (fsc *functionServiceCache) DeleteByPod(podName string) error {
+func (fsc *functionServiceCache) DeleteByPod(podName string, minAge time.Duration) (bool, error) {
+	responseChannel := make(chan *fscResponse)
+	fsc.requestChannel <- &fscRequest{
+		requestType:     DELETE_BY_POD,
+		podName:         podName,
+		age:             minAge,
+		responseChannel: responseChannel,
+	}
+	resp := <-responseChannel
+	return resp.deleted, resp.error
+}
+
+// _deleteByPod deletes the entry keyed by podName, 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)
 	if err != nil {
-		return err
+		return false, err
 	}
 	m := mI.(fission.Metadata)
 	fsvcI, err := fsc.byFunction.Get(m)
 	if err != nil {
-		return err
+		return false, err
 	}
 	fsvc := fsvcI.(*funcSvc)
 
+	if time.Now().Sub(fsvc.atime) < minAge {
+		return false, nil
+	}
+
 	fsc.byFunction.Delete(m)
 	fsc.byAddress.Delete(fsvc.address)
 	fsc.byPod.Delete(podName)
-	return nil
+	return true, nil
+}
+
+func (fsc *functionServiceCache) ListOld(age time.Duration) ([]string, error) {
+	responseChannel := make(chan *fscResponse)
+	fsc.requestChannel <- &fscRequest{
+		requestType:     LISTOLD,
+		age:             age,
+		responseChannel: responseChannel,
+	}
+	resp := <-responseChannel
+	return resp.podNames, resp.error
 }
 
 func (fsc *functionServiceCache) Log() {

poolmgr/gp.go

@@ -435,16 +435,47 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 		// our own.  TODO: this is grossly inefficient, improve it with some sort of state
 		// machine
 		log.Printf("func svc already exists: %v", existingFsvc.podName)
-		go gp.CleanupFunctionService(fsvc)
+		go gp.CleanupFunctionService(fsvc.podName)
 		return existingFsvc, nil
 	}
 	return fsvc, nil
 }
 
-func (gp *GenericPool) CleanupFunctionService(fsvc *funcSvc) {
+func (gp *GenericPool) CleanupFunctionService(podName string) error {
+	// remove ourselves from fsCache (only if we're still old)
+	deleted, err := gp.fsCache.DeleteByPod(podName, gp.idlePodReapTime)
+	if err != nil {
+		return err
+	}
+
+	if !deleted {
+		log.Printf("Not deleting %v, in use", podName)
+		return nil
+	}
+
 	// delete pod
-	// remove ourselves from fsCache
+	err = gp.kubernetesClient.Core().Pods(gp.namespace).Delete(podName, nil)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (gp *GenericPool) idlePodReaper() {
+	for {
+		time.Sleep(time.Minute)
+		podNames, err := gp.fsCache.ListOld(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)
+			if err != nil {
+				log.Printf("Error deleting idle pod '%v': %v", podName, err)
+			}
+		}
+	}
 }