Process async requests: receives a command-line payload to execute and when it finishes, sends a HTTP callback.

Process-AsyncRequest.ps1

function Process-AsyncRequest {
    # Receives a command-line payload to execute and when it finishes, sends a HTTP callback. 
    # Use-case: use with Azure Functions with connection to a Hybrid Connection. By installing this function on a server using HCM, 
    # we can pass code to the machines and get back JSON data at bulk.
    # Callback-URL can be a URL generated by Azure Logic Apps 'HTTP + Webhook' action.
    
    # Author: Daniel Ferreira (@daniel0x00)  
    # License: BSD 3-Clause

    <#
    .SYNOPSIS
        Receives a JSON schema with a command-line to be executed and a webhook to push results when command-line is completed.
   
    .EXAMPLE
        '{"commandline":"<command_line>", "sourcetype":"<sourcetype>", "callback":"<webhook_url>"}' | Process-AsyncRequest 

    .PARAMETER Input
        String. Mandatory. Pipeline enabled.

        The input JSON string. Valid schema:

            {
                "commandline":"<command_line>",
                "sourcetype":"<sourcetype>",
                "callback":"<webhook_url>"
            }
    #>   
    [CmdletBinding()]
    [OutputType([PSCustomObject])]
    param(    
        [Parameter(Position=0, Mandatory=$true, ValueFromPipeline=$true)]
        [string] $InputObject,

        [Parameter(Position=1, Mandatory=$false, ValueFromPipeline=$false)]
        [string] $StageStorageAccountName='AzStorageAccountName', # AzStorage Account name. Must be created upfront or passed-by.

        [Parameter(Position=2, Mandatory=$false, ValueFromPipeline=$false)]
        [string] $StageStorageContainerName='container', # AzStorage container name. Must be created upfront.

        [Parameter(Position=3, Mandatory=$false, ValueFromPipeline=$false)]
        [string] $StageStorageSAS='<sas_token>', #e.g: ?sv=2019-12-12&ss=bfqt&srt=o&sp=wac&se=2030-08-05T17:52:25Z&st=2020-08-05T09:52:25Z&spr=https&sig=<token>

        [Parameter(Position=4, Mandatory=$false, ValueFromPipeline=$false)]
        [string] $OutputLogPath='C:\temp\' #TODO: check existance of this path.
    )

    # Received timestamp:
    $ReceivedAt = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
    $LogOutputFile = $OutputLogPath + $ReceivedAt + '_AsyncRequest.txt'

    # Set vars:
    $OutputStageStorage = @()
    $ParsedInput = $InputObject | ConvertFrom-Json 
    $CommandLine = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($ParsedInput.commandline))
    $SourceType = $ParsedInput.sourcetype
    $CallbackURL = $ParsedInput.callback

    # Main try.
    # This try/catch prevents the process to fail without notifying the upstream orchestration
    try {

        # Log:
        $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
        "[$LogTimestamp] Execution started. " + [Environment]::NewLine | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append
        'Input object: ' + [Environment]::NewLine + $InputObject + [Environment]::NewLine + [Environment]::NewLine + 'CommandLine: ' + [Environment]::NewLine + $CommandLine + [Environment]::NewLine + [Environment]::NewLine + 'CallbackURL: ' + [Environment]::NewLine + $CallbackURL + [Environment]::NewLine | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

        # Stopwatch:
        $StopWatch = [System.Diagnostics.Stopwatch]::StartNew()

        # Execute the given command: 
        # $OutputExpression = Invoke-Expression -Command $CommandLine -ErrorAction Stop

        # Log:
        # $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
        #"[$LogTimestamp] Expression invoked." | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

        # Execute the command and break the output into files:
        $ContentOutputFile = [string]::Concat($OutputLogPath,$ReceivedAt,'_AsyncRequestContent')
        $ContentOutputFileCounter = 0;
        Invoke-Expression -Command $CommandLine -ErrorAction Stop | ConvertFrom-Json | ForEach-Object {
            $ContentOutputFileCounter++; 
            $Filename = [string]::concat($ContentOutputFile,'_',$ContentOutputFileCounter,'.json')

            # Log:
            # $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
            #"[$LogTimestamp] Writting file #$ContentOutputFileCounter to disk: $Filename" | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

            $_ | ConvertTo-Json -AsArray -Depth 99 | Out-File -FilePath $Filename -Encoding utf8 -NoNewline
        }

        # Log:
        $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
        "[$LogTimestamp] Expression invoked." | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

        # TODO:
        # Grouping of file contents to reduce AzStorage Throttling risk.
        # In the meantime, must use 'chunk' custom function to group output objects before calling Process-AsyncRequest function.

        # Create context to AzStorage:
        $AzStorageContext = New-AzStorageContext -StorageAccountName $StageStorageAccountName -SasToken $StageStorageSAS

        # Upload files into Azure Storage:
        $ContentOutputFileUploadCounter = 0;
        $ContentOutputFileCount = (Get-Item "$ContentOutputFile*" | Select-Object FullName, Name | Measure-Object).Count
        # Log how many files will be processed:
        "[$LogTimestamp] AsyncRequestContent to upload to AzStorage: $ContentOutputFileCount" | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

        $AzStorageThrottlingSeconds = 3;
        $FileSizes = @()
        Get-Item "$ContentOutputFile*" | Select-Object FullName, Name | ForEach-Object {
            $FullName = $_.FullName
            $Name = $SourceType + '/' + $_.Name -replace '-','/' -replace ' ','/'
            $FileSize = (Get-Item -Path $FullName).Length

            # Log:
            $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
            if ($FileSize -le 200) { "[$LogTimestamp] Skipping file - too small: $Name" | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append }

            # Upload files to AzStorage:
            if ($FileSize -gt 200) { 
                $FileSizes += $FileSize
                # Throttling delayer:
                $ContentOutputFileUploadCounter++;
                if (($ContentOutputFileUploadCounter % 8) -eq 0) { 
                    "[$LogTimestamp] Throttling file #$ContentOutputFileUploadCounter for $AzStorageThrottlingSeconds seconds to upload next file --> $Name" | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append
                    Start-Sleep -Seconds $AzStorageThrottlingSeconds
                }

                # Log:
                $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
                "[$LogTimestamp] Uploading file #$ContentOutputFileUploadCounter --> $Name" | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append

                # Upload:
                try { $OutputStageStorage += Set-AzStorageBlobContent -Context $AzStorageContext -File $FullName -Blob $Name -Container $StageStorageContainerName -Properties @{"ContentType" = "application/json"} -Force | Select-Object @{n='Uri';e={$_.ICloudBlob.Uri}}, Length }
                catch { "[$LogTimestamp] EXCEPTION when uploading file #$ContentOutputFileUploadCounter --> $Name. Exception message: "+$_.Exception.Message | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append }
            }
        }

        # Set the output payload:
        $OutputPayload = [PSCustomObject]@{
            Payload = $OutputStageStorage
            #PayloadBytes = ([System.Text.Encoding]::UTF8.GetByteCount(($OutputExpression | ConvertTo-Json -Depth 50 -Compress)))
            PayloadBytes = ($FileSizes | Measure-Object -Sum).Sum
        }

        # Send output to callback URL:
        $HTTPResponse = Invoke-WebRequest -Uri $CallbackURL -Method Post -Body ($OutputPayload | ConvertTo-Json -Depth 50 -Compress) -ContentType 'application/json'

        # Remove temp files:
        Get-Item "$ContentOutputFile*" | Remove-Item -Force

        # Stopwatch:
        $StopWatch.Stop()

        # Log:
        $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
        "[$LogTimestamp] StopWatch: " + ([int]$StopWatch.Elapsed.Hours) + 'h ' + ([int]$StopWatch.Elapsed.Minutes) + 'm ' + ([int]$StopWatch.Elapsed.Seconds) + 's.' + [Environment]::NewLine + [Environment]::NewLine + 'OutputPayload: ' + [Environment]::NewLine + ($OutputPayload | ConvertTo-Json -Depth 50 -Compress) + [Environment]::NewLine + 'HTTPResponse: ' + ($HTTPResponse.StatusCode | ConvertTo-Json -Depth 50 -Compress) + [Environment]::NewLine + [Environment]::NewLine | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append
        
    }
    catch {
        $Exception = $_.Exception.Message
        $ExceptionPayload = '{"exception":true,"function":"Process-AsyncRequest","message":"'+$Exception+'"}'

        $HTTPResponse = Invoke-WebRequest -Uri $CallbackURL -Method Post -Body ($ExceptionPayload | ConvertTo-Json -Compress) -ContentType 'application/json'
        
        # Log:
        $LogTimestamp = (get-date).ToLocalTime().ToString("yyyy-MM-dd HHmmss")
        "[$LogTimestamp] Exception triggered:" + [Environment]::NewLine + $ExceptionPayload | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append
        'HTTPResponse: ' + ($HTTPResponse.StatusCode | ConvertTo-Json -Compress) | Out-File -FilePath $LogOutputFile -Encoding UTF8 -Append
    }
}