Standardise Workflow messaging and optionally persist messages

This workflow captures the boilerplate that is copied and pasted for Zaqar
messages. It also adds the ability to persist the messages in Swift.

This can be manually tested with:

    $ cat input.json
    {
        "type": "test",
        "queue_name": "tripleo",
        "plan_name": "overcloud",
        "execution": {
            "id": "UUID"
        }
    }
    $ openstack workflow execution create tripleo.messaging.v1.send input.json

Co-Authored-By: Steven Hardy <shardy@redhat.com>
Related-Bug: #1757372
Implements: blueprint config-download-workflows
Depends-On: I78b4fc7673a4039ed7ef32b16bb2850411f7fcc6
Change-Id: Ib854689a3e606f18db7354427918fa8043ac28e0

workbooks/deployment.yaml

@@ -37,25 +37,20 @@ workflows:
           message: <% task().result %>
 
       send_message:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.deploy_on_server
-              payload:
-                status: <% $.get("status", "SUCCESS") %>
-                message: <% $.get("message", "") %>
-                server_uuid: <% $.server_uuid %>
-                server_name: <% $.server_name %>
-                config_name: <% $.config_name %>
-                status_code: <% $.get("status_code", "") %>
-                stdout: <% $.get("stdout", "") %>
-                stderr: <% $.get("stderr", "") %>
-                execution: <% execution() %>
-        on-success:
-          - fail: <% $.get('status') = "FAILED" %>
+          type: <% execution().name %>
+          execution: <% execution() %>
+          status: <% $.get("status", "SUCCESS") %>
+          message: <% $.get("message", "") %>
+          payload:
+            server_uuid: <% $.server_uuid %>
+            server_name: <% $.server_name %>
+            config_name: <% $.config_name %>
+            status_code: <% $.get("status_code", "") %>
+            stdout: <% $.get("stdout", "") %>
+            stderr: <% $.get("stderr", "") %>
 
   deploy_on_servers:
 
@@ -91,30 +86,21 @@ workflows:
           queue_name: <% $.queue_name %>
 
       send_success_message:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.deploy_on_servers
-              payload:
-                status: SUCCESS
-                execution: <% execution() %>
+          type: <% execution().name %>
+          status: SUCCESS
+          execution: <% execution() %>
 
       send_failed_message:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.deploy_on_servers
-              payload:
-                status: FAILED
-                message: <% task(deploy_on_servers).result %>
-                execution: <% execution() %>
-        on-success: fail
+          type: <% execution().name %>
+          status: FAILED
+          execution: <% execution() %>
+          message: <% task(deploy_on_servers).result %>
 
   deploy_plan:
 
@@ -132,6 +118,17 @@ workflows:
       - tripleo-common-managed
 
     tasks:
+      send_start_message:
+        workflow: tripleo.messaging.v1.send
+        input:
+          queue_name: <% $.queue_name %>
+          type: <% execution().name %>
+          status: RUNNING
+          deployment_status: DEPLOYING
+          message: "Workflow Started"
+          execution: <% execution() %>
+          plan_name: <% $.container %>
+        on-complete: add_validation_ssh_key
 
       add_validation_ssh_key:
         workflow: tripleo.validations.v1.add_validation_ssh_key_parameter
@@ -149,34 +146,28 @@ workflows:
             - 'pre-deployment'
           plan: <% $.container %>
           queue_name: <% $.queue_name %>
+        publish-on-error:
+          message: <% task().result %>
         on-success: create_swift_rings_backup_plan
-        on-error: set_validations_failed
-
-      set_validations_failed:
-        on-success: send_message
-        publish:
-          status: FAILED
-          message: <% task(run_validations).result %>
+        on-error: set_deployment_failed
 
       create_swift_rings_backup_plan:
         workflow: tripleo.swift_rings_backup.v1.create_swift_rings_backup_container_plan
-        on-success: cell_v2_discover_hosts
-        on-error: create_swift_rings_backup_plan_set_status_failed
         input:
           container: <% $.container %>
           queue_name: <% $.queue_name %>
           use_default_templates: true
