Support KEDA queue scale trigger for Logic App workflow app

Common/Constants.cs

@@ -145,7 +145,11 @@ public static TimeSpan MaxAllowedExecutionTime
         public const string DurableTaskStorageProvider = "storageProvider";
         public const string DurableTaskMicrosoftSqlProviderType = "MicrosoftSQL";
         public const string MicrosoftSqlScaler = "mssql";
+        public const string AzureQueueScaler = "azure-queue";
         public const string DurableTask = "durableTask";
+        public const string WorkflowAppKind = "workflowApp";
+        public const string WorkflowExtensionName = "workflow";
+        public const string WorkflowSettingsName = "Settings";
         public const string Extensions = "extensions";
         public const string SitePackages = "SitePackages";
         public const string PackageNameTxt = "packagename.txt";

Kudu.Contracts/IEnvironment.cs

@@ -34,5 +34,6 @@ public interface IEnvironment
         bool IsOnLinuxConsumption { get; }      // e.g. True on Linux Consumption. False on App Service.
         bool IsK8SEApp { get; }
         string K8SEAppName { get;  }
+        string K8SEAppKind { get; }
     }
 }

Kudu.Core/Environment.cs

@@ -43,6 +43,7 @@ public class Environment : IEnvironment
         private readonly string _secondaryJobsBinariesPath;
         private readonly string _k8seAppName;
         private readonly string _k8seAppNamespace;
+        private readonly string _k8seAppKind;
 
 
         // This ctor is used only in unit tests
