From c4287b75b2b727e7ffd78c99c8da2a56a553e9c1 Mon Sep 17 00:00:00 2001 From: Rob Watson Date: Fri, 28 Feb 2025 20:14:41 +0100 Subject: [PATCH] refactor: domain labels --- container/container.go | 18 +++++++++++++----- container/integration_test.go | 16 ++++++++-------- mediaserver/actor.go | 4 ++-- mediaserver/integration_test.go | 6 +++--- multiplexer/integration_test.go | 2 +- multiplexer/multiplexer.go | 7 +++++-- 6 files changed, 32 insertions(+), 21 deletions(-) diff --git a/container/container.go b/container/container.go index fae11ae..983cea6 100644 --- a/container/container.go +++ b/container/container.go @@ -48,6 +48,14 @@ type DockerClient interface { 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 // additional functionality such as exposing container stats. type Client struct { @@ -159,8 +167,8 @@ func (a *Client) RunContainer(ctx context.Context, params RunContainerParams) (< containerConfig := *params.ContainerConfig containerConfig.Labels = make(map[string]string) maps.Copy(containerConfig.Labels, params.ContainerConfig.Labels) - containerConfig.Labels["app"] = domain.AppName - containerConfig.Labels["app-id"] = a.id.String() + containerConfig.Labels[LabelApp] = domain.AppName + containerConfig.Labels[LabelAppID] = a.id.String() var name string if params.Name != "" { @@ -428,7 +436,7 @@ func (a *Client) ContainersWithLabels(extraLabels map[string]string) LabelOption // app instance. func AllContainers() LabelOptions { 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 { labels := map[string]string{ - "app": domain.AppName, - "app-id": a.id.String(), + LabelApp: domain.AppName, + LabelAppID: a.id.String(), } for _, el := range extraLabels { diff --git a/container/integration_test.go b/container/integration_test.go index f6c6eec..3595507 100644 --- a/container/integration_test.go +++ b/container/integration_test.go @@ -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, client.ContainersWithLabels(map[string]string{"component": component})) + running, err := client.ContainerRunning(ctx, client.ContainersWithLabels(map[string]string{container.LabelComponent: component})) require.NoError(t, err) assert.False(t, running) @@ -39,7 +39,7 @@ func TestIntegrationClientStartStop(t *testing.T) { ChanSize: 1, ContainerConfig: &typescontainer.Config{ Image: "netfluxio/mediamtx-alpine:latest", - Labels: map[string]string{"component": component}, + Labels: map[string]string{container.LabelComponent: component}, }, HostConfig: &typescontainer.HostConfig{ NetworkMode: "default", @@ -51,7 +51,7 @@ func TestIntegrationClientStartStop(t *testing.T) { require.Eventually( t, 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 }, 5*time.Second, @@ -62,7 +62,7 @@ func TestIntegrationClientStartStop(t *testing.T) { client.Close() 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) assert.False(t, running) } @@ -84,7 +84,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) { ChanSize: 1, ContainerConfig: &typescontainer.Config{ 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"}, }) @@ -95,7 +95,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) { ChanSize: 1, ContainerConfig: &typescontainer.Config{ 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"}, }) @@ -106,7 +106,7 @@ func TestIntegrationClientRemoveContainers(t *testing.T) { ChanSize: 1, ContainerConfig: &typescontainer.Config{ 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"}, }) @@ -186,7 +186,7 @@ func TestContainerRestart(t *testing.T) { ContainerConfig: &typescontainer.Config{ Image: "alpine:latest", Cmd: []string{"sleep", "1"}, - Labels: map[string]string{"component": component}, + Labels: map[string]string{container.LabelComponent: component}, }, HostConfig: &typescontainer.HostConfig{ NetworkMode: "default", diff --git a/mediaserver/actor.go b/mediaserver/actor.go index 5c59645..06598e4 100644 --- a/mediaserver/actor.go +++ b/mediaserver/actor.go @@ -96,7 +96,7 @@ func StartActor(ctx context.Context, params StartActorParams) *Actor { "MTX_API=yes", }, Labels: map[string]string{ - "component": componentName, + container.LabelComponent: componentName, }, Healthcheck: &typescontainer.HealthConfig{ 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 { if err := s.containerClient.RemoveContainers( context.Background(), - s.containerClient.ContainersWithLabels(map[string]string{"component": componentName}), + s.containerClient.ContainersWithLabels(map[string]string{container.LabelComponent: componentName}), ); err != nil { return fmt.Errorf("remove containers: %w", err) } diff --git a/mediaserver/integration_test.go b/mediaserver/integration_test.go index 032ebbe..93ad4e4 100644 --- a/mediaserver/integration_test.go +++ b/mediaserver/integration_test.go @@ -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(), 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) assert.False(t, running) @@ -42,7 +42,7 @@ func TestIntegrationMediaServerStartStop(t *testing.T) { require.Eventually( t, 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 }, time.Second*10, @@ -91,7 +91,7 @@ func TestIntegrationMediaServerStartStop(t *testing.T) { 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) assert.False(t, running) } diff --git a/multiplexer/integration_test.go b/multiplexer/integration_test.go index 96d886c..f75d721 100644 --- a/multiplexer/integration_test.go +++ b/multiplexer/integration_test.go @@ -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(), 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) assert.False(t, running) diff --git a/multiplexer/multiplexer.go b/multiplexer/multiplexer.go index 4463b10..ea076e5 100644 --- a/multiplexer/multiplexer.go +++ b/multiplexer/multiplexer.go @@ -80,7 +80,10 @@ func NewActor(ctx context.Context, params NewActorParams) *Actor { // ToggleDestination toggles the destination stream between on and off. func (a *Actor) ToggleDestination(url string) { 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 { a.logger.Info("Stopping live stream", "url", url) @@ -173,7 +176,7 @@ func (a *Actor) C() <-chan State { func (a *Actor) Close() error { if err := a.containerClient.RemoveContainers( context.Background(), - a.containerClient.ContainersWithLabels(map[string]string{"component": componentName}), + a.containerClient.ContainersWithLabels(map[string]string{container.LabelComponent: componentName}), ); err != nil { return fmt.Errorf("remove containers: %w", err) }