diff --git a/cmd/uncloud/image/push.go b/cmd/uncloud/image/push.go index f277205a..3504e831 100644 --- a/cmd/uncloud/image/push.go +++ b/cmd/uncloud/image/push.go @@ -1,14 +1,18 @@ package image import ( + "context" "fmt" + "github.com/docker/compose/v2/pkg/progress" "github.com/psviderski/uncloud/internal/cli" "github.com/spf13/cobra" ) type pushOptions struct { + image string machines []string + context string } func NewPushCommand() *cobra.Command { @@ -29,19 +33,37 @@ The image is uploaded to the machine which CLI is connected to (default) or the uc image push myapp:latest -m machine1,machine2,machine3`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { - image := args[0] + uncli := cmd.Context().Value("cli").(*cli.CLI) - machines := cli.ExpandCommaSeparatedValues(opts.machines) - - // TODO: Implement image push logic - fmt.Printf("Would push image %q to machines: %v\n", image, machines) - return fmt.Errorf("image push not yet implemented") + opts.image = args[0] + return push(cmd.Context(), uncli, opts) }, } cmd.Flags().StringSliceVarP(&opts.machines, "machine", "m", nil, "Machine names to push the image to. Can be specified multiple times or as a comma-separated "+ "list of machine names. (default is connected machine)") + cmd.Flags().StringVarP( + &opts.context, "context", "c", "", + "Name of the cluster context. (default is the current context)", + ) return cmd } + +func push(ctx context.Context, uncli *cli.CLI, opts pushOptions) error { + client, err := uncli.ConnectCluster(ctx, opts.context) + if err != nil { + return fmt.Errorf("connect to cluster: %w", err) + } + defer client.Close() + + machines := cli.ExpandCommaSeparatedValues(opts.machines) + + return progress.RunWithTitle(ctx, func(ctx context.Context) error { + if err = client.PushImage(ctx, opts.image, machines); err != nil { + return fmt.Errorf("push image to cluster: %w", err) + } + return nil + }, uncli.ProgressOut(), fmt.Sprintf("Pushing image %s to cluster", opts.image)) +} diff --git a/pkg/client/image.go b/pkg/client/image.go index 9bfff1c1..fb52955e 100644 --- a/pkg/client/image.go +++ b/pkg/client/image.go @@ -2,9 +2,29 @@ package client import ( "context" + "errors" + "fmt" + "net" + "strconv" + "strings" + "sync" + "time" + "github.com/charmbracelet/lipgloss" + "github.com/docker/compose/v2/pkg/progress" + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/image" dockerclient "github.com/docker/docker/client" + "github.com/docker/docker/pkg/jsonmessage" + "github.com/docker/go-connections/nat" + "github.com/psviderski/uncloud/internal/docker" + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/internal/machine/constants" + "github.com/psviderski/uncloud/internal/machine/network" + "github.com/psviderski/uncloud/internal/proxy" + "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/pkg/api" + netproxy "golang.org/x/net/proxy" ) func (cli *Client) InspectImage(ctx context.Context, id string) ([]api.MachineImage, error) { @@ -19,3 +39,349 @@ func (cli *Client) InspectImage(ctx context.Context, id string) ([]api.MachineIm func (cli *Client) InspectRemoteImage(ctx context.Context, id string) ([]api.MachineRemoteImage, error) { return cli.Docker.InspectRemoteImage(ctx, id) } + +// PushImage pushes a local Docker image to the specified machines. If no machines are specified, +// it pushes to the machine the client is connected to. +func (cli *Client) PushImage(ctx context.Context, image string, machineNamesOrIDs []string) error { + dockerCliWrapped, err := dockerclient.NewClientWithOpts(dockerclient.FromEnv, + dockerclient.WithAPIVersionNegotiation()) + if err != nil { + return fmt.Errorf("create Docker client: %w", err) + } + dockerCli := &docker.Client{Client: dockerCliWrapped} + defer dockerCli.Close() + + // Check if Docker image exists locally. + if _, _, err = dockerCli.ImageInspectWithRaw(ctx, image); err != nil { + if dockerclient.IsErrNotFound(err) { + return fmt.Errorf("image '%s' not found locally", image) + } + return fmt.Errorf("inspect image '%s' locally: %w", image, err) + } + + // Get the machine info for the specified machines or the connected machine if none are specified. + var machines []*pb.MachineInfo + if len(machineNamesOrIDs) > 0 { + machineMembers, err := cli.ListMachines(ctx, &api.MachineFilter{ + NamesOrIDs: machineNamesOrIDs, + }) + if err != nil { + return fmt.Errorf("list machines: %w", err) + } + + // Check if all specified machines were found. + if len(machineMembers) != len(machineNamesOrIDs) { + var notFound []string + for _, nameOrID := range machineNamesOrIDs { + if machineMembers.FindByNameOrID(nameOrID) == nil { + notFound = append(notFound, nameOrID) + } + } + + return fmt.Errorf("machines not found: %s", strings.Join(notFound, ", ")) + } + + for _, mm := range machineMembers { + machines = append(machines, mm.Machine) + } + } else { + // No machines specified, use the connected machine. + m, err := cli.MachineClient.Inspect(ctx, nil) + if err != nil { + return fmt.Errorf("inspect connected machine: %w", err) + } + + // If the machine has been renamed, the new name will only be stored in the cluster store. .Inspect will return + // the old name from the machine config. So we need to fetch the machine info from the cluster. + // TODO: make one source of truth for machine info. + mm, err := cli.InspectMachine(ctx, m.Id) + if err != nil { + return fmt.Errorf("inspect machine: %w", err) + } + machines = append(machines, mm.Machine) + } + + // Push image to all specified machines. + var wg sync.WaitGroup + errCh := make(chan error, len(machines)) + + // TODO: detect the target machine platform and figure out how to handle scenarios when local and target + // platforms differ. + for _, m := range machines { + wg.Go(func() { + if err := cli.pushImageToMachine(ctx, dockerCli, image, m); err != nil { + errCh <- fmt.Errorf("push image to machine '%s': %w", m.Name, err) + } + }) + } + + wg.Wait() + close(errCh) + + var errs []error + for err = range errCh { + errs = append(errs, err) + } + + return errors.Join(errs...) +} + +// pushImageToMachine pushes a local Docker image to a specific machine using local port forwarding to its unregistry. +func (cli *Client) pushImageToMachine( + ctx context.Context, dockerCli *docker.Client, imageName string, machine *pb.MachineInfo, +) error { + pw := progress.ContextWriter(ctx) + + machineSubnet, _ := machine.Network.Subnet.ToPrefix() + machineIP := network.MachineIP(machineSubnet) + unregistryAddr := net.JoinHostPort(machineIP.String(), strconv.Itoa(constants.UnregistryPort)) + + dialer, err := cli.connector.Dialer() + if err != nil { + return fmt.Errorf("get proxy dialer: %w", err) + } + + boldStyle := lipgloss.NewStyle().Bold(true) + proxyEventID := fmt.Sprintf("Proxy to unregistry on %s", boldStyle.Render(machine.Name)) + pw.Event(progress.StartingEvent(proxyEventID)) + + // Forward local port 127.0.0.1:PORT to the machine's unregistry over the established client connection. + unregProxy, err := newUnregistryProxy(ctx, unregistryAddr, dialer, func(err error) { + pw.Event(progress.NewEvent(proxyEventID, progress.Error, err.Error())) + }) + if err != nil { + pw.Event(progress.NewEvent(proxyEventID, progress.Error, err.Error())) + return fmt.Errorf("create local proxy to unregistry on machine '%s': %w", machine.Name, err) + } + // Get the local port the unregistry proxy is listening on. + proxyPort := unregProxy.Listener.Addr().(*net.TCPAddr).Port + + proxyCtx, cancelProxy := context.WithCancel(ctx) + proxyCtrID := "" + pushImageTag := "" + + // Cleanup function to remove temporary resources and stop proxies. + cleanup := func() { + // Remove temporary image tag. + if pushImageTag != "" { + dockerCli.ImageRemove(ctx, pushImageTag, image.RemoveOptions{}) + } + + // Remove socat proxy container. + if proxyCtrID != "" { + dockerCli.ContainerRemove(ctx, proxyCtrID, container.RemoveOptions{Force: true}) + } + + cancelProxy() + } + defer cleanup() + + go unregProxy.Run(proxyCtx) + + dockerVirtualised, err := isDockerVirtualised(ctx, dockerCli) + if err != nil { + return err + } + + if dockerVirtualised { + // Run socat proxy container to forward a localhost port from within the Docker VM to the host machine. + pw.Event(progress.Event{ + ID: proxyEventID, + Status: progress.Working, + StatusText: "Starting", + Text: "(detected virtualised Docker locally, starting socat container to proxy to Docker VM)", + }) + + proxyCtrID, proxyPort, err = runDockerVMProxyContainer(ctx, dockerCli, proxyPort) + if err != nil { + pw.Event(progress.NewEvent(proxyEventID, progress.Error, err.Error())) + return fmt.Errorf("run socat container to proxy unregistry to Docker VM: %w", err) + } + } + + pw.Event(progress.Event{ + ID: proxyEventID, + Status: progress.Done, + StatusText: "Started", + Text: fmt.Sprintf("(localhost:%d → %s)", proxyPort, unregistryAddr), + }) + + // Tag the image for pushing through the proxy. + pushImageTag = fmt.Sprintf("127.0.0.1:%d/%s", proxyPort, imageName) + if err = dockerCli.ImageTag(ctx, imageName, pushImageTag); err != nil { + return fmt.Errorf("tag image for push: %w", err) + } + + // Push the image through the proxy. + pushEventID := fmt.Sprintf("Pushing %s to %s", boldStyle.Render(imageName), boldStyle.Render(machine.Name)) + pw.Event(progress.NewEvent(pushEventID, progress.Working, "Pushing")) + + pushCh, err := dockerCli.PushImage(ctx, pushImageTag, image.PushOptions{}) + if err != nil { + pw.Event(progress.NewEvent(pushEventID, progress.Error, err.Error())) + return fmt.Errorf("push image: %w", err) + } + + // Wait for push to complete by reading all progress messages and converting them to events. + // If the context is cancelled, the pushCh will receive a context cancellation error. + for msg := range pushCh { + if msg.Err != nil { + pw.Event(progress.NewEvent(pushEventID, progress.Error, msg.Err.Error())) + return fmt.Errorf("push image: %w", msg.Err) + } + + // TODO: support quite mode like in compose: --quiet Push without printing progress information + if e := toPushProgressEvent(msg.Message); e != nil { + e.ID = fmt.Sprintf("Layer %s on %s:", e.ID, boldStyle.Render(machine.Name)) + e.ParentID = pushEventID + pw.Event(*e) + } + } + pw.Event(progress.NewEvent(pushEventID, progress.Done, "Pushed")) + + return nil +} + +func newUnregistryProxy( + ctx context.Context, remoteAddr string, dialer netproxy.ContextDialer, onError func(error), +) (*proxy.Proxy, error) { + // Test remote connectivity before creating a proxy. + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + testConn, err := dialer.DialContext(ctx, "tcp", remoteAddr) + if err != nil { + return nil, fmt.Errorf("connect to remote address '%s': %w", remoteAddr, err) + } + testConn.Close() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return nil, fmt.Errorf("listen on an available port on 127.0.0.1: %w", err) + } + + p := &proxy.Proxy{ + Listener: listener, + RemoteAddr: remoteAddr, + DialContext: dialer.DialContext, + OnError: onError, + } + + return p, nil +} + +// isDockerVirtualised checks if Docker is running in a virtualised environment like Docker Desktop on macOS. +func isDockerVirtualised(ctx context.Context, dockerCli *docker.Client) (bool, error) { + info, err := dockerCli.Info(ctx) + if err != nil { + return false, fmt.Errorf("get Docker info: %w", err) + } + + virtualisedHostnames := []string{"docker-desktop", "colima"} + for _, name := range virtualisedHostnames { + if strings.Contains(strings.ToLower(info.Name), name) { + return true, nil + } + } + + return false, nil +} + +// runDockerVMProxyContainer creates a socat container to proxy an available localhost port within the Docker VM +// (e.g. Docker Desktop on macOS) to the specified target port on the host machine. +// Returns the container ID and the localhost port the container port is bound to. +// TODO: accept custom image name. +func runDockerVMProxyContainer(ctx context.Context, dockerCli *docker.Client, targetPort int) (string, int, error) { + suffix, err := secret.RandomAlphaNumeric(4) + if err != nil { + return "", 0, fmt.Errorf("generate random suffix: %w", err) + } + containerName := fmt.Sprintf("uncloud-push-proxy-%s", suffix) + + containerPort := nat.Port("5000/tcp") + config := &container.Config{ + // TODO: make image configurable. + Image: "alpine/socat:latest", + // Reset the default entrypoint "socat". + Entrypoint: []string{}, + Cmd: []string{ + "timeout", "1800", // Auto-terminate socat after 30 minutes. + "socat", + "TCP-LISTEN:5000,fork,reuseaddr", + fmt.Sprintf("TCP-CONNECT:host.docker.internal:%d", targetPort), + }, + ExposedPorts: nat.PortSet{ + containerPort: {}, + }, + } + + // Get an available port on localhost to bind the container port to by creating a temporary listener and closing it. + // We need to explicitly specify the host port and not rely on Docker mapping because if not specified, + // 'docker push' from Docker Desktop is unable to reach the randomly mapped one for some reason. + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return "", 0, fmt.Errorf("reserve a local port: %w", err) + } + hostPort := listener.Addr().(*net.TCPAddr).Port + listener.Close() + + hostConfig := &container.HostConfig{ + AutoRemove: true, + PortBindings: nat.PortMap{ + containerPort: []nat.PortBinding{ + { + HostIP: "127.0.0.1", + HostPort: strconv.Itoa(hostPort), + }, + }, + }, + } + + resp, err := dockerCli.CreateContainerWithImagePull(ctx, containerName, config, hostConfig) + if err != nil { + return "", 0, fmt.Errorf("create socat proxy container: %w", err) + } + + cleanup := func() { + dockerCli.ContainerRemove(ctx, resp.ID, container.RemoveOptions{Force: true}) + } + + if err = dockerCli.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil { + // Clean up if start fails. + cleanup() + return "", 0, fmt.Errorf("start socat proxy container %s: %w", resp.ID, err) + } + + return resp.ID, hostPort, nil +} + +// toPushProgressEvent converts a JSON progress message from the Docker API to a progress event. +// It's based on toPushProgressEvent from Docker Compose. +func toPushProgressEvent(jm jsonmessage.JSONMessage) *progress.Event { + if jm.ID == "" || jm.Progress == nil { + return nil + } + + status := progress.Working + percent := 0 + + if jm.Progress.Total > 0 { + percent = int(jm.Progress.Current * 100 / jm.Progress.Total) + } + + switch jm.Status { + case "Pushed", "Layer already exists": + status = progress.Done + percent = 100 + } + + return &progress.Event{ + ID: jm.ID, + Current: jm.Progress.Current, + Total: jm.Progress.Total, + Percent: percent, + Text: jm.Status, + Status: status, + StatusText: jm.Progress.String(), + } +}