@@ -65,7 +66,8 @@ public Environment(
                 string requestId,
                 IHttpContextAccessor httpContextAccessor,
                 string k8seAppName = null,
-                string k8seAppNamespace = null)
+                string k8seAppNamespace = null,
+                string k8seAppKind = null)
         {
             if (repositoryPath == null)
             {
@@ -106,6 +108,7 @@ public Environment(
             _httpContextAccessor = httpContextAccessor;
             _k8seAppName = k8seAppName;
             _k8seAppNamespace = k8seAppNamespace;
+            _k8seAppKind = k8seAppKind;
         }
 
         public Environment(
@@ -116,7 +119,8 @@ public Environment(
                 string kuduConsoleFullPath,
                 IHttpContextAccessor httpContextAccessor,
                 string k8seAppName = null,
-                string k8seAppNamespace = null)
+                string k8seAppNamespace = null,
+                string k8seAppKind = null)
         {
             RootPath = rootPath;
 
@@ -133,6 +137,7 @@ public Environment(
             _locksPath = Path.Combine(SiteRootPath, Constants.LocksPath);
             _k8seAppName = k8seAppName;
             _k8seAppNamespace = k8seAppNamespace;
+            _k8seAppKind = k8seAppKind;
 
             if (OSDetector.IsOnWindows())
             {
@@ -499,6 +504,14 @@ public string K8SEAppNamespace
             }
         }
 
+        public string K8SEAppKind
+        {
+            get
+            {
+                return _k8seAppKind;
+            }
+        }
+
         public static string GetFreeSpaceHtml(string path)
         {
             try

Kudu.Core/Functions/KedaFunctionTriggerProvider.cs

@@ -1,4 +1,5 @@
-using Newtonsoft.Json.Linq;
+using Kudu.Core.Helpers;
+using Newtonsoft.Json.Linq;
 using System;
 using System.Collections.Generic;
 using System.IO;
@@ -9,7 +10,7 @@ namespace Kudu.Core.Functions
 {
     public static class KedaFunctionTriggerProvider
     {
-        public static IEnumerable<ScaleTrigger> GetFunctionTriggers(string zipFilePath)
+        public static IEnumerable<ScaleTrigger> GetFunctionTriggers(string zipFilePath, string appName = null, string appKind = null)
         {
             if (!File.Exists(zipFilePath))
             {
@@ -55,7 +56,18 @@ bool IsHostJson(string fullName)
                 return fullName.Equals(Constants.FunctionsHostConfigFile, StringComparison.OrdinalIgnoreCase);
             }
 
-            return CreateScaleTriggers(triggerBindings, hostJsonText);
+            var triggers = CreateScaleTriggers(triggerBindings, hostJsonText).ToList();
+
+            if (appKind?.ToLowerInvariant() == Constants.WorkflowAppKind.ToLowerInvariant())
+            {
+                // NOTE(haassyad) Check if the host json has the workflow extension loaded. If so we will add a queue scale trigger for the job dispatcher queue.
+                if (TryGetWorkflowKedaTrigger(hostJsonText, appName, out ScaleTrigger workflowScaleTrigger))
+                {
+                    triggers.Add(workflowScaleTrigger);
+                }
+            }
+
+            return triggers;
         }
 
         public static IEnumerable<ScaleTrigger> GetFunctionTriggers(IEnumerable<JObject> functionsJson, string hostJsonText)
@@ -222,6 +234,52 @@ internal static bool TryGetDurableKedaTrigger(string hostJsonText, out ScaleTrig
             return scaleTrigger != null;
         }
 
+        /// <summary>
+        /// Tries to add a scale trigger if the app is a workflow app.
+        /// </summary>
+        /// <param name="hostJsonText">The host.json text.</param>
+        /// <param name="appName">The app name.</param>
+        /// <param name="scaleTrigger">The scale trigger.</param>
+        /// <returns>true if a scale trigger was found</returns>
+        internal static bool TryGetWorkflowKedaTrigger(string hostJsonText, string appName, out ScaleTrigger scaleTrigger)
+        {
+            JObject hostJson = JObject.Parse(hostJsonText);
+
+            // Check the host.json file for workflow settings.
+            JObject workflowSettings = hostJson
+                .SelectToken(path: $"{Constants.Extensions}.{Constants.WorkflowExtensionName}.{Constants.WorkflowSettingsName}") as JObject;
+
+            // Get the queue length if specified, otherwise default to arbitrary value.
+            var queueLengthObject = workflowSettings?["Runtime.ScaleMonitor.KEDA.TargetQueueLength"];
+            var queueLength = queueLengthObject != null ? queueLengthObject.ToString() : "20";
+
+            // Get the host id if specified, otherwise default to app name.
+            var hostIdObject = workflowSettings?["Runtime.HostId"];
+            var hostId = hostIdObject != null ? hostIdObject.ToString() : appName;
+
+            // Hash the host id.
+            var hostSpecificStorageId = StringHelper
+                .EscapeAndTrimStorageKeyPrefix(HashHelper.MurmurHash64(hostId).ToString("X"), 32)
+                .ToLowerInvariant();
+
+            var queuePrefix = $"flow{hostSpecificStorageId}jobtriggers";
+
+            scaleTrigger = new ScaleTrigger
+            {
+                // Azure queue scaler reference: https://keda.sh/docs/2.2/scalers/azure-storage-queue/
+                Type = Constants.AzureQueueScaler,
+                Metadata = new Dictionary<string, string>
+                {
+                    // NOTE(haassyad): We only have one queue partition in single tenant.
+                    ["queueName"] = StringHelper.GetWorkflowQueueNameInternal(queuePrefix, 1),
+                    ["queueLength"] = queueLength,
+                    ["connectionStringFromEnv"] = "AzureWebJobsStorage",
+                }
+            };
+
+            return true;
+        }
+
         // match https://github.com/Azure/azure-functions-core-tools/blob/6bfab24b2743f8421475d996402c398d2fe4a9e0/src/Azure.Functions.Cli/Kubernetes/KEDA/V2/KedaV2Resource.cs#L91
         internal static IDictionary<string, string> PopulateMetadataDictionary(JToken t, string functionName)
         {

Kudu.Core/Helpers/HashHelper.cs

@@ -0,0 +1,138 @@
+using System.Text;
+
+namespace Kudu.Core.Helpers
+{
+    public static class HashHelper
+    {
+        /// <summary>
+        /// Computes 64-bit Murmur hash.
+        /// </summary>
+        /// <param name="str">The input string.</param>
+        /// <param name="seed">The input seed.</param>
+        public static ulong MurmurHash64(string str, uint seed = 0)
+        {
+            return HashHelper.MurmurHash64(Encoding.UTF8.GetBytes(str), seed);
+        }
+
+        /// <summary>
+        /// Computes 64-bit Murmur hash.
+        /// </summary>
+        /// <param name="data">The input data.</param>
+        /// <param name="seed">The input seed.</param>
+        public static ulong MurmurHash64(byte[] data, uint seed = 0)
+        {
+            const uint C1 = 0x239b961b;
+            const uint C2 = 0xab0e9789;
+            const uint C3 = 0x561ccd1b;
+            const uint C4 = 0x0bcaa747;
+            const uint C5 = 0x85ebca6b;
+            const uint C6 = 0xc2b2ae35;
+
+            int length = data.Length;
+
+            unchecked
+            {
+                uint h1 = seed;
+                uint h2 = seed;
+
+                int index = 0;
+                while (index + 7 < length)
+                {
+                    uint k1 = (uint)(data[index + 0] | data[index + 1] << 8 | data[index + 2] << 16 | data[index + 3] << 24);
+                    uint k2 = (uint)(data[index + 4] | data[index + 5] << 8 | data[index + 6] << 16 | data[index + 7] << 24);
+
+                    k1 *= C1;
+                    k1 = k1.RotateLeft32(15);
+                    k1 *= C2;
+                    h1 ^= k1;
+                    h1 = h1.RotateLeft32(19);
+                    h1 += h2;
+                    h1 = (h1 * 5) + C3;
+
+                    k2 *= C2;
+                    k2 = k2.RotateLeft32(17);
+                    k2 *= C1;
+                    h2 ^= k2;
+                    h2 = h2.RotateLeft32(13);
+                    h2 += h1;
+                    h2 = (h2 * 5) + C4;
+
+                    index += 8;
+                }
+
+                int tail = length - index;
+                if (tail > 0)
+                {
+                    uint k1 = (tail >= 4) ? (uint)(data[index + 0] | data[index + 1] << 8 | data[index + 2] << 16 | data[index + 3] << 24) :
+                              (tail == 3) ? (uint)(data[index + 0] | data[index + 1] << 8 | data[index + 2] << 16) :
+                              (tail == 2) ? (uint)(data[index + 0] | data[index + 1] << 8) :
+                                            (uint)data[index + 0];
+
+                    k1 *= C1;
+                    k1 = k1.RotateLeft32(15);
+                    k1 *= C2;
+                    h1 ^= k1;
+
+                    if (tail > 4)
+                    {
+                        uint k2 = (tail == 7) ? (uint)(data[index + 4] | data[index + 5] << 8 | data[index + 6] << 16) :
+                                  (tail == 6) ? (uint)(data[index + 4] | data[index + 5] << 8) :
+                                                (uint)data[index + 4];
+
+                        k2 *= C2;
+                        k2 = k2.RotateLeft32(17);
+                        k2 *= C1;
+                        h2 ^= k2;
+                    }
+                }
+
+                h1 ^= (uint)length;
+                h2 ^= (uint)length;
+
+                h1 += h2;
+                h2 += h1;
+
+                h1 ^= h1 >> 16;
+                h1 *= C5;
+                h1 ^= h1 >> 13;
+                h1 *= C6;
+                h1 ^= h1 >> 16;
+
+                h2 ^= h2 >> 16;
+                h2 *= C5;
+                h2 ^= h2 >> 13;
+                h2 *= C6;
+                h2 ^= h2 >> 16;
+
+                h1 += h2;
+                h2 += h1;
+
+                return ((ulong)h2 << 32) | (ulong)h1;
+            }
+        }
+
+        #region RotateLeft
+
+        /// <summary>
+        /// Rotates the bits in the provided value to the left (where the number of bits is specified).
+        /// </summary>
+        /// <param name="value">The value to be rotated.</param>
+        /// <param name="count">The number of bits to rotate.</param>
+        private static uint RotateLeft32(this uint value, int count)
+        {
+            return (value << count) | (value >> (32 - count));
+        }
+
+        /// <summary>
+        /// Rotates the bits in the provided value to the right (where the number of bits is specified).
+        /// </summary>
+        /// <param name="value">The value to be rotated.</param>
+        /// <param name="count">The number of bits to rotate.</param>
+        private static ulong RotateLeft64(this ulong value, int count)
+        {
+            return (value << count) | (value >> (64 - count));
+        }
+
+        #endregion
+    }
+}