func (e *Executor) LaunchTask(driver executor.ExecutorDriver, task *mesos.TaskInfo) {
	Logger.Infof("[LaunchTask] %s", task)

	Config.Read(task)

	runStatus := &mesos.TaskStatus{
		TaskId: task.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}

	if _, err := driver.SendStatusUpdate(runStatus); err != nil {
		Logger.Errorf("Failed to send status update: %s", runStatus)
	}

	go func() {
		e.producer = e.newSyslogProducer()
		e.producer.Start()
		<-e.close

		// finish task
		Logger.Infof("Finishing task %s", task.GetName())
		finStatus := &mesos.TaskStatus{
			TaskId: task.GetTaskId(),
			State:  mesos.TaskState_TASK_FINISHED.Enum(),
		}
		if _, err := driver.SendStatusUpdate(finStatus); err != nil {
			Logger.Errorf("Failed to send status update: %s", finStatus)
			os.Exit(1)
		}
		Logger.Infof("Task %s has finished", task.GetName())
		time.Sleep(time.Second)
		os.Exit(0)
	}()
}

func (self *DatabaseExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	log.Infoln("Launching task", taskInfo.GetName(), "with command", taskInfo.Command.GetValue())
	self.driver = driver
	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		log.Infoln("Got error", err)
	}

	self.tasksLaunched++
	log.Infoln("Total tasks launched ", self.tasksLaunched)
	//
	// this is where one would perform the requested task
	//
	self.runProg()
	time.Sleep(time.Second * 40)

	// finish task
	log.Infoln("Finishing task", taskInfo.GetName())
	finStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_FINISHED.Enum(),
	}
	_, err = driver.SendStatusUpdate(finStatus)
	if err != nil {
		log.Infoln("Got error", err)

	}
	log.Infoln("Task finished", taskInfo.GetName())
}

func (e *DiegoExecutor) watchTasks(driver exec.ExecutorDriver) {
	timeInIdle := 0 * time.Second
	for {
		select {
		case <-time.After(1 * time.Second):
			state, err := e.cellClient.State()
			if err != nil {
				fmt.Println("Get rep state error", err)
				break
			}

			repContainerSet := map[string]bool{}
			for _, lrp := range state.LRPs {
				repContainerSet[lrp.Identifier()] = true
			}
			// require patch to rep so diego tasks can be returned
			for _, task := range state.Tasks {
				repContainerSet[task.Identifier()] = true
			}

			e.lock.Lock()
			// update task to running status if needed
			for taskId, _ := range repContainerSet {
				taskState := e.taskStateMap[taskId]
				if taskState != mesos.TaskState_TASK_RUNNING {
					_, err := sendTaskStatusUpdate(driver, taskId, mesos.TaskState_TASK_RUNNING)
					if err == nil {
						e.taskStateMap[taskId] = mesos.TaskState_TASK_RUNNING
					}
				}
			}

			// find tasks not in cell state
			for taskId, _ := range e.taskStateMap {
				_, exists := repContainerSet[taskId]
				if !exists { // not found, report finished (or failed?)
					_, err := sendTaskStatusUpdate(driver, taskId, mesos.TaskState_TASK_FINISHED)
					if err == nil {
						delete(e.taskStateMap, taskId)
					}
				}
			}

			fmt.Println("repContainerSet: ", repContainerSet)
			fmt.Println("taskStateMap: ", e.taskStateMap)

			// nothing running, abort if been idle for a while
			if len(state.LRPs) == 0 && len(state.Tasks) == 0 && len(e.taskStateMap) == 0 {
				timeInIdle += (1 * time.Second)
				if timeInIdle >= 10*time.Second {
					driver.Abort()
				}
			} else {
				timeInIdle = 0
			}

			e.lock.Unlock()
		}
	}
}

func (k *QingYuanExecutor) sendStatus(driver bindings.ExecutorDriver, status *mesos.TaskStatus) {
	select {
	case <-k.done:
	default:
		k.outgoing <- func() (mesos.Status, error) { return driver.SendStatusUpdate(status) }
	}
}

