refactor: run_workflow approach & changed reverse ingestion approach …

…similar to automation runner;

ingestion/operators/docker/run_automation.py

@@ -16,7 +16,7 @@
 
 import yaml
 
-from metadata.automations.runner import execute
+from metadata.automations.execute_runner import execute
 from metadata.generated.schema.entity.automations.workflow import (
     Workflow as AutomationWorkflow,
 )

ingestion/src/metadata/automations/collate_runner.py

@@ -0,0 +1,13 @@
+#  Copyright 2021 Collate
+#  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.
+"""
+Run the Automation Workflow for Collate
+"""

ingestion/src/metadata/automations/execute_runner.py

@@ -0,0 +1,49 @@
+#  Copyright 2021 Collate
+#  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.
+"""
+Run the Automation Workflow
+"""
+from functools import singledispatch
+from typing import Any
+
+from metadata.generated.schema.entity.automations.workflow import (
+    Workflow as AutomationWorkflow,
+)
+from metadata.ingestion.ometa.ometa_api import OpenMetadata
+
+
+@singledispatch
+def run_workflow(request: Any, *_, **__) -> Any:
+    """
+    Main entrypoint to execute the automation workflow
+    """
+    raise NotImplementedError(f"Workflow runner not implemented for {type(request)}")
+
+
+def execute(encrypted_automation_workflow: AutomationWorkflow) -> Any:
+    """
+    Execute the automation workflow.
+    The implementation depends on the request body type
+    """
+    # Import all the functions defined for run_workflow
+    import metadata.automations.collate_runner  # pylint: disable=import-outside-toplevel
+    import metadata.automations.runner  # pylint: disable=import-outside-toplevel
+
+    # This will already instantiate the Secrets Manager
+    metadata = OpenMetadata(
+        config=encrypted_automation_workflow.openMetadataServerConnection
+    )
+
+    automation_workflow = metadata.get_by_name(
+        entity=AutomationWorkflow, fqn=encrypted_automation_workflow.name.root
+    )
+
+    return run_workflow(automation_workflow.request, automation_workflow, metadata)

ingestion/src/metadata/automations/runner.py

@@ -9,11 +9,10 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 """
-Run the Automation Workflow
+Run the Automation Workflow for OpenMetadata
 """
-from functools import singledispatch
-from typing import Any
 
+from metadata.automations.execute_runner import run_workflow
 from metadata.generated.schema.entity.automations.testServiceConnection import (
     TestServiceConnectionRequest,
 )
@@ -28,32 +27,6 @@
 from metadata.utils.ssl_manager import SSLManager, check_ssl_and_init
 
 
-def execute(encrypted_automation_workflow: AutomationWorkflow) -> Any:
-    """
-    Execute the automation workflow.
-    The implementation depends on the request body type
-    """
-
-    # This will already instantiate the Secrets Manager
-    metadata = OpenMetadata(
-        config=encrypted_automation_workflow.openMetadataServerConnection
-    )
-
-    automation_workflow = metadata.get_by_name(
-        entity=AutomationWorkflow, fqn=encrypted_automation_workflow.name.root
-    )
-
-    return run_workflow(automation_workflow.request, automation_workflow, metadata)
-
-
-@singledispatch
-def run_workflow(request: Any, *_, **__) -> Any:
-    """
-    Main entrypoint to execute the automation workflow
-    """
-    raise NotImplementedError(f"Workflow runner not implemented for {type(request)}")
-
-
 @run_workflow.register
 def _(
     request: TestServiceConnectionRequest,

ingestion/src/metadata/ingestion/api/parser.py

@@ -111,10 +111,6 @@
     PipelineMetadataConfigType,
     PipelineServiceMetadataPipeline,
 )
