mrpaulandrew / procfwk Goto Github PK

Where common expressions are used from lookup activities etc, refactor into variables so expression code isn't duplicated throughout the pipeline.

Having the last parameter value used captured along side existing metadata means the precursor stored procedure can inspect this value then update the next value for a given run.

Eg. Last value passed = 1.

Precursor update. New value = Last + 1.

This helps dynamically passing values to worker pipelines that use incremental date/time values.

For example store just the Principal Id in the database, but the Secret in Key Vault.

For parts of the Infant pipeline where SPN details are being passed around to interact with Workers use Secure Input/Output options for in scope activities so SPN values are not exposed in plain text as part of standard logs.

Update the [dbo].[ServicePrincipals] table to support Key Vault URLs as an option to be used instead of sorting SPN details in the metadata directly.

Drive different behaviour via a database property.

Extend affected Azure Functions to parse Key Vault URLs and return values from Key Vault using MSI authentication.

Hi, currently the framework can only handle one collection of pipelines (procfwk.Pipelines) and one execution (CurrentExecution) to be managed at any single time. It would be ideal if a higher level grouping to be allowed so that, we can define multiple sets of pipelines and dependencies, and then we have a capability to submit them as separate jobs, and have the framework to manage multiple jobs to be tracked by procfwk.CurrentExecution? Thank you.

Separate Metadata Integrity and Clean Up results into Two Stored Procedures & Pipeline Activities.

• One Lookup Activity
• One Stored Proc Activity

Can run in parallel before Clean Up ForEach activity within parent pipeline.

Then you can deploy the whole ADF code nicely with one line of code.

Pipeline passing false, should be true.

Checks for successful workers in current execution table.

Thanks @pfiadeiro for reporting this via Twitter.

When cancelling the parent or child pipelines (recursive cancel), the framework pipelines complete with a cancelled status, but the worker pipelines continue running (because cancelling does not allow cleanup). In the case where 50 workers in different data factories need cancelling too, this can be time consuming.

It would be nice if the framework had the option to define a cancellation policy, much like the continuation on error options, to allow cancellation of worker pipelines in the case an infant is cancelled.

I have been looking into pushing data factory events into event grid, to trigger a function for cancelled infants, that calls a new a cleanup pipeline to do just this. It cancels the worker, and waits for it to finish before updating the CurrentExecution table (only if required). I have PoC code that I can share.

I'm trying to setup the procfwk. Untill this moment everything works fine, but now I receive this message while executing DeployProcFwkComponents.ps1.

`Update-PropertiesFromFile : Could not find object: linkedService.SupportDatabase
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\public\Publish-AdfV2FromJson.ps1:160 char:9
Update-PropertiesFromFile -adf $adf -stage $Stage -option $op ...
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
CategoryInfo : NotSpecified: (:) [Write-Error], WriteErrorException
FullyQualifiedErrorId : Microsoft.PowerShell.Commands.WriteErrorException,Update-PropertiesFromFile

The property 'Body' cannot be found on this object. Verify that the property exists.
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\private\Update-PropertiesFromFile.ps1:68 char:9
$json = $o.Body
~~~~~~~~~~~~~~~
CategoryInfo : NotSpecified: (:) [], PropertyNotFoundException
FullyQualifiedErrorId : PropertyNotFoundStrict

The variable '$json' cannot be retrieved because it has not been set.
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\private\Update-PropertiesFromFile.ps1:69 char:23
if ($null -eq $json) {
~~~~~
CategoryInfo : InvalidOperation: (json:String) [], RuntimeException
FullyQualifiedErrorId : VariableIsUndefined

Update-PropertiesFromFile : Wrong path defined in config for object(path): linkedService.SupportDatabase(properties.typeProperties.connectionString.secretName)
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\public\Publish-AdfV2FromJson.ps1:160 char:9
Update-PropertiesFromFile -adf $adf -stage $Stage -option $op ...
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
CategoryInfo : NotSpecified: (:) [Write-Error], DataException
FullyQualifiedErrorId : System.Data.DataException,Update-PropertiesFromFile
The variable '$json' cannot be retrieved because it has not been set.
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\private\Update-PropertiesFromFile.ps1:88 char:44
Update-ObjectProperty -obj $json -path "properties.$p ...
~~~~~
CategoryInfo : InvalidOperation: (json:String) [], RuntimeException
FullyQualifiedErrorId : VariableIsUndefined

Save-AdfObjectAsFile : Cannot bind argument to parameter 'obj' because it is null.
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\private\Update-PropertiesFromFile.ps1:104 char:41
$f = (Save-AdfObjectAsFile -obj $o)
~~
CategoryInfo : InvalidData: (:) [Save-AdfObjectAsFile], ParameterBindingValidationException
FullyQualifiedErrorId : ParameterArgumentValidationErrorNullNotAllowed,Save-AdfObjectAsFile

The variable '$f' cannot be retrieved because it has not been set.
At C:\Program Files\WindowsPowerShell\Modules\azure.datafactory.tools\0.17.0\private\Update-PropertiesFromFile.ps1:105 char:23
$o.FileName = $f
~~
CategoryInfo : InvalidOperation: (f:String) [], RuntimeException
FullyQualifiedErrorId : VariableIsUndefined
`