func (k *KubernetesExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver, msg string) {
	select {
	case <-k.terminate:
	default:
		k.outgoing <- func() (mesos.Status, error) { return driver.SendFrameworkMessage(msg) }
	}
}

func (exec *exampleExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Println("Launching task", taskInfo.GetName(), "with command", taskInfo.Command.GetValue())

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}

	exec.tasksLaunched++
	fmt.Println("Total tasks launched ", exec.tasksLaunched)
	//
	// this is where one would perform the requested task
	//

	// finish task
	fmt.Println("Finishing task", taskInfo.GetName())
	finStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_FINISHED.Enum(),
	}
	_, err = driver.SendStatusUpdate(finStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}
	fmt.Println("Task finished", taskInfo.GetName())
}

func (k *KubernetesExecutor) sendStatus(driver bindings.ExecutorDriver, status *mesos.TaskStatus) {
	select {
	case <-k.terminate:
	default:
		k.outgoing <- func() (mesos.Status, error) { return driver.SendStatusUpdate(status) }
	}
}

func (e *RancherExecutor) LaunchTask(d executor.ExecutorDriver, task *mesosproto.TaskInfo) {
	taskId := task.TaskId
	s := mesosproto.TaskState_TASK_RUNNING
	d.SendStatusUpdate(&mesosproto.TaskStatus{TaskId: taskId, State: &s})
	var taskStruct TaskStruct
	json.Unmarshal(task.Data, &taskStruct)
	orchestrator := &orchestrator.Orchestrator{
		RosImg:    e.rosImg,
		RosHDD:    e.rosHDD,
		Iface:     e.iface,
		IfaceCIDR: e.ifaceCIDR,
		ImageRepo: taskStruct.ImageRepo,
		RegUrl:    taskStruct.RegistrationUrl,
		ImageTag:  taskStruct.ImageTag,
		HostUuid:  taskStruct.HostUuid,
		Hostname:  taskStruct.Hostname,
	}
	err := orchestrator.CreateAndBootstrap()
	//TBD: Read message type and add supp for DELETE
	s = mesosproto.TaskState_TASK_RUNNING
	if err != nil {
		log.Error(err)
		s = mesosproto.TaskState_TASK_ERROR
	}
	d.SendStatusUpdate(&mesosproto.TaskStatus{TaskId: taskId, State: &s})
}

func (e *Executor) LaunchTask(driver executor.ExecutorDriver, task *mesos.TaskInfo) {
	Logger.Infof("[LaunchTask] %s", task)

	runStatus := &mesos.TaskStatus{
		TaskId: task.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}

	if _, err := driver.SendStatusUpdate(runStatus); err != nil {
		Logger.Errorf("Failed to send status update: %s", runStatus)
		os.Exit(1) //TODO not sure if we should exit in this case, but probably yes
	}

	go func() {
		err := e.start()
		if err != nil {
			Logger.Errorf("Can't start executor: %s", err)
		}

		// finish task
		Logger.Infof("Finishing task %s", task.GetName())
		finStatus := &mesos.TaskStatus{
			TaskId: task.GetTaskId(),
			State:  mesos.TaskState_TASK_FINISHED.Enum(),
		}
		if _, err := driver.SendStatusUpdate(finStatus); err != nil {
			Logger.Errorf("Failed to send status update: %s", finStatus)
			os.Exit(1)
		}
		Logger.Infof("Task %s has finished", task.GetName())
	}()
}

