X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/de9d45e3a238df8e9f0b2833b86c5e54fec37c7a..b6d7efab2c4bffa3fabd55b166e44cca8ac1391f:/lib/dispatchcloud/worker/pool_test.go diff --git a/lib/dispatchcloud/worker/pool_test.go b/lib/dispatchcloud/worker/pool_test.go index 26087193f3..1948c1e874 100644 --- a/lib/dispatchcloud/worker/pool_test.go +++ b/lib/dispatchcloud/worker/pool_test.go @@ -5,13 +5,15 @@ package worker import ( - "io" + "sort" + "strings" "time" - "git.curoverse.com/arvados.git/lib/cloud" - "git.curoverse.com/arvados.git/lib/dispatchcloud/test" - "git.curoverse.com/arvados.git/sdk/go/arvados" - "github.com/sirupsen/logrus" + "git.arvados.org/arvados.git/lib/cloud" + "git.arvados.org/arvados.git/lib/dispatchcloud/test" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/ctxlog" + "github.com/prometheus/client_golang/prometheus" check "gopkg.in/check.v1" ) @@ -31,27 +33,132 @@ var less = &lessChecker{&check.CheckerInfo{Name: "less", Params: []string{"obtai type PoolSuite struct{} -func (suite *PoolSuite) SetUpSuite(c *check.C) { - logrus.StandardLogger().SetLevel(logrus.DebugLevel) -} +func (suite *PoolSuite) TestResumeAfterRestart(c *check.C) { + type1 := test.InstanceType(1) + type2 := test.InstanceType(2) + type3 := test.InstanceType(3) + waitForIdle := func(pool *Pool, notify <-chan struct{}) { + timeout := time.NewTimer(time.Second) + for { + instances := pool.Instances() + sort.Slice(instances, func(i, j int) bool { + return strings.Compare(instances[i].ArvadosInstanceType, instances[j].ArvadosInstanceType) < 0 + }) + if len(instances) == 3 && + instances[0].ArvadosInstanceType == type1.Name && + instances[0].WorkerState == StateIdle.String() && + instances[1].ArvadosInstanceType == type1.Name && + instances[1].WorkerState == StateIdle.String() && + instances[2].ArvadosInstanceType == type2.Name && + instances[2].WorkerState == StateIdle.String() { + return + } + select { + case <-timeout.C: + c.Logf("pool.Instances() == %#v", instances) + c.Error("timed out") + return + case <-notify: + } + } + } -func (suite *PoolSuite) TestStartContainer(c *check.C) { - // TODO: use an instanceSet stub with an SSH server -} + logger := ctxlog.TestLogger(c) + driver := &test.StubDriver{} + instanceSetID := cloud.InstanceSetID("test-instance-set-id") + is, err := driver.InstanceSet(nil, instanceSetID, nil, logger) + c.Assert(err, check.IsNil) + + newExecutor := func(cloud.Instance) Executor { + return &stubExecutor{ + response: map[string]stubResp{ + "crunch-run --list": stubResp{}, + "true": stubResp{}, + }, + } + } + + cluster := &arvados.Cluster{ + Containers: arvados.ContainersConfig{ + CloudVMs: arvados.CloudVMsConfig{ + BootProbeCommand: "true", + MaxProbesPerSecond: 1000, + ProbeInterval: arvados.Duration(time.Millisecond * 10), + SyncInterval: arvados.Duration(time.Millisecond * 10), + TagKeyPrefix: "testprefix:", + }, + }, + InstanceTypes: arvados.InstanceTypeMap{ + type1.Name: type1, + type2.Name: type2, + type3.Name: type3, + }, + } + + pool := NewPool(logger, arvados.NewClientFromEnv(), prometheus.NewRegistry(), instanceSetID, is, newExecutor, nil, cluster) + notify := pool.Subscribe() + defer pool.Unsubscribe(notify) + pool.Create(type1) + pool.Create(type1) + pool.Create(type2) + waitForIdle(pool, notify) + var heldInstanceID cloud.InstanceID + for _, inst := range pool.Instances() { + if inst.ArvadosInstanceType == type2.Name { + heldInstanceID = cloud.InstanceID(inst.Instance) + pool.SetIdleBehavior(heldInstanceID, IdleBehaviorHold) + } + } + // Wait for the tags to save to the cloud provider + tagKey := cluster.Containers.CloudVMs.TagKeyPrefix + tagKeyIdleBehavior + deadline := time.Now().Add(time.Second) + for !func() bool { + pool.mtx.RLock() + defer pool.mtx.RUnlock() + for _, wkr := range pool.workers { + if wkr.instType == type2 { + return wkr.instance.Tags()[tagKey] == string(IdleBehaviorHold) + } + } + return false + }() { + if time.Now().After(deadline) { + c.Fatal("timeout") + } + time.Sleep(time.Millisecond * 10) + } + pool.Stop() -func (suite *PoolSuite) TestVerifyHostKey(c *check.C) { - // TODO: use an instanceSet stub with an SSH server + c.Log("------- starting new pool, waiting to recover state") + + pool2 := NewPool(logger, arvados.NewClientFromEnv(), prometheus.NewRegistry(), instanceSetID, is, newExecutor, nil, cluster) + notify2 := pool2.Subscribe() + defer pool2.Unsubscribe(notify2) + waitForIdle(pool2, notify2) + for _, inst := range pool2.Instances() { + if inst.ArvadosInstanceType == type2.Name { + c.Check(inst.Instance, check.Equals, heldInstanceID) + c.Check(inst.IdleBehavior, check.Equals, IdleBehaviorHold) + } else { + c.Check(inst.IdleBehavior, check.Equals, IdleBehaviorRun) + } + } + pool2.Stop() } func (suite *PoolSuite) TestCreateUnallocShutdown(c *check.C) { - lameInstanceSet := &test.LameInstanceSet{Hold: make(chan bool)} + logger := ctxlog.TestLogger(c) + driver := test.StubDriver{HoldCloudOps: true} + instanceSet, err := driver.InstanceSet(nil, "test-instance-set-id", nil, logger) + c.Assert(err, check.IsNil) + type1 := arvados.InstanceType{Name: "a1s", ProviderType: "a1.small", VCPUs: 1, RAM: 1 * GiB, Price: .01} type2 := arvados.InstanceType{Name: "a2m", ProviderType: "a2.medium", VCPUs: 2, RAM: 2 * GiB, Price: .02} type3 := arvados.InstanceType{Name: "a2l", ProviderType: "a2.large", VCPUs: 4, RAM: 4 * GiB, Price: .04} pool := &Pool{ - logger: logrus.StandardLogger(), + logger: logger, newExecutor: func(cloud.Instance) Executor { return &stubExecutor{} }, - instanceSet: &throttledInstanceSet{InstanceSet: lameInstanceSet}, + instanceSet: &throttledInstanceSet{InstanceSet: instanceSet}, instanceTypes: arvados.InstanceTypeMap{ type1.Name: type1, type2.Name: type2, @@ -75,7 +182,7 @@ func (suite *PoolSuite) TestCreateUnallocShutdown(c *check.C) { c.Check(pool.Unallocated()[type3], check.Equals, 1) // Unblock the pending Create calls. - go lameInstanceSet.Release(4) + go driver.ReleaseCloudOps(4) // Wait for each instance to either return from its Create // call, or show up in a poll. @@ -89,7 +196,7 @@ func (suite *PoolSuite) TestCreateUnallocShutdown(c *check.C) { ivs := suite.instancesByType(pool, type3) c.Assert(ivs, check.HasLen, 1) type3instanceID := ivs[0].Instance - err := pool.SetIdleBehavior(type3instanceID, IdleBehaviorHold) + err = pool.SetIdleBehavior(type3instanceID, IdleBehaviorHold) c.Check(err, check.IsNil) // Check admin-hold behavior: refuse to shutdown, and don't @@ -155,7 +262,7 @@ func (suite *PoolSuite) TestCreateUnallocShutdown(c *check.C) { // if a node still appears in the provider list after a // previous attempt, so there might be more than 4 Destroy // calls to unblock. - go lameInstanceSet.Release(4444) + go driver.ReleaseCloudOps(4444) // Sync until all instances disappear from the provider list. suite.wait(c, pool, notify, func() bool { @@ -186,13 +293,3 @@ func (suite *PoolSuite) wait(c *check.C, pool *Pool, notify <-chan struct{}, rea } c.Check(ready(), check.Equals, true) } - -type stubExecutor struct{} - -func (*stubExecutor) SetTarget(cloud.ExecutorTarget) {} - -func (*stubExecutor) Execute(env map[string]string, cmd string, stdin io.Reader) ([]byte, []byte, error) { - return nil, nil, nil -} - -func (*stubExecutor) Close() {}