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

refactor: domain labels

Browse Source
This commit is contained in:
Rob Watson 2025-02-28 20:14:41 +01:00
parent 4bd594f54c
commit c4287b75b2
6 changed files with 32 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
container/container.go
View File

@ -48,6 +48,14 @@ type DockerClient interface {
NetworkRemove(context.Context, string) error NetworkRemove(context.Context, string) error
} }
const (
LabelPrefix = "io.netflux.octoplex."
LabelApp = LabelPrefix + "app"
LabelAppID = LabelPrefix + "app-id"
LabelComponent = LabelPrefix + "component"
LabelURL = LabelPrefix + "url"
)
// Client provides a thin wrapper around the Docker API client, and provides // Client provides a thin wrapper around the Docker API client, and provides
// additional functionality such as exposing container stats. // additional functionality such as exposing container stats.
type Client struct { type Client struct {
@ -159,8 +167,8 @@ func (a *Client) RunContainer(ctx context.Context, params RunContainerParams) (<
containerConfig := *params.ContainerConfig containerConfig := *params.ContainerConfig
containerConfig.Labels = make(map[string]string) containerConfig.Labels = make(map[string]string)
maps.Copy(containerConfig.Labels, params.ContainerConfig.Labels) maps.Copy(containerConfig.Labels, params.ContainerConfig.Labels)
containerConfig.Labels["app"] = domain.AppName containerConfig.Labels[LabelApp] = domain.AppName
containerConfig.Labels["app-id"] = a.id.String() containerConfig.Labels[LabelAppID] = a.id.String()
var name string var name string
if params.Name != "" { if params.Name != "" {
@ -428,7 +436,7 @@ func (a *Client) ContainersWithLabels(extraLabels map[string]string) LabelOption
// app instance. // app instance.
func AllContainers() LabelOptions { func AllContainers() LabelOptions {
return func() map[string]string { return func() map[string]string {
return map[string]string{"app": domain.AppName} return map[string]string{LabelApp: domain.AppName}
} }
} }
@ -445,8 +453,8 @@ func (a *Client) containersMatchingLabels(ctx context.Context, labels map[string
func (a *Client) instanceLabels(extraLabels ...map[string]string) map[string]string { func (a *Client) instanceLabels(extraLabels ...map[string]string) map[string]string {
labels := map[string]string{ labels := map[string]string{
"app": domain.AppName, LabelApp: domain.AppName,
"app-id": a.id.String(), LabelAppID: a.id.String(),
} }
for _, el := range extraLabels { for _, el := range extraLabels {

16
container/integration_test.go
View File

@ -30,7 +30,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
client, err := container.NewClient(ctx, apiClient, logger) client, err := container.NewClient(ctx, apiClient, logger)
require.NoError(t, err) require.NoError(t, err)
running, err := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component})) running, err := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{container.LabelComponent: component}))
require.NoError(t, err) require.NoError(t, err)
assert.False(t, running) assert.False(t, running)
@ -39,7 +39,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
ChanSize: 1, ChanSize: 1,
ContainerConfig: &typescontainer.Config{ ContainerConfig: &typescontainer.Config{
Image: "netfluxio/mediamtx-alpine:latest", Image: "netfluxio/mediamtx-alpine:latest",
Labels: map[string]string{"component": component}, Labels: map[string]string{container.LabelComponent: component},
}, },
HostConfig: &typescontainer.HostConfig{ HostConfig: &typescontainer.HostConfig{
NetworkMode: "default", NetworkMode: "default",
@ -51,7 +51,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
require.Eventually( require.Eventually(
t, t,
func() bool { func() bool {
running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component})) running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{container.LabelComponent: component}))
return err == nil && running return err == nil && running
}, },
5*time.Second, 5*time.Second,
@ -62,7 +62,7 @@ func TestIntegrationClientStartStop(t *testing.T) {
client.Close() client.Close()
require.NoError(t, <-errC) require.NoError(t, <-errC)
running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{"component": component})) running, err = client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{container.LabelComponent: component}))
require.NoError(t, err) require.NoError(t, err)
assert.False(t, running) assert.False(t, running)
} }
@ -84,7 +84,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
ChanSize: 1, ChanSize: 1,
ContainerConfig: &typescontainer.Config{ ContainerConfig: &typescontainer.Config{
Image: "netfluxio/mediamtx-alpine:latest", Image: "netfluxio/mediamtx-alpine:latest",
Labels: map[string]string{"component": component, "group": "test1"}, Labels: map[string]string{container.LabelComponent: component, "group": "test1"},
}, },
HostConfig: &typescontainer.HostConfig{NetworkMode: "default"}, HostConfig: &typescontainer.HostConfig{NetworkMode: "default"},
}) })
@ -95,7 +95,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
ChanSize: 1, ChanSize: 1,
ContainerConfig: &typescontainer.Config{ ContainerConfig: &typescontainer.Config{
Image: "netfluxio/mediamtx-alpine:latest", Image: "netfluxio/mediamtx-alpine:latest",
Labels: map[string]string{"component": component, "group": "test1"}, Labels: map[string]string{container.LabelComponent: component, "group": "test1"},
}, },
HostConfig: &typescontainer.HostConfig{NetworkMode: "default"}, HostConfig: &typescontainer.HostConfig{NetworkMode: "default"},
}) })
@ -106,7 +106,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) {
ChanSize: 1, ChanSize: 1,
ContainerConfig: &typescontainer.Config{ ContainerConfig: &typescontainer.Config{
Image: "netfluxio/mediamtx-alpine:latest", Image: "netfluxio/mediamtx-alpine:latest",
Labels: map[string]string{"component": component, "group": "test2"}, Labels: map[string]string{container.LabelComponent: component, "group": "test2"},
}, },
HostConfig: &typescontainer.HostConfig{NetworkMode: "default"}, HostConfig: &typescontainer.HostConfig{NetworkMode: "default"},
}) })
@ -186,7 +186,7 @@ func TestContainerRestart(t *testing.T) {
ContainerConfig: &typescontainer.Config{ ContainerConfig: &typescontainer.Config{
Image: "alpine:latest", Image: "alpine:latest",
Cmd: []string{"sleep", "1"}, Cmd: []string{"sleep", "1"},
Labels: map[string]string{"component": component}, Labels: map[string]string{container.LabelComponent: component},
}, },
HostConfig: &typescontainer.HostConfig{ HostConfig: &typescontainer.HostConfig{
NetworkMode: "default", NetworkMode: "default",

4
mediaserver/actor.go
View File

@ -96,7 +96,7 @@ func StartActor(ctx context.Context, params StartActorParams) *Actor {
"MTX_API=yes", "MTX_API=yes",
}, },
Labels: map[string]string{ Labels: map[string]string{
"component": componentName, container.LabelComponent: componentName,
}, },
Healthcheck: &typescontainer.HealthConfig{ Healthcheck: &typescontainer.HealthConfig{
Test: []string{"CMD", "curl", "-f", "http://localhost:9997/v3/paths/list"}, Test: []string{"CMD", "curl", "-f", "http://localhost:9997/v3/paths/list"},
@ -141,7 +141,7 @@ func (s *Actor) State() domain.Source {
func (s *Actor) Close() error { func (s *Actor) Close() error {
if err := s.containerClient.RemoveContainers( if err := s.containerClient.RemoveContainers(
context.Background(), context.Background(),
s.containerClient.ContainersWithLabels(map[string]string{"component": componentName}), s.containerClient.ContainersWithLabels(map[string]string{container.LabelComponent: componentName}),
); err != nil { ); err != nil {
return fmt.Errorf("remove containers: %w", err) return fmt.Errorf("remove containers: %w", err)
} }

6
mediaserver/integration_test.go
View File

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

2
multiplexer/integration_test.go
View File

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

7
multiplexer/multiplexer.go
View File

@ -80,7 +80,10 @@ func NewActor(ctx context.Context, params NewActorParams) *Actor {
// ToggleDestination toggles the destination stream between on and off. // ToggleDestination toggles the destination stream between on and off.
func (a *Actor) ToggleDestination(url string) { func (a *Actor) ToggleDestination(url string) {
a.actorC <- func() { a.actorC <- func() {
labels := map[string]string{"component": componentName, "url": url} labels := map[string]string{
container.LabelComponent: componentName,
container.LabelURL: url,
}
if _, ok := a.currURLs[url]; ok { if _, ok := a.currURLs[url]; ok {
a.logger.Info("Stopping live stream", "url", url) a.logger.Info("Stopping live stream", "url", url)
@ -173,7 +176,7 @@ func (a *Actor) C() <-chan State {
func (a *Actor) Close() error { func (a *Actor) Close() error {
if err := a.containerClient.RemoveContainers( if err := a.containerClient.RemoveContainers(
context.Background(), context.Background(),
a.containerClient.ContainersWithLabels(map[string]string{"component": componentName}), a.containerClient.ContainersWithLabels(map[string]string{container.LabelComponent: componentName}),
); err != nil { ); err != nil {
return fmt.Errorf("remove containers: %w", err) return fmt.Errorf("remove containers: %w", err)
} }