open-nomad/nomad/state/state_store_test.go

4296 lines
93 KiB
Go
Raw Normal View History

package state
2015-07-04 01:19:43 +00:00
import (
"fmt"
2015-07-04 01:19:43 +00:00
"os"
"reflect"
"sort"
2015-07-04 01:19:43 +00:00
"testing"
2015-12-19 01:51:30 +00:00
"time"
2015-07-04 01:19:43 +00:00
2017-02-07 00:46:23 +00:00
memdb "github.com/hashicorp/go-memdb"
"github.com/hashicorp/nomad/nomad/mock"
2015-07-04 01:19:43 +00:00
"github.com/hashicorp/nomad/nomad/structs"
)
func testStateStore(t *testing.T) *StateStore {
state, err := NewStateStore(os.Stderr)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if state == nil {
t.Fatalf("missing state")
}
return state
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpsertNode_Node(t *testing.T) {
2015-07-04 01:19:43 +00:00
state := testStateStore(t)
node := mock.Node()
2015-07-04 01:19:43 +00:00
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
_, err := state.NodeByID(ws, node.ID)
if err != nil {
t.Fatalf("bad: %v", err)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
err = state.UpsertNode(1000, node)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
2015-07-04 01:19:43 +00:00
2017-02-07 00:46:23 +00:00
ws = memdb.NewWatchSet()
out, err := state.NodeByID(ws, node.ID)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(node, out) {
t.Fatalf("bad: %#v %#v", node, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("nodes")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-04 01:19:43 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_DeleteNode_Node(t *testing.T) {
2015-07-04 01:19:43 +00:00
state := testStateStore(t)
node := mock.Node()
2015-07-04 01:19:43 +00:00
err := state.UpsertNode(1000, node)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.NodeByID(ws, node.ID); err != nil {
t.Fatalf("bad: %v", err)
}
err = state.DeleteNode(1001, node.ID)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.NodeByID(ws, node.ID)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
t.Fatalf("bad: %#v %#v", node, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("nodes")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-04 01:19:43 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpdateNodeStatus_Node(t *testing.T) {
2015-07-04 01:19:43 +00:00
state := testStateStore(t)
node := mock.Node()
2015-07-04 01:19:43 +00:00
err := state.UpsertNode(800, node)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that update node status fires the watch
ws := memdb.NewWatchSet()
if _, err := state.NodeByID(ws, node.ID); err != nil {
t.Fatalf("bad: %v", err)
}
err = state.UpdateNodeStatus(801, node.ID, structs.NodeStatusReady)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.NodeByID(ws, node.ID)
2015-07-04 01:19:43 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if out.Status != structs.NodeStatusReady {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 801 {
2015-07-04 01:19:43 +00:00
t.Fatalf("bad: %#v", out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("nodes")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 801 {
2015-09-07 02:51:50 +00:00
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-09-07 02:51:50 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpdateNodeDrain_Node(t *testing.T) {
2015-09-07 02:51:50 +00:00
state := testStateStore(t)
node := mock.Node()
err := state.UpsertNode(1000, node)
2015-09-07 02:51:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that update node drain fires the watch
ws := memdb.NewWatchSet()
if _, err := state.NodeByID(ws, node.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-09-07 02:51:50 +00:00
err = state.UpdateNodeDrain(1001, node.ID, true)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.NodeByID(ws, node.ID)
2015-09-07 02:51:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !out.Drain {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1001 {
t.Fatalf("bad: %#v", out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("nodes")
2015-09-07 02:51:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-04 01:19:43 +00:00
}
func TestStateStore_Nodes(t *testing.T) {
state := testStateStore(t)
var nodes []*structs.Node
for i := 0; i < 10; i++ {
node := mock.Node()
nodes = append(nodes, node)
err := state.UpsertNode(1000+uint64(i), node)
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that getters don't cause it to fire
ws := memdb.NewWatchSet()
iter, err := state.Nodes(ws)
if err != nil {
2017-02-07 00:46:23 +00:00
t.Fatalf("bad: %v", err)
}
var out []*structs.Node
for {
raw := iter.Next()
if raw == nil {
break
}
out = append(out, raw.(*structs.Node))
}
sort.Sort(NodeIDSort(nodes))
sort.Sort(NodeIDSort(out))
if !reflect.DeepEqual(nodes, out) {
t.Fatalf("bad: %#v %#v", nodes, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_NodesByIDPrefix(t *testing.T) {
state := testStateStore(t)
node := mock.Node()
node.ID = "11111111-662e-d0ab-d1c9-3e434af7bdb4"
err := state.UpsertNode(1000, node)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that getters don't cause it to fire
ws := memdb.NewWatchSet()
iter, err := state.NodesByIDPrefix(ws, node.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
gatherNodes := func(iter memdb.ResultIterator) []*structs.Node {
var nodes []*structs.Node
for {
raw := iter.Next()
if raw == nil {
break
}
node := raw.(*structs.Node)
nodes = append(nodes, node)
}
return nodes
}
nodes := gatherNodes(iter)
if len(nodes) != 1 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
iter, err = state.NodesByIDPrefix(ws, "11")
if err != nil {
t.Fatalf("err: %v", err)
}
nodes = gatherNodes(iter)
if len(nodes) != 1 {
t.Fatalf("err: %v", err)
}
node = mock.Node()
node.ID = "11222222-662e-d0ab-d1c9-3e434af7bdb4"
err = state.UpsertNode(1001, node)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
iter, err = state.NodesByIDPrefix(ws, "11")
if err != nil {
t.Fatalf("err: %v", err)
}
nodes = gatherNodes(iter)
if len(nodes) != 2 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
iter, err = state.NodesByIDPrefix(ws, "1111")
if err != nil {
t.Fatalf("err: %v", err)
}
nodes = gatherNodes(iter)
if len(nodes) != 1 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-07-07 16:41:05 +00:00
func TestStateStore_RestoreNode(t *testing.T) {
state := testStateStore(t)
2015-10-30 04:42:41 +00:00
node := mock.Node()
2015-07-07 16:41:05 +00:00
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.NodeRestore(node)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.NodeByID(ws, node.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, node) {
t.Fatalf("Bad: %#v %#v", out, node)
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpsertJob_Job(t *testing.T) {
2015-07-07 16:41:05 +00:00
state := testStateStore(t)
job := mock.Job()
2015-07-07 16:41:05 +00:00
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
_, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
2017-02-07 00:46:23 +00:00
t.Fatalf("bad: %v", err)
}
if err := state.UpsertJob(1000, job); err != nil {
2015-07-07 16:41:05 +00:00
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
2015-07-07 16:41:05 +00:00
2017-02-07 00:46:23 +00:00
ws = memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(job, out) {
t.Fatalf("bad: %#v %#v", job, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("jobs")
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != job.ID {
t.Fatalf("bad summary id: %v", summary.JobID)
}
_, ok := summary.Summary["web"]
if !ok {
t.Fatalf("nil summary for task group")
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2017-04-13 20:54:57 +00:00
// Check the job history
allVersions, err := state.JobHistoryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if len(allVersions) != 1 {
t.Fatalf("got %d; want 1", len(allVersions))
}
if a := allVersions[0]; a.ID != job.ID || a.Version != 0 {
t.Fatalf("bad: %v", a)
}
2015-07-07 16:41:05 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpdateUpsertJob_Job(t *testing.T) {
2015-07-07 16:41:05 +00:00
state := testStateStore(t)
job := mock.Job()
2015-07-07 16:41:05 +00:00
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
_, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
2017-02-07 00:46:23 +00:00
t.Fatalf("bad: %v", err)
}
if err := state.UpsertJob(1000, job); err != nil {
2015-07-07 16:41:05 +00:00
t.Fatalf("err: %v", err)
}
job2 := mock.Job()
job2.ID = job.ID
2017-04-13 20:54:57 +00:00
job2.AllAtOnce = true
err = state.UpsertJob(1001, job2)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(job2, out) {
t.Fatalf("bad: %#v %#v", job2, out)
}
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1001 {
t.Fatalf("bad: %#v", out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("jobs")
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
// Test that the job summary remains the same if the job is updated but
// count remains same
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != job.ID {
t.Fatalf("bad summary id: %v", summary.JobID)
}
_, ok := summary.Summary["web"]
if !ok {
t.Fatalf("nil summary for task group")
}
2017-04-13 20:54:57 +00:00
// Check the job history
allVersions, err := state.JobHistoryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if len(allVersions) != 2 {
t.Fatalf("got %d; want 1", len(allVersions))
}
if a := allVersions[0]; a.ID != job.ID || a.Version != 1 || !a.AllAtOnce {
t.Fatalf("bad: %+v", a)
}
if a := allVersions[1]; a.ID != job.ID || a.Version != 0 || a.AllAtOnce {
t.Fatalf("bad: %+v", a)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-07 16:41:05 +00:00
}
func TestStateStore_UpdateUpsertJob_PeriodicJob(t *testing.T) {
state := testStateStore(t)
job := mock.PeriodicJob()
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
_, err := state.JobByID(ws, job.ID)
if err != nil {
t.Fatalf("bad: %v", err)
}
if err := state.UpsertJob(1000, job); err != nil {
t.Fatalf("err: %v", err)
}
// Create a child and an evaluation
job2 := job.Copy()
job2.Periodic = nil
job2.ID = fmt.Sprintf("%v/%s-1490635020", job.ID, structs.PeriodicLaunchSuffix)
err = state.UpsertJob(1001, job2)
if err != nil {
t.Fatalf("err: %v", err)
}
eval := mock.Eval()
eval.JobID = job2.ID
err = state.UpsertEvals(1002, []*structs.Evaluation{eval})
if err != nil {
t.Fatalf("err: %v", err)
}
job3 := job.Copy()
job3.TaskGroups[0].Tasks[0].Name = "new name"
err = state.UpsertJob(1003, job3)
if err != nil {
t.Fatalf("err: %v", err)
}
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if s, e := out.Status, structs.JobStatusRunning; s != e {
t.Fatalf("got status %v; want %v", s, e)
}
}
// This test ensures that UpsertJob creates the EphemeralDisk is a job doesn't have
// one and clear out the task's disk resource asks
// COMPAT 0.4.1 -> 0.5
func TestStateStore_UpsertJob_NoEphemeralDisk(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
// Set the EphemeralDisk to nil and set the tasks's DiskMB to 150
job.TaskGroups[0].EphemeralDisk = nil
job.TaskGroups[0].Tasks[0].Resources.DiskMB = 150
err := state.UpsertJob(1000, job)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
// Expect the state store to create the EphemeralDisk and clear out Tasks's
// DiskMB
expected := job.Copy()
expected.TaskGroups[0].EphemeralDisk = &structs.EphemeralDisk{
SizeMB: 150,
}
expected.TaskGroups[0].Tasks[0].Resources.DiskMB = 0
if !reflect.DeepEqual(expected, out) {
t.Fatalf("bad: %#v %#v", expected, out)
}
}
2016-12-12 21:32:30 +00:00
// Upsert a job that is the child of a parent job and ensures its summary gets
// updated.
func TestStateStore_UpsertJob_ChildJob(t *testing.T) {
state := testStateStore(t)
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
2016-12-12 21:32:30 +00:00
parent := mock.Job()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
_, err := state.JobByID(ws, parent.ID)
if err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
if err := state.UpsertJob(1000, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
2017-02-07 00:46:23 +00:00
if err := state.UpsertJob(1001, child); err != nil {
2016-12-12 21:32:30 +00:00
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 1 || summary.Children.Running != 0 || summary.Children.Dead != 0 {
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
2017-04-13 20:54:57 +00:00
func TestStateStore_UpdateUpsertJob_JobHistory(t *testing.T) {
state := testStateStore(t)
// Create a job and mark it as stable
job := mock.Job()
job.Stable = true
job.Priority = 0
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
_, err := state.JobHistoryByID(ws, job.ID)
if err != nil {
t.Fatalf("bad: %v", err)
}
if err := state.UpsertJob(1000, job); err != nil {
t.Fatalf("err: %v", err)
}
if !watchFired(ws) {
t.Fatalf("bad")
}
var finalJob *structs.Job
for i := 1; i < 20; i++ {
finalJob = mock.Job()
finalJob.ID = job.ID
finalJob.Priority = i
err = state.UpsertJob(uint64(1000+i), finalJob)
if err != nil {
t.Fatalf("err: %v", err)
}
}
ws = memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(finalJob, out) {
t.Fatalf("bad: %#v %#v", finalJob, out)
}
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1019 {
t.Fatalf("bad: %#v", out)
}
if out.Version != 19 {
t.Fatalf("bad: %#v", out)
}
index, err := state.Index("job_histories")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1019 {
t.Fatalf("bad: %d", index)
}
// Check the job history
allVersions, err := state.JobHistoryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if len(allVersions) != structs.JobDefaultHistoricCount {
t.Fatalf("got %d; want 1", len(allVersions))
}
if a := allVersions[0]; a.ID != job.ID || a.Version != 19 || a.Priority != 19 {
t.Fatalf("bad: %+v", a)
}
if a := allVersions[1]; a.ID != job.ID || a.Version != 18 || a.Priority != 18 {
t.Fatalf("bad: %+v", a)
}
// Ensure we didn't delete the stable job
if a := allVersions[structs.JobDefaultHistoricCount-1]; a.ID != job.ID ||
a.Version != 0 || a.Priority != 0 || !a.Stable {
t.Fatalf("bad: %+v", a)
}
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_DeleteJob_Job(t *testing.T) {
2015-07-07 16:41:05 +00:00
state := testStateStore(t)
job := mock.Job()
2015-07-07 16:41:05 +00:00
err := state.UpsertJob(1000, job)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.JobByID(ws, job.ID); err != nil {
t.Fatalf("bad: %v", err)
}
err = state.DeleteJob(1001, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
t.Fatalf("bad: %#v %#v", job, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("jobs")
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if summary != nil {
t.Fatalf("expected summary to be nil, but got: %v", summary)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-07 16:41:05 +00:00
}
2016-12-12 21:32:30 +00:00
func TestStateStore_DeleteJob_ChildJob(t *testing.T) {
state := testStateStore(t)
parent := mock.Job()
if err := state.UpsertJob(998, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
if err := state.UpsertJob(999, child); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws, parent.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
err := state.DeleteJob(1001, child.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
2017-02-07 00:46:23 +00:00
ws = memdb.NewWatchSet()
summary, err := state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 0 || summary.Children.Running != 0 || summary.Children.Dead != 1 {
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
2015-07-07 16:41:05 +00:00
func TestStateStore_Jobs(t *testing.T) {
state := testStateStore(t)
var jobs []*structs.Job
for i := 0; i < 10; i++ {
job := mock.Job()
2015-07-07 16:41:05 +00:00
jobs = append(jobs, job)
err := state.UpsertJob(1000+uint64(i), job)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.Jobs(ws)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
var out []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
out = append(out, raw.(*structs.Job))
}
sort.Sort(JobIDSort(jobs))
sort.Sort(JobIDSort(out))
2015-07-07 16:41:05 +00:00
if !reflect.DeepEqual(jobs, out) {
t.Fatalf("bad: %#v %#v", jobs, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-07 16:41:05 +00:00
}
func TestStateStore_JobsByIDPrefix(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
job.ID = "redis"
err := state.UpsertJob(1000, job)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.JobsByIDPrefix(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
gatherJobs := func(iter memdb.ResultIterator) []*structs.Job {
var jobs []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
jobs = append(jobs, raw.(*structs.Job))
}
return jobs
}
jobs := gatherJobs(iter)
if len(jobs) != 1 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
iter, err = state.JobsByIDPrefix(ws, "re")
if err != nil {
t.Fatalf("err: %v", err)
}
jobs = gatherJobs(iter)
if len(jobs) != 1 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
job = mock.Job()
job.ID = "riak"
err = state.UpsertJob(1001, job)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
iter, err = state.JobsByIDPrefix(ws, "r")
if err != nil {
t.Fatalf("err: %v", err)
}
jobs = gatherJobs(iter)
if len(jobs) != 2 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
iter, err = state.JobsByIDPrefix(ws, "ri")
if err != nil {
t.Fatalf("err: %v", err)
}
jobs = gatherJobs(iter)
if len(jobs) != 1 {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-12-04 17:49:42 +00:00
func TestStateStore_JobsByPeriodic(t *testing.T) {
state := testStateStore(t)
var periodic, nonPeriodic []*structs.Job
for i := 0; i < 10; i++ {
job := mock.Job()
nonPeriodic = append(nonPeriodic, job)
err := state.UpsertJob(1000+uint64(i), job)
if err != nil {
t.Fatalf("err: %v", err)
}
}
for i := 0; i < 10; i++ {
job := mock.PeriodicJob()
periodic = append(periodic, job)
err := state.UpsertJob(2000+uint64(i), job)
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.JobsByPeriodic(ws, true)
2015-12-24 20:23:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-12-04 17:49:42 +00:00
var outPeriodic []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
outPeriodic = append(outPeriodic, raw.(*structs.Job))
}
2017-02-07 00:46:23 +00:00
iter, err = state.JobsByPeriodic(ws, false)
2016-06-16 21:32:07 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-12-24 20:23:05 +00:00
2015-12-04 17:49:42 +00:00
var outNonPeriodic []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
outNonPeriodic = append(outNonPeriodic, raw.(*structs.Job))
}
sort.Sort(JobIDSort(periodic))
sort.Sort(JobIDSort(nonPeriodic))
sort.Sort(JobIDSort(outPeriodic))
sort.Sort(JobIDSort(outNonPeriodic))
if !reflect.DeepEqual(periodic, outPeriodic) {
t.Fatalf("bad: %#v %#v", periodic, outPeriodic)
}
if !reflect.DeepEqual(nonPeriodic, outNonPeriodic) {
t.Fatalf("bad: %#v %#v", nonPeriodic, outNonPeriodic)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-04 17:49:42 +00:00
}
func TestStateStore_JobsByScheduler(t *testing.T) {
state := testStateStore(t)
var serviceJobs []*structs.Job
var sysJobs []*structs.Job
for i := 0; i < 10; i++ {
job := mock.Job()
serviceJobs = append(serviceJobs, job)
err := state.UpsertJob(1000+uint64(i), job)
if err != nil {
t.Fatalf("err: %v", err)
}
}
for i := 0; i < 10; i++ {
job := mock.SystemJob()
sysJobs = append(sysJobs, job)
err := state.UpsertJob(2000+uint64(i), job)
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.JobsByScheduler(ws, "service")
if err != nil {
t.Fatalf("err: %v", err)
}
var outService []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
outService = append(outService, raw.(*structs.Job))
}
2017-02-07 00:46:23 +00:00
iter, err = state.JobsByScheduler(ws, "system")
if err != nil {
t.Fatalf("err: %v", err)
}
var outSystem []*structs.Job
for {
raw := iter.Next()
if raw == nil {
break
}
outSystem = append(outSystem, raw.(*structs.Job))
}
sort.Sort(JobIDSort(serviceJobs))
sort.Sort(JobIDSort(sysJobs))
sort.Sort(JobIDSort(outService))
sort.Sort(JobIDSort(outSystem))
if !reflect.DeepEqual(serviceJobs, outService) {
t.Fatalf("bad: %#v %#v", serviceJobs, outService)
}
if !reflect.DeepEqual(sysJobs, outSystem) {
t.Fatalf("bad: %#v %#v", sysJobs, outSystem)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-12-15 03:20:57 +00:00
func TestStateStore_JobsByGC(t *testing.T) {
state := testStateStore(t)
var gc, nonGc []*structs.Job
for i := 0; i < 20; i++ {
var job *structs.Job
if i%2 == 0 {
job = mock.Job()
} else {
job = mock.PeriodicJob()
}
2015-12-15 03:20:57 +00:00
nonGc = append(nonGc, job)
if err := state.UpsertJob(1000+uint64(i), job); err != nil {
t.Fatalf("err: %v", err)
}
}
for i := 0; i < 10; i++ {
job := mock.Job()
job.Type = structs.JobTypeBatch
2015-12-15 03:20:57 +00:00
gc = append(gc, job)
if err := state.UpsertJob(2000+uint64(i), job); err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.JobsByGC(ws, true)
2015-12-15 03:20:57 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
var outGc []*structs.Job
for i := iter.Next(); i != nil; i = iter.Next() {
outGc = append(outGc, i.(*structs.Job))
}
2017-02-07 00:46:23 +00:00
iter, err = state.JobsByGC(ws, false)
2015-12-15 03:20:57 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
var outNonGc []*structs.Job
for i := iter.Next(); i != nil; i = iter.Next() {
outNonGc = append(outNonGc, i.(*structs.Job))
}
sort.Sort(JobIDSort(gc))
sort.Sort(JobIDSort(nonGc))
sort.Sort(JobIDSort(outGc))
sort.Sort(JobIDSort(outNonGc))
if !reflect.DeepEqual(gc, outGc) {
t.Fatalf("bad: %#v %#v", gc, outGc)
}
if !reflect.DeepEqual(nonGc, outNonGc) {
t.Fatalf("bad: %#v %#v", nonGc, outNonGc)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-15 03:20:57 +00:00
}
2015-07-07 16:41:05 +00:00
func TestStateStore_RestoreJob(t *testing.T) {
state := testStateStore(t)
2015-10-30 04:42:41 +00:00
job := mock.Job()
2015-07-07 16:41:05 +00:00
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.JobRestore(job)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
2015-07-07 16:41:05 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, job) {
t.Fatalf("Bad: %#v %#v", out, job)
}
}
// This test ensures that the state restore creates the EphemeralDisk for a job if
// it doesn't have one
// COMPAT 0.4.1 -> 0.5
func TestStateStore_Jobs_NoEphemeralDisk(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
// Set EphemeralDisk to nil and set the DiskMB to 150
job.TaskGroups[0].EphemeralDisk = nil
job.TaskGroups[0].Tasks[0].Resources.DiskMB = 150
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.JobRestore(job)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.JobByID(ws, job.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
// Expect job to have local disk and clear out the task's disk resource ask
expected := job.Copy()
expected.TaskGroups[0].EphemeralDisk = &structs.EphemeralDisk{
SizeMB: 150,
}
expected.TaskGroups[0].Tasks[0].Resources.DiskMB = 0
if !reflect.DeepEqual(out, expected) {
t.Fatalf("Bad: %#v %#v", out, job)
}
}
2015-12-19 01:51:30 +00:00
func TestStateStore_UpsertPeriodicLaunch(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
2015-12-16 21:46:09 +00:00
launch := &structs.PeriodicLaunch{ID: job.ID, Launch: time.Now()}
2015-12-19 01:51:30 +00:00
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
if _, err := state.PeriodicLaunchByID(ws, launch.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-12-19 01:51:30 +00:00
err := state.UpsertPeriodicLaunch(1000, launch)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.PeriodicLaunchByID(ws, job.ID)
2015-12-19 01:51:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-12-16 21:46:09 +00:00
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
2015-12-19 01:51:30 +00:00
if !reflect.DeepEqual(launch, out) {
t.Fatalf("bad: %#v %#v", job, out)
}
index, err := state.Index("periodic_launch")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-19 01:51:30 +00:00
}
func TestStateStore_UpdateUpsertPeriodicLaunch(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
2015-12-16 21:46:09 +00:00
launch := &structs.PeriodicLaunch{ID: job.ID, Launch: time.Now()}
2015-12-19 01:51:30 +00:00
err := state.UpsertPeriodicLaunch(1000, launch)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
if _, err := state.PeriodicLaunchByID(ws, launch.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-12-16 21:46:09 +00:00
launch2 := &structs.PeriodicLaunch{
ID: job.ID,
Launch: launch.Launch.Add(1 * time.Second),
}
2015-12-19 01:51:30 +00:00
err = state.UpsertPeriodicLaunch(1001, launch2)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.PeriodicLaunchByID(ws, job.ID)
2015-12-19 01:51:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-12-16 21:46:09 +00:00
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1001 {
t.Fatalf("bad: %#v", out)
}
2015-12-19 01:51:30 +00:00
if !reflect.DeepEqual(launch2, out) {
t.Fatalf("bad: %#v %#v", launch2, out)
}
index, err := state.Index("periodic_launch")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-19 01:51:30 +00:00
}
func TestStateStore_DeletePeriodicLaunch(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
2015-12-16 21:46:09 +00:00
launch := &structs.PeriodicLaunch{ID: job.ID, Launch: time.Now()}
2015-12-19 01:51:30 +00:00
err := state.UpsertPeriodicLaunch(1000, launch)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.PeriodicLaunchByID(ws, launch.ID); err != nil {
t.Fatalf("bad: %v", err)
}
err = state.DeletePeriodicLaunch(1001, launch.ID)
2015-12-19 01:51:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.PeriodicLaunchByID(ws, job.ID)
2015-12-19 01:51:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
t.Fatalf("bad: %#v %#v", job, out)
}
index, err := state.Index("periodic_launch")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-19 01:51:30 +00:00
}
2015-12-07 23:58:17 +00:00
func TestStateStore_PeriodicLaunches(t *testing.T) {
state := testStateStore(t)
var launches []*structs.PeriodicLaunch
for i := 0; i < 10; i++ {
job := mock.Job()
2015-12-16 21:46:09 +00:00
launch := &structs.PeriodicLaunch{ID: job.ID, Launch: time.Now()}
2015-12-07 23:58:17 +00:00
launches = append(launches, launch)
err := state.UpsertPeriodicLaunch(1000+uint64(i), launch)
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.PeriodicLaunches(ws)
2015-12-07 23:58:17 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
out := make(map[string]*structs.PeriodicLaunch, 10)
for {
raw := iter.Next()
if raw == nil {
break
}
launch := raw.(*structs.PeriodicLaunch)
if _, ok := out[launch.ID]; ok {
t.Fatalf("duplicate: %v", launch.ID)
}
out[launch.ID] = launch
}
for _, launch := range launches {
l, ok := out[launch.ID]
if !ok {
t.Fatalf("bad %v", launch.ID)
}
if !reflect.DeepEqual(launch, l) {
t.Fatalf("bad: %#v %#v", launch, l)
}
delete(out, launch.ID)
}
if len(out) != 0 {
t.Fatalf("leftover: %#v", out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-07 23:58:17 +00:00
}
func TestStateStore_RestorePeriodicLaunch(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
2015-12-16 21:46:09 +00:00
launch := &structs.PeriodicLaunch{ID: job.ID, Launch: time.Now()}
2015-12-07 23:58:17 +00:00
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.PeriodicLaunchRestore(launch)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.PeriodicLaunchByID(ws, job.ID)
2015-12-07 23:58:17 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, launch) {
t.Fatalf("Bad: %#v %#v", out, job)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-12-07 23:58:17 +00:00
}
2016-07-05 20:55:11 +00:00
func TestStateStore_RestoreJobSummary(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
jobSummary := &structs.JobSummary{
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{
Starting: 10,
2016-07-05 20:55:11 +00:00
},
},
}
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.JobSummaryRestore(jobSummary)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.JobSummaryByID(ws, job.ID)
2016-07-05 20:55:11 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, jobSummary) {
t.Fatalf("Bad: %#v %#v", out, jobSummary)
}
}
func TestStateStore_Indexes(t *testing.T) {
state := testStateStore(t)
node := mock.Node()
err := state.UpsertNode(1000, node)
if err != nil {
t.Fatalf("err: %v", err)
}
iter, err := state.Indexes()
if err != nil {
t.Fatalf("err: %v", err)
}
var out []*IndexEntry
for {
raw := iter.Next()
if raw == nil {
break
}
out = append(out, raw.(*IndexEntry))
}
expect := []*IndexEntry{
&IndexEntry{"nodes", 1000},
}
if !reflect.DeepEqual(expect, out) {
t.Fatalf("bad: %#v %#v", expect, out)
}
}
func TestStateStore_LatestIndex(t *testing.T) {
state := testStateStore(t)
if err := state.UpsertNode(1000, mock.Node()); err != nil {
t.Fatalf("err: %v", err)
}
exp := uint64(2000)
if err := state.UpsertJob(exp, mock.Job()); err != nil {
t.Fatalf("err: %v", err)
}
latest, err := state.LatestIndex()
if err != nil {
t.Fatalf("err: %v", err)
}
if latest != exp {
t.Fatalf("LatestIndex() returned %d; want %d", latest, exp)
}
}
2015-07-07 16:41:05 +00:00
func TestStateStore_RestoreIndex(t *testing.T) {
state := testStateStore(t)
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
2015-07-07 16:41:05 +00:00
index := &IndexEntry{"jobs", 1000}
err = restore.IndexRestore(index)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2015-09-07 03:51:01 +00:00
out, err := state.Index("jobs")
if err != nil {
t.Fatalf("err: %v", err)
}
2015-07-07 16:41:05 +00:00
if out != 1000 {
t.Fatalf("Bad: %#v %#v", out, 1000)
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpsertEvals_Eval(t *testing.T) {
2015-07-23 22:43:06 +00:00
state := testStateStore(t)
eval := mock.Eval()
2015-07-23 22:43:06 +00:00
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
if _, err := state.EvalByID(ws, eval.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-10-30 04:42:41 +00:00
err := state.UpsertEvals(1000, []*structs.Evaluation{eval})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval.ID)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(eval, out) {
t.Fatalf("bad: %#v %#v", eval, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("evals")
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-23 22:43:06 +00:00
}
func TestStateStore_UpsertEvals_CancelBlocked(t *testing.T) {
state := testStateStore(t)
// Create two blocked evals for the same job
j := "test-job"
b1, b2 := mock.Eval(), mock.Eval()
b1.JobID = j
b1.Status = structs.EvalStatusBlocked
b2.JobID = j
b2.Status = structs.EvalStatusBlocked
err := state.UpsertEvals(999, []*structs.Evaluation{b1, b2})
if err != nil {
t.Fatalf("err: %v", err)
}
// Create one complete and successful eval for the job
eval := mock.Eval()
eval.JobID = j
eval.Status = structs.EvalStatusComplete
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that the upsert of the complete eval
// fires the watch
ws := memdb.NewWatchSet()
if _, err := state.EvalByID(ws, b1.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalByID(ws, b2.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if err := state.UpsertEvals(1000, []*structs.Evaluation{eval}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(eval, out) {
t.Fatalf("bad: %#v %#v", eval, out)
}
index, err := state.Index("evals")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
// Get b1/b2 and check they are cancelled
2017-02-07 00:46:23 +00:00
out1, err := state.EvalByID(ws, b1.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
out2, err := state.EvalByID(ws, b2.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if out1.Status != structs.EvalStatusCancelled || out2.Status != structs.EvalStatusCancelled {
t.Fatalf("bad: %#v %#v", out1, out2)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_Update_UpsertEvals_Eval(t *testing.T) {
2015-07-23 22:43:06 +00:00
state := testStateStore(t)
eval := mock.Eval()
2015-07-23 22:43:06 +00:00
err := state.UpsertEvals(1000, []*structs.Evaluation{eval})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create a watchset so we can test that delete fires the watch
ws := memdb.NewWatchSet()
ws2 := memdb.NewWatchSet()
if _, err := state.EvalByID(ws, eval.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalsByJob(ws2, eval.JobID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-10-30 04:42:41 +00:00
eval2 := mock.Eval()
2015-07-23 22:43:06 +00:00
eval2.ID = eval.ID
eval2.JobID = eval.JobID
err = state.UpsertEvals(1001, []*structs.Evaluation{eval2})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
if !watchFired(ws2) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval.ID)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(eval2, out) {
t.Fatalf("bad: %#v %#v", eval2, out)
}
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
if out.ModifyIndex != 1001 {
t.Fatalf("bad: %#v", out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("evals")
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-23 22:43:06 +00:00
}
2016-12-12 21:32:30 +00:00
func TestStateStore_UpsertEvals_Eval_ChildJob(t *testing.T) {
state := testStateStore(t)
parent := mock.Job()
if err := state.UpsertJob(998, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
if err := state.UpsertJob(999, child); err != nil {
t.Fatalf("err: %v", err)
}
eval := mock.Eval()
eval.Status = structs.EvalStatusComplete
eval.JobID = child.ID
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
ws2 := memdb.NewWatchSet()
ws3 := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws, parent.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalByID(ws2, eval.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalsByJob(ws3, eval.JobID); err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
err := state.UpsertEvals(1000, []*structs.Evaluation{eval})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
if !watchFired(ws2) {
t.Fatalf("bad")
}
if !watchFired(ws3) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(eval, out) {
t.Fatalf("bad: %#v %#v", eval, out)
}
index, err := state.Index("evals")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 0 || summary.Children.Running != 0 || summary.Children.Dead != 1 {
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_DeleteEval_Eval(t *testing.T) {
2015-07-23 22:43:06 +00:00
state := testStateStore(t)
2015-10-30 04:42:41 +00:00
eval1 := mock.Eval()
eval2 := mock.Eval()
2015-10-30 04:42:41 +00:00
alloc1 := mock.Alloc()
alloc2 := mock.Alloc()
2015-07-23 22:43:06 +00:00
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that upsert fires the watch
watches := make([]memdb.WatchSet, 12)
for i := 0; i < 12; i++ {
watches[i] = memdb.NewWatchSet()
}
if _, err := state.EvalByID(watches[0], eval1.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalByID(watches[1], eval2.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalsByJob(watches[2], eval1.JobID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.EvalsByJob(watches[3], eval2.JobID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocByID(watches[4], alloc1.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocByID(watches[5], alloc2.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[6], alloc1.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[7], alloc2.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[8], alloc1.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[9], alloc2.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[10], alloc1.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[11], alloc2.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-10-30 04:42:41 +00:00
2016-07-21 21:43:21 +00:00
state.UpsertJobSummary(900, mock.JobSummary(eval1.JobID))
state.UpsertJobSummary(901, mock.JobSummary(eval2.JobID))
state.UpsertJobSummary(902, mock.JobSummary(alloc1.JobID))
state.UpsertJobSummary(903, mock.JobSummary(alloc2.JobID))
2015-10-30 04:42:41 +00:00
err := state.UpsertEvals(1000, []*structs.Evaluation{eval1, eval2})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-10-30 04:42:41 +00:00
err = state.UpsertAllocs(1001, []*structs.Allocation{alloc1, alloc2})
if err != nil {
t.Fatalf("err: %v", err)
}
2015-10-30 04:42:41 +00:00
err = state.DeleteEval(1002, []string{eval1.ID, eval2.ID}, []string{alloc1.ID, alloc2.ID})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
for i, ws := range watches {
if !watchFired(ws) {
t.Fatalf("bad %d", i)
}
}
ws := memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval1.ID)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
2015-10-30 04:42:41 +00:00
t.Fatalf("bad: %#v %#v", eval1, out)
2015-07-23 22:43:06 +00:00
}
2017-02-07 00:46:23 +00:00
out, err = state.EvalByID(ws, eval2.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
2015-10-30 04:42:41 +00:00
t.Fatalf("bad: %#v %#v", eval1, out)
}
2017-02-07 00:46:23 +00:00
outA, err := state.AllocByID(ws, alloc1.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
2015-10-30 04:42:41 +00:00
t.Fatalf("bad: %#v %#v", alloc1, outA)
}
2017-02-07 00:46:23 +00:00
outA, err = state.AllocByID(ws, alloc2.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
2015-10-30 04:42:41 +00:00
t.Fatalf("bad: %#v %#v", alloc1, outA)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("evals")
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1002 {
t.Fatalf("bad: %d", index)
}
2015-09-07 03:51:01 +00:00
index, err = state.Index("allocs")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1002 {
2015-07-23 22:43:06 +00:00
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-23 22:43:06 +00:00
}
2016-12-12 21:32:30 +00:00
func TestStateStore_DeleteEval_ChildJob(t *testing.T) {
state := testStateStore(t)
parent := mock.Job()
if err := state.UpsertJob(998, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
if err := state.UpsertJob(999, child); err != nil {
t.Fatalf("err: %v", err)
}
eval1 := mock.Eval()
eval1.JobID = child.ID
alloc1 := mock.Alloc()
alloc1.JobID = child.ID
err := state.UpsertEvals(1000, []*structs.Evaluation{eval1})
if err != nil {
t.Fatalf("err: %v", err)
}
err = state.UpsertAllocs(1001, []*structs.Allocation{alloc1})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws, parent.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
err = state.DeleteEval(1002, []string{eval1.ID}, []string{alloc1.ID})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
summary, err := state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 0 || summary.Children.Running != 0 || summary.Children.Dead != 1 {
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
2015-09-06 19:10:24 +00:00
func TestStateStore_EvalsByJob(t *testing.T) {
state := testStateStore(t)
eval1 := mock.Eval()
eval2 := mock.Eval()
eval2.JobID = eval1.JobID
eval3 := mock.Eval()
evals := []*structs.Evaluation{eval1, eval2}
err := state.UpsertEvals(1000, evals)
if err != nil {
t.Fatalf("err: %v", err)
}
err = state.UpsertEvals(1001, []*structs.Evaluation{eval3})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.EvalsByJob(ws, eval1.JobID)
2015-09-06 19:10:24 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
sort.Sort(EvalIDSort(evals))
sort.Sort(EvalIDSort(out))
if !reflect.DeepEqual(evals, out) {
t.Fatalf("bad: %#v %#v", evals, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-09-06 19:10:24 +00:00
}
2015-07-23 22:43:06 +00:00
func TestStateStore_Evals(t *testing.T) {
state := testStateStore(t)
var evals []*structs.Evaluation
for i := 0; i < 10; i++ {
eval := mock.Eval()
2015-07-23 22:43:06 +00:00
evals = append(evals, eval)
err := state.UpsertEvals(1000+uint64(i), []*structs.Evaluation{eval})
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.Evals(ws)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
var out []*structs.Evaluation
for {
raw := iter.Next()
if raw == nil {
break
}
out = append(out, raw.(*structs.Evaluation))
}
sort.Sort(EvalIDSort(evals))
sort.Sort(EvalIDSort(out))
if !reflect.DeepEqual(evals, out) {
t.Fatalf("bad: %#v %#v", evals, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-07-23 22:43:06 +00:00
}
func TestStateStore_EvalsByIDPrefix(t *testing.T) {
state := testStateStore(t)
var evals []*structs.Evaluation
ids := []string{
"aaaaaaaa-7bfb-395d-eb95-0685af2176b2",
"aaaaaaab-7bfb-395d-eb95-0685af2176b2",
"aaaaaabb-7bfb-395d-eb95-0685af2176b2",
"aaaaabbb-7bfb-395d-eb95-0685af2176b2",
"aaaabbbb-7bfb-395d-eb95-0685af2176b2",
"aaabbbbb-7bfb-395d-eb95-0685af2176b2",
"aabbbbbb-7bfb-395d-eb95-0685af2176b2",
"abbbbbbb-7bfb-395d-eb95-0685af2176b2",
"bbbbbbbb-7bfb-395d-eb95-0685af2176b2",
}
for i := 0; i < 9; i++ {
eval := mock.Eval()
eval.ID = ids[i]
evals = append(evals, eval)
}
err := state.UpsertEvals(1000, evals)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.EvalsByIDPrefix(ws, "aaaa")
if err != nil {
t.Fatalf("err: %v", err)
}
gatherEvals := func(iter memdb.ResultIterator) []*structs.Evaluation {
var evals []*structs.Evaluation
for {
raw := iter.Next()
if raw == nil {
break
}
evals = append(evals, raw.(*structs.Evaluation))
}
return evals
}
out := gatherEvals(iter)
if len(out) != 5 {
t.Fatalf("bad: expected five evaluations, got: %#v", out)
}
sort.Sort(EvalIDSort(evals))
for index, eval := range out {
if ids[index] != eval.ID {
t.Fatalf("bad: got unexpected id: %s", eval.ID)
}
}
2017-02-07 00:46:23 +00:00
iter, err = state.EvalsByIDPrefix(ws, "b-a7bfb")
if err != nil {
t.Fatalf("err: %v", err)
}
out = gatherEvals(iter)
if len(out) != 0 {
t.Fatalf("bad: unexpected zero evaluations, got: %#v", out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-07-23 22:43:06 +00:00
func TestStateStore_RestoreEval(t *testing.T) {
state := testStateStore(t)
2015-10-30 04:42:41 +00:00
eval := mock.Eval()
2015-07-23 22:43:06 +00:00
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
2015-10-30 04:42:41 +00:00
err = restore.EvalRestore(eval)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.EvalByID(ws, eval.ID)
2015-07-23 22:43:06 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2015-10-30 04:42:41 +00:00
if !reflect.DeepEqual(out, eval) {
t.Fatalf("Bad: %#v %#v", out, eval)
2015-07-23 22:43:06 +00:00
}
}
func TestStateStore_UpdateAllocsFromClient(t *testing.T) {
state := testStateStore(t)
2016-12-12 21:32:30 +00:00
parent := mock.Job()
if err := state.UpsertJob(998, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
if err := state.UpsertJob(999, child); err != nil {
t.Fatalf("err: %v", err)
}
alloc := mock.Alloc()
alloc.JobID = child.ID
alloc.Job = child
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
2017-02-07 19:35:38 +00:00
summary, err := state.JobSummaryByID(ws, parent.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 0 || summary.Children.Running != 1 || summary.Children.Dead != 0 {
t.Fatalf("bad children summary: %v", summary.Children)
}
// Create watchsets so we can test that update fires the watch
ws = memdb.NewWatchSet()
2017-02-07 00:46:23 +00:00
if _, err := state.JobSummaryByID(ws, parent.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
// Create the delta updates
2017-02-07 00:46:23 +00:00
ts := map[string]*structs.TaskState{"web": &structs.TaskState{State: structs.TaskStateRunning}}
2016-12-12 21:32:30 +00:00
update := &structs.Allocation{
ID: alloc.ID,
2017-02-07 19:35:38 +00:00
ClientStatus: structs.AllocClientStatusComplete,
2016-12-12 21:32:30 +00:00
TaskStates: ts,
JobID: alloc.JobID,
TaskGroup: alloc.TaskGroup,
}
err = state.UpdateAllocsFromClient(1001, []*structs.Allocation{update})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 19:35:38 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
2017-02-07 00:46:23 +00:00
ws = memdb.NewWatchSet()
2017-02-07 19:35:38 +00:00
summary, err = state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
2017-02-07 19:35:38 +00:00
if summary.Children.Pending != 0 || summary.Children.Running != 0 || summary.Children.Dead != 1 {
2016-12-12 21:32:30 +00:00
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
func TestStateStore_UpdateAllocsFromClient_ChildJob(t *testing.T) {
state := testStateStore(t)
2017-02-07 00:46:23 +00:00
alloc1 := mock.Alloc()
alloc2 := mock.Alloc()
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
if err := state.UpsertJob(999, alloc1.Job); err != nil {
t.Fatalf("err: %v", err)
}
if err := state.UpsertJob(999, alloc2.Job); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc1, alloc2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that update fires the watch
watches := make([]memdb.WatchSet, 8)
for i := 0; i < 8; i++ {
watches[i] = memdb.NewWatchSet()
}
if _, err := state.AllocByID(watches[0], alloc1.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocByID(watches[1], alloc2.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[2], alloc1.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[3], alloc2.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[4], alloc1.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[5], alloc2.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[6], alloc1.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[7], alloc2.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
// Create the delta updates
2016-03-01 22:09:25 +00:00
ts := map[string]*structs.TaskState{"web": &structs.TaskState{State: structs.TaskStatePending}}
update := &structs.Allocation{
2017-02-07 00:46:23 +00:00
ID: alloc1.ID,
ClientStatus: structs.AllocClientStatusFailed,
2016-03-01 22:09:25 +00:00
TaskStates: ts,
2017-02-07 00:46:23 +00:00
JobID: alloc1.JobID,
TaskGroup: alloc1.TaskGroup,
}
update2 := &structs.Allocation{
ID: alloc2.ID,
ClientStatus: structs.AllocClientStatusRunning,
2016-03-01 22:09:25 +00:00
TaskStates: ts,
JobID: alloc2.JobID,
TaskGroup: alloc2.TaskGroup,
}
err = state.UpdateAllocsFromClient(1001, []*structs.Allocation{update, update2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
for i, ws := range watches {
if !watchFired(ws) {
t.Fatalf("bad %d", i)
}
}
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc1.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
alloc1.CreateIndex = 1000
alloc1.ModifyIndex = 1001
alloc1.TaskStates = ts
alloc1.ClientStatus = structs.AllocClientStatusFailed
if !reflect.DeepEqual(alloc1, out) {
t.Fatalf("bad: %#v %#v", alloc1, out)
}
2017-02-07 00:46:23 +00:00
out, err = state.AllocByID(ws, alloc2.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
alloc2.ModifyIndex = 1000
alloc2.ModifyIndex = 1001
alloc2.ClientStatus = structs.AllocClientStatusRunning
2016-03-01 22:09:25 +00:00
alloc2.TaskStates = ts
if !reflect.DeepEqual(alloc2, out) {
t.Fatalf("bad: %#v %#v", alloc2, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("allocs")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
// Ensure summaries have been updated
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, alloc1.JobID)
if err != nil {
t.Fatalf("err: %v", err)
}
tgSummary := summary.Summary["web"]
if tgSummary.Failed != 1 {
t.Fatalf("expected failed: %v, actual: %v, summary: %#v", 1, tgSummary.Failed, tgSummary)
}
2017-02-07 00:46:23 +00:00
summary2, err := state.JobSummaryByID(ws, alloc2.JobID)
if err != nil {
t.Fatalf("err: %v", err)
}
tgSummary2 := summary2.Summary["web"]
2016-07-26 22:11:48 +00:00
if tgSummary2.Running != 1 {
t.Fatalf("expected running: %v, actual: %v", 1, tgSummary2.Running)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_UpdateMultipleAllocsFromClient(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
// Create the delta updates
ts := map[string]*structs.TaskState{"web": &structs.TaskState{State: structs.TaskStatePending}}
update := &structs.Allocation{
ID: alloc.ID,
ClientStatus: structs.AllocClientStatusRunning,
TaskStates: ts,
JobID: alloc.JobID,
TaskGroup: alloc.TaskGroup,
}
update2 := &structs.Allocation{
ID: alloc.ID,
ClientStatus: structs.AllocClientStatusPending,
TaskStates: ts,
JobID: alloc.JobID,
TaskGroup: alloc.TaskGroup,
}
err = state.UpdateAllocsFromClient(1001, []*structs.Allocation{update, update2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
alloc.CreateIndex = 1000
alloc.ModifyIndex = 1001
alloc.TaskStates = ts
alloc.ClientStatus = structs.AllocClientStatusPending
if !reflect.DeepEqual(alloc, out) {
t.Fatalf("bad: %#v , actual:%#v", alloc, out)
}
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, alloc.JobID)
expectedSummary := &structs.JobSummary{
JobID: alloc.JobID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{
Starting: 1,
},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
CreateIndex: 999,
ModifyIndex: 1001,
}
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(summary, expectedSummary) {
t.Fatalf("expected: %#v, actual: %#v", expectedSummary, summary)
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpsertAlloc_Alloc(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
2015-10-30 04:42:41 +00:00
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that update fires the watch
watches := make([]memdb.WatchSet, 4)
for i := 0; i < 4; i++ {
watches[i] = memdb.NewWatchSet()
}
if _, err := state.AllocByID(watches[0], alloc.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[1], alloc.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[2], alloc.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[3], alloc.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
for i, ws := range watches {
if !watchFired(ws) {
t.Fatalf("bad %d", i)
}
}
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(alloc, out) {
t.Fatalf("bad: %#v %#v", alloc, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("allocs")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
summary, err := state.JobSummaryByID(ws, alloc.JobID)
if err != nil {
t.Fatalf("err: %v", err)
}
tgSummary, ok := summary.Summary["web"]
if !ok {
t.Fatalf("no summary for task group web")
}
if tgSummary.Starting != 1 {
t.Fatalf("expected queued: %v, actual: %v", 1, tgSummary.Starting)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_UpsertAlloc_NoEphemeralDisk(t *testing.T) {
2016-09-02 00:41:50 +00:00
state := testStateStore(t)
alloc := mock.Alloc()
alloc.Job.TaskGroups[0].EphemeralDisk = nil
2016-09-02 00:41:50 +00:00
alloc.Job.TaskGroups[0].Tasks[0].Resources.DiskMB = 120
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
2016-09-02 00:41:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
expected := alloc.Copy()
expected.Job.TaskGroups[0].EphemeralDisk = &structs.EphemeralDisk{SizeMB: 120}
2016-09-02 00:41:50 +00:00
if !reflect.DeepEqual(expected, out) {
t.Fatalf("bad: %#v %#v", expected, out)
}
}
2016-12-12 21:32:30 +00:00
func TestStateStore_UpsertAlloc_ChildJob(t *testing.T) {
state := testStateStore(t)
parent := mock.Job()
if err := state.UpsertJob(998, parent); err != nil {
t.Fatalf("err: %v", err)
}
child := mock.Job()
child.ParentID = parent.ID
if err := state.UpsertJob(999, child); err != nil {
t.Fatalf("err: %v", err)
}
alloc := mock.Alloc()
alloc.JobID = child.ID
alloc.Job = child
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that delete fires the watch
ws := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws, parent.ID); err != nil {
t.Fatalf("bad: %v", err)
}
2016-12-12 21:32:30 +00:00
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
summary, err := state.JobSummaryByID(ws, parent.ID)
2016-12-12 21:32:30 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
if summary == nil {
t.Fatalf("nil summary")
}
if summary.JobID != parent.ID {
t.Fatalf("bad summary id: %v", parent.ID)
}
if summary.Children == nil {
t.Fatalf("nil children summary")
}
if summary.Children.Pending != 0 || summary.Children.Running != 1 || summary.Children.Dead != 0 {
t.Fatalf("bad children summary: %v", summary.Children)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-12-12 21:32:30 +00:00
}
2015-09-07 03:51:01 +00:00
func TestStateStore_UpdateAlloc_Alloc(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
summary, err := state.JobSummaryByID(ws, alloc.JobID)
if err != nil {
t.Fatalf("err: %v", err)
}
tgSummary := summary.Summary["web"]
if tgSummary.Starting != 1 {
t.Fatalf("expected starting: %v, actual: %v", 1, tgSummary.Starting)
}
alloc2 := mock.Alloc()
alloc2.ID = alloc.ID
alloc2.NodeID = alloc.NodeID + ".new"
2016-07-21 21:43:21 +00:00
state.UpsertJobSummary(1001, mock.JobSummary(alloc2.JobID))
2015-10-30 04:42:41 +00:00
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that update fires the watch
watches := make([]memdb.WatchSet, 4)
for i := 0; i < 4; i++ {
watches[i] = memdb.NewWatchSet()
}
if _, err := state.AllocByID(watches[0], alloc2.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByEval(watches[1], alloc2.EvalID); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByJob(watches[2], alloc2.JobID, false); err != nil {
t.Fatalf("bad: %v", err)
}
if _, err := state.AllocsByNode(watches[3], alloc2.NodeID); err != nil {
t.Fatalf("bad: %v", err)
}
2015-10-30 04:42:41 +00:00
2016-07-21 21:43:21 +00:00
err = state.UpsertAllocs(1002, []*structs.Allocation{alloc2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
for i, ws := range watches {
if !watchFired(ws) {
t.Fatalf("bad %d", i)
}
}
ws = memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(alloc2, out) {
t.Fatalf("bad: %#v %#v", alloc2, out)
}
if out.CreateIndex != 1000 {
t.Fatalf("bad: %#v", out)
}
2016-07-21 21:43:21 +00:00
if out.ModifyIndex != 1002 {
t.Fatalf("bad: %#v", out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("allocs")
if err != nil {
t.Fatalf("err: %v", err)
}
2016-07-21 21:43:21 +00:00
if index != 1002 {
t.Fatalf("bad: %d", index)
}
2015-10-30 04:42:41 +00:00
// Ensure that summary hasb't changed
2017-02-07 00:46:23 +00:00
summary, err = state.JobSummaryByID(ws, alloc.JobID)
if err != nil {
t.Fatalf("err: %v", err)
}
tgSummary = summary.Summary["web"]
if tgSummary.Starting != 1 {
t.Fatalf("expected starting: %v, actual: %v", 1, tgSummary.Starting)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
// This test ensures that the state store will mark the clients status as lost
// when set rather than preferring the existing status.
func TestStateStore_UpdateAlloc_Lost(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
alloc.ClientStatus = "foo"
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
alloc2 := new(structs.Allocation)
*alloc2 = *alloc
alloc2.ClientStatus = structs.AllocClientStatusLost
if err := state.UpsertAllocs(1001, []*structs.Allocation{alloc2}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc2.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if out.ClientStatus != structs.AllocClientStatusLost {
t.Fatalf("bad: %#v", out)
}
}
// This test ensures an allocation can be updated when there is no job
// associated with it. This will happen when a job is stopped by an user which
// has non-terminal allocations on clients
func TestStateStore_UpdateAlloc_NoJob(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
// Upsert a job
state.UpsertJobSummary(998, mock.JobSummary(alloc.JobID))
if err := state.UpsertJob(999, alloc.Job); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
if err := state.DeleteJob(1001, alloc.JobID); err != nil {
t.Fatalf("err: %v", err)
}
// Update the desired state of the allocation to stop
allocCopy := alloc.Copy()
allocCopy.DesiredStatus = structs.AllocDesiredStatusStop
if err := state.UpsertAllocs(1002, []*structs.Allocation{allocCopy}); err != nil {
t.Fatalf("err: %v", err)
}
// Update the client state of the allocation to complete
allocCopy1 := allocCopy.Copy()
allocCopy1.ClientStatus = structs.AllocClientStatusComplete
if err := state.UpdateAllocsFromClient(1003, []*structs.Allocation{allocCopy1}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, _ := state.AllocByID(ws, alloc.ID)
// Update the modify index of the alloc before comparing
allocCopy1.ModifyIndex = 1003
if !reflect.DeepEqual(out, allocCopy1) {
t.Fatalf("expected: %#v \n actual: %#v", allocCopy1, out)
}
}
func TestStateStore_JobSummary(t *testing.T) {
state := testStateStore(t)
// Add a job
job := mock.Job()
state.UpsertJob(900, job)
// Get the job back
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
outJob, _ := state.JobByID(ws, job.ID)
if outJob.CreateIndex != 900 {
t.Fatalf("bad create index: %v", outJob.CreateIndex)
}
2017-02-07 00:46:23 +00:00
summary, _ := state.JobSummaryByID(ws, job.ID)
if summary.CreateIndex != 900 {
t.Fatalf("bad create index: %v", summary.CreateIndex)
}
2017-02-07 00:46:23 +00:00
// Upsert an allocation
alloc := mock.Alloc()
alloc.JobID = job.ID
alloc.Job = job
state.UpsertAllocs(910, []*structs.Allocation{alloc})
// Update the alloc from client
alloc1 := alloc.Copy()
alloc1.ClientStatus = structs.AllocClientStatusPending
alloc1.DesiredStatus = ""
state.UpdateAllocsFromClient(920, []*structs.Allocation{alloc})
alloc3 := alloc.Copy()
alloc3.ClientStatus = structs.AllocClientStatusRunning
alloc3.DesiredStatus = ""
state.UpdateAllocsFromClient(930, []*structs.Allocation{alloc3})
// Upsert the alloc
alloc4 := alloc.Copy()
alloc4.ClientStatus = structs.AllocClientStatusPending
alloc4.DesiredStatus = structs.AllocDesiredStatusRun
state.UpsertAllocs(950, []*structs.Allocation{alloc4})
// Again upsert the alloc
alloc5 := alloc.Copy()
alloc5.ClientStatus = structs.AllocClientStatusPending
alloc5.DesiredStatus = structs.AllocDesiredStatusRun
state.UpsertAllocs(970, []*structs.Allocation{alloc5})
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
expectedSummary := structs.JobSummary{
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{
Running: 1,
},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
CreateIndex: 900,
ModifyIndex: 930,
}
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job.ID)
if !reflect.DeepEqual(&expectedSummary, summary) {
t.Fatalf("expected: %#v, actual: %v", expectedSummary, summary)
}
// De-register the job.
state.DeleteJob(980, job.ID)
// Shouldn't have any effect on the summary
alloc6 := alloc.Copy()
alloc6.ClientStatus = structs.AllocClientStatusRunning
alloc6.DesiredStatus = ""
state.UpdateAllocsFromClient(990, []*structs.Allocation{alloc6})
// We shouldn't have any summary at this point
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job.ID)
if summary != nil {
t.Fatalf("expected nil, actual: %#v", summary)
}
// Re-register the same job
job1 := mock.Job()
job1.ID = job.ID
state.UpsertJob(1000, job1)
2017-02-07 00:46:23 +00:00
outJob2, _ := state.JobByID(ws, job1.ID)
if outJob2.CreateIndex != 1000 {
t.Fatalf("bad create index: %v", outJob2.CreateIndex)
}
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job1.ID)
if summary.CreateIndex != 1000 {
t.Fatalf("bad create index: %v", summary.CreateIndex)
}
// Upsert an allocation
alloc7 := alloc.Copy()
alloc7.JobID = outJob.ID
alloc7.Job = outJob
alloc7.ClientStatus = structs.AllocClientStatusComplete
alloc7.DesiredStatus = structs.AllocDesiredStatusRun
state.UpdateAllocsFromClient(1020, []*structs.Allocation{alloc7})
expectedSummary = structs.JobSummary{
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
CreateIndex: 1000,
ModifyIndex: 1000,
}
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job1.ID)
if !reflect.DeepEqual(&expectedSummary, summary) {
t.Fatalf("expected: %#v, actual: %#v", expectedSummary, summary)
}
}
func TestStateStore_ReconcileJobSummary(t *testing.T) {
state := testStateStore(t)
2016-08-04 01:08:37 +00:00
// Create an alloc
alloc := mock.Alloc()
2016-08-04 01:08:37 +00:00
// Add another task group to the job
tg2 := alloc.Job.TaskGroups[0].Copy()
tg2.Name = "db"
alloc.Job.TaskGroups = append(alloc.Job.TaskGroups, tg2)
state.UpsertJob(100, alloc.Job)
2016-08-04 01:08:37 +00:00
// Create one more alloc for the db task group
alloc2 := mock.Alloc()
alloc2.TaskGroup = "db"
alloc2.JobID = alloc.JobID
alloc2.Job = alloc.Job
2016-08-04 01:08:37 +00:00
// Upserts the alloc
state.UpsertAllocs(110, []*structs.Allocation{alloc, alloc2})
// Change the state of the first alloc to running
alloc3 := alloc.Copy()
alloc3.ClientStatus = structs.AllocClientStatusRunning
state.UpdateAllocsFromClient(120, []*structs.Allocation{alloc3})
//Add some more allocs to the second tg
alloc4 := mock.Alloc()
alloc4.JobID = alloc.JobID
alloc4.Job = alloc.Job
alloc4.TaskGroup = "db"
alloc5 := alloc4.Copy()
alloc5.ClientStatus = structs.AllocClientStatusRunning
alloc6 := mock.Alloc()
alloc6.JobID = alloc.JobID
alloc6.Job = alloc.Job
alloc6.TaskGroup = "db"
alloc7 := alloc6.Copy()
alloc7.ClientStatus = structs.AllocClientStatusComplete
alloc8 := mock.Alloc()
alloc8.JobID = alloc.JobID
alloc8.Job = alloc.Job
alloc8.TaskGroup = "db"
alloc9 := alloc8.Copy()
alloc9.ClientStatus = structs.AllocClientStatusFailed
alloc10 := mock.Alloc()
alloc10.JobID = alloc.JobID
alloc10.Job = alloc.Job
alloc10.TaskGroup = "db"
alloc11 := alloc10.Copy()
alloc11.ClientStatus = structs.AllocClientStatusLost
state.UpsertAllocs(130, []*structs.Allocation{alloc4, alloc6, alloc8, alloc10})
state.UpdateAllocsFromClient(150, []*structs.Allocation{alloc5, alloc7, alloc9, alloc11})
// DeleteJobSummary is a helper method and doesn't modify the indexes table
state.DeleteJobSummary(130, alloc.Job.ID)
state.ReconcileJobSummaries(120)
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
summary, _ := state.JobSummaryByID(ws, alloc.Job.ID)
expectedSummary := structs.JobSummary{
JobID: alloc.Job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{
2016-08-04 01:08:37 +00:00
Running: 1,
},
"db": structs.TaskGroupSummary{
Starting: 1,
Running: 1,
2016-08-04 01:08:37 +00:00
Failed: 1,
Complete: 1,
Lost: 1,
},
},
CreateIndex: 100,
ModifyIndex: 120,
}
if !reflect.DeepEqual(&expectedSummary, summary) {
t.Fatalf("expected: %v, actual: %v", expectedSummary, summary)
}
}
func TestStateStore_UpdateAlloc_JobNotPresent(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
state.UpsertJob(100, alloc.Job)
state.UpsertAllocs(200, []*structs.Allocation{alloc})
// Delete the job
state.DeleteJob(300, alloc.Job.ID)
// Update the alloc
alloc1 := alloc.Copy()
alloc1.ClientStatus = structs.AllocClientStatusRunning
// Updating allocation should not throw any error
if err := state.UpdateAllocsFromClient(400, []*structs.Allocation{alloc1}); err != nil {
t.Fatalf("expect err: %v", err)
}
// Re-Register the job
state.UpsertJob(500, alloc.Job)
// Update the alloc again
alloc2 := alloc.Copy()
alloc2.ClientStatus = structs.AllocClientStatusComplete
if err := state.UpdateAllocsFromClient(400, []*structs.Allocation{alloc1}); err != nil {
t.Fatalf("expect err: %v", err)
}
// Job Summary of the newly registered job shouldn't account for the
// allocation update for the older job
expectedSummary := structs.JobSummary{
JobID: alloc1.JobID,
Summary: map[string]structs.TaskGroupSummary{
"web": structs.TaskGroupSummary{},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
CreateIndex: 500,
ModifyIndex: 500,
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
summary, _ := state.JobSummaryByID(ws, alloc.Job.ID)
if !reflect.DeepEqual(&expectedSummary, summary) {
t.Fatalf("expected: %v, actual: %v", expectedSummary, summary)
}
}
2015-09-07 03:51:01 +00:00
func TestStateStore_EvictAlloc_Alloc(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
2016-07-21 21:43:21 +00:00
state.UpsertJobSummary(999, mock.JobSummary(alloc.JobID))
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
if err != nil {
t.Fatalf("err: %v", err)
}
2015-08-25 23:26:34 +00:00
alloc2 := new(structs.Allocation)
*alloc2 = *alloc
alloc2.DesiredStatus = structs.AllocDesiredStatusEvict
err = state.UpsertAllocs(1001, []*structs.Allocation{alloc2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
2015-08-25 23:26:34 +00:00
if out.DesiredStatus != structs.AllocDesiredStatusEvict {
t.Fatalf("bad: %#v %#v", alloc, out)
}
2015-09-07 03:51:01 +00:00
index, err := state.Index("allocs")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
}
2015-08-05 01:32:50 +00:00
func TestStateStore_AllocsByNode(t *testing.T) {
state := testStateStore(t)
var allocs []*structs.Allocation
for i := 0; i < 10; i++ {
alloc := mock.Alloc()
2015-08-05 01:32:50 +00:00
alloc.NodeID = "foo"
allocs = append(allocs, alloc)
}
2016-07-21 21:43:21 +00:00
for idx, alloc := range allocs {
state.UpsertJobSummary(uint64(900+idx), mock.JobSummary(alloc.JobID))
}
err := state.UpsertAllocs(1000, allocs)
2015-08-05 01:32:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocsByNode(ws, "foo")
2015-08-05 01:32:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
sort.Sort(AllocIDSort(allocs))
sort.Sort(AllocIDSort(out))
if !reflect.DeepEqual(allocs, out) {
t.Fatalf("bad: %#v %#v", allocs, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-08-05 01:32:50 +00:00
}
func TestStateStore_AllocsByNodeTerminal(t *testing.T) {
state := testStateStore(t)
var allocs, term, nonterm []*structs.Allocation
for i := 0; i < 10; i++ {
alloc := mock.Alloc()
alloc.NodeID = "foo"
if i%2 == 0 {
alloc.DesiredStatus = structs.AllocDesiredStatusStop
term = append(term, alloc)
} else {
nonterm = append(nonterm, alloc)
}
allocs = append(allocs, alloc)
}
2016-07-21 21:43:21 +00:00
for idx, alloc := range allocs {
state.UpsertJobSummary(uint64(900+idx), mock.JobSummary(alloc.JobID))
}
err := state.UpsertAllocs(1000, allocs)
if err != nil {
t.Fatalf("err: %v", err)
}
// Verify the terminal allocs
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocsByNodeTerminal(ws, "foo", true)
if err != nil {
t.Fatalf("err: %v", err)
}
sort.Sort(AllocIDSort(term))
sort.Sort(AllocIDSort(out))
if !reflect.DeepEqual(term, out) {
t.Fatalf("bad: %#v %#v", term, out)
}
// Verify the non-terminal allocs
2017-02-07 00:46:23 +00:00
out, err = state.AllocsByNodeTerminal(ws, "foo", false)
if err != nil {
t.Fatalf("err: %v", err)
}
sort.Sort(AllocIDSort(nonterm))
sort.Sort(AllocIDSort(out))
if !reflect.DeepEqual(nonterm, out) {
t.Fatalf("bad: %#v %#v", nonterm, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
2015-08-07 00:36:10 +00:00
func TestStateStore_AllocsByJob(t *testing.T) {
state := testStateStore(t)
var allocs []*structs.Allocation
for i := 0; i < 10; i++ {
alloc := mock.Alloc()
2015-08-07 00:36:10 +00:00
alloc.JobID = "foo"
allocs = append(allocs, alloc)
}
2016-07-21 21:43:21 +00:00
for i, alloc := range allocs {
state.UpsertJobSummary(uint64(900+i), mock.JobSummary(alloc.JobID))
}
err := state.UpsertAllocs(1000, allocs)
2015-08-07 00:36:10 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocsByJob(ws, "foo", false)
2015-08-07 00:36:10 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
sort.Sort(AllocIDSort(allocs))
sort.Sort(AllocIDSort(out))
if !reflect.DeepEqual(allocs, out) {
t.Fatalf("bad: %#v %#v", allocs, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2015-08-07 00:36:10 +00:00
}
func TestStateStore_AllocsForRegisteredJob(t *testing.T) {
state := testStateStore(t)
var allocs []*structs.Allocation
var allocs1 []*structs.Allocation
job := mock.Job()
job.ID = "foo"
state.UpsertJob(100, job)
for i := 0; i < 3; i++ {
alloc := mock.Alloc()
alloc.Job = job
alloc.JobID = job.ID
allocs = append(allocs, alloc)
}
if err := state.UpsertAllocs(200, allocs); err != nil {
t.Fatalf("err: %v", err)
}
if err := state.DeleteJob(250, job.ID); err != nil {
t.Fatalf("err: %v", err)
}
job1 := mock.Job()
job1.ID = "foo"
job1.CreateIndex = 50
state.UpsertJob(300, job1)
for i := 0; i < 4; i++ {
alloc := mock.Alloc()
alloc.Job = job1
alloc.JobID = job1.ID
allocs1 = append(allocs1, alloc)
}
if err := state.UpsertAllocs(1000, allocs1); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocsByJob(ws, job1.ID, true)
if err != nil {
t.Fatalf("err: %v", err)
}
expected := len(allocs) + len(allocs1)
if len(out) != expected {
t.Fatalf("expected: %v, actual: %v", expected, len(out))
}
2017-02-07 00:46:23 +00:00
out1, err := state.AllocsByJob(ws, job1.ID, false)
expected = len(allocs1)
if len(out1) != expected {
t.Fatalf("expected: %v, actual: %v", expected, len(out1))
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_AllocsByIDPrefix(t *testing.T) {
state := testStateStore(t)
var allocs []*structs.Allocation
ids := []string{
"aaaaaaaa-7bfb-395d-eb95-0685af2176b2",
"aaaaaaab-7bfb-395d-eb95-0685af2176b2",
"aaaaaabb-7bfb-395d-eb95-0685af2176b2",
"aaaaabbb-7bfb-395d-eb95-0685af2176b2",
"aaaabbbb-7bfb-395d-eb95-0685af2176b2",
"aaabbbbb-7bfb-395d-eb95-0685af2176b2",
"aabbbbbb-7bfb-395d-eb95-0685af2176b2",
"abbbbbbb-7bfb-395d-eb95-0685af2176b2",
"bbbbbbbb-7bfb-395d-eb95-0685af2176b2",
}
for i := 0; i < 9; i++ {
alloc := mock.Alloc()
alloc.ID = ids[i]
allocs = append(allocs, alloc)
}
2016-07-21 21:43:21 +00:00
for i, alloc := range allocs {
state.UpsertJobSummary(uint64(900+i), mock.JobSummary(alloc.JobID))
}
err := state.UpsertAllocs(1000, allocs)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.AllocsByIDPrefix(ws, "aaaa")
if err != nil {
t.Fatalf("err: %v", err)
}
gatherAllocs := func(iter memdb.ResultIterator) []*structs.Allocation {
var allocs []*structs.Allocation
for {
raw := iter.Next()
if raw == nil {
break
}
allocs = append(allocs, raw.(*structs.Allocation))
}
return allocs
}
out := gatherAllocs(iter)
if len(out) != 5 {
t.Fatalf("bad: expected five allocations, got: %#v", out)
}
sort.Sort(AllocIDSort(allocs))
for index, alloc := range out {
if ids[index] != alloc.ID {
t.Fatalf("bad: got unexpected id: %s", alloc.ID)
}
}
2017-02-07 00:46:23 +00:00
iter, err = state.AllocsByIDPrefix(ws, "b-a7bfb")
if err != nil {
t.Fatalf("err: %v", err)
}
out = gatherAllocs(iter)
if len(out) != 0 {
t.Fatalf("bad: unexpected zero allocations, got: %#v", out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_Allocs(t *testing.T) {
state := testStateStore(t)
var allocs []*structs.Allocation
for i := 0; i < 10; i++ {
alloc := mock.Alloc()
allocs = append(allocs, alloc)
}
2016-07-21 21:43:21 +00:00
for i, alloc := range allocs {
state.UpsertJobSummary(uint64(900+i), mock.JobSummary(alloc.JobID))
}
err := state.UpsertAllocs(1000, allocs)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
iter, err := state.Allocs(ws)
if err != nil {
t.Fatalf("err: %v", err)
}
var out []*structs.Allocation
for {
raw := iter.Next()
if raw == nil {
break
}
out = append(out, raw.(*structs.Allocation))
}
sort.Sort(AllocIDSort(allocs))
sort.Sort(AllocIDSort(out))
if !reflect.DeepEqual(allocs, out) {
t.Fatalf("bad: %#v %#v", allocs, out)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_RestoreAlloc(t *testing.T) {
state := testStateStore(t)
2015-10-30 15:27:47 +00:00
alloc := mock.Alloc()
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.AllocRestore(alloc)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, alloc) {
t.Fatalf("Bad: %#v %#v", out, alloc)
}
2015-10-30 15:27:47 +00:00
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_RestoreAlloc_NoEphemeralDisk(t *testing.T) {
2016-09-02 00:41:50 +00:00
state := testStateStore(t)
alloc := mock.Alloc()
alloc.Job.TaskGroups[0].EphemeralDisk = nil
2016-09-02 00:41:50 +00:00
alloc.Job.TaskGroups[0].Tasks[0].Resources.DiskMB = 120
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.AllocRestore(alloc)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.AllocByID(ws, alloc.ID)
2016-09-02 00:41:50 +00:00
if err != nil {
t.Fatalf("err: %v", err)
}
expected := alloc.Copy()
expected.Job.TaskGroups[0].EphemeralDisk = &structs.EphemeralDisk{SizeMB: 120}
2016-09-02 00:41:50 +00:00
expected.Job.TaskGroups[0].Tasks[0].Resources.DiskMB = 0
if !reflect.DeepEqual(out, expected) {
t.Fatalf("Bad: %#v %#v", out, expected)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
2016-09-02 00:41:50 +00:00
}
2016-01-09 02:22:59 +00:00
func TestStateStore_SetJobStatus_ForceStatus(t *testing.T) {
2016-01-12 01:34:25 +00:00
state := testStateStore(t)
txn := state.db.Txn(true)
// Create and insert a mock job.
2016-01-09 02:22:59 +00:00
job := mock.Job()
job.Status = ""
2016-01-12 01:34:25 +00:00
job.ModifyIndex = 0
if err := txn.Insert("jobs", job); err != nil {
t.Fatalf("job insert failed: %v", err)
}
2016-01-09 02:22:59 +00:00
exp := "foobar"
2016-01-12 01:34:25 +00:00
index := uint64(1000)
2017-02-07 00:46:23 +00:00
if err := state.setJobStatus(index, txn, job, false, exp); err != nil {
2016-01-09 02:22:59 +00:00
t.Fatalf("setJobStatus() failed: %v", err)
}
2016-01-12 01:34:25 +00:00
i, err := txn.First("jobs", "id", job.ID)
if err != nil {
t.Fatalf("job lookup failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
updated := i.(*structs.Job)
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
if updated.Status != exp {
t.Fatalf("setJobStatus() set %v; expected %v", updated.Status, exp)
}
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
if updated.ModifyIndex != index {
t.Fatalf("setJobStatus() set %d; expected %d", updated.ModifyIndex, index)
}
}
func TestStateStore_SetJobStatus_NoOp(t *testing.T) {
2016-01-09 02:22:59 +00:00
state := testStateStore(t)
2016-01-12 01:34:25 +00:00
txn := state.db.Txn(true)
// Create and insert a mock job that should be pending.
job := mock.Job()
job.Status = structs.JobStatusPending
job.ModifyIndex = 10
if err := txn.Insert("jobs", job); err != nil {
t.Fatalf("job insert failed: %v", err)
}
index := uint64(1000)
2017-02-07 00:46:23 +00:00
if err := state.setJobStatus(index, txn, job, false, ""); err != nil {
2016-01-09 02:22:59 +00:00
t.Fatalf("setJobStatus() failed: %v", err)
}
2016-01-12 01:34:25 +00:00
i, err := txn.First("jobs", "id", job.ID)
if err != nil {
t.Fatalf("job lookup failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
updated := i.(*structs.Job)
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
if updated.ModifyIndex == index {
t.Fatalf("setJobStatus() should have been a no-op")
}
}
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
func TestStateStore_SetJobStatus(t *testing.T) {
2016-01-09 02:22:59 +00:00
state := testStateStore(t)
2016-01-12 01:34:25 +00:00
txn := state.db.Txn(true)
// Create and insert a mock job that should be pending but has an incorrect
// status.
job := mock.Job()
job.Status = "foobar"
job.ModifyIndex = 10
if err := txn.Insert("jobs", job); err != nil {
t.Fatalf("job insert failed: %v", err)
}
index := uint64(1000)
2017-02-07 00:46:23 +00:00
if err := state.setJobStatus(index, txn, job, false, ""); err != nil {
2016-01-09 02:22:59 +00:00
t.Fatalf("setJobStatus() failed: %v", err)
}
2016-01-12 01:34:25 +00:00
i, err := txn.First("jobs", "id", job.ID)
if err != nil {
t.Fatalf("job lookup failed: %v", err)
}
updated := i.(*structs.Job)
if updated.Status != structs.JobStatusPending {
t.Fatalf("setJobStatus() set %v; expected %v", updated.Status, structs.JobStatusPending)
}
if updated.ModifyIndex != index {
t.Fatalf("setJobStatus() set %d; expected %d", updated.ModifyIndex, index)
2016-01-09 02:22:59 +00:00
}
}
2016-01-12 01:34:25 +00:00
func TestStateStore_GetJobStatus_NoEvalsOrAllocs(t *testing.T) {
2016-01-09 02:22:59 +00:00
job := mock.Job()
2016-01-12 01:34:25 +00:00
state := testStateStore(t)
txn := state.db.Txn(false)
status, err := state.getJobStatus(txn, job, false)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
}
if status != structs.JobStatusPending {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusPending)
}
}
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
func TestStateStore_GetJobStatus_NoEvalsOrAllocs_Periodic(t *testing.T) {
job := mock.PeriodicJob()
2016-01-09 02:22:59 +00:00
state := testStateStore(t)
txn := state.db.Txn(false)
2016-01-12 01:34:25 +00:00
status, err := state.getJobStatus(txn, job, false)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
if status != structs.JobStatusRunning {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusRunning)
2016-01-09 02:22:59 +00:00
}
}
2016-01-12 01:34:25 +00:00
func TestStateStore_GetJobStatus_NoEvalsOrAllocs_EvalDelete(t *testing.T) {
job := mock.Job()
2016-01-09 02:22:59 +00:00
state := testStateStore(t)
2016-01-12 01:34:25 +00:00
txn := state.db.Txn(false)
status, err := state.getJobStatus(txn, job, true)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
}
2016-01-09 02:22:59 +00:00
2016-01-12 01:34:25 +00:00
if status != structs.JobStatusDead {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusDead)
}
}
func TestStateStore_GetJobStatus_DeadEvalsAndAllocs(t *testing.T) {
state := testStateStore(t)
2016-01-09 02:22:59 +00:00
job := mock.Job()
// Create a mock alloc that is dead.
alloc := mock.Alloc()
alloc.JobID = job.ID
2016-07-20 23:07:15 +00:00
alloc.DesiredStatus = structs.AllocDesiredStatusStop
2016-07-21 21:43:21 +00:00
state.UpsertJobSummary(999, mock.JobSummary(alloc.JobID))
2016-01-09 02:22:59 +00:00
if err := state.UpsertAllocs(1000, []*structs.Allocation{alloc}); err != nil {
t.Fatalf("err: %v", err)
}
// Create a mock eval that is complete
eval := mock.Eval()
eval.JobID = job.ID
eval.Status = structs.EvalStatusComplete
if err := state.UpsertEvals(1001, []*structs.Evaluation{eval}); err != nil {
t.Fatalf("err: %v", err)
}
txn := state.db.Txn(false)
2016-01-12 01:34:25 +00:00
status, err := state.getJobStatus(txn, job, false)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
if status != structs.JobStatusDead {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusDead)
2016-01-09 02:22:59 +00:00
}
}
2016-01-12 01:34:25 +00:00
func TestStateStore_GetJobStatus_RunningAlloc(t *testing.T) {
2016-01-09 02:22:59 +00:00
state := testStateStore(t)
job := mock.Job()
// Create a mock alloc that is running.
alloc := mock.Alloc()
alloc.JobID = job.ID
alloc.DesiredStatus = structs.AllocDesiredStatusRun
2016-07-21 21:43:21 +00:00
state.UpsertJobSummary(999, mock.JobSummary(alloc.JobID))
2016-01-09 02:22:59 +00:00
if err := state.UpsertAllocs(1000, []*structs.Allocation{alloc}); err != nil {
t.Fatalf("err: %v", err)
}
txn := state.db.Txn(false)
2016-01-12 01:34:25 +00:00
status, err := state.getJobStatus(txn, job, true)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
if status != structs.JobStatusRunning {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusRunning)
2016-01-09 02:22:59 +00:00
}
}
func TestStateStore_SetJobStatus_PendingEval(t *testing.T) {
state := testStateStore(t)
job := mock.Job()
// Create a mock eval that is pending.
eval := mock.Eval()
eval.JobID = job.ID
eval.Status = structs.EvalStatusPending
if err := state.UpsertEvals(1000, []*structs.Evaluation{eval}); err != nil {
t.Fatalf("err: %v", err)
}
txn := state.db.Txn(false)
2016-01-12 01:34:25 +00:00
status, err := state.getJobStatus(txn, job, true)
if err != nil {
t.Fatalf("getJobStatus() failed: %v", err)
2016-01-09 02:22:59 +00:00
}
2016-01-12 01:34:25 +00:00
if status != structs.JobStatusPending {
t.Fatalf("getJobStatus() returned %v; expected %v", status, structs.JobStatusPending)
2016-01-09 02:22:59 +00:00
}
}
func TestStateJobSummary_UpdateJobCount(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
job := alloc.Job
job.TaskGroups[0].Count = 3
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that upsert fires the watch
ws := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws, job.ID); err != nil {
t.Fatalf("bad: %v", err)
}
if err := state.UpsertJob(1000, job); err != nil {
t.Fatalf("err: %v", err)
}
if err := state.UpsertAllocs(1001, []*structs.Allocation{alloc}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
summary, _ := state.JobSummaryByID(ws, job.ID)
expectedSummary := structs.JobSummary{
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": {
Starting: 1,
},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
CreateIndex: 1000,
ModifyIndex: 1001,
}
if !reflect.DeepEqual(summary, &expectedSummary) {
t.Fatalf("expected: %v, actual: %v", expectedSummary, summary)
}
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that upsert fires the watch
ws2 := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws2, job.ID); err != nil {
t.Fatalf("bad: %v", err)
}
alloc2 := mock.Alloc()
alloc2.Job = job
alloc2.JobID = job.ID
alloc3 := mock.Alloc()
alloc3.Job = job
alloc3.JobID = job.ID
if err := state.UpsertAllocs(1002, []*structs.Allocation{alloc2, alloc3}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws2) {
t.Fatalf("bad")
}
outA, _ := state.AllocByID(ws, alloc3.ID)
2016-07-21 21:43:21 +00:00
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job.ID)
expectedSummary = structs.JobSummary{
2016-07-12 20:37:51 +00:00
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": {
Starting: 3,
},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
2016-07-21 21:43:21 +00:00
CreateIndex: job.CreateIndex,
ModifyIndex: outA.ModifyIndex,
2016-07-12 20:37:51 +00:00
}
if !reflect.DeepEqual(summary, &expectedSummary) {
t.Fatalf("expected summary: %v, actual: %v", expectedSummary, summary)
}
2017-02-07 00:46:23 +00:00
// Create watchsets so we can test that upsert fires the watch
ws3 := memdb.NewWatchSet()
if _, err := state.JobSummaryByID(ws3, job.ID); err != nil {
t.Fatalf("bad: %v", err)
}
alloc4 := mock.Alloc()
alloc4.ID = alloc2.ID
alloc4.Job = alloc2.Job
alloc4.JobID = alloc2.JobID
2016-07-12 20:27:45 +00:00
alloc4.ClientStatus = structs.AllocClientStatusComplete
alloc5 := mock.Alloc()
alloc5.ID = alloc3.ID
alloc5.Job = alloc3.Job
alloc5.JobID = alloc3.JobID
2016-07-12 20:27:45 +00:00
alloc5.ClientStatus = structs.AllocClientStatusComplete
if err := state.UpdateAllocsFromClient(1004, []*structs.Allocation{alloc4, alloc5}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws2) {
t.Fatalf("bad")
}
outA, _ = state.AllocByID(ws, alloc5.ID)
summary, _ = state.JobSummaryByID(ws, job.ID)
2016-07-12 20:37:51 +00:00
expectedSummary = structs.JobSummary{
JobID: job.ID,
Summary: map[string]structs.TaskGroupSummary{
"web": {
Complete: 2,
Starting: 1,
},
},
2016-12-16 18:21:56 +00:00
Children: new(structs.JobChildrenSummary),
2016-07-21 21:43:21 +00:00
CreateIndex: job.CreateIndex,
ModifyIndex: outA.ModifyIndex,
2016-07-12 20:37:51 +00:00
}
if !reflect.DeepEqual(summary, &expectedSummary) {
t.Fatalf("expected: %v, actual: %v", expectedSummary, summary)
}
}
func TestJobSummary_UpdateClientStatus(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
job := alloc.Job
job.TaskGroups[0].Count = 3
alloc2 := mock.Alloc()
alloc2.Job = job
alloc2.JobID = job.ID
alloc3 := mock.Alloc()
alloc3.Job = job
alloc3.JobID = job.ID
err := state.UpsertJob(1000, job)
if err != nil {
t.Fatalf("err: %v", err)
}
if err := state.UpsertAllocs(1001, []*structs.Allocation{alloc, alloc2, alloc3}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
summary, _ := state.JobSummaryByID(ws, job.ID)
if summary.Summary["web"].Starting != 3 {
t.Fatalf("bad job summary: %v", summary)
}
alloc4 := mock.Alloc()
alloc4.ID = alloc2.ID
alloc4.Job = alloc2.Job
alloc4.JobID = alloc2.JobID
alloc4.ClientStatus = structs.AllocClientStatusComplete
alloc5 := mock.Alloc()
alloc5.ID = alloc3.ID
alloc5.Job = alloc3.Job
alloc5.JobID = alloc3.JobID
alloc5.ClientStatus = structs.AllocClientStatusFailed
alloc6 := mock.Alloc()
alloc6.ID = alloc.ID
alloc6.Job = alloc.Job
alloc6.JobID = alloc.JobID
alloc6.ClientStatus = structs.AllocClientStatusRunning
if err := state.UpdateAllocsFromClient(1002, []*structs.Allocation{alloc4, alloc5, alloc6}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
summary, _ = state.JobSummaryByID(ws, job.ID)
if summary.Summary["web"].Running != 1 || summary.Summary["web"].Failed != 1 || summary.Summary["web"].Complete != 1 {
t.Fatalf("bad job summary: %v", summary)
}
alloc7 := mock.Alloc()
alloc7.Job = alloc.Job
alloc7.JobID = alloc.JobID
if err := state.UpsertAllocs(1003, []*structs.Allocation{alloc7}); err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
summary, _ = state.JobSummaryByID(ws, job.ID)
if summary.Summary["web"].Starting != 1 || summary.Summary["web"].Running != 1 || summary.Summary["web"].Failed != 1 || summary.Summary["web"].Complete != 1 {
t.Fatalf("bad job summary: %v", summary)
}
}
func TestStateStore_UpsertVaultAccessors(t *testing.T) {
state := testStateStore(t)
a := mock.VaultAccessor()
a2 := mock.VaultAccessor()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
if _, err := state.VaultAccessor(ws, a.Accessor); err != nil {
t.Fatalf("err: %v", err)
}
if _, err := state.VaultAccessor(ws, a2.Accessor); err != nil {
t.Fatalf("err: %v", err)
}
err := state.UpsertVaultAccessor(1000, []*structs.VaultAccessor{a, a2})
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.VaultAccessor(ws, a.Accessor)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(a, out) {
t.Fatalf("bad: %#v %#v", a, out)
}
2017-02-07 00:46:23 +00:00
out, err = state.VaultAccessor(ws, a2.Accessor)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(a2, out) {
t.Fatalf("bad: %#v %#v", a2, out)
}
2017-02-07 00:46:23 +00:00
iter, err := state.VaultAccessors(ws)
if err != nil {
t.Fatalf("err: %v", err)
}
count := 0
for {
raw := iter.Next()
if raw == nil {
break
}
count++
accessor := raw.(*structs.VaultAccessor)
if !reflect.DeepEqual(accessor, a) && !reflect.DeepEqual(accessor, a2) {
t.Fatalf("bad: %#v", accessor)
}
}
if count != 2 {
t.Fatalf("bad: %d", count)
}
index, err := state.Index("vault_accessors")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_DeleteVaultAccessors(t *testing.T) {
state := testStateStore(t)
a1 := mock.VaultAccessor()
a2 := mock.VaultAccessor()
accessors := []*structs.VaultAccessor{a1, a2}
err := state.UpsertVaultAccessor(1000, accessors)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
if _, err := state.VaultAccessor(ws, a1.Accessor); err != nil {
t.Fatalf("err: %v", err)
}
err = state.DeleteVaultAccessors(1001, accessors)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
if !watchFired(ws) {
t.Fatalf("bad")
}
ws = memdb.NewWatchSet()
out, err := state.VaultAccessor(ws, a1.Accessor)
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
t.Fatalf("bad: %#v %#v", a1, out)
}
2017-02-07 00:46:23 +00:00
out, err = state.VaultAccessor(ws, a2.Accessor)
if err != nil {
t.Fatalf("err: %v", err)
}
if out != nil {
t.Fatalf("bad: %#v %#v", a2, out)
}
index, err := state.Index("vault_accessors")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1001 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_VaultAccessorsByAlloc(t *testing.T) {
state := testStateStore(t)
alloc := mock.Alloc()
var accessors []*structs.VaultAccessor
var expected []*structs.VaultAccessor
for i := 0; i < 5; i++ {
accessor := mock.VaultAccessor()
accessor.AllocID = alloc.ID
expected = append(expected, accessor)
accessors = append(accessors, accessor)
}
for i := 0; i < 10; i++ {
accessor := mock.VaultAccessor()
accessors = append(accessors, accessor)
}
err := state.UpsertVaultAccessor(1000, accessors)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.VaultAccessorsByAlloc(ws, alloc.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if len(expected) != len(out) {
t.Fatalf("bad: %#v %#v", len(expected), len(out))
}
index, err := state.Index("vault_accessors")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_VaultAccessorsByNode(t *testing.T) {
state := testStateStore(t)
node := mock.Node()
var accessors []*structs.VaultAccessor
var expected []*structs.VaultAccessor
for i := 0; i < 5; i++ {
accessor := mock.VaultAccessor()
accessor.NodeID = node.ID
expected = append(expected, accessor)
accessors = append(accessors, accessor)
}
for i := 0; i < 10; i++ {
accessor := mock.VaultAccessor()
accessors = append(accessors, accessor)
}
err := state.UpsertVaultAccessor(1000, accessors)
if err != nil {
t.Fatalf("err: %v", err)
}
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.VaultAccessorsByNode(ws, node.ID)
if err != nil {
t.Fatalf("err: %v", err)
}
if len(expected) != len(out) {
t.Fatalf("bad: %#v %#v", len(expected), len(out))
}
index, err := state.Index("vault_accessors")
if err != nil {
t.Fatalf("err: %v", err)
}
if index != 1000 {
t.Fatalf("bad: %d", index)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
}
}
func TestStateStore_RestoreVaultAccessor(t *testing.T) {
state := testStateStore(t)
a := mock.VaultAccessor()
restore, err := state.Restore()
if err != nil {
t.Fatalf("err: %v", err)
}
err = restore.VaultAccessorRestore(a)
if err != nil {
t.Fatalf("err: %v", err)
}
restore.Commit()
2017-02-07 00:46:23 +00:00
ws := memdb.NewWatchSet()
out, err := state.VaultAccessor(ws, a.Accessor)
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(out, a) {
t.Fatalf("Bad: %#v %#v", out, a)
}
2017-02-07 00:46:23 +00:00
if watchFired(ws) {
t.Fatalf("bad")
2015-10-30 15:27:47 +00:00
}
}
func TestStateStore_Abandon(t *testing.T) {
s := testStateStore(t)
abandonCh := s.AbandonCh()
s.Abandon()
select {
case <-abandonCh:
default:
t.Fatalf("bad")
}
}
2017-02-07 00:46:23 +00:00
// watchFired is a helper for unit tests that returns if the given watch set
// fired (it doesn't care which watch actually fired). This uses a fixed
// timeout since we already expect the event happened before calling this and
// just need to distinguish a fire from a timeout. We do need a little time to
// allow the watch to set up any goroutines, though.
func watchFired(ws memdb.WatchSet) bool {
timedOut := ws.Watch(time.After(50 * time.Millisecond))
return !timedOut
2015-10-30 04:42:41 +00:00
}
// NodeIDSort is used to sort nodes by ID
type NodeIDSort []*structs.Node
func (n NodeIDSort) Len() int {
return len(n)
}
func (n NodeIDSort) Less(i, j int) bool {
return n[i].ID < n[j].ID
}
func (n NodeIDSort) Swap(i, j int) {
n[i], n[j] = n[j], n[i]
}
2015-07-07 16:41:05 +00:00
// JobIDis used to sort jobs by id
type JobIDSort []*structs.Job
2015-07-07 16:41:05 +00:00
func (n JobIDSort) Len() int {
2015-07-07 16:41:05 +00:00
return len(n)
}
func (n JobIDSort) Less(i, j int) bool {
return n[i].ID < n[j].ID
2015-07-07 16:41:05 +00:00
}
func (n JobIDSort) Swap(i, j int) {
2015-07-07 16:41:05 +00:00
n[i], n[j] = n[j], n[i]
}
2015-07-23 22:43:06 +00:00
// EvalIDis used to sort evals by id
type EvalIDSort []*structs.Evaluation
func (n EvalIDSort) Len() int {
return len(n)
}
func (n EvalIDSort) Less(i, j int) bool {
return n[i].ID < n[j].ID
}
func (n EvalIDSort) Swap(i, j int) {
n[i], n[j] = n[j], n[i]
}
// AllocIDsort used to sort allocations by id
type AllocIDSort []*structs.Allocation
func (n AllocIDSort) Len() int {
return len(n)
}
func (n AllocIDSort) Less(i, j int) bool {
return n[i].ID < n[j].ID
}
func (n AllocIDSort) Swap(i, j int) {
n[i], n[j] = n[j], n[i]
}