I tried many things. What I've changed now in the PS1 script is remove the first 14 lines and added this:

$tenantId = "mytenantid"
$subscriptionId = "myscubscriptionid"
$spId = "azureclientid"
$spKey = "azureclientsecret"
$resourceGroupName = "myresourcename"
$dataFactoryName = "myadfname"
$region = "West Europe" (but also tried "WestEurope" or "West-Europe")

And later I found:
$Env:SQLDatabase = ""
There I added this;
Server=tcp:adfframework.database.windows.net,1433;Initial Catalog=frameworkmetadatadb;Persist Security Info=False;User ID=MYUSERNAME;Password=MYPASSWORD;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=False;Connection Timeout=30;

I'm pretty sure that I am making a mistake. But I don't know which one and how to fix my problem. Can I get support here?

Add Tenant and Subscription value to a metadata table allowing the framework to reach across both tenants and subscriptions when calling worker pipelines.

Remove tenant as a property.
Remove subscription as a property.

Using the PowerShell script blogged about here: https://mrpaulandrew.com/2020/11/09/best-practices-for-implementing-azure-data-factory-auto-checker-script-v0-1/

Export an ARM template for the framework factory and apply fixes to any exposed issues.

Add better metadata integrity checks for encrypted SPN details in the event property values change.

Use procfwkTesting schema within separate solution project.

Create fake pipeline status values during a restart that force the parent pipeline clean activities to run with correct outcomes.

In the event of a worker pipeline failure develop the ability to allow the worker to automatically retry.

Configure via the properties table:

• Enabled.
• Retry attempts.

Describe the bug
When alerting is setup using the bit mask to specify specific alerts only, eg Errors, the proc to determine if alerts should be processed returned before the outcome was known. This meant a successful worker configured to only alert on failure would make the framework child pipeline fail when processing the alerts as it would attempt to send the alert without having a valid address to send it to.

Affected services
Which resource within the processing framework does this affect?

• All of them

To Reproduce
Steps to reproduce the behavior:

1. Modify procfwk.PipelineAlertLink to setup a worker pipeline to alert an email address only when an error occurs (OutcomesBitValue = 4). Ensure no other alerts are configured for the pipeline
2. Run the framework to process the pipeline successfully
3. In the ADF monitor view, the framework child pipeline will have a status of error from trying to send an email to a blank address

Expected behaviour
The framework should be checking the outcome of the worker pipeline to determine if there are any alerts to be sent based a bit-wise AND with procfwk.PipelineAlertLink.OutcomesBitValue.

Additional context
I have a branch on my fork with a fix, will submit a PR with the details

Generic worker pipelines, datasets and linked services can all be configured to use parameters for dynamic behavior. If some completely data-driven solutions, generic pipelines can have large string parameters that are not currently supported by the framework.

ADF.procwfk should support large pipeline parameters so these generic pipelines can be orchestrated via the framework.

If a parameter exists in the metadata, but the procfwk.PipelineParameters.ParameterValue column is NULL, then no parameters will be sent to that pipeline.

This affects the execution of worker pipelines in data factory.

Steps to reproduce the behavior:

1. Create a pipeline with more than one parameter.
2. Change the value of one parameter in the procfwk.PipelineParameters table to NULL, but leave the pipeline in place.
3. Execute the pipeline.

I would expect one of two outcomes:
Null parameters are simply not passed to the pipeline.
A constraint on the procfwk.PipelineParameters.ParameterValue column would prevent NULLs.

