Delete zombie sessions and instances (#71)

services/check_swarm_status_task.go

@@ -9,7 +9,7 @@ import (
 type checkSwarmStatusTask struct {
 }
 
-func (c checkSwarmStatusTask) Run(i *Instance) {
+func (c checkSwarmStatusTask) Run(i *Instance) error {
 	if info, err := GetDaemonInfo(i); err == nil {
 		if info.Swarm.LocalNodeState != swarm.LocalNodeStateInactive && info.Swarm.LocalNodeState != swarm.LocalNodeStateLocked {
 			i.IsManager = &info.Swarm.ControlAvailable
@@ -18,6 +18,7 @@ func (c checkSwarmStatusTask) Run(i *Instance) {
 		}
 	} else {
 		log.Println(err)
+		return err
 	}
-
+	return nil
 }

services/check_swarm_used_ports.go

@@ -5,11 +5,13 @@ import "log"
 type checkSwarmUsedPortsTask struct {
 }
 
-func (c checkSwarmUsedPortsTask) Run(i *Instance) {
+func (c checkSwarmUsedPortsTask) Run(i *Instance) error {
 	if i.IsManager != nil && *i.IsManager {
 		// This is a swarm manager instance, then check for ports
 		if err := SetInstanceSwarmPorts(i); err != nil {
 			log.Println(err)
+			return err
 		}
 	}
+	return nil
 }

services/check_used_ports_task.go

@@ -1,12 +1,18 @@
 package services
 
+import "log"
+
 type checkUsedPortsTask struct {
 }
 
-func (c checkUsedPortsTask) Run(i *Instance) {
+func (c checkUsedPortsTask) Run(i *Instance) error {
 	if ports, err := GetUsedPorts(i); err == nil {
 		for _, p := range ports {
 			i.setUsedPort(p)
 		}
+	} else {
+		log.Println(err)
+		return err
 	}
+	return nil
 }

services/collect_stats_task.go

@@ -19,18 +19,18 @@ type collectStatsTask struct {
 	previousSystem uint64
 }
 
-func (c collectStatsTask) Run(i *Instance) {
+func (c collectStatsTask) Run(i *Instance) error {
 	reader, err := GetContainerStats(i.Name)
 	if err != nil {
 		log.Println("Error while trying to collect instance stats", err)
-		return
+		return err
 	}
 	dec := json.NewDecoder(reader)
 	var v *types.StatsJSON
 	e := dec.Decode(&v)
 	if e != nil {
 		log.Println("Error while trying to collect instance stats", e)
-		return
+		return err
 	}
 	// Memory
 	if v.MemoryStats.Limit != 0 {
@@ -46,6 +46,8 @@ func (c collectStatsTask) Run(i *Instance) {
 	c.previousSystem = v.PreCPUStats.SystemUsage
 	c.cpuPercent = calculateCPUPercentUnix(c.previousCPU, c.previousSystem, v)
 	i.Cpu = fmt.Sprintf("%.2f%%", c.cpuPercent)
+
+	return nil
 }
 
 func calculateCPUPercentUnix(previousCPU, previousSystem uint64, v *types.StatsJSON) float64 {

services/instance.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"log"
 	"os"
+	"strings"
 	"sync"
 
 	"golang.org/x/text/encoding"
@@ -22,7 +23,6 @@ type Instance struct {
 	IP           string                  `json:"ip"`
 	conn         *types.HijackedResponse `json:"-"`
 	ctx          context.Context         `json:"-"`
-	statsReader  io.ReadCloser           `json:"-"`
 	dockerClient *client.Client          `json:"-"`
 	IsManager    *bool                   `json:"is_manager"`
 	Mem          string                  `json:"mem"`
@@ -129,22 +129,25 @@ func (i *Instance) Attach() {
 	}
 }
 func GetInstance(session *Session, name string) *Instance {
-	//TODO: Use redis
 	return session.Instances[name]
 }
 func DeleteInstance(session *Session, instance *Instance) error {
-	// stop collecting stats
+	if instance.conn != nil {
-	if instance.statsReader != nil {
+		instance.conn.Close()
-		instance.statsReader.Close()
 	}
-
-	//TODO: Use redis
-	delete(session.Instances, instance.Name)
 	err := DeleteContainer(instance.Name)
+	if !strings.Contains(err.Error(), "No such container") {
+		log.Println(err)
+		return err
+	}
 
 	wsServer.BroadcastTo(session.Id, "delete instance", instance.Name)
 
+	delete(session.Instances, instance.Name)
+	if err := saveSessionsToDisk(); err != nil {
+		return err
+	}
 	setGauges()
 
-	return err
+	return nil
 }

services/session.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -112,7 +113,16 @@ func (s *Session) SchedulePeriodicTasks() {
 				go func() {
 					defer wg.Done()
 					for _, t := range periodicTasks {
-						t.Run(i)
+						err := t.Run(i)
+						if err != nil {
+							if strings.Contains(err.Error(), "No such container") {
+								log.Printf("Container for instance [%s] doesn't exist any more. Deleting from session.\n", i.IP)
+								DeleteInstance(i.session, i)
+							} else {
+								log.Println(err)
+							}
+							break
+						}
 					}
 				}()
 			}
@@ -161,24 +171,26 @@ func CloseSession(s *Session) error {
 	}
 	log.Printf("Starting clean up of session [%s]\n", s.Id)
 	for _, i := range s.Instances {
-		if i.conn != nil {
+		err := DeleteInstance(s, i)
-			i.conn.Close()
+		if err != nil {
-		}
-		if err := DeleteContainer(i.Name); err != nil {
 			log.Println(err)
 			return err
 		}
 	}
 	// Disconnect PWD daemon from the network
 	if err := DisconnectNetwork("pwd", s.Id); err != nil {
+		if !strings.Contains(err.Error(), "is not connected to the network") {
 			log.Println("ERROR NETWORKING")
 			return err
 		}
-	log.Printf("Connected pwd to network [%s]\n", s.Id)
+	}
+	log.Printf("Disconnected pwd from network [%s]\n", s.Id)
 	if err := DeleteNetwork(s.Id); err != nil {
+		if !strings.Contains(err.Error(), "not found") {
 			log.Println(err)
 			return err
 		}
+	}
 	delete(sessions, s.Id)
 
 	// We store sessions as soon as we delete one
@@ -296,15 +308,21 @@ func LoadSessionsFromDisk() error {
 
 			// Connect PWD daemon to the new network
 			if err := ConnectNetwork("pwd", s.Id); err != nil {
-				log.Println("ERROR NETWORKING")
+				if strings.Contains(err.Error(), "Could not attach to network") {
+					log.Printf("Network for session [%s] doesn't exist. Removing all instances and session.", s.Id)
+					CloseSession(s)
+				} else {
+					log.Println("ERROR NETWORKING", err)
 					return err
 				}
+			} else {
 				log.Printf("Connected pwd to network [%s]\n", s.Id)
 
 				// Schedule peridic tasks execution
 				s.SchedulePeriodicTasks()
 			}
 		}
+	}
 	file.Close()
 	setGauges()
 	return err

services/task.go

@@ -1,7 +1,7 @@
 package services
 
 type periodicTask interface {
-	Run(i *Instance)
+	Run(i *Instance) error
 }
 
 var periodicTasks []periodicTask