add is ready to API.

OpenApi.yml

@@ -411,6 +411,9 @@ components:
         running:
           type: boolean
           example: true
+        is_ready:
+          type: boolean
+          example: true
         status:
           type: string
           example: Up 2 seconds

internal/room/containers.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 	"time"
 
 	dockerTypes "github.com/docker/docker/api/types"
@@ -19,14 +20,17 @@ func (manager *RoomManagerCtx) containerToEntry(container dockerTypes.Container)
 		return nil, err
 	}
 
+	roomId := container.ID[:12]
+
 	entry := &types.RoomEntry{
-		ID:             container.ID[:12],
+		ID:             roomId,
 		URL:            labels.URL,
 		Name:           labels.Name,
 		NekoImage:      labels.NekoImage,
 		IsOutdated:     labels.NekoImage != container.Image,
 		MaxConnections: labels.Epr.Max - labels.Epr.Min + 1,
 		Running:        container.State == "running",
+		IsReady:        manager.events.IsRoomReady(roomId) || strings.Contains(container.Status, "healthy"),
 		Status:         container.Status,
 		Created:        time.Unix(container.Created, 0),
 		Labels:         labels.UserDefined,

internal/room/events.go

@@ -3,109 +3,254 @@ package room
 import (
 	"context"
 	"fmt"
+	"io"
+	"strings"
+	"sync"
 
+	"github.com/m1k1o/neko-rooms/internal/config"
 	"github.com/m1k1o/neko-rooms/internal/types"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 
 	dockerTypes "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/filters"
+	dockerClient "github.com/docker/docker/client"
 )
 
-func (manager *RoomManagerCtx) EventsLoopStart() {
-	ctx, cancel := context.WithCancel(context.Background())
-	manager.eventsLoopCtx = ctx
-	manager.eventsLoopCancel = cancel
+type events struct {
+	wg sync.WaitGroup
 
-	msgs, errs := manager.client.Events(ctx, dockerTypes.EventsOptions{
+	logger zerolog.Logger
+	config *config.Room
+	client *dockerClient.Client
+
+	roomsReadyCh chan string
+	roomsReadyMu sync.Mutex
+	roomsReady   map[string]struct{}
+
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	listeners   []chan types.RoomEvent
+	listenersMu sync.Mutex
+}
+
+func newEvents(config *config.Room, client *dockerClient.Client) *events {
+	return &events{
+		logger: log.With().Str("module", "events").Logger(),
+		config: config,
+		client: client,
+
+		roomsReadyCh: make(chan string),
+		roomsReady:   make(map[string]struct{}),
+	}
+}
+
+func (e *events) Start() {
+	ctx, cancel := context.WithCancel(context.Background())
+	e.ctx = ctx
+	e.cancel = cancel
+
+	msgs, errs := e.client.Events(ctx, dockerTypes.EventsOptions{
 		Filters: filters.NewArgs(
 			filters.Arg("type", "container"),
-			filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", manager.config.InstanceName)),
+			filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", e.config.InstanceName)),
 			filters.Arg("event", "create"),
 			filters.Arg("event", "start"),
-			//filters.Arg("event", "health_status"),
+			filters.Arg("event", "health_status"),
 			filters.Arg("event", "stop"),
 			filters.Arg("event", "destroy"),
 		),
 	})
 
-	manager.eventsWg.Add(1)
+	e.wg.Add(1)
 	go func() {
-		defer manager.eventsWg.Done()
+		defer e.wg.Done()
 
 		for {
 			select {
 			case err, ok := <-errs:
 				if !ok {
-					manager.logger.Fatal().Msg("docker event error channel closed")
+					e.logger.Error().Msg("docker event error channel closed")
 					return
 				}
 
-				manager.logger.Err(err).Msg("got docker event error")
+				e.logger.Err(err).Msg("got docker event error")
 				return
+			case roomId := <-e.roomsReadyCh:
+				e.logger.Info().Str("id", roomId).Msg("room ready")
+
+				// ignore if room was already ready
+				if !e.setRoomReady(roomId) {
+					continue
+				}
+
+				e.broadcast(types.RoomEvent{
+					ID:     roomId,
+					Action: "ready",
+				})
 			case msg, ok := <-msgs:
 				if !ok {
-					manager.logger.Fatal().Msg("docker event message channel closed")
+					e.logger.Error().Msg("docker event message channel closed")
 					return
 				}
 
-				e := types.RoomEvent{
-					ID:     msg.Actor.ID[:12],
-					Action: msg.Action,
-					Labels: msg.Actor.Attributes,
-				}
+				roomId := msg.Actor.ID[:12]
 
-				manager.logger.Info().
-					Str("id", e.ID).
-					Str("action", e.Action).
+				e.logger.Info().
+					Str("id", roomId).
+					Str("action", msg.Action).
 					Msg("got docker event")
 
-				// broadcast event
-				manager.eventsMu.Lock()
-				for _, listener := range manager.eventsListeners {
-					listener <- e
+				action := ""
+				switch msg.Action {
+				case "create":
+					action = "created"
+				case "start":
+					action = "started"
+					e.waitForRoomReady(roomId)
+				case "health_status: healthy":
+					action = "ready"
+					// ignore if room was already ready
+					if !e.setRoomReady(roomId) {
+						continue
+					}
+				case "stop":
+					action = "stopped"
+					e.setRoomNotReady(roomId)
+				case "destroy":
+					action = "destroyed"
 				}
-				manager.eventsMu.Unlock()
+
+				e.broadcast(types.RoomEvent{
+					ID:     roomId,
+					Action: action,
+					Labels: msg.Actor.Attributes,
+				})
 			}
 		}
 	}()
 }
 
-func (manager *RoomManagerCtx) EventsLoopStop() error {
-	manager.eventsLoopCancel()
-	manager.eventsWg.Wait()
+func (e *events) Shutdown() error {
+	e.cancel()
+	close(e.roomsReadyCh)
+	e.wg.Wait()
 	return nil
 }
 
-func (manager *RoomManagerCtx) Events(ctx context.Context) (<-chan types.RoomEvent, <-chan error) {
+//
+// room ready
+//
+
+func (e *events) waitForRoomReady(roomId string) {
+	e.wg.Add(1)
+	go func() {
+		defer e.wg.Done()
+
+		// check if room is ready
+		exec, err := e.client.ContainerExecCreate(e.ctx, roomId, dockerTypes.ExecConfig{
+			AttachStdout: true,
+			Cmd: []string{
+				"/bin/bash", "-c",
+				"for ((a=1; a<=5; a++)); do (echo > /dev/tcp/localhost/8080) >/dev/null && echo -n OK && exit; sleep 1; done; exit",
+			},
+		})
+		if err != nil {
+			e.logger.Err(err).Msg("failed to create exec")
+			return
+		}
+
+		conn, err := e.client.ContainerExecAttach(e.ctx, exec.ID, dockerTypes.ExecStartCheck{})
+		if err != nil {
+			e.logger.Err(err).Msg("failed to attach exec")
+			return
+		}
+		defer conn.Close()
+
+		data, err := io.ReadAll(conn.Reader)
+		if err != nil {
+			e.logger.Err(err).Msg("failed to read exec")
+			return
+		}
+
+		if strings.HasSuffix(string(data), "OK") {
+			e.logger.Info().Str("id", roomId).Msg("room ready")
+			e.roomsReadyCh <- roomId
+			return
+		}
+
+		e.logger.Info().Str("id", roomId).Str("data", string(data)).Msg("room not ready")
+	}()
+}
+
+func (e *events) setRoomReady(roomId string) bool {
+	e.roomsReadyMu.Lock()
+	defer e.roomsReadyMu.Unlock()
+
+	_, ok := e.roomsReady[roomId]
+	e.roomsReady[roomId] = struct{}{}
+	return !ok
+}
+
+func (e *events) setRoomNotReady(roomId string) {
+	e.roomsReadyMu.Lock()
+	defer e.roomsReadyMu.Unlock()
+
+	delete(e.roomsReady, roomId)
+}
+
+func (e *events) IsRoomReady(roomId string) bool {
+	e.roomsReadyMu.Lock()
+	defer e.roomsReadyMu.Unlock()
+
+	_, ok := e.roomsReady[roomId]
+	return ok
+}
+
+//
+// events
+//
+
+func (e *events) broadcast(event types.RoomEvent) {
+	e.listenersMu.Lock()
+	for _, listener := range e.listeners {
+		listener <- event
+	}
+	e.listenersMu.Unlock()
+}
+
+func (e *events) Events(ctx context.Context) (<-chan types.RoomEvent, <-chan error) {
 	messages := make(chan types.RoomEvent)
 	errs := make(chan error, 1)
 
 	// add listener
-	manager.eventsMu.Lock()
-	manager.eventsListeners = append(manager.eventsListeners, messages)
-	manager.eventsMu.Unlock()
+	e.listenersMu.Lock()
+	e.listeners = append(e.listeners, messages)
+	e.listenersMu.Unlock()
 
-	manager.eventsWg.Add(1)
+	e.wg.Add(1)
 	go func() {
-		defer manager.eventsWg.Done()
+		defer e.wg.Done()
 		defer close(errs)
 
 		select {
 		case <-ctx.Done():
 			errs <- ctx.Err()
-		case <-manager.eventsLoopCtx.Done():
-			errs <- fmt.Errorf("room manager shutdown")
+		case <-e.ctx.Done():
+			errs <- fmt.Errorf("room events shutdown")
 			return
 		}
 
 		// remove listener
-		manager.eventsMu.Lock()
-		for i, listener := range manager.eventsListeners {
+		e.listenersMu.Lock()
+		for i, listener := range e.listeners {
 			if listener == messages {
-				manager.eventsListeners = append(manager.eventsListeners[:i], manager.eventsListeners[i+1:]...)
+				e.listeners = append(e.listeners[:i], e.listeners[i+1:]...)
 				break
 			}
 		}
-		manager.eventsMu.Unlock()
+		e.listenersMu.Unlock()
 	}()
 
 	return messages, errs

internal/room/manager.go

@@ -12,7 +12,6 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/docker/cli/opts"
@@ -49,6 +48,7 @@ func New(client *dockerClient.Client, config *config.Room) *RoomManagerCtx {
 		logger: logger,
 		config: config,
 		client: client,
+		events: newEvents(config, client),
 	}
 }
 
@@ -56,12 +56,7 @@ type RoomManagerCtx struct {
 	logger zerolog.Logger
 	config *config.Room
 	client *dockerClient.Client
-
-	eventsMu         sync.Mutex
-	eventsWg         sync.WaitGroup
-	eventsListeners  []chan types.RoomEvent
-	eventsLoopCtx    context.Context
-	eventsLoopCancel context.CancelFunc
+	events *events
 }
 
 func (manager *RoomManagerCtx) Config() types.RoomsConfig {
@@ -1015,3 +1010,17 @@ func (manager *RoomManagerCtx) Restart(ctx context.Context, id string) error {
 	// Restart the actual container
 	return manager.client.ContainerRestart(ctx, id, container.StopOptions{})
 }
+
+// events
+
+func (manager *RoomManagerCtx) EventsLoopStart() {
+	manager.events.Start()
+}
+
+func (manager *RoomManagerCtx) EventsLoopStop() error {
+	return manager.events.Shutdown()
+}
+
+func (manager *RoomManagerCtx) Events(ctx context.Context) (<-chan types.RoomEvent, <-chan error) {
+	return manager.events.Events(ctx)
+}

internal/types/room.go

@@ -23,6 +23,7 @@ type RoomEntry struct {
 	IsOutdated     bool              `json:"is_outdated"`
 	MaxConnections uint16            `json:"max_connections"` // 0 when using mux
 	Running        bool              `json:"running"`
+	IsReady        bool              `json:"is_ready"`
 	Status         string            `json:"status"`
 	Created        time.Time         `json:"created"`
 	Labels         map[string]string `json:"labels,omitempty"`