X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/98db65de63c9e2acfeae6636ccc619171635bda0..245a36c76baa54ca6cf75f501d5367d976c68a38:/lib/dispatchcloud/test/stub_driver.go

diff --git a/lib/dispatchcloud/test/stub_driver.go b/lib/dispatchcloud/test/stub_driver.go
index 41eb20763c..037580f8d9 100644
--- a/lib/dispatchcloud/test/stub_driver.go
+++ b/lib/dispatchcloud/test/stub_driver.go
@@ -18,7 +18,9 @@ import (
 	"time"
 
 	"git.arvados.org/arvados.git/lib/cloud"
+	"git.arvados.org/arvados.git/lib/crunchrun"
 	"git.arvados.org/arvados.git/sdk/go/arvados"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/crypto/ssh"
 )
@@ -61,7 +63,7 @@ type StubDriver struct {
 }
 
 // InstanceSet returns a new *StubInstanceSet.
-func (sd *StubDriver) InstanceSet(params json.RawMessage, id cloud.InstanceSetID, _ cloud.SharedResourceTags, logger logrus.FieldLogger) (cloud.InstanceSet, error) {
+func (sd *StubDriver) InstanceSet(params json.RawMessage, id cloud.InstanceSetID, _ cloud.SharedResourceTags, logger logrus.FieldLogger, reg *prometheus.Registry) (cloud.InstanceSet, error) {
 	if sd.holdCloudOps == nil {
 		sd.holdCloudOps = make(chan bool)
 	}
@@ -107,7 +109,7 @@ type StubInstanceSet struct {
 	lastInstanceID     int
 }
 
-func (sis *StubInstanceSet) Create(it arvados.InstanceType, image cloud.ImageID, tags cloud.InstanceTags, cmd cloud.InitCommand, authKey ssh.PublicKey) (cloud.Instance, error) {
+func (sis *StubInstanceSet) Create(it arvados.InstanceType, image cloud.ImageID, tags cloud.InstanceTags, initCommand cloud.InitCommand, authKey ssh.PublicKey) (cloud.Instance, error) {
 	if sis.driver.HoldCloudOps {
 		sis.driver.holdCloudOps <- true
 	}
@@ -126,12 +128,12 @@ func (sis *StubInstanceSet) Create(it arvados.InstanceType, image cloud.ImageID,
 	}
 	sis.lastInstanceID++
 	svm := &StubVM{
+		InitCommand:  initCommand,
 		sis:          sis,
 		id:           cloud.InstanceID(fmt.Sprintf("inst%d,%s", sis.lastInstanceID, it.ProviderType)),
 		tags:         copyTags(tags),
 		providerType: it.ProviderType,
-		initCommand:  cmd,
-		running:      map[string]int64{},
+		running:      map[string]stubProcess{},
 		killing:      map[string]bool{},
 	}
 	svm.SSHService = SSHService{
@@ -170,6 +172,15 @@ func (sis *StubInstanceSet) Stop() {
 	sis.stopped = true
 }
 
+func (sis *StubInstanceSet) StubVMs() (svms []*StubVM) {
+	sis.mtx.Lock()
+	defer sis.mtx.Unlock()
+	for _, vm := range sis.servers {
+		svms = append(svms, vm)
+	}
+	return
+}
+
 type RateLimitError struct{ Retry time.Time }
 
 func (e RateLimitError) Error() string            { return fmt.Sprintf("rate limited until %s", e.Retry) }
@@ -189,21 +200,35 @@ type StubVM struct {
 	CrunchRunMissing      bool
 	CrunchRunCrashRate    float64
 	CrunchRunDetachDelay  time.Duration
+	ArvMountMaxExitLag    time.Duration
+	ArvMountDeadlockRate  float64
 	ExecuteContainer      func(arvados.Container) int
 	CrashRunningContainer func(arvados.Container)
+	ExtraCrunchRunArgs    string // extra args expected after "crunch-run --detach --stdin-config "
+
+	// Populated by (*StubInstanceSet)Create()
+	InitCommand cloud.InitCommand
 
 	sis          *StubInstanceSet
 	id           cloud.InstanceID
 	tags         cloud.InstanceTags
-	initCommand  cloud.InitCommand
 	providerType string
 	SSHService   SSHService
-	running      map[string]int64
+	running      map[string]stubProcess
 	killing      map[string]bool
 	lastPID      int64
+	deadlocked   string
 	sync.Mutex
 }
 
+type stubProcess struct {
+	pid int64
+
+	// crunch-run has exited, but arv-mount process (or something)
+	// still holds lock in /var/run/
+	exited bool
+}
+
 func (svm *StubVM) Instance() stubInstance {
 	svm.Lock()
 	defer svm.Unlock()
@@ -240,15 +265,15 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 		fmt.Fprint(stderr, "crunch-run: command not found\n")
 		return 1
 	}
-	if strings.HasPrefix(command, "crunch-run --detach --stdin-env ") {
-		var stdinKV map[string]string
-		err := json.Unmarshal(stdinData, &stdinKV)
+	if strings.HasPrefix(command, "crunch-run --detach --stdin-config "+svm.ExtraCrunchRunArgs) {
+		var configData crunchrun.ConfigData
+		err := json.Unmarshal(stdinData, &configData)
 		if err != nil {
 			fmt.Fprintf(stderr, "unmarshal stdin: %s (stdin was: %q)\n", err, stdinData)
 			return 1
 		}
 		for _, name := range []string{"ARVADOS_API_HOST", "ARVADOS_API_TOKEN"} {
-			if stdinKV[name] == "" {
+			if configData.Env[name] == "" {
 				fmt.Fprintf(stderr, "%s env var missing from stdin %q\n", name, stdinData)
 				return 1
 			}
@@ -256,7 +281,7 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 		svm.Lock()
 		svm.lastPID++
 		pid := svm.lastPID
-		svm.running[uuid] = pid
+		svm.running[uuid] = stubProcess{pid: pid}
 		svm.Unlock()
 		time.Sleep(svm.CrunchRunDetachDelay)
 		fmt.Fprintf(stderr, "starting %s\n", uuid)
@@ -273,16 +298,13 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 				logger.Print("[test] exiting crunch-run stub")
 				svm.Lock()
 				defer svm.Unlock()
-				if svm.running[uuid] != pid {
-					if !completed {
-						bugf := svm.sis.driver.Bugf
-						if bugf == nil {
-							bugf = logger.Warnf
-						}
-						bugf("[test] StubDriver bug or caller bug: pid %d exiting, running[%s]==%d", pid, uuid, svm.running[uuid])
+				if svm.running[uuid].pid != pid {
+					bugf := svm.sis.driver.Bugf
+					if bugf == nil {
+						bugf = logger.Warnf
 					}
-				} else {
-					delete(svm.running, uuid)
+					bugf("[test] StubDriver bug or caller bug: pid %d exiting, running[%s].pid==%d", pid, uuid, svm.running[uuid].pid)
+					return
 				}
 				if !completed {
 					logger.WithField("State", ctr.State).Print("[test] crashing crunch-run stub")
@@ -290,6 +312,15 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 						svm.CrashRunningContainer(ctr)
 					}
 				}
+				sproc := svm.running[uuid]
+				sproc.exited = true
+				svm.running[uuid] = sproc
+				svm.Unlock()
+				time.Sleep(svm.ArvMountMaxExitLag * time.Duration(math_rand.Float64()))
+				svm.Lock()
+				if math_rand.Float64() >= svm.ArvMountDeadlockRate {
+					delete(svm.running, uuid)
+				}
 			}()
 
 			crashluck := math_rand.Float64()
@@ -305,7 +336,7 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 			time.Sleep(time.Duration(math_rand.Float64()*20) * time.Millisecond)
 
 			svm.Lock()
-			killed := svm.running[uuid] != pid
+			killed := svm.killing[uuid]
 			svm.Unlock()
 			if killed || wantCrashEarly {
 				return
@@ -335,38 +366,31 @@ func (svm *StubVM) Exec(env map[string]string, command string, stdin io.Reader,
 	if command == "crunch-run --list" {
 		svm.Lock()
 		defer svm.Unlock()
-		for uuid := range svm.running {
-			fmt.Fprintf(stdout, "%s\n", uuid)
+		for uuid, sproc := range svm.running {
+			if sproc.exited {
+				fmt.Fprintf(stdout, "%s stale\n", uuid)
+			} else {
+				fmt.Fprintf(stdout, "%s\n", uuid)
+			}
 		}
 		if !svm.ReportBroken.IsZero() && svm.ReportBroken.Before(time.Now()) {
 			fmt.Fprintln(stdout, "broken")
 		}
+		fmt.Fprintln(stdout, svm.deadlocked)
 		return 0
 	}
 	if strings.HasPrefix(command, "crunch-run --kill ") {
 		svm.Lock()
-		pid, running := svm.running[uuid]
-		if running && !svm.killing[uuid] {
+		sproc, running := svm.running[uuid]
+		if running && !sproc.exited {
 			svm.killing[uuid] = true
-			go func() {
-				time.Sleep(time.Duration(math_rand.Float64()*30) * time.Millisecond)
-				svm.Lock()
-				defer svm.Unlock()
-				if svm.running[uuid] == pid {
-					// Kill only if the running entry
-					// hasn't since been killed and
-					// replaced with a different one.
-					delete(svm.running, uuid)
-				}
-				delete(svm.killing, uuid)
-			}()
 			svm.Unlock()
 			time.Sleep(time.Duration(math_rand.Float64()*2) * time.Millisecond)
 			svm.Lock()
-			_, running = svm.running[uuid]
+			sproc, running = svm.running[uuid]
 		}
 		svm.Unlock()
-		if running {
+		if running && !sproc.exited {
 			fmt.Fprintf(stderr, "%s: container is running\n", uuid)
 			return 1
 		}
@@ -458,3 +482,7 @@ func copyTags(src cloud.InstanceTags) cloud.InstanceTags {
 	}
 	return dst
 }
+
+func (si stubInstance) PriceHistory(arvados.InstanceType) []cloud.InstancePrice {
+	return nil
+}