Add cleanup.go

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