rob/octoplex
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: startup check

Browse Source
This commit is contained in:
Rob Watson 2025-02-26 22:01:25 +01:00 committed by Rob Watson
parent 3d5e1a091d
commit 4e3a72893b
8 changed files with 124 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
app/app.go
View File

@ -45,6 +45,18 @@ func Run(
}
defer containerClient.Close()
if exists, err := containerClient.ContainerRunning(ctx, container.AllContainers()); err != nil {
return fmt.Errorf("check existing containers: %w", err)
} else if exists {
if <-ui.ShowStartupCheckModal() {
if err = containerClient.RemoveContainers(ctx, container.AllContainers()); err != nil {
return fmt.Errorf("remove existing containers: %w", err)
}
} else {
return nil
}
}
srv := mediaserver.StartActor(ctx, mediaserver.StartActorParams{
ContainerClient: containerClient,
Logger: logger.With("component", "mediaserver"),

49
container/container.go
View File

@ -344,7 +344,7 @@ func (a *Client) Close() error {
ctx, cancel := context.WithTimeout(context.Background(), stopTimeout)
defer cancel()
containerList, err := a.containersMatchingLabels(ctx, nil)
containerList, err := a.containersMatchingLabels(ctx, a.instanceLabels())
if err != nil {
return fmt.Errorf("container list: %w", err)
}
@ -382,8 +382,8 @@ func (a *Client) removeContainer(ctx context.Context, id string) error {
}
// ContainerRunning checks if a container with the given labels is running.
func (a *Client) ContainerRunning(ctx context.Context, labels map[string]string) (bool, error) {
containers, err := a.containersMatchingLabels(ctx, labels)
func (a *Client) ContainerRunning(ctx context.Context, labelOptions LabelOptions) (bool, error) {
containers, err := a.containersMatchingLabels(ctx, labelOptions())
if err != nil {
return false, fmt.Errorf("container list: %w", err)
}
@ -398,8 +398,8 @@ func (a *Client) ContainerRunning(ctx context.Context, labels map[string]string)
}
// RemoveContainers removes all containers with the given labels.
func (a *Client) RemoveContainers(ctx context.Context, labels map[string]string) error {
containers, err := a.containersMatchingLabels(ctx, labels)
func (a *Client) RemoveContainers(ctx context.Context, labelOptions LabelOptions) error {
containers, err := a.containersMatchingLabels(ctx, labelOptions())
if err != nil {
return fmt.Errorf("container list: %w", err)
}
@ -413,11 +413,27 @@ func (a *Client) RemoveContainers(ctx context.Context, labels map[string]string)
return nil
}
// LabelOptions is a function that returns a map of labels.
type LabelOptions func() map[string]string
// ContainersWithLabels returns a LabelOptions function that returns the labels for
// this app instance.
func (a *Client) ContainersWithLabels(extraLabels map[string]string) LabelOptions {
return func() map[string]string {
return a.instanceLabels(extraLabels)
}
}
// AllContainers returns a LabelOptions function that returns the labels for any
// app instance.
func AllContainers() LabelOptions {
return func() map[string]string {
return map[string]string{"app": domain.AppName}
}
}
func (a *Client) containersMatchingLabels(ctx context.Context, labels map[string]string) ([]container.Summary, error) {
filterArgs := filters.NewArgs(
filters.Arg("label", "app="+domain.AppName),
filters.Arg("label", "app-id="+a.id.String()),
)
filterArgs := filters.NewArgs()
for k, v := range labels {
filterArgs.Add("label", k+"="+v)
}
@ -427,6 +443,21 @@ func (a *Client) containersMatchingLabels(ctx context.Context, labels map[string
})
}
func (a *Client) instanceLabels(extraLabels ...map[string]string) map[string]string {
labels := map[string]string{
"app": domain.AppName,
"app-id": a.id.String(),
}
for _, el := range extraLabels {
for k, v := range el {
labels[k] = v
}
}
return labels
}
func shortID(id string) string {
if len(id) < 12 {
return id

16
container/integration_test.go
View File

@ -30,7 +30,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
client, err := container.NewClient(ctx, apiClient, logger)
require.NoError(t, err)
running, err := client.ContainerRunning(ctx, map[string]string{"component": component})
running, err := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component}))
require.NoError(t, err)
assert.False(t, running)
@ -51,7 +51,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
require.Eventually(
t,
func() bool {
running, err = client.ContainerRunning(ctx, map[string]string{"component": component})
running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component}))
return err == nil && running
},
5*time.Second,
@ -62,7 +62,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
client.Close()
require.NoError(t, <-errC)
running, err = client.ContainerRunning(ctx, map[string]string{"component": component})
running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component}))
require.NoError(t, err)
assert.False(t, running)
}
@ -117,7 +117,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
require.Eventually(
t,
func() bool {
running, _ := client.ContainerRunning(ctx, map[string]string{"group": "test1"})
running, _ := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"group": "test1"}))
return running
},
5*time.Second,
@ -128,7 +128,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
require.Eventually(
t,
func() bool {
running, _ := client.ContainerRunning(ctx, map[string]string{"group": "test2"})
running, _ := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"group": "test2"}))
return running
},
2*time.Second,
@ -137,7 +137,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
)
// remove group 1
err = client.RemoveContainers(ctx, map[string]string{"group": "test1"})
err = client.RemoveContainers(ctx, client.ContainersWithLabels(map[string]string{"group": "test1"}))
require.NoError(t, err)
// check group 1 is not running
@ -145,7 +145,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
t,
func() bool {
var running bool
running, err = client.ContainerRunning(ctx, map[string]string{"group": "test1"})
running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"group": "test1"}))
return err == nil && !running
},
2*time.Second,
@ -154,7 +154,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
)
// check group 2 is still running
running, err := client.ContainerRunning(ctx, map[string]string{"group": "test2"})
running, err := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"group": "test2"}))
require.NoError(t, err)
assert.True(t, running)

