// RunPlugins executes a set of plugins. The plugin configurations are given in a map with pluginId as key.
// Outputs the results of running the plugins, indexed by pluginId.
func RunPlugins(
	context context.T,
	documentID string,
	plugins map[string]*contracts.Configuration,
	pluginRegistry plugin.PluginRegistry,
	sendReply SendResponse,
	cancelFlag task.CancelFlag,
) (pluginOutputs map[string]*contracts.PluginResult) {

	requestReboot := false

	pluginOutputs = make(map[string]*contracts.PluginResult)
	for pluginID, pluginConfig := range plugins {
		// populate plugin start time and status
		pluginOutputs[pluginID] = &contracts.PluginResult{
			Status:        contracts.ResultStatusInProgress,
			StartDateTime: time.Now(),
		}
		if pluginConfig.OutputS3BucketName != "" {
			pluginOutputs[pluginID].OutputS3BucketName = pluginConfig.OutputS3BucketName
			if pluginConfig.OutputS3KeyPrefix != "" {
				pluginOutputs[pluginID].OutputS3KeyPrefix = pluginConfig.OutputS3KeyPrefix

			}
		}
		p, ok := pluginRegistry[pluginID]
		if !ok {
			err := fmt.Errorf("Plugin with id %s not found!", pluginID)
			pluginOutputs[pluginID].Status = contracts.ResultStatusFailed
			pluginOutputs[pluginID].Error = err
			context.Log().Error(err)
		} else {
			r := runPlugin(context, p, pluginID, *pluginConfig, cancelFlag)
			pluginOutputs[pluginID].Code = r.Code
			pluginOutputs[pluginID].Status = r.Status
			pluginOutputs[pluginID].Error = r.Error
			pluginOutputs[pluginID].Output = r.Output

			if r.Status == contracts.ResultStatusSuccessAndReboot {
				requestReboot = true
			}
		}
		// set end time.
		pluginOutputs[pluginID].EndDateTime = time.Now()

		context.Log().Infof("Sending response on plugin completion: %v", pluginID)
		sendReply(documentID, pluginID, pluginOutputs)

	}

	// request reboot if any of the plugins have requested a reboot
	if requestReboot {
		context.Log().Debug("Requesting reboot...")
		go rebooter.RequestPendingReboot()
	}

	return
}

// processCancelCommandMessage processes a single send command message received from MDS.
func (p *Processor) processCancelCommandMessage(context context.T,
	mdsService service.Service,
	sendCommandPool task.Pool,
	msg ssmmds.Message) {

	log := context.Log()
	log.Debug("Processing cancel command message ", jsonutil.Indent(*msg.Payload))

	var parsedMessage messageContracts.CancelPayload
	err := json.Unmarshal([]byte(*msg.Payload), &parsedMessage)
	if err != nil {
		log.Error("format of received cancel message is invalid ", err)
		err = mdsService.FailMessage(log, *msg.MessageId, service.InternalHandlerException)
		if err != nil {
			sdkutil.HandleAwsError(log, err, p.processorStopPolicy)
		}
		return
	}
	log.Debugf("ParsedMessage is %v", parsedMessage)

	//persist in current folder here
	cancelCmd := initializeCancelCommandState(msg, parsedMessage)
	commandID := getCommandID(*msg.MessageId)
	// persist new interim command state in current folder
	commandStateHelper.PersistData(log, commandID, *msg.Destination, appconfig.DefaultLocationOfCurrent, cancelCmd)

	//remove from pending folder
	commandStateHelper.RemoveData(log, commandID, *msg.Destination, appconfig.DefaultLocationOfPending)

	log.Debugf("Canceling job with id %v...", parsedMessage.CancelMessageID)

	if found := sendCommandPool.Cancel(parsedMessage.CancelMessageID); !found {
		log.Debugf("Job with id %v not found (possibly completed)", parsedMessage.CancelMessageID)
		cancelCmd.DebugInfo = fmt.Sprintf("Command %v couldn't be cancelled", cancelCmd.CancelCommandID)
		cancelCmd.Status = contracts.ResultStatusFailed
	} else {
		cancelCmd.DebugInfo = fmt.Sprintf("Command %v cancelled", cancelCmd.CancelCommandID)
		cancelCmd.Status = contracts.ResultStatusSuccess
	}

	//persist the final status of cancel-message in current folder
	commandStateHelper.PersistData(log, commandID, *msg.Destination, appconfig.DefaultLocationOfCurrent, cancelCmd)

	//persist : commands execution in completed folder (terminal state folder)
	log.Debugf("Execution of %v is over. Moving interimState file from Current to Completed folder", *msg.MessageId)

	commandStateHelper.MoveCommandState(log,
		commandID,
		*msg.Destination,
		appconfig.DefaultLocationOfCurrent,
		appconfig.DefaultLocationOfCompleted)

	log.Debugf("Deleting message")
	err = mdsService.DeleteMessage(log, *msg.MessageId)
	if err != nil {
		sdkutil.HandleAwsError(log, err, p.processorStopPolicy)
	}
}