+        publish-on-error:
+          message: <% task().result %>
+        on-success: cell_v2_discover_hosts
+        on-error: set_deployment_failed
 
       cell_v2_discover_hosts:
-        on-success: deploy
-        on-error: cell_v2_discover_hosts_failed
         action: tripleo.baremetal.cell_v2_discover_hosts
-
-      cell_v2_discover_hosts_failed:
-        on-success: send_message
-        publish:
-          status: FAILED
-          message: <% task(cell_v2_discover_hosts).result %>
+        publish-on-error:
+          message: <% task().result %>
+        on-success: deploy
+        on-error: set_deployment_failed
 
       deploy:
         action: tripleo.deployment.deploy
@@ -184,35 +175,33 @@ workflows:
           timeout: <% $.timeout %>
           container: <% $.container %>
           skip_deploy_identifier: <% $.skip_deploy_identifier %>
-        on-success: send_message
+        publish-on-error:
+          message: <% task().result %>
+        on-success: set_deployment_success
         on-error: set_deployment_failed
 
-      create_swift_rings_backup_plan_set_status_failed:
-        on-success: send_message
-        publish:
-          status: FAILED
-          message: <% task(create_swift_rings_backup_plan).result %>
-
       set_deployment_failed:
         on-success: send_message
         publish:
           status: FAILED
-          message: <% task(deploy).result %>
+          deployment_status: DEPLOY_FAILED
+
+      set_deployment_success:
+        on-success: send_message
+        publish:
+          status: SUCCESS
+          deployment_status: DEPLOY_SUCCESS
 
       send_message:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.deploy_plan
-              payload:
-                status: <% $.get('status', 'SUCCESS') %>
-                message: <% $.get('message', '') %>
-                execution: <% execution() %>
-        on-success:
-          - fail: <% $.get('status') = "FAILED" %>
+          type: <% execution().name %>
+          status: <% $.status %>
+          message: <% $.get('message', '') %>
+          execution: <% execution() %>
+          plan_name: <% $.container %>
+          deployment_status: <% $.deployment_status %>
 
   get_horizon_url:
 
@@ -236,26 +225,21 @@ workflows:
           stack_id: <% $.stack %>
         publish:
           horizon_url: <% task().result.outputs.where($.output_key = "EndpointMap").output_value.HorizonPublic.uri.single() %>
-        on-success: notify_zaqar
+        on-success: send_message
         publish-on-error:
           status: FAILED
           message: <% task().result %>
 
-      notify_zaqar:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+      send_message:
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.get_horizon_url
-              payload:
-                horizon_url: <% $.get('horizon_url', '') %>
-                status: <% $.get('status', 'SUCCESS') %>
-                message: <% $.get('message', '') %>
-                execution: <% execution() %>
-        on-success:
-          - fail: <% $.get('status') = "FAILED" %>
+          type: <% execution().name %>
+          status: <% $.get('status', 'SUCCESS') %>
+          execution: <% execution() %>
+          message: <% $.get('message', '') %>
+          payload:
+            horizon_url: <% $.get('horizon_url', '') %>
 
   config_download_deploy:
 
@@ -295,16 +279,14 @@ workflows:
           message: <% task().result %>
 
       send_msg_config_download:
-        action: zaqar.queue_post
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.config_download
-              payload:
-                status: <% $.get('status', 'RUNNING') %>
-                message: Config downloaded at <% $.get('work_dir') %>/<% execution().id %>
-                execution: <% execution() %>
+          type: <% execution().name %>
+          status: <% $.get('status', 'RUNNING') %>
+          execution: <% execution() %>
+          message: Config downloaded at <% $.get('work_dir') %>/<% execution().id %>
+          plan_name: <% $.plan_name %>
         on-success: get_private_key
 
       get_private_key:
@@ -332,32 +314,28 @@ workflows:
           message: <% task().result %>
 
       send_msg_generate_inventory:
-        action: zaqar.queue_post
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.config_download
-              payload:
-                status: <% $.get('status', 'RUNNING') %>
-                message: Inventory generated at <% $.get('inventory') %>
-                execution: <% execution() %>
+          type: <% execution().name %>
+          status: <% $.get('status', 'RUNNING') %>
+          execution: <% execution() %>
+          message: Inventory generated at <% $.get('inventory') %>
+          plan_name: <% $.plan_name %>
         on-success: send_msg_run_ansible
 
       send_msg_run_ansible:
-        action: zaqar.queue_post
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.config_download
-              payload:
-                status: <% $.get('status', 'RUNNING') %>
-                message: >
-                  Running ansible playbook at <% $.get('work_dir') %>/<% execution().id %>/deploy_steps_playbook.yaml.
-                  See log file at <% $.get('work_dir') %>/<% execution().id %>/ansible.log for progress.
-                  ...
-                execution: <% execution() %>
+          type: <% execution().name %>
+          status: <% $.get('status', 'RUNNING') %>
+          execution: <% execution() %>
+          plan_name: <% $.plan_name %>
+          message: >
+            Running ansible playbook at <% $.get('work_dir') %>/<% execution().id %>/deploy_steps_playbook.yaml.
+            See log file at <% $.get('work_dir') %>/<% execution().id %>/ansible.log for progress.
+            ...
         on-success: run_ansible
 
       run_ansible:
@@ -397,19 +375,14 @@ workflows:
           message: Ansible failed, check log at <% $.get('work_dir') %>/<% execution().id %>/ansible.log.
 
       send_message:
-        action: zaqar.queue_post
-        retry: count=5 delay=1
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.config_download
-              payload:
-                status: <% $.get('status', 'SUCCESS') %>
-                message: <% $.get('message', '') %>
-                execution: <% execution() %>
-        on-success:
-          - fail: <% $.get('status') = "FAILED" %>
+          type: <% execution().name %>
+          status: <% $.get('status', 'SUCCESS') %>
+          execution: <% execution() %>
+          message: <% $.get('message', '') %>
+          plan_name: <% $.plan_name %>
 
   create_overcloudrc:
     description: >
@@ -428,7 +401,7 @@ workflows:
         action: tripleo.deployment.overcloudrc
         input:
           container: <% $.container %>
-        on-complete: notify_zaqar
+        on-complete: send_message
         publish-on-error:
           status: FAILED
           message: <% task().result %>
@@ -436,16 +409,11 @@ workflows:
           status: SUCCESS
           message: <% task().result %>
 
-      notify_zaqar:
-        action: zaqar.queue_post
+      send_message:
+        workflow: tripleo.messaging.v1.send
         input:
           queue_name: <% $.queue_name %>
-          messages:
-            body:
-              type: tripleo.deployment.v1.create_overcloudrc
-              payload:
-                status: <% $.status %>
-                message: <% $.get('message', '') %>
-                execution: <% execution() %>
-        on-success:
-          - fail: <% $.get('status') = "FAILED" %>
+          type: <% execution().name %>
+          status: <% $.status %>
+          execution: <% execution() %>
+          message: <% $.get('message', '') %>

workbooks/messaging.yaml