func (exec *ExecutorCore) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	exec.lock.Lock()
	defer exec.lock.Unlock()
	fmt.Println("Launching task", taskInfo.GetName(), "with command", taskInfo.Command.GetValue())
	os.Args[0] = fmt.Sprintf("executor - %s", taskInfo.TaskId.GetValue())

	//fmt.Println("Other hilarious facts: ", taskInfo)

	//
	// this is where one would perform the requested task
	//
	fmt.Println("Starting task")

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.TaskId,
		State:  mesos.TaskState_TASK_STARTING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)

	if err != nil {
		log.Panic("Got error", err)
	}

	if exec.riakNode != nil {
		log.Fatalf("Task being started, twice, existing task: %+v, new task: %+v", exec.riakNode)
	}
	exec.riakNode = NewRiakNode(taskInfo, exec)
	exec.riakNode.Run()

}

func (k *QingYuanExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver, msg string) {
	select {
	case <-k.done:
	default:
		k.outgoing <- func() (mesos.Status, error) { return driver.SendFrameworkMessage(msg) }
	}
}

// mesos.Executor interface method.
// Invoked when a task has been launched on this executor.
func (this *TransformExecutor) LaunchTask(driver executor.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Printf("Launching task %s with command %s\n", taskInfo.GetName(), taskInfo.Command.GetValue())

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}

	if _, err := driver.SendStatusUpdate(runStatus); err != nil {
		fmt.Printf("Failed to send status update: %s\n", runStatus)
	}

	go func() {
		//        this.avroDecoder = kafka.NewKafkaAvroDecoder(this.config.SchemaRegistryUrl)
		this.startHTTPServer()
		this.startProducer()
		<-this.close
		close(this.incoming)

		// finish task
		fmt.Printf("Finishing task %s\n", taskInfo.GetName())
		finStatus := &mesos.TaskStatus{
			TaskId: taskInfo.GetTaskId(),
			State:  mesos.TaskState_TASK_FINISHED.Enum(),
		}
		if _, err := driver.SendStatusUpdate(finStatus); err != nil {
			fmt.Printf("Failed to send status update: %s\n", finStatus)
		}
		fmt.Printf("Task %s has finished\n", taskInfo.GetName())
	}()
}

func (exec *exampleExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Println("Launching task", taskInfo.GetName(), "with command", taskInfo.Command.GetValue())

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}

	exec.tasksLaunched++
	fmt.Println("Total tasks launched ", exec.tasksLaunched)
	//
	// this is where one would perform the requested task
	//

	// rexray.UpdateLogLevel()
	// rexray.InitDriverManagers()
	// sdm := rexray.GetSdm()
	//
	// allVolumes, err := sdm.GetVolume("", "")
	// if err != nil {
	// 	log.Fatal(err)
	// }
	//
	// if len(allVolumes) > 0 {
	// 	yamlOutput, err := yaml.Marshal(&allVolumes)
	// 	if err != nil {
	// 		log.Fatal(err)
	// 	}
	// 	fmt.Printf(string(yamlOutput))
	// }

	client, _ := plugins.NewClient("unix:///run/mesos/executor/rexray.sock", tlsconfig.Options{InsecureSkipVerify: true})
	vd := volumeDriverProxy{client}
	err = vd.Create("test", nil)
	if err != nil {
		fmt.Println("Got error", err)
	}

	// finish task
	fmt.Println("Finishing task", taskInfo.GetName())
	finStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_FINISHED.Enum(),
	}
	_, err = driver.SendStatusUpdate(finStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}
	fmt.Println("Task finished", taskInfo.GetName())
}

func (exec *exampleExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Printf("Launching task %v with data [%#x]\n", taskInfo.GetName(), taskInfo.Data)

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}

	exec.tasksLaunched++
	fmt.Println("Total tasks launched ", exec.tasksLaunched)

	// Download image
	fileName, err := downloadImage(string(taskInfo.Data))
	if err != nil {
		fmt.Printf("Failed to download image with error: %v\n", err)
		return
	}
	fmt.Printf("Downloaded image: %v\n", fileName)

	// Process image
	fmt.Printf("Processing image: %v\n", fileName)
	outFile, err := procImage(fileName)
	if err != nil {
		fmt.Printf("Failed to process image with error: %v\n", err)
		return
	}

	// Upload image
	fmt.Printf("Uploading image: %v\n", outFile)
	if err = uploadImage("http://127.0.0.1:12345/", outFile); err != nil {
		fmt.Printf("Failed to upload image with error: %v\n", err)
		return
	} else {
		fmt.Printf("Uploaded image: %v\n", outFile)
	}

	// Finish task
	fmt.Println("Finishing task", taskInfo.GetName())
	finStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_FINISHED.Enum(),
	}
	_, err = driver.SendStatusUpdate(finStatus)
	if err != nil {
		fmt.Println("Got error", err)
		return
	}

	fmt.Println("Task finished", taskInfo.GetName())
}