// NewHealthCheck creates a new health check core plugin.
func NewHealthCheck(context context.T) *HealthCheck {
	healthContext := context.With("[" + name + "]")
	healthCheckStopPolicy := sdkutil.NewStopPolicy(name, 10)
	svc := ssm.NewService()

	return &HealthCheck{
		context:               healthContext,
		healthCheckStopPolicy: healthCheckStopPolicy,
		service:               svc,
	}
}

// Our intent here is to look for array of commands which will be executed as a part of this document and replace the incompatible code.
func removeDependencyOnInstanceMetadataForManagedInstance(context context.T, parsedMessage messageContracts.SendCommandPayload) (messageContracts.SendCommandPayload, error) {
	log := context.Log()
	var properties []interface{}
	var parsedDocumentProperties ManagedInstanceDocumentProperties

	err := jsonutil.Remarshal(parsedMessage.DocumentContent.RuntimeConfig[appconfig.PluginNameAwsRunScript].Properties, &properties)
	if err != nil {
		log.Errorf("Invalid format of properties in %v document. error: %v", parsedMessage.DocumentName, err)
		return parsedMessage, err
	}

	// Since 'Properties' is an array and we use only one property block for the above documents, array location '0' of 'Properties' is used.
	err = jsonutil.Remarshal(properties[0], &parsedDocumentProperties)
	if err != nil {
		log.Errorf("Invalid format of properties in %v document. error: %v", parsedMessage.DocumentName, err)
		return parsedMessage, err
	}

	region, err := platform.Region()
	if err != nil {
		log.Errorf("Error retrieving agent region. error: %v", err)
		return parsedMessage, err
	}

	// Comment or replace the incompatible code from this document.
	log.Info("Replacing managed instance incompatible code for AWS SSM Document.")
	for i, command := range parsedDocumentProperties.RunCommand {
		if strings.Contains(command, "$metadataLocation = 'http://169.254.169.254/latest/dynamic/instance-identity/document/region'") {
			parsedDocumentProperties.RunCommand[i] = strings.Replace(command, "$metadataLocation = 'http://169.254.169.254/latest/dynamic/instance-identity/document/region'", "# $metadataLocation = 'http://169.254.169.254/latest/dynamic/instance-identity/document/region' (This is done to make it managed instance compatible)", 1)
		}

		if strings.Contains(command, "$metadata = (New-Object Net.WebClient).DownloadString($metadataLocation)") {
			parsedDocumentProperties.RunCommand[i] = strings.Replace(command, "$metadata = (New-Object Net.WebClient).DownloadString($metadataLocation)", "# $metadata = (New-Object Net.WebClient).DownloadString($metadataLocation) (This is done to make it managed instance compatible)", 1)
		}

		if strings.Contains(command, "$region = (ConvertFrom-JSON $metadata).region") {
			parsedDocumentProperties.RunCommand[i] = strings.Replace(command, "$region = (ConvertFrom-JSON $metadata).region", "$region = '"+region+"'", 1)
		}
	}

	// Plug-in the compatible 'Properties' block back to the document.
	properties[0] = parsedDocumentProperties
	var documentProperties interface{} = properties
	parsedMessage.DocumentContent.RuntimeConfig[appconfig.PluginNameAwsRunScript].Properties = documentProperties

	// For debug purposes.
	parsedMessageContent, err := jsonutil.Marshal(parsedMessage)
	if err != nil {
		log.Errorf("Error marshalling %v document. error: %v", parsedMessage.DocumentName, err)
		return parsedMessage, err
	}
	log.Debug("ParsedMessage after removing dependency on instance metadata for managed instance is ", jsonutil.Indent(parsedMessageContent))
	return parsedMessage, nil
}

