Merge pull request #298 from SUSE/fix-stale-pods

Fix stale pods

src/jetstream/plugins/kubernetes/get_release.go

@@ -3,11 +3,8 @@ package kubernetes
 import (
 	"encoding/json"
 	"fmt"
-
 	"time"
 
-	//"fmt"
-
 	"github.com/gorilla/websocket"
 	"github.com/labstack/echo"
 	log "github.com/sirupsen/logrus"
@@ -105,9 +102,7 @@ func (c *KubernetesSpecification) GetReleaseStatus(ec echo.Context) error {
 	// this back incrementally
 
 	// Parse the manifest
-	log.Info("Got release")
 	rel := helm.NewHelmRelease(res, endpointGUID, userID)
-	log.Info("Done")
 
 	graph := helm.NewHelmReleaseGraph(rel)
 
@@ -200,7 +195,6 @@ func readLoop(c *websocket.Conn, stopchan chan<- bool) {
 }
 
 func sendResource(ws *websocket.Conn, kind string, data interface{}) error {
-
 	var err error
 	var txt []byte
 	if txt, err = json.Marshal(data); err == nil {

src/jetstream/plugins/kubernetes/helm/release.go

@@ -5,9 +5,8 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"reflect"
-
-	//	"io"
 	"strings"
 
 	log "github.com/sirupsen/logrus"
@@ -105,6 +104,9 @@ func getResourceIdentifier(typeMeta metav1.TypeMeta, objectMeta metav1.ObjectMet
 func (r *HelmRelease) setResource(res KubeResource) {
 	r.Resources[res.getID()] = res
 }
+func (r *HelmRelease) deleteResource(res KubeResource) {
+	delete(r.Resources, res.getID())
+}
 
 // GetResources gets all fo the resources for the release
 func (r *HelmRelease) GetResources() []interface{} {
@@ -211,6 +213,8 @@ func (r *HelmRelease) UpdatePods(jetstream interfaces.PortalProxy) {
 		jobs = append(jobs, job)
 	}
 
+	pods := make(map[string]*KubeResource)
+
 	runner := NewKubeAPIJob(jetstream, jobs)
 	res := runner.Run()
 	for _, j := range res {
@@ -229,12 +233,22 @@ func (r *HelmRelease) UpdatePods(jetstream interfaces.PortalProxy) {
 				podCopy := &v1.Pod{}
 				*podCopy = pod
 				res.Resource = podCopy
+				pods[res.getID()] = &res
 
 				r.setResource(res)
 				r.processPodOwners(pod)
 			}
 		}
 	}
+
+	// Now remove all pods that have not just been retrieved
+	// These are stale pods
+	for _, res := range r.Resources {
+		_, exists := pods[res.getID()]
+		if res.Kind == "Pod" && !exists {
+			r.deleteResource(res)
+		}
+	}
 }
 
 // Pods can be owned by a ReplicaSet - these are not represented in the manifest, as they
@@ -288,17 +302,26 @@ func (r *HelmRelease) UpdateResources(jetstream interfaces.PortalProxy) {
 	res := runner.Run()
 	for _, j := range res {
 
-		// TODO: If the status was 404, then we should remove the resource
-
 		// Add a kube resource
 		res := KubeResource{
 			Kind:       j.Kind,
 			APIVersion: j.APIVersion,
 		}
 		res.Metadata.Name = j.Name
 
-		// TODO: This should carry over
-		res.Manifest = false
+		// TODO: If the status was 404, then we should remove the resource
+		if j.StatusCode == http.StatusNotFound {
+			log.Debugf("Resource has been deleted - removing: %s -> %s", j.Kind, j.Name)
+			r.deleteResource(res)
+		}
+
+		// Manifest should carry over - indicates if the resource was in the Helm manifest
+		// Pods are an example of a reosurce which is not in the manifest
+		if existing, ok := r.Resources[res.getID()]; ok {
+			res.Manifest = existing.Manifest
+		} else {
+			res.Manifest = false
+		}
 
 		decode := scheme.Codecs.UniversalDeserializer().Decode
 		obj, _, err := decode(j.Data, nil, nil)
@@ -326,7 +349,5 @@ func getRestURL(namespace, kind, apiVersion, name string) string {
 		}
 	}
 	restURL = fmt.Sprintf("/%s/%s/namespaces/%s/%ss/%s", base, apiVersion, namespace, strings.ToLower(kind), name)
-
-	//log.Errorf("%s %s %s -> %s", kind, apiVersion, name, restURL)
 	return restURL
 }