func sendTaskStatusUpdate(driver exec.ExecutorDriver, taskId string, state mesos.TaskState) (mesos.Status, error) {
	taskStatus := mesos.TaskStatus{
		TaskId: &mesos.TaskID{Value: proto.String(taskId)},
		State:  &state,
	}
	driverStatus, err := driver.SendStatusUpdate(&taskStatus)
	if err != nil {
		fmt.Printf("Send task status error, driverStatus: %v, err: %v\n", driverStatus.String(), err)
	}
	return driverStatus, err
}

// assumes that caller has obtained state lock
func (k *KubernetesExecutor) doShutdown(driver bindings.ExecutorDriver) {
	defer func() {
		log.Errorf("exiting with unclean shutdown: %v", recover())
		if k.exitFunc != nil {
			k.exitFunc(1)
		}
	}()

	(&k.state).transitionTo(terminalState)

	// signal to all listeners that this KubeletExecutor is done!
	close(k.terminate)
	close(k.updateChan)
	close(k.nodeInfos)

	if k.shutdownAlert != nil {
		func() {
			util.HandleCrash()
			k.shutdownAlert()
		}()
	}

	log.Infoln("Stopping executor driver")
	_, err := driver.Stop()
	if err != nil {
		log.Warningf("failed to stop executor driver: %v", err)
	}

	log.Infoln("Shutdown the executor")

	// according to docs, mesos will generate TASK_LOST updates for us
	// if needed, so don't take extra time to do that here.
	k.tasks = map[string]*kuberTask{}

	select {
	// the main Run() func may still be running... wait for it to finish: it will
	// clear the pod configuration cleanly, telling k8s "there are no pods" and
	// clean up resources (pods, volumes, etc).
	case <-k.kubeletFinished:

	//TODO(jdef) attempt to wait for events to propagate to API server?

	// TODO(jdef) extract constant, should be smaller than whatever the
	// slave graceful shutdown timeout period is.
	case <-time.After(15 * time.Second):
		log.Errorf("timed out waiting for kubelet Run() to die")
	}
	log.Infoln("exiting")
	if k.exitFunc != nil {
		k.exitFunc(0)
	}
}

func (exec *ExecutorCore) KillTask(driver exec.ExecutorDriver, taskId *mesos.TaskID) {
	exec.lock.Lock()
	defer exec.lock.Unlock()
	fmt.Println("Kill task")
	runStatus := &mesos.TaskStatus{
		TaskId: exec.riakNode.taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_KILLED.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)

	if err != nil {
		log.Panic("Got error", err)
	}
}

func (exec *scraperExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Printf("Launching task %v with data [%#x]\n", taskInfo.GetName(), taskInfo.Data)

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}

	exec.tasksLaunched++
	fmt.Println("Total tasks launched ", exec.tasksLaunched)

	// Download html
	uri := string(taskInfo.Data)
	fileName, url, err := downloadHTML(uri)

	if err != nil {
		fmt.Printf("Failed to scrape html with error: %v\n", err)
		return
	}
	fmt.Printf("Scraped URI: %v\n", fileName)

	// Upload html
	path := base64.StdEncoding.EncodeToString([]byte(url))
	fmt.Printf("Uploading html: %v\n", fileName)
	if err = uploadImageToS3(path, fileName); err != nil {
		fmt.Printf("Failed to upload html with error: %v\n", err)
		return
	} else {
		fmt.Printf("Uploaded html: %v\n", fileName)
	}

	// Finish task
	fmt.Println("Finishing task", taskInfo.GetName())
	finStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_FINISHED.Enum(),
		Data:   []byte(uri),
	}
	_, err = driver.SendStatusUpdate(finStatus)
	if err != nil {
		fmt.Println("Got error", err)
		return
	}

	fmt.Println("Task finished", taskInfo.GetName())
}