// Execute starts the scheduling of the message processor plugin
func (p *Processor) Execute(context context.T) (err error) {

	log := p.context.Log()
	log.Infof("starting mdsprocessor polling")
	//process the older messages from Current & Pending folder
	p.processOlderMessages()

	if p.messagePollJob, err = scheduler.Every(pollMessageFrequencyMinutes).Minutes().Run(p.loop); err != nil {
		context.Log().Errorf("unable to schedule message processor. %v", err)
	}
	return
}

// Execute runs multiple sets of commands and returns their outputs.
// res.Output will contain a slice of RunCommandPluginOutput.
func (p *Plugin) Execute(context context.T, config contracts.Configuration, cancelFlag task.CancelFlag) (res contracts.PluginResult) {
	log := context.Log()
	log.Info("RunCommand started with configuration ", config)
	util := new(updateutil.Utility)
	manager := new(updateManager)

	res.StartDateTime = time.Now()
	defer func() { res.EndDateTime = time.Now() }()

	//loading Properties as list since aws:updateSsmAgent uses properties as list
	var properties []interface{}
	if properties, res = pluginutil.LoadParametersAsList(log, config.Properties); res.Code != 0 {
		return res
	}

	out := make([]UpdatePluginOutput, len(properties))
	for i, prop := range properties {
		// check if a reboot has been requested
		if rebooter.RebootRequested() {
			log.Info("A plugin has requested a reboot.")
			break
		}

		if cancelFlag.ShutDown() {
			out[i] = UpdatePluginOutput{}
			out[i].Errors = []string{"Execution canceled due to ShutDown"}
			break
		} else if cancelFlag.Canceled() {
			out[i] = UpdatePluginOutput{}
			out[i].Errors = []string{"Execution canceled"}
			break
		}

		out[i] = updateAgent(p,
			config,
			log,
			manager,
			util,
			prop,
			cancelFlag,
			config.OutputS3BucketName,
			config.OutputS3KeyPrefix,
			res.StartDateTime)

		res.Code = out[i].ExitCode
		res.Status = out[i].Status
		res.Output = fmt.Sprintf("%v", out[i].String())
	}

	return
}

// Execute runs multiple sets of commands and returns their outputs.
// res.Output will contain a slice of RunCommandPluginOutput.
func (p *Plugin) Execute(context context.T, config contracts.Configuration, cancelFlag task.CancelFlag) (res contracts.PluginResult) {
	log := context.Log()
	log.Infof("%v started with configuration %v", Name(), config)
	res.StartDateTime = time.Now()
	defer func() { res.EndDateTime = time.Now() }()

	//loading Properties as list since aws:runPowershellScript & aws:runShellScript uses properties as list
	var properties []interface{}
	if properties, res = pluginutil.LoadParametersAsList(log, config.Properties); res.Code != 0 {

		pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)
		return res
	}

	out := make([]contracts.PluginOutput, len(properties))
	for i, prop := range properties {
		// check if a reboot has been requested
		if rebooter.RebootRequested() {
			log.Info("A plugin has requested a reboot.")
			return
		}

		if cancelFlag.ShutDown() {
			out[i] = contracts.PluginOutput{Errors: []string{"Execution canceled due to ShutDown"}}
			out[i].ExitCode = 1
			out[i].Status = contracts.ResultStatusFailed
			break
		}

		if cancelFlag.Canceled() {
			out[i] = contracts.PluginOutput{Errors: []string{"Execution canceled"}}
			out[i].ExitCode = 1
			out[i].Status = contracts.ResultStatusCancelled
			break
		}

		out[i] = p.runCommandsRawInput(log, prop, config.OrchestrationDirectory, cancelFlag, config.OutputS3BucketName, config.OutputS3KeyPrefix)
	}

	// TODO: instance here we have to do more result processing, where individual sub properties results are merged smartly into plugin response.
	// Currently assuming we have only one work.
	if len(properties) > 0 {
		res.Code = out[0].ExitCode
		res.Status = out[0].Status
		res.Output = out[0].String()
	}

	pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)

	return res
}

