handle interrupts

README.md

@@ -302,14 +302,12 @@ Credit should go to all the authors and maintainers for their open source tools,
 
 <!--
 
-fix distroless kube
-
 solve the machine and add how to after docker https://github.com/juice-shop/juice-shop#docker-container
 
 * test all catalog
 * discord + social links
 * replace task/htb example with thm
-* verify/support kube config relative path
+* verify/support kube config relative path + remote cluster
 * update cloud pkg
 * update platform prs
 * verify network connectivity between boxes/tasks i.e. kube.svc
@@ -331,8 +329,7 @@ TODO
     - debug `htb-postman`
     - lab inputs
     - add flow example
-    - verify kube/cloud distroless support
-    - verify kube/cloud no-shell support
+    - verify cloud no-shell support
     - play htb: linux/win
     - RELEASE example https://github.com/boz/kail#homebrew
     - docker release and gh-action
@@ -377,7 +374,6 @@ TODO
     - print/event shared directory, same as envs, ports etc.
     - review tty resize
     - expose copy from/to ???
-    - kube: add distroless support
     - kube: verify if `close()` is needed or `return nil`
     - kube: `execBox` deployment always check/scale replica to 1 before exec (test with replica=0)
     - kube: update resources sizes + comparison

pkg/box/docker/docker.go

@@ -2,10 +2,6 @@ package docker
 
 import (
 	"fmt"
-	"os"
-	"os/signal"
-	"syscall"
-
 	"github.com/pkg/errors"
 	"golang.org/x/exp/maps"
 	"golang.org/x/exp/slices"
@@ -16,6 +12,7 @@ import (
 	commonDocker "github.com/hckops/hckctl/pkg/common/docker"
 	commonModel "github.com/hckops/hckctl/pkg/common/model"
 	"github.com/hckops/hckctl/pkg/schema"
+	"github.com/hckops/hckctl/pkg/util"
 )
 
 func newDockerBoxClient(commonOpts *boxModel.CommonBoxOptions, dockerOpts *commonModel.DockerOptions) (*DockerBoxClient, error) {
@@ -189,7 +186,6 @@ func (box *DockerBoxClient) searchBox(name string) (*boxModel.BoxInfo, error) {
 }
 
 func (box *DockerBoxClient) execBox(template *boxModel.BoxV1, info boxModel.BoxInfo, streamOpts *commonModel.StreamOptions, deleteOnExit bool) error {
-	box.eventBus.Publish(newContainerExecDockerEvent(info.Id, info.Name, template.Shell))
 
 	// attempt to restart all associated sidecars
 	sidecars, err := box.dockerCommon.SidecarList(info.Name)
@@ -283,7 +279,7 @@ func (box *DockerBoxClient) execBox(template *boxModel.BoxV1, info boxModel.BoxI
 			}
 		},
 	}
-
+	box.eventBus.Publish(newContainerExecDockerEvent(info.Id, info.Name, template.Shell))
 	return box.client.ContainerExec(execOpts)
 }
 
@@ -301,26 +297,22 @@ func (box *DockerBoxClient) publishPortInfo(networkMap map[string]boxModel.BoxPo
 func (box *DockerBoxClient) logsBox(info boxModel.BoxInfo, streamOpts *commonModel.StreamOptions, deleteOnExit bool) error {
 
 	if deleteOnExit {
-		// captures CTRL+C
-		signalChan := make(chan os.Signal, 1)
-		signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)
-		go func() {
-			<-signalChan
+		util.InterruptHandler(func() {
 			box.deleteBox(info)
-		}()
+		})
 	}
 
 	opts := &docker.ContainerLogsOpts{
 		ContainerId: info.Id,
 		OutStream:   streamOpts.Out,
-		ErrStream:   streamOpts.Err,
 		OnStreamCloseCallback: func() {
-			box.eventBus.Publish(newContainerExecExitDockerEvent(info.Id))
+			box.eventBus.Publish(newContainerLogsExitDockerEvent(info.Id))
 		},
 		OnStreamErrorCallback: func(err error) {
-			box.eventBus.Publish(newContainerExecErrorDockerEvent(info.Id, err))
+			box.eventBus.Publish(newContainerLogsErrorDockerEvent(info.Id, err))
 		},
 	}
+	box.eventBus.Publish(newContainerLogsDockerEvent(info.Id))
 	return box.client.ContainerLogs(opts)
 }
 

pkg/box/docker/events.go

@@ -84,6 +84,18 @@ func newContainerExecErrorDockerEvent(containerId string, err error) *dockerBoxE
 	return &dockerBoxEvent{kind: event.LogError, value: fmt.Sprintf("container exec error: containerId=%s error=%v", containerId, err)}
 }
 