35
mediaserver/actor.go
View File

@ -139,7 +139,10 @@ func (s *Actor) State() domain.Source {
// Close closes the media server actor.
func (s *Actor) Close() error {
if err := s.containerClient.RemoveContainers(context.Background(), map[string]string{"component": componentName}); err != nil {
if err := s.containerClient.RemoveContainers(
context.Background(),
s.containerClient.ContainersWithLabels(map[string]string{"component": componentName}),
); err != nil {
return fmt.Errorf("remove containers: %w", err)
}
@ -169,6 +172,11 @@ func (s *Actor) actorLoop(containerStateC <-chan domain.Container, errC <-chan e
case containerState := <-containerStateC:
s.state.Container = containerState
if s.state.Container.State == "exited" {
fetchStateT.Stop()
s.handleContainerExit(nil)
}
sendState()
continue
@ -185,17 +193,7 @@ func (s *Actor) actorLoop(containerStateC <-chan domain.Container, errC <-chan e
}
fetchStateT.Stop()
if s.state.Container.ExitCode != nil {
s.state.ExitReason = fmt.Sprintf("Server process exited with code %d.", *s.state.Container.ExitCode)
} else {
s.state.ExitReason = "Server process exited unexpectedly."
}
if err != nil {
s.state.ExitReason += "\n\n" + err.Error()
}
s.state.Live = false
s.handleContainerExit(err)
sendState()
case <-fetchStateT.C:
@ -235,6 +233,19 @@ func (s *Actor) actorLoop(containerStateC <-chan domain.Container, errC <-chan e
}
}
func (s *Actor) handleContainerExit(err error) {
if s.state.Container.ExitCode != nil {
s.state.ExitReason = fmt.Sprintf("Server process exited with code %d.", *s.state.Container.ExitCode)
} else {
s.state.ExitReason = "Server process exited unexpectedly."
}
if err != nil {
s.state.ExitReason += "\n\n" + err.Error()
}
s.state.Live = false
}
// rtmpURL returns the RTMP URL for the media server, accessible from the host.
func (s *Actor) rtmpURL() string {
return fmt.Sprintf("rtmp://localhost:%d/%s", s.rtmpPort, rtmpPath)

6
mediaserver/integration_test.go
View File

@ -25,7 +25,7 @@ func TestIntegrationMediaServerStartStop(t *testing.T) {
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, containerClient.Close()) })
running, err := containerClient.ContainerRunning(t.Context(), map[string]string{"component": component})
running, err := containerClient.ContainerRunning(t.Context(), containerClient.ContainersWithLabels(map[string]string{"component": component}))
require.NoError(t, err)
assert.False(t, running)
@ -42,7 +42,7 @@ func TestIntegrationMediaServerStartStop(t *testing.T) {
require.Eventually(
t,
func() bool {
running, err = containerClient.ContainerRunning(t.Context(), map[string]string{"component": component})
running, err = containerClient.ContainerRunning(t.Context(), containerClient.ContainersWithLabels(map[string]string{"component": component}))
return err == nil && running
},
time.Second*10,
@ -91,7 +91,7 @@ func TestIntegrationMediaServerStartStop(t *testing.T) {
mediaServer.Close()
running, err = containerClient.ContainerRunning(t.Context(), map[string]string{"component": component})
running, err = containerClient.ContainerRunning(t.Context(), containerClient.ContainersWithLabels(map[string]string{"component": component}))
require.NoError(t, err)
assert.False(t, running)
}

2
multiplexer/integration_test.go
View File

@ -26,7 +26,7 @@ func TestIntegrationMultiplexer(t *testing.T) {
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, containerClient.Close()) })
running, err := containerClient.ContainerRunning(t.Context(), map[string]string{"component": component})
running, err := containerClient.ContainerRunning(t.Context(), containerClient.ContainersWithLabels(map[string]string{"component": component}))
require.NoError(t, err)
assert.False(t, running)

7
multiplexer/multiplexer.go
View File

@ -85,7 +85,7 @@ func (a *Actor) ToggleDestination(url string) {
if _, ok := a.currURLs[url]; ok {
a.logger.Info("Stopping live stream", "url", url)
if err := a.containerClient.RemoveContainers(a.ctx, labels); err != nil {
if err := a.containerClient.RemoveContainers(a.ctx, a.containerClient.ContainersWithLabels(labels)); err != nil {
// TODO: error handling
a.logger.Error("Failed to stop live stream", "url", url, "err", err)
}
@ -171,7 +171,10 @@ func (a *Actor) C() <-chan State {
// Close closes the actor.
func (a *Actor) Close() error {
if err := a.containerClient.RemoveContainers(context.Background(), map[string]string{"component": componentName}); err != nil {
if err := a.containerClient.RemoveContainers(
context.Background(),
a.containerClient.ContainersWithLabels(map[string]string{"component": componentName}),
); err != nil {
return fmt.Errorf("remove containers: %w", err)
}

33
terminal/actor.go
View File

@ -212,6 +212,37 @@ func (a *Actor) actorLoop(ctx context.Context) {
}
}
// ShowStartupCheckModal shows a modal dialog to the user, asking if they want
// to kill a running instance of Octoplex.
//
// The method will block until the user has made a choice, after which the
// channel will receive true if the user wants to quit the other instance, or
// false to quit this instance.
func (a *Actor) ShowStartupCheckModal() <-chan bool {
done := make(chan bool)
modal := tview.NewModal()
modal.SetText("Another instance of Octoplex may already be running. Pressing continue will close that instance. Continue?").
AddButtons([]string{"Continue", "Exit"}).
SetBackgroundColor(tcell.ColorBlack).
SetTextColor(tcell.ColorWhite).
SetDoneFunc(func(buttonIndex int, _ string) {
if buttonIndex == 0 {
done <- true
a.pages.RemovePage("modal")
a.app.SetFocus(a.destView)
} else {
done <- false
}
})
modal.SetBorderStyle(tcell.StyleDefault.Background(tcell.ColorBlack).Foreground(tcell.ColorWhite))
a.pages.AddPage("modal", modal, true, true)
a.app.Draw()
return done
}
// SetState sets the state of the terminal user interface.
func (a *Actor) SetState(state domain.AppState) {
a.ch <- func() {
@ -221,7 +252,7 @@ func (a *Actor) SetState(state domain.AppState) {
AddButtons([]string{"Quit"}).
SetBackgroundColor(tcell.ColorBlack).
SetTextColor(tcell.ColorWhite).
SetDoneFunc(func(buttonIndex int, buttonLabel string) {
SetDoneFunc(func(int, string) {
// TODO: improve app cleanup
a.app.Stop()
})