Merge pull request #5038 from hashicorp/b-drivermanager-tests

WIP: fix failing tests caused by async driver manager

api/fs_test.go

@@ -46,6 +46,9 @@ func TestFS_Logs(t *testing.T) {
 		if nodes[0].Status != "ready" {
 			return false, fmt.Errorf("node not ready: %s", nodes[0].Status)
 		}
+		if _, ok := nodes[0].Drivers["mock_driver"]; !ok {
+			return false, fmt.Errorf("mock_driver not ready")
+		}
 		return true, nil
 	}, func(err error) {
 		t.Fatalf("err: %v", err)

client/client.go

@@ -237,6 +237,10 @@ type Client struct {
 
 	// batchNodeUpdates is used to batch initial updates to the node
 	batchNodeUpdates *batchNodeUpdates
+
+	// fpInitialized chan is closed when the first batch of fingerprints are
+	// applied to the node and the server is updated
+	fpInitialized chan struct{}
 }
 
 var (
@@ -281,6 +285,7 @@ func NewClient(cfg *config.Config, consulCatalog consul.CatalogAPI, consulServic
 		triggerDiscoveryCh:   make(chan struct{}),
 		triggerNodeUpdate:    make(chan struct{}, 8),
 		triggerEmitNodeEvent: make(chan *structs.NodeEvent, 8),
+		fpInitialized:        make(chan struct{}),
 	}
 
 	c.batchNodeUpdates = newBatchNodeUpdates(
@@ -436,6 +441,11 @@ func NewClient(cfg *config.Config, consulCatalog consul.CatalogAPI, consulServic
 	return c, nil
 }
 
+// Ready returns a chan that is closed when the client is fully initialized
+func (c *Client) Ready() <-chan struct{} {
+	return c.fpInitialized
+}
+
 // init is used to initialize the client and perform any setup
 // needed before we begin starting its various components.
 func (c *Client) init() error {

client/node_updater.go

@@ -18,7 +18,8 @@ var (
 )
 
 // batchFirstFingerprints waits for the first fingerprint response from all
-// plugin managers and sends a single Node update for all fingerprints
+// plugin managers and sends a single Node update for all fingerprints. It
+// should only ever be called once
 func (c *Client) batchFirstFingerprints() {
 	ctx, cancel := context.WithTimeout(context.Background(), batchFirstFingerprintsTimeout)
 	defer cancel()
@@ -63,6 +64,8 @@ SEND_BATCH:
 	if driverChanged || devicesChanged {
 		c.updateNodeLocked()
 	}
+
+	close(c.fpInitialized)
 }
 
 // updateNodeFromDriver receives a DriverInfo struct for the driver and updates

command/alloc_status_test.go

@@ -96,7 +96,8 @@ func TestAllocStatusCommand_Run(t *testing.T) {
 			return false, err
 		}
 		for _, node := range nodes {
-			if node.Status == structs.NodeStatusReady {
+			if _, ok := node.Drivers["mock_driver"]; ok &&
+				node.Status == structs.NodeStatusReady {
 				return true, nil
 			}
 		}

command/job_status_test.go

@@ -26,6 +26,21 @@ func TestJobStatusCommand_Run(t *testing.T) {
 	t.Parallel()
 	srv, client, url := testServer(t, true, nil)
 	defer srv.Shutdown()
+	testutil.WaitForResult(func() (bool, error) {
+		nodes, _, err := client.Nodes().List(nil)
+		if err != nil {
+			return false, err
+		}
+		if len(nodes) == 0 {
+			return false, fmt.Errorf("missing node")
+		}
+		if _, ok := nodes[0].Drivers["mock_driver"]; !ok {
+			return false, fmt.Errorf("mock_driver not ready")
+		}
+		return true, nil
+	}, func(err error) {
+		t.Fatalf("err: %s", err)
+	})
 
 	ui := new(cli.MockUi)
 	cmd := &JobStatusCommand{Meta: Meta{Ui: ui}}
@@ -247,6 +262,24 @@ func TestJobStatusCommand_WithAccessPolicy(t *testing.T) {
 	token := srv.RootToken
 	assert.NotNil(token, "failed to bootstrap ACL token")
 
+	// Wait for client ready
+	client.SetSecretID(token.SecretID)
+	testutil.WaitForResult(func() (bool, error) {
+		nodes, _, err := client.Nodes().List(nil)
+		if err != nil {
+			return false, err
+		}
+		if len(nodes) == 0 {
+			return false, fmt.Errorf("missing node")
+		}
+		if _, ok := nodes[0].Drivers["mock_driver"]; !ok {
+			return false, fmt.Errorf("mock_driver not ready")
+		}
+		return true, nil
+	}, func(err error) {
+		t.Fatalf("err: %s", err)
+	})
+
 	// Register a job
 	j := testJob("job1_sfx")
 

command/node_drain_test.go

@@ -113,6 +113,9 @@ func TestNodeDrainCommand_Monitor(t *testing.T) {
 		if len(nodes) == 0 {
 			return false, fmt.Errorf("missing node")
 		}
+		if _, ok := nodes[0].Drivers["mock_driver"]; !ok {
+			return false, fmt.Errorf("mock_driver not ready")
+		}
 		nodeID = nodes[0].ID
 		return true, nil
 	}, func(err error) {

nomad/client_stats_endpoint_test.go

@@ -2,6 +2,7 @@ package nomad
 
 import (
 	"testing"
+	"time"
 
 	msgpackrpc "github.com/hashicorp/net-rpc-msgpackrpc"
 	"github.com/hashicorp/nomad/acl"
@@ -188,6 +189,13 @@ func TestClientStats_Stats_Remote(t *testing.T) {
 	})
 	defer cleanup()
 
+	// Wait for client initialization
+	select {
+	case <-c.Ready():
+	case <-time.After(10 * time.Second):
+		require.Fail("client timedout on initialize")
+	}
+
 	testutil.WaitForResult(func() (bool, error) {
 		nodes := s2.connectedNodes()
 		return len(nodes) == 1, nil