runtime: add debug console service

Add `kata-runtime exec` to enter guest OS
through shell started by agent

Fixes: #245

Signed-off-by: bin liu <bin@hyper.sh>

src/runtime/cli/config-generated.go.in

@@ -39,6 +39,7 @@ var version = "@VERSION@"
 // project-specific command names
 var envCmd = fmt.Sprintf("%s-env", projectPrefix)
 var checkCmd = fmt.Sprintf("%s-check", projectPrefix)
+var execCmd = "exec"
 
 // project-specific option names
 var configFilePathOption = fmt.Sprintf("%s-config", projectPrefix)

src/runtime/cli/kata-exec.go

@@ -0,0 +1,208 @@
+// Copyright (c) 2017-2019 Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package main
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+
+	"sync"
+	"time"
+
+	"github.com/containerd/console"
+	"github.com/kata-containers/kata-containers/src/runtime/pkg/katautils"
+	clientUtils "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/agent/protocols/client"
+	"github.com/pkg/errors"
+	"github.com/urfave/cli"
+)
+
+const (
+
+	// The buffer size used to specify the buffer for IO streams copy
+	bufSize = 32 << 10
+
+	defaultTimeout = 3 * time.Second
+)
+
+var (
+	bufPool = sync.Pool{
+		New: func() interface{} {
+			buffer := make([]byte, bufSize)
+			return &buffer
+		},
+	}
+)
+
+var kataExecCLICommand = cli.Command{
+	Name:  execCmd,
+	Usage: "Enter into guest by debug console",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "monitor-addr",
+			Usage: "Kata monitor listen address.",
+		},
+		cli.Uint64Flag{
+			Name:  "debug-port",
+			Usage: "Port that debug console is listening on.",
+		},
+	},
+	Action: func(context *cli.Context) error {
+		ctx, err := cliContextToContext(context)
+		if err != nil {
+			return err
+		}
+		span, _ := katautils.Trace(ctx, "exec")
+		defer span.Finish()
+
+		endPoint := context.String("monitor-addr")
+		if endPoint == "" {
+			endPoint = "http://localhost:8090"
+		}
+
+		port := context.Uint64("debug-port")
+		if port == 0 {
+			port = 1026
+		}
+
+		sandboxID := context.Args().Get(0)
+		if sandboxID == "" {
+			return fmt.Errorf("SandboxID not found")
+		}
+
+		conn, err := getConn(endPoint, sandboxID, port)
+		if err != nil {
+			return err
+		}
+		defer conn.Close()
+
+		con := console.Current()
+		defer con.Reset()
+
+		if err := con.SetRaw(); err != nil {
+			return err
+		}
+
+		iostream := &iostream{
+			conn:   conn,
+			exitch: make(chan struct{}),
+			closed: false,
+		}
+
+		ioCopy(iostream, con)
+
+		<-iostream.exitch
+		return nil
+	},
+}
+
+func ioCopy(stream *iostream, con console.Console) {
+	var wg sync.WaitGroup
+
+	// stdin
+	go func() {
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+		io.CopyBuffer(stream, con, *p)
+	}()
+
+	// stdout
+	wg.Add(1)
+	go func() {
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+		io.CopyBuffer(os.Stdout, stream, *p)
+		wg.Done()
+	}()
+
+	wg.Wait()
+	close(stream.exitch)
+}
+
+type iostream struct {
+	conn   net.Conn
+	exitch chan struct{}
+	closed bool
+}
+
+func (s *iostream) Write(data []byte) (n int, err error) {
+	if s.closed {
+		return 0, errors.New("stream closed")
+	}
+	return s.conn.Write(data)
+}
+
+func (s *iostream) Close() error {
+	if s.closed {
+		return errors.New("stream closed")
+	}
+
+	err := s.conn.Close()
+	if err == nil {
+		s.closed = true
+	}
+
+	return err
+}
+
+func (s *iostream) Read(data []byte) (n int, err error) {
+	if s.closed {
+		return 0, errors.New("stream closed")
+	}
+
+	return s.conn.Read(data)
+}
+
+func getConn(endPoint, sandboxID string, port uint64) (net.Conn, error) {
+	shimURL := fmt.Sprintf("%s/agent-url?sandbox=%s", endPoint, sandboxID)
+	resp, err := http.Get(shimURL)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("Failed to get %s: %d", shimURL, resp.StatusCode)
+	}
+
+	defer resp.Body.Close()
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	sock := strings.TrimSuffix(string(data), "\n")
+	addr, err := url.Parse(sock)
+	if err != nil {
+		return nil, err
+	}
+
+	// validate more
+	switch addr.Scheme {
+	case clientUtils.VSockSocketScheme:
+		// vsock://31513974:1024
+		shimAddr := clientUtils.VSockSocketScheme + ":" + addr.Host
+		shimAddr = strings.Replace(shimAddr, ":1024", fmt.Sprintf(":%d", port), -1)
+		return clientUtils.VsockDialer(shimAddr, defaultTimeout)
+
+	case clientUtils.HybridVSockScheme:
+		// addr: hvsock:///run/vc/firecracker/340b412c97bf1375cdda56bfa8f18c8a/root/kata.hvsock:1024
+		hvsocket := strings.Split(addr.Path, ":")
+		if len(hvsocket) != 2 {
+			return nil, fmt.Errorf("Invalid hybrid vsock scheme: %s", sock)
+		}
+
+		// hvsock:///run/vc/firecracker/340b412c97bf1375cdda56bfa8f18c8a/root/kata.hvsock
+		shimAddr := fmt.Sprintf("%s:%s:%d", clientUtils.HybridVSockScheme, hvsocket[0], port)
+		return clientUtils.HybridVSockDialer(shimAddr, defaultTimeout)
+	}
+
+	return nil, fmt.Errorf("schema %s not found", addr.Scheme)
+}

