Merge pull request #501 from bergwolf/qemu

virtcontainers: keep qmp connection whenever possible

virtcontainers/api.go

@@ -35,7 +35,12 @@ func SetLogger(logger logrus.FieldLogger) {
 // CreateSandbox is the virtcontainers sandbox creation entry point.
 // CreateSandbox creates a sandbox and its containers. It does not start them.
 func CreateSandbox(sandboxConfig SandboxConfig, factory Factory) (VCSandbox, error) {
-	return createSandboxFromConfig(sandboxConfig, factory)
+	s, err := createSandboxFromConfig(sandboxConfig, factory)
+	if err == nil {
+		s.releaseStatelessSandbox()
+	}
+
+	return s, err
 }
 
 func createSandboxFromConfig(sandboxConfig SandboxConfig, factory Factory) (*Sandbox, error) {
@@ -86,6 +91,7 @@ func DeleteSandbox(sandboxID string) (VCSandbox, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Delete it.
 	if err := s.Delete(); err != nil {
@@ -97,7 +103,8 @@ func DeleteSandbox(sandboxID string) (VCSandbox, error) {
 
 // FetchSandbox is the virtcontainers sandbox fetching entry point.
 // FetchSandbox will find out and connect to an existing sandbox and
-// return the sandbox structure.
+// return the sandbox structure. The caller is responsible of calling
+// VCSandbox.Release() after done with it.
 func FetchSandbox(sandboxID string) (VCSandbox, error) {
 	if sandboxID == "" {
 		return nil, errNeedSandboxID
@@ -110,21 +117,22 @@ func FetchSandbox(sandboxID string) (VCSandbox, error) {
 	defer unlockSandbox(lockFile)
 
 	// Fetch the sandbox from storage and create it.
-	sandbox, err := fetchSandbox(sandboxID)
+	s, err := fetchSandbox(sandboxID)
 	if err != nil {
 		return nil, err
 	}
 
 	// If the proxy is KataBuiltInProxyType type, it needs to restart the proxy to watch the
 	// guest console if it hadn't been watched.
-	if isProxyBuiltIn(sandbox.config.ProxyType) {
-		err = sandbox.startProxy()
+	if isProxyBuiltIn(s.config.ProxyType) {
+		err = s.startProxy()
 		if err != nil {
+			s.Release()
 			return nil, err
 		}
 	}
 
-	return sandbox, nil
+	return s, nil
 }
 
 // StartSandbox is the virtcontainers sandbox starting entry point.
@@ -147,6 +155,7 @@ func StartSandbox(sandboxID string) (VCSandbox, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	return startSandbox(s)
 }
@@ -184,6 +193,7 @@ func StopSandbox(sandboxID string) (VCSandbox, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Stop it.
 	err = s.stop()
@@ -211,6 +221,7 @@ func RunSandbox(sandboxConfig SandboxConfig, factory Factory) (VCSandbox, error)
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	lockFile, err := rwLockSandbox(s.id)
 	if err != nil {
@@ -269,6 +280,7 @@ func StatusSandbox(sandboxID string) (SandboxStatus, error) {
 		unlockSandbox(lockFile)
 		return SandboxStatus{}, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// We need to potentially wait for a separate container.stop() routine
 	// that needs to be terminated before we return from this function.
@@ -320,6 +332,7 @@ func CreateContainer(sandboxID string, containerConfig ContainerConfig) (VCSandb
 	if err != nil {
 		return nil, nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	c, err := s.CreateContainer(containerConfig)
 	if err != nil {
@@ -351,6 +364,7 @@ func DeleteContainer(sandboxID, containerID string) (VCContainer, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	return s.DeleteContainer(containerID)
 }
@@ -376,6 +390,7 @@ func StartContainer(sandboxID, containerID string) (VCContainer, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	c, err := s.StartContainer(containerID)
 	if err != nil {
@@ -406,6 +421,7 @@ func StopContainer(sandboxID, containerID string) (VCContainer, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Fetch the container.
 	c, err := s.findContainer(containerID)
@@ -443,6 +459,7 @@ func EnterContainer(sandboxID, containerID string, cmd Cmd) (VCSandbox, VCContai
 	if err != nil {
 		return nil, nil, nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	c, process, err := s.EnterContainer(containerID, cmd)
 	if err != nil {
@@ -473,6 +490,7 @@ func StatusContainer(sandboxID, containerID string) (ContainerStatus, error) {
 		unlockSandbox(lockFile)
 		return ContainerStatus{}, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// We need to potentially wait for a separate container.stop() routine
 	// that needs to be terminated before we return from this function.
@@ -559,6 +577,7 @@ func KillContainer(sandboxID, containerID string, signal syscall.Signal, all boo
 	if err != nil {
 		return err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Fetch the container.
 	c, err := s.findContainer(containerID)
@@ -608,6 +627,7 @@ func ProcessListContainer(sandboxID, containerID string, options ProcessListOpti
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Fetch the container.
 	c, err := s.findContainer(containerID)
@@ -639,6 +659,7 @@ func UpdateContainer(sandboxID, containerID string, resources specs.LinuxResourc
 	if err != nil {
 		return err
 	}
+	defer s.releaseStatelessSandbox()
 
 	return s.UpdateContainer(containerID, resources)
 }
@@ -664,6 +685,7 @@ func StatsContainer(sandboxID, containerID string) (ContainerStats, error) {
 	if err != nil {
 		return ContainerStats{}, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	return s.StatsContainer(containerID)
 }
@@ -687,6 +709,7 @@ func togglePauseContainer(sandboxID, containerID string, pause bool) error {
 	if err != nil {
 		return err
 	}
+	defer s.releaseStatelessSandbox()
 
 	// Fetch the container.
 	c, err := s.findContainer(containerID)

virtcontainers/api_test.go

@@ -1811,6 +1811,7 @@ func TestStatusContainerStateReady(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer p2.releaseStatelessSandbox()
 
 	expectedStatus := ContainerStatus{
 		ID: contID,
@@ -1884,6 +1885,7 @@ func TestStatusContainerStateRunning(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer p2.releaseStatelessSandbox()
 
 	expectedStatus := ContainerStatus{
 		ID: contID,
@@ -2333,7 +2335,23 @@ func TestFetchSandbox(t *testing.T) {
 
 	fetched, err := FetchSandbox(s.ID())
 	assert.Nil(t, err, "%v", err)
-	assert.True(t, fetched == s, "fetched sandboxed do not match")
+	assert.True(t, fetched != s, "fetched stateless sandboxes should not match")
+}
+
+func TestFetchStatefulSandbox(t *testing.T) {
+	cleanUp()
+
+	config := newTestSandboxConfigNoop()
+
+	config.Stateful = true
+	s, err := CreateSandbox(config, nil)
+	if s == nil || err != nil {
+		t.Fatal(err)
+	}
+
+	fetched, err := FetchSandbox(s.ID())
+	assert.Nil(t, err, "%v", err)
+	assert.Equal(t, fetched, s, "fetched stateful sandboxed should match")
 }
 
 func TestFetchNonExistingSandbox(t *testing.T) {

virtcontainers/hypervisor.go

@@ -547,5 +547,6 @@ type hypervisor interface {
 	hotplugAddDevice(devInfo interface{}, devType deviceType) (interface{}, error)
 	hotplugRemoveDevice(devInfo interface{}, devType deviceType) (interface{}, error)
 	getSandboxConsole(sandboxID string) (string, error)
+	disconnect()
 	capabilities() capabilities
 }

virtcontainers/mock_hypervisor.go

@@ -73,3 +73,6 @@ func (m *mockHypervisor) hotplugRemoveDevice(devInfo interface{}, devType device
 func (m *mockHypervisor) getSandboxConsole(sandboxID string) (string, error) {
 	return "", nil
 }
+
+func (m *mockHypervisor) disconnect() {
+}

virtcontainers/mock_hypervisor_test.go

@@ -104,3 +104,9 @@ func TestMockHypervisorSaveSandbox(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestMockHypervisorDisconnect(t *testing.T) {
+	var m *mockHypervisor
+
+	m.disconnect()
+}

virtcontainers/qemu.go

@@ -507,12 +507,6 @@ func (q *qemu) startSandbox() error {
 
 // waitSandbox will wait for the Sandbox's VM to be up and running.
 func (q *qemu) waitSandbox(timeout int) error {
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
 	if timeout < 0 {
 		return fmt.Errorf("Invalid timeout %ds", timeout)
 	}
@@ -537,8 +531,9 @@ func (q *qemu) waitSandbox(timeout int) error {
 
 		time.Sleep(time.Duration(50) * time.Millisecond)
 	}
-
 	q.qmpMonitorCh.qmp = qmp
+	defer q.qmpShutdown()
+
 	qemuMajorVersion = ver.Major
 	qemuMinorVersion = ver.Minor
 
@@ -559,23 +554,14 @@ func (q *qemu) waitSandbox(timeout int) error {
 
 // stopSandbox will stop the Sandbox's VM.
 func (q *qemu) stopSandbox() error {
-	cfg := govmmQemu.QMPConfig{Logger: newQMPLogger()}
-	disconnectCh := make(chan struct{})
-
 	q.Logger().Info("Stopping Sandbox")
-	qmp, _, err := govmmQemu.QMPStart(q.qmpMonitorCh.ctx, q.qmpMonitorCh.path, cfg, disconnectCh)
+
+	err := q.qmpSetup()
 	if err != nil {
-		q.Logger().WithError(err).Error("Failed to connect to QEMU instance")
 		return err
 	}
 
-	err = qmp.ExecuteQMPCapabilities(q.qmpMonitorCh.ctx)
-	if err != nil {
-		q.Logger().WithError(err).Error(qmpCapErrMsg)
-		return err
-	}
-
-	err = qmp.ExecuteQuit(q.qmpMonitorCh.ctx)
+	err = q.qmpMonitorCh.qmp.ExecuteQuit(q.qmpMonitorCh.ctx)
 	if err != nil {
 		q.Logger().WithError(err).Error("Fail to execute qmp QUIT")
 		return err
@@ -590,28 +576,8 @@ func (q *qemu) stopSandbox() error {
 }
 
 func (q *qemu) togglePauseSandbox(pause bool) error {
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
-	cfg := govmmQemu.QMPConfig{Logger: newQMPLogger()}
-
-	// Auto-closed by QMPStart().
-	disconnectCh := make(chan struct{})
-
-	qmp, _, err := govmmQemu.QMPStart(q.qmpMonitorCh.ctx, q.qmpMonitorCh.path, cfg, disconnectCh)
+	err := q.qmpSetup()
 	if err != nil {
-		q.Logger().WithError(err).Error("Failed to connect to QEMU instance")
-		return err
-	}
-
-	q.qmpMonitorCh.qmp = qmp
-
-	err = qmp.ExecuteQMPCapabilities(q.qmpMonitorCh.ctx)
-	if err != nil {
-		q.Logger().WithError(err).Error(qmpCapErrMsg)
 		return err
 	}
 
@@ -628,7 +594,11 @@ func (q *qemu) togglePauseSandbox(pause bool) error {
 	return nil
 }
 
-func (q *qemu) qmpSetup() (*govmmQemu.QMP, error) {
+func (q *qemu) qmpSetup() error {
+	if q.qmpMonitorCh.qmp != nil {
+		return nil
+	}
+
 	cfg := govmmQemu.QMPConfig{Logger: newQMPLogger()}
 
 	// Auto-closed by QMPStart().
@@ -637,16 +607,25 @@ func (q *qemu) qmpSetup() (*govmmQemu.QMP, error) {
 	qmp, _, err := govmmQemu.QMPStart(q.qmpMonitorCh.ctx, q.qmpMonitorCh.path, cfg, disconnectCh)
 	if err != nil {
 		q.Logger().WithError(err).Error("Failed to connect to QEMU instance")
-		return nil, err
+		return err
 	}
 
 	err = qmp.ExecuteQMPCapabilities(q.qmpMonitorCh.ctx)
 	if err != nil {
+		qmp.Shutdown()
 		q.Logger().WithError(err).Error(qmpCapErrMsg)
-		return nil, err
+		return err
 	}
+	q.qmpMonitorCh.qmp = qmp
 
-	return qmp, nil
+	return nil
+}
+
+func (q *qemu) qmpShutdown() {
+	if q.qmpMonitorCh.qmp != nil {
+		q.qmpMonitorCh.qmp.Shutdown()
+		q.qmpMonitorCh.qmp = nil
+	}
 }
 
 func (q *qemu) addDeviceToBridge(ID string) (string, Bridge, error) {
@@ -678,19 +657,11 @@ func (q *qemu) removeDeviceFromBridge(ID string) error {
 }
 
 func (q *qemu) hotplugBlockDevice(drive *deviceDrivers.Drive, op operation) error {
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
-	qmp, err := q.qmpSetup()
+	err := q.qmpSetup()
 	if err != nil {
 		return err
 	}
 
-	q.qmpMonitorCh.qmp = qmp
-
 	devID := "virtio-" + drive.ID
 
 	if op == addDevice {
@@ -747,19 +718,11 @@ func (q *qemu) hotplugBlockDevice(drive *deviceDrivers.Drive, op operation) erro
 }
 
 func (q *qemu) hotplugVFIODevice(device deviceDrivers.VFIODevice, op operation) error {
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
-	qmp, err := q.qmpSetup()
+	err := q.qmpSetup()
 	if err != nil {
 		return err
 	}
 
-	q.qmpMonitorCh.qmp = qmp
-
 	devID := "vfio-" + device.DeviceInfo.ID
 
 	if op == addDevice {
@@ -829,19 +792,11 @@ func (q *qemu) hotplugCPUs(vcpus uint32, op operation) (uint32, error) {
 		return 0, nil
 	}
 
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
-	qmp, err := q.qmpSetup()
+	err := q.qmpSetup()
 	if err != nil {
 		return 0, err
 	}
 
-	q.qmpMonitorCh.qmp = qmp
-
 	if op == addDevice {
 		return q.hotplugAddCPUs(vcpus)
 	}
@@ -964,22 +919,12 @@ func (q *qemu) hotplugMemory(memDev *memoryDevice, op operation) error {
 }
 
 func (q *qemu) hotplugAddMemory(memDev *memoryDevice) error {
-	// setup qmp channel if necessary
-	if q.qmpMonitorCh.qmp == nil {
-		qmp, err := q.qmpSetup()
-		if err != nil {
-			return err
-		}
-
-		q.qmpMonitorCh.qmp = qmp
-
-		defer func() {
-			qmp.Shutdown()
-			q.qmpMonitorCh.qmp = nil
-		}()
+	err := q.qmpSetup()
+	if err != nil {
+		return err
 	}
 
-	err := q.qmpMonitorCh.qmp.ExecHotplugMemory(q.qmpMonitorCh.ctx, "memory-backend-ram", "mem"+strconv.Itoa(memDev.slot), "", memDev.sizeMB)
+	err = q.qmpMonitorCh.qmp.ExecHotplugMemory(q.qmpMonitorCh.ctx, "memory-backend-ram", "mem"+strconv.Itoa(memDev.slot), "", memDev.sizeMB)
 	if err != nil {
 		q.Logger().WithError(err).Error("hotplug memory")
 		return err
@@ -1032,30 +977,10 @@ func (q *qemu) getSandboxConsole(id string) (string, error) {
 }
 
 func (q *qemu) saveSandbox() error {
-	defer func(qemu *qemu) {
-		if q.qmpMonitorCh.qmp != nil {
-			q.qmpMonitorCh.qmp.Shutdown()
-		}
-	}(q)
-
 	q.Logger().Info("save sandbox")
 
-	cfg := govmmQemu.QMPConfig{Logger: newQMPLogger()}
-
-	// Auto-closed by QMPStart().
-	disconnectCh := make(chan struct{})
-
-	qmp, _, err := govmmQemu.QMPStart(q.qmpMonitorCh.ctx, q.qmpMonitorCh.path, cfg, disconnectCh)
+	err := q.qmpSetup()
 	if err != nil {
-		q.Logger().WithError(err).Error("Failed to connect to QEMU instance")
-		return err
-	}
-
-	q.qmpMonitorCh.qmp = qmp
-
-	err = qmp.ExecuteQMPCapabilities(q.qmpMonitorCh.ctx)
-	if err != nil {
-		q.Logger().WithError(err).Error(qmpCapErrMsg)
 		return err
 	}
 
@@ -1083,6 +1008,10 @@ func (q *qemu) saveSandbox() error {
 	return nil
 }
 
+func (q *qemu) disconnect() {
+	q.qmpShutdown()
+}
+
 // genericAppendBridges appends to devices the given bridges
 func genericAppendBridges(devices []govmmQemu.Device, bridges []Bridge, machineType string) []govmmQemu.Device {
 	bus := defaultPCBridgeBus

virtcontainers/sandbox.go

@@ -350,6 +350,10 @@ type SandboxConfig struct {
 
 	// SharePidNs sets all containers to share the same sandbox level pid namespace.
 	SharePidNs bool
+
+	// Stateful keeps sandbox resources in memory across APIs. Users will be responsible
+	// for calling Release() to release the memory resources.
+	Stateful bool
 }
 
 func (s *Sandbox) startProxy() error {
@@ -468,6 +472,7 @@ type Sandbox struct {
 
 	shmSize    uint64
 	sharePidNs bool
+	stateful   bool
 }
 
 // ID returns the sandbox identifier string.
@@ -541,13 +546,23 @@ func (s *Sandbox) GetContainer(containerID string) VCContainer {
 
 // Release closes the agent connection and removes sandbox from internal list.
 func (s *Sandbox) Release() error {
+	s.Logger().Info("release sandbox")
 	globalSandboxList.removeSandbox(s.id)
 	if s.monitor != nil {
 		s.monitor.stop()
 	}
+	s.hypervisor.disconnect()
 	return s.agent.disconnect()
 }
 
+func (s *Sandbox) releaseStatelessSandbox() error {
+	if s.stateful {
+		return nil
+	}
+
+	return s.Release()
+}
+
 // Status gets the status of the sandbox
 // TODO: update container status properly, see kata-containers/runtime#253
 func (s *Sandbox) Status() SandboxStatus {
@@ -750,6 +765,7 @@ func newSandbox(sandboxConfig SandboxConfig, factory Factory) (*Sandbox, error)
 		wg:              &sync.WaitGroup{},
 		shmSize:         sandboxConfig.ShmSize,
 		sharePidNs:      sandboxConfig.SharePidNs,
+		stateful:        sandboxConfig.Stateful,
 	}
 
 	if err = globalSandboxList.addSandbox(s); err != nil {
@@ -808,6 +824,7 @@ func (s *Sandbox) storeSandbox() error {
 
 // fetchSandbox fetches a sandbox config from a sandbox ID and returns a sandbox.
 func fetchSandbox(sandboxID string) (sandbox *Sandbox, err error) {
+	virtLog.WithField("sandbox-id", sandboxID).Info("fetch sandbox")
 	if sandboxID == "" {
 		return nil, errNeedSandboxID
 	}
@@ -1388,6 +1405,7 @@ func togglePauseSandbox(sandboxID string, pause bool) (*Sandbox, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer s.releaseStatelessSandbox()
 
 	if pause {
 		err = s.Pause()