1311 lines
35 KiB
Go
1311 lines
35 KiB
Go
package client
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"runtime"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hashicorp/go-msgpack/codec"
|
|
"github.com/hashicorp/nomad/acl"
|
|
"github.com/hashicorp/nomad/client/config"
|
|
cstructs "github.com/hashicorp/nomad/client/structs"
|
|
"github.com/hashicorp/nomad/helper/pluginutils/catalog"
|
|
"github.com/hashicorp/nomad/helper/uuid"
|
|
"github.com/hashicorp/nomad/nomad"
|
|
"github.com/hashicorp/nomad/nomad/mock"
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
nstructs "github.com/hashicorp/nomad/nomad/structs"
|
|
nconfig "github.com/hashicorp/nomad/nomad/structs/config"
|
|
"github.com/hashicorp/nomad/plugins/drivers"
|
|
"github.com/hashicorp/nomad/testutil"
|
|
"github.com/stretchr/testify/require"
|
|
"golang.org/x/sys/unix"
|
|
)
|
|
|
|
func TestAllocations_Restart(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
client, cleanup := TestClient(t, nil)
|
|
defer cleanup()
|
|
|
|
a := mock.Alloc()
|
|
a.Job.TaskGroups[0].Tasks[0].Driver = "mock_driver"
|
|
a.Job.TaskGroups[0].RestartPolicy = &nstructs.RestartPolicy{
|
|
Attempts: 0,
|
|
Mode: nstructs.RestartPolicyModeFail,
|
|
}
|
|
a.Job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "10s",
|
|
}
|
|
require.Nil(client.addAlloc(a, ""))
|
|
|
|
// Try with bad alloc
|
|
req := &nstructs.AllocRestartRequest{}
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp)
|
|
require.Error(err)
|
|
|
|
// Try with good alloc
|
|
req.AllocID = a.ID
|
|
|
|
testutil.WaitForResult(func() (bool, error) {
|
|
var resp2 nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp2)
|
|
if err != nil && strings.Contains(err.Error(), "not running") {
|
|
return false, err
|
|
}
|
|
|
|
return true, nil
|
|
}, func(err error) {
|
|
t.Fatalf("err: %v", err)
|
|
})
|
|
}
|
|
|
|
func TestAllocations_Restart_ACL(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
server, addr, root, cleanupS := testACLServer(t, nil)
|
|
defer cleanupS()
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{addr}
|
|
c.ACLEnabled = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
|
|
|
|
// Try request without a token and expect failure
|
|
{
|
|
req := &nstructs.AllocRestartRequest{}
|
|
req.AllocID = alloc.ID
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp)
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with an invalid token and expect failure
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{}))
|
|
req := &nstructs.AllocRestartRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp)
|
|
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with a valid token
|
|
{
|
|
policyHCL := mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilityAllocLifecycle})
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1007, "valid", policyHCL)
|
|
require.NotNil(token)
|
|
req := &nstructs.AllocRestartRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
req.Namespace = nstructs.DefaultNamespace
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp)
|
|
require.NoError(err)
|
|
//require.True(nstructs.IsErrUnknownAllocation(err), "Expected unknown alloc, found: %v", err)
|
|
}
|
|
|
|
// Try request with a management token
|
|
{
|
|
req := &nstructs.AllocRestartRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = root.SecretID
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Restart", &req, &resp)
|
|
// Depending on how quickly the alloc restarts there may be no
|
|
// error *or* a task not running error; either is fine.
|
|
if err != nil {
|
|
require.Contains(err.Error(), "Task not running", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAllocations_GarbageCollectAll(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
client, cleanup := TestClient(t, nil)
|
|
defer cleanup()
|
|
|
|
req := &nstructs.NodeSpecificRequest{}
|
|
var resp nstructs.GenericResponse
|
|
require.Nil(client.ClientRPC("Allocations.GarbageCollectAll", &req, &resp))
|
|
}
|
|
|
|
func TestAllocations_GarbageCollectAll_ACL(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
server, addr, root, cleanupS := testACLServer(t, nil)
|
|
defer cleanupS()
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{addr}
|
|
c.ACLEnabled = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
// Try request without a token and expect failure
|
|
{
|
|
req := &nstructs.NodeSpecificRequest{}
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollectAll", &req, &resp)
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with an invalid token and expect failure
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
|
|
req := &nstructs.NodeSpecificRequest{}
|
|
req.AuthToken = token.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollectAll", &req, &resp)
|
|
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with a valid token
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1007, "valid", mock.NodePolicy(acl.PolicyWrite))
|
|
req := &nstructs.NodeSpecificRequest{}
|
|
req.AuthToken = token.SecretID
|
|
var resp nstructs.GenericResponse
|
|
require.Nil(client.ClientRPC("Allocations.GarbageCollectAll", &req, &resp))
|
|
}
|
|
|
|
// Try request with a management token
|
|
{
|
|
req := &nstructs.NodeSpecificRequest{}
|
|
req.AuthToken = root.SecretID
|
|
var resp nstructs.GenericResponse
|
|
require.Nil(client.ClientRPC("Allocations.GarbageCollectAll", &req, &resp))
|
|
}
|
|
}
|
|
|
|
func TestAllocations_GarbageCollect(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
client, cleanup := TestClient(t, func(c *config.Config) {
|
|
c.GCDiskUsageThreshold = 100.0
|
|
})
|
|
defer cleanup()
|
|
|
|
a := mock.Alloc()
|
|
a.Job.TaskGroups[0].Tasks[0].Driver = "mock_driver"
|
|
rp := &nstructs.RestartPolicy{
|
|
Attempts: 0,
|
|
Mode: nstructs.RestartPolicyModeFail,
|
|
}
|
|
a.Job.TaskGroups[0].RestartPolicy = rp
|
|
a.Job.TaskGroups[0].Tasks[0].RestartPolicy = rp
|
|
a.Job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "10ms",
|
|
}
|
|
require.Nil(client.addAlloc(a, ""))
|
|
|
|
// Try with bad alloc
|
|
req := &nstructs.AllocSpecificRequest{}
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
|
|
require.NotNil(err)
|
|
|
|
// Try with good alloc
|
|
req.AllocID = a.ID
|
|
testutil.WaitForResult(func() (bool, error) {
|
|
// Check if has been removed first
|
|
if ar, ok := client.allocs[a.ID]; !ok || ar.IsDestroyed() {
|
|
return true, nil
|
|
}
|
|
|
|
var resp2 nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp2)
|
|
return err == nil, err
|
|
}, func(err error) {
|
|
t.Fatalf("err: %v", err)
|
|
})
|
|
}
|
|
|
|
func TestAllocations_GarbageCollect_ACL(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
server, addr, root, cleanupS := testACLServer(t, nil)
|
|
defer cleanupS()
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{addr}
|
|
c.ACLEnabled = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
}
|
|
|
|
noSuchAllocErr := fmt.Errorf("No such allocation on client or allocation not eligible for GC")
|
|
|
|
// Wait for client to be running job
|
|
alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
|
|
|
|
// Try request without a token and expect failure
|
|
{
|
|
req := &nstructs.AllocSpecificRequest{}
|
|
req.AllocID = alloc.ID
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with an invalid token and expect failure
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
|
|
req := &nstructs.AllocSpecificRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
|
|
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with a valid token
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
|
|
mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilitySubmitJob}))
|
|
req := &nstructs.AllocSpecificRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
req.Namespace = nstructs.DefaultNamespace
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
|
|
require.Error(err, noSuchAllocErr)
|
|
}
|
|
|
|
// Try request with a management token
|
|
{
|
|
req := &nstructs.AllocSpecificRequest{}
|
|
req.AuthToken = root.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
|
|
require.Error(err, noSuchAllocErr)
|
|
}
|
|
}
|
|
|
|
func TestAllocations_Signal(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
client, cleanup := TestClient(t, nil)
|
|
defer cleanup()
|
|
|
|
a := mock.Alloc()
|
|
require.Nil(t, client.addAlloc(a, ""))
|
|
|
|
// Try with bad alloc
|
|
req := &nstructs.AllocSignalRequest{}
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Signal", &req, &resp)
|
|
require.NotNil(t, err)
|
|
require.True(t, nstructs.IsErrUnknownAllocation(err))
|
|
|
|
// Try with good alloc
|
|
req.AllocID = a.ID
|
|
|
|
var resp2 nstructs.GenericResponse
|
|
err = client.ClientRPC("Allocations.Signal", &req, &resp2)
|
|
|
|
require.Error(t, err, "Expected error, got: %s, resp: %#+v", err, resp2)
|
|
require.Contains(t, err.Error(), "Failed to signal task: web, err: Task not running")
|
|
}
|
|
|
|
func TestAllocations_Signal_ACL(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
server, addr, root, cleanupS := testACLServer(t, nil)
|
|
defer cleanupS()
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{addr}
|
|
c.ACLEnabled = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
|
|
|
|
// Try request without a token and expect failure
|
|
{
|
|
req := &nstructs.AllocSignalRequest{}
|
|
req.AllocID = alloc.ID
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Signal", &req, &resp)
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with an invalid token and expect failure
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
|
|
req := &nstructs.AllocSignalRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Signal", &req, &resp)
|
|
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with a valid token
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
|
|
mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilityAllocLifecycle}))
|
|
req := &nstructs.AllocSignalRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
req.Namespace = nstructs.DefaultNamespace
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Signal", &req, &resp)
|
|
require.NoError(err)
|
|
}
|
|
|
|
// Try request with a management token
|
|
{
|
|
req := &nstructs.AllocSignalRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = root.SecretID
|
|
|
|
var resp nstructs.GenericResponse
|
|
err := client.ClientRPC("Allocations.Signal", &req, &resp)
|
|
require.NoError(err)
|
|
}
|
|
}
|
|
|
|
func TestAllocations_Stats(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
client, cleanup := TestClient(t, nil)
|
|
defer cleanup()
|
|
|
|
a := mock.Alloc()
|
|
require.Nil(client.addAlloc(a, ""))
|
|
|
|
// Try with bad alloc
|
|
req := &cstructs.AllocStatsRequest{}
|
|
var resp cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp)
|
|
require.NotNil(err)
|
|
|
|
// Try with good alloc
|
|
req.AllocID = a.ID
|
|
testutil.WaitForResult(func() (bool, error) {
|
|
var resp2 cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp2)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if resp2.Stats == nil {
|
|
return false, fmt.Errorf("invalid stats object")
|
|
}
|
|
|
|
return true, nil
|
|
}, func(err error) {
|
|
t.Fatalf("err: %v", err)
|
|
})
|
|
}
|
|
|
|
func TestAllocations_Stats_ACL(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
server, addr, root, cleanupS := testACLServer(t, nil)
|
|
defer cleanupS()
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{addr}
|
|
c.ACLEnabled = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
|
|
|
|
// Try request without a token and expect failure
|
|
{
|
|
req := &cstructs.AllocStatsRequest{}
|
|
req.AllocID = alloc.ID
|
|
var resp cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp)
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with an invalid token and expect failure
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
|
|
req := &cstructs.AllocStatsRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
|
|
var resp cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp)
|
|
|
|
require.NotNil(err)
|
|
require.EqualError(err, nstructs.ErrPermissionDenied.Error())
|
|
}
|
|
|
|
// Try request with a valid token
|
|
{
|
|
token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
|
|
mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilityReadJob}))
|
|
req := &cstructs.AllocStatsRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = token.SecretID
|
|
req.Namespace = nstructs.DefaultNamespace
|
|
|
|
var resp cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp)
|
|
require.NoError(err)
|
|
}
|
|
|
|
// Try request with a management token
|
|
{
|
|
req := &cstructs.AllocStatsRequest{}
|
|
req.AllocID = alloc.ID
|
|
req.AuthToken = root.SecretID
|
|
|
|
var resp cstructs.AllocStatsResponse
|
|
err := client.ClientRPC("Allocations.Stats", &req, &resp)
|
|
require.NoError(err)
|
|
}
|
|
}
|
|
|
|
func TestAlloc_ExecStreaming(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
// Start a server and client
|
|
s, cleanupS := nomad.TestServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
c, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
})
|
|
defer cleanupC()
|
|
|
|
expectedStdout := "Hello from the other side\n"
|
|
expectedStderr := "Hello from the other side\n"
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
"exec_command": map[string]interface{}{
|
|
"run_for": "1ms",
|
|
"stdout_string": expectedStdout,
|
|
"stderr_string": expectedStderr,
|
|
"exit_code": 3,
|
|
},
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
testutil.WaitForRunning(t, s.RPC, job)
|
|
|
|
// Get the allocation ID
|
|
args := nstructs.AllocListRequest{}
|
|
args.Region = "global"
|
|
resp := nstructs.AllocListResponse{}
|
|
require.NoError(s.RPC("Alloc.List", &args, &resp))
|
|
require.Len(resp.Allocations, 1)
|
|
allocID := resp.Allocations[0].ID
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: allocID,
|
|
Task: job.TaskGroups[0].Tasks[0].Name,
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{Region: "global"},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := c.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(encoder.Encode(req))
|
|
|
|
timeout := time.After(3 * time.Second)
|
|
|
|
exitCode := -1
|
|
receivedStdout := ""
|
|
receivedStderr := ""
|
|
|
|
OUTER:
|
|
for {
|
|
select {
|
|
case <-timeout:
|
|
// time out report
|
|
require.Equal(expectedStdout, receivedStderr, "didn't receive expected stdout")
|
|
require.Equal(expectedStderr, receivedStderr, "didn't receive expected stderr")
|
|
require.Equal(3, exitCode, "failed to get exit code")
|
|
require.FailNow("timed out")
|
|
case err := <-errCh:
|
|
require.NoError(err)
|
|
case f := <-frames:
|
|
switch {
|
|
case f.Stdout != nil && len(f.Stdout.Data) != 0:
|
|
receivedStdout += string(f.Stdout.Data)
|
|
case f.Stderr != nil && len(f.Stderr.Data) != 0:
|
|
receivedStderr += string(f.Stderr.Data)
|
|
case f.Exited && f.Result != nil:
|
|
exitCode = int(f.Result.ExitCode)
|
|
default:
|
|
t.Logf("received unrelevant frame: %v", f)
|
|
}
|
|
|
|
if expectedStdout == receivedStdout && expectedStderr == receivedStderr && exitCode == 3 {
|
|
break OUTER
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAlloc_ExecStreaming_NoAllocation(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
// Start a server and client
|
|
s, cleanupS := nomad.TestServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
c, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
})
|
|
defer cleanupC()
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: uuid.Generate(),
|
|
Task: "testtask",
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{Region: "global"},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := c.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(encoder.Encode(req))
|
|
|
|
timeout := time.After(3 * time.Second)
|
|
|
|
select {
|
|
case <-timeout:
|
|
require.FailNow("timed out")
|
|
case err := <-errCh:
|
|
require.True(nstructs.IsErrUnknownAllocation(err), "expected no allocation error but found: %v", err)
|
|
case f := <-frames:
|
|
require.Fail("received unexpected frame", "frame: %#v", f)
|
|
}
|
|
}
|
|
|
|
func TestAlloc_ExecStreaming_DisableRemoteExec(t *testing.T) {
|
|
t.Parallel()
|
|
require := require.New(t)
|
|
|
|
// Start a server and client
|
|
s, cleanupS := nomad.TestServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
c, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
c.DisableRemoteExec = true
|
|
})
|
|
defer cleanupC()
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: uuid.Generate(),
|
|
Task: "testtask",
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{Region: "global"},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := c.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(encoder.Encode(req))
|
|
|
|
timeout := time.After(3 * time.Second)
|
|
|
|
select {
|
|
case <-timeout:
|
|
require.FailNow("timed out")
|
|
case err := <-errCh:
|
|
require.True(nstructs.IsErrPermissionDenied(err), "expected permission denied error but found: %v", err)
|
|
case f := <-frames:
|
|
require.Fail("received unexpected frame", "frame: %#v", f)
|
|
}
|
|
}
|
|
|
|
func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
// Start a server and client
|
|
s, root, cleanupS := nomad.TestACLServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.ACLEnabled = true
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
})
|
|
defer cleanupC()
|
|
|
|
// Create a bad token
|
|
policyBad := mock.NamespacePolicy("other", "", []string{acl.NamespaceCapabilityDeny})
|
|
tokenBad := mock.CreatePolicyAndToken(t, s.State(), 1005, "invalid", policyBad)
|
|
|
|
policyGood := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec, acl.NamespaceCapabilityReadFS})
|
|
tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
|
|
|
|
cases := []struct {
|
|
Name string
|
|
Token string
|
|
ExpectedError string
|
|
}{
|
|
{
|
|
Name: "bad token",
|
|
Token: tokenBad.SecretID,
|
|
ExpectedError: nstructs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "good token",
|
|
Token: tokenGood.SecretID,
|
|
ExpectedError: "task not found",
|
|
},
|
|
{
|
|
Name: "root token",
|
|
Token: root.SecretID,
|
|
ExpectedError: "task not found",
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: alloc.ID,
|
|
Task: "testtask",
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{
|
|
Region: "global",
|
|
AuthToken: c.Token,
|
|
Namespace: nstructs.DefaultNamespace,
|
|
},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := client.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(t, err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(t, encoder.Encode(req))
|
|
|
|
select {
|
|
case <-time.After(3 * time.Second):
|
|
require.FailNow(t, "timed out")
|
|
case err := <-errCh:
|
|
require.Contains(t, err.Error(), c.ExpectedError)
|
|
case f := <-frames:
|
|
require.Fail(t, "received unexpected frame", "frame: %#v", f)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestAlloc_ExecStreaming_ACL_WithIsolation_Image asserts that token only needs
|
|
// alloc-exec acl policy when image isolation is used
|
|
func TestAlloc_ExecStreaming_ACL_WithIsolation_Image(t *testing.T) {
|
|
t.Parallel()
|
|
isolation := drivers.FSIsolationImage
|
|
|
|
// Start a server and client
|
|
s, root, cleanupS := nomad.TestACLServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
client, cleanupC := TestClient(t, func(c *config.Config) {
|
|
c.ACLEnabled = true
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
|
|
pluginConfig := []*nconfig.PluginConfig{
|
|
{
|
|
Name: "mock_driver",
|
|
Config: map[string]interface{}{
|
|
"fs_isolation": string(isolation),
|
|
},
|
|
},
|
|
}
|
|
|
|
c.PluginLoader = catalog.TestPluginLoaderWithOptions(t, "", map[string]string{}, pluginConfig)
|
|
})
|
|
defer cleanupC()
|
|
|
|
// Create a bad token
|
|
policyBad := mock.NamespacePolicy("other", "", []string{acl.NamespaceCapabilityDeny})
|
|
tokenBad := mock.CreatePolicyAndToken(t, s.State(), 1005, "invalid", policyBad)
|
|
|
|
policyAllocExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec})
|
|
tokenAllocExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyAllocExec)
|
|
|
|
policyAllocNodeExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec, acl.NamespaceCapabilityAllocNodeExec})
|
|
tokenAllocNodeExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyAllocNodeExec)
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
"exec_command": map[string]interface{}{
|
|
"run_for": "1ms",
|
|
"stdout_string": "some output",
|
|
},
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)
|
|
|
|
// Get the allocation ID
|
|
args := nstructs.AllocListRequest{}
|
|
args.Region = "global"
|
|
args.AuthToken = root.SecretID
|
|
args.Namespace = nstructs.DefaultNamespace
|
|
resp := nstructs.AllocListResponse{}
|
|
require.NoError(t, s.RPC("Alloc.List", &args, &resp))
|
|
require.Len(t, resp.Allocations, 1)
|
|
allocID := resp.Allocations[0].ID
|
|
|
|
cases := []struct {
|
|
Name string
|
|
Token string
|
|
ExpectedError string
|
|
}{
|
|
{
|
|
Name: "bad token",
|
|
Token: tokenBad.SecretID,
|
|
ExpectedError: nstructs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "alloc-exec token",
|
|
Token: tokenAllocExec.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
{
|
|
Name: "alloc-node-exec token",
|
|
Token: tokenAllocNodeExec.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
{
|
|
Name: "root token",
|
|
Token: root.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: allocID,
|
|
Task: job.TaskGroups[0].Tasks[0].Name,
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{
|
|
Region: "global",
|
|
AuthToken: c.Token,
|
|
Namespace: nstructs.DefaultNamespace,
|
|
},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := client.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(t, err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(t, encoder.Encode(req))
|
|
|
|
select {
|
|
case <-time.After(3 * time.Second):
|
|
case err := <-errCh:
|
|
if c.ExpectedError == "" {
|
|
require.NoError(t, err)
|
|
} else {
|
|
require.Contains(t, err.Error(), c.ExpectedError)
|
|
}
|
|
case f := <-frames:
|
|
// we are good if we don't expect an error
|
|
if c.ExpectedError != "" {
|
|
require.Fail(t, "unexpected frame", "frame: %#v", f)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestAlloc_ExecStreaming_ACL_WithIsolation_Chroot asserts that token only needs
|
|
// alloc-exec acl policy when chroot isolation is used
|
|
func TestAlloc_ExecStreaming_ACL_WithIsolation_Chroot(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
if runtime.GOOS != "linux" || unix.Geteuid() != 0 {
|
|
t.Skip("chroot isolation requires linux root")
|
|
}
|
|
|
|
isolation := drivers.FSIsolationChroot
|
|
|
|
// Start a server and client
|
|
s, root, cleanupS := nomad.TestACLServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
client, cleanup := TestClient(t, func(c *config.Config) {
|
|
c.ACLEnabled = true
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
|
|
pluginConfig := []*nconfig.PluginConfig{
|
|
{
|
|
Name: "mock_driver",
|
|
Config: map[string]interface{}{
|
|
"fs_isolation": string(isolation),
|
|
},
|
|
},
|
|
}
|
|
|
|
c.PluginLoader = catalog.TestPluginLoaderWithOptions(t, "", map[string]string{}, pluginConfig)
|
|
})
|
|
defer cleanup()
|
|
|
|
// Create a bad token
|
|
policyBad := mock.NamespacePolicy("other", "", []string{acl.NamespaceCapabilityDeny})
|
|
tokenBad := mock.CreatePolicyAndToken(t, s.State(), 1005, "invalid", policyBad)
|
|
|
|
policyAllocExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec})
|
|
tokenAllocExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "alloc-exec", policyAllocExec)
|
|
|
|
policyAllocNodeExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec, acl.NamespaceCapabilityAllocNodeExec})
|
|
tokenAllocNodeExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "alloc-node-exec", policyAllocNodeExec)
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
"exec_command": map[string]interface{}{
|
|
"run_for": "1ms",
|
|
"stdout_string": "some output",
|
|
},
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)
|
|
|
|
// Get the allocation ID
|
|
args := nstructs.AllocListRequest{}
|
|
args.Region = "global"
|
|
args.AuthToken = root.SecretID
|
|
args.Namespace = nstructs.DefaultNamespace
|
|
resp := nstructs.AllocListResponse{}
|
|
require.NoError(t, s.RPC("Alloc.List", &args, &resp))
|
|
require.Len(t, resp.Allocations, 1)
|
|
allocID := resp.Allocations[0].ID
|
|
|
|
cases := []struct {
|
|
Name string
|
|
Token string
|
|
ExpectedError string
|
|
}{
|
|
{
|
|
Name: "bad token",
|
|
Token: tokenBad.SecretID,
|
|
ExpectedError: nstructs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "alloc-exec token",
|
|
Token: tokenAllocExec.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
{
|
|
Name: "alloc-node-exec token",
|
|
Token: tokenAllocNodeExec.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
{
|
|
Name: "root token",
|
|
Token: root.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: allocID,
|
|
Task: job.TaskGroups[0].Tasks[0].Name,
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{
|
|
Region: "global",
|
|
AuthToken: c.Token,
|
|
Namespace: nstructs.DefaultNamespace,
|
|
},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := client.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(t, err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(t, encoder.Encode(req))
|
|
|
|
select {
|
|
case <-time.After(3 * time.Second):
|
|
case err := <-errCh:
|
|
if c.ExpectedError == "" {
|
|
require.NoError(t, err)
|
|
} else {
|
|
require.Contains(t, err.Error(), c.ExpectedError)
|
|
}
|
|
case f := <-frames:
|
|
// we are good if we don't expect an error
|
|
if c.ExpectedError != "" {
|
|
require.Fail(t, "unexpected frame", "frame: %#v", f)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestAlloc_ExecStreaming_ACL_WithIsolation_None asserts that token needs
|
|
// alloc-node-exec acl policy as well when no isolation is used
|
|
func TestAlloc_ExecStreaming_ACL_WithIsolation_None(t *testing.T) {
|
|
t.Parallel()
|
|
isolation := drivers.FSIsolationNone
|
|
|
|
// Start a server and client
|
|
s, root, cleanupS := nomad.TestACLServer(t, nil)
|
|
defer cleanupS()
|
|
testutil.WaitForLeader(t, s.RPC)
|
|
|
|
client, cleanup := TestClient(t, func(c *config.Config) {
|
|
c.ACLEnabled = true
|
|
c.Servers = []string{s.GetConfig().RPCAddr.String()}
|
|
|
|
pluginConfig := []*nconfig.PluginConfig{
|
|
{
|
|
Name: "mock_driver",
|
|
Config: map[string]interface{}{
|
|
"fs_isolation": string(isolation),
|
|
},
|
|
},
|
|
}
|
|
|
|
c.PluginLoader = catalog.TestPluginLoaderWithOptions(t, "", map[string]string{}, pluginConfig)
|
|
})
|
|
defer cleanup()
|
|
|
|
// Create a bad token
|
|
policyBad := mock.NamespacePolicy("other", "", []string{acl.NamespaceCapabilityDeny})
|
|
tokenBad := mock.CreatePolicyAndToken(t, s.State(), 1005, "invalid", policyBad)
|
|
|
|
policyAllocExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec})
|
|
tokenAllocExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "alloc-exec", policyAllocExec)
|
|
|
|
policyAllocNodeExec := mock.NamespacePolicy(structs.DefaultNamespace, "",
|
|
[]string{acl.NamespaceCapabilityAllocExec, acl.NamespaceCapabilityAllocNodeExec})
|
|
tokenAllocNodeExec := mock.CreatePolicyAndToken(t, s.State(), 1009, "alloc-node-exec", policyAllocNodeExec)
|
|
|
|
job := mock.BatchJob()
|
|
job.TaskGroups[0].Count = 1
|
|
job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
|
|
"run_for": "20s",
|
|
"exec_command": map[string]interface{}{
|
|
"run_for": "1ms",
|
|
"stdout_string": "some output",
|
|
},
|
|
}
|
|
|
|
// Wait for client to be running job
|
|
testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)
|
|
|
|
// Get the allocation ID
|
|
args := nstructs.AllocListRequest{}
|
|
args.Region = "global"
|
|
args.AuthToken = root.SecretID
|
|
args.Namespace = nstructs.DefaultNamespace
|
|
resp := nstructs.AllocListResponse{}
|
|
require.NoError(t, s.RPC("Alloc.List", &args, &resp))
|
|
require.Len(t, resp.Allocations, 1)
|
|
allocID := resp.Allocations[0].ID
|
|
|
|
cases := []struct {
|
|
Name string
|
|
Token string
|
|
ExpectedError string
|
|
}{
|
|
{
|
|
Name: "bad token",
|
|
Token: tokenBad.SecretID,
|
|
ExpectedError: nstructs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "alloc-exec token",
|
|
Token: tokenAllocExec.SecretID,
|
|
ExpectedError: nstructs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "alloc-node-exec token",
|
|
Token: tokenAllocNodeExec.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
{
|
|
Name: "root token",
|
|
Token: root.SecretID,
|
|
ExpectedError: "",
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
|
|
// Make the request
|
|
req := &cstructs.AllocExecRequest{
|
|
AllocID: allocID,
|
|
Task: job.TaskGroups[0].Tasks[0].Name,
|
|
Tty: true,
|
|
Cmd: []string{"placeholder command"},
|
|
QueryOptions: nstructs.QueryOptions{
|
|
Region: "global",
|
|
AuthToken: c.Token,
|
|
Namespace: nstructs.DefaultNamespace,
|
|
},
|
|
}
|
|
|
|
// Get the handler
|
|
handler, err := client.StreamingRpcHandler("Allocations.Exec")
|
|
require.Nil(t, err)
|
|
|
|
// Create a pipe
|
|
p1, p2 := net.Pipe()
|
|
defer p1.Close()
|
|
defer p2.Close()
|
|
|
|
errCh := make(chan error)
|
|
frames := make(chan *drivers.ExecTaskStreamingResponseMsg)
|
|
|
|
// Start the handler
|
|
go handler(p2)
|
|
go decodeFrames(t, p1, frames, errCh)
|
|
|
|
// Send the request
|
|
encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
|
|
require.Nil(t, encoder.Encode(req))
|
|
|
|
select {
|
|
case <-time.After(3 * time.Second):
|
|
case err := <-errCh:
|
|
if c.ExpectedError == "" {
|
|
require.NoError(t, err)
|
|
} else {
|
|
require.Contains(t, err.Error(), c.ExpectedError)
|
|
}
|
|
case f := <-frames:
|
|
// we are good if we don't expect an error
|
|
if c.ExpectedError != "" {
|
|
require.Fail(t, "unexpected frame", "frame: %#v", f)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func decodeFrames(t *testing.T, p1 net.Conn, frames chan<- *drivers.ExecTaskStreamingResponseMsg, errCh chan<- error) {
|
|
// Start the decoder
|
|
decoder := codec.NewDecoder(p1, nstructs.MsgpackHandle)
|
|
|
|
for {
|
|
var msg cstructs.StreamErrWrapper
|
|
if err := decoder.Decode(&msg); err != nil {
|
|
if err == io.EOF || strings.Contains(err.Error(), "closed") {
|
|
return
|
|
}
|
|
t.Logf("received error decoding: %#v", err)
|
|
|
|
errCh <- fmt.Errorf("error decoding: %v", err)
|
|
return
|
|
}
|
|
|
|
if msg.Error != nil {
|
|
errCh <- msg.Error
|
|
continue
|
|
}
|
|
|
|
var frame drivers.ExecTaskStreamingResponseMsg
|
|
if err := json.Unmarshal(msg.Payload, &frame); err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
t.Logf("received message: %#v", msg)
|
|
frames <- &frame
|
|
}
|
|
}
|