Adding external cosmos db scaler

Keda.CosmosDB.Scaler/Dockerfile

@@ -0,0 +1,17 @@
+FROM mcr.microsoft.com/dotnet/core/aspnet:3.1-buster-slim AS base
+WORKDIR /app
+EXPOSE 80
+EXPOSE 443
+
+FROM mcr.microsoft.com/dotnet/core/sdk:3.1-buster AS build
+WORKDIR /src
+COPY ["src/Keda.CosmosDB.Scaler.csproj", "src/"]
+RUN dotnet restore "src/Keda.CosmosDB.Scaler.csproj"
+COPY . .
+WORKDIR "/src/"
+RUN dotnet publish "src/Keda.CosmosDB.Scaler.csproj" -c Release -o /app/publish
+
+FROM base AS final
+WORKDIR /app
+COPY --from=build /app/publish .
+ENTRYPOINT ["dotnet", "Keda.CosmosDB.Scaler.dll"]

Keda.CosmosDB.Scaler/Keda.CosmosDB.Scaler.sln

@@ -0,0 +1,31 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.30907.101
+MinimumVisualStudioVersion = 10.0.40219.1
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Keda.CosmosDB.Scaler", "src\Keda.CosmosDB.Scaler.csproj", "{DF930E63-7F41-43B3-932D-2FD4E810FBA0}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Keda.CosmosDB.Scaler.Tests", "test\Keda.CosmosDB.Scaler.Tests.csproj", "{31AF2207-6571-4587-964C-0EE3951D86C2}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{DF930E63-7F41-43B3-932D-2FD4E810FBA0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DF930E63-7F41-43B3-932D-2FD4E810FBA0}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DF930E63-7F41-43B3-932D-2FD4E810FBA0}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DF930E63-7F41-43B3-932D-2FD4E810FBA0}.Release|Any CPU.Build.0 = Release|Any CPU
+		{31AF2207-6571-4587-964C-0EE3951D86C2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{31AF2207-6571-4587-964C-0EE3951D86C2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{31AF2207-6571-4587-964C-0EE3951D86C2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{31AF2207-6571-4587-964C-0EE3951D86C2}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {E7B53967-47F0-4201-B16E-4F231B896245}
+	EndGlobalSection
+EndGlobal

Keda.CosmosDB.Scaler/src/Constants.cs

@@ -0,0 +1,22 @@
+namespace Keda.CosmosDB.Scaler
+{
+    public class Constants
+    {
+        public const string AccountKey = "AccountKey";
+        public const string AccountEndpoint = "AccountEndpoint";
+        public const string AzureCosmosDBMetricPrefix = "azure-cosmosDB";
+        public const string DefaultHostName = "defaultName";
+
+        public const string CollectionNameMetadata = "collectionName";
+        public const string DatabaseNameMetadata = "databaseName";
+        public const string ConnectionStringMetadata ="cosmosDBConnectionString";
+        public const string AccountNameMetadata = "accountName";
+
+        public const string LeasesConnectionStringMetadata = "leasesCosmosDBConnectionString";
+        public const string LeaseDatabaseNameMetadata = "leaseDatabaseName";
+        public const string LeaseCollectionNameMetadata = "leaseCollectionName";
+        public const string LeaseCollectionPrefixMetadata = "leaseCollectionPrefix";
+
+        public const int GrpcPort = 4050;
+    }
+}

Keda.CosmosDB.Scaler/src/Helpers/StringHelpers.cs

@@ -0,0 +1,10 @@
+namespace Keda.CosmosDB.Scaler.Extensions
+{
+    public static class StringHelpers
+    {
+        public static string NormalizeString(string inputString)
+        {
+            return inputString?.Replace("/", "-").Replace(".", "-").Replace(":", "-").Replace("%", "-").ToLower();
+        }
+    }
+}

Keda.CosmosDB.Scaler/src/Keda.CosmosDB.Scaler.csproj

@@ -0,0 +1,22 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Grpc.AspNetCore" Version="2.34.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Hosting" Version="2.2.7" />
+    <PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Azure.Cosmos" Version="3.18.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="5.0.0" />
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <Protobuf Include="Protos\externalscaler.proto" GrpcServices="Server" />
+  </ItemGroup>
+
+</Project>

Keda.CosmosDB.Scaler/src/Program.cs

@@ -0,0 +1,32 @@
+using System.Net;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Server.Kestrel.Core;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Keda.CosmosDB.Scaler
+{
+    class Program
+    {
+        static void Main(string[] args)
+        {
+            CreateHostBuilder(args).Build().Run();
+        }
+
+        public static IHostBuilder CreateHostBuilder(string[] args) =>
+            Host.CreateDefaultBuilder(args)
+                .ConfigureLogging(logging =>
+                {
+                    logging.ClearProviders();
+                    logging.AddConsole();
+                })
+            .ConfigureWebHostDefaults(webBuilder =>
+            {
+                webBuilder.ConfigureKestrel(options =>
+                {
+                    options.Listen(IPAddress.Any, Constants.GrpcPort, o => o.Protocols = HttpProtocols.Http2);
+                });
+                webBuilder.UseStartup<Startup>();
+            });
+    }
+}

Keda.CosmosDB.Scaler/src/Protos/externalscaler.proto

@@ -0,0 +1,51 @@
+syntax = "proto3";
+
+option csharp_namespace = "Keda.CosmosDB.Scaler.Protos";
+
+package externalscaler;
+option go_package = ".;externalscaler";
+
+service ExternalScaler {
+    rpc IsActive(ScaledObjectRef) returns (IsActiveResponse) {}
+    rpc StreamIsActive(ScaledObjectRef) returns (stream IsActiveResponse) {}
+    rpc GetMetricSpec(ScaledObjectRef) returns (GetMetricSpecResponse) {}
+    rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}
+}
+
+message ScaledObjectRef {
+    string name = 1;
+    string namespace = 2;
+    map<string, string> scalerMetadata = 3;
+}
+
+message IsActiveResponse {
+    bool result = 1;
+}
+
+message GetMetricSpecResponse {
+    repeated MetricSpec metricSpecs = 1;
+}
+
+message MetricSpec {
+    string metricName = 1;
+    int64 targetSize = 2;
+}
+
+message GetMetricsRequest {
+    ScaledObjectRef scaledObjectRef = 1;
+    string metricName = 2;
+}
+
+message GetMetricsResponse {
+    repeated MetricValue metricValues = 1;
+}
+
+message MetricValue {
+    string metricName = 1;
+    int64 metricValue = 2;
+}
+
+message NewRequest {
+	ScaledObjectRef scaledObjectRef = 1;
+	map<string, string> metadata = 2;
+}

Keda.CosmosDB.Scaler/src/Repository/CosmosDBEstimator.cs

@@ -0,0 +1,70 @@
+using Keda.CosmosDB.Scaler.Services;
+using Microsoft.Azure.Cosmos;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Keda.CosmosDB.Scaler.Repository
+{
+    public class CosmosDBEstimator : ICosmosDBEstimator
+    {
+        private ConcurrentDictionary<CosmosDBTrigger, ChangeFeedEstimator> _changeFeedBuilderMap;
+
+        public CosmosDBEstimator()
+        {
+            _changeFeedBuilderMap = new ConcurrentDictionary<CosmosDBTrigger, ChangeFeedEstimator>(new CosmosDBTriggerComparer());
+        }
+
+        internal ChangeFeedEstimator GetOrCreateEstimator(CosmosDBTrigger trigger)
+        {
+            if (_changeFeedBuilderMap.TryGetValue(trigger, out ChangeFeedEstimator estimator))
+            {
+                return estimator;
+            }
+
+            CosmosClientOptions clientOptions = new CosmosClientOptions()
+            {
+                ConnectionMode = ConnectionMode.Gateway
+            };
+
+            CosmosClient monitoredCosmosDBService = new CosmosClient(trigger.CosmosDBConnectionString, clientOptions);
+            CosmosClient leaseCosmosDBService;
+
+            if (trigger.CosmosDBConnectionString.Equals(
+                trigger.Lease.LeasesCosmosDBConnectionString, StringComparison.OrdinalIgnoreCase))
+            {
+                leaseCosmosDBService = monitoredCosmosDBService;
+            }
+            else
+            {
+                leaseCosmosDBService = new CosmosClient(trigger.Lease.LeasesCosmosDBConnectionString, clientOptions);
+            }
+
+            var monitoredContainer = monitoredCosmosDBService.GetContainer(trigger.DatabaseName, trigger.CollectionName);
+            var leaseContainer = leaseCosmosDBService.GetContainer(trigger.Lease.LeaseDatabaseName, trigger.Lease.LeaseCollectionName);
+
+            estimator = monitoredContainer.GetChangeFeedEstimator(trigger.Lease.LeaseCollectionPrefix ?? string.Empty, leaseContainer);
+
+            _changeFeedBuilderMap.TryAdd(trigger, estimator);
+            return estimator;
+        }
+
+        public async Task<long> GetEstimatedWork(CosmosDBTrigger trigger)
+        {
+            ChangeFeedEstimator estimator = GetOrCreateEstimator(trigger);
+            List<ChangeFeedProcessorState> partitionWorkList = new List<ChangeFeedProcessorState>();
+
+            using FeedIterator<ChangeFeedProcessorState> estimatorIterator = estimator.GetCurrentStateIterator();
+            {
+                while (estimatorIterator.HasMoreResults)
+                {
+                    FeedResponse<ChangeFeedProcessorState> response = await estimatorIterator.ReadNextAsync();
+                    partitionWorkList.AddRange(response);
+                }
+            }
+            return partitionWorkList.Sum(item => item.EstimatedLag); ;
+        }
+    }
+}

Keda.CosmosDB.Scaler/src/Repository/CosmosDBTriggerComparer.cs

@@ -0,0 +1,32 @@
+using Keda.CosmosDB.Scaler.Services;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Keda.CosmosDB.Scaler.Repository
+{
+    public class CosmosDBTriggerComparer : IEqualityComparer<CosmosDBTrigger>
+    {
+        public bool Equals([AllowNull] CosmosDBTrigger x, [AllowNull] CosmosDBTrigger y)
+        {
+            return (x.AccountName.Equals(y.AccountName, StringComparison.OrdinalIgnoreCase) &&
+                x.CollectionName.Equals(y.CollectionName, StringComparison.OrdinalIgnoreCase) &&
+                x.CosmosDBConnectionString.Equals(y.CosmosDBConnectionString, StringComparison.OrdinalIgnoreCase) &&
+                x.DatabaseName.Equals(y.DatabaseName, StringComparison.OrdinalIgnoreCase) &&
+                x.Lease.LeasesCosmosDBConnectionString.Equals(y.Lease.LeasesCosmosDBConnectionString, StringComparison.OrdinalIgnoreCase) &&
+                x.Lease.LeaseDatabaseName.Equals(y.Lease.LeaseDatabaseName, StringComparison.OrdinalIgnoreCase) &&
+                x.Lease.LeaseCollectionName.Equals(y.Lease.LeaseCollectionName, StringComparison.OrdinalIgnoreCase));
+        }
+
+        public int GetHashCode([DisallowNull] CosmosDBTrigger obj)
+        {
+            if (obj == null)
+            {
+                return 0;
+            }
+
+            return obj.AccountName.GetHashCode() ^ obj.CollectionName.GetHashCode() ^ obj.CosmosDBConnectionString.GetHashCode()
+                ^ obj.DatabaseName.GetHashCode() ^ obj.Lease.LeasesCosmosDBConnectionString.GetHashCode() ^ obj.Lease.LeaseDatabaseName.GetHashCode() ^ obj.Lease.LeaseCollectionName.GetHashCode();
+        }
+    }
+}

Keda.CosmosDB.Scaler/src/Repository/ICosmosDBEstimator.cs

@@ -0,0 +1,10 @@
+using Keda.CosmosDB.Scaler.Services;
+using System.Threading.Tasks;
+
+namespace Keda.CosmosDB.Scaler.Repository
+{
+    public interface ICosmosDBEstimator
+    {
+        Task<long> GetEstimatedWork(CosmosDBTrigger trigger);
+    }
+}