Add standby instance support for Vetu (#719)

Author: edigaryev

.cirrus.yml

@@ -57,6 +57,36 @@ task:
   test_script:
     - go test -v -p 1 ./...
 
+task:
+  name: Test (Linux with Vetu)
+  alias: Tests
+  aws_credentials:
+    role_arn: arn:aws:iam::944424729675:role/VetuIntegrationTests
+    role_session_name: cirrus
+    region: us-east-2
+  ec2_instance:
+    image: ubuntu/images/hvm-ssd/ubuntu-jammy-22.04-arm64-server-*
+    architecture: arm64
+    region: us-east-2
+    type: a1.metal
+    block_device_mappings:
+      - device_name: /dev/sda1
+        ebs:
+          volume_size: 100
+  install_vetu_script:
+    - sudo apt-get update && sudo apt-get -y install apt-transport-https ca-certificates
+    - echo "deb [trusted=yes] https://apt.fury.io/cirruslabs/ /" | sudo tee /etc/apt/sources.list.d/cirruslabs.list
+    - sudo apt-get update && sudo apt-get -y install vetu
+  test_script:
+    - wget --no-verbose -O - https://go.dev/dl/go1.22.2.linux-arm64.tar.gz | tar -C /usr/local -xz
+    - export PATH=$PATH:/usr/local/go/bin
+    - go test -v -p 1 ./...
+  env:
+    CIRRUS_INTERNAL_VETU_VM: ghcr.io/cirruslabs/ubuntu-runner-arm64:latest
+    CIRRUS_INTERNAL_VETU_SSH_PASSWORD: admin
+    CIRRUS_INTERNAL_VETU_SSH_USER: admin
+    HOME: /root
+
 task:
   name: Release (Dry Run)
   only_if: $CIRRUS_TAG == ''

PERSISTENT-WORKERS.md

@@ -306,4 +306,22 @@ persistent_worker:
       memory: 12
 ```
 
-Currently only Tart isolation is supported.
+Vetu configuration on `arm64` is similar:
+
+```yaml
+standby:
+  isolation:
+    vetu:
+      image: ghcr.io/cirruslabs/ubuntu-runner-arm64:latest
+      user: admin
+      password: admin
+      cpu: 16
+      memory: 48
+      networking:
+        host: {}
+      disk_size: 100
+```
+
+On `amd64`, simply replace the `image` with `ghcr.io/cirruslabs/ubuntu-runner-amd64:latest`.
+
+Currently only Tart and Vetu isolations are supported for standby.

internal/commands/run.go

@@ -67,13 +67,11 @@ var vetuLazyPull bool
 // Flags useful for debugging.
 var debugNoCleanup bool
 
-var baseEnvironment map[string]string = eenvironment.Merge(
-	eenvironment.Static(),
-	eenvironment.BuildID(),
-	eenvironment.ProjectSpecific(projectDir),
-)
-
-func readYaml(ctx context.Context, userSpecifiedEnvironment map[string]string) (*parser.Result, error) {
+func readYaml(
+	ctx context.Context,
+	baseEnvironment map[string]string,
+	userSpecifiedEnvironment map[string]string,
+) (*parser.Result, error) {
 	// Retrieve the combined YAML configuration
 	combinedYAML, err := helpers.ReadCombinedConfig(
 		ctx,
@@ -119,6 +117,8 @@ func readYaml(ctx context.Context, userSpecifiedEnvironment map[string]string) (
 }
 
 func run(cmd *cobra.Command, args []string) error {
+	baseEnvironment := makeBaseEnvironment()
+
 	userSpecifiedEnvironment, err := makeUserSpecifiedEnvironment()
 	if err != nil {
 		return fmt.Errorf("%v: %v", ErrRun, err)
@@ -127,7 +127,7 @@ func run(cmd *cobra.Command, args []string) error {
 	// https://github.com/spf13/cobra/issues/340#issuecomment-374617413
 	cmd.SilenceUsage = true
 
-	result, err := readYaml(cmd.Context(), userSpecifiedEnvironment)
+	result, err := readYaml(cmd.Context(), baseEnvironment, userSpecifiedEnvironment)
 	if err != nil {
 		return err
 	}
@@ -212,12 +212,14 @@ func newRunCmd() *cobra.Command {
 		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 			completions := []string{}
 
+			baseEnvironment := makeBaseEnvironment()
+
 			userSpecifiedEnvironment, err := makeUserSpecifiedEnvironment()
 			if err != nil {
 				return completions, cobra.ShellCompDirectiveError
 			}
 
-			result, err := readYaml(cmd.Context(), userSpecifiedEnvironment)
+			result, err := readYaml(cmd.Context(), baseEnvironment, userSpecifiedEnvironment)
 			if err != nil {
 				return completions, cobra.ShellCompDirectiveError
 			}
@@ -297,6 +299,14 @@ func newRunCmd() *cobra.Command {
 	return cmd
 }
 
+func makeBaseEnvironment() map[string]string {
+	return eenvironment.Merge(
+		eenvironment.Static(),
+		eenvironment.BuildID(),
+		eenvironment.ProjectSpecific(projectDir),
+	)
+}
+
 func makeUserSpecifiedEnvironment() (map[string]string, error) {
 	var result map[string]string
 	var err error

internal/commands/run_test.go

@@ -286,6 +286,10 @@ func TestRunYAMLAndStarlarkHooks(t *testing.T) {
 
 // TestRunContainerPull ensures that container images are pulled by default.
 func TestRunContainerPull(t *testing.T) {
+	if _, ok := os.LookupEnv("CIRRUS_CONTAINER_BACKEND"); !ok {
+		t.Skip("no container backend configured")
+	}
+
 	backend, err := containerbackend.New(containerbackend.BackendTypeAuto)
 	if err != nil {
 		t.Fatal(err)
@@ -432,6 +436,10 @@ func TestRunPrebuiltImageTemplate(t *testing.T) {
 }
 
 func TestAffectedFiles(t *testing.T) {
+	if _, ok := os.LookupEnv("CIRRUS_CONTAINER_BACKEND"); !ok {
+		t.Skip("no container backend configured")
+	}
+
 	testutil.TempChdirPopulatedWith(t, "testdata/run-affected-files")
 
 	// Create os.Stderr writer that duplicates it's output to buf
@@ -464,6 +472,10 @@ func TestHasStaticEnvironment(t *testing.T) {
 }
 
 func TestRunGitHubAnnotations(t *testing.T) {
+	if _, ok := os.LookupEnv("CIRRUS_CONTAINER_BACKEND"); !ok {
+		t.Skip("no container backend configured")
+	}
+
 	testutil.TempChdirPopulatedWith(t, "testdata/run-github-annotations")
 
 	t.Setenv("GITHUB_ACTIONS", "true")
@@ -492,6 +504,10 @@ func TestRunGitHubAnnotations(t *testing.T) {
 }
 
 func TestRunArtifactsDir(t *testing.T) {
+	if _, ok := os.LookupEnv("CIRRUS_CONTAINER_BACKEND"); !ok {
+		t.Skip("no container backend configured")
+	}
+
 	testutil.TempChdirPopulatedWith(t, "testdata/run-artifacts-dir")
 
 	// Create os.Stderr writer that duplicates it's output to buf

internal/executor/instance/abstract/abstract.go

@@ -8,13 +8,13 @@ import (
 )
 
 type Instance interface {
-	Run(context.Context, *runconfig.RunConfig) error
+	Run(ctx context.Context, config *runconfig.RunConfig) error
 	WorkingDirectory(projectDir string, dirtyMode bool) string
 	Close(ctx context.Context) error
 	Attributes() []attribute.KeyValue
 }
 
 type WarmableInstance interface {
 	// Warmup can be optionally called in case of a persistent worker is configured to be warm
-	Warmup(context.Context, string, map[string]string, *echelon.Logger) error
+	Warmup(ctx context.Context, ident string, env map[string]string, logger *echelon.Logger) error
 }

internal/executor/instance/persistentworker/isolation/vetu/vetu.go

@@ -9,12 +9,14 @@ import (
 	"github.com/cirruslabs/cirrus-cli/internal/executor/instance/runconfig"
 	"github.com/cirruslabs/cirrus-cli/internal/executor/platform"
 	"github.com/cirruslabs/cirrus-cli/internal/logger"
+	"github.com/cirruslabs/echelon"
 	"github.com/getsentry/sentry-go"
 	"github.com/google/uuid"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/crypto/ssh"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -39,6 +41,8 @@ type Vetu struct {
 	diskSize         uint32
 	bridgedInterface string
 	hostNetworking   bool
+
+	vm *VM
 }
 
 func New(
@@ -83,72 +87,93 @@ func (vetu *Vetu) Attributes() []attribute.KeyValue {
 	}
 }
 
-func (vetu *Vetu) Run(ctx context.Context, config *runconfig.RunConfig) error {
+func (vetu *Vetu) Warmup(
+	ctx context.Context,
+	ident string,
+	env map[string]string,
+	logger *echelon.Logger,
+) error {
+	return vetu.bootVM(ctx, ident, env, false, logger)
+}
+
+func (vetu *Vetu) bootVM(
+	ctx context.Context,
+	ident string,
+	env map[string]string,
+	lazyPull bool,
+	logger *echelon.Logger,
+) error {
 	ctx, prepareInstanceSpan := tracer.Start(ctx, "prepare-instance")
 	defer prepareInstanceSpan.End()
 
-	tmpVMName := fmt.Sprintf("%s%d-", vmNamePrefix, config.TaskID) + uuid.NewString()
-	vm, err := NewVMClonedFrom(ctx,
-		vetu.vmName, tmpVMName,
-		config.VetuOptions.LazyPull,
-		config.AdditionalEnvironment,
-		config.Logger(),
-	)
+	var identToBeInjected string
+	if ident != "" {
+		identToBeInjected = fmt.Sprintf("%s-", ident)
+	}
+
+	tmpVMName := vmNamePrefix + identToBeInjected + uuid.NewString()
+
+	vm, err := NewVMClonedFrom(ctx, vetu.vmName, tmpVMName, lazyPull, env, logger)
 	if err != nil {
 		return fmt.Errorf("%w: failed to create VM cloned from %q: %v", ErrFailed, vetu.vmName, err)
 	}
-	defer func() {
-		if localHub := sentry.GetHubFromContext(ctx); localHub != nil {
-			localHub.AddBreadcrumb(&sentry.Breadcrumb{
-				Message: fmt.Sprintf("stopping and deleting the VM %s", vm.ident),
-			}, nil)
-		}
 
-		_ = vm.Close()
-	}()
+	vetu.vm = vm
 
-	if err := vm.Configure(ctx, vetu.cpu, vetu.memory, vetu.diskSize, config.Logger()); err != nil {
+	if err := vm.Configure(ctx, vetu.cpu, vetu.memory, vetu.diskSize, logger); err != nil {
 		return fmt.Errorf("%w: failed to configure VM %q: %v", ErrFailed, vm.Ident(), err)
 	}
 
 	// Start the VM (asynchronously)
 	vm.Start(ctx, vetu.bridgedInterface, vetu.hostNetworking)
 
-	// Wait for the VM to start and get its IP address
-	bootLogger := config.Logger().Scoped("boot virtual machine")
+	// Wait for the VM to start and get it's DHCP address
+	bootLogger := logger.Scoped("boot virtual machine")
 
-	var ip string
+	ipCtx, ipCtxCancel := context.WithTimeoutCause(ctx, 10*time.Minute,
+		fmt.Errorf("timed out while trying to retrieve the VM %s IP", vm.Ident()))
+	defer ipCtxCancel()
 
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case err := <-vm.ErrChan():
-			return err
-		default:
-			time.Sleep(time.Second)
-		}
+	ip, err := vetu.retrieveIPLoop(ipCtx, vm)
+	if err != nil {
+		return err
+	}
 
-		ip, err = vm.RetrieveIP(ctx)
-		if err != nil {
-			vetu.logger.Debugf("failed to retrieve VM %s IP: %v\n", vm.Ident(), err)
-			continue
-		}
+	bootLogger.Errorf("VM was assigned with %s IP", ip)
 
-		break
+	sshClient, err := remoteagent.WaitForSSH(ipCtx, fmt.Sprintf("%s:%d", ip, vetu.sshPort), vetu.sshUser,
+		vetu.sshPassword, logger)
+	if err != nil {
+		return err
 	}
+	_ = sshClient.Close()
 
-	vetu.logger.Debugf("IP %s retrieved from VM %s, running agent...", ip, vm.Ident())
-
-	bootLogger.Errorf("VM was assigned with %s IP", ip)
 	bootLogger.Finish(true)
 
-	prepareInstanceSpan.End()
+	return nil
+}
+
+func (vetu *Vetu) Run(ctx context.Context, config *runconfig.RunConfig) error {
+	if vetu.vm == nil {
+		err := vetu.bootVM(ctx, strconv.FormatInt(config.TaskID, 10), config.AdditionalEnvironment,
+			config.VetuOptions.LazyPull, config.Logger())
+		if err != nil {
+			return err
+		}
+	}
+
+	ip, err := vetu.vm.RetrieveIP(ctx)
+	if err != nil {
+		return err
+	}
+
+	// Wait for the VM to start and get its IP address
+	vetu.logger.Debugf("IP %s retrieved from VM %s, running agent...", ip, vetu.vm.Ident())
 
 	err = remoteagent.WaitForAgent(ctx, vetu.logger, fmt.Sprintf("%s:%d", ip, vetu.sshPort),
 		vetu.sshUser, vetu.sshPassword, "linux", runtime.GOARCH,
 		config, true, vetu.initializeHooks(config), nil, "",
-		map[string]string{"CIRRUS_VM_ID": vm.Ident()})
+		map[string]string{"CIRRUS_VM_ID": vetu.vm.Ident()})
 	if err != nil {
 		return err
 	}
@@ -164,8 +189,18 @@ func (vetu *Vetu) WorkingDirectory(projectDir string, dirtyMode bool) string {
 	return platform.NewUnix().GenericWorkingDir()
 }
 
-func (vetu *Vetu) Close(context.Context) error {
-	return nil
+func (vetu *Vetu) Close(ctx context.Context) error {
+	if vetu.vm == nil {
+		return nil
+	}
+
+	if localHub := sentry.GetHubFromContext(ctx); localHub != nil {
+		localHub.AddBreadcrumb(&sentry.Breadcrumb{
+			Message: fmt.Sprintf("stopping and deleting the VM %s", vetu.vm.ident),
+		}, nil)
+	}
+
+	return vetu.vm.Close()
 }
 
 func Cleanup() error {
@@ -208,3 +243,24 @@ func (vetu *Vetu) initializeHooks(config *runconfig.RunConfig) []remoteagent.Wai
 
 	return hooks
 }
+
+func (vetu *Vetu) retrieveIPLoop(ctx context.Context, vm *VM) (string, error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case err := <-vm.ErrChan():
+			return "", err
+		default:
+			time.Sleep(time.Second)
+		}
+
+		ip, err := vm.RetrieveIP(ctx)
+		if err != nil {
+			vetu.logger.Debugf("failed to retrieve VM %s IP: %v\n", vm.Ident(), err)
+			continue
+		}
+
+		return ip, nil
+	}
+}