Fix Virtlet restarts

cmd/vmwrapper/vmwrapper.go

@@ -71,13 +71,17 @@ func main() {
  }
  }
 
- // FIXME: move the pid of qemu instance out of /kubepods/podxxxxxxx
- // for some cases it will be killed by kubelet after the virtlet pod is deleted/recreated
+ // FIXME: move the pid of qemu instance out of kubelet-managed
+ // for cgroups that aren't managed by libvirt.
+ // If we don't do this, the VM pod will be killed by kubelet when Virtlet pod
+ // is removed dnd cgroup-per-qos is enabled in kubelet settings.
  cm := cgroups.NewManager(os.Getpid(), nil)
- if _, err := cm.GetProcessController("hugetlb"); err == nil {
- err = cm.MoveProcess("hugetlb", "/")
- if err != nil {
- glog.Warningf("failed to move pid into hugetlb path /: %v", err)
+ for _, ctl := range []string{"hugetlb", "systemd", "pids"} {
+ if _, err := cm.GetProcessController(ctl); err == nil {
+ err = cm.MoveProcess(ctl, "/")
+ if err != nil {
+ glog.Warningf("failed to move pid into cgroup %q path /: %v", ctl, err)
+ }
  }
  }
 

pkg/network/csn.go

@@ -42,7 +42,9 @@ type InterfaceDescription struct {
  // namespace or to control file in sysfs for sr-iov VF.
  // It may be nil if the interface was recovered after restarting Virtlet.
  // It's only needed during the initial VM startup.
- Fo *os.File
+ // The json tag is here so that bogus File object doesn't get stored
+ // in the metadata db.
+ Fo *os.File `json:"-"`
  // Name contains original interface name for sr-iov interface.
  Name string
  // HardwareAddr contains original hardware address for CNI-created

pkg/tapmanager/tapfdsource.go

@@ -341,8 +341,8 @@ func (s *TapFDSource) Recover(key string, data []byte) error {
  })
 }
 
-// RetrieveFDs retrieve the FDs
-// It is only the case if VM exited but recover didn't populate the FDs
+// RetrieveFDs retrieves the FDs.
+// It's only used in case if VM exited but Recover() didn't populate the FDs
 func (s *TapFDSource) RetrieveFDs(key string) ([]int, error) {
  var podNet *podNetwork
  var fds []int

pkg/utils/cgroups/controllers.go

@@ -94,8 +94,15 @@ func (c *RealManager) GetProcessControllers() (map[string]string, error) {
  // "6:memory:/user.slice/user-xxx.slice/session-xx.scope"
  parts := strings.SplitN(line, ":", 3)
 
+ name := parts[1]
+ if strings.HasPrefix(name, "name=") {
+ // Handle named cgroup hierarchies like name=systemd
+ // The corresponding directory tree will be /sys/fs/cgroup/systemd
+ name = name[5:]
+ }
+
  // use second part as controller name and third as its path
- ctrls[parts[1]] = parts[2]
+ ctrls[name] = parts[2]
 
  if err == io.EOF {
  break

tests/e2e/basic_test.go

@@ -29,14 +29,14 @@ import (
  . "github.com/Mirantis/virtlet/tests/e2e/ginkgo-ext"
 )
 
-var _ = Describe("Virtlet [Basic cirros tests]", func() {
+var _ = Describe("Virtlet [Basic tests]", func() {
  var (
  vm *framework.VMInterface
  vmPod *framework.PodInterface
  )
 
  BeforeAll(func() {
- vm = controller.VM("cirros-vm")
+ vm = controller.VM("test-vm")
  Expect(vm.CreateAndWait(VMOptions{}.ApplyDefaults(), time.Minute*5, nil)).To(Succeed())
  var err error
  vmPod, err = vm.Pod()
@@ -69,6 +69,7 @@ var _ = Describe("Virtlet [Basic cirros tests]", func() {
  var (
  logPath string
  nodeExecutor framework.Executor
+ ssh framework.Executor
  )
 
  BeforeAll(func() {
@@ -85,29 +86,33 @@ var _ = Describe("Virtlet [Basic cirros tests]", func() {
  }
  }
  Expect(logPath).NotTo(BeEmpty())
+ ssh = waitSSH(vm)
+ _, err = framework.RunSimple(ssh, "echo ++foo++ | sudo tee /dev/console")
+ Expect(err).NotTo(HaveOccurred())
  })
 
- It("Should contain login string in pod log and each line of that log must be a valid JSON", func() {
- Eventually(func() error {
+ It("Should contain the echoed string and each line of the log must be a valid JSON", func() {
+ Eventually(func() (string, error) {
  out, err := framework.RunSimple(nodeExecutor, "cat", logPath)
  if err != nil {
- return err
+ return "", err
  }
- found := 0
+ var b strings.Builder
  for _, line := range strings.Split(out, "\n") {
  var entry map[string]string
  if err := json.Unmarshal([]byte(line), &entry); err != nil {
- return fmt.Errorf("error unmarshalling json: %v", err)
+ return "", fmt.Errorf("error unmarshalling json: %v", err)
  }
- if strings.HasPrefix(entry["log"], "login as 'cirros' user. default password") {
- found++
- }
- }
- if found != 1 {
- return fmt.Errorf("expected login prompt to appear exactly once in the log, but got %d occurrences", found)
+ b.WriteString(line + "\n")
  }
- return nil
- })
+ return b.String(), nil
+ }, 120, 5).Should(ContainSubstring("++foo++"))
+ })
+
+ It("Should be readable via Kubernetes API", func() {
+ c, err := vmPod.Container("")
+ Expect(err).NotTo(HaveOccurred())
+ Eventually(c.Logs, 120, 5).Should(ContainSubstring("++foo++"))
  })
  })
 

tests/e2e/common.go

@@ -141,15 +141,26 @@ func deleteVM(vm *framework.VMInterface) {
  }
 }
 
-// do asserts that function with multiple return values doesn't fail
-// considering we have func `foo(something) (something, error)`
+// do asserts that function with multiple return values doesn't fail.
+// Considering we have func `foo(something) (something, error)`:
 //
 // `x := do(foo(something))` is equivalent to
 // val, err := fn(something)
 // Expect(err).To(Succeed())
 // x = val
+//
+// The rule is that the function must return at least 2 values,
+// of which the first one is returned as the first value of do(),
+// and the last value is interpreted as error (the second value of do()).
 func do(value interface{}, extra ...interface{}) interface{} {
- ExpectWithOffset(1, value, extra...).To(BeAnything())
+ if len(extra) == 0 {
+ panic("bad usage of do() -- no extra values")
+ }
+ lastValue := extra[len(extra)-1]
+ if lastValue != nil {
+ err := lastValue.(error)
+ Expect(err).NotTo(HaveOccurred())
+ }
  return value
 }
 
@@ -323,7 +334,7 @@ func withCeph(monitorIP, secret *string, kubeSecret string) {
  return err
  }
  return fmt.Errorf("secret %s was not deleted", kubeSecret)
- })
+ }).Should(Succeed())
  }
  })
 }

tests/e2e/framework/common.go

@@ -54,6 +54,7 @@ type Executor interface {
  io.Closer
  Run(stdin io.Reader, stdout, stderr io.Writer, command ...string) error
  Start(stdin io.Reader, stdout, stderr io.Writer, command ...string) (Command, error)
+ Logs() (string, error)
 }
 
 // Run executes command with the given executor, returns stdout/stderr as strings

tests/e2e/framework/docker_interface.go

@@ -18,6 +18,7 @@ package framework
 
 import (
  "context"
+ "errors"
  "fmt"
  "io"
  "io/ioutil"
@@ -198,7 +199,12 @@ func (*DockerContainerExecInterface) Close() error {
 
 // Start is a placeholder for fulfilling Executor interface
 func (*DockerContainerExecInterface) Start(stdin io.Reader, stdout, stderr io.Writer, command ...string) (Command, error) {
- return nil, fmt.Errorf("Not Implemented")
+ return nil, errors.New("not implemented")
+}
+
+// Logs is a placeholder for fulfilling Executor interface
+func (*DockerContainerExecInterface) Logs() (string, error) {
+ return "", errors.New("not implemented")
 }
 
 func containerHandleDataPiping(resp types.HijackedResponse, inStream io.Reader, outStream, errorStream io.Writer) error {

tests/e2e/framework/localcmd_interface.go

@@ -2,6 +2,7 @@ package framework
 
 import (
  "context"
+ "errors"
  "io"
  "os/exec"
  "syscall"
@@ -52,6 +53,11 @@ func (l *LocalCmd) Close() error {
  return nil
 }
 
+// Logs is a placeholder for fulfilling Executor interface
+func (l *LocalCmd) Logs() (string, error) {
+ return "", errors.New("not implemented")
+}
+
 type localCommand struct {
  cmd *exec.Cmd
 }

tests/e2e/framework/pod_interface.go

@@ -21,6 +21,7 @@ import (
  "errors"
  "fmt"
  "io"
+ "io/ioutil"
  "net/http"
  "os"
  "os/signal"
@@ -162,7 +163,9 @@ func (pi *PodInterface) WaitForDestruction(timing ...time.Duration) error {
  }, timeout, pollPeriond, consistencyPeriod)
 }
 
-// Container returns interface to execute commands in one of pod's containers
+// Container returns an interface to handle one of the pod's
+// containers. If name is empty, it takes the first container
+// of the pod.
 func (pi *PodInterface) Container(name string) (Executor, error) {
  if name == "" && len(pi.Pod.Spec.Containers) > 0 {
  name = pi.Pod.Spec.Containers[0].Name
@@ -320,3 +323,28 @@ func (*containerInterface) Close() error {
 func (*containerInterface) Start(stdin io.Reader, stdout, stderr io.Writer, command ...string) (Command, error) {
  return nil, errors.New("Not Implemented")
 }
+
+// Logs returns the logs of the container as a string.
+func (ci *containerInterface) Logs() (string, error) {
+ restClient := ci.podInterface.controller.client.RESTClient()
+ req := restClient.Get().
+ Name(ci.podInterface.Pod.Name).
+ Namespace(ci.podInterface.Pod.Namespace).
+ Resource("pods").
+ SubResource("log")
+ req.VersionedParams(&v1.PodLogOptions{
+ Container: ci.name,
+ }, scheme.ParameterCodec)
+ stream, err := req.Stream()
+ if err != nil {
+ return "", err
+ }
+ defer stream.Close()
+
+ bs, err := ioutil.ReadAll(stream)
+ if err != nil {
+ return "", fmt.Errorf("ReadAll(): %v", err)
+ }
+
+ return string(bs), nil
+}

tests/e2e/framework/ssh_interface.go

@@ -17,6 +17,7 @@ limitations under the License.
 package framework
 
 import (
+ "errors"
  "fmt"
  "io"
  "strings"
@@ -116,6 +117,11 @@ func (si *sshInterface) Close() error {
  return nil
 }
 
+// Logs is a placeholder for fulfilling Executor interface
+func (*sshInterface) Logs() (string, error) {
+ return "", errors.New("not implemented")
+}
+
 type sshCommand struct {
  session *ssh.Session
 }

tests/e2e/ginkgo-ext/scopes.go

@@ -124,8 +124,8 @@ func AfterAll(body func()) bool {
  return true
 }
 
-//JustAfterEach runs the function just after each test, before all AfterEeach,
-//AfterFailed and AfterAll
+// JustAfterEach runs the function just after each test, before all AfterEach,
+// AfterFailed and AfterAll
 func JustAfterEach(body func()) bool {
  if currentScope != nil {
  if body == nil {

tests/e2e/resources_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package e2e
 
 import (
- "fmt"
  "regexp"
  "strconv"
  "time"
@@ -43,12 +42,12 @@ var _ = Describe("VM resources", func() {
  do(vm.Pod())
  })
 
+ scheduleWaitSSH(&vm, &ssh)
+
  AfterAll(func() {
  deleteVM(vm)
  })
 
- scheduleWaitSSH(&vm, &ssh)
-
  It("Should have CPU count as set for the domain [Conformance]", func() {
  checkCPUCount(vm, ssh, 2)
  })
@@ -67,17 +66,14 @@ var _ = Describe("VM resources", func() {
  })
 
  It("Should grow the root volume size if requested", func() {
- Eventually(func() error {
+ minSize := 3900
+ Eventually(func() (int, error) {
  sizeStr := do(framework.RunSimple(ssh, "/bin/sh", "-c", `df -m / | tail -1 | awk "{print \$2}"`)).(string)
  size, err := strconv.Atoi(sizeStr)
  if err != nil {
- return err
- }
- minSize := 3900
- if size < minSize {
- return fmt.Errorf("the size is %d but needs to be at least %d", size, minSize)
+ return 0, err
  }
- return nil
- })
+ return size, nil
+ }).Should(BeNumerically(">=", minSize))
  })
 })