Skip to content

Commit 1e07daa

Browse files
catsbycatsby
committed
drivers: Add/Use go-getter to fetch remote binaries
Updates Qemu, Java drivers to use go-getter to fetch binaries Adds remote artifact support for Exec, Raw Exec drivers
1 parent c8bd79c commit 1e07daa

File tree

13 files changed

+407
-74
lines changed

13 files changed

+407
-74
lines changed

‎client/driver/driver.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package driver
33
import (
44
"fmt"
55
"log"
6+
"path/filepath"
67
"sync"
78

89
"github.com/hashicorp/nomad/client/allocdir"
@@ -114,6 +115,12 @@ func TaskEnvironmentVariables(ctx *ExecContext, task *structs.Task) environment.
114115

115116
if ctx.AllocDir != nil {
116117
env.SetAllocDir(ctx.AllocDir.SharedDir)
118+
taskdir, ok := ctx.AllocDir.TaskDirs[task.Name]
119+
if !ok {
120+
// TODO: Update this to return an error
121+
}
122+
123+
env.SetTaskLocalDir(filepath.Join(taskdir, allocdir.TaskLocal))
117124
}
118125

119126
if task.Resources != nil {

‎client/driver/exec.go‎

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,15 @@ package driver
22

33
import (
44
"fmt"
5+
"log"
6+
"path"
7+
"path/filepath"
58
"runtime"
69
"syscall"
710
"time"
811

12+
"github.com/hashicorp/go-getter"
13+
"github.com/hashicorp/nomad/client/allocdir"
914
"github.com/hashicorp/nomad/client/config"
1015
"github.com/hashicorp/nomad/client/executor"
1116
"github.com/hashicorp/nomad/nomad/structs"
@@ -41,12 +46,40 @@ func (d *ExecDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool,
4146
}
4247

4348
func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
44-
// Get the command
49+
// Get the command to be ran
4550
command, ok := task.Config["command"]
4651
if !ok || command == "" {
4752
return nil, fmt.Errorf("missing command for exec driver")
4853
}
4954

55+
// Check if an artificat is specified and attempt to download it
56+
source, ok := task.Config["artifact_source"]
57+
if ok && source != "" {
58+
// Proceed to download an artifact to be executed.
59+
// We use go-getter to support a variety of protocols, but need to change
60+
// file permissions of the resulted download to be executable
61+
62+
// Create a location to download the artifact.
63+
taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
64+
if !ok {
65+
return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
66+
}
67+
destDir := filepath.Join(taskDir, allocdir.TaskLocal)
68+
69+
artifactName := path.Base(source)
70+
artifactFile := filepath.Join(destDir, artifactName)
71+
if err := getter.GetFile(artifactFile, source); err != nil {
72+
return nil, fmt.Errorf("Error downloading artifact for Exec driver: %s", err)
73+
}
74+
75+
// Add execution permissions to the newly downloaded artifact
76+
if runtime.GOOS != "windows" {
77+
if err := syscall.Chmod(artifactFile, 0755); err != nil {
78+
log.Printf("[ERR] driver.Exec: Error making artifact executable: %s", err)
79+
}
80+
}
81+
}
82+
5083
// Get the environment variables.
5184
envVars := TaskEnvironmentVariables(ctx, task)
5285

‎client/driver/exec_test.go‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"io/ioutil"
66
"path/filepath"
77
"reflect"
8+
"runtime"
89
"testing"
910
"time"
1011

@@ -120,6 +121,104 @@ func TestExecDriver_Start_Wait(t *testing.T) {
120121
}
121122
}
122123

124+
func TestExecDriver_Start_Artifact_basic(t *testing.T) {
125+
ctestutils.ExecCompatible(t)
126+
var file string
127+
switch runtime.GOOS {
128+
case "darwin":
129+
file = "hi_darwin_amd64"
130+
default:
131+
file = "hi_linux_amd64"
132+
}
133+
134+
task := &structs.Task{
135+
Name: "sleep",
136+
Config: map[string]string{
137+
"artifact_source": fmt.Sprintf("https://dl.dropboxusercontent.com/u/47675/jar_thing/%s", file),
138+
"command": filepath.Join("$NOMAD_TASK_DIR", file),
139+
},
140+
Resources: basicResources,
141+
}
142+
143+
driverCtx := testDriverContext(task.Name)
144+
ctx := testDriverExecContext(task, driverCtx)
145+
defer ctx.AllocDir.Destroy()
146+
d := NewExecDriver(driverCtx)
147+
148+
handle, err := d.Start(ctx, task)
149+
if err != nil {
150+
t.Fatalf("err: %v", err)
151+
}
152+
if handle == nil {
153+
t.Fatalf("missing handle")
154+
}
155+
156+
// Update should be a no-op
157+
err = handle.Update(task)
158+
if err != nil {
159+
t.Fatalf("err: %v", err)
160+
}
161+
162+
// Task should terminate quickly
163+
select {
164+
case err := <-handle.WaitCh():
165+
if err != nil {
166+
t.Fatalf("err: %v", err)
167+
}
168+
case <-time.After(5 * time.Second):
169+
t.Fatalf("timeout")
170+
}
171+
}
172+
173+
func TestExecDriver_Start_Artifact_expanded(t *testing.T) {
174+
ctestutils.ExecCompatible(t)
175+
var file string
176+
switch runtime.GOOS {
177+
case "darwin":
178+
file = "hi_darwin_amd64"
179+
default:
180+
file = "hi_linux_amd64"
181+
}
182+
183+
task := &structs.Task{
184+
Name: "sleep",
185+
Config: map[string]string{
186+
"artifact_source": fmt.Sprintf("https://dl.dropboxusercontent.com/u/47675/jar_thing/%s", file),
187+
"command": "/bin/bash",
188+
"args": fmt.Sprintf("-c '/bin/sleep 1 && %s'", filepath.Join("$NOMAD_TASK_DIR", file)),
189+
},
190+
Resources: basicResources,
191+
}
192+
193+
driverCtx := testDriverContext(task.Name)
194+
ctx := testDriverExecContext(task, driverCtx)
195+
defer ctx.AllocDir.Destroy()
196+
d := NewExecDriver(driverCtx)
197+
198+
handle, err := d.Start(ctx, task)
199+
if err != nil {
200+
t.Fatalf("err: %v", err)
201+
}
202+
if handle == nil {
203+
t.Fatalf("missing handle")
204+
}
205+
206+
// Update should be a no-op
207+
err = handle.Update(task)
208+
if err != nil {
209+
t.Fatalf("err: %v", err)
210+
}
211+
212+
// Task should terminate quickly
213+
select {
214+
case err := <-handle.WaitCh():
215+
if err != nil {
216+
t.Fatalf("err: %v", err)
217+
}
218+
case <-time.After(5 * time.Second):
219+
t.Fatalf("timeout")
220+
}
221+
}
123222
func TestExecDriver_Start_Wait_AllocDir(t *testing.T) {
124223
ctestutils.ExecCompatible(t)
125224

‎client/driver/java.go‎

Lines changed: 8 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,6 @@ package driver
33
import (
44
"bytes"
55
"fmt"
6-
"io"
7-
"net/http"
8-
"os"
96
"os/exec"
107
"path"
118
"path/filepath"
@@ -14,6 +11,7 @@ import (
1411
"syscall"
1512
"time"
1613

14+
"github.com/hashicorp/go-getter"
1715
"github.com/hashicorp/nomad/client/allocdir"
1816
"github.com/hashicorp/nomad/client/config"
1917
"github.com/hashicorp/nomad/client/executor"
@@ -97,37 +95,18 @@ func (d *JavaDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
9795
return nil, fmt.Errorf("missing jar source for Java Jar driver")
9896
}
9997

100-
// Attempt to download the thing
101-
// Should be extracted to some kind of Http Fetcher
102-
// Right now, assume publicly accessible HTTP url
103-
resp, err := http.Get(source)
104-
if err != nil {
105-
return nil, fmt.Errorf("Error downloading source for Java driver: %s", err)
106-
}
107-
108-
// Get the tasks local directory.
10998
taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
11099
if !ok {
111100
return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
112101
}
113-
taskLocal := filepath.Join(taskDir, allocdir.TaskLocal)
114102

115-
// Create a location to download the binary.
116-
fName := path.Base(source)
117-
fPath := filepath.Join(taskLocal, fName)
118-
f, err := os.OpenFile(fPath, os.O_CREATE|os.O_WRONLY, 0666)
119-
if err != nil {
120-
return nil, fmt.Errorf("Error opening file to download to: %s", err)
121-
}
122-
123-
defer f.Close()
124-
defer resp.Body.Close()
103+
destDir := filepath.Join(taskDir, allocdir.TaskLocal)
125104

126-
// Copy remote file to local directory for execution
127-
// TODO: a retry of sort if io.Copy fails, for large binaries
128-
_, ioErr := io.Copy(f, resp.Body)
129-
if ioErr != nil {
130-
return nil, fmt.Errorf("Error copying jar from source: %s", ioErr)
105+
// Create a location to download the binary.
106+
jarName := path.Base(source)
107+
jarPath := filepath.Join(destDir, jarName)
108+
if err := getter.GetFile(jarPath, source); err != nil {
109+
return nil, fmt.Errorf("Error downloading source for Java driver: %s", err)
131110
}
132111

133112
// Get the environment variables.
@@ -141,10 +120,8 @@ func (d *JavaDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
141120
args = append(args, jvm_options)
142121
}
143122

144-
// Build the argument list
145-
args = append(args, "-jar", filepath.Join(allocdir.TaskLocal, fName))
146-
147123
// Build the argument list.
124+
args = append(args, "-jar", filepath.Join(allocdir.TaskLocal, jarName))
148125
if argRaw, ok := task.Config["args"]; ok {
149126
args = append(args, argRaw)
150127
}

‎client/driver/qemu.go‎

Lines changed: 9 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import (
88
"fmt"
99
"io"
1010
"log"
11-
"net/http"
1211
"os"
1312
"os/exec"
1413
"path/filepath"
@@ -19,6 +18,7 @@ import (
1918
"syscall"
2019
"time"
2120

21+
"github.com/hashicorp/go-getter"
2222
"github.com/hashicorp/nomad/client/allocdir"
2323
"github.com/hashicorp/nomad/client/config"
2424
"github.com/hashicorp/nomad/nomad/structs"
@@ -94,45 +94,25 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
9494
return nil, fmt.Errorf("Missing required Task Resource: Memory")
9595
}
9696

97-
// Attempt to download the thing
98-
// Should be extracted to some kind of Http Fetcher
99-
// Right now, assume publicly accessible HTTP url
100-
resp, err := http.Get(source)
101-
if err != nil {
102-
return nil, fmt.Errorf("Error downloading source for Qemu driver: %s", err)
103-
}
104-
10597
// Get the tasks local directory.
10698
taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
10799
if !ok {
108100
return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
109101
}
110-
taskLocal := filepath.Join(taskDir, allocdir.TaskLocal)
111102

112-
// Create a location in the local directory to download and store the image.
113-
// TODO: Caching
103+
// Create a location to download the binary.
104+
destDir := filepath.Join(taskDir, allocdir.TaskLocal)
114105
vmID := fmt.Sprintf("qemu-vm-%s-%s", structs.GenerateUUID(), filepath.Base(source))
115-
fPath := filepath.Join(taskLocal, vmID)
116-
vmPath, err := os.OpenFile(fPath, os.O_CREATE|os.O_WRONLY, 0666)
117-
if err != nil {
118-
return nil, fmt.Errorf("Error opening file to download to: %s", err)
119-
}
120-
121-
defer vmPath.Close()
122-
defer resp.Body.Close()
123-
124-
// Copy remote file to local AllocDir for execution
125-
// TODO: a retry of sort if io.Copy fails, for large binaries
126-
_, ioErr := io.Copy(vmPath, resp.Body)
127-
if ioErr != nil {
128-
return nil, fmt.Errorf("Error copying Qemu image from source: %s", ioErr)
106+
vmPath := filepath.Join(destDir, vmID)
107+
if err := getter.GetFile(vmPath, source); err != nil {
108+
return nil, fmt.Errorf("Error downloading artifact for Qemu driver: %s", err)
129109
}
130110

131111
// compute and check checksum
132112
if check, ok := task.Config["checksum"]; ok {
133113
d.logger.Printf("[DEBUG] Running checksum on (%s)", vmID)
134114
hasher := sha256.New()
135-
file, err := os.Open(vmPath.Name())
115+
file, err := os.Open(vmPath)
136116
if err != nil {
137117
return nil, fmt.Errorf("Failed to open file for checksum")
138118
}
@@ -163,7 +143,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
163143
"-machine", "type=pc,accel=" + accelerator,
164144
"-name", vmID,
165145
"-m", mem,
166-
"-drive", "file=" + vmPath.Name(),
146+
"-drive", "file=" + vmPath,
167147
"-nodefconfig",
168148
"-nodefaults",
169149
"-nographic",
@@ -240,7 +220,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
240220
// Create and Return Handle
241221
h := &qemuHandle{
242222
proc: cmd.Process,
243-
vmID: vmPath.Name(),
223+
vmID: vmPath,
244224
doneCh: make(chan struct{}),
245225
waitCh: make(chan error, 1),
246226
}

0 commit comments

Comments
 (0)