-from metadata.generated.schema.metadataIngestion.reverseIngestionPipeline import (
-    ReverseIngestionPipeline,
-    ReverseIngestionType,
-)
 from metadata.generated.schema.metadataIngestion.searchServiceMetadataPipeline import (
     SearchMetadataConfigType,
     SearchServiceMetadataPipeline,
@@ -162,7 +158,6 @@
     StorageMetadataConfigType.StorageMetadata.value: StorageServiceMetadataPipeline,
     SearchMetadataConfigType.SearchMetadata.value: SearchServiceMetadataPipeline,
     DbtConfigType.DBT.value: DbtPipeline,
-    ReverseIngestionType.ReverseIngestion.value: ReverseIngestionPipeline,
 }
 
 DBT_CONFIG_TYPE_MAP = {

ingestion/src/metadata/utils/class_helper.py

@@ -56,10 +56,6 @@
 from metadata.generated.schema.metadataIngestion.pipelineServiceMetadataPipeline import (
     PipelineServiceMetadataPipeline,
 )
-from metadata.generated.schema.metadataIngestion.reverseIngestionPipeline import (
-    ReverseIngestionPipeline,
-    ReverseIngestionType,
-)
 from metadata.generated.schema.metadataIngestion.searchServiceMetadataPipeline import (
     SearchServiceMetadataPipeline,
 )
@@ -86,7 +82,6 @@
     ServiceType.Storage.value: "storageService",
     # We use test suites as "services" for DQ Ingestion Pipelines
     TestSuiteServiceType.TestSuite.value: "testSuite",
-    ReverseIngestionType.ReverseIngestion.value: "reverseIngestion",
 }
 
 SOURCE_CONFIG_TYPE_INGESTION = {
@@ -102,7 +97,6 @@
     StorageServiceMetadataPipeline.__name__: PipelineType.metadata,
     SearchServiceMetadataPipeline.__name__: PipelineType.metadata,
     TestSuitePipeline.__name__: PipelineType.TestSuite,
-    ReverseIngestionPipeline.__name__: PipelineType.reverseIngestion,
     MetadataToElasticSearchPipeline.__name__: PipelineType.elasticSearchReindex,
     DataInsightPipeline.__name__: PipelineType.dataInsight,
     DbtPipeline.__name__: PipelineType.dbt,
@@ -137,8 +131,6 @@ def _get_service_type_from(  # pylint: disable=inconsistent-return-statements
 ) -> ServiceType:
     if service_subtype.lower() == "testsuite":
         return TestSuiteServiceType.TestSuite
-    if service_subtype.lower() == "reverseingestion":
-        return ReverseIngestionType.ReverseIngestion
 
     for service_type in ServiceType:
         if service_subtype.lower() in [

ingestion/src/metadata/utils/service_spec/service_spec.py

@@ -52,7 +52,6 @@ class BaseSpec(BaseModel):
     lineage_source_class: Optional[str] = None
     usage_source_class: Optional[str] = None
     sampler_class: Optional[str] = None
-    reverse_ingestion_source_class: Optional[str] = None
 
     @model_validator(mode="before")
     @classmethod
@@ -95,7 +94,7 @@ def import_source_class(
     service_type: ServiceType, source_type: str, from_: str = "ingestion"
 ) -> Type[Source]:
     source_class_type = source_type.split(TYPE_SEPARATOR)[-1]
-    if source_class_type in ["usage", "lineage", "reverse_ingestion"]:
+    if source_class_type in ["usage", "lineage"]:
         field = f"{source_class_type}_source_class"
     else:
         field = "metadata_source_class"

openmetadata-airflow-apis/openmetadata_managed_apis/api/routes/run_automation.py

@@ -19,7 +19,7 @@
 from openmetadata_managed_apis.utils.logger import routes_logger
 from pydantic import ValidationError
 
-from metadata.automations.runner import execute
+from metadata.automations.execute_runner import execute
 from metadata.ingestion.api.parser import parse_automation_workflow_gracefully
 from metadata.utils.secrets.secrets_manager_factory import SecretsManagerFactory
 

openmetadata-service/src/main/java/org/openmetadata/service/secrets/converter/WorkflowClassConverter.java

@@ -14,9 +14,9 @@
 package org.openmetadata.service.secrets.converter;
 
 import java.util.List;
-import org.openmetadata.schema.api.services.ingestionPipelines.CreateIngestionPipeline;
 import org.openmetadata.schema.entity.automations.TestServiceConnectionRequest;
 import org.openmetadata.schema.entity.automations.Workflow;
+import org.openmetadata.schema.metadataIngestion.ReverseIngestionPipeline;
 import org.openmetadata.schema.services.connections.metadata.OpenMetadataConnection;
 import org.openmetadata.service.util.JsonUtils;
 
@@ -33,7 +33,7 @@ public Object convert(Object object) {
 
     tryToConvertOrFail(
             workflow.getRequest(),
-            List.of(TestServiceConnectionRequest.class, CreateIngestionPipeline.class))
+            List.of(TestServiceConnectionRequest.class, ReverseIngestionPipeline.class))
         .ifPresent(workflow::setRequest);
 
     if (workflow.getOpenMetadataServerConnection() != null) {

openmetadata-spec/src/main/resources/json/schema/api/automations/createWorkflow.json

@@ -32,7 +32,7 @@
           "$ref": "../../entity/automations/testServiceConnection.json"
         },
         {
-          "$ref": "../services/ingestionPipelines/createIngestionPipeline.json"
+          "$ref": "../../metadataIngestion/reverseIngestionPipeline.json"
         }
       ]
     },

