diff --git a/OpenApi.yml b/OpenApi.yml index f7a426c..704061f 100644 --- a/OpenApi.yml +++ b/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 diff --git a/internal/room/containers.go b/internal/room/containers.go index 882a924..05bfcd1 100644 --- a/internal/room/containers.go +++ b/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, diff --git a/internal/room/events.go b/internal/room/events.go index e01bd08..772965d 100644 --- a/internal/room/events.go +++ b/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 diff --git a/internal/room/manager.go b/internal/room/manager.go index 34d6569..9be2445 100644 --- a/internal/room/manager.go +++ b/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) +} diff --git a/internal/types/room.go b/internal/types/room.go index 2365b9f..89032eb 100644 --- a/internal/types/room.go +++ b/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"`