func()

in agent/acs/session/payload_responder.go [97:231]
104 lines of code
16 McCabe index (conditional complexity)

func (pmHandler *payloadMessageHandler) addPayloadTasks(payload *ecsacs.PayloadMessage) (
	[]*ecsacs.IAMRoleCredentialsAckRequest, bool) {
	// Verify that we were able to work with all tasks in this payload.
	// This is so we know whether to ACK the whole thing or not.
	allTasksOK := true

	validTasks := make([]*apitask.Task, 0, len(payload.Tasks))
	for _, task := range payload.Tasks {
		if task == nil {
			logger.Critical("Received nil task for message", logger.Fields{
				loggerfield.MessageID: aws.ToString(payload.MessageId),
			})
			allTasksOK = false
			continue
		}

		// Note: If we receive an EBS-backed task, we'll also receive an incomplete volume configuration in the list of Volumes
		// To accommodate this, we'll first check if the task IS EBS-backed then we'll mark the corresponding Volume object to be
		// of type "attachment". This volume object will be replaced by the newly created EBS volume configuration when we parse
		// through the task attachments.
		volName, ok := hasEBSAttachment(task)
		if ok {
			initializeAttachmentTypeVolume(task, volName)
		}

		apiTask, err := apitask.TaskFromACS(task, payload)
		if err != nil {
			pmHandler.handleInvalidTask(task, err, payload)
			allTasksOK = false
			continue
		}

		logger.Info("Received task payload from ACS", logger.Fields{
			loggerfield.TaskARN:       apiTask.Arn,
			loggerfield.TaskVersion:   apiTask.Version,
			loggerfield.DesiredStatus: apiTask.GetDesiredStatus().String(),
		})

		if apiTask.IsFaultInjectionEnabled() {
			logger.Info("Fault Injection Enabled for task", logger.Fields{
				loggerfield.TaskARN: apiTask.Arn,
			})
		}

		if task.RoleCredentials != nil {
			// The payload from ACS for the task has credentials for the
			// task. Add those to the credentials manager and set the
			// credentials id for the task as well.
			taskIAMRoleCredentials := credentials.IAMRoleCredentialsFromACS(task.RoleCredentials,
				credentials.ApplicationRoleType)
			err = pmHandler.credentialsManager.SetTaskCredentials(
				&(credentials.TaskIAMRoleCredentials{
					ARN:                aws.ToString(task.Arn),
					IAMRoleCredentials: taskIAMRoleCredentials,
				}))
			if err != nil {
				pmHandler.handleInvalidTask(task, err, payload)
				allTasksOK = false
				continue
			}
			logger.Info("Found application credentials for task", logger.Fields{
				loggerfield.TaskARN:       apiTask.Arn,
				loggerfield.TaskVersion:   apiTask.Version,
				loggerfield.RoleARN:       taskIAMRoleCredentials.RoleArn,
				loggerfield.RoleType:      taskIAMRoleCredentials.RoleType,
				loggerfield.CredentialsID: taskIAMRoleCredentials.CredentialsID,
			})
			apiTask.SetCredentialsID(taskIAMRoleCredentials.CredentialsID)
		}

		// Add ENI information to the task struct.
		for _, acsENI := range task.ElasticNetworkInterfaces {
			eni, err := ni.InterfaceFromACS(acsENI)
			if err != nil {
				pmHandler.handleInvalidTask(task, err, payload)
				allTasksOK = false
				continue
			}
			apiTask.AddTaskENI(eni)
		}

		// Add the app mesh information to task struct.
		if task.ProxyConfiguration != nil {
			appmesh, err := nlappmesh.AppMeshFromACS(task.ProxyConfiguration)
			if err != nil {
				pmHandler.handleInvalidTask(task, err, payload)
				allTasksOK = false
				continue
			}
			apiTask.SetAppMesh(appmesh)
		}

		if task.ExecutionRoleCredentials != nil {
			// The payload message contains execution credentials for the task.
			// Add the credentials to the credentials manager and set the
			// task executionCredentials id.
			taskExecutionIAMRoleCredentials := credentials.IAMRoleCredentialsFromACS(task.ExecutionRoleCredentials,
				credentials.ExecutionRoleType)
			err = pmHandler.credentialsManager.SetTaskCredentials(
				&(credentials.TaskIAMRoleCredentials{
					ARN:                aws.ToString(task.Arn),
					IAMRoleCredentials: taskExecutionIAMRoleCredentials,
				}))
			if err != nil {
				pmHandler.handleInvalidTask(task, err, payload)
				allTasksOK = false
				continue
			}
			logger.Info("Found execution credentials for task", logger.Fields{
				loggerfield.TaskARN:       apiTask.Arn,
				loggerfield.TaskVersion:   apiTask.Version,
				loggerfield.RoleARN:       taskExecutionIAMRoleCredentials.RoleArn,
				loggerfield.RoleType:      taskExecutionIAMRoleCredentials.RoleType,
				loggerfield.CredentialsID: taskExecutionIAMRoleCredentials.CredentialsID,
			})
			apiTask.SetExecutionRoleCredentialsID(taskExecutionIAMRoleCredentials.CredentialsID)
		}

		validTasks = append(validTasks, apiTask)
	}

	// Add 'stop' transitions first to allow seqnum ordering to work out
	// Because a 'start' sequence number should only be proceeded if all 'stop's
	// of the same sequence number have completed, the 'start' events need to be
	// added after the 'stop' events are there to block them.
	stoppedTasksCredentialsAcks, stoppedTasksAddedOK := pmHandler.addTasks(payload, validTasks, isTaskStatusNotStopped)
	newTasksCredentialsAcks, newTasksAddedOK := pmHandler.addTasks(payload, validTasks, isTaskStatusStopped)
	if !stoppedTasksAddedOK || !newTasksAddedOK {
		allTasksOK = false
	}

	// Construct a slice with credentials acks from all tasks.
	credentialsAcks := append(stoppedTasksCredentialsAcks, newTasksCredentialsAcks...)
	return credentialsAcks, allTasksOK
}