Merge pull request #683 from austinvazquez/wait-for-vm-up

fix: extend vsock ack message timeout for remote snapshot dials

Author: austinvazquez

.buildkite/pipeline.yml

@@ -100,12 +100,13 @@ steps:
     env:
       DOCKER_IMAGE_TAG: "$BUILDKITE_BUILD_NUMBER"
       NUMBER_OF_VMS: 10
-      EXTRAGOARGS: "-v -count=1 -race"
+      EXTRAGOARGS: "-v -count=1 -race -timeout 3m"
       FICD_DM_VOLUME_GROUP: fcci-vg
     artifact_paths:
       - "snapshotter/logs/*"
     command:
       - make -C snapshotter integ-test FICD_DM_POOL=build_${BUILDKITE_BUILD_NUMBER}_snapshotter
+    timeout_in_minutes: 10
 
   - wait
 

snapshotter/app/service.go

@@ -22,6 +22,7 @@ import (
 	"strconv"
 	"strings"
 	"syscall"
+	"time"
 
 	snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
 	"github.com/containerd/containerd/contrib/snapshotservice"
@@ -175,8 +176,10 @@ func initSnapshotter(ctx context.Context, config config.Config, cache cache.Cach
 		if err != nil {
 			return nil, err
 		}
+
+		ackMsgTimeout := time.Duration(config.Snapshotter.Dialer.AckMsgTimeoutInSeconds) * time.Second
 		snapshotterDialer := func(ctx context.Context, namespace string) (net.Conn, error) {
-			return vsock.DialContext(ctx, host, uint32(port), vsock.WithLogger(log.G(ctx)))
+			return vsock.DialContext(ctx, host, uint32(port), vsock.WithLogger(log.G(ctx)), vsock.WithAckMsgTimeout(ackMsgTimeout))
 		}
 
 		var metricsProxy *metrics.Proxy

snapshotter/config/config.go

@@ -32,6 +32,7 @@ type Config struct {
 
 type snapshotter struct {
 	Listener listener `toml:"listener"`
+	Dialer   dialer   `toml:"dialer"`
 	Proxy    proxy    `toml:"proxy"`
 	Metrics  metrics  `toml:"metrics"`
 }
@@ -41,6 +42,10 @@ type listener struct {
 	Address string `toml:"address" default:"/var/lib/demux-snapshotter/snapshotter.sock"`
 }
 
+type dialer struct {
+	AckMsgTimeoutInSeconds int `toml:"ack_msg_timeout_in_seconds" default:"1"`
+}
+
 type proxy struct {
 	Address address `toml:"address"`
 }

snapshotter/config/config_test.go

@@ -62,6 +62,9 @@ func defaultConfig() error {
 				Network: "unix",
 				Address: "/var/lib/demux-snapshotter/snapshotter.sock",
 			},
+			Dialer: dialer{
+				AckMsgTimeoutInSeconds: 1,
+			},
 			Metrics: metrics{
 				Enable: false,
 			},
@@ -79,6 +82,8 @@ func parseExampleConfig() error {
 	  [snapshotter.listener]
 	    network = "unix"
 	    address = "/var/lib/demux-snapshotter/non-default-snapshotter.vsock"
+	  [snapshotter.dialer]
+	    ack_msg_timeout_in_seconds = 4
 	  [snapshotter.proxy.address.resolver]
 	    type = "http"
 	    address = "localhost:10001"
@@ -96,6 +101,9 @@ func parseExampleConfig() error {
 				Network: "unix",
 				Address: "/var/lib/demux-snapshotter/non-default-snapshotter.vsock",
 			},
+			Dialer: dialer{
+				AckMsgTimeoutInSeconds: 4,
+			},
 			Proxy: proxy{
 				Address: address{
 					Resolver: resolver{

snapshotter/service_integ_test.go

@@ -18,7 +18,6 @@ import (
 	"fmt"
 	"strconv"
 	"testing"
-	"time"
 
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/namespaces"
@@ -53,23 +52,14 @@ func TestLaunchContainerWithRemoteSnapshotter_Isolated(t *testing.T) {
 	integtest.Prepare(t, integtest.WithDefaultNetwork())
 
 	vmID := 0
-
-	testTimeout := 300 * time.Second
-	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
-	defer cancel()
-
-	err := launchContainerWithRemoteSnapshotterInVM(ctx, strconv.Itoa(vmID))
+	err := launchContainerWithRemoteSnapshotterInVM(context.Background(), strconv.Itoa(vmID))
 	require.NoError(t, err)
 }
 
 func TestLaunchMultipleContainersWithRemoteSnapshotter_Isolated(t *testing.T) {
 	integtest.Prepare(t, integtest.WithDefaultNetwork())
 
-	testTimeout := 600 * time.Second
-	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
-	defer cancel()
-
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := errgroup.WithContext(context.Background())
 
 	numberOfVms := integtest.NumberOfVms
 	for vmID := 0; vmID < numberOfVms; vmID++ {
@@ -126,7 +116,7 @@ func launchContainerWithRemoteSnapshotterInVM(ctx context.Context, vmID string)
 		ContainerCount: 1,
 	})
 	if err != nil {
-		return fmt.Errorf("Failed to create microVM[%s] [%v]", vmID, err)
+		return fmt.Errorf("Failed to create VM[%s]: %w", vmID, err)
 	}
 	defer fcClient.StopVM(ctx, &proto.StopVMRequest{VMID: vmID})
 
@@ -135,16 +125,16 @@ func launchContainerWithRemoteSnapshotterInVM(ctx context.Context, vmID string)
 		Metadata: fmt.Sprintf(dockerMetadataTemplate, "ghcr.io", noAuth, noAuth),
 	})
 	if err != nil {
-		return fmt.Errorf("Failed to configure VM metadata for registry resolution [%v]", err)
+		return fmt.Errorf("Failed to configure VM metadata for registry resolution: %w", err)
 	}
 
 	image, err := client.Pull(ctx, al2stargz,
 		containerd.WithPullUnpack,
 		containerd.WithPullSnapshotter(snapshotterName),
-		containerd.WithImageHandlerWrapper(source.AppendDefaultLabelsHandlerWrapper(al2stargz, 10*1024*1024)),
+		containerd.WithImageHandlerWrapper(source.AppendDefaultLabelsHandlerWrapper(al2stargz, 10*mib)),
 	)
 	if err != nil {
-		return fmt.Errorf("Failed to pull image for VM: %s [%v]", vmID, err)
+		return fmt.Errorf("Failed to pull image for VM[%s]: %w", vmID, err)
 	}
 	defer client.ImageService().Delete(ctx, image.Name())
 
@@ -160,13 +150,13 @@ func launchContainerWithRemoteSnapshotterInVM(ctx context.Context, vmID string)
 		),
 	)
 	if err != nil {
-		return fmt.Errorf("Failed to create container in VM: %s, [%v]", vmID, err)
+		return fmt.Errorf("Failed to create container in VM[%s]: %w", vmID, err)
 	}
 	defer container.Delete(ctx, containerd.WithSnapshotCleanup)
 
 	_, err = integtest.RunTask(ctx, container)
 	if err != nil {
-		return fmt.Errorf("Failed to run task in VM: %s [%v]", vmID, err)
+		return fmt.Errorf("Failed to run task in VM[%s]: %w", vmID, err)
 	}
 	return nil
 }

tools/docker/entrypoint.sh

@@ -42,6 +42,9 @@ EOF
 
 mkdir -p /etc/demux-snapshotter /var/lib/demux-snapshotter
 cat > /etc/demux-snapshotter/config.toml <<EOF
+[snapshotter.dialer]
+  ack_msg_timeout_in_seconds = 2
+
 [snapshotter.proxy.address.resolver]
   type = "http"
   address = "http://127.0.0.1:10001"