proxy: add kataProxyBuiltin

When specified, it does not spawn a new process to proxy kata grpc
connections. Instead, the yamux multiplexing functionality is builtin
in the kata agent dialer.

Signed-off-by: Peng Tao <bergwolf@gmail.com>

virtcontainers/kata_agent.go

@@ -73,8 +73,9 @@ func (s *kataVSOCK) String() string {
 // KataAgentState is the structure describing the data stored from this
 // agent implementation.
 type KataAgentState struct {
-	ProxyPid int
-	URL      string
+	ProxyPid     int
+	ProxyBuiltIn bool
+	URL          string
 }
 
 type kataAgent struct {
@@ -417,6 +418,7 @@ func (k *kataAgent) startPod(pod Pod) error {
 
 	proxyParams := proxyParams{
 		agentURL: agentURL,
+		logger:   k.Logger().WithField("pod-id", pod.id),
 	}
 
 	// Start the proxy here
@@ -427,12 +429,17 @@ func (k *kataAgent) startPod(pod Pod) error {
 
 	// Fill agent state with proxy information, and store them.
 	k.state.ProxyPid = pid
+	k.state.ProxyBuiltIn = isProxyBuiltIn(pod.config.ProxyType)
 	k.state.URL = uri
 	if err := pod.storage.storeAgentState(pod.id, k.state); err != nil {
 		return err
 	}
 
-	k.Logger().WithField("proxy-pid", pid).Info("proxy started")
+	k.Logger().WithFields(logrus.Fields{
+		"pod-id":    pod.id,
+		"proxy-pid": pid,
+		"proxy-url": uri,
+	}).Info("proxy started")
 
 	hostname := pod.config.Hostname
 	if len(hostname) > maxHostnameLen {
@@ -805,7 +812,7 @@ func (k *kataAgent) connect() error {
 		return nil
 	}
 
-	client, err := kataclient.NewAgentClient(k.state.URL)
+	client, err := kataclient.NewAgentClient(k.state.URL, k.state.ProxyBuiltIn)
 	if err != nil {
 		return err
 	}

virtcontainers/kata_builtin_proxy.go

@@ -0,0 +1,103 @@
+//
+// Copyright (c) 2018 HyperHQ Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package virtcontainers
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+
+	"github.com/sirupsen/logrus"
+)
+
+// This is a kata builtin proxy implementation of the proxy interface. Kata proxy
+// functionality is implemented inside the virtcontainers library.
+type kataBuiltInProxy struct {
+	podID string
+	conn  net.Conn
+}
+
+// start is the proxy start implementation for kata builtin proxy.
+// It starts the console watcher for the guest.
+// It returns agentURL to let agent connect directly.
+func (p *kataBuiltInProxy) start(pod Pod, params proxyParams) (int, string, error) {
+	if p.conn != nil {
+		return -1, "", fmt.Errorf("kata builtin proxy running for pod %s", p.podID)
+	}
+
+	p.podID = pod.id
+	console := pod.hypervisor.getPodConsole(pod.id)
+	err := p.watchConsole(consoleProtoUnix, console, params.logger)
+	if err != nil {
+		return -1, "", err
+	}
+
+	return -1, params.agentURL, nil
+}
+
+// stop is the proxy stop implementation for kata builtin proxy.
+func (p *kataBuiltInProxy) stop(pod Pod, pid int) error {
+	if p.conn != nil {
+		p.conn.Close()
+		p.conn = nil
+		p.podID = ""
+	}
+	return nil
+}
+
+func (p *kataBuiltInProxy) watchConsole(proto, console string, logger *logrus.Entry) (err error) {
+	var (
+		scanner *bufio.Scanner
+		conn    net.Conn
+	)
+
+	switch proto {
+	case consoleProtoUnix:
+		conn, err = net.Dial("unix", console)
+		if err != nil {
+			return err
+		}
+	// TODO: add pty console support for kvmtools
+	case consoleProtoPty:
+		fallthrough
+	default:
+		return fmt.Errorf("unknown console proto %s", proto)
+	}
+
+	p.conn = conn
+
+	go func() {
+		scanner = bufio.NewScanner(conn)
+		for scanner.Scan() {
+			fmt.Printf("[POD-%s] vmconsole: %s\n", p.podID, scanner.Text())
+		}
+
+		if err := scanner.Err(); err != nil {
+			if err == io.EOF {
+				logger.Info("console watcher quits")
+			} else {
+				logger.WithError(err).WithFields(logrus.Fields{
+					"console-protocol": proto,
+					"console-socket":   console,
+				}).Error("Failed to read agent logs")
+			}
+		}
+	}()
+
+	return nil
+}

virtcontainers/proxy.go

@@ -21,6 +21,7 @@ import (
 	"path/filepath"
 
 	"github.com/mitchellh/mapstructure"
+	"github.com/sirupsen/logrus"
 )
 
 // ProxyConfig is a structure storing information needed from any
@@ -34,6 +35,7 @@ type ProxyConfig struct {
 // for the execution of the proxy binary.
 type proxyParams struct {
 	agentURL string
+	logger   *logrus.Entry
 }
 
 // ProxyType describes a proxy type.
@@ -51,6 +53,9 @@ const (
 
 	// KataProxyType is the kataProxy.
 	KataProxyType ProxyType = "kataProxy"
+
+	// KataBuiltInProxyType is the kataBuiltInProxy.
+	KataBuiltInProxyType ProxyType = "kataBuiltInProxy"
 )
 
 const (
@@ -59,6 +64,14 @@ const (
 	waitForProxyTimeoutSecs = 5.0
 )
 
+const (
+	// unix socket type of console
+	consoleProtoUnix = "unix"
+
+	// pty type of console. Used mostly by kvmtools.
+	consoleProtoPty = "pty"
+)
+
 // Set sets a proxy type based on the input string.
 func (pType *ProxyType) Set(value string) error {
 	switch value {
@@ -74,6 +87,9 @@ func (pType *ProxyType) Set(value string) error {
 	case "kataProxy":
 		*pType = KataProxyType
 		return nil
+	case "kataBuiltInProxy":
+		*pType = KataBuiltInProxyType
+		return nil
 	default:
 		return fmt.Errorf("Unknown proxy type %s", value)
 	}
@@ -90,6 +106,8 @@ func (pType *ProxyType) String() string {
 		return string(CCProxyType)
 	case KataProxyType:
 		return string(KataProxyType)
+	case KataBuiltInProxyType:
+		return string(KataBuiltInProxyType)
 	default:
 		return ""
 	}
@@ -106,6 +124,8 @@ func newProxy(pType ProxyType) (proxy, error) {
 		return &ccProxy{}, nil
 	case KataProxyType:
 		return &kataProxy{}, nil
+	case KataBuiltInProxyType:
+		return &kataBuiltInProxy{}, nil
 	default:
 		return &noopProxy{}, nil
 	}
@@ -148,6 +168,10 @@ func defaultProxyURL(pod Pod, socketType string) (string, error) {
 	}
 }
 
+func isProxyBuiltIn(pType ProxyType) bool {
+	return pType == KataBuiltInProxyType
+}
+
 // proxy is the virtcontainers proxy interface.
 type proxy interface {
 	// start launches a proxy instance for the specified pod, returning