func runPlugin(
	context context.T,
	p plugin.T,
	pluginID string,
	config contracts.Configuration,
	cancelFlag task.CancelFlag,
) (res contracts.PluginResult) {
	// create a new context that includes plugin ID
	context = context.With("[pluginID=" + pluginID + "]")

	log := context.Log()
	defer func() {
		// recover in case the plugin panics
		// this should handle some kind of seg fault errors.
		if err := recover(); err != nil {
			res.Status = contracts.ResultStatusFailed
			res.Code = 1
			res.Error = fmt.Errorf("Plugin crashed with message %v!", err)
			log.Error(res.Error)
		}
	}()
	log.Debug("Running plugin")
	return p.Execute(context, config, cancelFlag)
}

// GetUpdatePluginConfig returns the default values for the update plugin
func GetUpdatePluginConfig(context context.T) UpdatePluginConfig {
	log := context.Log()
	region, err := platform.Region()
	if err != nil {
		log.Errorf("Error retrieving agent region in update plugin config. error: %v", err)
	}

	var manifestUrl string
	if region == "cn-north-1" {
		manifestUrl = "https://s3.cn-north-1.amazonaws.com.cn/amazon-ssm-cn-north-1/ssm-agent-manifest.json"
	} else {
		manifestUrl = "https://amazon-ssm-{Region}.s3.amazonaws.com/ssm-agent-manifest.json"
	}

	return UpdatePluginConfig{
		ManifestLocation:      manifestUrl,
		StdoutFileName:        "stdout",
		StderrFileName:        "stderr",
		MaxStdoutLength:       2500,
		MaxStderrLength:       2500,
		OutputTruncatedSuffix: "--output truncated--",
	}
}

// loadPlatformDependentPlugins registers platform dependent plugins
func loadPlatformDependentPlugins(context context.T) PluginRegistry {
	log := context.Log()
	var workerPlugins = PluginRegistry{}

	// registering aws:psModule plugin
	psModulePluginName := psmodule.Name()
	psModulePlugin, err := psmodule.NewPlugin(pluginutil.DefaultPluginConfig())
	if err != nil {
		log.Errorf("failed to create plugin %s %v", psModulePluginName, err)
	} else {
		workerPlugins[psModulePluginName] = psModulePlugin
	}

	// registering aws:applications plugin
	applicationPluginName := application.Name()
	applicationPlugin, err := application.NewPlugin(pluginutil.DefaultPluginConfig())
	if err != nil {
		log.Errorf("failed to create plugin %s %v", applicationPluginName, err)
	} else {
		workerPlugins[applicationPluginName] = applicationPlugin
	}

	return workerPlugins
}

// loadPlatformIndependentPlugins registers plugins common to all platforms
func loadPlatformIndependentPlugins(context context.T) PluginRegistry {
	log := context.Log()
	var workerPlugins = PluginRegistry{}

	// registering aws:runPowerShellScript & aws:runShellScript plugin
	runcommandPluginName := runcommand.Name()
	runcommandPlugin, err := runcommand.NewPlugin(pluginutil.DefaultPluginConfig())
	if err != nil {
		log.Errorf("failed to create plugin %s %v", runcommandPluginName, err)
	} else {
		workerPlugins[runcommandPluginName] = runcommandPlugin
	}

	// registering aws:updateSsmAgent plugin
	updateAgentPluginName := updatessmagent.Name()
	updateAgentPlugin, err := updatessmagent.NewPlugin(updatessmagent.GetUpdatePluginConfig(context))
	if err != nil {
		log.Errorf("failed to create plugin %s %v", updateAgentPluginName, err)
	} else {
		workerPlugins[updateAgentPluginName] = updateAgentPlugin
	}

	return workerPlugins
}

