open-nomad/api/nodes_test.go

package api

import (
	"context"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/hashicorp/nomad/api/internal/testutil"
	"github.com/stretchr/testify/require"
)

func TestNodes_List(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	var qm *QueryMeta
	var out []*NodeListStub
	var err error

	testutil.WaitForResult(func() (bool, error) {
		out, qm, err = nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Check that we got valid QueryMeta.
	assertQueryMeta(t, qm)
}

func TestNodes_PrefixList(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	var qm *QueryMeta
	var out []*NodeListStub
	var err error

	// Get the node ID
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Find node based on four character prefix
	out, qm, err = nodes.PrefixList(nodeID[:4])
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	if n := len(out); n != 1 {
		t.Fatalf("expected 1 node, got: %d ", n)
	}

	// Check that we got valid QueryMeta.
	assertQueryMeta(t, qm)
}

// TestNodes_List_Resources asserts that ?resources=true includes allocated and
// reserved resources in the response.
func TestNodes_List_Resources(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	var out []*NodeListStub
	var err error

	testutil.WaitForResult(func() (bool, error) {
		out, _, err = nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// By default resources should *not* be included
	require.Nil(t, out[0].NodeResources)
	require.Nil(t, out[0].ReservedResources)

	qo := &QueryOptions{
		Params: map[string]string{"resources": "true"},
	}
	out, _, err = nodes.List(qo)
	require.NoError(t, err)
	require.NotNil(t, out[0].NodeResources)
	require.NotNil(t, out[0].ReservedResources)
}

func TestNodes_Info(t *testing.T) {
	t.Parallel()
	startTime := time.Now().Unix()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	// Retrieving a nonexistent node returns error
	_, _, err := nodes.Info("12345678-abcd-efab-cdef-123456789abc", nil)
	if err == nil || !strings.Contains(err.Error(), "not found") {
		t.Fatalf("expected not found error, got: %#v", err)
	}

	// Get the node ID
	var nodeID, dc string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		dc = out[0].Datacenter
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Querying for existing nodes returns properly
	result, qm, err := nodes.Info(nodeID, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	assertQueryMeta(t, qm)

	// Check that the result is what we expect
	if result.ID != nodeID || result.Datacenter != dc {
		t.Fatalf("expected %s (%s), got: %s (%s)",
			nodeID, dc,
			result.ID, result.Datacenter)
	}

	require.Equal(t, 20000, result.NodeResources.MinDynamicPort)
	require.Equal(t, 32000, result.NodeResources.MaxDynamicPort)

	// Check that the StatusUpdatedAt field is being populated correctly
	if result.StatusUpdatedAt < startTime {
		t.Fatalf("start time: %v, status updated: %v", startTime, result.StatusUpdatedAt)
	}

	if len(result.Events) < 1 {
		t.Fatalf("Expected at minimum the node register event to be populated: %+v", result)
	}
}

func TestNodes_NoSecretID(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	// Get the node ID
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// perform a raw http call and make sure that:
	// - "ID" to make sure that raw decoding is working correctly
	// - "SecretID" to make sure it's not present
	resp := make(map[string]interface{})
	_, err := c.query("/v1/node/"+nodeID, &resp, nil)
	require.NoError(t, err)
	require.Equal(t, nodeID, resp["ID"])
	require.Empty(t, resp["SecretID"])
}

func TestNodes_ToggleDrain(t *testing.T) {
	t.Parallel()
	require := require.New(t)
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	// Wait for node registration and get the ID
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Check for drain mode
	out, _, err := nodes.Info(nodeID, nil)
	require.Nil(err)
	require.False(out.Drain)
	require.Nil(out.LastDrain)

	// Toggle it on
	timeBeforeDrain := time.Now().Add(-1 * time.Second)
	spec := &DrainSpec{
		Deadline: 10 * time.Second,
	}
	drainMeta := map[string]string{
		"reason": "this node needs to go",
	}
	drainOut, err := nodes.UpdateDrainOpts(nodeID, &DrainOptions{
		DrainSpec:    spec,
		MarkEligible: false,
		Meta:         drainMeta,
	}, nil)
	require.Nil(err)
	assertWriteMeta(t, &drainOut.WriteMeta)

	// Drain may have completed before we can check, use event stream
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	streamCh, err := c.EventStream().Stream(ctx, map[Topic][]string{
		TopicNode: {nodeID},
	}, 0, nil)
	require.NoError(err)

	// we expect to see the node change to Drain:true and then back to Drain:false+ineligible
	var sawDraining, sawDrainComplete uint64
	for sawDrainComplete == 0 {
		select {
		case events := <-streamCh:
			require.NoError(events.Err)
			for _, e := range events.Events {
				node, err := e.Node()
				require.NoError(err)
				require.Equal(node.DrainStrategy != nil, node.Drain)
				require.True(!node.Drain || node.SchedulingEligibility == NodeSchedulingIneligible) // node.Drain => "ineligible"
				if node.Drain && node.SchedulingEligibility == NodeSchedulingIneligible {
					require.NotNil(node.LastDrain)
					require.Equal(DrainStatusDraining, node.LastDrain.Status)
					now := time.Now()
					require.False(node.LastDrain.StartedAt.Before(timeBeforeDrain),
						"wanted %v <= %v", node.LastDrain.StartedAt, timeBeforeDrain)
					require.False(node.LastDrain.StartedAt.After(now),
						"wanted %v <= %v", node.LastDrain.StartedAt, now)
					require.Equal(drainMeta, node.LastDrain.Meta)
					sawDraining = node.ModifyIndex
				} else if sawDraining != 0 && !node.Drain && node.SchedulingEligibility == NodeSchedulingIneligible {
					require.NotNil(node.LastDrain)
					require.Equal(DrainStatusComplete, node.LastDrain.Status)
					require.True(!node.LastDrain.UpdatedAt.Before(node.LastDrain.StartedAt))
					require.Equal(drainMeta, node.LastDrain.Meta)
					sawDrainComplete = node.ModifyIndex
				}
			}
		case <-time.After(5 * time.Second):
			require.Fail("failed waiting for event stream event")
		}
	}

	// Toggle off again
	drainOut, err = nodes.UpdateDrain(nodeID, nil, true, nil)
	require.Nil(err)
	assertWriteMeta(t, &drainOut.WriteMeta)

	// Check again
	out, _, err = nodes.Info(nodeID, nil)
	require.Nil(err)
	require.False(out.Drain)
	require.Nil(out.DrainStrategy)
	require.Equal(NodeSchedulingEligible, out.SchedulingEligibility)
}

func TestNodes_ToggleEligibility(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	// Wait for node registration and get the ID
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Check for eligibility
	out, _, err := nodes.Info(nodeID, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	if out.SchedulingEligibility != NodeSchedulingEligible {
		t.Fatalf("node should be eligible")
	}

	// Toggle it off
	eligOut, err := nodes.ToggleEligibility(nodeID, false, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	assertWriteMeta(t, &eligOut.WriteMeta)

	// Check again
	out, _, err = nodes.Info(nodeID, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	if out.SchedulingEligibility != NodeSchedulingIneligible {
		t.Fatalf("bad eligibility: %v vs %v", out.SchedulingEligibility, NodeSchedulingIneligible)
	}

	// Toggle on
	eligOut, err = nodes.ToggleEligibility(nodeID, true, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	assertWriteMeta(t, &eligOut.WriteMeta)

	// Check again
	out, _, err = nodes.Info(nodeID, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	if out.SchedulingEligibility != NodeSchedulingEligible {
		t.Fatalf("bad eligibility: %v vs %v", out.SchedulingEligibility, NodeSchedulingEligible)
	}
	if out.DrainStrategy != nil {
		t.Fatalf("drain strategy should be unset")
	}
}

func TestNodes_Allocations(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, nil)
	defer s.Stop()
	nodes := c.Nodes()

	// Looking up by a nonexistent node returns nothing. We
	// don't check the index here because it's possible the node
	// has already registered, in which case we will get a non-
	// zero result anyways.
	allocs, _, err := nodes.Allocations("nope", nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	if n := len(allocs); n != 0 {
		t.Fatalf("expected 0 allocs, got: %d", n)
	}
}

func TestNodes_ForceEvaluate(t *testing.T) {
	t.Parallel()
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()
	nodes := c.Nodes()

	// Force-eval on a nonexistent node fails
	_, _, err := nodes.ForceEvaluate("12345678-abcd-efab-cdef-123456789abc", nil)
	if err == nil || !strings.Contains(err.Error(), "not found") {
		t.Fatalf("expected not found error, got: %#v", err)
	}

	// Wait for node registration and get the ID
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := nodes.List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Try force-eval again. We don't check the WriteMeta because
	// there are no allocations to process, so we would get an index
	// of zero. Same goes for the eval ID.
	_, _, err = nodes.ForceEvaluate(nodeID, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}
}

func TestNodes_Sort(t *testing.T) {
	t.Parallel()
	nodes := []*NodeListStub{
		{CreateIndex: 2},
		{CreateIndex: 1},
		{CreateIndex: 5},
	}
	sort.Sort(NodeIndexSort(nodes))

	expect := []*NodeListStub{
		{CreateIndex: 5},
		{CreateIndex: 2},
		{CreateIndex: 1},
	}
	if !reflect.DeepEqual(nodes, expect) {
		t.Fatalf("\n\n%#v\n\n%#v", nodes, expect)
	}
}

// Unittest monitorDrainMultiplex when an error occurs
func TestNodes_MonitorDrain_Multiplex_Bad(t *testing.T) {
	t.Parallel()
	require := require.New(t)

	ctx := context.Background()
	multiplexCtx, cancel := context.WithCancel(ctx)

	// monitorDrainMultiplex doesn't require anything on *Nodes, so we
	// don't need to use a full Client
	var nodeClient *Nodes

	outCh := make(chan *MonitorMessage, 8)
	nodeCh := make(chan *MonitorMessage, 1)
	allocCh := make(chan *MonitorMessage, 8)
	exitedCh := make(chan struct{})
	go func() {
		defer close(exitedCh)
		nodeClient.monitorDrainMultiplex(ctx, cancel, outCh, nodeCh, allocCh)
	}()

	// Fake an alloc update
	msg := Messagef(0, "alloc update")
	allocCh <- msg
	require.Equal(msg, <-outCh)

	// Fake a node update
	msg = Messagef(0, "node update")
	nodeCh <- msg
	require.Equal(msg, <-outCh)

	// Fake an error that should shut everything down
	msg = Messagef(MonitorMsgLevelError, "fake error")
	nodeCh <- msg
	require.Equal(msg, <-outCh)

	_, ok := <-exitedCh
	require.False(ok)

	_, ok = <-outCh
	require.False(ok)

	// Exiting should also cancel the context that would be passed to the
	// node & alloc watchers
	select {
	case <-multiplexCtx.Done():
	case <-time.After(100 * time.Millisecond):
		t.Fatalf("context wasn't canceled")
	}

}

// Unittest monitorDrainMultiplex when drain finishes
func TestNodes_MonitorDrain_Multiplex_Good(t *testing.T) {
	t.Parallel()
	require := require.New(t)

	ctx := context.Background()
	multiplexCtx, cancel := context.WithCancel(ctx)

	// monitorDrainMultiplex doesn't require anything on *Nodes, so we
	// don't need to use a full Client
	var nodeClient *Nodes

	outCh := make(chan *MonitorMessage, 8)
	nodeCh := make(chan *MonitorMessage, 1)
	allocCh := make(chan *MonitorMessage, 8)
	exitedCh := make(chan struct{})
	go func() {
		defer close(exitedCh)
		nodeClient.monitorDrainMultiplex(ctx, cancel, outCh, nodeCh, allocCh)
	}()

	// Fake a node updating and finishing
	msg := Messagef(MonitorMsgLevelInfo, "node update")
	nodeCh <- msg
	close(nodeCh)
	require.Equal(msg, <-outCh)

	// Nothing else should have exited yet
	select {
	case msg, ok := <-outCh:
		if ok {
			t.Fatalf("unexpected output: %q", msg)
		}
		t.Fatalf("out channel closed unexpectedly")
	case <-exitedCh:
		t.Fatalf("multiplexer exited unexpectedly")
	case <-multiplexCtx.Done():
		t.Fatalf("multiplexer context canceled unexpectedly")
	case <-time.After(10 * time.Millisecond):
		t.Logf("multiplexer still running as expected")
	}

	// Fake an alloc update coming in after the node monitor has finished
	msg = Messagef(0, "alloc update")
	allocCh <- msg
	require.Equal(msg, <-outCh)

	// Closing the allocCh should cause everything to exit
	close(allocCh)

	_, ok := <-exitedCh
	require.False(ok)

	_, ok = <-outCh
	require.False(ok)

	// Exiting should also cancel the context that would be passed to the
	// node & alloc watchers
	select {
	case <-multiplexCtx.Done():
	case <-time.After(100 * time.Millisecond):
		t.Fatalf("context wasn't canceled")
	}

}

func TestNodes_DrainStrategy_Equal(t *testing.T) {
	t.Parallel()
	require := require.New(t)

	// nil
	var d *DrainStrategy
	require.True(d.Equal(nil))

	o := &DrainStrategy{}
	require.False(d.Equal(o))
	require.False(o.Equal(d))

	d = &DrainStrategy{}
	require.True(d.Equal(o))

	// ForceDeadline
	d.ForceDeadline = time.Now()
	require.False(d.Equal(o))

	o.ForceDeadline = d.ForceDeadline
	require.True(d.Equal(o))

	// Deadline
	d.Deadline = 1
	require.False(d.Equal(o))

	o.Deadline = 1
	require.True(d.Equal(o))

	// IgnoreSystemJobs
	d.IgnoreSystemJobs = true
	require.False(d.Equal(o))

	o.IgnoreSystemJobs = true
	require.True(d.Equal(o))
}

func TestNodes_Purge(t *testing.T) {
	t.Parallel()
	require := require.New(t)
	c, s := makeClient(t, nil, func(c *testutil.TestServerConfig) {
		c.DevMode = true
	})
	defer s.Stop()

	// Purge on a nonexistent node fails.
	_, _, err := c.Nodes().Purge("12345678-abcd-efab-cdef-123456789abc", nil)
	if err == nil || !strings.Contains(err.Error(), "not found") {
		t.Fatalf("expected not found error, got: %#v", err)
	}

	// Wait for node registration and get the ID so we can attempt to purge a
	// node that exists.
	var nodeID string
	testutil.WaitForResult(func() (bool, error) {
		out, _, err := c.Nodes().List(nil)
		if err != nil {
			return false, err
		}
		if n := len(out); n != 1 {
			return false, fmt.Errorf("expected 1 node, got: %d", n)
		}
		nodeID = out[0].ID
		return true, nil
	}, func(err error) {
		t.Fatalf("err: %s", err)
	})

	// Perform the node purge and check the response objects.
	out, meta, err := c.Nodes().Purge(nodeID, nil)
	require.Nil(err)
	require.NotNil(out)

	// We can't use assertQueryMeta here, as the RPC response does not populate
	// the known leader field.
	require.Greater(meta.LastIndex, uint64(0))
}

func TestNodeStatValueFormatting(t *testing.T) {
	t.Parallel()

	cases := []struct {
		expected string
		value    StatValue
	}{
		{
			"true",
			StatValue{BoolVal: boolToPtr(true)},
		},
		{
			"false",
			StatValue{BoolVal: boolToPtr(false)},
		},
		{
			"myvalue",
			StatValue{StringVal: stringToPtr("myvalue")},
		},
		{
			"2.718",
			StatValue{
				FloatNumeratorVal: float64ToPtr(2.718),
			},
		},
		{
			"2.718 / 3.14",
			StatValue{
				FloatNumeratorVal:   float64ToPtr(2.718),
				FloatDenominatorVal: float64ToPtr(3.14),
			},
		},
		{
			"2.718 MHz",
			StatValue{
				FloatNumeratorVal: float64ToPtr(2.718),
				Unit:              "MHz",
			},
		},
		{
			"2.718 / 3.14 MHz",
			StatValue{
				FloatNumeratorVal:   float64ToPtr(2.718),
				FloatDenominatorVal: float64ToPtr(3.14),
				Unit:                "MHz",
			},
		},
		{
			"2",
			StatValue{
				IntNumeratorVal: int64ToPtr(2),
			},
		},
		{
			"2 / 3",
			StatValue{
				IntNumeratorVal:   int64ToPtr(2),
				IntDenominatorVal: int64ToPtr(3),
			},
		},
		{
			"2 MHz",
			StatValue{
				IntNumeratorVal: int64ToPtr(2),
				Unit:            "MHz",
			},
		},
		{
			"2 / 3 MHz",
			StatValue{
				IntNumeratorVal:   int64ToPtr(2),
				IntDenominatorVal: int64ToPtr(3),
				Unit:              "MHz",
			},
		},
	}

	for i, c := range cases {
		t.Run(fmt.Sprintf("case %d %v", i, c.expected), func(t *testing.T) {
			formatted := c.value.String()
			require.Equal(t, c.expected, formatted)
		})
	}
}