+func newContainerLogsDockerEvent(containerId string) *dockerBoxEvent {
+	return &dockerBoxEvent{kind: event.LogDebug, value: fmt.Sprintf("container logs: containerId=%s", containerId)}
+}
+
+func newContainerLogsExitDockerEvent(containerId string) *dockerBoxEvent {
+	return &dockerBoxEvent{kind: event.LogDebug, value: fmt.Sprintf("container logs exit: containerId=%s", containerId)}
+}
+
+func newContainerLogsErrorDockerEvent(containerId string, err error) *dockerBoxEvent {
+	return &dockerBoxEvent{kind: event.LogError, value: fmt.Sprintf("container logs error: containerId=%s error=%v", containerId, err)}
+}
+
 func newContainerListDockerEvent(index int, containerName string, containerId string, healthy bool) *dockerBoxEvent {
 	return &dockerBoxEvent{kind: event.LogDebug, value: fmt.Sprintf("container list: (%d) containerName=%s containerId=%s healthy=%v", index, containerName, containerId, healthy)}
 }

pkg/box/kubernetes/events.go

@@ -92,6 +92,22 @@ func newPodExecKubeLoaderEvent() *kubeBoxEvent {
 	return &kubeBoxEvent{kind: event.LoaderStop, value: "waiting"}
 }
 
+func newPodLogsKubeEvent(namespace string, name string) *kubeBoxEvent {
+	return &kubeBoxEvent{kind: event.LogDebug, value: fmt.Sprintf("pod logs: namespace=%s name=%s", namespace, name)}
+}
+
+func newPodLogsExitKubeEvent(namespace string, name string) *kubeBoxEvent {
+	return &kubeBoxEvent{kind: event.LogDebug, value: fmt.Sprintf("pod logs exit: namespace=%s name=%s", namespace, name)}
+}
+
+func newPodLogsExitKubeConsoleEvent() *kubeBoxEvent {
+	return &kubeBoxEvent{kind: event.LoaderUpdate, value: "killing"}
+}
+
+func newPodLogsErrorKubeEvent(namespace string, name string, err error) *kubeBoxEvent {
+	return &kubeBoxEvent{kind: event.LogError, value: fmt.Sprintf("pod logs error: namespace=%s name=%s error=%v", namespace, name, err)}
+}
+
 func newPodPortForwardIgnoreKubeEvent(namespace string, name string) *kubeBoxEvent {
 	return &kubeBoxEvent{kind: event.LogWarning, value: fmt.Sprintf("pod port-forward ignored: namespace=%s name=%s", namespace, name)}
 }

pkg/box/kubernetes/kubernetes.go