Use Case:

If the metadata of some pipelines, for some reason, isn't updated, when a new execution is triggered and the Clean Up Previous Run activity runs, the status of all pipelines not updated will be changed to Success if they completed successfully.

Once that activity completes, the Execution Wrapper activity will start a new execution that will be pointless since all pipelines have finished successfully and nothing will happen during this run, it will only serve to clean the CurrentExecution table at the end.

Suggestion:

Tweak the ExecutionWrapper stored procedure to consider this particular case.

One possible solution is after the check for a running execution is done, to add the following piece of code

--- check for all pipelines' status being success after a clean run
IF EXISTS
(
    select 1 from [procfwk].[CurrentExecution]
)
AND NOT EXISTS
(
    select 1 from [procfwk].[CurrentExecution] where [PipelineStatus] <> 'Success'
)
BEGIN
    EXEC [procfwk].[UpdateExecutionLog]
		@PerformErrorCheck = 0;
END;

Currently, if a worker pipeline fails it blocks all other pipeline executions, failing the entire execution. A feature has been put on the backlog to allow this failure to be ignored and allow all other pipelines to execute. In my experience, what is needed is a bit more complex.

For a single data platform you will have several to many data products that are generated. These products each require certain pipelines to run, in order. Often, data products can depend on common pipelines, but this is not always the case.

For example, consider the following scenario:

id | pipeline name        | stage
----------------------------------
1  | data_source1_http    | 1
2  | trans_source1_stage  | 2
3  | trans_product1_load  | 3
4  | data_source2_odata   | 1
5  | trans_source2_stage  | 2
6  | trans_product2_load  | 3
7  | trans_product3_dwh   | 4

trans_product3_dwh pulls data from the staging area for source1 and source 2. trans_product1_load only uses data from source1. trans_product2_load only uses data from source 2.

If pipeline data_source1_http fails for some reason, then the execution for pipelines 2, 3 and 7 should be held up, but not 4, 5 and 6. Those pipelines are independent of the failed pipeline and the subsequent products can be created without a problem.

This type of pipeline dependency/heirarchy would be very helpful as part of the orchestration framework.

Following initial investigations:
https://mrpaulandrew.com/2020/06/03/adf-procfwk-and-azure-synapse-orchestrate-preview-and-limitations/

Add support for Azure Synapse.

Include support for Worker and Framework pipelines in either Data Factory or Synapse.

Scrap a list of pipelines from an ADF instance via Powershell and use it to populate the metadata database with a basic set of information using the stored procedure [procfwkHelpers].[AddPipelineViaPowerShell].

Sudo code...
Get-AzDataFactoryV2Pipeline
EXEC [procfwkHelpers].[AddPipelineViaPowerShell].

Describe the bug
If you cancel a running worker pipeline, the framework stops in the same way as if an error happened (in simple failure handling mode) . If you then restart the framework processing, instead of re-running the worker pipeline, it fails to reset the state and the parent framework pipeline fails without triggering any child pipelines.

Affected services
Which resource within the processing framework does this affect?

• All of them

To Reproduce
Steps to reproduce the behavior:

1. Trigger a new run in Data Factory
2. Cancel one of the worker pipelines
3. Wait for the current framework execution to complete
4. Check the current executions table - the cancelled pipeline should have a status of Cancelled/Unknown
5. Re-run the framework parent pipeline. The framework fails to restart the processing cleanly like it does for a failed execution.

Expected behaviour
A cancelled pipeline should be restart-able like a failed pipeline.

Additional context
I have a branch on my fork with a fix, will submit a PR with the details

Move worker pipeline execute activities into infant pipeline along side run handling, making the call to run a worker pipeline an isolated operation.

The child level pipeline can then be made simpler and used just for the purpose of scaling out executions.

Once done. Refactor infant to use local variables replacing hard coded activity output expressions where possible.

Include TRY/CATCH blocks in Azure Functions that feedback into wider framework failures.

1. Add new table [PipelineDependencies] allowing 1:many relationships for Pipelines.

| PipelineId | DependantPipelineId |

2. Add a new Property called 'FailureHandling'. This property could have 3 possible values....

None = The failure status is updated in the Current Execution table and the framework carries on regardless.

