Add checksum field to wf definition

We have no way to determine whether the workflow definition has
changed or not.
This complicates the work with Mistral for external services.
A 'checksum' field has been added to workflow_definitions_v2
to solve this issue.

Implements: blueprint mistral-add-checksum-field-to-wf-definition

Change-Id: I99d095c6c74c62f134aedfa3a63308a0712db38d

mistral/api/controllers/v2/resources.py

@@ -95,6 +95,7 @@ class Workflow(resource.Resource, ScopedResource):
     interface = types.jsontype
     "input and output of the workflow"
     definition = wtypes.text
+    checksum = wtypes.text
     "workflow text written in Mistral v2 language"
     tags = [wtypes.text]
     scope = SCOPE_TYPES

mistral/db/sqlalchemy/migration/alembic_migrations/versions/043_add_checksum_to_wf_definition.py

@@ -0,0 +1,38 @@
+# Copyright 2023 OpenStack Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""add_checksum_to_wf_definition
+
+Revision ID: 043
+Revises: 042
+Create Date: 2023-10-16 12:02:23.374515
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '043'
+down_revision = '042'
+
+from alembic import op
+from mistral.db.utils import column_exists
+import sqlalchemy as sa
+
+
+def upgrade():
+    if not column_exists('workflow_definitions_v2', 'checksum'):
+        op.add_column(
+            'workflow_definitions_v2',
+            sa.Column('checksum', sa.String(length=32), nullable=True)
+        )

mistral/db/utils.py

@@ -17,7 +17,9 @@ import decorator
 import functools
 import inspect
 
+from alembic import op
 from sqlalchemy import exc as sqla_exc
+from sqlalchemy import inspect as ins
 
 from oslo_db import exception as db_exc
 from oslo_log import log as logging
@@ -211,3 +213,10 @@ def tx_cached(use_args=None, ignore_args=None):
         return result
 
     return _decorator
+
+
+def column_exists(table_name, column_name):
+    bind = op.get_context().bind
+    insp = ins(bind)
+    columns = insp.get_columns(table_name)
+    return any(c["name"] == column_name for c in columns)

mistral/db/v2/sqlalchemy/models.py

@@ -156,6 +156,7 @@ class WorkflowDefinition(Definition):
     """Contains info about workflow (including definition in Mistral DSL)."""
 
     __tablename__ = 'workflow_definitions_v2'
+    checksum = sa.Column(sa.String(32), nullable=True)
     __table_args__ = (
         sa.UniqueConstraint(
             'name',

mistral/services/workflows.py

@@ -13,6 +13,7 @@
 #    limitations under the License.
 
 
+import hashlib
 from mistral.db.v2 import api as db_api
 from mistral import exceptions as exc
 from mistral.lang import parser as spec_parser
@@ -183,12 +184,17 @@ def update_workflow_execution_env(wf_ex, env):
     return wf_ex
 
 
+def get_workflow_definition_checksum(definition):
+    return hashlib.md5(definition.encode('utf-8')).hexdigest()
+
+
 def _get_workflow_values(wf_spec, definition, scope, namespace=None,
                          is_system=False):
     values = {
         'name': wf_spec.get_name(),
         'tags': wf_spec.get_tags(),
         'definition': definition,
+        'checksum': get_workflow_definition_checksum(definition),
         'spec': wf_spec.to_dict(),
         'scope': scope,
         'namespace': namespace,

mistral/tests/unit/api/v2/test_workflows.py

@@ -15,6 +15,7 @@
 
 import copy
 import datetime
+import hashlib
 from unittest import mock
 
 import json
@@ -42,7 +43,20 @@ flow:
     task1:
       action: std.echo output="Hi"
 """
+WF_DEFINITION_UPDATED = """
+---
+version: '2.0'
 
+flow:
+  type: direct
+  input:
+    - param1
+    - param2
+
+  tasks:
+    task1:
+      action: std.echo output="Hi"
+"""
 WF_DB = models.WorkflowDefinition(
     id='123e4567-e89b-12d3-a456-426655440000',
     name='flow',
@@ -217,6 +231,7 @@ FIRST_WF = {
     'spec': FIRST_WF_DICT,
     'scope': 'private',
     'namespace': '',
+    'checksum': '1b786ecb96b9358f67718a407c274885',
     'is_system': False
 }
 
@@ -239,6 +254,7 @@ SECOND_WF = {
     'spec': SECOND_WF_DICT,
     'scope': 'private',
     'namespace': '',
+    'checksum': '5803661ccfdf226c95254b2a8a295226',
     'is_system': False
 }
 
@@ -899,3 +915,47 @@ class TestWorkflowsController(base.APITest):
             '/v2/workflows/123e4567-e89b-12d3-a456-426655440000')
         self.assertEqual(200, resp.status_int)
         self.assertIn('project_id', resp.json)
+
+    def test_post_checksum_workflow_added(self):
+        resp = self.app.post(
+            '/v2/workflows',
+            WF_DEFINITION,
+            headers={'Content-Type': 'text/plain'}
+        )
+        body = resp.json
+        checksum = body['workflows'][0]['checksum']
+        self.assertIsNotNone(checksum)
+
+    def test_put_checksum_workflow_updated(self):
+        resp = self.app.post(
+            '/v2/workflows',
+            WF_DEFINITION,
+            headers={'Content-Type': 'text/plain'}
+        )
+        body = resp.json
+        checksum_old = body['workflows'][0]['checksum']
+        resp = self.app.put(
+            '/v2/workflows',
+            WF_DEFINITION_UPDATED,
+            headers={'Content-Type': 'text/plain'}
+        )
+        body = resp.json
+        checksum = body['workflows'][0]['checksum']
+        self.assertTrue(checksum != checksum_old)
+
+    def test_checksum_has_md5_format(self):
+        resp = self.app.post(
+            '/v2/workflows',
+            WF_DEFINITION,
+            headers={'Content-Type': 'text/plain'}
+        )
+        body = resp.json
+        checksum = body['workflows'][0]['checksum']
+        self.assertTrue(self.is_valid_checksum(checksum))
+
+    def is_valid_checksum(self, checksum):
+        try:
+            hashlib.md5(checksum.encode())
+            return True
+        except Exception:
+            return False