diff --git a/client/driver/docker.go b/client/driver/docker.go
index bc0c5ec89b82..f2bedc0ee712 100644
--- a/client/driver/docker.go
+++ b/client/driver/docker.go
@@ -785,7 +785,7 @@ func (d *DockerDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle
 	}
 
 	if err := d.createImage(driverConfig, client, taskDir); err != nil {
-		return nil, fmt.Errorf("failed to create image: %v", err)
+		return nil, err
 	}
 
 	image := driverConfig.ImageName
diff --git a/client/driver/docker_test.go b/client/driver/docker_test.go
index e4cd34ad34c9..b97142236421 100644
--- a/client/driver/docker_test.go
+++ b/client/driver/docker_test.go
@@ -288,6 +288,46 @@ func TestDockerDriver_Start_LoadImage(t *testing.T) {
 
 }
 
+func TestDockerDriver_Start_BadPull_Recoverable(t *testing.T) {
+	if !testutil.DockerIsConnected(t) {
+		t.SkipNow()
+	}
+	task := &structs.Task{
+		Name: "busybox-demo",
+		Config: map[string]interface{}{
+			"image":   "127.0.1.1:32121/foo", // bad path
+			"command": "/bin/echo",
+			"args": []string{
+				"hello",
+			},
+		},
+		LogConfig: &structs.LogConfig{
+			MaxFiles:      10,
+			MaxFileSizeMB: 10,
+		},
+		Resources: &structs.Resources{
+			MemoryMB: 256,
+			CPU:      512,
+		},
+	}
+
+	driverCtx, execCtx := testDriverContexts(task)
+	driverCtx.config.Options = map[string]string{"docker.cleanup.image": "false"}
+	defer execCtx.AllocDir.Destroy()
+	d := NewDockerDriver(driverCtx)
+
+	_, err := d.Start(execCtx, task)
+	if err == nil {
+		t.Fatalf("want err: %v", err)
+	}
+
+	if rerr, ok := err.(*structs.RecoverableError); !ok {
+		t.Fatalf("want recoverable error: %+v", err)
+	} else if !rerr.Recoverable {
+		t.Fatalf("error not recoverable: %+v", err)
+	}
+}
+
 func TestDockerDriver_Start_Wait_AllocDir(t *testing.T) {
 	// This test requires that the alloc dir be mounted into docker as a volume.
 	// Because this cannot happen when docker is run remotely, e.g. when running
diff --git a/client/driver/mock_driver.go b/client/driver/mock_driver.go
index ad6e2486e7c5..e13cad580453 100644
--- a/client/driver/mock_driver.go
+++ b/client/driver/mock_driver.go
@@ -27,6 +27,13 @@ func init() {
 // MockDriverConfig is the driver configuration for the MockDriver
 type MockDriverConfig struct {
 
+	// StartErr specifies the error that should be returned when starting the
+	// mock driver.
+	StartErr string `mapstructure:"start_error"`
+
+	// StartErrRecoverable marks the error returned is recoverable
+	StartErrRecoverable bool `mapstructure:"start_error_recoverable"`
+
 	// KillAfter is the duration after which the mock driver indicates the task
 	// has exited after getting the initial SIGINT signal
 	KillAfter time.Duration `mapstructure:"kill_after"`
@@ -83,6 +90,10 @@ func (m *MockDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
 		return nil, err
 	}
 
+	if driverConfig.StartErr != "" {
+		return nil, structs.NewRecoverableError(errors.New(driverConfig.StartErr), driverConfig.StartErrRecoverable)
+	}
+
 	h := mockDriverHandle{
 		taskName:    task.Name,
 		runFor:      driverConfig.RunFor,
diff --git a/client/task_runner.go b/client/task_runner.go
index 144166fb1f13..60f78c741646 100644
--- a/client/task_runner.go
+++ b/client/task_runner.go
@@ -1018,8 +1018,17 @@ func (r *TaskRunner) startTask() error {
 	// Start the job
 	handle, err := driver.Start(r.ctx, r.task)
 	if err != nil {
-		return fmt.Errorf("failed to start task '%s' for alloc '%s': %v",
+		wrapped := fmt.Errorf("failed to start task '%s' for alloc '%s': %v",
 			r.task.Name, r.alloc.ID, err)
+
+		r.logger.Printf("[INFO] client: %v", wrapped)
+
+		if rerr, ok := err.(*structs.RecoverableError); ok {
+			return structs.NewRecoverableError(wrapped, rerr.Recoverable)
+		}
+
+		return wrapped
+
 	}
 
 	r.handleLock.Lock()
diff --git a/client/task_runner_test.go b/client/task_runner_test.go
index 8e1ec942387c..7db4f7ef7e61 100644
--- a/client/task_runner_test.go
+++ b/client/task_runner_test.go
@@ -109,6 +109,45 @@ func TestTaskRunner_SimpleRun(t *testing.T) {
 	}
 }
 
+func TestTaskRunner_Run_RecoverableStartError(t *testing.T) {
+	alloc := mock.Alloc()
+	task := alloc.Job.TaskGroups[0].Tasks[0]
+	task.Driver = "mock_driver"
+	task.Config = map[string]interface{}{
+		"exit_code":               0,
+		"start_error":             "driver failure",
+		"start_error_recoverable": true,
+	}
+
+	upd, tr := testTaskRunnerFromAlloc(true, alloc)
+	tr.MarkReceived()
+	go tr.Run()
+	defer tr.Destroy(structs.NewTaskEvent(structs.TaskKilled))
+	defer tr.ctx.AllocDir.Destroy()
+
+	testutil.WaitForResult(func() (bool, error) {
+		if l := len(upd.events); l < 3 {
+			return false, fmt.Errorf("Expect at least three  events; got %v", l)
+		}
+
+		if upd.events[0].Type != structs.TaskReceived {
+			return false, fmt.Errorf("First Event was %v; want %v", upd.events[0].Type, structs.TaskReceived)
+		}
+
+		if upd.events[1].Type != structs.TaskDriverFailure {
+			return false, fmt.Errorf("Second Event was %v; want %v", upd.events[1].Type, structs.TaskDriverFailure)
+		}
+
+		if upd.events[2].Type != structs.TaskRestarting {
+			return false, fmt.Errorf("Second Event was %v; want %v", upd.events[2].Type, structs.TaskRestarting)
+		}
+
+		return true, nil
+	}, func(err error) {
+		t.Fatalf("err: %v", err)
+	})
+}
+
 func TestTaskRunner_Destroy(t *testing.T) {
 	ctestutil.ExecCompatible(t)
 	upd, tr := testTaskRunner(true)