Merge pull request #105 from chungjin/release

rebase develop branch to release branch

Author: chungjin

config/config.go

@@ -59,6 +59,7 @@ const (
 	HTTP_TIMEOUT                         = "launchtask.httptimeout"
 	COMPOSE_STOP_TIMEOUT                 = "cleanpod.timeout"
 	CONFIG_OVERRIDE_PREFIX               = "config."
+	monitorName                          = "podMonitor.monitorName"
 )
 
 // Read from default configuration file and set config as key/values
@@ -135,6 +136,7 @@ func setDefaultConfig(conf *viper.Viper) {
 	conf.SetDefault(TIMEOUT, "500s")
 	conf.SetDefault(COMPOSE_STOP_TIMEOUT, 10)
 	conf.SetDefault(HTTP_TIMEOUT, "20s")
+	conf.SetDefault(monitorName, "default")
 }
 
 func GetAppFolder() string {

config/config.yaml

@@ -28,3 +28,5 @@ dockerdump:
    containerpidfile: /run/docker/libcontainerd/docker-containerd.pid
    dockerlogpath: /var/log/upstart/docker.log
 dockercomposeverbose: false
+podMonitor:
+   monitorName: default

dce/main.go

@@ -31,11 +31,11 @@ import (
 
 	"github.com/paypal/dce-go/config"
 	"github.com/paypal/dce-go/dce/monitor"
+	_ "github.com/paypal/dce-go/dce/monitor/plugin/default"
 	"github.com/paypal/dce-go/plugin"
 	_ "github.com/paypal/dce-go/pluginimpl/example"
 	_ "github.com/paypal/dce-go/pluginimpl/general"
 	"github.com/paypal/dce-go/types"
-	"github.com/paypal/dce-go/utils"
 	fileUtils "github.com/paypal/dce-go/utils/file"
 	"github.com/paypal/dce-go/utils/pod"
 	"github.com/paypal/dce-go/utils/wait"
@@ -71,7 +71,8 @@ func (exec *dockerComposeExecutor) Disconnected(exec.ExecutorDriver) {
 }
 
 func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskInfo *mesos.TaskInfo) {
-	log.SetOutput(config.CreateFileAppendMode(types.DCE_OUT))
+	ctx := context.Background()
+	initlogger()
 	appStartTime := time.Now()
 
 	log.Println("====================Mesos LaunchTask====================")
@@ -104,7 +105,7 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 	pod.SetPodStatus(types.POD_STARTING)
 
 	// Update mesos state TO STARTING
-	pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_STARTING.Enum())
+	pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_STARTING.Enum())
 
 	// Get required compose file list
 	pod.ComposeFiles, _ = fileUtils.GetFiles(taskInfo)
@@ -120,13 +121,11 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 
 	// Create context with timeout
 	// Wait for pod launching until timeout
-	var ctx context.Context
 
 	var cancel context.CancelFunc
-	ctx = context.Background()
 	ctx, cancel = context.WithTimeout(ctx, config.GetLaunchTimeout())
 
-	go pod.WaitOnPod(&ctx)
+	go pod.WaitOnPod(ctx)
 
 	// Get order of plugins from config or mesos labels
 	pluginOrder, err := fileUtils.GetPluginOrder(taskInfo)
@@ -141,50 +140,50 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 	extpoints = plugin.GetOrderedExtpoints(pluginOrder)
 
 	// Executing LaunchTaskPreImagePull in order
