From df4ab40e12249a8cbcbb95d3fec41af32e6a763e Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Thu, 10 Apr 2025 13:50:08 +1000 Subject: [PATCH] chore(volumes): add client methods for volume management, e2e test --- internal/machine/docker/server.go | 6 +- pkg/api/client.go | 2 +- pkg/api/container.go | 5 - pkg/api/machine.go | 20 ++++ pkg/api/volume.go | 13 +++ pkg/client/cluster.go | 8 +- pkg/client/image.go | 1 + pkg/client/print.go | 14 +++ pkg/client/volume.go | 124 ++++++++++++++++++++++++ test/e2e/assert.go | 29 ++++++ test/e2e/service_test.go | 153 +++++++++++++++--------------- 11 files changed, 286 insertions(+), 89 deletions(-) create mode 100644 pkg/api/machine.go create mode 100644 pkg/client/print.go create mode 100644 pkg/client/volume.go diff --git a/internal/machine/docker/server.go b/internal/machine/docker/server.go index a731b010..ca3df723 100644 --- a/internal/machine/docker/server.go +++ b/internal/machine/docker/server.go @@ -463,7 +463,7 @@ func (s *Server) CreateServiceContainer( config.Labels[api.LabelServicePorts] = strings.Join(encodedPorts, ",") } - mounts, err := toDockerMounts(spec.Volumes, spec.Container.VolumeMounts) + mounts, err := ToDockerMounts(spec.Volumes, spec.Container.VolumeMounts) if err != nil { return nil, err } @@ -537,7 +537,7 @@ func (s *Server) CreateServiceContainer( return &pb.CreateContainerResponse{Response: respBytes}, nil } -func toDockerMounts(volumes []api.VolumeSpec, mounts []api.VolumeMount) ([]mount.Mount, error) { +func ToDockerMounts(volumes []api.VolumeSpec, mounts []api.VolumeMount) ([]mount.Mount, error) { dockerMounts := make([]mount.Mount, 0, len(mounts)) for _, m := range mounts { idx := slices.IndexFunc(volumes, func(v api.VolumeSpec) bool { @@ -627,6 +627,8 @@ func (s *Server) verifyDockerVolumesExist(ctx context.Context, mounts []mount.Mo return status.Errorf(codes.Internal, "inspect volume '%s': %v", m.Source, err.Error()) } // TODO: check if the volume driver and options are the same as in the mount and fail if not. + // Should we even ignore driver-specific options in the volume spec for externally managed volumes? + // Instead, just inspect the existing volume and construct the mount from it. } return nil diff --git a/pkg/api/client.go b/pkg/api/client.go index 60f62009..66a2ea82 100644 --- a/pkg/api/client.go +++ b/pkg/api/client.go @@ -39,7 +39,7 @@ type ImageClient interface { type MachineClient interface { InspectMachine(ctx context.Context, id string) (*pb.MachineMember, error) - ListMachines(ctx context.Context) ([]*pb.MachineMember, error) + ListMachines(ctx context.Context) (MachineMembersList, error) } type ServiceClient interface { diff --git a/pkg/api/container.go b/pkg/api/container.go index 9938b6b1..6b591d0d 100644 --- a/pkg/api/container.go +++ b/pkg/api/container.go @@ -110,11 +110,6 @@ type ServiceContainer struct { ServiceSpec ServiceSpec } -type MachineContainer struct { - MachineID string - Container Container -} - // ServiceID returns the ID of the service this container belongs to. func (c *ServiceContainer) ServiceID() string { return c.Config.Labels[LabelServiceID] diff --git a/pkg/api/machine.go b/pkg/api/machine.go new file mode 100644 index 00000000..6978982b --- /dev/null +++ b/pkg/api/machine.go @@ -0,0 +1,20 @@ +package api + +import "github.com/psviderski/uncloud/internal/machine/api/pb" + +type MachineMembersList []*pb.MachineMember + +func (m MachineMembersList) FindByManagementIP(ip string) *pb.MachineMember { + for _, machine := range m { + addr, err := machine.Machine.Network.ManagementIp.ToAddr() + if err != nil { + continue + } + + if addr.String() == ip { + return machine + } + } + + return nil +} diff --git a/pkg/api/volume.go b/pkg/api/volume.go index e52f79d6..4e71ef0e 100644 --- a/pkg/api/volume.go +++ b/pkg/api/volume.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/docker/docker/api/types/mount" + "github.com/docker/docker/api/types/volume" ) const ( @@ -43,6 +44,8 @@ type BindOptions struct { // VolumeOptions represents options for a managed volume. type VolumeOptions struct { // Driver specifies the volume driver and its options for volume creation. + // TODO: It seems we don't really need Driver and Labels if we only support externally managed volumes. + // However we may need them in the future if we add support for isolated container-scoped volumes. Driver *mount.Driver `json:",omitempty"` // Labels are key-value metadata to apply to the volume if creating a new volume. Labels map[string]string `json:",omitempty"` @@ -182,3 +185,13 @@ func sortVolumeMounts(mounts []VolumeMount) { return false }) } + +// MachineVolume represents a volume on a specific machine. +type MachineVolume struct { + // MachineID is the ID of the machine where the volume exists. + MachineID string + // MachineName is the name of the machine where the volume exists. + MachineName string + // Volume is the Docker volume model. + Volume volume.Volume +} diff --git a/pkg/client/cluster.go b/pkg/client/cluster.go index 17c0f8cf..70bdfbf9 100644 --- a/pkg/client/cluster.go +++ b/pkg/client/cluster.go @@ -2,19 +2,20 @@ package client import ( "context" + "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" "google.golang.org/protobuf/types/known/emptypb" ) -func (cli *Client) InspectMachine(ctx context.Context, id string) (*pb.MachineMember, error) { +func (cli *Client) InspectMachine(ctx context.Context, nameOrID string) (*pb.MachineMember, error) { machines, err := cli.ListMachines(ctx) if err != nil { return nil, err } for _, m := range machines { - if m.Machine.Id == id || m.Machine.Name == id { + if m.Machine.Id == nameOrID || m.Machine.Name == nameOrID { return m, nil } } @@ -22,7 +23,8 @@ func (cli *Client) InspectMachine(ctx context.Context, id string) (*pb.MachineMe return nil, api.ErrNotFound } -func (cli *Client) ListMachines(ctx context.Context) ([]*pb.MachineMember, error) { +// ListMachines returns a list of all machines registered in the cluster. +func (cli *Client) ListMachines(ctx context.Context) (api.MachineMembersList, error) { resp, err := cli.ClusterClient.ListMachines(ctx, &emptypb.Empty{}) if err != nil { return nil, err diff --git a/pkg/client/image.go b/pkg/client/image.go index 91f22cba..9bfff1c1 100644 --- a/pkg/client/image.go +++ b/pkg/client/image.go @@ -2,6 +2,7 @@ package client import ( "context" + dockerclient "github.com/docker/docker/client" "github.com/psviderski/uncloud/pkg/api" ) diff --git a/pkg/client/print.go b/pkg/client/print.go new file mode 100644 index 00000000..7db8a188 --- /dev/null +++ b/pkg/client/print.go @@ -0,0 +1,14 @@ +package client + +import ( + "fmt" + "os" + + "github.com/charmbracelet/lipgloss" +) + +func PrintWarning(msg string) { + style := lipgloss.NewStyle().Foreground(lipgloss.Color("11")) // Bright yellow. + styledMsg := style.Render(fmt.Sprintf("WARNING: %s", msg)) + fmt.Fprintln(os.Stderr, styledMsg) +} diff --git a/pkg/client/volume.go b/pkg/client/volume.go new file mode 100644 index 00000000..3f3af68d --- /dev/null +++ b/pkg/client/volume.go @@ -0,0 +1,124 @@ +package client + +import ( + "context" + "fmt" + + "github.com/docker/compose/v2/pkg/progress" + "github.com/docker/docker/api/types/volume" + dockerclient "github.com/docker/docker/client" + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" +) + +// CreateVolume creates a new volume on the specified machine. +func (cli *Client) CreateVolume( + ctx context.Context, machineNameOrID string, opts volume.CreateOptions, +) (api.MachineVolume, error) { + var resp api.MachineVolume + + if opts.Name == "" { + return resp, fmt.Errorf("volume name is required (anonymous volumes are not supported)") + } + + machine, err := cli.InspectMachine(ctx, machineNameOrID) + if err != nil { + return resp, fmt.Errorf("inspect machine '%s': %w", machineNameOrID, err) + } + // Proxy Docker gRPC requests to the selected machine. + ctx = proxyToMachine(ctx, machine.Machine) + + pw := progress.ContextWriter(ctx) + eventID := fmt.Sprintf("Volume %s on %s", opts.Name, machine.Machine.Name) + pw.Event(progress.CreatingEvent(eventID)) + + vol, err := cli.Docker.CreateVolume(ctx, opts) + if err != nil { + return resp, fmt.Errorf("create volume on machine '%s': %w", machine.Machine.Name, err) + } + + resp = api.MachineVolume{ + MachineID: machine.Machine.Id, + MachineName: machine.Machine.Name, + Volume: vol, + } + pw.Event(progress.CreatedEvent(eventID)) + + return resp, nil +} + +// ListVolumes returns a list of all volumes on the cluster machines. +func (cli *Client) ListVolumes(ctx context.Context) ([]api.MachineVolume, error) { + machines, err := cli.ListMachines(ctx) + if err != nil { + return nil, fmt.Errorf("list machines: %w", err) + } + + // Broadcast the volume list request to all machines. + listCtx, err := api.ProxyMachinesContext(ctx, cli, nil) + if err != nil { + return nil, fmt.Errorf("create request context to broadcast to all machines: %w", err) + } + + machineVolumes, err := cli.Docker.ListVolumes(listCtx, volume.ListOptions{}) + if err != nil { + return nil, fmt.Errorf("list volumes: %w", err) + } + + var volumes []api.MachineVolume + // Process responses from all machines. + for _, mv := range machineVolumes { + if mv.Metadata != nil && mv.Metadata.Error != "" { + // TODO: return failed machines in the response. + PrintWarning(fmt.Sprintf("failed to list volumes on machine '%s': %s", + mv.Metadata.Machine, mv.Metadata.Error)) + continue + } + + var m *pb.MachineMember + if mv.Metadata == nil { + // ListVolumes was proxied to only one machine. + m = machines[0] + } else { + m = machines.FindByManagementIP(mv.Metadata.Machine) + if m == nil { + return nil, fmt.Errorf("machine not found by management IP: %s", mv.Metadata.Machine) + } + } + + for _, vol := range mv.Response.Volumes { + volumes = append(volumes, api.MachineVolume{ + MachineID: m.Machine.Id, + MachineName: m.Machine.Name, + Volume: *vol, + }) + } + } + + return volumes, nil +} + +// RemoveVolume removes a volume from the specified machine. +func (cli *Client) RemoveVolume(ctx context.Context, machineNameOrID, volumeName string, force bool) error { + machine, err := cli.InspectMachine(ctx, machineNameOrID) + if err != nil { + return fmt.Errorf("inspect machine '%s': %w", machineNameOrID, err) + } + // Proxy Docker gRPC requests to the selected machine. + ctx = proxyToMachine(ctx, machine.Machine) + + pw := progress.ContextWriter(ctx) + eventID := fmt.Sprintf("Volume %s on %s", volumeName, machine.Machine.Name) + pw.Event(progress.RemovingEvent(eventID)) + + if err = cli.Docker.RemoveVolume(ctx, volumeName, force); err != nil { + if dockerclient.IsErrNotFound(err) { + return api.ErrNotFound + } + return fmt.Errorf("remove volume '%s' from machine '%s': %w", + volumeName, machine.Machine.Name, err) + } + pw.Event(progress.RemovedEvent(eventID)) + + return nil +} diff --git a/test/e2e/assert.go b/test/e2e/assert.go index 6447751f..54518a8a 100644 --- a/test/e2e/assert.go +++ b/test/e2e/assert.go @@ -1,16 +1,21 @@ package e2e import ( + "reflect" + "slices" "strconv" + "strings" "testing" mapset "github.com/deckarep/golang-set/v2" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/mount" "github.com/docker/go-connections/nat" machinedocker "github.com/psviderski/uncloud/internal/machine/docker" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client/deploy" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpec) { @@ -34,6 +39,7 @@ func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api spec = spec.SetDefaults() // Verify labels. + assert.True(t, api.ValidateServiceID(ctr.Config.Labels[api.LabelServiceID])) assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName]) assert.Equal(t, spec.Mode, ctr.Config.Labels[api.LabelServiceMode]) assert.Contains(t, ctr.Config.Labels, api.LabelManaged) @@ -54,7 +60,11 @@ func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api assert.Equal(t, spec.Container.Image, ctr.Config.Image) assert.Equal(t, spec.Container.Init, ctr.HostConfig.Init) + assert.True(t, strings.HasPrefix(ctr.Name, spec.Name+"-")) + + assert.Empty(t, ctr.HostConfig.Binds, "Expected empty binds as all volumes should be mapped to mounts") assert.ElementsMatch(t, spec.Container.Volumes, ctr.HostConfig.Binds) + assertContainerMountsMatchSpec(t, ctr.HostConfig.Mounts, spec) // Compare host ports. portBindings := make(nat.PortMap) @@ -84,6 +94,25 @@ func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName) } +func assertContainerMountsMatchSpec(t *testing.T, mounts []mount.Mount, spec api.ServiceSpec) { + expectedMounts, err := machinedocker.ToDockerMounts(spec.Volumes, spec.Container.VolumeMounts) + require.NoError(t, err) + + sortMounts(mounts) + sortMounts(expectedMounts) + + assert.Len(t, mounts, len(expectedMounts), "Expected %d mounts", len(expectedMounts)) + for i, m := range mounts { + assert.True(t, reflect.DeepEqual(m, expectedMounts[i]), "Expected mount type=%s,src=%s,dst=%s to match spec") + } +} + +func sortMounts(mounts []mount.Mount) { + slices.SortFunc(mounts, func(a, b mount.Mount) int { + return strings.Compare(a.Target, b.Target) + }) +} + // serviceContainersByMachine returns a map of machine ID to service containers on that machine. func serviceContainersByMachine(t *testing.T, svc api.Service) map[string][]api.ServiceContainer { containers := make(map[string][]api.ServiceContainer) diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index e80c04f2..db94b946 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -4,13 +4,12 @@ import ( "context" "errors" "net/netip" - "strings" "testing" "github.com/docker/docker/api/types/container" - "github.com/docker/go-connections/nat" + "github.com/docker/docker/api/types/mount" + "github.com/docker/docker/api/types/volume" "github.com/psviderski/uncloud/internal/machine/api/pb" - machinedocker "github.com/psviderski/uncloud/internal/machine/docker" "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/pkg/api" @@ -651,34 +650,7 @@ func TestServiceLifecycle(t *testing.T) { // Verify container configuration. mc, err := cli.InspectContainer(ctx, serviceID, resp.ID) require.NoError(t, err) - ctr := mc.Container - - assert.True(t, strings.HasPrefix(ctr.Name, "container-spec-defaults-")) - assert.Equal(t, "portainer/pause:latest", ctr.Config.Image) - - // Verify default settings. - assert.Empty(t, ctr.Config.Cmd) - assert.EqualValues(t, []string{"/pause"}, ctr.Config.Entrypoint) // Populated by the image. - assert.Equal(t, []string{"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"}, ctr.Config.Env) - - assert.Nil(t, ctr.HostConfig.Init) - assert.Empty(t, ctr.HostConfig.Binds) - assert.Empty(t, ctr.HostConfig.PortBindings) - assert.Equal(t, container.RestartPolicy{ - Name: container.RestartPolicyAlways, - MaximumRetryCount: 0, - }, ctr.HostConfig.RestartPolicy) - - // Verify labels. - assert.Equal(t, serviceID, ctr.Config.Labels[api.LabelServiceID]) - assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName]) - assert.Equal(t, api.ServiceModeReplicated, ctr.Config.Labels[api.LabelServiceMode]) - assert.NotContains(t, ctr.Config.Labels, api.LabelServicePorts) // No ports set. - assert.Contains(t, ctr.Config.Labels, api.LabelManaged) - - // Verify network settings. - assert.Len(t, ctr.NetworkSettings.Networks, 1) - assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName) + assertContainerMatchesSpec(t, mc.Container, spec) }) t.Run("create container with full spec", func(t *testing.T) { @@ -699,9 +671,28 @@ func TestServiceLifecycle(t *testing.T) { "BOOL": "true", "": "ignored", }, - Image: "portainer/pause:latest", - Init: &init, - Volumes: []string{"/host/path:/container/path:ro"}, + Image: "portainer/pause:latest", + Init: &init, + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "hostpath", + ContainerPath: "/volumes/hostpath", + ReadOnly: true, + }, + { + VolumeName: "container-spec-full-default-volume", + ContainerPath: "/volumes/default-volume", + }, + { + VolumeName: "custom-volume", + ContainerPath: "/volumes/custom-volume", + ReadOnly: true, + }, + { + VolumeName: "tmpfs", + ContainerPath: "/volumes/tmpfs", + }, + }, }, Ports: []api.PortSpec{ { @@ -718,6 +709,55 @@ func TestServiceLifecycle(t *testing.T) { Mode: api.PortModeIngress, }, }, + Volumes: []api.VolumeSpec{ + { + Name: "hostpath", + Type: api.VolumeTypeBind, + BindOptions: &api.BindOptions{ + HostPath: "/tmp/container-spec-full/host/path", + CreateHostPath: true, + }, + }, + { + Name: "container-spec-full-default-volume", + Type: api.VolumeTypeVolume, + }, + { + Name: "custom-volume", + Type: api.VolumeTypeVolume, + VolumeOptions: &api.VolumeOptions{ + Driver: &mount.Driver{ + Name: "local", + }, + Labels: map[string]string{ + "key": "value", + }, + Name: "container-spec-full-custom-volume", + NoCopy: true, + }, + }, + { + Name: "tmpfs", + Type: api.VolumeTypeTmpfs, + }, + }, + } + + // Create the volumes before creating the container as they must be managed externally. + volumeNames := []string{ + "container-spec-full-default-volume", + "container-spec-full-custom-volume", + } + for _, name := range volumeNames { + _, err = cli.CreateVolume(ctx, c.Machines[0].Name, volume.CreateOptions{Name: name}) + require.NoError(t, err) + + t.Cleanup(func() { + err := cli.RemoveVolume(ctx, c.Machines[0].Name, name, false) + if !errors.Is(err, api.ErrNotFound) { + require.NoError(t, err) + } + }) } resp, err := cli.CreateContainer(ctx, serviceID, spec, c.Machines[0].Name) @@ -734,50 +774,7 @@ func TestServiceLifecycle(t *testing.T) { // Verify container configuration. mc, err := cli.InspectContainer(ctx, serviceID, resp.ID) require.NoError(t, err) - ctr := mc.Container - - assert.True(t, strings.HasPrefix(ctr.Name, "container-spec-full-")) - assert.Equal(t, "portainer/pause:latest", ctr.Config.Image) - - assert.EqualValues(t, spec.Container.Command, ctr.Config.Cmd) - assert.EqualValues(t, spec.Container.Entrypoint, ctr.Config.Entrypoint) - - expectedEnv := []string{ - "BOOL=true", - "EMTPY=", - "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", - "VAR=value", - } - assert.ElementsMatch(t, expectedEnv, ctr.Config.Env) - - assert.True(t, *ctr.HostConfig.Init) - assert.Len(t, ctr.HostConfig.Binds, 1) - assert.Contains(t, ctr.HostConfig.Binds, spec.Container.Volumes[0]) - - assert.Len(t, ctr.HostConfig.PortBindings, 1) - expectedPort := []nat.PortBinding{ - { - HostIP: "127.0.0.1", - HostPort: "80", - }, - } - assert.Equal(t, expectedPort, ctr.HostConfig.PortBindings[nat.Port("8080/tcp")]) - - assert.Equal(t, container.RestartPolicy{ - Name: container.RestartPolicyAlways, - MaximumRetryCount: 0, - }, ctr.HostConfig.RestartPolicy) - - // Verify labels. - assert.Equal(t, serviceID, ctr.Config.Labels[api.LabelServiceID]) - assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName]) - assert.Equal(t, api.ServiceModeGlobal, ctr.Config.Labels[api.LabelServiceMode]) - assert.Equal(t, "127.0.0.1:80:8080/tcp@host,app.example.com:8000/https", ctr.Config.Labels[api.LabelServicePorts]) - assert.Contains(t, ctr.Config.Labels, api.LabelManaged) - - // Verify network settings. - assert.Len(t, ctr.NetworkSettings.Networks, 1) - assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName) + assertContainerMatchesSpec(t, mc.Container, spec) }) t.Run("create container invalid service", func(t *testing.T) {