Add Get-TaskResult CmdLet for DF (#786)

Co-authored-by: Greg Roll <[email protected]>
Azure · Apr 18, 2022 · d184e72 · d184e72
1 parent 6dd32c3
commit d184e72
Show file tree

Hide file tree

Showing 13 changed files with 263 additions and 16 deletions.
diff --git a/release_notes.md b/release_notes.md
@@ -1 +1,3 @@
-* Bug fix: Activity Functions can now use output bindings (https://github.com/Azure/azure-functions-powershell-worker/issues/646)
+* Bug fix: [Context.InstanceId can now be accessed](https://github.com/Azure/azure-functions-powershell-worker/issues/727)
+* Bug fix: [Data in External Events is now read and returned to orchestrator](https://github.com/Azure/azure-functions-powershell-worker/issues/68)
+* New feature (external contribution): [Get-TaskResult can now be used to obtain the result of an already-completed Durable Functions Task](https://github.com/Azure/azure-functions-powershell-worker/pull/786)
diff --git a/src/Durable/Commands/GetDurableTaskResult.cs b/src/Durable/Commands/GetDurableTaskResult.cs
@@ -0,0 +1,36 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+#pragma warning disable 1591 // Missing XML comment for publicly visible type or member 'member'
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Durable.Commands
+{
+    using System.Collections;
+    using System.Management.Automation;
+    using Microsoft.Azure.Functions.PowerShellWorker.Durable.Tasks;
+
+    [Cmdlet("Get", "DurableTaskResult")]
+    public class GetDurableTaskResultCommand : PSCmdlet
+    {
+        [Parameter(Mandatory = true)]
+        [ValidateNotNull]
+        public DurableTask[] Task { get; set; }
+
+        private readonly DurableTaskHandler _durableTaskHandler = new DurableTaskHandler();
+
+        protected override void EndProcessing()
+        {
+            var privateData = (Hashtable)MyInvocation.MyCommand.Module.PrivateData;
+            var context = (OrchestrationContext)privateData[SetFunctionInvocationContextCommand.ContextKey];
+
+            _durableTaskHandler.GetTaskResult(Task, context, WriteObject);
+        }
+
+        protected override void StopProcessing()
+        {
+            _durableTaskHandler.Stop();
+        }
+    }
+}
diff --git a/src/Durable/DurableTaskHandler.cs b/src/Durable/DurableTaskHandler.cs
@@ -195,6 +195,21 @@ public void WaitAny(
             }
         }
 
+        public void GetTaskResult(
+            IReadOnlyCollection<DurableTask> tasksToQueryResultFor,
+            OrchestrationContext context,
+            Action<object> output)
+        {
+            foreach (var task in tasksToQueryResultFor) {
+                var scheduledHistoryEvent = task.GetScheduledHistoryEvent(context, true);
+                var processedHistoryEvent = task.GetCompletedHistoryEvent(context, scheduledHistoryEvent, true);
+                if (processedHistoryEvent != null)
+                {
+                    output(GetEventResult(processedHistoryEvent));
+                }
+            }
+        }
+
         public void Stop()
         {
             _waitForStop.Set();

diff --git a/src/Durable/Tasks/ActivityInvocationTask.cs b/src/Durable/Tasks/ActivityInvocationTask.cs
@@ -37,15 +37,15 @@ internal ActivityInvocationTask(string functionName, object functionInput)
         {
         }
 
-        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context)
+        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context, bool processed)
         {
             return context.History.FirstOrDefault(
                 e => e.EventType == HistoryEventType.TaskScheduled &&
                      e.Name == FunctionName &&
-                     !e.IsProcessed);
+                     e.IsProcessed == processed);
         }
 
-        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent)
+        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent, bool processed)
         {
             return scheduledHistoryEvent == null
                 ? null

diff --git a/src/Durable/Tasks/DurableTask.cs b/src/Durable/Tasks/DurableTask.cs
@@ -11,9 +11,9 @@ namespace Microsoft.Azure.Functions.PowerShellWorker.Durable.Tasks
 
     public abstract class DurableTask
     {
-        internal abstract HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context);
+        internal abstract HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context, bool processed = false);
 
-        internal abstract HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent);
+        internal abstract HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent, bool processed = false);
 
         internal abstract OrchestrationAction CreateOrchestrationAction();
     }

diff --git a/src/Durable/Tasks/DurableTimerTask.cs b/src/Durable/Tasks/DurableTimerTask.cs
@@ -28,15 +28,15 @@ internal DurableTimerTask(
             Action = new CreateDurableTimerAction(FireAt);
         }
 
