proxy: implement (re)store of proxy's state

Introduce the high availability feature of cc-proxy by implementing
store/restore of proxy's state to/from disk. This feature depends
on the ability of shim to reconnect to cc-proxy if connection is lost.

Fixes #4.

Signed-off-by: Dmitry Voytik <dmitry.voytik@huawei.com>

Author: Dmitry Voytik

Makefile

@@ -16,6 +16,7 @@ LOCALSTATEDIR := /var
 
 SOURCES := $(shell find . 2>&1 | grep -E '.*\.(c|h|go)$$')
 PROXY_SOCKET := $(LOCALSTATEDIR)/run/clear-containers/proxy.sock
+STORE_STATE_DIR := $(LOCALSTATEDIR)/lib/clear-containers/proxy/
 
 DESCRIBE := $(shell git describe 2> /dev/null || true)
 DESCRIBE_DIRTY := $(if $(shell git status --porcelain --untracked-files=no 2> /dev/null),${DESCRIBE}-dirty,${DESCRIBE})
@@ -53,7 +54,7 @@ all: cc-proxy $(UNIT_FILES)
 
 cc-proxy: $(SOURCES) Makefile
 	$(QUIET_GOBUILD)go build -i -o $@ -ldflags \
-		"-X main.DefaultSocketPath=$(PROXY_SOCKET) -X main.Version=$(VERSION)"
+		"-X main.DefaultSocketPath=$(PROXY_SOCKET) -X main.Version=$(VERSION) -X main.storeStateDir=$(STORE_STATE_DIR)"
 
 #
 # Tests

proxy.go

@@ -20,6 +20,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	_ "net/http/pprof"
@@ -36,6 +37,13 @@ import (
 	"github.com/clearcontainers/proxy/api"
 )
 
+// storeStateDir is populated at link time with the value of:
+//   $(LOCALSTATEDIR)/lib/clear-containers/proxy/"
+var storeStateDir string = "/var/lib/clearcontainers/proxy/"
+
+const proxyStateDirPerm = 0755
+const proxyStateFilesPerm = 0640
+
 // tokenState  tracks if an I/O token has been claimed by a shim.
 type tokenState int
 
@@ -84,6 +92,20 @@ type proxy struct {
 	wg sync.WaitGroup
 }
 
+// proxyStateOnDisk is used to re(store) proxy state on disk
+type proxyStateOnDisk struct {
+	Version         string   `json:"version"`
+	SocketPath      string   `json:"socket_path"`
+	EnableVMConsole bool     `json:"enable_vm_console"`
+	ContainerIDs    []string `json:"container_ids"`
+}
+
+// vmStateOnDisk is used to re(store) vm struct on disk
+type vmStateOnDisk struct {
+	RegisterVM api.RegisterVM `json:"registerVM"`
+	Tokens     []string       `json:"tokens"`
+}
+
 type clientKind int
 
 const (
@@ -129,6 +151,33 @@ func newClient(proxy *proxy, conn net.Conn) *client {
 	}
 }
 