-	if _, err := utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+	if _, err := pod.PluginPanicHandler(pod.ConditionFunc(func() (string, error) {
 		for i, ext := range extpoints {
 
 			if ext == nil {
 				logger.Errorln("Error getting plugins from plugin registration pools")
 				return "", errors.New("plugin is nil")
 			}
 			granularMetricStepName := fmt.Sprintf("%s_LaunchTaskPreImagePull", ext.Name())
-			utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, granularMetricStepName, "Starting")
+			pod.StartStep(pod.StepMetrics, granularMetricStepName)
 
-			err = ext.LaunchTaskPreImagePull(&ctx, &pod.ComposeFiles, executorId, taskInfo)
+			err = ext.LaunchTaskPreImagePull(ctx, &pod.ComposeFiles, executorId, taskInfo)
 			if err != nil {
 				logger.Errorf("Error executing LaunchTaskPreImagePull of plugin : %v", err)
-				utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Error")
+				pod.EndStep(pod.StepMetrics, granularMetricStepName, nil, err)
 				return "", err
 			}
-			utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Success")
+			pod.EndStep(pod.StepMetrics, granularMetricStepName, nil, nil)
 
 			if config.EnableComposeTrace() {
-				fileUtils.DumpPluginModifiedComposeFiles(ctx, pluginOrder[i], "LaunchTaskPreImagePull", i)
+				fileUtils.DumpPluginModifiedComposeFiles(pluginOrder[i], "LaunchTaskPreImagePull", i)
 			}
 		}
 		return "", err
 	})); err != nil {
 		logger.Errorf("error while executing task pre image pull: %s", err)
 		pod.SetPodStatus(types.POD_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 		return
 	}
 	// Write updated compose files into pod folder
-	err = fileUtils.WriteChangeToFiles(ctx)
+	err = fileUtils.WriteChangeToFiles()
 	if err != nil {
 		logger.Errorf("Failure writing updated compose files : %v", err)
 		pod.SetPodStatus(types.POD_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 	}
 
 	//Validate Compose files
 	if err := validateComposeFiles(); err != nil {
 		pod.SetPodStatus(types.POD_COMPOSE_CHECK_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 		return
 	}
 
@@ -193,98 +192,92 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 	if err != nil {
 		pod.SetPodStatus(types.POD_PULL_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 		return
 	}
 
 	timeElapsed := time.Since(appStartTime)
 	logger.Printf("Time elapsed since App launch: %.3fs", timeElapsed.Seconds())
 
 	// Executing LaunchTaskPostImagePull in order
-	if _, err := utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+	if _, err := pod.PluginPanicHandler(pod.ConditionFunc(func() (string, error) {
 		for i, ext := range extpoints {
 			if ext == nil {
 				logger.Errorln("Error getting plugins from plugin registration pools")
 				return "", errors.New("plugin is nil")
 			}
 			granularMetricStepName := fmt.Sprintf("%s_LaunchTaskPostImagePull", ext.Name())
-			utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, granularMetricStepName, "Starting")
+			pod.StartStep(pod.StepMetrics, granularMetricStepName)
 
-			err = ext.LaunchTaskPostImagePull(&ctx, &pod.ComposeFiles, executorId, taskInfo)
+			err = ext.LaunchTaskPostImagePull(ctx, &pod.ComposeFiles, executorId, taskInfo)
 			if err != nil {
 				logger.Errorf("Error executing LaunchTaskPreImagePull of plugin : %v", err)
-				utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Error")
+				pod.EndStep(pod.StepMetrics, granularMetricStepName, nil, err)
 				return "", err
 			}
 
-			utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Success")
+			pod.EndStep(pod.StepMetrics, granularMetricStepName, nil, nil)
 
 			if config.EnableComposeTrace() {
-				fileUtils.DumpPluginModifiedComposeFiles(ctx, pluginOrder[i], "LaunchTaskPostImagePull", i)
+				fileUtils.DumpPluginModifiedComposeFiles(pluginOrder[i], "LaunchTaskPostImagePull", i)
 			}
 		}
 		return "", err
 	})); err != nil {
 		pod.SetPodStatus(types.POD_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 		return
 	}
 
 	// Service list from all compose files
-	podServices := getServices(ctx)
+	podServices := getServices()
 	logger.Printf("pod service list: %v", podServices)
 
 	// Write updated compose files into pod folder
-	err = fileUtils.WriteChangeToFiles(ctx)
+	err = fileUtils.WriteChangeToFiles()
 	if err != nil {
 		logger.Errorf("Failure writing updated compose files : %v", err)
 		pod.SetPodStatus(types.POD_FAILED)
 		cancel()
-		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		pod.SendMesosStatus(ctx, driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 	}
 
-	utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, "Launch_Pod", "Starting")
+	pod.StartStep(pod.StepMetrics, "Launch_Pod")
 
 	// Launch pod
 	replyPodStatus, err := pod.LaunchPod(pod.ComposeFiles)
-	if err != nil {
-		utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), "Launch_Pod", "Error")
-	} else {
-		utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), "Launch_Pod", "Success")
-	}
+	pod.EndStep(pod.StepMetrics, "Launch_Pod", nil, err)
 
 	logger.Printf("Pod status returned by LaunchPod : %s", replyPodStatus.String())
 
 	// Take an action depends on different status
 	switch replyPodStatus {
 	case types.POD_FAILED:
 		cancel()
-		pod.SendPodStatus(types.POD_FAILED)
+		pod.SendPodStatus(ctx, types.POD_FAILED)
 
 	case types.POD_STARTING:
 		// Initial health check
 		res, err := initHealthCheck(podServices)
 		if err != nil || res == types.POD_FAILED {
 			cancel()
-			pod.SendPodStatus(types.POD_FAILED)
+			pod.SendPodStatus(ctx, types.POD_FAILED)
 		}
 
 		// Temp status keeps the pod status returned by PostLaunchTask
-		tempStatus, err := utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+		tempStatus, err := pod.PluginPanicHandler(pod.ConditionFunc(func() (string, error) {
 			var tempStatus string
 			for _, ext := range extpoints {
 				logger.Println("Executing post launch task plugin")
 
 				granularMetricStepName := fmt.Sprintf("%s_PostLaunchTask", ext.Name())
-				utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, granularMetricStepName, "Starting")
+				pod.StartStep(pod.StepMetrics, granularMetricStepName)
 
-				tempStatus, err = ext.PostLaunchTask(&ctx, pod.ComposeFiles, taskInfo)
+				tempStatus, err = ext.PostLaunchTask(ctx, pod.ComposeFiles, taskInfo)
+				pod.EndStep(pod.StepMetrics, granularMetricStepName, nil, err)
 				if err != nil {
 					logger.Errorf("Error executing PostLaunchTask : %v", err)
-					utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Error")
-				} else {
-					utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Success")
 				}
 
 				logger.Printf("Get pod status : %s returned by PostLaunchTask", tempStatus)
@@ -300,20 +293,26 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 		}
 		if tempStatus == types.POD_FAILED.String() {
 			cancel()
-			pod.SendPodStatus(types.POD_FAILED)
+			pod.SendPodStatus(ctx, types.POD_FAILED)
 			return
 		}
 		if res == types.POD_RUNNING {
 			cancel()
 			if pod.GetPodStatus() != types.POD_RUNNING {
-				pod.SendPodStatus(types.POD_RUNNING)
-				go monitor.MonitorPoller()
+				pod.SendPodStatus(ctx, types.POD_RUNNING)
+				go func() {
+					status, err := monitor.MonitorPoller(ctx)
+					if err != nil {
+						log.Errorf("failure from monitor: %s", err)
+					}
+					pod.SendPodStatus(ctx, status)
+				}()
 			}
 		}
 		//For adhoc job, send finished to mesos if job already finished during init health check
 		if res == types.POD_FINISHED {
 			cancel()
-			pod.SendPodStatus(types.POD_FINISHED)
+			pod.SendPodStatus(ctx, types.POD_FINISHED)
 		}
 
 	default:
@@ -324,6 +323,7 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 }
 
 func (exec *dockerComposeExecutor) KillTask(driver exec.ExecutorDriver, taskId *mesos.TaskID) {
+	ctx := context.Background()
 	log.Println("====================Mesos KillTask====================")
 
 	defer func() {
@@ -345,12 +345,12 @@ func (exec *dockerComposeExecutor) KillTask(driver exec.ExecutorDriver, taskId *
 		logKill.Printf("Mesos Kill Task : Current task status is %s , continue killTask", status)
 		pod.SetPodStatus(types.POD_KILLED)
 
-		err := pod.StopPod(pod.ComposeFiles)
+		err := pod.StopPod(ctx, pod.ComposeFiles)
 		if err != nil {
 			logKill.Errorf("Error cleaning up pod : %v", err.Error())
 		}
 
-		err = pod.SendMesosStatus(driver, taskId, mesos.TaskState_TASK_KILLED.Enum())
+		err = pod.SendMesosStatus(ctx, driver, taskId, mesos.TaskState_TASK_KILLED.Enum())
 		if err != nil {
 			logKill.Errorf("Error during kill Task : %v", err.Error())
 		}
@@ -370,7 +370,7 @@ func (exec *dockerComposeExecutor) FrameworkMessage(driver exec.ExecutorDriver,
 func (exec *dockerComposeExecutor) Shutdown(driver exec.ExecutorDriver) {
 	// Execute shutdown plugin extensions in order
 	for _, ext := range extpoints {
-		ext.Shutdown(pod.ComposeExecutorDriver)
+		ext.Shutdown(pod.ComposeTaskInfo, pod.ComposeExecutorDriver)
 	}
 	log.Println("====================Stop ExecutorDriver====================")
 	driver.Stop()
@@ -395,18 +395,11 @@ func pullImage() error {
 	logger.Println("====================Pulling Image====================")
 
 	if !config.SkipPullImages() {
-		count := 0
 		err := wait.PollRetry(config.GetPullRetryCount(), config.GetPollInterval(), func() (string, error) {
-			utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, fmt.Sprintf("Image_Pull_%v", count), "Starting")
+			pod.StartStep(pod.StepMetrics, "Image_Pull")
 			err := pod.PullImage(pod.ComposeFiles)
-			if err != nil {
-				utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), fmt.Sprintf("Image_Pull_%v", count), "Error")
-			} else {
-				utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), fmt.Sprintf("Image_Pull_%v", count), "Success")
-			}
-			count++
+			pod.EndStep(pod.StepMetrics, "Image_Pull", nil, err)
 			return "", err
-
 		})
 
 		if err != nil {
@@ -420,19 +413,21 @@ func pullImage() error {
 
 func initHealthCheck(podServices map[string]bool) (types.PodStatus, error) {
 	res, err := wait.WaitUntil(config.GetLaunchTimeout(), func(healthCheckReply chan string) {
+		// wait until timeout or receive any result from healthCheckReply
+		// healthCheckReply is to stored the pod status
 		pod.HealthCheck(pod.ComposeFiles, podServices, healthCheckReply)
 	})
 
 	if err != nil {
 		log.Printf("POD_INIT_HEALTH_CHECK_TIMEOUT -- %v", err)
 		return types.POD_FAILED, err
 	}
-	return utils.ToPodStatus(res), err
+	return pod.ToPodStatus(res), err
 }
 
-func getServices(ctx context.Context) map[string]bool {
+func getServices() map[string]bool {
+	filesMap := pod.GetServiceDetail()
 	podService := make(map[string]bool)
-	filesMap := ctx.Value(types.SERVICE_DETAIL).(types.ServiceDetail)
 
 	for _, file := range pod.ComposeFiles {
 		servMap := filesMap[file][types.SERVICES].(map[interface{}]interface{})
@@ -507,3 +502,15 @@ func switchDebugMode() {
 		log.SetLevel(log.DebugLevel)
 	}
 }
+
+// redirect the output, and set the loglevel
+func initlogger() {
+	log.SetOutput(config.CreateFileAppendMode(types.DCE_OUT))
+	loglevel := config.GetConfig().GetString(types.LOGLEVEL)
+	ll, err := log.ParseLevel(loglevel)
+	if err == nil {
+		log.SetLevel(ll)
+	} else {
+		log.SetLevel(log.InfoLevel)
+	}
+}

dce/monitor/monitor.go

@@ -16,142 +16,27 @@
 package monitor
 
 import (
-	"fmt"
+	"context"
 
 	"github.com/paypal/dce-go/config"
+	"github.com/paypal/dce-go/plugin"
 	"github.com/paypal/dce-go/types"
-	"github.com/paypal/dce-go/utils"
-	"github.com/paypal/dce-go/utils/pod"
-	"github.com/paypal/dce-go/utils/wait"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 )
 
-// Watching pod status and notifying executor if any container in the pod goes wrong
-func podMonitor(systemProxyId string) types.PodStatus {
-	logger := log.WithFields(log.Fields{
-		"func": "monitor.podMonitor",
-	})
-
-	var err error
-
-	for i := 0; i < len(pod.MonitorContainerList); i++ {
-		var healthy types.HealthStatus
-		var exitCode int
-		var running bool
-
-		if hc, ok := pod.HealthCheckListId[pod.MonitorContainerList[i]]; ok && hc {
-			healthy, running, exitCode, err = pod.CheckContainer(pod.MonitorContainerList[i], true)
-			logger.Debugf("container %s has health check, health status: %s, exitCode: %d, err : %v",
-				pod.MonitorContainerList[i], healthy.String(), exitCode, err)
-		} else {
-			healthy, running, exitCode, err = pod.CheckContainer(pod.MonitorContainerList[i], false)
-			log.Debugf("container %s doesn't have health check, status: %s, exitCode: %d, err : %v",
-				pod.MonitorContainerList[i], healthy.String(), exitCode, err)
-		}
-
-		if err != nil {
-			logger.Errorf(fmt.Sprintf("POD_MONITOR_HEALTH_CHECK_FAILED -- Error inspecting container with id : %s, %v", pod.MonitorContainerList[i], err.Error()))
-			logger.Errorln("POD_MONITOR_FAILED -- Send Failed")
-			return types.POD_FAILED
-		}
-
-		if exitCode != 0 {
-			logger.Println("POD_MONITOR_APP_EXIT -- Stop pod monitor and send Failed")
-			return types.POD_FAILED
-		}
-
-		if healthy == types.UNHEALTHY {
-			err = pod.PrintInspectDetail(pod.MonitorContainerList[i])
-			if err != nil {
-				log.Warnf("Error during docker inspect: ", err)
-			}
-			if config.GetConfigSection(config.CLEANPOD) == nil ||
-				config.GetConfigSection(config.CLEANPOD)[types.UNHEALTHY.String()] == "true" {
-				logger.Println("POD_MONITOR_HEALTH_CHECK_FAILED -- Stop pod monitor and send Failed")
-				return types.POD_FAILED
-			}
-			logger.Warnf("Container %s became unhealthy, but pod won't be killed due to cleanpod config", pod.MonitorContainerList[i])
-		}
-
-		if exitCode == 0 && !running {
-			logger.Printf("Removed finished(exit with 0) container %s from monitor list", pod.MonitorContainerList[i])
-			pod.MonitorContainerList = append(pod.MonitorContainerList[:i], pod.MonitorContainerList[i+1:]...)
-			i--
-
-		}
-	}
-
-	// Send finished to mesos IF no container running or ONLY system proxy is running in the pod
-	isService := config.IsService()
-	if len(pod.MonitorContainerList) == 0 && !isService {
-		logger.Println("Task is ADHOC job. All containers in the pod exit with code 0, sending FINISHED")
-		return types.POD_FINISHED
-	}
-
-	if len(pod.MonitorContainerList) == 0 {
-		logger.Println("Task is SERVICE. All containers in the pod exit with code 0, sending FAILED")
-		return types.POD_FAILED
-	}
-
-	if len(pod.MonitorContainerList) == 1 && pod.MonitorContainerList[0] == systemProxyId && !isService {
-		logger.Println("Task is ADHOC job. Only infra container is running in the pod, sending FINISHED")
-		return types.POD_FINISHED
-	}
-
-	if len(pod.MonitorContainerList) == 1 && pod.MonitorContainerList[0] == systemProxyId {
-		logger.Println("Task is SERVICE. Only infra container is running in the pod, sending FAILED")
-		return types.POD_FAILED
-	}
-
-	return types.POD_EMPTY
-}
-
-// Polling pod monitor periodically
-func MonitorPoller() {
+// MonitorPoller inspects pod status periodically
+func MonitorPoller(ctx context.Context) (types.PodStatus, error) {
 	logger := log.WithFields(log.Fields{
 		"func": "monitor.MonitorPoller",
 	})
 
 	logger.Println("====================Pod Monitor Poller====================")
 
-	// Get infra container id
-	var infraContainerId string
-	var err error
-	if !config.GetConfig().GetBool(types.RM_INFRA_CONTAINER) {
-		infraContainerId, err = pod.GetContainerIdByService(pod.ComposeFiles, types.INFRA_CONTAINER)
-		if err != nil {
-			logger.Errorf("Error getting container id of service %s: %v", types.INFRA_CONTAINER, err)
-			logger.Errorln("POD_MONITOR_FAILED -- Send Failed")
-			pod.SendPodStatus(types.POD_FAILED)
-			return
-		}
-		logger.Printf("Infra container id: %s", infraContainerId)
-	}
-
-	res, err := wait.PollForever(config.GetPollInterval(), nil, func() (string, error) {
-		return podMonitor(infraContainerId).String(), nil
-	})
-
-	logger.Printf("Pod Monitor Receiver : Received  message %s", res)
-
-	curPodStatus := pod.GetPodStatus()
-	// Once mesos send admin kill to the pod, monitor also will find containers exist
-	// have following check to avoid sending redundant or invalid status
-	if curPodStatus == types.POD_KILLED || curPodStatus == types.POD_FAILED {
-		logger.Println("====================Pod Monitor Stopped====================")
-		return
-	}
-
-	if err != nil {
-		pod.SendPodStatus(types.POD_FAILED)
-		return
-	}
-
-	switch utils.ToPodStatus(res) {
-	case types.POD_FAILED:
-		pod.SendPodStatus(types.POD_FAILED)
-
-	case types.POD_FINISHED:
-		pod.SendPodStatus(types.POD_FINISHED)
+	name := config.GetConfig().GetString("podMonitor.monitorName")
+	monitor := plugin.Monitors.Lookup(name)
+	if monitor == nil {
+		return types.POD_FAILED, errors.Errorf("monitor plugin %s doesn't exist", name)
 	}
+	return monitor.Start(ctx)
 }

dce/monitor/plugin/default/monitor.go

@@ -0,0 +1,106 @@
+package _default
+
+import (
+	"context"
+
+	"github.com/paypal/dce-go/config"
+	"github.com/paypal/dce-go/plugin"
+	"github.com/paypal/dce-go/types"
+	"github.com/paypal/dce-go/utils/pod"
+	"github.com/paypal/dce-go/utils/wait"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+const name = "default"
+
+type monitor struct{}
+
+func init() {
+	// Register default monitor plugin
+	log.SetOutput(config.CreateFileAppendMode(types.DCE_OUT))
+	plugin.Monitors.Register(&monitor{}, name)
+	log.Infof("Registered monitor plugin %s", name)
+}
+
+func (m *monitor) Start(ctx context.Context) (types.PodStatus, error) {
+	logger := log.WithFields(log.Fields{
+		"monitor": name,
+	})
+	// Get infra container ID
+	var infraContainerId string
+	var err error
+	if !config.GetConfig().GetBool(types.RM_INFRA_CONTAINER) {
+		infraContainerId, err = pod.GetContainerIdByService(pod.ComposeFiles, types.INFRA_CONTAINER)
+		if err != nil {
+			return types.POD_FAILED, errors.Wrap(err, "fail to get infra container ID")
+		}
+		logger.Debugf("Infra container ID: %s", infraContainerId)
+	}
+
+	run := func() (types.PodStatus, error) {
+		for i := 0; i < len(pod.MonitorContainerList); i++ {
+			hc, ok := pod.HealthCheckListId[pod.MonitorContainerList[i]]
+			_, healthy, running, exitCode, err := pod.CheckContainer(pod.MonitorContainerList[i], ok && hc)
+			if err != nil {
+				return types.POD_FAILED, err
+			}
+			logger.Debugf("container %s has health check, health status: %s, exitCode: %d, err : %v",
+				pod.MonitorContainerList[i], healthy.String(), exitCode, err)
+
+			if exitCode != 0 {
+				return types.POD_FAILED, nil
+			}
+
+			if healthy == types.UNHEALTHY {
+				err = pod.PrintInspectDetail(pod.MonitorContainerList[i])
+				if err != nil {
+					log.Warnf("failed to get container detail: %s ", err)
+				}
+				return types.POD_FAILED, nil
+			}
+
+			if exitCode == 0 && !running {
+				logger.Infof("Removed finished(exit with 0) container %s from monitor list",
+					pod.MonitorContainerList[i])
+				pod.MonitorContainerList = append(pod.MonitorContainerList[:i], pod.MonitorContainerList[i+1:]...)
+				i--
+
+			}
+		}
+
+		// Send finished to mesos IF no container running or ONLY system proxy is running in the pod
+		switch config.IsService() {
+		case true:
+			if len(pod.MonitorContainerList) == 0 {
+				logger.Error("Task is SERVICE. All containers in the pod exit with code 0, sending FAILED")
+				return types.POD_FAILED, nil
+			}
+			if len(pod.MonitorContainerList) == 1 && pod.MonitorContainerList[0] == infraContainerId {
+				logger.Error("Task is SERVICE. Only infra container is running in the pod, sending FAILED")
+				return types.POD_FAILED, nil
+			}
+		case false:
+			if len(pod.MonitorContainerList) == 0 {
+				logger.Info("Task is ADHOC job. All containers in the pod exit with code 0, sending FINISHED")
+				return types.POD_FINISHED, nil
+			}
+			if len(pod.MonitorContainerList) == 1 && pod.MonitorContainerList[0] == infraContainerId {
+				logger.Info("Task is ADHOC job. Only infra container is running in the pod, sending FINISHED")
+				return types.POD_FINISHED, nil
+			}
+		}
+		return types.POD_EMPTY, nil
+	}
+
+	res, err := wait.PollForever(config.GetPollInterval(), nil, func() (string, error) {
+		status, err := run()
+		if err != nil {
+			// Error won't be considered as pod failure unless pod status is failed
+			log.Warnf("error from monitor periodical check: %s", err)
+		}
+		return status.String(), nil
+	})
+
+	return pod.ToPodStatus(res), err
+}

docs/how-to-develop.md

@@ -65,16 +65,16 @@ type ComposePlugin interface {
        PreLaunchTask(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error
        
        // PostLaunchTask is invoked after pod launch in context of executor LaunchTask callback.
-       PostLaunchTask(ctx *context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error)
+       PostLaunchTask(ctx context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error)
        
        // PreKillTask is invoked prior to killing pod in context of executor KillTask callback. 
-       PreKillTask(taskInfo *mesos.TaskInfo) error
+       PreKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error
        
        // PostKillTask is invoked after killing pod in context of executor KillTask callback. 
-       PostKillTask(taskInfo *mesos.TaskInfo) error
+       PostKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error
        
        // Shutdown is invoked prior to executor shutdown in context of Shutdown callback. 
-       Shutdown(executor.ExecutorDriver) error
+       Shutdown(taskInfo *mesos.TaskInfo, ed executor.ExecutorDriver)
 }
 ```
 PreLaunchTask and PostLaunchTask have Context object as first parameter. This is used to pass around parsed compose files so as to avoid loading from files by individual plugins.
@@ -112,22 +112,22 @@ func (ex *exampleExt) PreLaunchTask(ctx *context.Context, composeFiles *[]string
 	return nil
 }
 
-func (ex *exampleExt) PostLaunchTask(ctx *context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error) {
+func (ex *exampleExt) PostLaunchTask(ctx context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error) {
 	logger.Println("PostLaunchTask Starting")
 	return "", nil
 }
 
-func (ex *exampleExt) PreKillTask(taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) PreKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PreKillTask Starting")
 	return nil
 }
 
-func (ex *exampleExt) PostKillTask(taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) PostKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PostKillTask Starting")
 	return nil
 }
 
-func (ex *exampleExt) Shutdown(executor.ExecutorDriver) error {
+func (ex *exampleExt) Shutdown(taskInfo *mesos.TaskInfo, ed executor.ExecutorDriver) {
 	logger.Println("Shutdown Starting")
 	return nil
 }

go.mod

@@ -31,7 +31,6 @@ require (
 	gopkg.in/ini.v1 v1.57.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 // indirect
-
 )
 
 replace (

go.sum

@@ -11,16 +11,12 @@ cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqCl
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-git.apache.org/thrift.git v0.14.0 h1:oOfDeaoEWsWForcJbGRqiJXbjNu9DKQaQvyH9rFoj4U=
-git.apache.org/thrift.git v0.14.0/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/apache/thrift v0.14.0 h1:vqZ2DP42i8th2OsgCcYZkirtbzvpZEFx53LiWDJXIAs=
-github.com/apache/thrift v0.14.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
@@ -148,8 +144,6 @@ github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/paypal/gorealis v1.1.0 h1:MDjA+vuHu8FlBWVFb3gcn6nhP+kFMDEIsSmZRV6Q8a4=
 github.com/paypal/gorealis v1.1.0/go.mod h1:uWjenketTx0csB8smzIsZk1cEhDGHnq1LLq0+fXsMUk=
-github.com/paypal/gorealis v1.23.0 h1:xIQFNjS7+pf1JKCgk4lLXNNK89S2GGoiaZrdz9lsOR8=
-github.com/paypal/gorealis v1.23.0/go.mod h1:+XWLwhxwZ47Vld8fH3oppB9LGFuQO/0CdMM0chVHvN4=
 github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
@@ -176,7 +170,6 @@ github.com/rdelval/thrift v0.0.0-20161221203622-b2a4d4ae21c7/go.mod h1:6lRYofZ1V
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
-github.com/samuel/go-zookeeper v0.0.0-20171117190445-471cd4e61d7a/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/samuel/go-zookeeper v0.0.0-20200724154423-2164a8ac840e h1:CGjiMQ0wMH4wtNWrlj6kiTbkPt2F3rbYnhGX6TWLfco=
 github.com/samuel/go-zookeeper v0.0.0-20200724154423-2164a8ac840e/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
@@ -207,7 +200,6 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
-github.com/stretchr/testify v1.2.0/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=

plugin/extpoints.go

@@ -55,6 +55,7 @@ func UnregisterExtension(name string) []string {
 	return ifaces
 }
 
+
 // Base extension point
 
 type extensionPoint struct {
@@ -176,6 +177,7 @@ func (ep *composePluginExt) Names() []string {
 	return names
 }
 
+
 // PodStatusHook
 
 var PodStatusHooks = &podStatusHookExt{
@@ -225,3 +227,56 @@ func (ep *podStatusHookExt) Names() []string {
 	}
 	return names
 }
+
+
+// Monitor
+
+var Monitors = &monitorExt{
+	newExtensionPoint(new(Monitor)),
+}
+
+type monitorExt struct {
+	*extensionPoint
+}
+
+func (ep *monitorExt) Unregister(name string) bool {
+	return ep.unregister(name)
+}
+
+func (ep *monitorExt) Register(extension Monitor, name string) bool {
+	return ep.register(extension, name)
+}
+
+func (ep *monitorExt) Lookup(name string) Monitor {
+	ext := ep.lookup(name)
+	if ext == nil {
+		return nil
+	}
+	return ext.(Monitor)
+}
+
+func (ep *monitorExt) Select(names []string) []Monitor {
+	var selected []Monitor
+	for _, name := range names {
+		selected = append(selected, ep.Lookup(name))
+	}
+	return selected
+}
+
+func (ep *monitorExt) All() map[string]Monitor {
+	all := make(map[string]Monitor)
+	for k, v := range ep.all() {
+		all[k] = v.(Monitor)
+	}
+	return all
+}
+
+func (ep *monitorExt) Names() []string {
+	var names []string
+	for k := range ep.all() {
+		names = append(names, k)
+	}
+	return names
+}
+
+

plugin/type.go

@@ -12,37 +12,38 @@
  * limitations under the License.
  */
 
-//go:generate go-extpoints . ComposePlugin PodStatusHook
+//go:generate go-extpoints . ComposePlugin PodStatusHook Monitor
 package plugin
 
 import (
 	"context"
 
 	"github.com/mesos/mesos-go/executor"
 	mesos "github.com/mesos/mesos-go/mesosproto"
+	"github.com/paypal/dce-go/types"
 )
 
 type ComposePlugin interface {
-	// Get the name of the plugin
+	// Name gets the name of the plugin
 	Name() string
 
 	// execute some tasks before the Image is pulled
-	LaunchTaskPreImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error
+	LaunchTaskPreImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error
 
 	// execute some tasks after the Image is pulled
-	LaunchTaskPostImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error
+	LaunchTaskPostImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error
 
 	// execute the tasks after the pod is launched
-	PostLaunchTask(ctx *context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error)
+	PostLaunchTask(ctx context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error)
 
 	// execute the task before we send a Kill to Mesos
-	PreKillTask(taskInfo *mesos.TaskInfo) error
+	PreKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error
 
 	// execute the task after we send a Kill to Mesos
-	PostKillTask(taskInfo *mesos.TaskInfo) error
+	PostKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error
 
 	// execute the task to shutdown the pod
-	Shutdown(executor.ExecutorDriver) error
+	Shutdown(taskInfo *mesos.TaskInfo, ed executor.ExecutorDriver) error
 }
 
 // PodStatusHook allows custom implementations to be plugged when a Pod (mesos task) status changes. Currently this is
@@ -51,5 +52,12 @@ type PodStatusHook interface {
 	// Execute is invoked when the pod.taskStatusCh channel has a new status. It returns an error on failure,
 	// and also a flag "failExec" indicating if the error needs to fail the execution when a series of hooks are executed
 	// This is to support cases where a few hooks can be executed in a best effort manner and need not fail the executor
-	Execute(podStatus string, data interface{}) (failExec bool, err error)
+	Execute(ctx context.Context, podStatus string, data interface{}) (failExec bool, err error)
+}
+
+// Monitor inspects pods periodically until pod failed or terminated It also defines when to consider a pod as failed.
+// Move monitor as a plugin provides flexibility to replace default monitor logic.
+// Monitor name presents in config `monitorName` will be used, otherwise, default monitor will be used.
+type Monitor interface {
+	Start(ctx context.Context) (types.PodStatus, error)
 }

pluginimpl/example/impl.go

@@ -22,6 +22,7 @@ import (
 	"github.com/paypal/dce-go/config"
 	"github.com/paypal/dce-go/plugin"
 	"github.com/paypal/dce-go/types"
+	"github.com/paypal/dce-go/utils/pod"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -45,14 +46,14 @@ func (p *exampleExt) Name() string {
 	return "example"
 }
 
-func (ex *exampleExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) LaunchTaskPreImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
 	logger.Println("LaunchTaskPreImagePull begin")
 	// docker compose YML files are saved in context as type SERVICE_DETAIL which is map[interface{}]interface{}.
 	// Massage YML files and save it in context.
 	// Then pass to next plugin.
 
 	// Get value from context
-	filesMap := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
+	filesMap := pod.GetServiceDetail()
 
 	// Add label in each service, in each compose YML file
 	for _, file := range *composeFiles {
@@ -69,17 +70,17 @@ func (ex *exampleExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	}
 
 	// Save the changes back to context
-	*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, filesMap)
+	pod.SetServiceDetail(filesMap)
 
 	return nil
 }
 
-func (ex *exampleExt) LaunchTaskPostImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) LaunchTaskPostImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
 	logger.Println("LaunchTaskPostImagePull begin")
 	return nil
 }
 
-func (ex *exampleExt) PostLaunchTask(ctx *context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error) {
+func (ex *exampleExt) PostLaunchTask(ctx context.Context, composeFiles []string, taskInfo *mesos.TaskInfo) (string, error) {
 	logger.Println("PostLaunchTask begin")
 	return "", nil
 }
@@ -89,17 +90,17 @@ func (ex *exampleExt) PreStopPod() error {
 	return nil
 }
 
-func (ex *exampleExt) PreKillTask(taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) PreKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PreKillTask begin")
 	return nil
 }
 
-func (ex *exampleExt) PostKillTask(taskInfo *mesos.TaskInfo) error {
+func (ex *exampleExt) PostKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PostKillTask begin")
 	return nil
 }
 
-func (ex *exampleExt) Shutdown(executor.ExecutorDriver) error {
+func (ex *exampleExt) Shutdown(taskInfo *mesos.TaskInfo, ed executor.ExecutorDriver) error {
 	logger.Println("Shutdown begin")
 	return nil
 }

pluginimpl/general/editor.go

@@ -38,11 +38,11 @@ const (
 	DEFAULT_VERSION = "2.1"
 )
 
-func editComposeFile(ctx *context.Context, file string, executorId string, taskId string, ports *list.Element,
+func editComposeFile(file string, executorId string, taskId string, ports *list.Element,
 	extraHosts map[interface{}]bool) (string, *list.Element, error) {
 	var err error
 
-	filesMap := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
+	filesMap := pod.GetServiceDetail()
 	if filesMap[file][types.SERVICES] == nil {
 		log.Printf("Services is empty for file %s \n", file)
 		return "", ports, nil
@@ -69,7 +69,7 @@ func editComposeFile(ctx *context.Context, file string, executorId string, taskI
 		delete(filesMap, file)
 		file = file + utils.FILE_POSTFIX
 	}
-	*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, filesMap)
+	pod.SetServiceDetail(filesMap)
 
 	logger.Printf("Updated compose files, current context: %v\n", filesMap)
 	return file, ports, err
@@ -240,9 +240,9 @@ func updateServiceSessions(serviceName, file, executorId, taskId string, filesMa
 	return ports, nil
 }
 
-func postEditComposeFile(ctx *context.Context, file string) error {
+func postEditComposeFile(file string) error {
 	var err error
-	filesMap := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
+	filesMap := pod.GetServiceDetail()
 	if filesMap[file][types.SERVICES] == nil {
 		return nil
 	}
@@ -254,9 +254,8 @@ func postEditComposeFile(ctx *context.Context, file string) error {
 			return err
 		}
 	}
-	*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, filesMap)
-
-	err = utils.WriteChangeToFiles(*ctx)
+	pod.SetServiceDetail(filesMap)
+	err = utils.WriteChangeToFiles()
 	if err != nil {
 		log.Errorf("Failure writing updated compose files : %v", err)
 		return err
@@ -304,12 +303,8 @@ func scanForExtraHostsSection(containerDetails map[interface{}]interface{}, extr
 	}
 }
 
-func addExtraHostsSection(ctx *context.Context, file, svcName string, extraHostsCollection map[interface{}]bool) {
-	filesMap, ok := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
-	if !ok {
-		log.Warnln("Couldn't get service detail")
-		return
-	}
+func addExtraHostsSection(ctx context.Context, file, svcName string, extraHostsCollection map[interface{}]bool) {
+	filesMap := pod.GetServiceDetail()
 	servMap, ok := filesMap[file][types.SERVICES].(map[interface{}]interface{})
 	if !ok {
 		log.Warnf("Couldn't get content of compose file %s\n", file)

pluginimpl/general/editor_test.go

@@ -48,12 +48,12 @@ func TestGenerateEditComposeFile(t *testing.T) {
 
 	var extraHosts = make(map[interface{}]bool)
 	ctx = context.WithValue(ctx, types.SERVICE_DETAIL, servDetail)
-	_, curPort, _ := editComposeFile(&ctx, "testdata/test.yml", "executorId", "taskId", ports.Front(), extraHosts)
+	_, curPort, _ := editComposeFile(ctx, "testdata/test.yml", "executorId", "taskId", ports.Front(), extraHosts)
 
 	if curPort == nil || strconv.FormatUint(curPort.Value.(uint64), 10) != "3000" {
 		t.Errorf("expected current port to be 3000 but got %v", curPort)
 	}
-	err = file.WriteChangeToFiles(ctx)
+	err = file.WriteChangeToFiles()
 	if err != nil {
 		t.Fatalf("Failed to write to files %v", err)
 	}
@@ -83,7 +83,7 @@ func Test_editComposeFile(t *testing.T) {
 	assert.Equal(t, nil, containerDetails[types.LABELS], "Before editing compose file")
 
 	var extraHosts = make(map[interface{}]bool)
-	_, curPort, _ := editComposeFile(&ctx, "testdata/test.yml", "executorId", "taskId", ports.Front(), extraHosts)
+	_, curPort, _ := editComposeFile(ctx, "testdata/test.yml", "executorId", "taskId", ports.Front(), extraHosts)
 
 	// After edit compose file
 	if curPort == nil || strconv.FormatUint(curPort.Value.(uint64), 10) != "3000" {
@@ -136,7 +136,7 @@ func Test_addExtraHostsSection(t *testing.T) {
 	containerDetail := make(map[interface{}]interface{})
 	containerDetail[types.EXTRA_HOSTS] = []interface{}{"redis2:0.0.0.2"}
 	scanForExtraHostsSection(containerDetail, extraHosts)
-	addExtraHostsSection(&ctx, "testdata/docker-extra-host.yml", "redis", extraHosts)
+	addExtraHostsSection(ctx, "testdata/docker-extra-host.yml", "redis", extraHosts)
 	filesMap, ok := ctx.Value(types.SERVICE_DETAIL).(types.ServiceDetail)
 	if !ok {
 		t.Error("Couldn't get service detail")
@@ -157,6 +157,6 @@ func Test_addExtraHostsSection(t *testing.T) {
 	}
 
 	preCtx := ctx
-	addExtraHostsSection(&ctx, "testdata/docker-extra-host.yml", "fake", extraHosts)
+	addExtraHostsSection(ctx, "testdata/docker-extra-host.yml", "fake", extraHosts)
 	assert.Equal(t, preCtx, ctx, "Adding extra host to non exist service")
 }

pluginimpl/general/impl.go

@@ -58,7 +58,7 @@ func (p *generalExt) Name() string {
 	return "general"
 }
 
-func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
+func (ge *generalExt) LaunchTaskPreImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
 	logger.Println("LaunchTaskPreImagePull begin")
 
 	if composeFiles == nil || len(*composeFiles) == 0 {
@@ -69,19 +69,19 @@ func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	var err error
 
 	logger.Println("====================context in====================")
-	logger.Println((*ctx).Value(types.SERVICE_DETAIL))
+	logger.Printf("SERVICE_DETAIL: %+v", pod.GetServiceDetail())
 
 	logger.Printf("Current compose files list: %v", *composeFiles)
 
-	if (*ctx).Value(types.SERVICE_DETAIL) == nil {
+	if len(pod.GetServiceDetail()) == 0 {
 		var servDetail types.ServiceDetail
 		servDetail, err = utils.ParseYamls(composeFiles)
 		if err != nil {
 			log.Errorf("Error parsing yaml files : %v", err)
 			return err
 		}
 
-		*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, servDetail)
+		pod.SetServiceDetail(servDetail)
 	}
 
 	currentPort := pod.GetPorts(taskInfo)
@@ -100,7 +100,7 @@ func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	for i, file := range *composeFiles {
 		logger.Printf("Starting Edit compose file %s", file)
 		var editedFile string
-		editedFile, currentPort, err = editComposeFile(ctx, file, executorId, taskInfo.GetTaskId().GetValue(), currentPort, extraHosts)
+		editedFile, currentPort, err = editComposeFile(file, executorId, taskInfo.GetTaskId().GetValue(), currentPort, extraHosts)
 		if err != nil {
 			logger.Errorln("Error editing compose file : ", err.Error())
 			return err
@@ -119,9 +119,9 @@ func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	// Remove infra container yml file if network mode is host
 	if config.GetConfig().GetBool(types.RM_INFRA_CONTAINER) {
 		logger.Printf("Remove file: %s\n", types.INFRA_CONTAINER_GEN_YML)
-		filesMap := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
+		filesMap := pod.GetServiceDetail()
 		delete(filesMap, editedFiles[indexInfra])
-		*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, filesMap)
+		pod.SetServiceDetail(filesMap)
 		editedFiles = append(editedFiles[:indexInfra], editedFiles[indexInfra+1:]...)
 		err = utils.DeleteFile(types.INFRA_CONTAINER_YML)
 		if err != nil {
@@ -133,7 +133,7 @@ func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	}
 
 	logger.Println("====================context out====================")
-	logger.Println((*ctx).Value(types.SERVICE_DETAIL))
+	logger.Printf("SERVICE_DETAIL: %+v", pod.GetServiceDetail())
 
 	*composeFiles = editedFiles
 
@@ -146,15 +146,15 @@ func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles
 	return nil
 }
 
-func (gp *generalExt) LaunchTaskPostImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
+func (gp *generalExt) LaunchTaskPostImagePull(ctx context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
 	logger.Println("LaunchTaskPostImagePull begin")
 	return nil
 }
 
-func (gp *generalExt) PostLaunchTask(ctx *context.Context, files []string, taskInfo *mesos.TaskInfo) (string, error) {
+func (gp *generalExt) PostLaunchTask(ctx context.Context, files []string, taskInfo *mesos.TaskInfo) (string, error) {
 	logger.Println("PostLaunchTask begin")
 	if pod.SinglePort {
-		err := postEditComposeFile(ctx, infraYmlPath)
+		err := postEditComposeFile(infraYmlPath)
 		if err != nil {
 			log.Errorf("PostLaunchTask: Error editing compose file : %v", err)
 			return types.POD_FAILED.String(), err
@@ -163,15 +163,15 @@ func (gp *generalExt) PostLaunchTask(ctx *context.Context, files []string, taskI
 	return "", nil
 }
 
-func (gp *generalExt) PreKillTask(taskInfo *mesos.TaskInfo) error {
+func (gp *generalExt) PreKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PreKillTask begin")
 	return nil
 }
 
 // PostKillTask cleans up containers, volumes, images if task is killed by mesos
 // Failed tasks will be cleaned up based on config cleanpod.cleanvolumeandcontaineronmesoskill and cleanpod.cleanimageonmesoskill
 // Non pre-existing networks will always be removed
-func (gp *generalExt) PostKillTask(taskInfo *mesos.TaskInfo) error {
+func (gp *generalExt) PostKillTask(ctx context.Context, taskInfo *mesos.TaskInfo) error {
 	logger.Println("PostKillTask begin, pod status:", pod.GetPodStatus())
 	var err error
 	if !pod.LaunchCmdAttempted {
@@ -228,15 +228,15 @@ func (gp *generalExt) PostKillTask(taskInfo *mesos.TaskInfo) error {
 	return err
 }
 
-func (gp *generalExt) Shutdown(executor.ExecutorDriver) error {
+func (gp *generalExt) Shutdown(taskInfo *mesos.TaskInfo, ed executor.ExecutorDriver) error {
 	logger.Println("Shutdown begin")
 	return nil
 }
 
-func CreateInfraContainer(ctx *context.Context, path string) (string, error) {
+func CreateInfraContainer(ctx context.Context, path string) (string, error) {
 	containerDetail := make(map[interface{}]interface{})
 	service := make(map[interface{}]interface{})
-	_yaml := make(map[interface{}]interface{})
+	_yaml := make(map[string]interface{})
 
 	containerDetail[types.CONTAINER_NAME] = config.GetConfigSection(config.INFRA_CONTAINER)[types.CONTAINER_NAME]
 	containerDetail[types.IMAGE] = config.GetConfigSection(config.INFRA_CONTAINER)[types.IMAGE]
@@ -284,13 +284,10 @@ func CreateInfraContainer(ctx *context.Context, path string) (string, error) {
 		return "", err
 	}
 
-	fileMap, ok := (*ctx).Value(types.SERVICE_DETAIL).(types.ServiceDetail)
-	if !ok {
-		log.Warningln("SERVICE_DETAIL missing in context value")
-		fileMap = types.ServiceDetail{}
-	}
+	fileMap := pod.GetServiceDetail()
 
 	fileMap[fileName] = _yaml
-	*ctx = context.WithValue(*ctx, types.SERVICE_DETAIL, fileMap)
+
+	pod.SetServiceDetail(fileMap)
 	return fileName, nil
 }

pluginimpl/general/impl_test.go

@@ -29,7 +29,7 @@ import (
 func TestCreateInfraContainer(t *testing.T) {
 	config.GetConfig().SetDefault(types.NO_FOLDER, true)
 	ctx := context.Background()
-	_, err := CreateInfraContainer(&ctx, "testdata/docker-infra-container.yml")
+	_, err := CreateInfraContainer(ctx, "testdata/docker-infra-container.yml")
 	if err != nil {
 		t.Errorf("expected no error, but got %v", err)
 	}

types/types.go

@@ -84,6 +84,7 @@ func (status HealthStatus) String() string {
 }
 
 const (
+	LOGLEVEL                = "loglevel"
 	CONTAINER_NAME          = "container_name"
 	NETWORK_MODE            = "network_mode"
 	HEALTH_CHECK            = "healthcheck"
@@ -124,7 +125,8 @@ const (
 	DCE_ERR                 = "dce.err"
 )
 
-type ServiceDetail map[interface{}](map[interface{}]interface{})
+// ServiceDetail key is filepath, value is map to store Unmarshal the docker-compose.yaml
+type ServiceDetail map[string]map[string]interface{}
 
 type CmdResult struct {
 	Result  error
@@ -160,3 +162,14 @@ func (c *ContainerStatusDetails) SetComposeTaskId(composeTaskId *mesosproto.Task
 type err string
 
 const NoComposeFile err = "no compose file specified"
+
+type StepData struct {
+	RetryID    int                    `json:"retryID,omitempty"`
+	StepName   string                 `json:"stepName,omitempty"`
+	ErrorMsg   error                  `json:"errorMsg,omitempty"`
+	Status     string                 `json:"status,omitempty"`
+	Tags       map[string]interface{} `json:"tags,omitempty"`
+	StartTime  int64                  `json:"startTime,omitempty"`
+	EndTime    int64                  `json:"endTime,omitempty"`
+	ExecTimeMS int64                  `json:"execTimeMS,omitempty"`
+}

utils/file/file.go

@@ -18,8 +18,6 @@ import (
 	"bufio"
 	"bytes"
 	"container/list"
-	"context"
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -29,6 +27,7 @@ import (
 	"strings"
 
 	mesos "github.com/mesos/mesos-go/mesosproto"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 
@@ -223,33 +222,26 @@ func DeleteFile(file string) error {
 	return os.Remove(file)
 }
 
-func WriteChangeToFiles(ctx context.Context) error {
-	filesMap := ctx.Value(types.SERVICE_DETAIL).(types.ServiceDetail)
+func WriteChangeToFiles() error {
+	filesMap := pod.GetServiceDetail()
 	for file := range filesMap {
 		content, err := yaml.Marshal(filesMap[file])
 		if err != nil {
 			log.Errorf("error occured while marshalling file from fileMap: %s", err)
 		}
-		_, err = WriteToFile(file.(string), content)
+		_, err = WriteToFile(file, content)
 		if err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func DumpPluginModifiedComposeFiles(ctx context.Context, plugin, funcName string, pluginOrder int) {
-	if ctx.Value(types.SERVICE_DETAIL) == nil {
-
-	}
-	filesMap, ok := ctx.Value(types.SERVICE_DETAIL).(types.ServiceDetail)
-	if !ok {
-		log.Printf("Skip dumping modified compose file by plugin %s", plugin)
-		return
-	}
+func DumpPluginModifiedComposeFiles(plugin, funcName string, pluginOrder int) {
+	filesMap := pod.GetServiceDetail()
 	for file := range filesMap {
 		content, _ := yaml.Marshal(filesMap[file])
-		fParts := strings.Split(file.(string), PATH_DELIMITER)
+		fParts := strings.Split(file, PATH_DELIMITER)
 		if len(fParts) < 2 {
 			log.Printf("Skip dumping modified compose file by plugin %s, since file name is invalid %s", plugin, file)
 			return
@@ -512,17 +504,21 @@ func DeFolderPath(filepaths []string) []string {
 	return filenames
 }
 
-func ParseYamls(files *[]string) (map[interface{}](map[interface{}]interface{}), error) {
-	res := make(map[interface{}](map[interface{}]interface{}))
+func ParseYamls(files *[]string) (map[string]map[string]interface{}, error) {
+	res := make(map[string]map[string]interface{})
 	for _, file := range *files {
 		data, err := ioutil.ReadFile(file)
 		if err != nil {
 			log.Errorf("Error reading file %s : %v", file, err)
 		}
-		m := make(map[interface{}]interface{})
+		// unmarshal the docker-compose.yaml
+		m := make(map[string]interface{})
 		err = yaml.Unmarshal(data, &m)
 		if err != nil {
-			log.Errorf("Error unmarshalling %v", err)
+			return res, errors.Errorf("Error unmarshalling %v", err)
+		}
+		if len(FolderPath(strings.Fields(file))[0]) == 0 {
+			return res, errors.Errorf("folder %s under %+v is empty", strings.Fields(file), FolderPath)
 		}
 		res[FolderPath(strings.Fields(file))[0]] = m
 	}
@@ -678,7 +674,7 @@ func ConvertMapToArray(m map[interface{}]interface{}) []interface{} {
 // CreateMapValuesArray creates an interface array of all the values of the given map
 // examples: Map= {"1":"a", "2":"b", "3":"c"}
 // array returned will be: ["a", "b", "c"]
-func CreateMapValuesArray(m map[interface{}]interface{}) []interface{} {
+func CreateMapValuesArray(m map[string][]*types.StepData) []interface{} {
 	var a []interface{}
 	for _, v := range m {
 		a = append(a, v)

utils/http/http.go

@@ -96,7 +96,6 @@ func GetRequest(ctx context.Context, transport http.RoundTripper, url string) ([
 		return nil, err
 	}
 	err = resp.Body.Close()
-	err = resp.Body.Close()
 	if err != nil {
 		log.Errorf("Failed to close response body :%v", err)
 		return nil, err

utils/pod/pod.go

@@ -31,7 +31,6 @@ import (
 	"github.com/paypal/dce-go/config"
 	"github.com/paypal/dce-go/plugin"
 	"github.com/paypal/dce-go/types"
-	"github.com/paypal/dce-go/utils"
 	waitUtil "github.com/paypal/dce-go/utils/wait"
 	"github.com/paypal/gorealis/gen-go/apache/aurora"
 	"github.com/pkg/errors"
@@ -59,14 +58,23 @@ var PluginOrder []string
 var HealthCheckListId = make(map[string]bool)
 var MonitorContainerList []string
 var SinglePort bool
-var StepMetrics = make(map[interface{}]interface{})
+var StepMetrics = make(map[string][]*types.StepData)
+
+// store all docker-composed yaml file, key is filepath, value is the yaml unmarshelled object
+var ServiceDetail = make(map[string]map[string]interface{})
 
 // LaunchCmdAttempted indicates that an attempt to run the command to launch the pod (docker compose up with params) was
 // made. This does not indicate that the result of the command execution.
 var LaunchCmdAttempted = false
 
 // taskStatusCh is pushed with the task status sent to Mesos, so any custom pod task status hooks can be executed
-var taskStatusCh = make(chan string, 1)
+
+type TaskStatus struct {
+	Ctx    context.Context
+	Status string
+}
+
+var taskStatusCh = make(chan TaskStatus, 1)
 
 // Check exit code of all the containers in the pod.
 // If all the exit codes are zero, then assign zero as pod's exit code,
@@ -345,7 +353,7 @@ func dockerLogToPodLogFile(files []string, retry bool) {
 
 // Stop pod
 // docker-compose stop
-func StopPod(files []string) error {
+func StopPod(ctx context.Context, files []string) error {
 	logger := log.WithFields(log.Fields{
 		"files": files,
 		"func":  "pod.StopPod",
@@ -358,9 +366,9 @@ func StopPod(files []string) error {
 	logger.Printf("Plugin order: %s", PluginOrder)
 
 	// Executing PreKillTask in order
-	_, err := utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+	_, err := PluginPanicHandler(ConditionFunc(func() (string, error) {
 		for i, ext := range plugins {
-			if err := ext.PreKillTask(ComposeTaskInfo); err != nil {
+			if err := ext.PreKillTask(ctx, ComposeTaskInfo); err != nil {
 				logger.Errorf("Error executing PreKillTask in %dth plugin: %v", i, err)
 			}
 		}
@@ -396,23 +404,23 @@ func StopPod(files []string) error {
 		}
 	}
 
-	err = callAllPluginsPostKillTask()
+	err = callAllPluginsPostKillTask(ctx)
 	if err != nil {
 		logger.Error(err)
 	}
 
 	return nil
 }
 
-func callAllPluginsPostKillTask() error {
+func callAllPluginsPostKillTask(ctx context.Context) error {
 	// Select plugin extension points from plugin pools
 	plugins := plugin.GetOrderedExtpoints(PluginOrder)
 	log.Printf("Plugin order: %s", PluginOrder)
 
 	// Executing PostKillTask plugin extensions in order
-	utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+	PluginPanicHandler(ConditionFunc(func() (string, error) {
 		for _, ext := range plugins {
-			err := ext.PostKillTask(ComposeTaskInfo)
+			err := ext.PostKillTask(ctx, ComposeTaskInfo)
 			if err != nil {
 				log.Errorf("Error executing PostKillTask of plugin : %v", err)
 			}
@@ -595,32 +603,32 @@ func PullImage(files []string) error {
 
 //CheckContainer does check container details
 //return healthy,run,err
-func CheckContainer(containerId string, healthCheck bool) (types.HealthStatus, bool, int, error) {
+func CheckContainer(containerId string, healthCheck bool) (types.ContainerStatusDetails, types.HealthStatus, bool, int, error) {
 	containerDetail, err := InspectContainerDetails(containerId, healthCheck)
 	if err != nil {
 		log.Printf("CheckContainer : Error inspecting container with id : %s, %v", containerId, err.Error())
-		return types.UNHEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, err
+		return containerDetail, types.UNHEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, err
 	}
 
 	if containerDetail.ExitCode != 0 {
 		log.Printf("CheckContainer : Container %s is finished with exit code %v\n", containerId, containerDetail.ExitCode)
-		return types.UNHEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
+		return containerDetail, types.UNHEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
 	}
 
 	if healthCheck {
 		if containerDetail.IsRunning {
 			//log.Printf("CheckContainer : Primary container %s is running , %s\n", containerId, containerDetail.HealthStatus)
-			return utils.ToHealthStatus(containerDetail.HealthStatus), containerDetail.IsRunning, containerDetail.ExitCode, nil
+			return containerDetail, ToHealthStatus(containerDetail.HealthStatus), containerDetail.IsRunning, containerDetail.ExitCode, nil
 		}
-		return utils.ToHealthStatus(containerDetail.HealthStatus), containerDetail.IsRunning, containerDetail.ExitCode, nil
+		return containerDetail, ToHealthStatus(containerDetail.HealthStatus), containerDetail.IsRunning, containerDetail.ExitCode, nil
 	}
 
 	if containerDetail.IsRunning {
 		//log.Printf("CheckContainer : Regular container %s is running\n", containerId)
-		return types.HEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
+		return containerDetail, types.HEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
 	}
 
-	return types.HEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
+	return containerDetail, types.HEALTHY, containerDetail.IsRunning, containerDetail.ExitCode, nil
 }
 
 func KillContainer(sig string, containerId string) error {
@@ -814,7 +822,7 @@ func updatePodLaunched() {
 	log.Printf("Updated Current Pod Status with Pod Launched ")
 }
 
-func SendPodStatus(status types.PodStatus) {
+func SendPodStatus(ctx context.Context, status types.PodStatus) {
 	logger := log.WithFields(log.Fields{
 		"status": status,
 		"func":   "pod.SendPodStatus",
@@ -831,46 +839,46 @@ func SendPodStatus(status types.PodStatus) {
 	switch status {
 	case types.POD_RUNNING:
 		updatePodLaunched()
-		SendMesosStatus(ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_RUNNING.Enum())
+		SendMesosStatus(ctx, ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_RUNNING.Enum())
 	case types.POD_FINISHED:
 		// Stop pod after sending status to mesos
 		// To kill system proxy container
 		if len(MonitorContainerList) > 0 {
 			logger.Printf("Stop containers still running in the pod: %v", MonitorContainerList)
-			err := StopPod(ComposeFiles)
+			err := StopPod(ctx, ComposeFiles)
 			if err != nil {
 				logger.Errorf("Error stop pod: %v", err)
 			}
 		} else {
 			// Adding this else part to clean the adhoc tasks.
-			err := callAllPluginsPostKillTask()
+			err := callAllPluginsPostKillTask(ctx)
 			if err != nil {
 				logger.Error(err)
 			}
 		}
-		SendMesosStatus(ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FINISHED.Enum())
+		SendMesosStatus(ctx, ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FINISHED.Enum())
 	case types.POD_FAILED:
 		if LaunchCmdAttempted {
-			err := StopPod(ComposeFiles)
+			err := StopPod(ctx, ComposeFiles)
 			if err != nil {
 				logger.Errorf("Error cleaning up pod : %v\n", err.Error())
 			}
 		}
-		SendMesosStatus(ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		SendMesosStatus(ctx, ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 	case types.POD_PULL_FAILED:
-		callAllPluginsPostKillTask()
-		SendMesosStatus(ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		callAllPluginsPostKillTask(ctx)
+		SendMesosStatus(ctx, ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 
 	case types.POD_COMPOSE_CHECK_FAILED:
-		callAllPluginsPostKillTask()
-		SendMesosStatus(ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
+		callAllPluginsPostKillTask(ctx)
+		SendMesosStatus(ctx, ComposeExecutorDriver, ComposeTaskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 	}
 
 	logger.Printf("MesosStatus %s completed", status)
 }
 
 //Update mesos and pod status
-func SendMesosStatus(driver executor.ExecutorDriver, taskId *mesos.TaskID, state *mesos.TaskState) error {
+func SendMesosStatus(ctx context.Context, driver executor.ExecutorDriver, taskId *mesos.TaskID, state *mesos.TaskState) error {
 	logger := log.WithFields(log.Fields{
 		"state": state.Enum().String(),
 		"func":  "pod.SendMesosStatus",
@@ -907,22 +915,46 @@ func SendMesosStatus(driver executor.ExecutorDriver, taskId *mesos.TaskID, state
 	time.Sleep(5 * time.Second)
 
 	// Push the state to Task status channel so any further steps on a given task status can be executed
-	taskStatusCh <- state.Enum().String()
+	taskStatusCh <- TaskStatus{
+		Ctx:    ctx,
+		Status: state.Enum().String(),
+	}
+
+	return nil
+}
 
+// dumpContainerInspect dumps docker inspect output of containers
+func dumpContainerInspect() error {
+	ids, err := GetPodContainerIds(ComposeFiles)
+	if err != nil {
+		return err
+	}
+	log.Debugf("found container ids %+v", ids)
+	for i := range ids {
+		if err := PrintInspectDetail(ids[i]); err != nil {
+			log.Warnf("unable to inspect container %s", ids[i])
+		}
+	}
 	return nil
 }
 
 // Wait for pod running/finished until timeout or failed
-func WaitOnPod(ctx *context.Context) {
+func WaitOnPod(ctx context.Context) {
 	select {
-	case <-(*ctx).Done():
-		if (*ctx).Err() == context.DeadlineExceeded {
+	case <-(ctx).Done():
+		if (ctx).Err() == context.DeadlineExceeded {
 			log.Println("POD_LAUNCH_TIMEOUT")
 			if dump, ok := config.GetConfig().GetStringMap("dockerdump")["enable"].(bool); ok && dump {
 				DockerDump()
 			}
-			SendPodStatus(types.POD_FAILED)
-		} else if (*ctx).Err() == context.Canceled {
+			log.Debug("Dumping containers state...")
+			// dump docker inspect for each for container
+			if err := dumpContainerInspect(); err != nil {
+				log.Warnf("unable to inspect containers --%v", err)
+			}
+			log.Debug("Completed dumping containers state")
+			SendPodStatus(ctx, types.POD_FAILED)
+		} else if (ctx).Err() == context.Canceled {
 			log.Println("Stop waitUtil on pod, since pod is running/finished/failed")
 		}
 	}
@@ -1125,6 +1157,8 @@ func HealthCheck(files []string, podServices map[string]bool, out chan<- string)
 
 healthCheck:
 	for len(containers) != healthCount {
+		StartStep(StepMetrics, "HealthCheck")
+		tag := make(map[string]interface{})
 		healthCount = 0
 
 		for i := 0; i < len(containers); i++ {
@@ -1134,12 +1168,12 @@ healthCheck:
 			var running bool
 
 			if hc, ok := HealthCheckListId[containers[i]]; ok && hc {
-				healthy, running, exitCode, err = CheckContainer(containers[i], true)
+				tag[containers[i]], healthy, running, exitCode, err = CheckContainer(containers[i], true)
 			} else {
 				if hc, err = isHealthCheckConfigured(containers[i]); hc {
-					healthy, running, exitCode, err = CheckContainer(containers[i], true)
+					tag[containers[i]], healthy, running, exitCode, err = CheckContainer(containers[i], true)
 				} else {
-					healthy, running, exitCode, err = CheckContainer(containers[i], false)
+					tag[containers[i]], healthy, running, exitCode, err = CheckContainer(containers[i], false)
 				}
 			}
 
@@ -1169,6 +1203,7 @@ healthCheck:
 				break healthCheck
 			}
 		}
+		EndStep(StepMetrics, "HealthCheck", tag, err)
 
 		if len(containers) != healthCount {
 			time.Sleep(interval)
@@ -1240,11 +1275,11 @@ func ListenOnTaskStatus(driver executor.ExecutorDriver, taskInfo *mesos.TaskInfo
 	cachedTaskInfo := &taskInfo
 	for {
 		select {
-		case status, ok := <-taskStatusCh: // wait for task status from LaunchTask
+		case taskStatus, ok := <-taskStatusCh: // wait for task status from LaunchTask
 			if ok {
-				switch status {
+				switch taskStatus.Status {
 				case mesos.TaskState_TASK_RUNNING.String():
-					if err := execPodStatusHooks(status, *cachedTaskInfo); err != nil {
+					if err := execPodStatusHooks(taskStatus.Ctx, taskStatus.Status, *cachedTaskInfo); err != nil {
 						logger.Errorf("executing hooks failed %v ", err)
 					}
 				case mesos.TaskState_TASK_FAILED.String():
@@ -1254,7 +1289,7 @@ func ListenOnTaskStatus(driver executor.ExecutorDriver, taskInfo *mesos.TaskInfo
 						2. Health Monitor or any plugin monitors and fails after the task has been running for
 						   a longtime (PodStatus.Launched = true, and marked as failed later)
 					*/
-					if err := execPodStatusHooks(status, *cachedTaskInfo); err != nil {
+					if err := execPodStatusHooks(taskStatus.Ctx, taskStatus.Status, *cachedTaskInfo); err != nil {
 						logger.Errorf("executing hooks failed %v ", err)
 					}
 					stopDriver(driver)
@@ -1263,7 +1298,7 @@ func ListenOnTaskStatus(driver executor.ExecutorDriver, taskInfo *mesos.TaskInfo
 					stopDriver(driver)
 					break
 				default:
-					log.Infof("Nothing to do on task status %s", status)
+					log.Infof("Nothing to do on task status %s", taskStatus.Status)
 				}
 			} else {
 				log.Errorln("failure reading from task status channel")
@@ -1274,7 +1309,7 @@ func ListenOnTaskStatus(driver executor.ExecutorDriver, taskInfo *mesos.TaskInfo
 
 // execPodStatusHooks finds the hooks (implementations of ExecutorHook interface) configured for executor phase and executes them
 // error is returned if any of the hooks failed, and ExecutorHook.BestEffort() returns true
-func execPodStatusHooks(status string, taskInfo *mesos.TaskInfo) error {
+func execPodStatusHooks(ctx context.Context, status string, taskInfo *mesos.TaskInfo) error {
 	logger := log.WithFields(log.Fields{
 		"requuid":   GetLabel("requuid", taskInfo),
 		"tenant":    GetLabel("tenant", taskInfo),
@@ -1288,14 +1323,14 @@ func execPodStatusHooks(status string, taskInfo *mesos.TaskInfo) error {
 		return nil
 	}
 	logger.Infof("Executor Post Hooks found: %v", podStatusHooks)
-	if _, err := utils.PluginPanicHandler(utils.ConditionFunc(func() (string, error) {
+	if _, err := PluginPanicHandler(ConditionFunc(func() (string, error) {
 		for _, name := range podStatusHooks {
 			hook := plugin.PodStatusHooks.Lookup(name)
 			if hook == nil {
 				logger.Errorf("Hook %s is nil, not initialized? still continuing with available hooks", name)
 				continue
 			}
-			if failExec, pherr := hook.Execute(status, taskInfo); pherr != nil {
+			if failExec, pherr := hook.Execute(ctx, status, taskInfo); pherr != nil {
 				logger.Errorf(
 					"PodStatusHook %s failed with %v and is not best effort, so stopping further execution ",
 					name, pherr)

utils/pod/pod_test.go

@@ -15,6 +15,7 @@
 package pod
 
 import (
+	"context"
 	"log"
 	"strings"
 	"testing"
@@ -33,28 +34,24 @@ import (
 func TestLaunchPod(t *testing.T) {
 	// file doesn't exist, should fail
 	files := []string{"docker-fail.yml"}
-	res := LaunchPod(files)
-	if res != types.POD_FAILED {
-		t.Fatalf("expected pod status to be POD_FAILED, but got %s", res)
-	}
+	res, err := LaunchPod(files)
+	assert.NoError(t, err)
+	assert.EqualValues(t, res, types.POD_FAILED)
 
 	// adhoc job
 	files = []string{"testdata/docker-adhoc.yml"}
-	res = LaunchPod(files)
-	if res != types.POD_STARTING {
-		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
-	}
+	res, err = LaunchPod(files)
+	assert.NoError(t, err)
+	assert.EqualValues(t, res, types.POD_STARTING)
 
 	// long running job
 	files = []string{"testdata/docker-long.yml"}
-	res = LaunchPod(files)
-	if res != types.POD_STARTING {
-		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
-	}
-	err := ForceKill(files)
-	if err != nil {
-		t.Errorf("expected no errors, but got %v", err)
-	}
+	res, err = LaunchPod(files)
+	assert.NoError(t, err)
+	assert.EqualValues(t, res, types.POD_STARTING)
+
+	err = ForceKill(files)
+	assert.NoError(t, err)
 }
 
 func TestGetContainerNetwork(t *testing.T) {
@@ -84,32 +81,33 @@ func TestRemoveNetwork(t *testing.T) {
 
 func TestForceKill(t *testing.T) {
 	files := []string{"testdata/docker-long.yml"}
-	res := LaunchPod(files)
+	res, err := LaunchPod(files)
+	assert.NoError(t, err)
 	if res != types.POD_STARTING {
 		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
 	}
-	err := ForceKill(files)
-	if err != nil {
-		t.Errorf("expected no errors, but got %v", err)
-	}
+	err = ForceKill(files)
+	assert.NoError(t, err)
 }
 
 func TestStopPod(t *testing.T) {
 	files := []string{"testdata/docker-long.yml"}
-	res := LaunchPod(files)
+	res, err := LaunchPod(files)
+	assert.NoError(t, err)
 	if res != types.POD_STARTING {
 		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
 	}
 	config.GetConfig().SetDefault(types.RM_INFRA_CONTAINER, true)
-	err := StopPod(files)
+	err = StopPod(context.TODO(), files)
+	assert.NoError(t, err)
 	if err != nil {
 		t.Errorf("expected no errors, but got %v", err)
 	}
 }
 
 func TestGetContainerIdByService(t *testing.T) {
 	files := []string{"testdata/docker-long.yml"}
-	res := LaunchPod(files)
+	res, err := LaunchPod(files)
 	if res != types.POD_STARTING {
 		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
 	}
@@ -134,7 +132,8 @@ func TestKillContainer(t *testing.T) {
 	assert.Error(t, err, "test kill invalid container")
 
 	files := []string{"testdata/docker-long.yml"}
-	res := LaunchPod(files)
+	res, err := LaunchPod(files)
+	assert.NoError(t, err)
 	if res != types.POD_STARTING {
 		t.Fatalf("expected pod status to be POD_STARTING, but got %s", res)
 	}
@@ -144,9 +143,11 @@ func TestKillContainer(t *testing.T) {
 	err = KillContainer("SIGUSR1", id)
 	assert.NoError(t, err, "Test sending kill signal to container")
 	err = KillContainer("", id)
+	assert.NoError(t, err)
 
 	config.GetConfig().Set(types.RM_INFRA_CONTAINER, true)
-	StopPod(files)
+	err = StopPod(context.Background(), files)
+	assert.NoError(t, err)
 }
 
 func TestGetAndRemoveLabel(t *testing.T) {
@@ -215,6 +216,7 @@ func TestGetAndRemoveLabel(t *testing.T) {
 }
 
 func TestExecHooks(t *testing.T) {
+	ctx := context.Background()
 	//Register plugin with name
 	if ok := plugin.PodStatusHooks.Register(&happyHook{}, "happyHook"); !ok {
 		log.Fatalf("failed to register plugin %s", "happyHook")
@@ -229,29 +231,29 @@ func TestExecHooks(t *testing.T) {
 	}
 
 	config.GetConfig().Set("podstatushooks.TASK_RUNNING", []string{"happyHook"})
-	assert.NoError(t, execPodStatusHooks("TASK_RUNNING", nil), "happy hook can't fail")
+	assert.NoError(t, execPodStatusHooks(ctx, "TASK_RUNNING", nil), "happy hook can't fail")
 
 	config.GetConfig().Set("podstatushooks.TASK_FAILED", []string{"happyHook", "mandatoryHook"})
-	assert.Error(t, execPodStatusHooks("TASK_FAILED", nil), "mandatory hook can't succeed")
+	assert.Error(t, execPodStatusHooks(ctx, "TASK_FAILED", nil), "mandatory hook can't succeed")
 
 	config.GetConfig().Set("podstatushooks.TASK_FAILED", []string{"panicHook"})
-	assert.Error(t, execPodStatusHooks("TASK_FAILED", nil), "panicHook hook can't succeed")
+	assert.Error(t, execPodStatusHooks(ctx, "TASK_FAILED", nil), "panicHook hook can't succeed")
 }
 
 // dummy executor hooks for unit test
 type happyHook struct{}
 type mandatoryHook struct{}
 type panicHook struct{}
 
-func (p *happyHook) Execute(podStatus string, data interface{}) (failExec bool, err error) {
+func (p *happyHook) Execute(ctx context.Context, podStatus string, data interface{}) (failExec bool, err error) {
 	return true, nil
 }
 
-func (p *mandatoryHook) Execute(status string, data interface{}) (failExec bool, err error) {
+func (p *mandatoryHook) Execute(ctx context.Context, status string, data interface{}) (failExec bool, err error) {
 	return true, errors.New("failure test case")
 }
 
-func (p *panicHook) Execute(status string, data interface{}) (failExec bool, err error) {
+func (p *panicHook) Execute(ctx context.Context, status string, data interface{}) (failExec bool, err error) {
 	panic("unit test panic")
 	return false, errors.New("panic test case")
 }

utils/pod/service.go

@@ -0,0 +1,13 @@
+package pod
+
+import (
+	"github.com/paypal/dce-go/types"
+)
+
+func GetServiceDetail() types.ServiceDetail {
+	return ServiceDetail
+}
+
+func SetServiceDetail(sd types.ServiceDetail) {
+	ServiceDetail = sd
+}

utils/pod/service_test.go

@@ -0,0 +1,59 @@
+package pod
+
+import (
+	"testing"
+
+	"gopkg.in/yaml.v2"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUpdateServiceDetail(t *testing.T) {
+	// prepare taskInfo
+	t.Run("manual-set", func(t *testing.T) {
+		s := GetServiceDetail()
+		assert.Empty(t, s)
+
+		s["a1"] = make(map[string]interface{})
+		s["a1"]["b1"] = "c1"
+		s["a1"]["b2"] = 2
+		s["a1"]["b3"] = []interface{}{"c1, c2"}
+
+		SetServiceDetail(s)
+
+		s1 := GetServiceDetail()
+		assert.EqualValues(t, s, s1)
+	})
+
+	t.Run("parse-yaml", func(t *testing.T) {
+
+		m := make(map[string]interface{})
+		sd := GetServiceDetail()
+
+		var yamlFile = []byte(`
+Hacker: true
+name: steve
+hobbies:
+- skateboarding
+- snowboarding
+- go
+clothing:
+  jacket: leather
+  trousers: denim
+age: 35
+eyes : brown
+beard: true
+`)
+
+		err := yaml.Unmarshal(yamlFile, &m)
+		assert.NoError(t, err)
+
+		sd["f1"] = m
+
+		SetServiceDetail(sd)
+
+		s1 := GetServiceDetail()
+		assert.EqualValues(t, sd, s1)
+	})
+
+}

utils/pod/util.go

@@ -0,0 +1,117 @@
+package pod
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/paypal/dce-go/types"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+type ConditionFunc func() (string, error)
+
+func PluginPanicHandler(condition ConditionFunc) (res string, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("Recover : %v \n", r)
+			err = errors.New(fmt.Sprintf("Recover : %v \n", r))
+		}
+	}()
+
+	if res, err = condition(); err != nil {
+		log.Errorf("Error executing plugins: %v \n", err)
+		return res, err
+	}
+	return res, err
+}
+
+func ToPodStatus(s string) types.PodStatus {
+	switch s {
+	case "POD_STAGING":
+		return types.POD_STAGING
+	case "POD_STARTING":
+		return types.POD_STARTING
+	case "POD_RUNNING":
+		return types.POD_RUNNING
+	case "POD_FAILED":
+		return types.POD_FAILED
+	case "POD_KILLED":
+		return types.POD_KILLED
+	case "POD_FINISHED":
+		return types.POD_FINISHED
+	case "POD_PULL_FAILED":
+		return types.POD_PULL_FAILED
+	case "POD_COMPOSE_CHECK_FAILED":
+		return types.POD_COMPOSE_CHECK_FAILED
+	}
+
+	return types.POD_EMPTY
+}
+
+func ToHealthStatus(s string) types.HealthStatus {
+	switch s {
+	case "starting":
+		return types.STARTING
+	case "healthy":
+		return types.HEALTHY
+	case "unhealthy":
+		return types.UNHEALTHY
+	}
+
+	return types.UNKNOWN_HEALTH_STATUS
+}
+
+// StartStep start a step of dce, and add a new item into the values, key is the stepName
+// In the map, key will be step, value will be each retry result, and duration.
+func StartStep(stepData map[string][]*types.StepData, stepName string) {
+	if len(stepName) == 0 {
+		log.Error("error while updating step data for Granular Metrics: step name can't be empty for stepData")
+	}
+	var ok bool
+
+	stepValues, ok := stepData[stepName]
+	if !ok {
+		stepValues = []*types.StepData{}
+		stepData[stepName] = stepValues
+	}
+	stepValue := &types.StepData{}
+
+	stepValue.StepName = stepName
+	stepValue.RetryID = len(stepValues)
+	stepValue.StartTime = time.Now().Unix()
+	stepValue.Status = "Starting"
+	stepValues = append(stepValues, stepValue)
+	stepData[stepName] = stepValues
+}
+
+// EndStep ends the current dce step, and update the result, duraiton.
+// current dce step can be fetch from stepData, key is the stepName, value is each retry results. Update the latest result
+func EndStep(stepData map[string][]*types.StepData, stepName string, tag map[string]interface{}, err error) {
+	if len(stepName) == 0 {
+		log.Error("error while updating step data for Granular Metrics: step name can't be empty for stepData")
+		return
+	}
+	var ok bool
+
+	stepValues, ok := stepData[stepName]
+	if !ok {
+		log.Errorf("key %s not exist in stepData %+v", stepName, stepData)
+		return
+	}
+	if len(stepValues) < 1 {
+		log.Errorf("len of stepValues is %d, less than 1", len(stepValues))
+		return
+	}
+
+	step := stepValues[len(stepValues)-1]
+	step.Tags = tag
+	step.EndTime = time.Now().Unix()
+	step.ErrorMsg = err
+	step.ExecTimeMS = (step.EndTime - step.StartTime) * 1000
+	if err != nil {
+		step.Status = "Error"
+	} else {
+		step.Status = "Success"
+	}
+}

utils/pod/util_test.go

@@ -0,0 +1,67 @@
+package pod
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/paypal/dce-go/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPluginPanicHandler(t *testing.T) {
+	_, err := PluginPanicHandler(ConditionFunc(func() (string, error) {
+		panic("Test panic error")
+	}))
+	if err == nil {
+		t.Error("Expected err not be nil, but got nil")
+	}
+}
+
+func TestSetStepData(t *testing.T) {
+	testErr := errors.New("unit-test")
+	example := map[string][]*types.StepData{
+		"Image_Pull": {
+			{
+				RetryID:  0,
+				StepName: "Image_Pull",
+				Status:   "Error",
+				ErrorMsg: testErr,
+			},
+			{
+				RetryID:  1,
+				StepName: "Image_Pull",
+				Status:   "Success",
+			},
+		},
+		"HealthCheck": {
+			{
+				RetryID:  0,
+				StepName: "HealthCheck",
+				Status:   "Success",
+			},
+		},
+	}
+	StartStep(StepMetrics, "Image_Pull")
+	EndStep(StepMetrics, "Image_Pull", nil, testErr)
+
+	StartStep(StepMetrics, "Image_Pull")
+	EndStep(StepMetrics, "Image_Pull", nil, nil)
+
+	StartStep(StepMetrics, "HealthCheck")
+	EndStep(StepMetrics, "HealthCheck", nil, nil)
+
+	for k, v1 := range example {
+		v2, ok := StepMetrics[k]
+		assert.True(t, ok)
+		assert.Equal(t, len(v1), len(v2))
+		for i, s1 := range v1 {
+			s2 := v2[i]
+			assert.Equal(t, (s2.EndTime-s2.StartTime)*1000, s2.ExecTimeMS)
+			assert.Equal(t, s1.RetryID, s2.RetryID)
+			assert.Equal(t, s1.Status, s2.Status)
+			assert.Equal(t, s1.StepName, s2.StepName)
+			assert.Equal(t, s1.ErrorMsg, s2.ErrorMsg)
+			assert.Equal(t, s1.Tags, s2.Tags)
+		}
+	}
+}