// Execute starts the scheduling of the health check plugin
func (h *HealthCheck) Execute(context context.T) (err error) {
	if h.healthJob, err = scheduler.Every(h.scheduleInMinutes()).Minutes().Run(h.updateHealth); err != nil {
		context.Log().Errorf("unable to schedule health update. %v", err)
	}
	return
}

// Execute runs multiple sets of commands and returns their outputs.
// res.Output will contain a slice of PluginOutput.
func (p *Plugin) Execute(context context.T, config contracts.Configuration, cancelFlag task.CancelFlag) (res contracts.PluginResult) {
	log := context.Log()
	log.Infof("%v started with configuration %v", Name(), config)
	res.StartDateTime = time.Now()
	defer func() { res.EndDateTime = time.Now() }()

	//loading Properties as list since aws:applications uses properties as list
	var properties []interface{}
	if properties, res = pluginutil.LoadParametersAsList(log, config.Properties); res.Code != 0 {

		pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)
		return res
	}

	msiFailureCount := 0
	atleastOneRequestedReboot := false
	finalStdOut := ""
	finalStdErr := ""
	out := make([]ApplicationPluginOutput, len(properties))
	for i, prop := range properties {
		// check if a reboot has been requested
		if rebooter.RebootRequested() {
			log.Infof("Stopping execution of %v plugin due to an external reboot request.", Name())
			return
		}

		if cancelFlag.ShutDown() {
			res.Code = 1
			res.Status = contracts.ResultStatusFailed
			pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)
			return
		}

		if cancelFlag.Canceled() {
			res.Code = 1
			res.Status = contracts.ResultStatusCancelled
			pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)
			return
		}

		out[i] = p.runCommandsRawInput(log, prop, config.OrchestrationDirectory, cancelFlag, config.OutputS3BucketName, config.OutputS3KeyPrefix)

		if out[i].Status == contracts.ResultStatusFailed {
			msiFailureCount++

			if out[i].Stdout != "" {
				finalStdOut = fmt.Sprintf("%v\n%v", finalStdOut, out[i].Stdout)
			}

			if out[i].Stderr != "" {
				finalStdErr = fmt.Sprintf("%v\n%v", finalStdErr, out[i].Stderr)
			}
		}

		if out[i].Status == contracts.ResultStatusSuccessAndReboot {
			atleastOneRequestedReboot = true
			res.Code = out[i].ExitCode
		}
	}

	if atleastOneRequestedReboot {
		res.Status = contracts.ResultStatusSuccessAndReboot
	} else {
		res.Status = contracts.ResultStatusSuccess
		res.Code = 0
	}

	if msiFailureCount > 0 {
		finalStdOut = fmt.Sprintf("Number of Failures: %v\n%v", msiFailureCount, finalStdOut)
		res.Status = contracts.ResultStatusFailed
		res.Code = 1
	}

	finalOut := contracts.PluginOutput{
		Stdout: finalStdOut,
		Stderr: finalStdErr,
	}

	res.Output = finalOut.String()
	pluginutil.PersistPluginInformationToCurrent(log, Name(), config, res)

	return res
}