+func (proxy *proxy) restoreTokens(vm *vm, tokens []string) error {
+	if vm == nil {
+		return fmt.Errorf("vm parameter must be not nil")
+	}
+
+	for _, token := range tokens {
+		token, err := vm.AllocateTokenAs(Token(token))
+		if err != nil {
+			return err
+		}
+		proxy.Lock()
+		proxy.tokenToVM[token] = &tokenInfo{
+			state: tokenStateAllocated,
+			vm:    vm,
+		}
+		proxy.Unlock()
+
+		session := vm.findSessionByToken(token)
+		if session == nil {
+			return fmt.Errorf("unknown token %s", token)
+		}
+		// Signal that the process is already started
+		close(session.processStarted)
+	}
+	return nil
+}
+
 func (proxy *proxy) allocateTokens(vm *vm, numIOStreams int) (*api.IOResponse, error) {
 	url := url.URL{
 		Scheme: "unix",
@@ -193,6 +242,229 @@ func (proxy *proxy) releaseToken(token Token) (*tokenInfo, error) {
 	return info, nil
 }
 
+// returns false if it's a clean start (i.e. no state is stored) or restoring failed
+func (proxy *proxy) restoreState() bool {
+	proxyStateFilePath := storeStateDir + "proxy_state.json"
+	if _, err := os.Stat(storeStateDir); os.IsNotExist(err) {
+		err := os.MkdirAll(storeStateDir, proxyStateDirPerm)
+		if err != nil {
+			proxyLog.Errorf("Couldn't create directory %s: %v",
+				storeStateDir, err)
+		}
+		return false
+	}
+
+	fdata, err := ioutil.ReadFile(proxyStateFilePath)
+	if err != nil {
+		proxyLog.Errorf("Couldn't recover from %s: %v", proxyStateFilePath, err)
+		return false
+	}
+
+	var proxyState proxyStateOnDisk
+	err = json.Unmarshal(fdata, &proxyState)
+	if err != nil {
+		proxyLog.Errorf("Couldn't unmarshal %s: %v", proxyStateFilePath, err)
+		return false
+	}
+	proxyLog.Debugf("proxy: %+v", proxyState)
+
+	if len(proxyState.ContainerIDs) == 0 {
+		return false
+	}
+	proxyLog.Warn("Recovering proxy state from: ", proxyStateFilePath)
+	if proxyState.Version != Version {
+		proxyLog.Warnf("Stored state version (%s) mismatches proxy"+
+			" version (%s). Aborting", proxyState.Version, Version)
+		return false
+	}
+
+	proxy.socketPath = proxyState.SocketPath
+	proxy.enableVMConsole = proxyState.EnableVMConsole
+
+	for _, contID := range proxyState.ContainerIDs {
+		go restoreVMState(proxy, contID)
+	}
+
+	return true
+}
+
+func (proxy *proxy) storeState() {
+	proxyStateFilePath := storeStateDir + "proxy_state.json"
+	proxy.Lock()
+	defer proxy.Unlock()
+
+	// if there are 0 VMs then remove state from disk
+	if (len(proxy.vms)) == 0 {
+		if err := os.Remove(proxyStateFilePath); err != nil {
+			proxyLog.Errorf("Couldn't remove %s: %v",
+				proxyStateFilePath, err)
+		}
+		return
+	}
+
+	proxyState := &proxyStateOnDisk{
+		Version:         Version,
+		SocketPath:      proxy.socketPath,
+		EnableVMConsole: proxy.enableVMConsole,
+		ContainerIDs:    make([]string, 0, len(proxy.vms)),
+	}
+	for cid := range proxy.vms {
+		proxyState.ContainerIDs = append(proxyState.ContainerIDs, cid)
+	}
+
+	data, err := json.MarshalIndent(proxyState, "", "\t")
+	if err != nil {
+		proxyLog.Errorf("Couldn't marshal proxy state %+v", proxyState)
+	}
+	err = ioutil.WriteFile(proxyStateFilePath, data, proxyStateFilesPerm)
+	if err != nil {
+		proxyLog.Errorf("Couldn't store proxy state to %s: %v",
+			proxyStateFilePath, err)
+	}
+}
+
+func vmStateFilePath(id string) string {
+	return storeStateDir + "vm_" + id + ".json"
+}
+
+func storeVMState(vm *vm, tokens []string) {
+	odVM := vmStateOnDisk{
+		RegisterVM: api.RegisterVM{
+			ContainerID: vm.containerID,
+			CtlSerial:   vm.hyperHandler.GetCtlSockPath(),
+			IoSerial:    vm.hyperHandler.GetIoSockPath(),
+			Console:     vm.console.socketPath,
+		},
+		Tokens: tokens,
+	}
+	o, err := json.MarshalIndent(&odVM, "", "\t")
+	if err != nil {
+		proxyLog.WithField("vm", vm.containerID).Warnf(
+			"Couldn't marshal VM state: %v", err)
+		return
+	}
+	storeFile := vmStateFilePath(vm.containerID)
+	err = ioutil.WriteFile(storeFile, o, proxyStateFilesPerm)
+	if err != nil {
+		proxyLog.WithField("vm", vm.containerID).Warnf(
+			"Couldn't store VM state to %s: %v", storeFile, err)
+	}
+}
+
+func delVMAndState(proxy *proxy, vm *vm) {
+	if proxy == nil {
+		proxyLog.Error("proxy parameter must be not nil")
+		return
+	}
+	if vm == nil {
+		proxyLog.Error("vm parameter must be not nil")
+		return
+	}
+	proxyLog.Infof("Removing on-disk state of %s", vm.containerID)
+	proxy.Lock()
+	delete(proxy.vms, vm.containerID)
+	proxy.Unlock()
+	proxy.storeState()
+	storeFile := vmStateFilePath(vm.containerID)
+	if err := os.Remove(storeFile); err != nil {
+		proxyLog.WithField("vm", vm.containerID).Warnf(
+			"Couldn't remove file %s: %v", storeFile, err)
+	}
+}
+
+func readVMState(containerID string) *vmStateOnDisk {
+	if containerID == "" {
+		proxyLog.Errorf("containerID parameter must be not empty")
+		return nil
+	}
+	vmStateFilePath := vmStateFilePath(containerID)
+	fdata, err := ioutil.ReadFile(vmStateFilePath)
+	if err != nil {
+		proxyLog.Errorf("Couldn't read %s: %v", vmStateFilePath, err)
+		return nil
+	}
+
+	var vmState vmStateOnDisk
+	err = json.Unmarshal(fdata, &vmState)
+	if err != nil {
+		proxyLog.Errorf("Couldn't unmarshal %s: %v", vmStateFilePath, err)
+		return nil
+	}
+	proxyLog.Debugf("restoring vm state: %+v", vmState)
+	return &vmState
+}
+
+func restoreTokens(proxy *proxy, vmState *vmStateOnDisk, vm *vm) {
+	if err := proxy.restoreTokens(vm, vmState.Tokens); err != nil {
+		proxyLog.Errorf("Failed to restore tokens: %v", err)
+		return
+	}
+
+	for _, token := range vmState.Tokens {
+		session := vm.findSessionByToken(Token(token))
+		if session == nil {
+			proxyLog.Errorf("Session must be not nil")
+			delVMAndState(proxy, vm)
+			return
+		}
+		if err := session.WaitForShim(); err != nil {
+			proxyLog.Errorf("Failed to re-connect with shim: %v", err)
+			delVMAndState(proxy, vm)
+			return
+		}
+	}
+}
+
+func restoreVMState(proxy *proxy, containerID string) {
+	if proxy == nil {
+		proxyLog.Errorf("proxy parameter must be not nil")
+		return
+	}
+
+	vmState := readVMState(containerID)
+	if vmState == nil {
+		return
+	}
+
+	regVM := vmState.RegisterVM
+	if regVM.ContainerID == "" || regVM.CtlSerial == "" || regVM.IoSerial == "" {
+		proxyLog.Errorf("wrong VM parameters")
+		return
+	}
+
+	proxy.Lock()
+	if _, ok := proxy.vms[regVM.ContainerID]; ok {
+		proxy.Unlock()
+		proxyLog.Errorf("%s: container already registered", regVM.ContainerID)
+		return
+	}
+	vm := newVM(regVM.ContainerID, regVM.CtlSerial, regVM.IoSerial)
+	proxy.vms[regVM.ContainerID] = vm
+	proxy.Unlock()
+
+	proxyLog.Infof("restoreVMState(containerId=%s,ctlSerial=%s,ioSerial=%s,console=%s)",
+		regVM.ContainerID, regVM.CtlSerial, regVM.IoSerial, regVM.Console)
+
+	if regVM.Console != "" && proxy.enableVMConsole {
+		vm.setConsole(regVM.Console)
+	}
+
+	restoreTokens(proxy, vmState, vm)
+	if err := vm.Reconnect(true); err != nil {
+		proxyLog.Errorf("Failed to connect: %v", err)
+		delVMAndState(proxy, vm)
+		return
+	}
+
+	// We start one goroutine per-VM to monitor the qemu process
+	proxy.wg.Add(1)
+	go func() {
+		<-vm.OnVMLost()
+		vm.Close()
+		proxy.wg.Done()
+	}()
+}
+
 // "RegisterVM"
 func registerVM(data []byte, userData interface{}, response *handlerResponse) {
 	client := userData.(*client)
@@ -247,6 +519,8 @@ func registerVM(data []byte, userData interface{}, response *handlerResponse) {
 	}
 
 	client.vm = vm
+	storeVMState(vm, io.Tokens)
+	proxy.storeState()
 
 	if proxyKSM != nil {
 		proxyKSM.kick()
@@ -323,9 +597,7 @@ func unregisterVM(data []byte, userData interface{}, response *handlerResponse)
 
 	client.log.Info("UnregisterVM()")
 
-	proxy.Lock()
-	delete(proxy.vms, vm.containerID)
-	proxy.Unlock()
+	delVMAndState(proxy, vm)
 
 	client.vm = nil
 }
@@ -617,12 +889,13 @@ func (proxy *proxy) init() error {
 	var l net.Listener
 	var err error
 
-	// flags
-	proxy.enableVMConsole = logrus.GetLevel() == logrus.DebugLevel
+	if !proxy.restoreState() {
+		// flags
+		proxy.enableVMConsole = logrus.GetLevel() == logrus.DebugLevel
 
-	// Open the proxy socket
-	if proxy.socketPath, err = getSocketPath(); err != nil {
-		return fmt.Errorf("couldn't get a rigth socket path: %v", err)
+		if proxy.socketPath, err = getSocketPath(); err != nil {
+			return fmt.Errorf("couldn't get a right socket path: %v", err)
+		}
 	}
 	fds := listenFds()