openmetadata-spec/src/main/resources/json/schema/entity/automations/workflow.json

@@ -67,7 +67,7 @@
           "$ref": "testServiceConnection.json"
         },
         {
-          "$ref": "../../api/services/ingestionPipelines/createIngestionPipeline.json"
+          "$ref": "../../metadataIngestion/reverseIngestionPipeline.json"
         }
       ]
     },

openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/ingestionPipeline.json

@@ -23,8 +23,7 @@
         "dataInsight",
         "elasticSearchReindex",
         "dbt",
-        "application",
-        "reverseIngestion"
+        "application"
       ]
     },
     "pipelineStatus": {

openmetadata-spec/src/main/resources/json/schema/metadataIngestion/reverseIngestionPipeline.json

@@ -2,7 +2,7 @@
     "$id": "https://open-metadata.org/schema/entity/applications/metadataIngestion/reverseIngestionPipeline.json",
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "reverseIngestionPipeline",
-    "javaType": "org.openmetadata.schema.metadataIngestion.ReverseMetadataOperationRequest",
+    "javaType": "org.openmetadata.schema.metadataIngestion.ReverseIngestionPipeline",
     "description": "Apply a set of operations on a service",
     "type": "object",
     "definitions": {

openmetadata-spec/src/main/resources/json/schema/metadataIngestion/workflow.json

@@ -62,9 +62,6 @@
             },
             {
               "$ref": "apiServiceMetadataPipeline.json"
-            },
-            {
-              "$ref": "reverseIngestionPipeline.json"
             }
           ]
         }
@@ -92,10 +89,7 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "type",
-        "sourceConfig"
-      ]
+      "required": ["type", "sourceConfig"]
     },
     "processor": {
       "description": "Configuration for Processor Component in the OpenMetadata Ingestion Framework.",
@@ -110,9 +104,7 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "type"
-      ]
+      "required": ["type"]
     },
     "stage": {
       "description": "Configuration for Stage Component in the OpenMetadata Ingestion Framework.",
@@ -127,9 +119,7 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "type"
-      ]
+      "required": ["type"]
     },
     "sink": {
       "description": "Configuration for Sink Component in the OpenMetadata Ingestion Framework.",
@@ -144,9 +134,7 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "type"
-      ]
+      "required": ["type"]
     },
     "bulkSink": {
       "description": "Configuration for BulkSink Component in the OpenMetadata Ingestion Framework.",
@@ -161,20 +149,13 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "type"
-      ]
+      "required": ["type"]
     },
     "logLevels": {
       "description": "Supported logging levels",
       "javaType": "org.openmetadata.schema.metadataIngestion.LogLevels",
       "type": "string",
-      "enum": [
-        "DEBUG",
-        "INFO",
-        "WARN",
-        "ERROR"
-      ],
+      "enum": ["DEBUG", "INFO", "WARN", "ERROR"],
       "default": "INFO"
     },
     "workflowConfig": {
@@ -201,9 +182,7 @@
         }
       },
       "additionalProperties": false,
-      "required": [
-        "openMetadataServerConfig"
-      ]
+      "required": ["openMetadataServerConfig"]
     }
   },
   "properties": {
@@ -246,29 +225,16 @@
           "$ref": "../type/basic.json#/definitions/uuid"
         }
       },
-      "required": [
-        "source",
-        "workflowConfig"
-      ],
+      "required": ["source", "workflowConfig"],
       "additionalProperties": false
     }
   },
   "oneOf": [
     {
-      "required": [
-        "name",
-        "openMetadataWorkflowConfig",
-        "source",
-        "sink"
-      ]
+      "required": ["name", "openMetadataWorkflowConfig", "source", "sink"]
     },
     {
-      "required": [
-        "name",
-        "openMetadataWorkflowConfig",
-        "source",
-        "bulkSink"
-      ]
+      "required": ["name", "openMetadataWorkflowConfig", "source", "bulkSink"]
     }
   ],
   "additionalProperties": false