Merge pull request #88 from paypal/develop

Granular Metrics (develop to release)

dce/main.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"encoding/json"
 	"flag"
+	"fmt"
 	"os"
 	"os/signal"
 	"strings"
@@ -141,21 +142,28 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 	// Executing LaunchTaskPreImagePull in order
 	if _, err := utils.PluginPanicHandler(utils.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")
+
 			err = ext.LaunchTaskPreImagePull(&ctx, &pod.ComposeFiles, executorId, taskInfo)
+			utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Complete")
 			if err != nil {
 				logger.Errorf("Error executing LaunchTaskPreImagePull of plugin : %v", err)
 				return "", err
 			}
+
 			if config.EnableComposeTrace() {
 				fileUtils.DumpPluginModifiedComposeFiles(ctx, 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())
@@ -178,8 +186,12 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 		return
 	}
 
+	utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, "Image_Pull", "Starting")
+
 	// Pull image
-	if err := pullImage(); err != nil {
+	err = pullImage()
+	utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), "Image_Pull", "Completed")
+	if err != nil {
 		pod.SetPodStatus(types.POD_PULL_FAILED)
 		cancel()
 		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
@@ -196,11 +208,16 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 				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")
+
 			err = ext.LaunchTaskPostImagePull(&ctx, &pod.ComposeFiles, executorId, taskInfo)
+			utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Completed")
 			if err != nil {
 				logger.Errorf("Error executing LaunchTaskPreImagePull of plugin : %v", err)
 				return "", err
 			}
+
 			if config.EnableComposeTrace() {
 				fileUtils.DumpPluginModifiedComposeFiles(ctx, pluginOrder[i], "LaunchTaskPostImagePull", i)
 			}
@@ -226,8 +243,11 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 		pod.SendMesosStatus(driver, taskInfo.GetTaskId(), mesos.TaskState_TASK_FAILED.Enum())
 	}
 
+	utils.SetStepData(pod.StepMetrics, time.Now().Unix(), 0, "Launch_Pod", "Starting")
+
 	// Launch pod
 	replyPodStatus := pod.LaunchPod(pod.ComposeFiles)
+	utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), "Launch_Pod", "Completed")
 
 	logger.Printf("Pod status returned by LaunchPod : %s", replyPodStatus.String())
 
@@ -251,10 +271,15 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 			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")
+
 				tempStatus, err = ext.PostLaunchTask(&ctx, pod.ComposeFiles, taskInfo)
+				utils.SetStepData(pod.StepMetrics, 0, time.Now().Unix(), granularMetricStepName, "Completed")
 				if err != nil {
 					logger.Errorf("Error executing PostLaunchTask : %v", err)
 				}
+
 				logger.Printf("Get pod status : %s returned by PostLaunchTask", tempStatus)
 
 				if tempStatus == types.POD_FAILED.String() {
@@ -286,7 +311,6 @@ func (exec *dockerComposeExecutor) LaunchTask(driver exec.ExecutorDriver, taskIn
 
 	default:
 		logger.Printf("default: Unknown status -- %s from pullAndLaunchPod ", replyPodStatus)
-
 	}
 
 	logger.Println("====================Mesos LaunchTask Returned====================")

plugin/type.go

@@ -23,11 +23,25 @@ import (
 )
 
 type ComposePlugin interface {
+	// Get 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
+
+	// execute some tasks after the Image is pulled
 	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)
+
+	// execute the task before we send a Kill to Mesos
 	PreKillTask(taskInfo *mesos.TaskInfo) error
+
+	// execute the task after we send a Kill to Mesos
 	PostKillTask(taskInfo *mesos.TaskInfo) error
+
+	// execute the task to shutdown the pod
 	Shutdown(executor.ExecutorDriver) error
 }
 
@@ -38,4 +52,4 @@ type PodStatusHook interface {
 	// 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)
-}
+}

pluginimpl/example/impl.go

@@ -41,6 +41,10 @@ func init() {
 	plugin.ComposePlugins.Register(new(exampleExt), "example")
 }
 
+func (p *exampleExt) Name() string {
+	return "example"
+}
+
 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{}.

pluginimpl/general/impl.go

@@ -54,6 +54,10 @@ func init() {
 	config.ConfigInit(utils.SearchFile(".", "general.yaml"))
 }
 
+func (p *generalExt) Name() string {
+	return "general"
+}
+
 func (ge *generalExt) LaunchTaskPreImagePull(ctx *context.Context, composeFiles *[]string, executorId string, taskInfo *mesos.TaskInfo) error {
 	logger.Println("LaunchTaskPreImagePull begin")
 

utils/file/file.go

@@ -226,8 +226,11 @@ func DeleteFile(file string) error {
 func WriteChangeToFiles(ctx context.Context) error {
 	filesMap := ctx.Value(types.SERVICE_DETAIL).(types.ServiceDetail)
 	for file := range filesMap {
-		content, _ := yaml.Marshal(filesMap[file])
-		_, err := WriteToFile(file.(string), content)
+		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)
 		if err != nil {
 			return err
 		}
@@ -671,3 +674,14 @@ func ConvertMapToArray(m map[interface{}]interface{}) []interface{} {
 	}
 	return a
 }
+
+// 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{} {
+	var a []interface{}
+	for _, v := range m {
+		a = append(a, v)
+	}
+	return a
+}

utils/pod/pod.go

@@ -60,6 +60,7 @@ var PluginOrder []string
 var HealthCheckListId = make(map[string]bool)
 var MonitorContainerList []string
 var SinglePort bool
+var StepMetrics = make(map[interface{}]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.

utils/util.go

@@ -60,3 +60,33 @@ func ToHealthStatus(s string) types.HealthStatus {
 
 	return types.UNKNOWN_HEALTH_STATUS
 }
+
+/*
+SetStepData sets the metricsof the DCE step in the map. If the step already exist in the map, it just updates the end time
+*/
+func SetStepData(stepData map[interface{}]interface{}, startTime, endTime int64, stepName, status string) {
+
+	if len(stepName) == 0 {
+		log.Error("error while updating step data for Granular Metrics: step name can't be empty for stepData")
+	}
+	var stepValue map[string]interface{}
+	var ok bool
+
+	stepValue, ok = stepData[stepName].(map[string]interface{})
+	if !ok {
+		stepValue = make(map[string]interface{})
+	}
+
+	stepValue["stepName"] = stepName
+	if startTime != 0 {
+		stepValue["startTime"] = startTime
+	}
+	if endTime != 0 {
+		stepValue["endTime"] = endTime
+		stepValue["execTimeMS"] = (stepValue["endTime"].(int64) - stepValue["startTime"].(int64)) * 1000
+	}
+	if len(status) > 0 {
+		stepValue["status"] = status
+	}
+	stepData[stepName] = stepValue
+}