-        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context)
+        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context, bool processed)
         {
             return context.History.FirstOrDefault(
                 e => e.EventType == HistoryEventType.TimerCreated &&
                      e.FireAt.Equals(FireAt) &&
                      !e.IsProcessed);
         }
 
-        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent)
+        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent scheduledHistoryEvent, bool processed)
         {
             return scheduledHistoryEvent == null
                 ? null

diff --git a/src/Durable/Tasks/ExternalEventTask.cs b/src/Durable/Tasks/ExternalEventTask.cs
@@ -21,17 +21,17 @@ public ExternalEventTask(string externalEventName)
         }
 
         // There is no corresponding history event for an expected external event
-        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context)
+        internal override HistoryEvent GetScheduledHistoryEvent(OrchestrationContext context, bool processed)
         {
             return null;
         }
 
-        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent taskScheduled)
+        internal override HistoryEvent GetCompletedHistoryEvent(OrchestrationContext context, HistoryEvent taskScheduled, bool processed)
         {
             return context.History.FirstOrDefault(
                     e => e.EventType == HistoryEventType.EventRaised &&
                          e.Name == ExternalEventName &&
-                         !e.IsProcessed);
+                         e.IsPlayed == processed);
         }
 
         internal override OrchestrationAction CreateOrchestrationAction()

