c24f4d9925
The client allocation GC API returns a misleading error message when the allocation exists but is not yet eligible for GC. Make this clear in the error response. Note in the docs that the allocation will still show on the server responses.
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: structs.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: structs.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: structs.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: structs.ErrPermissionDenied.Error(),
|
|
},
|
|
{
|
|
Name: "alloc-exec token",
|
|
Token: tokenAllocExec.SecretID,
|
|
ExpectedError: structs.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
|
|
}
|
|
}
|