ms/blueprintsprocessor/functions/ansible-awx-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/ansible/executor/ComponentRemoteAnsibleExecutor.kt

   1 /*
   2  *  Copyright © 2019 Bell Canada.
   3  *  Modifications Copyright © 2018-2019 IBM.
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  */
  17
  18 package org.onap.ccsdk.cds.blueprintsprocessor.functions.ansible.executor
  19
  20 import com.fasterxml.jackson.databind.JsonNode
  21 import com.fasterxml.jackson.databind.ObjectMapper
  22 import com.fasterxml.jackson.databind.node.TextNode
  23 import kotlinx.coroutines.delay
  24 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
  25 import org.onap.ccsdk.cds.blueprintsprocessor.rest.service.BluePrintRestLibPropertyService
  26 import org.onap.ccsdk.cds.blueprintsprocessor.rest.service.BlueprintWebClientService
  27 import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractComponentFunction
  28 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonNode
  29 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
  30 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
  31 import org.onap.ccsdk.cds.controllerblueprints.core.isNullOrMissing
  32 import org.onap.ccsdk.cds.controllerblueprints.core.returnNullIfMissing
  33 import org.onap.ccsdk.cds.controllerblueprints.core.rootFieldsToMap
  34 import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
  35 import org.slf4j.LoggerFactory
  36 import org.springframework.beans.factory.config.ConfigurableBeanFactory
  37 import org.springframework.context.annotation.Scope
  38 import org.springframework.http.HttpMethod
  39 import org.springframework.stereotype.Component
  40 import java.net.URI
  41 import java.net.URLEncoder
  42 import java.util.NoSuchElementException
  43
  44 /**
  45  * ComponentRemoteAnsibleExecutor
  46  *
  47  * Component that launches a run of a job template (INPUT_JOB_TEMPLATE_NAME) representing an Ansible playbook,
  48  * and its parameters, via the AWX server identified by the INPUT_ENDPOINT_SELECTOR parameter.
  49  *
  50  * It supports extra_vars, limit, tags, skip-tags, inventory (by name or Id) Ansible parameters.
  51  * It reports the results of the execution via properties, named execute-command-status and execute-command-logs
  52  *
  53  * @author Serge Simard
  54  */
  55 @Component("component-remote-ansible-executor")
  56 @Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  57 open class ComponentRemoteAnsibleExecutor(
  58     private val blueprintRestLibPropertyService: BluePrintRestLibPropertyService,
  59     private val mapper: ObjectMapper
  60 ) :
  61     AbstractComponentFunction() {
  62
  63     // HTTP related constants
  64     private val HTTP_SUCCESS = 200..202
  65     private val GET = HttpMethod.GET.name
  66     private val POST = HttpMethod.POST.name
  67     private val plainTextHeaders = mapOf("Accept" to "text/plain")
  68
  69     var checkDelay: Long = 15_000
  70
  71     companion object {
  72
  73         private val log = LoggerFactory.getLogger(ComponentRemoteAnsibleExecutor::class.java)
  74
  75         // input fields names accepted by this executor
  76         const val INPUT_ENDPOINT_SELECTOR = "endpoint-selector"
  77         const val INPUT_JOB_TEMPLATE_NAME = "job-template-name"
  78         const val ANSIBLE_FIRE_FAILURE = "ansible-fire-failure"
  79         const val ANSIBLE_FAILED_STATUS = "failed"
  80         const val INPUT_WORKFLOW_JOB_TEMPLATE_NAME = "workflow-job-template-id"
  81         const val INPUT_LIMIT_TO_HOST = "limit"
  82         const val INPUT_INVENTORY = "inventory"
  83         const val INPUT_EXTRA_VARS = "extra-vars"
  84         const val INPUT_TAGS = "tags"
  85         const val INPUT_SKIP_TAGS = "skip-tags"
  86
  87         // output fields names (and values) populated by this executor; aligned with job details status field values.
  88         const val ATTRIBUTE_EXEC_CMD_ARTIFACTS = "ansible-artifacts"
  89         const val ATTRIBUTE_EXEC_CMD_STATUS = "ansible-command-status"
  90         const val ATTRIBUTE_EXEC_CMD_LOG = "ansible-command-logs"
  91         const val ATTRIBUTE_EXEC_CMD_STATUS_ERROR = "error"
  92     }
  93
  94     override suspend fun processNB(executionRequest: ExecutionServiceInput) {
  95
  96         try {
  97             val restClientService = getAWXRestClient()
  98
  99             // Get either a job template name or a workflow template name property
 100             var workflowURIPrefix = ""
 101             var jobTemplateName = getOperationInput(INPUT_JOB_TEMPLATE_NAME).returnNullIfMissing()?.textValue() ?: ""
 102             val isWorkflowJT = jobTemplateName.isBlank()
 103             if (isWorkflowJT) {
 104                 jobTemplateName = getOperationInput(INPUT_WORKFLOW_JOB_TEMPLATE_NAME).asText()
 105                 workflowURIPrefix = "workflow_"
 106             }
 107             var isAnsibleFireFailure = false
 108             if (getOptionalOperationInput(ANSIBLE_FIRE_FAILURE) != null) {
 109                 isAnsibleFireFailure = getOperationInput(ANSIBLE_FIRE_FAILURE).asBoolean()
 110             }
 111
 112             val jtId = lookupJobTemplateIDByName(restClientService, jobTemplateName, workflowURIPrefix)
 113             if (jtId.isNotEmpty()) {
 114                 runJobTemplateOnAWX(restClientService, jobTemplateName, jtId, workflowURIPrefix, isAnsibleFireFailure)
 115             } else {
 116                 val message = "Workflow/Job template $jobTemplateName does not exists"
 117                 log.error(message)
 118                 setNodeOutputErrors(ATTRIBUTE_EXEC_CMD_STATUS_ERROR, message)
 119             }
 120         } catch (e: Exception) {
 121             log.error("Failed to process on remote executor (${e.message})", e)
 122             setNodeOutputErrors(ATTRIBUTE_EXEC_CMD_STATUS_ERROR, "Failed to process on remote executor (${e.message})")
 123         }
 124     }
 125
 126     override suspend fun recoverNB(runtimeException: RuntimeException, executionRequest: ExecutionServiceInput) {
 127         val message = "Error in ComponentRemoteAnsibleExecutor : ${runtimeException.message}"
 128         log.error(message, runtimeException)
 129         setNodeOutputErrors(ATTRIBUTE_EXEC_CMD_STATUS_ERROR, message)
 130     }
 131
 132     /** Creates a TokenAuthRestClientService, since this executor expect type property to be "token-auth" and the
 133      * token to be an OAuth token (access_token response field) generated via the AWX /api/o/token rest endpoint
 134      * The token field is of the form "Bearer access_token_from_response", for example :
 135      *  "blueprintsprocessor.restclient.awx.type=token-auth"
 136      *  "blueprintsprocessor.restclient.awx.url=http://awx-endpoint"
 137      *  "blueprintsprocessor.restclient.awx.token=Bearer J9gEtMDzxcqw25574fioY9VAhLDIs1"
 138      *
 139      * Also supports json endpoint definition via DSL entry, e.g.:
 140      *     "ansible-remote-endpoint": {
 141      *        "type": "token-auth",
 142      *        "url": "http://awx-endpoint",
 143      *        "token": "Bearer J9gEtMDzxcqw25574fioY9VAhLDIs1"
 144      *     }
 145      */
 146     private fun getAWXRestClient(): BlueprintWebClientService {
 147
 148         val endpointSelector = getOperationInput(INPUT_ENDPOINT_SELECTOR)
 149
 150         try {
 151             return blueprintRestLibPropertyService.blueprintWebClientService(endpointSelector)
 152         } catch (e: NoSuchElementException) {
 153             throw IllegalArgumentException("No value provided for input selector $endpointSelector", e)
 154         }
 155     }
 156
 157     /**
 158      * Finds the job template ID based on the job template name provided in the request
 159      */
 160     private fun lookupJobTemplateIDByName(
 161         awxClient: BlueprintWebClientService,
 162         job_template_name: String?,
 163         workflowPrefix: String
 164     ): String {
 165         val encodedJTName = URI(
 166             null, null,
 167             "/api/v2/${workflowPrefix}job_templates/$job_template_name/",
 168             null, null
 169         ).rawPath
 170
 171         // Get Job Template details by name
 172         var response = awxClient.exchangeResource(GET, encodedJTName, "")
 173         val jtDetails: JsonNode = mapper.readTree(response.body)
 174         return jtDetails.at("/id").asText()
 175     }
 176
 177     /**
 178      * Performs the job template execution on AWX, ie. prepare arguments as per job template
 179      * requirements (ask fields) and provided overriding values. Then it launches the run, and monitors
 180      * its execution. Finally, it retrieves the job results via the stdout api.
 181      * The status and output attributes are populated in the process.
 182      */
 183     private suspend fun runJobTemplateOnAWX(
 184         awxClient: BlueprintWebClientService,
 185         job_template_name: String?,
 186         jtId: String,
 187         workflowPrefix: String,
 188         isAnsibleFireFailure: Boolean
 189     ) {
 190         setNodeOutputProperties("preparing".asJsonPrimitive(), "".asJsonPrimitive(), "".asJsonPrimitive())
 191
 192         // Get Job Template requirements
 193         var response = awxClient.exchangeResource(GET, "/api/v2/${workflowPrefix}job_templates/$jtId/launch/", "")
 194         // FIXME: handle non-successful SC
 195         val jtLaunchReqs: JsonNode = mapper.readTree(response.body)
 196         val payload = prepareLaunchPayload(awxClient, jtLaunchReqs, workflowPrefix.isNotBlank())
 197
 198         log.info("Running job with $payload, for requestId $processId.")
 199
 200         // Launch the job for the targeted template
 201         var jtLaunched: JsonNode = JacksonUtils.objectMapper.createObjectNode()
 202         response = awxClient.exchangeResource(POST, "/api/v2/${workflowPrefix}job_templates/$jtId/launch/", payload)
 203         if (response.status in HTTP_SUCCESS) {
 204             jtLaunched = mapper.readTree(response.body)
 205             val fieldsIgnored: JsonNode = jtLaunched.at("/ignored_fields")
 206             if (fieldsIgnored.rootFieldsToMap().isNotEmpty()) {
 207                 log.warn("Ignored fields : $fieldsIgnored, for requestId $processId.")
 208             }
 209         }
 210
 211         if (response.status in HTTP_SUCCESS) {
 212             val jobId: String = jtLaunched.at("/id").asText()
 213
 214             // Poll current job status while job is not executed
 215             var jobStatus = "unknown"
 216             var jobEndTime = "null"
 217             while (jobEndTime == "null") {
 218                 response = awxClient.exchangeResource(GET, "/api/v2/${workflowPrefix}jobs/$jobId/", "")
 219                 val jobLaunched: JsonNode = mapper.readTree(response.body)
 220                 jobStatus = jobLaunched.at("/status").asText()
 221                 jobEndTime = jobLaunched.at("/finished").asText()
 222                 delay(checkDelay)
 223             }
 224
 225             log.info("Execution of job template $job_template_name in job #$jobId finished with status ($jobStatus) for requestId $processId")
 226
 227             if (isAnsibleFireFailure && jobStatus == ANSIBLE_FAILED_STATUS) {
 228                 val message = "Execution of job template $job_template_name failed for requestId $processId." + " (Response: ${response.body}) "
 229                 log.error(message)
 230                 setNodeOutputErrors(ATTRIBUTE_EXEC_CMD_STATUS_ERROR, message)
 231             }
 232
 233             populateJobRunResponse(awxClient, jobId, workflowPrefix, jobStatus)
 234         } else {
 235             // The job template requirements were not fulfilled with the values passed in. The message below will
 236             // provide more information via the response, like the ignored_fields, or variables_needed_to_start,
 237             // or resources_needed_to_start, in order to help user pinpoint the problems with the request.
 238             val message = "Execution of job template $job_template_name could not be started for requestId $processId." +
 239                 " (Response: ${response.body}) "
 240             log.error(message)
 241             setNodeOutputErrors(ATTRIBUTE_EXEC_CMD_STATUS_ERROR, message)
 242         }
 243     }
 244
 245     /**
 246      * Extracts the output from either a job stdout call OR collects the workflow run output, as well as the artifacts
 247      * and populate the component corresponding output properties
 248      */
 249     private fun populateJobRunResponse(
 250         awxClient: BlueprintWebClientService,
 251         jobId: String,
 252         workflowPrefix: String,
 253         jobStatus: String
 254     ) {
 255
 256         val collectedResponses = StringBuilder(4096)
 257         val artifacts: MutableMap<String, JsonNode> = mutableMapOf()
 258
 259         collectJobIdsRelatedToJobRun(awxClient, jobId, workflowPrefix).forEach { aJobId ->
 260
 261             // Collect the response text from the corresponding jobIds
 262             var response = awxClient.exchangeResource(GET, "/api/v2/jobs/$aJobId/stdout/?format=txt", "", plainTextHeaders)
 263             if (response.status in HTTP_SUCCESS) {
 264                 val jobOutput = response.body
 265                 collectedResponses
 266                     .append("Output for Job $aJobId :" + System.lineSeparator())
 267                     .append(jobOutput)
 268                     .append(System.lineSeparator())
 269                 log.info("Response for job $aJobId: \n $jobOutput \n")
 270             } else {
 271                 log.warn("Could not gather response for job $aJobId. Status=${response.status}")
 272             }
 273
 274             // Collect artifacts variables from each job and gather them up in one json node
 275             response = awxClient.exchangeResource(GET, "/api/v2/jobs/$aJobId/", "")
 276             if (response.status in HTTP_SUCCESS) {
 277                 val jobArtifacts = mapper.readTree(response.body).at("/artifacts")
 278                 if (jobArtifacts != null) {
 279                     artifacts.putAll(jobArtifacts.rootFieldsToMap())
 280                 }
 281             }
 282         }
 283
 284         log.info("Artifacts for job $jobId: \n $artifacts \n")
 285
 286         setNodeOutputProperties(jobStatus.asJsonPrimitive(), collectedResponses.toString().asJsonPrimitive(), artifacts.asJsonNode())
 287     }
 288
 289     /**
 290      * List all the job Ids for a give workflow, i.e. sub jobs, or the jobId if not a workflow instance
 291      */
 292     private fun collectJobIdsRelatedToJobRun(awxClient: BlueprintWebClientService, jobId: String, workflowPrefix: String): Array<String> {
 293
 294         var jobIds: Array<String>
 295
 296         if (workflowPrefix.isNotEmpty()) {
 297             var response = awxClient.exchangeResource(GET, "/api/v2/${workflowPrefix}jobs/$jobId/workflow_nodes/", "")
 298             val jobDetails = mapper.readTree(response.body).at("/results")
 299
 300             // gather up job Id of all actual job nodes that ran during the workflow
 301             jobIds = emptyArray()
 302             for (jobDetail in jobDetails.elements()) {
 303                 if (jobDetail.at("/do_not_run").asText() == "false") {
 304                     jobIds = jobIds.plus(jobDetail.at("/summary_fields/job/id").asText())
 305                 }
 306             }
 307         } else {
 308             jobIds = arrayOf(jobId)
 309         }
 310         return jobIds
 311     }
 312
 313     /**
 314      * Prepares the JSON payload expected by the job template api,
 315      * by applying the overrides that were provided
 316      * and allowed by the template definition flags in jtLaunchReqs
 317      */
 318     private fun prepareLaunchPayload(
 319         awxClient: BlueprintWebClientService,
 320         jtLaunchReqs: JsonNode,
 321         isWorkflow: Boolean
 322     ): String {
 323         val payload = JacksonUtils.objectMapper.createObjectNode()
 324
 325         // Parameter defaults
 326         val inventoryProp = getOptionalOperationInput(INPUT_INVENTORY)
 327         val extraArgs = getOperationInput(INPUT_EXTRA_VARS)
 328
 329         if (!isWorkflow) {
 330             val limitProp = getOptionalOperationInput(INPUT_LIMIT_TO_HOST)
 331             val tagsProp = getOptionalOperationInput(INPUT_TAGS)
 332             val skipTagsProp = getOptionalOperationInput(INPUT_SKIP_TAGS)
 333
 334             val askLimitOnLaunch = jtLaunchReqs.at("/ask_limit_on_launch").asBoolean()
 335             if (askLimitOnLaunch && !limitProp.isNullOrMissing()) {
 336                 payload.set<JsonNode>(INPUT_LIMIT_TO_HOST, limitProp)
 337             }
 338             val askTagsOnLaunch = jtLaunchReqs.at("/ask_tags_on_launch").asBoolean()
 339             if (askTagsOnLaunch && !tagsProp.isNullOrMissing()) {
 340                 payload.set<JsonNode>(INPUT_TAGS, tagsProp)
 341             }
 342             if (askTagsOnLaunch && !skipTagsProp.isNullOrMissing()) {
 343                 payload.set<JsonNode>("skip_tags", skipTagsProp)
 344             }
 345         }
 346
 347         val askInventoryOnLaunch = jtLaunchReqs.at("/ask_inventory_on_launch").asBoolean()
 348         if (askInventoryOnLaunch && !inventoryProp.isNullOrMissing()) {
 349             var inventoryKeyId = if (inventoryProp is TextNode) {
 350                 resolveInventoryIdByName(awxClient, inventoryProp.textValue())?.asJsonPrimitive()
 351             } else {
 352                 inventoryProp
 353             }
 354             payload.set<JsonNode>(INPUT_INVENTORY, inventoryKeyId)
 355         }
 356
 357         payload.set<JsonNode>("extra_vars", extraArgs)
 358
 359         return payload.asJsonString(false)
 360     }
 361
 362     private fun resolveInventoryIdByName(awxClient: BlueprintWebClientService, inventoryProp: String): Int? {
 363         var invId: Int? = null
 364
 365         // Get Inventory by name
 366         val encoded = URLEncoder.encode(inventoryProp)
 367         val response = awxClient.exchangeResource(GET, "/api/v2/inventories/?name=$encoded", "")
 368         if (response.status in HTTP_SUCCESS) {
 369             // Extract the inventory ID from response
 370             val invDetails = mapper.readTree(response.body)
 371             val nbInvFound = invDetails.at("/count").asInt()
 372             if (nbInvFound == 1) {
 373                 invId = invDetails["results"][0]["id"].asInt()
 374                 log.info("Resolved inventory $inventoryProp to ID #: $invId")
 375             }
 376         }
 377
 378         if (invId == null) {
 379             val message = "Could not resolve inventory $inventoryProp by name..."
 380             log.error(message)
 381             throw IllegalArgumentException(message)
 382         }
 383
 384         return invId
 385     }
 386
 387     /**
 388      * Utility function to set the output properties of the executor node
 389      */
 390     private fun setNodeOutputProperties(status: JsonNode, message: JsonNode, artifacts: JsonNode) {
 391         setAttribute(ATTRIBUTE_EXEC_CMD_STATUS, status)
 392         log.info("Executor status   : $status")
 393         setAttribute(ATTRIBUTE_EXEC_CMD_ARTIFACTS, artifacts)
 394         log.info("Executor artifacts: $artifacts")
 395         setAttribute(ATTRIBUTE_EXEC_CMD_LOG, message)
 396         log.info("Executor message  : $message")
 397     }
 398
 399     /**
 400      * Utility function to set the output properties and errors of the executor node, in cas of errors
 401      */
 402     private fun setNodeOutputErrors(status: String, message: String, artifacts: JsonNode = "".asJsonPrimitive()) {
 403         setAttribute(ATTRIBUTE_EXEC_CMD_STATUS, status.asJsonPrimitive())
 404         setAttribute(ATTRIBUTE_EXEC_CMD_LOG, message.asJsonPrimitive())
 405         setAttribute(ATTRIBUTE_EXEC_CMD_ARTIFACTS, artifacts)
 406
 407         addError(status, ATTRIBUTE_EXEC_CMD_LOG, message)
 408     }
 409 }