Simple = The failure status is updated in the Current Execution table. Downstream execution stages are marked as blocked and processing is stopped after the stage where the failure occurred. This is basically what currently happens.

Informed = The failure status is updated in the Current Execution table. Downstream pipelines are marked as blocked (using the DependantPipelineId attribute in the PipelineDependencies table). However, processing is allowed to continue. When the next execution stage starts only pipelines NOT blocked will be prepared and ran.

Include:

• Scoping template
• New KPI's
• Better visuals

Control pipeline run frequency via the framework metadata. Eg. Dynamic enable/disable of pipelines for a given pattern. Maybe use a precursor procedure with custom code to drive if a pipeline/stage should be executed daily/weekly/monthly etc?

Add tests that deal with the framework restart logic given each failure handling type, included cancelled pipelines.

Describe the bug
When the OverideRestart property is set to 1 and there's a current execution where some pipelines have a status <> 'Success', the creation of a new execution will fail. This happens since the ExecutionWrapper stored procedure calls the UpdateExecutionLog stored procedure and when the OverideRestart property is set to 1, the process will only work if all pipelines in the Current Execution table have a status of 'Success' (in theory, they wouldn't even be there and moved to the final log table once framework completes), otherwise it will always throw an error.

Affected services
Which resource within the processing framework does this affect?

• Data Factory
• SQL Database

To Reproduce
Steps to reproduce the behavior:

1. Change a pipeline parameter to force a specific pipeline to fail. For example, in a wait activity set the value 'abc'
2. Trigger a new run in Data Factory and wait for it to finish with a failed pipeline
3. Set the property OverideRestart to 1
4. Trigger another new run in Data Factory
5. Framework will fail when running the Execution Wrapper activity.

Expected behaviour
The creation of a new run should be possible (by setting OverideProperty = 1) even when the previous run didn't finish successfully

Complete first pass documentation via GitHub pages.

Include all current framework features and functionality as a base line and complete set of details.

When reusing worker pipeline for different reasons provide an extra field to include details/values that support understanding and debugging.

In the event of a platform failure validate the status of pipelines that have become disconnected from the execution run metadata.

Get the actual status of the pipeline and decide on the best action to take before the execution is restarted.

When a Worker pipeline is cancelled during an execution run ensure the metadata and logic handles this correctly.

For a given execution run use global variables to set parameters that don't change and reduce the number of lookup activity calls without the processing.

Add a scalar function to return the latest property value. Use this internally replacing complete select queries against the properties view.

Reduce all framework pipeline activity timeout values from the service default of 7 days.

In the event of a wider platform failure, the framework processes should not "get stuck" and be in a situation of waiting 7 days before a failure is realised.

Building on community feedback in the following issues:

#61
#62

Allow the framework to support multiple executions of the parent pipeline using a new higher level concept called Batches.

• A batch sits above execution stages.
• Batches can be linked to 1/many execution stages.
• Be enabled/disabled via properties.
• Batches can run concurrently. Batch 1 = Running, Batch 2 = Running.
• A single batch ID cannot run concurrently.

Examples of Batch names:

1. Hourly
2. Daily
3. Weekly
4. Monthly

Finally, it is expected that a Trigger hitting the parent pipeline will have the batch name included.

Describe the bug
When the parent pipeline is triggered and the framework has not reached the stage of registering into the framework db, if the user triggers again, there is a potential chance that subsequent trigger(s) will overwrite the preceding runs

Affected services
Which resource within the processing framework does this affect?

• Data Factory
• SQL Database (I believe the subsequent triggered pipeline will overwrite records of the preceding one)
• Functions
• All of them
• Other

To Reproduce
Steps to reproduce the behavior:

1. Trigger 02-Parent. Then immediately, trigger 02-Parent again. Technically, this would create two trigger runs. Recognizing that the current framework is not designed to allow multiple triggering of the same "job", so this is technically not an allowed function. But the current integrity check relies on the existence of the framework database records, and in this scenario, these records are not written/committed to the db yet hence the framework will not know that there is a job already triggered

PS: this is one of the reasons for Enhancement Request #61

Expected behaviour
The 2nd triggered run should fail (as if the integrity check has detected via db records)

Screenshots
Sorry I no longer have screen shots as I have created my own work around detection for this

Additional context
My current work around is that I created an additional function to check the ADF run log if a triggered job has already occurred (in my version, i have implemented a version of the enhancement request #61)