Merge pull request #96 from noxiouz/porto_wait

isolate(porto): add waitLoop to cleanup dead containers

Author: noxiouz

isolate/porto/box.go

@@ -57,6 +57,8 @@ type Box struct {
 	muContainers sync.Mutex
 	containers   map[string]*container
 	blobRepo     BlobRepository
+
+	onClose context.CancelFunc
 }
 
 // NewBox creates new Box
@@ -111,19 +113,95 @@ func NewBox(ctx context.Context, cfg isolate.BoxConfig) (isolate.Box, error) {
 		},
 	}
 
+	ctx, onClose := context.WithCancel(ctx)
 	box := &Box{
 		config:     config,
 		instanceID: uuid.New(),
 		transport:  tr,
 		spawnSM:    semaphore.New(config.SpawnConcurrency),
 		containers: make(map[string]*container),
+		onClose:    onClose,
 
 		blobRepo: blobRepo,
 	}
 
+	go box.waitLoop(ctx)
+
 	return box, nil
 }
 
+func (b *Box) waitLoop(ctx context.Context) {
+	apexctx.GetLogger(ctx).Info("start waitLoop")
+	var (
+		portoConn porto.API
+		err       error
+	)
+
+	var waitTimeout = 30 * time.Second
+
+	closed := func(portoConn porto.API) bool {
+		select {
+		case <-ctx.Done():
+			if portoConn != nil {
+				portoConn.Close()
+			}
+			return true
+		default:
+			return false
+		}
+	}
+
+LOOP:
+	for {
+		apexctx.GetLogger(ctx).Info("next iteration of waitLoop")
+		if closed(portoConn) {
+			return
+		}
+		// Connect to Porto if we have not connected yet.
+		// In case of error: wait either a fixed timeout or closing of Box
+		if portoConn == nil {
+			apexctx.GetLogger(ctx).Info("waitLoop: connect to Portod")
+			portoConn, err = porto.Connect()
+			if err != nil {
+				apexctx.GetLogger(ctx).WithError(err).Warn("unable to connect to Portod")
+				select {
+				case <-time.After(time.Second):
+					continue LOOP
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+
+		// * means all containers
+		// if no containers dead for waitTimeout, name will be an empty string
+		containerName, err := portoConn.Wait([]string{"*"}, 30*waitTimeout)
+		if err != nil {
+			portoConn.Close()
+			portoConn = nil
+			continue LOOP
+		}
+
+		if containerName != "" {
+			apexctx.GetLogger(ctx).Infof("Wait reports %s to be dead", containerName)
+			b.muContainers.Lock()
+			container, ok := b.containers[containerName]
+			if ok {
+				delete(b.containers, containerName)
+			}
+			rest := len(b.containers)
+			b.muContainers.Unlock()
+			if ok {
+				if err = container.Kill(); err != nil {
+					apexctx.GetLogger(ctx).WithError(err).Errorf("Killing %s error", containerName)
+				}
+			}
+
+			apexctx.GetLogger(ctx).Infof("%d containers are being tracked now", rest)
+		}
+	}
+}
+
 func (b *Box) appLayerName(appname string) string {
 	return b.instanceID + appname
 }
@@ -271,5 +349,6 @@ func (b *Box) Spawn(ctx context.Context, config isolate.SpawnConfig, output io.W
 // Close releases all resources such as idle connections from http.Transport
 func (b *Box) Close() error {
 	b.transport.CloseIdleConnections()
+	b.onClose()
 	return nil
 }

isolate/porto/container.go

@@ -158,6 +158,7 @@ func (c *container) start(portoConn porto.API, output io.Writer) error {
 
 func (c *container) Kill() (err error) {
 	defer apexctx.GetLogger(c.ctx).WithField("id", c.containerID).Trace("Kill container").Stop(&err)
+	containersKilledCounter.Inc(1)
 	portoConn, err := porto.Connect()
 	if err != nil {
 		return err

isolate/porto/metrics.go

@@ -12,6 +12,7 @@ var (
 	containersCreatedCounter = metrics.NewCounter()
 	// containers that crashed during spawning
 	containersErroredCounter = metrics.NewCounter()
+	containersKilledCounter  = metrics.NewCounter()
 
 	totalSpawnTimer = metrics.NewTimer()
 )
@@ -21,5 +22,6 @@ func init() {
 	registry.Register("spawning_queue_size", spawningQueueSize)
 	registry.Register("containers_created", containersCreatedCounter)
 	registry.Register("containers_errored", containersErroredCounter)
+	registry.Register("containers_killed", containersKilledCounter)
 	registry.Register("total_spawn_timer", totalSpawnTimer)
 }