Merge pull request #5690 from hashicorp/f-nomad-exec-part-04-rkt

implement nomad exec for rkt

command/node_config.go

@@ -82,7 +82,7 @@ func (c *NodeConfigCommand) Run(args []string) int {
 	if updateServers {
 		// Get the server addresses
 		if len(args) == 0 {
-			c.Ui.Error("If the '-update-servers' flag is set, atleast one server argument must be provided")
+			c.Ui.Error("If the '-update-servers' flag is set, at least one server argument must be provided")
 			c.Ui.Error(commandErrorText(c))
 			return 1
 		}

drivers/exec/driver.go

@@ -506,7 +506,7 @@ func (d *Driver) SignalTask(taskID string, signal string) error {
 
 func (d *Driver) ExecTask(taskID string, cmd []string, timeout time.Duration) (*drivers.ExecTaskResult, error) {
 	if len(cmd) == 0 {
-		return nil, fmt.Errorf("error cmd must have atleast one value")
+		return nil, fmt.Errorf("error cmd must have at least one value")
 	}
 	handle, ok := d.tasks.Get(taskID)
 	if !ok {
@@ -540,7 +540,7 @@ func (d *Driver) ExecTaskStreamingRaw(ctx context.Context,
 	stream drivers.ExecTaskStream) error {
 
 	if len(command) == 0 {
-		return fmt.Errorf("error cmd must have atleast one value")
+		return fmt.Errorf("error cmd must have at least one value")
 	}
 	handle, ok := d.tasks.Get(taskID)
 	if !ok {

drivers/java/driver.go

@@ -563,7 +563,7 @@ func (d *Driver) ExecTaskStreamingRaw(ctx context.Context,
 	stream drivers.ExecTaskStream) error {
 
 	if len(command) == 0 {
-		return fmt.Errorf("error cmd must have atleast one value")
+		return fmt.Errorf("error cmd must have at least one value")
 	}
 	handle, ok := d.tasks.Get(taskID)
 	if !ok {

drivers/rkt/driver.go

@@ -863,7 +863,7 @@ func (d *Driver) SignalTask(taskID string, signal string) error {
 
 func (d *Driver) ExecTask(taskID string, cmdArgs []string, timeout time.Duration) (*drivers.ExecTaskResult, error) {
 	if len(cmdArgs) == 0 {
-		return nil, fmt.Errorf("error cmd must have atleast one value")
+		return nil, fmt.Errorf("error cmd must have at least one value")
 	}
 	handle, ok := d.tasks.Get(taskID)
 	if !ok {
@@ -891,6 +891,28 @@ func (d *Driver) ExecTask(taskID string, cmdArgs []string, timeout time.Duration
 
 }
 
+var _ drivers.ExecTaskStreamingRawDriver = (*Driver)(nil)
+
+func (d *Driver) ExecTaskStreamingRaw(ctx context.Context,
+	taskID string,
+	command []string,
+	tty bool,
+	stream drivers.ExecTaskStream) error {
+
+	if len(command) == 0 {
+		return fmt.Errorf("error cmd must have at least one value")
+	}
+	handle, ok := d.tasks.Get(taskID)
+	if !ok {
+		return drivers.ErrTaskNotFound
+	}
+
+	enterCmd := []string{rktCmd, "enter", handle.uuid, handle.env.ReplaceEnv(command[0])}
+	enterCmd = append(enterCmd, handle.env.ParseAndReplace(command[1:])...)
+
+	return handle.exec.ExecStreaming(ctx, enterCmd, tty, stream)
+}
+
 // GetAbsolutePath returns the absolute path of the passed binary by resolving
 // it in the path and following symlinks.
 func GetAbsolutePath(bin string) (string, error) {

drivers/rkt/driver_test.go

@@ -913,3 +913,67 @@ config {
 
 	require.EqualValues(t, expected, tc)
 }
+
+func TestRkt_ExecTaskStreaming(t *testing.T) {
+	ctestutil.RktCompatible(t)
+	if !testutil.IsCI() {
+		t.Parallel()
+	}
+
+	require := require.New(t)
+	d := NewRktDriver(testlog.HCLogger(t))
+	harness := dtestutil.NewDriverHarness(t, d)
+
+	task := &drivers.TaskConfig{
+		ID:      uuid.Generate(),
+		AllocID: uuid.Generate(),
+		Name:    "etcd",
+		Resources: &drivers.Resources{
+			NomadResources: &structs.AllocatedTaskResources{
+				Memory: structs.AllocatedMemoryResources{
+					MemoryMB: 128,
+				},
+				Cpu: structs.AllocatedCpuResources{
+					CpuShares: 100,
+				},
+			},
+			LinuxResources: &drivers.LinuxResources{
+				MemoryLimitBytes: 134217728,
+				CPUShares:        100,
+			},
+		},
+	}
+
+	tc := &TaskConfig{
+		ImageName: "docker://busybox:1.29.3",
+		Command:   "/bin/sleep",
+		Args:      []string{"1000"},
+		Net:       []string{"none"},
+	}
+	require.NoError(task.EncodeConcreteDriverConfig(&tc))
+	testtask.SetTaskConfigEnv(task)
+
+	cleanup := harness.MkAllocDir(task, true)
+	defer cleanup()
+
+	_, _, err := harness.StartTask(task)
+	require.NoError(err)
+	defer d.DestroyTask(task.ID, true)
+
+	// wait for container to be up and executable
+	testutil.WaitForResult(func() (bool, error) {
+		res, err := d.ExecTask(task.ID, []string{"/bin/sh", "-c", "echo hi"}, time.Second)
+		if err != nil {
+			return false, fmt.Errorf("failed to exec: %#v", err)
+		}
+		if !res.ExitResult.Successful() {
+			return false, fmt.Errorf("ps failed: %#v %#v", res.ExitResult, res)
+		}
+		return true, nil
+	}, func(err error) {
+		require.NoError(err)
+	})
+
+	dtestutil.ExecTaskStreamingConformanceTests(t, harness, task.ID)
+
+}