src/runtime/cli/kata-monitor/main.go

@@ -36,6 +36,7 @@ func main() {
 	m := http.NewServeMux()
 	m.Handle("/metrics", http.HandlerFunc(km.ProcessMetricsRequest))
 	m.Handle("/sandboxes", http.HandlerFunc(km.ListSandboxes))
+	m.Handle("/agent-url", http.HandlerFunc(km.GetAgentURL))
 
 	// for debug shim process
 	m.Handle("/debug/vars", http.HandlerFunc(km.ExpvarHandler))

src/runtime/cli/main.go

@@ -125,6 +125,7 @@ var runtimeCommands = []cli.Command{
 	// Kata Containers specific extensions
 	kataCheckCLICommand,
 	kataEnvCLICommand,
+	kataExecCLICommand,
 	factoryCLICommand,
 }
 

src/runtime/containerd-shim-v2/shim_management.go

@@ -8,6 +8,7 @@ package containerdshim
 import (
 	"context"
 	"expvar"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/pprof"
@@ -34,6 +35,18 @@ var (
 	shimMgtLog               = shimLog.WithField("subsystem", "shim-management")
 )
 
+// agentURL returns URL for agent
+func (s *service) agentURL(w http.ResponseWriter, r *http.Request) {
+	url, err := s.sandbox.GetAgentURL()
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(err.Error()))
+		return
+	}
+
+	fmt.Fprint(w, url)
+}
+
 // serveMetrics handle /metrics requests
 func (s *service) serveMetrics(w http.ResponseWriter, r *http.Request) {
 
@@ -139,6 +152,7 @@ func (s *service) startManagementServer(ctx context.Context, ociSpec *specs.Spec
 	// bind hanlder
 	m := http.NewServeMux()
 	m.Handle("/metrics", http.HandlerFunc(s.serveMetrics))
+	m.Handle("/agent-url", http.HandlerFunc(s.agentURL))
 	s.mountPprofHandle(m, ociSpec)
 
 	// register shim metrics

src/runtime/pkg/kata-monitor/metrics.go

@@ -10,7 +10,6 @@ import (
 	"compress/gzip"
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"path/filepath"
 	"sort"
@@ -236,33 +235,7 @@ func (km *KataMonitor) aggregateSandboxMetrics(encoder expfmt.Encoder) error {
 
 // getSandboxMetrics will get sandbox's metrics from shim
 func (km *KataMonitor) getSandboxMetrics(sandboxID, namespace string) ([]*dto.MetricFamily, error) {
-	socket, err := km.getMonitorAddress(sandboxID, namespace)
-	if err != nil {
-		return nil, err
-	}
-
-	transport := &http.Transport{
-		DisableKeepAlives: true,
-		Dial: func(proto, addr string) (conn net.Conn, err error) {
-			return net.Dial("unix", "\x00"+socket)
-		},
-	}
-
-	client := http.Client{
-		Timeout:   3 * time.Second,
-		Transport: transport,
-	}
-
-	resp, err := client.Get("http://shim/metrics")
-	if err != nil {
-		return nil, err
-	}
-
-	defer func() {
-		resp.Body.Close()
-	}()
-
-	body, err := ioutil.ReadAll(resp.Body)
+	body, err := km.doGet(sandboxID, namespace, defaultTimeout, "metrics")
 	if err != nil {
 		return nil, err
 	}

src/runtime/pkg/kata-monitor/monitor.go

@@ -80,6 +80,28 @@ func (km *KataMonitor) initSandboxCache() error {
 	return nil
 }
 
+// GetAgentURL returns agent URL
+func (km *KataMonitor) GetAgentURL(w http.ResponseWriter, r *http.Request) {
+	sandboxID, err := getSandboxIdFromReq(r)
+	if err != nil {
+		commonServeError(w, http.StatusBadRequest, err)
+		return
+	}
+	namespace, err := km.getSandboxNamespace(sandboxID)
+	if err != nil {
+		commonServeError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	data, err := km.doGet(sandboxID, namespace, defaultTimeout, "agent-url")
+	if err != nil {
+		commonServeError(w, http.StatusBadRequest, err)
+		return
+	}
+
+	fmt.Fprintln(w, string(data))
+}
+
 // ListSandboxes list all sandboxes running in Kata
 func (km *KataMonitor) ListSandboxes(w http.ResponseWriter, r *http.Request) {
 	sandboxes := km.getSandboxList()

src/runtime/pkg/kata-monitor/pprof.go

@@ -12,14 +12,6 @@ import (
 	"net/http"
 )
 
-func getSandboxIdFromReq(r *http.Request) (string, error) {
-	sandbox := r.URL.Query().Get("sandbox")
-	if sandbox != "" {
-		return sandbox, nil
-	}
-	return "", fmt.Errorf("sandbox not found in %+v", r.URL.Query())
-}
-
 func serveError(w http.ResponseWriter, status int, txt string) {
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
 	w.Header().Set("X-Go-Pprof", "1")

src/runtime/pkg/kata-monitor/shim_client.go

@@ -0,0 +1,81 @@
+// Copyright (c) 2020 Ant Financial
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package katamonitor
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"time"
+)
+
+const (
+	defaultTimeout = 3 * time.Second
+)
+
+func commonServeError(w http.ResponseWriter, status int, err error) {
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	w.WriteHeader(status)
+	if err != nil {
+		fmt.Fprintln(w, err.Error())
+	}
+}
+
+func getSandboxIdFromReq(r *http.Request) (string, error) {
+	sandbox := r.URL.Query().Get("sandbox")
+	if sandbox != "" {
+		return sandbox, nil
+	}
+	return "", fmt.Errorf("sandbox not found in %+v", r.URL.Query())
+}
+
+func (km *KataMonitor) buildShimClient(sandboxID, namespace string, timeout time.Duration) (*http.Client, error) {
+	socket, err := km.getMonitorAddress(sandboxID, namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	transport := &http.Transport{
+		DisableKeepAlives: true,
+		Dial: func(proto, addr string) (conn net.Conn, err error) {
+			return net.Dial("unix", "\x00"+socket)
+		},
+	}
+
+	client := &http.Client{
+		Transport: transport,
+	}
+
+	if timeout > 0 {
+		client.Timeout = timeout
+	}
+
+	return client, nil
+}
+
+func (km *KataMonitor) doGet(sandboxID, namespace string, timeoutInSeconds time.Duration, urlPath string) ([]byte, error) {
+	client, err := km.buildShimClient(sandboxID, namespace, timeoutInSeconds)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := client.Get(fmt.Sprintf("http://shim/%s", urlPath))
+	if err != nil {
+		return nil, err
+	}
+
+	defer func() {
+		resp.Body.Close()
+	}()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	return body, nil
+}

src/runtime/virtcontainers/interfaces.go

@@ -75,6 +75,7 @@ type VCSandbox interface {
 
 	UpdateRuntimeMetrics() error
 	GetAgentMetrics() (string, error)
+	GetAgentURL() (string, error)
 }
 
 // VCContainer is the Container interface

src/runtime/virtcontainers/pkg/agent/protocols/client/client.go

@@ -178,7 +178,7 @@ func parse(sock string) (string, *url.URL, error) {
 func agentDialer(addr *url.URL) dialer {
 	switch addr.Scheme {
 	case VSockSocketScheme:
-		return vsockDialer
+		return VsockDialer
 	case HybridVSockScheme:
 		return HybridVSockDialer
 	case MockHybridVSockScheme:
@@ -278,7 +278,7 @@ func commonDialer(timeout time.Duration, dialFunc func() (net.Conn, error), time
 	return conn, nil
 }
 
-func vsockDialer(sock string, timeout time.Duration) (net.Conn, error) {
+func VsockDialer(sock string, timeout time.Duration) (net.Conn, error) {
 	cid, port, err := parseGrpcVsockAddr(sock)
 	if err != nil {
 		return nil, err

src/runtime/virtcontainers/pkg/annotations/annotations.go

@@ -245,7 +245,7 @@ const (
 	// The following example can be used to load two kernel modules with parameters
 	///
 	//   annotations:
-	//     io.kata-containers.config.agent.kernel_modules: "e1000e InterruptThrottleRate=3000,3000,3000 EEE=1; i915 enable_ppgtt=0"
+	//     io.katacontainers.config.agent.kernel_modules: "e1000e InterruptThrottleRate=3000,3000,3000 EEE=1; i915 enable_ppgtt=0"
 	//
 	// The first word is considered as the module name and the rest as its parameters.
 	//

src/runtime/virtcontainers/pkg/vcmock/sandbox.go

@@ -247,3 +247,10 @@ func (s *Sandbox) Stats() (vc.SandboxStats, error) {
 	}
 	return vc.SandboxStats{}, nil
 }
+
+func (s *Sandbox) GetAgentURL() (string, error) {
+	if s.GetAgentURLFunc != nil {
+		return s.GetAgentURLFunc()
+	}
+	return "", nil
+}

src/runtime/virtcontainers/pkg/vcmock/types.go

@@ -67,6 +67,7 @@ type Sandbox struct {
 	UpdateRuntimeMetricsFunc func() error
 	GetAgentMetricsFunc      func() (string, error)
 	StatsFunc                func() (vc.SandboxStats, error)
+	GetAgentURLFunc          func() (string, error)
 }
 
 // Container is a fake Container type used for testing

src/runtime/virtcontainers/sandbox.go

@@ -2270,3 +2270,7 @@ func (s *Sandbox) GetPatchedOCISpec() *specs.Spec {
 func (s *Sandbox) GetOOMEvent() (string, error) {
 	return s.agent.getOOMEvent()
 }
+
+func (s *Sandbox) GetAgentURL() (string, error) {
+	return s.agent.getAgentURL()
+}