@@ -204,7 +204,15 @@ func (box *KubeBoxClient) searchBox(name string) (*boxModel.BoxInfo, error) {
 }
 
 func (box *KubeBoxClient) execBox(template *boxModel.BoxV1, info *boxModel.BoxInfo, streamOpts *commonModel.StreamOptions, deleteOnExit bool) error {
-	box.eventBus.Publish(newPodExecKubeEvent(template.Name, box.clientOpts.Namespace, info.Id, template.Shell))
+
+	// TODO if BoxInfo not Healthy attempt scale 1
+
+	if template.Shell == boxModel.BoxShellNone {
+		// stop loader
+		box.eventBus.Publish(newPodExecKubeLoaderEvent())
+
+		return box.logsBox(template, info, deleteOnExit)
+	}
 
 	// exec
 	opts := &kubernetes.PodExecOpts{
@@ -226,9 +234,30 @@ func (box *KubeBoxClient) execBox(template *boxModel.BoxV1, info *boxModel.BoxIn
 		defer box.deleteBox(info.Name)
 	}
 
+	box.eventBus.Publish(newPodExecKubeEvent(template.Name, box.clientOpts.Namespace, info.Id, template.Shell))
 	return box.client.PodExecShell(opts)
 }
 
+func (box *KubeBoxClient) logsBox(template *boxModel.BoxV1, info *boxModel.BoxInfo, deleteOnExit bool) error {
+	namespace := box.clientOpts.Namespace
+
+	if deleteOnExit {
+		util.InterruptHandler(func() {
+			box.eventBus.Publish(newPodLogsExitKubeEvent(namespace, info.Id))
+			box.eventBus.Publish(newPodLogsExitKubeConsoleEvent())
+			box.deleteBox(info.Name)
+		})
+	}
+
+	opts := &kubernetes.PodLogsOpts{
+		Namespace:     namespace,
+		PodName:       info.Id,
+		ContainerName: template.MainContainerName(),
+	}
+	box.eventBus.Publish(newPodLogsKubeEvent(namespace, info.Id))
+	return box.client.PodLogsStd(opts)
+}
+
 func (box *KubeBoxClient) podPortForward(template *boxModel.BoxV1, boxInfo *boxModel.BoxInfo, isWait bool) error {
 	namespace := box.clientOpts.Namespace
 

pkg/client/docker/client.go

@@ -5,11 +5,9 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"os/signal"
 	"path/filepath"
 	"strings"
 	"sync"
-	"syscall"
 	"time"
 
 	"github.com/pkg/errors"
@@ -99,13 +97,9 @@ func (client *DockerClient) ContainerCreate(opts *ContainerCreateOpts) (string,
 	}
 
 	if opts.CaptureInterrupt {
-		// captures CTRL+C
-		signalChan := make(chan os.Signal, 1)
-		signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)
-		go func() {
-			<-signalChan
+		util.InterruptHandler(func() {
 			opts.OnContainerInterruptCallback(newContainer.ID)
-		}()
+		})
 	}
 
 	if err := opts.OnContainerCreateCallback(newContainer.ID); err != nil {
@@ -355,8 +349,9 @@ func (client *DockerClient) ContainerLogs(opts *ContainerLogsOpts) error {
 
 	var once sync.Once
 	go func() {
+		// do not copy ErrStream because it fails with "Unrecognized input header"
 		if _, err := io.Copy(opts.OutStream, outStream); err != nil {
-			opts.OnStreamErrorCallback(errors.Wrap(err, "error copy stdout and stderr docker->local"))
+			opts.OnStreamErrorCallback(errors.Wrap(err, "error copy stdout docker->local"))
 		}
 		once.Do(onStreamCloseCallback)
 	}()

pkg/client/docker/options.go

@@ -75,7 +75,6 @@ type ContainerExecOpts struct {
 type ContainerLogsOpts struct {
 	ContainerId           string
 	OutStream             io.Writer
-	ErrStream             io.Writer
 	OnStreamCloseCallback func()
 	OnStreamErrorCallback func(error)
 }

pkg/client/kubernetes/client.go

@@ -7,11 +7,9 @@ import (
 	"io"
 	"net/http"
 	"os"
-	"os/signal"
 	"path/filepath"
 	"strconv"
 	"strings"
-	"syscall"
 
 	"github.com/pkg/errors"
 
@@ -582,13 +580,9 @@ func (client *KubeClient) JobCreate(opts *JobCreateOpts) error {
 	}
 
 	if opts.CaptureInterrupt {
-		// captures CTRL+C
-		signalChan := make(chan os.Signal, 1)
-		signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)
-		go func() {
-			<-signalChan
+		util.InterruptHandler(func() {
 			opts.OnContainerInterruptCallback(job.Name)
-		}()
+		})
 	}
 
 	// blocks until the job is ready, then stop watching

pkg/task/kubernetes/events.go

@@ -40,6 +40,10 @@ func newJobDeleteKubeEvent(namespace string, name string) *kubeTaskEvent {
 	return &kubeTaskEvent{kind: event.LogInfo, value: fmt.Sprintf("job delete: namespace=%s name=%s", namespace, name)}
 }
 
+func newJobDeleteKubeConsoleEvent() *kubeTaskEvent {
+	return &kubeTaskEvent{kind: event.LoaderUpdate, value: "killing"}
+}
+
 func newPodNameKubeEvent(namespace string, name string, containerName string) *kubeTaskEvent {
 	return &kubeTaskEvent{kind: event.LogInfo, value: fmt.Sprintf("found unique pod: namespace=%s name=%s containerName=%s", namespace, name, containerName)}
 }

pkg/task/kubernetes/kubernetes.go

@@ -91,7 +91,7 @@ func (task *KubeTaskClient) runTask(opts *taskModel.RunOptions) error {
 		CaptureInterrupt: true,
 		OnContainerInterruptCallback: func(name string) {
 			// ignore error when interrupted: it will attempt to delete the job twice
-			task.eventBus.Publish(newJobDeleteKubeEvent(namespace, jobName))
+			task.eventBus.Publish(newJobDeleteKubeConsoleEvent())
 			task.client.JobDelete(namespace, name)
 		},
 		OnStatusEventCallback: func(event string) {

pkg/util/interrupt.go

@@ -0,0 +1,17 @@
+package util
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func InterruptHandler(callback func()) {
+	// captures CTRL+C
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-signalChan
+		callback()
+	}()
+}