// LaunchTask called when executor launch tasks
func (builder *ImageBuilder) LaunchTask(driver executor.ExecutorDriver, taskInfo *mesosproto.TaskInfo) {
	fmt.Printf("Launching task %v with ID %v\n", taskInfo.GetName(), taskInfo.GetTaskId().GetValue())

	status := &mesosproto.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesosproto.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(status)
	if err != nil {
		fmt.Println("Send task running status error: ", err)
	}

	// Download context tar file
	// use data in task info
	contextTar, err := utils.Download(string(taskInfo.Data))
	if err != nil {
		fmt.Printf("Download context error: %v", err)
	}

	// Untar context file(filename.tar -> filename)
	contextDir := strings.TrimSuffix(contextTar, ".tar")
	err = utils.UnTar(contextTar, contextDir)
	if err != nil {
		fmt.Printf("Untar context error: %v", err)
	}

	// Build image with context
	var buf bytes.Buffer
	opts := docker.BuildImageOptions{
		Name:           taskInfo.GetTaskId().GetValue(),
		ContextDir:     contextDir,
		SuppressOutput: false,
		OutputStream:   &buf,
	}
	err = builder.client.BuildImage(opts)
	if err != nil {
		fmt.Printf("Build image error: %v\n", err)
	}
	fmt.Println(buf.String())

	fmt.Println("Task finished", taskInfo.GetName())
	status.State = mesosproto.TaskState_TASK_FINISHED.Enum()
	_, err = driver.SendStatusUpdate(status)
	if err != nil {
		fmt.Println("Send task finished status error: ", err)
	}
}

func (exec *ghsvisExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
	fmt.Println("Launching task", taskInfo.GetName(), "with command", taskInfo.Command.GetValue())
	fmt.Println(taskInfo)

	runStatus := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_RUNNING.Enum(),
	}
	_, err := driver.SendStatusUpdate(runStatus)
	if err != nil {
		fmt.Println("Got error", err)
	}

	exec.tasksLaunched++
	fmt.Println("Total tasks launched ", exec.tasksLaunched)
	//
	// this is where one would perform the requested task
	//
	finishTask := func() {
		// finish task
		fmt.Println("Finishing task", taskInfo.GetName())
		finStatus := &mesos.TaskStatus{
			TaskId: taskInfo.GetTaskId(),
			State:  mesos.TaskState_TASK_FINISHED.Enum(),
		}
		if _, err := driver.SendStatusUpdate(finStatus); err != nil {
			fmt.Println("error sending FINISHED", err)
		}
		fmt.Println("Task finished", taskInfo.GetName())
	}

	starting := &mesos.TaskStatus{
		TaskId: taskInfo.GetTaskId(),
		State:  mesos.TaskState_TASK_STARTING.Enum(),
	}
	if _, err := driver.SendStatusUpdate(starting); err != nil {
		fmt.Println("error sending STARTING", err)
	}
	go func() {
		running := &mesos.TaskStatus{
			TaskId: taskInfo.GetTaskId(),
			State:  mesos.TaskState_TASK_RUNNING.Enum(),
		}
		if _, err := driver.SendStatusUpdate(running); err != nil {
			fmt.Println("error sending RUNNING", err)
		}
		for {
			select {
			case <-time.Tick(5 * time.Second):
				fmt.Println("In loop", taskInfo.String)
				exec.Log(graph.GraphID{}, graph.AlgorithmID{}, taskInfo.GetTaskId().String(), "Hello")
			}
		}
		finishTask()
	}()
}