diff --git a/client/driver/docker.go b/client/driver/docker.go
index bc0c5ec89..f2bedc0ee 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/mock_driver.go b/client/driver/mock_driver.go
index ad6e2486e..e13cad580 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 144166fb1..60f78c741 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 8e1ec9423..7db4f7ef7 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)