diff --git a/...icrosoft.Azure.Functions.PowerShellWorker/Microsoft.Azure.Functions.PowerShellWorker.psd1 b/...icrosoft.Azure.Functions.PowerShellWorker/Microsoft.Azure.Functions.PowerShellWorker.psd1
@@ -59,6 +59,7 @@ FunctionsToExport = @(
 # Cmdlets to export from this module, for best performance, do not use wildcards and do not delete the entry, use an empty array if there are no cmdlets to export.
 CmdletsToExport = @(
     'Get-OutputBinding',
+    'Get-DurableTaskResult'
     'Invoke-DurableActivity',
     'Push-OutputBinding',
     'Set-DurableCustomStatus',

diff --git a/...icrosoft.Azure.Functions.PowerShellWorker/Microsoft.Azure.Functions.PowerShellWorker.psm1 b/...icrosoft.Azure.Functions.PowerShellWorker/Microsoft.Azure.Functions.PowerShellWorker.psm1
@@ -235,6 +235,8 @@ function New-DurableOrchestrationCheckStatusResponse {
     The TaskHubName of the orchestration instance that will handle the external event.
 .PARAMETER ConnectionName
     The name of the connection string associated with TaskHubName
+.PARAMETER AppCode
+    The Azure Functions system key
 #>
 function Send-DurableExternalEvent {
     [CmdletBinding()]
@@ -263,12 +265,16 @@ function Send-DurableExternalEvent {
 
         [Parameter(
             ValueFromPipelineByPropertyName=$true)]
-        [string] $ConnectionName
+        [string] $ConnectionName,
+
+        [Parameter(
+            ValueFromPipelineByPropertyName=$true)]
+        [string] $AppCode
     )
 
     $DurableClient = GetDurableClientFromModulePrivateData
 
-    $RequestUrl = GetRaiseEventUrl -DurableClient $DurableClient -InstanceId $InstanceId -EventName $EventName -TaskHubName $TaskHubName -ConnectionName $ConnectionName
+    $RequestUrl = GetRaiseEventUrl -DurableClient $DurableClient -InstanceId $InstanceId -EventName $EventName -TaskHubName $TaskHubName -ConnectionName $ConnectionName -AppCode $AppCode
 
     $Body = $EventData | ConvertTo-Json -Compress
 
@@ -280,7 +286,8 @@ function GetRaiseEventUrl(
     [string] $InstanceId,
     [string] $EventName,
     [string] $TaskHubName,
-    [string] $ConnectionName) {
+    [string] $ConnectionName,
+    [string] $AppCode) {
 
     $RequestUrl = $DurableClient.BaseUrl + "/instances/$InstanceId/raiseEvent/$EventName"
 
@@ -291,6 +298,9 @@ function GetRaiseEventUrl(
     if ($null -eq $ConnectionName) {
         $query += "connection=$ConnectionName"
     }
+    if ($null -eq $AppCode) {
+        $query += "code=$AppCode"
+    }
     if ($query.Count -gt 0) {
         $RequestUrl += "?" + [string]::Join("&", $query)
     }

diff --git a/...nctions.PowerShellWorker.E2E/Azure.Functions.PowerShellWorker.E2E/DurableEndToEndTests.cs b/...nctions.PowerShellWorker.E2E/Azure.Functions.PowerShellWorker.E2E/DurableEndToEndTests.cs
@@ -147,6 +147,172 @@ public async Task LegacyDurableCommandNamesStillWork()
             }
         }
 
+        [Fact]
+        public async Task OrchestratationContextHasAllExpectedProperties()
+        {
+            var initialResponse = await Utilities.GetHttpTriggerResponse("DurableClientOrchContextProperties", queryString: string.Empty);
+            Assert.Equal(HttpStatusCode.Accepted, initialResponse.StatusCode);
+
+            var initialResponseBody = await initialResponse.Content.ReadAsStringAsync();
+            dynamic initialResponseBodyObject = JsonConvert.DeserializeObject(initialResponseBody);
+            var statusQueryGetUri = (string)initialResponseBodyObject.statusQueryGetUri;
+
+            var startTime = DateTime.UtcNow;
+
+            using (var httpClient = new HttpClient())
+            {
+                while (true)
+                {
+                    var statusResponse = await httpClient.GetAsync(statusQueryGetUri);
+                    switch (statusResponse.StatusCode)
+                    {
+                        case HttpStatusCode.Accepted:
+                            {
+                                var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                                var runtimeStatus = (string)statusResponseBody.runtimeStatus;
+                                Assert.True(
+                                    runtimeStatus == "Running" || runtimeStatus == "Pending",
+                                    $"Unexpected runtime status: {runtimeStatus}");
+
+                                if (DateTime.UtcNow > startTime + _orchestrationCompletionTimeout)
+                                {
+                                    Assert.True(false, $"The orchestration has not completed after {_orchestrationCompletionTimeout}");
+                                }
+
+                                await Task.Delay(TimeSpan.FromSeconds(2));
+                                break;
+                            }
+
+                        case HttpStatusCode.OK:
+                            {
+                                var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                                Assert.Equal("Completed", (string)statusResponseBody.runtimeStatus);
+                                Assert.Equal("True", statusResponseBody.output[0].ToString());
+                                Assert.Equal("Hello myInstanceId", statusResponseBody.output[1].ToString());
+                                Assert.Equal("False", statusResponseBody.output[2].ToString());
+                                return;
+                            }
+
+                        default:
+                            Assert.True(false, $"Unexpected orchestration status code: {statusResponse.StatusCode}");
+                            break;
+                    }
+                }
+            }
+        }
+
+        [Fact]
+        public async Task ExternalEventReturnsData()
+        {
+            var initialResponse = await Utilities.GetHttpTriggerResponse("DurableClient", queryString: "?FunctionName=DurableOrchestratorRaiseEvent");
+            Assert.Equal(HttpStatusCode.Accepted, initialResponse.StatusCode);
+
+            var initialResponseBody = await initialResponse.Content.ReadAsStringAsync();
+            dynamic initialResponseBodyObject = JsonConvert.DeserializeObject(initialResponseBody);
+            var statusQueryGetUri = (string)initialResponseBodyObject.statusQueryGetUri;
+            var raiseEventUri = (string)initialResponseBodyObject.sendEventPostUri;
+
+            raiseEventUri = raiseEventUri.Replace("{eventName}", "TESTEVENTNAME");
+
+            var startTime = DateTime.UtcNow;
+
+            using (var httpClient = new HttpClient())
+            {
+                while (true)
+                {
+                    // Send external event payload
+                    var json = JsonConvert.SerializeObject("helloWorld!");
+                    var httpContent = new StringContent(json, Encoding.UTF8, "application/json");
+                    await httpClient.PostAsync(raiseEventUri, httpContent);
+
+                    var statusResponse = await httpClient.GetAsync(statusQueryGetUri);
+                    switch (statusResponse.StatusCode)
+                    {
+                        case HttpStatusCode.Accepted:
+                            {
+                                var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                                var runtimeStatus = (string)statusResponseBody.runtimeStatus;
+                                Assert.True(
+                                    runtimeStatus == "Running" || runtimeStatus == "Pending",
+                                    $"Unexpected runtime status: {runtimeStatus}");
+
+                                if (DateTime.UtcNow > startTime + _orchestrationCompletionTimeout)
+                                {
+                                    Assert.True(false, $"The orchestration has not completed after {_orchestrationCompletionTimeout}");
+                                }
+
+                                await Task.Delay(TimeSpan.FromSeconds(2));
+                                break;
+                            }
+
+                        case HttpStatusCode.OK:
+                            {
+                                var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                                Assert.Equal("Completed", (string)statusResponseBody.runtimeStatus);
+                                Assert.Equal("helloWorld!", statusResponseBody.output.ToString());
+                                return;
+                            }
+
+                        default:
+                            Assert.True(false, $"Unexpected orchestration status code: {statusResponse.StatusCode}");
+                            break;
+                    }
+                }
+            }
+        }
+
+        [Fact]
+        public async Task OrchestratationCanAlwaysObtainTaskResult()
+        {
+            var initialResponse = await Utilities.GetHttpTriggerResponse("DurableClient", queryString: "?FunctionName=DurableOrchestratorGetTaskResult");
+            Assert.Equal(HttpStatusCode.Accepted, initialResponse.StatusCode);
+
+            var initialResponseBody = await initialResponse.Content.ReadAsStringAsync();
+            dynamic initialResponseBodyObject = JsonConvert.DeserializeObject(initialResponseBody);
+            var statusQueryGetUri = (string)initialResponseBodyObject.statusQueryGetUri;
+
+            var startTime = DateTime.UtcNow;
+
+            using (var httpClient = new HttpClient())
+            {
+                while (true)
+                {
+                    var statusResponse = await httpClient.GetAsync(statusQueryGetUri);
+                    switch (statusResponse.StatusCode)
+                    {
+                        case HttpStatusCode.Accepted:
+                        {
+                            var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                            var runtimeStatus = (string)statusResponseBody.runtimeStatus;
+                            Assert.True(
+                                runtimeStatus == "Running" || runtimeStatus == "Pending",
+                                $"Unexpected runtime status: {runtimeStatus}");
+
+                            if (DateTime.UtcNow > startTime + _orchestrationCompletionTimeout)
+                            {
+                                Assert.True(false, $"The orchestration has not completed after {_orchestrationCompletionTimeout}");
+                            }
+
+                            await Task.Delay(TimeSpan.FromSeconds(2));
+                            break;
+                        }
+
+                        case HttpStatusCode.OK:
+                        {
+                            var statusResponseBody = await GetResponseBodyAsync(statusResponse);
+                            Assert.Equal("Completed", (string)statusResponseBody.runtimeStatus);
+                            Assert.Equal("Hello world", statusResponseBody.output.ToString());
+                            return;
+                        }
+
+                        default:
+                            Assert.True(false, $"Unexpected orchestration status code: {statusResponse.StatusCode}");
+                            break;
+                    }
+                }
+            }
+        }
+
         [Fact]
         public async Task ActivityCanHaveQueueBinding()
         {

diff --git a/test/E2E/TestFunctionApp/DurableOrchestratorGetTaskResult/function.json b/test/E2E/TestFunctionApp/DurableOrchestratorGetTaskResult/function.json
@@ -0,0 +1,9 @@
+{
+  "bindings": [
+    {
+      "name": "Context",
+      "type": "orchestrationTrigger",
+      "direction": "in"
+    }
+  ]
+}
diff --git a/test/E2E/TestFunctionApp/DurableOrchestratorGetTaskResult/run.ps1 b/test/E2E/TestFunctionApp/DurableOrchestratorGetTaskResult/run.ps1
@@ -0,0 +1,8 @@
+param($Context)
+
+$output = @()
+
+$task = Invoke-DurableActivity -FunctionName 'DurableActivity' -Input "world" -NoWait
+$firstTask = Wait-DurableTask -Task @($task) -Any
+$output += Get-DurableTaskResult -Task @($firstTask)
+$output
diff --git a/test/Unit/Durable/ActivityInvocationTaskTests.cs b/test/Unit/Durable/ActivityInvocationTaskTests.cs
@@ -225,8 +225,8 @@ public void GetCompletedHistoryEvent_ReturnsTaskCompletedOrTaskFailed(bool succe
             var orchestrationContext = new OrchestrationContext { History = history };
 
             var task = new ActivityInvocationTask(FunctionName, FunctionInput);
-            var scheduledEvent = task.GetScheduledHistoryEvent(orchestrationContext);
-            var completedEvent = task.GetCompletedHistoryEvent(orchestrationContext, scheduledEvent);
+            var scheduledEvent = task.GetScheduledHistoryEvent(orchestrationContext, false);
+            var completedEvent = task.GetCompletedHistoryEvent(orchestrationContext, scheduledEvent, false);
 
             Assert.Equal(scheduledEvent.EventId, completedEvent.TaskScheduledId);
         }