feat: adds terminal mode to attach or exec shell on a container (#3726)

internal/agent/client.go

@@ -395,6 +395,14 @@ func (c *Client) ContainerAction(ctx context.Context, containerId string, action
 	return err
 }
 
+func (c *Client) ContainerAttach(ctx context.Context, containerId string) (io.WriteCloser, io.Reader, error) {
+	panic("not implemented")
+}
+
+func (c *Client) ContainerExec(ctx context.Context, containerId string, cmd []string) (io.WriteCloser, io.Reader, error) {
+	panic("not implemented")
+}
+
 func (c *Client) Close() error {
 	return c.conn.Close()
 }

internal/container/client.go

@@ -38,4 +38,6 @@ type Client interface {
 	Ping(context.Context) error
 	Host() Host
 	ContainerActions(ctx context.Context, action ContainerAction, containerID string) error
+	ContainerAttach(ctx context.Context, id string) (io.WriteCloser, io.Reader, error)
+	ContainerExec(ctx context.Context, id string, cmd []string) (io.WriteCloser, io.Reader, error)
 }

internal/docker/client.go

@@ -33,6 +33,10 @@ type DockerCLI interface {
 	ContainerStart(ctx context.Context, containerID string, options docker.StartOptions) error
 	ContainerStop(ctx context.Context, containerID string, options docker.StopOptions) error
 	ContainerRestart(ctx context.Context, containerID string, options docker.StopOptions) error
+	ContainerAttach(ctx context.Context, containerID string, options docker.AttachOptions) (types.HijackedResponse, error)
+	ContainerExecCreate(ctx context.Context, containerID string, options docker.ExecOptions) (docker.ExecCreateResponse, error)
+	ContainerExecAttach(ctx context.Context, execID string, config docker.ExecAttachOptions) (types.HijackedResponse, error)
+	ContainerExecResize(ctx context.Context, execID string, options docker.ResizeOptions) error
 	Info(ctx context.Context) (system.Info, error)
 }
 
@@ -297,6 +301,54 @@ func (d *DockerClient) Host() container.Host {
 	return d.host
 }
 
+func (d *DockerClient) ContainerAttach(ctx context.Context, id string) (io.WriteCloser, io.Reader, error) {
+	log.Debug().Str("id", id).Str("host", d.host.Name).Msg("Attaching to container")
+	options := docker.AttachOptions{
+		Stream: true,
+		Stdin:  true,
+		Stdout: true,
+		Stderr: true,
+	}
+
+	waiter, err := d.cli.ContainerAttach(ctx, id, options)
+
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return waiter.Conn, waiter.Reader, nil
+}
+
+func (d *DockerClient) ContainerExec(ctx context.Context, id string, cmd []string) (io.WriteCloser, io.Reader, error) {
+	log.Debug().Str("id", id).Str("host", d.host.Name).Msg("Executing command in container")
+	options := docker.ExecOptions{
+		AttachStdout: true,
+		AttachStderr: true,
+		AttachStdin:  true,
+		Cmd:          cmd,
+		Tty:          true,
+	}
+
+	execID, err := d.cli.ContainerExecCreate(ctx, id, options)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	waiter, err := d.cli.ContainerExecAttach(ctx, execID.ID, docker.ExecAttachOptions{})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if err = d.cli.ContainerExecResize(ctx, execID.ID, docker.ResizeOptions{
+		Width:  100,
+		Height: 40,
+	}); err != nil {
+		return nil, nil, err
+	}
+
+	return waiter.Conn, waiter.Reader, nil
+}
+
 func newContainer(c docker.Summary, host string) container.Container {
 	name := "no name"
 	if c.Labels["dev.dozzle.name"] != "" {

internal/k8s/client.go

@@ -241,8 +241,15 @@ func (k *K8sClient) Host() container.Host {
 }
 
 func (k *K8sClient) ContainerActions(ctx context.Context, action container.ContainerAction, containerID string) error {
-	// Implementation for container actions (start, stop, restart, etc.)
-	return nil
+	panic("not implemented")
+}
+
+func (k *K8sClient) ContainerAttach(ctx context.Context, id string) (io.WriteCloser, io.Reader, error) {
+	panic("not implemented")
+}
+
+func (k *K8sClient) ContainerExec(ctx context.Context, id string, cmd []string) (io.WriteCloser, io.Reader, error) {
+	panic("not implemented")
 }
 
 // Helper function to parse pod and container names from container ID

internal/support/container/agent_service.go

@@ -70,3 +70,11 @@ func (d *agentService) SubscribeContainersStarted(ctx context.Context, container
 func (a *agentService) ContainerAction(ctx context.Context, container container.Container, action container.ContainerAction) error {
 	return a.client.ContainerAction(ctx, container.ID, action)
 }
+
+func (a *agentService) Attach(ctx context.Context, container container.Container, stdin io.Reader, stdout io.Writer) error {
+	panic("not implemented")
+}
+
+func (a *agentService) Exec(ctx context.Context, container container.Container, cmd []string, stdin io.Reader, stdout io.Writer) error {
+	panic("not implemented")
+}

internal/support/container/client_service.go

@@ -16,13 +16,17 @@ type ClientService interface {
 	Host(ctx context.Context) (container.Host, error)
 	ContainerAction(ctx context.Context, container container.Container, action container.ContainerAction) error
 	LogsBetweenDates(ctx context.Context, container container.Container, from time.Time, to time.Time, stdTypes container.StdType) (<-chan *container.LogEvent, error)
-	RawLogs(ctx context.Context, container container.Container, from time.Time, to time.Time, stdTypes container.StdType) (io.ReadCloser, error)
+	RawLogs(context.Context, container.Container, time.Time, time.Time, container.StdType) (io.ReadCloser, error)
 
 	// Subscriptions
-	SubscribeStats(ctx context.Context, stats chan<- container.ContainerStat)
-	SubscribeEvents(ctx context.Context, events chan<- container.ContainerEvent)
-	SubscribeContainersStarted(ctx context.Context, containers chan<- container.Container)
+	SubscribeStats(context.Context, chan<- container.ContainerStat)
+	SubscribeEvents(context.Context, chan<- container.ContainerEvent)
+	SubscribeContainersStarted(context.Context, chan<- container.Container)
 
 	// Blocking streaming functions that should be used in a goroutine
-	StreamLogs(ctx context.Context, container container.Container, from time.Time, stdTypes container.StdType, events chan<- *container.LogEvent) error
+	StreamLogs(context.Context, container.Container, time.Time, container.StdType, chan<- *container.LogEvent) error
+
+	// Terminal
+	Attach(context.Context, container.Container, io.Reader, io.Writer) error
+	Exec(context.Context, container.Container, []string, io.Reader, io.Writer) error
 }

internal/support/container/container_service.go

@@ -35,3 +35,11 @@ func (c *ContainerService) StreamLogs(ctx context.Context, from time.Time, stdTy
 func (c *ContainerService) Action(ctx context.Context, action container.ContainerAction) error {
 	return c.clientService.ContainerAction(ctx, c.Container, action)
 }
+
+func (c *ContainerService) Attach(ctx context.Context, stdin io.Reader, stdout io.Writer) error {
+	return c.clientService.Attach(ctx, c.Container, stdin, stdout)
+}
+
+func (c *ContainerService) Exec(ctx context.Context, cmd []string, stdin io.Reader, stdout io.Writer) error {
+	return c.clientService.Exec(ctx, c.Container, cmd, stdin, stdout)
+}

internal/support/docker/docker_service.go

@@ -3,6 +3,7 @@ package docker_support
 import (
 	"context"
 	"io"
+	"sync"
 	"time"
 
 	"github.com/amir20/dozzle/internal/container"
@@ -109,3 +110,75 @@ func (d *DockerClientService) SubscribeEvents(ctx context.Context, events chan<-
 func (d *DockerClientService) SubscribeContainersStarted(ctx context.Context, containers chan<- container.Container) {
 	d.store.SubscribeNewContainers(ctx, containers)
 }
+
+func (d *DockerClientService) Attach(ctx context.Context, container container.Container, stdin io.Reader, stdout io.Writer) error {
+	cancelCtx, cancel := context.WithCancel(ctx)
+	containerWriter, containerReader, err := d.client.ContainerAttach(cancelCtx, container.ID)
+	if err != nil {
+		cancel()
+		return err
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		if _, err := io.Copy(containerWriter, stdin); err != nil {
+			log.Error().Err(err).Msg("error while reading from ws")
+		}
+		cancel()
+		containerWriter.Close()
+	}()
+
+	go func() {
+		defer wg.Done()
+		if container.Tty {
+			if _, err := io.Copy(stdout, containerReader); err != nil {
+				log.Error().Err(err).Msg("error while writing to ws")
+			}
+		} else {
+			if _, err := stdcopy.StdCopy(stdout, stdout, containerReader); err != nil {
+				log.Error().Err(err).Msg("error while writing to ws")
+			}
+		}
+		cancel()
+	}()
+
+	wg.Wait()
+
+	return nil
+}
+
+func (d *DockerClientService) Exec(ctx context.Context, container container.Container, cmd []string, stdin io.Reader, stdout io.Writer) error {
+	cancelCtx, cancel := context.WithCancel(ctx)
+	containerWriter, containerReader, err := d.client.ContainerExec(cancelCtx, container.ID, cmd)
+	if err != nil {
+		cancel()
+		return err
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		if _, err := io.Copy(containerWriter, stdin); err != nil {
+			log.Error().Err(err).Msg("error while reading from ws")
+		}
+		cancel()
+		containerWriter.Close()
+	}()
+
+	go func() {
+		defer wg.Done()
+		if _, err := stdcopy.StdCopy(stdout, stdout, containerReader); err != nil {
+			log.Error().Err(err).Msg("error while writing to ws")
+		}
+		cancel()
+	}()
+
+	wg.Wait()
+
+	return nil
+}

internal/support/k8s/k8s_service.go

@@ -90,3 +90,11 @@ func (k *K8sClientService) SubscribeEvents(ctx context.Context, events chan<- co
 func (k *K8sClientService) SubscribeContainersStarted(ctx context.Context, containers chan<- container.Container) {
 	k.store.SubscribeNewContainers(ctx, containers)
 }
+
+func (k *K8sClientService) Attach(ctx context.Context, container container.Container, stdin io.Reader, stdout io.Writer) error {
+	panic("not implemented")
+}
+
+func (k *K8sClientService) Exec(ctx context.Context, container container.Container, cmd []string, stdin io.Reader, stdout io.Writer) error {
+	panic("not implemented")
+}

internal/web/routes.go

@@ -107,6 +107,8 @@ func createRouter(h *handler) *chi.Mux {
 				r.Get("/hosts/{host}/containers/{id}/logs/stream", h.streamContainerLogs)
 				r.Get("/hosts/{host}/logs/stream", h.streamHostLogs)
 				r.Get("/hosts/{host}/containers/{id}/logs", h.fetchLogsBetweenDates)
+				r.Get("/hosts/{host}/containers/{id}/attach", h.attach)
+				r.Get("/hosts/{host}/containers/{id}/exec", h.exec)
 				r.Get("/hosts/{host}/logs/mergedStream/{ids}", h.streamLogsMerged)
 				r.Get("/containers/{hostIds}/download", h.downloadLogs) // formatted as host:container,host:container
 				r.Get("/stacks/{stack}/logs/stream", h.streamStackLogs)

internal/web/terminal.go

@@ -0,0 +1,120 @@
+package web
+
+import (
+	"net/http"
+
+	"github.com/amir20/dozzle/internal/auth"
+	"github.com/go-chi/chi/v5"
+	"github.com/gorilla/websocket"
+	"github.com/rs/zerolog/log"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
+
+func (h *handler) attach(w http.ResponseWriter, r *http.Request) {
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Error().Err(err).Msg("error while trying to upgrade connection")
+		return
+	}
+	defer conn.Close()
+
+	id := chi.URLParam(r, "id")
+	userLabels := h.config.Labels
+	if h.config.Authorization.Provider != NONE {
+		user := auth.UserFromContext(r.Context())
+		if user.ContainerLabels.Exists() {
+			userLabels = user.ContainerLabels
+		}
+	}
+
+	containerService, err := h.hostService.FindContainer(hostKey(r), id, userLabels)
+
+	if err != nil {
+		log.Error().Err(err).Msg("error while trying to find container")
+		return
+	}
+
+	wsReader := &webSocketReader{conn: conn}
+	wsWriter := &webSocketWriter{conn: conn}
+	if err = containerService.Attach(r.Context(), wsReader, wsWriter); err != nil {
+		log.Error().Err(err).Msg("error while trying to attach to container")
+		conn.WriteMessage(websocket.TextMessage, []byte("🚨 Error while trying to attach to container\r\n"))
+		return
+	}
+}
+
+func (h *handler) exec(w http.ResponseWriter, r *http.Request) {
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Error().Err(err).Msg("error while trying to upgrade connection")
+		return
+	}
+	defer conn.Close()
+
+	id := chi.URLParam(r, "id")
+	userLabels := h.config.Labels
+	if h.config.Authorization.Provider != NONE {
+		user := auth.UserFromContext(r.Context())
+		if user.ContainerLabels.Exists() {
+			userLabels = user.ContainerLabels
+		}
+	}
+
+	containerService, err := h.hostService.FindContainer(hostKey(r), id, userLabels)
+	if err != nil {
+		log.Error().Err(err).Msg("error while trying to find container")
+		return
+	}
+
+	wsReader := &webSocketReader{conn: conn}
+	wsWriter := &webSocketWriter{conn: conn}
+	if err = containerService.Exec(r.Context(), []string{"sh", "-c", "command -v bash >/dev/null 2>&1 && exec bash || exec sh"}, wsReader, wsWriter); err != nil {
+		log.Error().Err(err).Msg("error while trying to attach to container")
+		conn.WriteMessage(websocket.TextMessage, []byte("🚨 Error while trying to attach to container\r\n"))
+		return
+	}
+}
+
+type webSocketWriter struct {
+	conn *websocket.Conn
+}
+
+func (w *webSocketWriter) Write(p []byte) (int, error) {
+	err := w.conn.WriteMessage(websocket.TextMessage, p)
+	return len(p), err
+}
+
+type webSocketReader struct {
+	conn   *websocket.Conn
+	buffer []byte
+}
+
+func (r *webSocketReader) Read(p []byte) (n int, err error) {
+	if len(r.buffer) > 0 {
+		n = copy(p, r.buffer)
+		r.buffer = r.buffer[n:]
+		return n, nil
+	}
+
+	// Otherwise, read a new message
+	_, message, err := r.conn.ReadMessage()
+	if err != nil {
+		return 0, err
+	}
+
+	n = copy(p, message)
+
+	// If we couldn't copy the entire message, store the rest in our buffer
+	if n < len(message) {
+		r.buffer = message[n:]
+	}
+
+	return n, nil
+}