A draft commit of Kata support

daemon/pod/decommission.go

@@ -10,10 +10,10 @@ import (
 	dockertypes "github.com/docker/engine-api/types"
 
 	"github.com/hyperhq/hyperd/utils"
-	"github.com/hyperhq/runv/hypervisor"
+	vc "github.com/kata-containers/runtime/virtcontainers"
 )
 
-type sandboxOp func(sb *hypervisor.Vm) error
+type sandboxOp func(sb *vc.Sandbox) error
 type stateValidator func(state PodState) bool
 
 func (p *XPod) DelayDeleteOn() bool {
@@ -37,9 +37,9 @@ func (p *XPod) Stop(graceful int) error {
 
 func (p *XPod) ForceQuit() {
 	err := p.protectedSandboxOperation(
-		func(sb *hypervisor.Vm) error {
-			sb.Kill()
-			return nil
+		func(sb *vc.Sandbox) error {
+			_, err := vc.StopSandbox(sb.ID())
+			return err
 		},
 		time.Second*5,
 		"kill pod")
@@ -118,8 +118,8 @@ func (p *XPod) Pause() error {
 	p.statusLock.Unlock()
 
 	err := p.protectedSandboxOperation(
-		func(sb *hypervisor.Vm) error {
-			return sb.Pause(true)
+		func(sb *vc.Sandbox) error {
+			return sb.Pause()
 		},
 		time.Second*5,
 		"pause pod")
@@ -148,8 +148,8 @@ func (p *XPod) UnPause() error {
 	p.statusLock.Unlock()
 
 	err := p.protectedSandboxOperation(
-		func(sb *hypervisor.Vm) error {
-			return sb.Pause(false)
+		func(sb *vc.Sandbox) error {
+			return sb.Pause()
 		},
 		time.Second*5,
 		"resume pod")
@@ -176,8 +176,8 @@ func (p *XPod) KillContainer(id string, sig int64) error {
 	}
 	c.setKill()
 	return p.protectedSandboxOperation(
-		func(sb *hypervisor.Vm) error {
-			return sb.KillContainer(id, syscall.Signal(sig))
+		func(sb *vc.Sandbox) error {
+			return vc.KillContainer(sb.ID(), id, syscall.Signal(sig), true)
 		},
 		time.Second*5,
 		fmt.Sprintf("Kill container %s with %d", id, sig))
@@ -307,7 +307,7 @@ func (p *XPod) RemoveContainer(id string) error {
 // protectedSandboxOperation() protect the hypervisor operations, which may
 // panic or hang too long time.
 func (p *XPod) protectedSandboxOperation(op sandboxOp, timeout time.Duration, comment string) error {
-	dangerousOp := func(sb *hypervisor.Vm, errChan chan<- error) {
+	dangerousOp := func(sb *vc.Sandbox, errChan chan<- error) {
 		defer func() {
 			err := recover()
 			if err != nil {
@@ -393,13 +393,13 @@ func (p *XPod) doStopPod(graceful int) error {
 	}
 
 	p.Log(INFO, "stop container success, shutdown sandbox")
-	result := p.sandbox.Shutdown()
-	if result.IsSuccess() {
+	_, err = vc.StopSandbox(p.sandbox.ID())
+	if err == nil {
 		p.Log(INFO, "pod is stopped")
 		return nil
 	}
 
-	err = fmt.Errorf("failed to shuting down: %s", result.Message())
+	err = fmt.Errorf("failed to shuting down: %s", err)
 	p.Log(ERROR, err)
 	return err
 }
@@ -448,13 +448,20 @@ func (p *XPod) stopContainers(cList []string, graceful int) error {
 		}
 		future.Add(c.Id(), func() error {
 			var toc <-chan time.Time
+			var retch = make(chan int32)
+
 			if int64(graceful) < 0 {
 				toc = make(chan time.Time)
 			} else {
 				toc = time.After(waitTime)
 			}
+
 			forceKill := graceful == 0
-			resChan := p.sandbox.WaitProcess(true, []string{c.Id()}, -1)
+			go func(retch chan int32, c *Container) {
+				ret, _ := p.sandbox.WaitProcess(c.Id(), c.Id())
+				retch <- ret
+			}(retch, c)
+
 			c.Log(DEBUG, "now, stop container")
 			err := c.terminate(forceKill)
 			// TODO filter container/process can't find error
@@ -464,20 +471,11 @@ func (p *XPod) stopContainers(cList []string, graceful int) error {
 					return err
 				}
 			}
-			if resChan == nil {
-				err := fmt.Errorf("cannot wait container %s", c.Id())
-				p.Log(ERROR, err)
-				return err
-			}
+
 			for {
 				select {
-				case ex, ok := <-resChan:
-					if !ok {
-						err := fmt.Errorf("chan broken while waiting container: %s", c.Id())
-						p.Log(WARNING, err)
-						return err
-					}
-					p.Log(DEBUG, "container %s stopped (%v)", ex.Id, ex.Code)
+				case ret := <-retch:
+					p.Log(DEBUG, "container %s stopped (%d)", c.Id(), ret)
 					return nil
 				case <-toc:
 					if forceKill {
@@ -493,6 +491,7 @@ func (p *XPod) stopContainers(cList []string, graceful int) error {
 				}
 			}
 			return nil
+
 		})
 	}
 
@@ -532,7 +531,8 @@ func (p *XPod) waitVMStop() {
 	}
 	p.statusLock.RUnlock()
 
-	_, _ = <-p.sandbox.WaitVm(-1)
+	monitor, _ := p.sandbox.Monitor()
+	_ = <-monitor
 	p.Log(INFO, "got vm exit event")
 	p.cleanup()
 }

daemon/pod/exec.go

@@ -8,11 +8,9 @@ import (
 	"time"
 
 	"github.com/docker/docker/pkg/stdcopy"
-
 	"github.com/hyperhq/hypercontainer-utils/hlog"
 	"github.com/hyperhq/hyperd/utils"
-	"github.com/hyperhq/runv/api"
-	"github.com/hyperhq/runv/hypervisor"
+	vc "github.com/kata-containers/runtime/virtcontainers"
 )
 
 type Exec struct {
@@ -57,7 +55,7 @@ func (p *XPod) CreateExec(containerId, cmds string, terminal bool) (string, erro
 	p.statusLock.Lock()
 	p.execs[execId] = &Exec{
 		Container: containerId,
-		Id:        execId,
+		Id:        "",
 		Cmds:      command,
 		Terminal:  terminal,
 		ExitCode:  255,
@@ -85,6 +83,7 @@ type writeCloser struct {
 }
 
 func (p *XPod) StartExec(stdin io.ReadCloser, stdout io.WriteCloser, containerId, execId string) error {
+
 	c, ok := p.containers[containerId]
 	if !ok {
 		err := fmt.Errorf("no container %s available for exec %s", containerId, execId)
@@ -103,7 +102,7 @@ func (p *XPod) StartExec(stdin io.ReadCloser, stdout io.WriteCloser, containerId
 	}
 
 	wReader := &waitClose{ReadCloser: stdin, wait: make(chan bool)}
-	tty := &hypervisor.TtyIO{
+	tty := &TtyIO{
 		Stdin:  wReader,
 		Stdout: stdout,
 	}
@@ -124,21 +123,39 @@ func (p *XPod) StartExec(stdin io.ReadCloser, stdout io.WriteCloser, containerId
 		}
 	}
 
+	cmd := vc.Cmd{
+		Args:         es.Cmds,
+		Envs:         c.cmdEnvs([]vc.EnvVar{}),
+		WorkDir:      c.spec.Workdir,
+		Interactive:  es.Terminal,
+		Detach:       !es.Terminal,
+		User:         "0", //set the default user and group
+		PrimaryGroup: "0",
+	}
+
+	if c.spec.User != nil {
+		cmd.User = c.spec.User.Name
+		cmd.PrimaryGroup = c.spec.User.Group
+	}
+
+	_, process, err := p.sandbox.EnterContainer(containerId, cmd)
+	if err != nil {
+		err := fmt.Errorf("cannot enter container %s, with err %s", containerId, err)
+		p.Log(ERROR, err)
+		return err
+	}
+	es.Id = process.Token
+
 	go func(es *Exec) {
-		result := p.sandbox.WaitProcess(false, []string{execId}, -1)
-		if result == nil {
+		ret, err := p.sandbox.WaitProcess(containerId, es.Id)
+		if err == nil {
 			es.Log(ERROR, "can not wait exec")
 			return
 		}
 
-		r, ok := <-result
-		if !ok {
-			es.Log(ERROR, "waiting exec interrupted")
-			return
-		}
+		es.Log(DEBUG, "exec terminated at %v with code %d", time.Now(), int(ret))
+		es.ExitCode = uint8(ret)
 
-		es.Log(DEBUG, "exec terminated at %v with code %d", r.FinishedAt, r.Code)
-		es.ExitCode = uint8(r.Code)
 		select {
 		case es.finChan <- true:
 			es.Log(DEBUG, "wake exec stopped chan")
@@ -147,30 +164,17 @@ func (p *XPod) StartExec(stdin io.ReadCloser, stdout io.WriteCloser, containerId
 		}
 	}(es)
 
-	var envs []string
-	for e, v := range c.descript.Envs {
-		envs = append(envs, fmt.Sprintf("%s=%s", e, v))
-	}
-
-	process := &api.Process{
-		Container: es.Container,
-		Id:        es.Id,
-		Terminal:  es.Terminal,
-		Args:      es.Cmds,
-		Envs:      envs,
-		Workdir:   c.descript.Workdir,
-	}
-
-	if c.descript.UGI != nil {
-		process.User = c.descript.UGI.User
-		process.Group = c.descript.UGI.Group
-		process.AdditionalGroup = c.descript.UGI.AdditionalGroups
+	cstdin, cstdout, cstderr, err := p.sandbox.IOStream(containerId, es.Id)
+	if err != nil {
+		c.Log(ERROR, err)
+		return err
 	}
 
-	err := p.sandbox.AddProcess(process, tty)
+	go streamCopy(tty, cstdin, cstdout, cstderr)
 
 	<-wReader.wait
-	return err
+
+	return nil
 }
 
 func (p *XPod) GetExecExitCode(containerId, execId string) (uint8, error) {
@@ -208,8 +212,8 @@ func (p *XPod) KillExec(execId string, sig int64) error {
 	}
 
 	return p.protectedSandboxOperation(
-		func(sb *hypervisor.Vm) error {
-			return sb.SignalProcess(es.Container, es.Id, syscall.Signal(sig))
+		func(sb *vc.Sandbox) error {
+			return sb.SignalProcess(es.Container, es.Id, syscall.Signal(sig), true)
 		},
 		time.Second*5,
 		fmt.Sprintf("Kill process %s with %d", es.Id, sig))
@@ -228,16 +232,20 @@ func (p *XPod) CleanupExecs() {
 }
 
 func (p *XPod) ExecVM(cmd string, stdin io.ReadCloser, stdout, stderr io.WriteCloser) (int, error) {
-	wReader := &waitClose{ReadCloser: stdin, wait: make(chan bool)}
-	tty := &hypervisor.TtyIO{
-		Stdin:  wReader,
-		Stdout: stdout,
-		Stderr: stderr,
-	}
-	res, err := p.sandbox.HyperstartExec(cmd, tty)
-	if err != nil {
-		return res, err
-	}
-	<-wReader.wait
-	return res, err
+	/*
+		wReader := &waitClose{ReadCloser: stdin, wait: make(chan bool)}
+		tty := &hypervisor.TtyIO{
+			Stdin:  wReader,
+			Stdout: stdout,
+			Stderr: stderr,
+		}
+
+		res, err := p.sandbox.HyperstartExec(cmd, tty)
+		if err != nil {
+			return res, err
+		}
+		<-wReader.wait
+	*/
+	//	return res, err
+	return 0, nil
 }

daemon/pod/networks.go

@@ -76,11 +76,14 @@ func (inf *Interface) add() error {
 		inf.Log(ERROR, err)
 		return err
 	}
-	err := inf.p.sandbox.AddNic(inf.descript)
-	if err != nil {
-		inf.Log(ERROR, "failed to add NIC: %v", err)
-	}
-	return err
+	/*
+		err := inf.p.sandbox.AddNic(inf.descript)
+		if err != nil {
+			inf.Log(ERROR, "failed to add NIC: %v", err)
+		}
+		return err
+	*/
+	return nil
 }
 
 func (inf *Interface) cleanup() error {