// processSendCommandMessage processes a single send command message received from MDS.
func (p *Processor) processSendCommandMessage(context context.T,
	mdsService service.Service,
	messagesOrchestrationRootDir string,
	runPlugins PluginRunner,
	cancelFlag task.CancelFlag,
	buildReply replyBuilder,
	sendResponse engine.SendResponse,
	msg ssmmds.Message) {

	commandID := getCommandID(*msg.MessageId)

	log := context.Log()
	log.Debug("Processing send command message ", *msg.MessageId)
	log.Trace("Processing send command message ", jsonutil.Indent(*msg.Payload))

	parsedMessage, err := parser.ParseMessageWithParams(log, *msg.Payload)
	if err != nil {
		log.Error("format of received message is invalid ", err)
		err = mdsService.FailMessage(log, *msg.MessageId, service.InternalHandlerException)
		if err != nil {
			sdkutil.HandleAwsError(log, err, p.processorStopPolicy)
		}
		return
	}

	parsedMessageContent, _ := jsonutil.Marshal(parsedMessage)
	log.Debug("ParsedMessage is ", jsonutil.Indent(parsedMessageContent))

	// adapt plugin configuration format from MDS to plugin expected format
	s3KeyPrefix := path.Join(parsedMessage.OutputS3KeyPrefix, parsedMessage.CommandID, *msg.Destination)

	messageOrchestrationDirectory := filepath.Join(messagesOrchestrationRootDir, commandID)

	// Check if it is a managed instance and its executing managed instance incompatible AWS SSM public document.
	// A few public AWS SSM documents contain code which is not compatible when run on managed instances.
	// isManagedInstanceIncompatibleAWSSSMDocument makes sure to find such documents at runtime and replace the incompatible code.
	isMI, err := isManagedInstance()
	if err != nil {
		log.Errorf("Error determining managed instance. error: %v", err)
	}

	if isMI && isManagedInstanceIncompatibleAWSSSMDocument(parsedMessage.DocumentName) {
		log.Debugf("Running Incompatible AWS SSM Document %v on managed instance", parsedMessage.DocumentName)
		if parsedMessage, err = removeDependencyOnInstanceMetadataForManagedInstance(context, parsedMessage); err != nil {
			return
		}
	}

	pluginConfigurations := getPluginConfigurations(
		parsedMessage.DocumentContent.RuntimeConfig,
		messageOrchestrationDirectory,
		parsedMessage.OutputS3BucketName,
		s3KeyPrefix,
		*msg.MessageId)

	//persist : all information in current folder
	log.Info("Persisting message in current execution folder")

	//Data format persisted in Current Folder is defined by the struct - CommandState
	interimCmdState := initializeCommandState(pluginConfigurations, msg, parsedMessage)

	// persist new interim command state in current folder
	commandStateHelper.PersistData(log, commandID, *msg.Destination, appconfig.DefaultLocationOfCurrent, interimCmdState)

	//Deleting from pending folder since the command is getting executed
	commandStateHelper.RemoveData(log, commandID, *msg.Destination, appconfig.DefaultLocationOfPending)

	log.Debug("Running plugins...")
	outputs := runPlugins(context, *msg.MessageId, pluginConfigurations, sendResponse, cancelFlag)
	pluginOutputContent, _ := jsonutil.Marshal(outputs)
	log.Debugf("Plugin outputs %v", jsonutil.Indent(pluginOutputContent))

	payloadDoc := buildReply("", outputs)

	//check if document isn't supported by SSM -> update the DocumentLevel status message & send reply accordingly
	ssmDocName := parsedMessage.DocumentName
	if IsDocumentNotSupportedBySsmAgent(ssmDocName) {
		log.Infof("%s is not yet supported by aws-ssm-agent, setting up Document level response accordingly", ssmDocName)
		payloadDoc.DocumentTraceOutput = fmt.Sprintf("%s document is not yet supported by amazon-ssm-agent.", ssmDocName)
		p.sendDocLevelResponse(*msg.MessageId, contracts.ResultStatusFailed, payloadDoc.DocumentTraceOutput)
	}

	//update documentInfo in interim cmd state file
	documentInfo := commandStateHelper.GetDocumentInfo(log, commandID, *msg.Destination, appconfig.DefaultLocationOfCurrent)

	// set document level information which wasn't set previously
	documentInfo.AdditionalInfo = payloadDoc.AdditionalInfo
	documentInfo.DocumentStatus = payloadDoc.DocumentStatus
	documentInfo.DocumentTraceOutput = payloadDoc.DocumentTraceOutput
	documentInfo.RuntimeStatus = payloadDoc.RuntimeStatus

	//persist final documentInfo.
	commandStateHelper.PersistDocumentInfo(log,
		documentInfo,
		commandID,
		*msg.Destination,
		appconfig.DefaultLocationOfCurrent)

	// Skip sending response when the document requires a reboot
//.........这里部分代码省略.........

// runCmdsUsingCmdState takes commandState as an input and executes only those plugins which haven't yet executed. This is functionally
// very similar to processSendCommandMessage because everything to do with cmd execution is part of that function right now.
func (p *Processor) runCmdsUsingCmdState(context context.T,
	mdsService service.Service,
	runPlugins PluginRunner,
	cancelFlag task.CancelFlag,
	buildReply replyBuilder,
	sendResponse engine.SendResponse,
	command messageContracts.CommandState) {

	log := context.Log()
	var pluginConfigurations map[string]*contracts.Configuration
	pendingPlugins := false
	pluginConfigurations = make(map[string]*contracts.Configuration)

	//iterate through all plugins to find all plugins that haven't executed yet.
	for k, v := range command.PluginsInformation {
		if v.HasExecuted {
			log.Debugf("skipping execution of Plugin - %v of command - %v since it has already executed.", k, command.DocumentInformation.CommandID)
		} else {
			log.Debugf("Plugin - %v of command - %v will be executed", k, command.DocumentInformation.CommandID)
			pluginConfigurations[k] = &v.Configuration
			pendingPlugins = true
		}
	}

	//execute plugins that haven't been executed yet
	//individual plugins after execution will update interim cmd state file accordingly
	if pendingPlugins {

		log.Debugf("executing following plugins of command - %v", command.DocumentInformation.CommandID)
		for k := range pluginConfigurations {
			log.Debugf("Plugin: %v", k)
		}

		//Since only some plugins of a cmd gets executed here - there is no need to get output from engine & construct the sendReply output.
		//Instead after all plugins of a command get executed, use persisted data to construct sendReply payload
		runPlugins(context, command.DocumentInformation.MessageID, pluginConfigurations, sendResponse, cancelFlag)
	}

	//read from persisted file
	newCmdState := commandStateHelper.GetCommandInterimState(log,
		command.DocumentInformation.CommandID,
		command.DocumentInformation.Destination,
		appconfig.DefaultLocationOfCurrent)

	//construct sendReply payload
	outputs := make(map[string]*contracts.PluginResult)

	for k, v := range newCmdState.PluginsInformation {
		outputs[k] = &v.Result
	}

	pluginOutputContent, _ := jsonutil.Marshal(outputs)
	log.Debugf("plugin outputs %v", jsonutil.Indent(pluginOutputContent))

	payloadDoc := buildReply("", outputs)

	//update interim cmd state file with document level information
	var documentInfo messageContracts.DocumentInfo

	// set document level information which wasn't set previously
	documentInfo.AdditionalInfo = payloadDoc.AdditionalInfo
	documentInfo.DocumentStatus = payloadDoc.DocumentStatus
	documentInfo.DocumentTraceOutput = payloadDoc.DocumentTraceOutput
	documentInfo.RuntimeStatus = payloadDoc.RuntimeStatus

	//persist final documentInfo.
	commandStateHelper.PersistDocumentInfo(log,
		documentInfo,
		command.DocumentInformation.CommandID,
		command.DocumentInformation.Destination,
		appconfig.DefaultLocationOfCurrent)

	// Skip sending response when the document requires a reboot
	if documentInfo.DocumentStatus == contracts.ResultStatusSuccessAndReboot {
		log.Debug("skipping sending response of %v since the document requires a reboot", newCmdState.DocumentInformation.MessageID)
		return
	}

	//send document level reply
	log.Debug("sending reply on message completion ", outputs)
	sendResponse(command.DocumentInformation.MessageID, "", outputs)

	//persist : commands execution in completed folder (terminal state folder)
	log.Debugf("execution of %v is over. Moving interimState file from Current to Completed folder", newCmdState.DocumentInformation.MessageID)

	commandStateHelper.MoveCommandState(log,
		newCmdState.DocumentInformation.CommandID,
		newCmdState.DocumentInformation.Destination,
		appconfig.DefaultLocationOfCurrent,
		appconfig.DefaultLocationOfCompleted)

	log.Debugf("deleting message")
	isUpdate := false
	for pluginName := range pluginConfigurations {
		if pluginName == appconfig.PluginNameAwsAgentUpdate {
			isUpdate = true
		}
	}
//.........这里部分代码省略.........

// NewProcessor initializes a new mds processor with the given parameters.
func NewProcessor(context context.T) *Processor {
	messageContext := context.With("[" + name + "]")
	log := messageContext.Log()
	config := messageContext.AppConfig()

	instanceID, err := platform.InstanceID()
	if instanceID == "" {
		log.Errorf("no instanceID provided, %v", err)
		return nil
	}

	mdsService := newMdsService(config)

	agentInfo := contracts.AgentInfo{
		Lang:      config.Os.Lang,
		Name:      config.Agent.Name,
		Version:   config.Agent.Version,
		Os:        config.Os.Name,
		OsVersion: config.Os.Version,
	}

	agentConfig := contracts.AgentConfiguration{
		AgentInfo:  agentInfo,
		InstanceID: instanceID,
	}

	// sendCommand and cancelCommand will be processed by separate worker pools
	// so we can define the number of workers per each
	cancelWaitDuration := 10000 * time.Millisecond
	clock := times.DefaultClock
	sendCommandTaskPool := task.NewPool(log, config.Mds.CommandWorkersLimit, cancelWaitDuration, clock)
	cancelCommandTaskPool := task.NewPool(log, CancelWorkersLimit, cancelWaitDuration, clock)

	// create new message processor
	orchestrationRootDir := path.Join(appconfig.DefaultDataStorePath, instanceID, appconfig.DefaultCommandRootDirName, config.Agent.OrchestrationRootDir)

	replyBuilder := func(pluginID string, results map[string]*contracts.PluginResult) messageContracts.SendReplyPayload {
		runtimeStatuses := parser.PrepareRuntimeStatuses(log, results)
		return parser.PrepareReplyPayload(pluginID, runtimeStatuses, clock.Now(), agentConfig.AgentInfo)
	}

	statusReplyBuilder := func(agentInfo contracts.AgentInfo, resultStatus contracts.ResultStatus, documentTraceOutput string) messageContracts.SendReplyPayload {
		return parser.PrepareReplyPayloadToUpdateDocumentStatus(agentInfo, resultStatus, documentTraceOutput)

	}
	// create a stop policy where we will stop after 10 consecutive errors and if time period expires.
	processorStopPolicy := newStopPolicy()

	// SendResponse is used to send response on plugin completion.
	// If pluginID is empty it will send responses of all plugins.
	// If pluginID is specified, response will be sent of that particular plugin.
	sendResponse := func(messageID string, pluginID string, results map[string]*contracts.PluginResult) {
		payloadDoc := replyBuilder(pluginID, results)
		processSendReply(log, messageID, mdsService, payloadDoc, processorStopPolicy)
	}

	// SendDocLevelResponse is used to send document level update
	// Specify a new status of the document
	sendDocLevelResponse := func(messageID string, resultStatus contracts.ResultStatus, documentTraceOutput string) {
		payloadDoc := statusReplyBuilder(agentInfo, resultStatus, documentTraceOutput)
		processSendReply(log, messageID, mdsService, payloadDoc, processorStopPolicy)
	}

	// PersistData is used to persist the data into a bookkeeping folder
	persistData := func(msg *ssmmds.Message, bookkeeping string) {
		commandStateHelper.PersistData(log, getCommandID(*msg.MessageId), *msg.Destination, bookkeeping, *msg)
	}

	return &Processor{
		context:              messageContext,
		stopSignal:           make(chan bool),
		config:               agentConfig,
		service:              mdsService,
		pluginRunner:         pluginRunner,
		sendCommandPool:      sendCommandTaskPool,
		cancelCommandPool:    cancelCommandTaskPool,
		buildReply:           replyBuilder,
		sendResponse:         sendResponse,
		sendDocLevelResponse: sendDocLevelResponse,
		orchestrationRootDir: orchestrationRootDir,
		persistData:          persistData,
		processorStopPolicy:  processorStopPolicy,
	}
}