@@ -0,0 +1,151 @@
+---
+version: '2.0'
+name: tripleo.messaging.v1
+description: TripleO Zaqar Workflows
+
+workflows:
+
+  send:
+
+    description: >-
+        Send a message to a Zaqar queue and optionally persist it to Swift
+
+        This workflow sends a standard message to Zaqar (taking care of the
+        error handling and retry logic) and optionally persists the message
+        to Swift. The output of the workflow directly matches the input, this
+        means it can be used to send a message and set the output of the parent
+        calling workflow.
+
+        If plan_name is provided the message will be persisted to Swift. In a
+        container named "{plan_name}-messages". The swift objects will be
+        named:
+            {TYPE}/{TIMESTAMP}.yaml
+
+        If a deployment_status is provided, the top-level deployment_status.yaml
+        will also be updated, which will contain the deployment_status and the
+        related message.
+
+        The standard message format will be:
+
+            body: {
+              type: 'tripleo.workflow.name', # Matches the workflow name
+              payload: {
+                status: 'STATUS', # One of RUNNING, SUCCESS, FAILED
+                execution: {execution details}, # Deprecated and will be removed in the future.
+                execution_id: 'UUID',
+                message: "Human readable description",
+                < Arbitrary data. This should match the workflow output data >
+              }
+            }
+
+        Workflow Input:
+          queue_name - The Zaqar queue name to post to.
+          type - The message type, this should match the calling workflows name
+          execution - Details about the workflow execution. Should be passed by using <% execution() %>
+          status - Optional. The status of the message. SUCCESS/RUNNING/FAILED.
+          message - Optional. A human readable message to be included
+          payload - Optional. A dictionary output data to be sent in the message.
+          plan_name - Optional. The deployment plan name. This is used for the swift messages container.
+          deployment_status - Optional. If set the top-level deployment_status.yaml will be updated.
+
+    input:
+      - queue_name
+      - type
+      - execution
+      - status: 'SUCCESS'
+      - message: null
+      - payload: {}
+      - plan_name: null
+      - deployment_status: null
+
+
+    tags:
+      - tripleo-common-managed
+
+    output:
+      type: <% $.type %>
+      payload: <% $.payload %>
+
+    tasks:
+
+      merge_payload:
+        on-success:
+          publish:
+            branch:
+              # The payload with arbitrary keys is merged with the status, message and execution.
+              payload: <% {status => $.status, message => $.message, execution => $.execution, execution_id => $.execution.id, plan_name => $.plan_name, deployment_status => $.deployment_status} + $.payload %>
+          next: prepare_messages
+
+      prepare_messages:
+        on-success:
+          publish:
+            branch:
+              swift_message: <% {type => $.type, payload => $.payload} %>
+              deployment_status_message: <% {deployment_status => $.deployment_status, workflow_status => {type => $.type, payload => $.payload}} %>
+              container: <% "{0}-messages".format($.plan_name) %>
+          next: branch_workflow
+
+      # It should be possible for this to happen in the next section above, but
+      # there seems to be a Mistral bug... to be confirmed...
+      branch_workflow:
+        on-success:
+          - send_message
+          - complete_swift: <% not bool($.plan_name) %>
+          - verify_container_exists: <% bool($.plan_name) %>
+
+      send_message:
+        action: zaqar.queue_post
+        retry: count=5 delay=1
+        input:
+          queue_name: <% $.queue_name %>
+          messages:
+            body: <% {type => $.type, payload => $.payload} %>
+        on-success: check_status
+
+      verify_container_exists:
+        action: swift.head_container container=<% $.container %>
+        on-error:
+          - create_container
+        on-success:
+          - wait_for_swift: <% not bool($.deployment_status) %>
+          - persist_to_swift_plan_latest: <% bool($.deployment_status) %>
+          - persist_to_swift
+
+      create_container:
+        action: swift.put_container container=<% $.container %>
+        on-success:
+          - wait_for_swift: <% not bool($.deployment_status) %>
+          - persist_to_swift_plan_latest: <% bool($.deployment_status) %>
+          - persist_to_swift
+
+      persist_to_swift:
+        action: swift.put_object
+        input:
+          container: <% $.container %>
+          obj: <% "{0}/{1}.yaml".format($.type, now().format("%Y-%m-%d_%H:%M:%S")) %>
+          contents: <% yaml_dump($.swift_message) %>
+        on-success: wait_for_swift
+
+      persist_to_swift_plan_latest:
+        action: swift.put_object
+        input:
+          container: <% $.container %>
+          obj: <% "deployment_status.yaml" %>
+          contents: <% yaml_dump($.deployment_status_message) %>
+        on-success: wait_for_swift
+
+      wait_for_swift:
+        # We want persist_to_swift and either persist_to_swift_plan_latest or
+        # create_container to join here. Two of the three tasks.
+        join: 2
+        on-success: complete_swift
+
+      complete_swift:
+        on-success: check_status
+
+      check_status:
+        # We want both complete_swift and send_message to join here. This means
+        # that zaqar and swift (if enabled) will all be finished.
+        join: all
+        on-complete:
+          - fail(msg="Workflow failed due to message status"): <% $.get('status') = "FAILED" %>