console-watcher: use console watcher to monitor guest console outputs

Import new console watcher to monitor guest console outputs, and will be
only effective when we turn on enable_debug option.
Guest console outputs may include guest kernel debug info, agent debug info,
etc.

Fixes: #389

Signed-off-by: Penny Zheng penny.zheng@arm.com

src/runtime/virtcontainers/acrn.go

@@ -254,7 +254,7 @@ func (a *Acrn) buildDevices(imagePath string) ([]Device, error) {
 		return nil, fmt.Errorf("Image Path should not be empty: %s", imagePath)
 	}
 
-	console, err := a.getSandboxConsole(a.id)
+	_, console, err := a.getSandboxConsole(a.id)
 	if err != nil {
 		return nil, err
 	}
@@ -643,11 +643,16 @@ func (a *Acrn) addDevice(devInfo interface{}, devType deviceType) error {
 
 // getSandboxConsole builds the path of the console where we can read
 // logs coming from the sandbox.
-func (a *Acrn) getSandboxConsole(id string) (string, error) {
+func (a *Acrn) getSandboxConsole(id string) (string, string, error) {
 	span, _ := a.trace("getSandboxConsole")
 	defer span.Finish()
 
-	return utils.BuildSocketPath(a.store.RunVMStoragePath(), id, acrnConsoleSocket)
+	consoleURL, err := utils.BuildSocketPath(a.store.RunVMStoragePath(), id, acrnConsoleSocket)
+	if err != nil {
+		return consoleProtoUnix, "", err
+	}
+
+	return consoleProtoUnix, consoleURL, nil
 }
 
 func (a *Acrn) saveSandbox() error {

src/runtime/virtcontainers/acrn_test.go

@@ -205,9 +205,10 @@ func TestAcrnGetSandboxConsole(t *testing.T) {
 	sandboxID := "testSandboxID"
 	expected := filepath.Join(a.store.RunVMStoragePath(), sandboxID, consoleSocket)
 
-	result, err := a.getSandboxConsole(sandboxID)
+	proto, result, err := a.getSandboxConsole(sandboxID)
 	assert.NoError(err)
 	assert.Equal(result, expected)
+	assert.Equal(proto, consoleProtoUnix)
 }
 
 func TestAcrnCreateSandbox(t *testing.T) {

src/runtime/virtcontainers/clh.go

@@ -9,8 +9,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -21,6 +19,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/containerd/console"
 	persistapi "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/persist/api"
 	chclient "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/cloud-hypervisor/client"
 	"github.com/opencontainers/selinux/go-selinux/label"
@@ -123,6 +122,7 @@ type cloudHypervisor struct {
 	vmconfig  chclient.VmConfig
 	virtiofsd Virtiofsd
 	store     persistapi.PersistDriver
+	console   console.Console
 }
 
 var clhKernelParams = []Param{
@@ -357,13 +357,12 @@ func (clh *cloudHypervisor) startSandbox(timeout int) error {
 		return errors.New("cloud-hypervisor only supports virtio based file sharing")
 	}
 
-	var strErr string
-	strErr, pid, err := clh.LaunchClh()
+	pid, err := clh.LaunchClh()
 	if err != nil {
 		if shutdownErr := clh.virtiofsd.Stop(); shutdownErr != nil {
 			clh.Logger().WithField("error", shutdownErr).Warn("error shutting down Virtiofsd")
 		}
-		return fmt.Errorf("failed to launch cloud-hypervisor: %q, hypervisor output:\n%s", err, strErr)
+		return fmt.Errorf("failed to launch cloud-hypervisor: %q", err)
 	}
 	clh.state.PID = pid
 
@@ -377,9 +376,16 @@ func (clh *cloudHypervisor) startSandbox(timeout int) error {
 
 // getSandboxConsole builds the path of the console where we can read
 // logs coming from the sandbox.
-func (clh *cloudHypervisor) getSandboxConsole(id string) (string, error) {
+func (clh *cloudHypervisor) getSandboxConsole(id string) (string, string, error) {
 	clh.Logger().WithField("function", "getSandboxConsole").WithField("id", id).Info("Get Sandbox Console")
-	return "", nil
+	master, slave, err := console.NewPty()
+	if err != nil {
+		clh.Logger().Debugf("Error create pseudo tty: %v", err)
+		return consoleProtoPty, "", err
+	}
+	clh.console = master
+
+	return consoleProtoPty, slave, nil
 }
 
 func (clh *cloudHypervisor) disconnect() {
@@ -885,11 +891,11 @@ func (clh *cloudHypervisor) getAvailableVersion() error {
 
 }
 
-func (clh *cloudHypervisor) LaunchClh() (string, int, error) {
+func (clh *cloudHypervisor) LaunchClh() (int, error) {
 
 	clhPath, err := clh.clhPath()
 	if err != nil {
-		return "", -1, err
+		return -1, err
 	}
 
 	args := []string{cscAPIsocket, clh.state.apiSocket}
@@ -913,14 +919,12 @@ func (clh *cloudHypervisor) LaunchClh() (string, int, error) {
 	clh.Logger().WithField("args", strings.Join(args, " ")).Info()
 
 	cmdHypervisor := exec.Command(clhPath, args...)
-	var hypervisorOutput io.ReadCloser
 	if clh.config.Debug {
 		cmdHypervisor.Env = os.Environ()
 		cmdHypervisor.Env = append(cmdHypervisor.Env, "RUST_BACKTRACE=full")
-		// Get  StdoutPipe only for debug, without debug golang will redirect to /dev/null
-		hypervisorOutput, err = cmdHypervisor.StdoutPipe()
-		if err != nil {
-			return "", -1, err
+		if clh.console != nil {
+			cmdHypervisor.Stderr = clh.console
+			cmdHypervisor.Stdout = clh.console
 		}
 	}
 
@@ -928,37 +932,15 @@ func (clh *cloudHypervisor) LaunchClh() (string, int, error) {
 
 	err = utils.StartCmd(cmdHypervisor)
 	if err != nil {
-		return "", -1, err
+		return -1, err
 	}
 
 	if err := clh.waitVMM(clhTimeout); err != nil {
 		clh.Logger().WithField("error", err).Warn("cloud-hypervisor init failed")
-		var output string
-
-		if hypervisorOutput != nil {
-			b, errRead := ioutil.ReadAll(hypervisorOutput)
-			if errRead != nil {
-				output = "failed to read hypervisor output to get error information"
-			} else {
-				output = string(b)
-			}
-		} else {
-			output = "Please enable hypervisor logging to get stdout information"
+		return -1, err
 	}
 
-		return output, -1, err
-	}
-
-	if clh.config.Debug {
-		cmdLogger := utils.NewProgramLogger("kata-hypervisor")
-		clh.Logger().Debugf("Starting process logger(%s) for hypervisor", cmdLogger)
-		if err := cmdLogger.StartLogger(hypervisorOutput); err != nil {
-			// Not critical to run a container, but output wont be logged
-			clh.Logger().Warnf("Failed start process logger(%s) %s", cmdLogger, err)
-		}
-	}
-
-	return "", cmdHypervisor.Process.Pid, nil
+	return cmdHypervisor.Process.Pid, nil
 }
 
 //###########################################################################

src/runtime/virtcontainers/fc.go

@@ -10,7 +10,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -25,7 +24,6 @@ import (
 
 	"github.com/kata-containers/kata-containers/src/runtime/virtcontainers/device/config"
 	persistapi "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/persist/api"
-	kataclient "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/agent/protocols/client"
 	"github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/firecracker/client"
 	models "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/firecracker/client/models"
 	ops "github.com/kata-containers/kata-containers/src/runtime/virtcontainers/pkg/firecracker/client/operations"
@@ -154,6 +152,8 @@ type firecracker struct {
 
 	fcConfigPath string
 	fcConfig     *types.FcConfig // Parameters configured before VM starts
+
+	console console.Console
 }
 
 type firecrackerDevice struct {
@@ -398,13 +398,8 @@ func (fc *firecracker) fcInit(timeout int) error {
 	}
 
 	if fc.config.Debug {
-		stdin, err := fc.watchConsole()
-		if err != nil {
-			return err
-		}
-
-		cmd.Stderr = stdin
-		cmd.Stdout = stdin
+		cmd.Stderr = fc.console
+		cmd.Stdout = fc.console
 	}
 
 	fc.Logger().WithField("hypervisor args", args).Debug()
@@ -1101,8 +1096,15 @@ func (fc *firecracker) hotplugRemoveDevice(devInfo interface{}, devType deviceTy
 
 // getSandboxConsole builds the path of the console where we can read
 // logs coming from the sandbox.
-func (fc *firecracker) getSandboxConsole(id string) (string, error) {
-	return fmt.Sprintf("%s://%s:%d", kataclient.HybridVSockScheme, filepath.Join(fc.jailerRoot, defaultHybridVSocketName), vSockLogsPort), nil
+func (fc *firecracker) getSandboxConsole(id string) (string, string, error) {
+	master, slave, err := console.NewPty()
+	if err != nil {
+		fc.Logger().Debugf("Error create pseudo tty: %v", err)
+		return consoleProtoPty, "", err
+	}
+	fc.console = master
+
+	return consoleProtoPty, slave, nil
 }
 
 func (fc *firecracker) disconnect() {
@@ -1214,40 +1216,6 @@ func (fc *firecracker) generateSocket(id string) (interface{}, error) {
 	}, nil
 }
 
-func (fc *firecracker) watchConsole() (*os.File, error) {
-	master, slave, err := console.NewPty()
-	if err != nil {
-		fc.Logger().WithField("Error create pseudo tty", err).Debug()
-		return nil, err
-	}
-
-	stdio, err := os.OpenFile(slave, syscall.O_RDWR, 0700)
-	if err != nil {
-		fc.Logger().WithError(err).Debugf("open pseudo tty %s", slave)
-		return nil, err
-	}
-
-	go func() {
-		scanner := bufio.NewScanner(master)
-		for scanner.Scan() {
-			fc.Logger().WithFields(logrus.Fields{
-				"sandbox":   fc.id,
-				"vmconsole": scanner.Text(),
-			}).Infof("reading guest console")
-		}
-
-		if err := scanner.Err(); err != nil {
-			if err == io.EOF {
-				fc.Logger().Info("console watcher quits")
-			} else {
-				fc.Logger().WithError(err).Error("Failed to read guest console")
-			}
-		}
-	}()
-
-	return stdio, nil
-}
-
 func (fc *firecracker) isRateLimiterBuiltin() bool {
 	return true
 }

src/runtime/virtcontainers/hypervisor.go

@@ -764,7 +764,7 @@ type hypervisor interface {
 	hotplugRemoveDevice(devInfo interface{}, devType deviceType) (interface{}, error)
 	resizeMemory(memMB uint32, memoryBlockSizeMB uint32, probe bool) (uint32, memoryDevice, error)
 	resizeVCPUs(vcpus uint32) (uint32, uint32, error)
-	getSandboxConsole(sandboxID string) (string, error)
+	getSandboxConsole(sandboxID string) (string, string, error)
 	disconnect()
 	capabilities() types.Capabilities
 	hypervisorConfig() HypervisorConfig

src/runtime/virtcontainers/mock_hypervisor.go

@@ -82,8 +82,8 @@ func (m *mockHypervisor) hotplugRemoveDevice(devInfo interface{}, devType device
 	return nil, nil
 }
 
-func (m *mockHypervisor) getSandboxConsole(sandboxID string) (string, error) {
-	return "", nil
+func (m *mockHypervisor) getSandboxConsole(sandboxID string) (string, string, error) {
+	return "", "", nil
 }
 
 func (m *mockHypervisor) resizeMemory(memMB uint32, memorySectionSizeMB uint32, probe bool) (uint32, memoryDevice, error) {

src/runtime/virtcontainers/mock_hypervisor_test.go

@@ -66,9 +66,11 @@ func TestMockHypervisorGetSandboxConsole(t *testing.T) {
 	var m *mockHypervisor
 
 	expected := ""
-	result, err := m.getSandboxConsole("testSandboxID")
+	expectedProto := ""
+	proto, result, err := m.getSandboxConsole("testSandboxID")
 	assert.NoError(t, err)
 	assert.Equal(t, result, expected)
+	assert.Equal(t, proto, expectedProto)
 }
 
 func TestMockHypervisorSaveSandbox(t *testing.T) {

src/runtime/virtcontainers/qemu.go

@@ -380,7 +380,7 @@ func (q *qemu) createQmpSocket() ([]govmmQemu.QMPSocket, error) {
 func (q *qemu) buildDevices(initrdPath string) ([]govmmQemu.Device, *govmmQemu.IOThread, error) {
 	var devices []govmmQemu.Device
 
-	console, err := q.getSandboxConsole(q.id)
+	_, console, err := q.getSandboxConsole(q.id)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -1764,11 +1764,16 @@ func (q *qemu) addDevice(devInfo interface{}, devType deviceType) error {
 
 // getSandboxConsole builds the path of the console where we can read
 // logs coming from the sandbox.
-func (q *qemu) getSandboxConsole(id string) (string, error) {
+func (q *qemu) getSandboxConsole(id string) (string, string, error) {
 	span, _ := q.trace("getSandboxConsole")
 	defer span.Finish()
 
-	return utils.BuildSocketPath(q.store.RunVMStoragePath(), id, consoleSocket)
+	consoleURL, err := utils.BuildSocketPath(q.store.RunVMStoragePath(), id, consoleSocket)
+	if err != nil {
+		return consoleProtoUnix, "", err
+	}
+
+	return consoleProtoUnix, consoleURL, nil
 }
 
 func (q *qemu) saveSandbox() error {

src/runtime/virtcontainers/qemu_test.go

@@ -314,9 +314,10 @@ func TestQemuGetSandboxConsole(t *testing.T) {
 	sandboxID := "testSandboxID"
 	expected := filepath.Join(q.store.RunVMStoragePath(), sandboxID, consoleSocket)
 
-	result, err := q.getSandboxConsole(sandboxID)
+	proto, result, err := q.getSandboxConsole(sandboxID)
 	assert.NoError(err)
 	assert.Equal(result, expected)
+	assert.Equal(proto, consoleProtoUnix)
 }
 
 func TestQemuCapabilities(t *testing.T) {

src/runtime/virtcontainers/sandbox.go

@@ -6,6 +6,7 @@
 package virtcontainers
 
 import (
+	"bufio"
 	"context"
 	"fmt"
 	"io"
@@ -189,6 +190,8 @@ type Sandbox struct {
 	cgroupMgr *vccgroups.Manager
 
 	ctx context.Context
+
+	cw *consoleWatcher
 }
 
 // ID returns the sandbox identifier string.
@@ -929,6 +932,103 @@ func (s *Sandbox) ListRoutes() ([]*pbTypes.Route, error) {
 	return s.agent.listRoutes()
 }
 
+const (
+	// unix socket type of console
+	consoleProtoUnix = "unix"
+
+	// pty type of console.
+	consoleProtoPty = "pty"
+)
+
+// console watcher is designed to monitor guest console output.
+type consoleWatcher struct {
+	proto      string
+	consoleURL string
+	conn       net.Conn
+	ptyConsole *os.File
+}
+
+func newConsoleWatcher(s *Sandbox) (*consoleWatcher, error) {
+	var (
+		err error
+		cw  consoleWatcher
+	)
+
+	cw.proto, cw.consoleURL, err = s.hypervisor.getSandboxConsole(s.id)
+	if err != nil {
+		return nil, err
+	}
+
+	return &cw, nil
+}
+
+// start the console watcher
+func (cw *consoleWatcher) start(s *Sandbox) (err error) {
+	if cw.consoleWatched() {
+		return fmt.Errorf("console watcher has already watched for sandbox %s", s.id)
+	}
+
+	var scanner *bufio.Scanner
+
+	switch cw.proto {
+	case consoleProtoUnix:
+		cw.conn, err = net.Dial("unix", cw.consoleURL)
+		if err != nil {
+			return err
+		}
+		scanner = bufio.NewScanner(cw.conn)
+	case consoleProtoPty:
+		// read-only
+		cw.ptyConsole, err = os.Open(cw.consoleURL)
+		scanner = bufio.NewScanner(cw.ptyConsole)
+	default:
+		return fmt.Errorf("unknown console proto %s", cw.proto)
+	}
+
+	go func() {
+		for scanner.Scan() {
+			s.Logger().WithFields(logrus.Fields{
+				"console-protocol": cw.proto,
+				"console-url":      cw.consoleURL,
+				"sandbox":          s.id,
+				"vmconsole":        scanner.Text(),
+			}).Debug("reading guest console")
+		}
+
+		if err := scanner.Err(); err != nil {
+			if err == io.EOF {
+				s.Logger().Info("console watcher quits")
+			} else {
+				s.Logger().WithError(err).WithFields(logrus.Fields{
+					"console-protocol": cw.proto,
+					"console-url":      cw.consoleURL,
+					"sandbox":          s.id,
+				}).Error("Failed to read guest console logs")
+			}
+		}
+	}()
+
+	return nil
+}
+
+// check if the console watcher has already watched the vm console.
+func (cw *consoleWatcher) consoleWatched() bool {
+	return cw.conn != nil || cw.ptyConsole != nil
+}
+
+// stop the console watcher.
+func (cw *consoleWatcher) stop() {
+	if cw.conn != nil {
+		cw.conn.Close()
+		cw.conn = nil
+	}
+
+	if cw.ptyConsole != nil {
+		cw.ptyConsole.Close()
+		cw.ptyConsole = nil
+	}
+}
+
 // startVM starts the VM.
 func (s *Sandbox) startVM() (err error) {
 	span, ctx := s.trace("startVM")
@@ -936,6 +1036,15 @@ func (s *Sandbox) startVM() (err error) {
 
 	s.Logger().Info("Starting VM")
 
+	if s.config.HypervisorConfig.Debug {
+		// create console watcher
+		consoleWatcher, err := newConsoleWatcher(s)
+		if err != nil {
+			return err
+		}
+		s.cw = consoleWatcher
+	}
+
 	if err := s.network.Run(s.networkNS.NetNsPath, func() error {
 		if s.factory != nil {
 			vm, err := s.factory.GetVM(ctx, VMConfig{
@@ -980,6 +1089,14 @@ func (s *Sandbox) startVM() (err error) {
 
 	s.Logger().Info("VM started")
 
+	if s.cw != nil {
+		s.Logger().Debug("console watcher starts")
+		if err := s.cw.start(s); err != nil {
+			s.cw.stop()
+			return err
+		}
+	}
+
 	// Once the hypervisor is done starting the sandbox,
 	// we want to guarantee that it is manageable.
 	// For that we need to ask the agent to start the
@@ -1478,6 +1595,12 @@ func (s *Sandbox) Stop(force bool) error {
 		return err
 	}
 
+	// shutdown console watcher if exists
+	if s.cw != nil {
+		s.Logger().Debug("stop the sandbox")
+		s.cw.stop()
+	}
+
 	if err := s.setSandboxState(types.StateStopped); err != nil {
 		return err
 	}

src/runtime/virtcontainers/utils/utils.go

@@ -9,7 +9,6 @@ import (
 	"crypto/rand"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -303,26 +302,6 @@ const (
 	GiB          = MiB << 10
 )
 
-// Binary to use to log program output
-const LoggerBinaryName = "systemd-cat"
-
-type ProgramLogger struct {
-	cmd *exec.Cmd
-}
-
-func NewProgramLogger(loggerLabel string) ProgramLogger {
-	return ProgramLogger{cmd: exec.Command(LoggerBinaryName, "-t", loggerLabel)}
-}
-
-func (p *ProgramLogger) StartLogger(output io.ReadCloser) error {
-	p.cmd.Stdin = output
-	return StartCmd(p.cmd)
-}
-
-func (p ProgramLogger) String() string {
-	return p.cmd.Path
-}
-
 func ConvertNetlinkFamily(netlinkFamily int32) pbTypes.IPFamily {
 	switch netlinkFamily {
 	case netlink.FAMILY_V6: