Akka.Cluster.Metrics extension implementation

src/Akka.sln

@@ -207,6 +207,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Akka.Docs.Tests", "core\Akk
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Akka.Docs.Tutorials", "core\Akka.Docs.Tutorials\Akka.Docs.Tutorials.csproj", "{9AE636F2-988C-42E4-9B40-FF1F7177DF91}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Cluster.Metrics", "contrib\cluster\Akka.Cluster.Metrics\Akka.Cluster.Metrics.csproj", "{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}"
+EndProject
 Global
 	GlobalSection(Performance) = preSolution
 		HasPerformanceSessions = true
@@ -909,6 +911,18 @@ Global
 		{9AE636F2-988C-42E4-9B40-FF1F7177DF91}.Release|x64.Build.0 = Release|Any CPU
 		{9AE636F2-988C-42E4-9B40-FF1F7177DF91}.Release|x86.ActiveCfg = Release|Any CPU
 		{9AE636F2-988C-42E4-9B40-FF1F7177DF91}.Release|x86.Build.0 = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|x64.Build.0 = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Debug|x86.Build.0 = Debug|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|x64.ActiveCfg = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|x64.Build.0 = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|x86.ActiveCfg = Release|Any CPU
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -998,6 +1012,7 @@ Global
 		{6F8FECD6-6E39-473E-9B9A-9EE22CBF479F} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
 		{159F6312-3F12-4A84-9E7C-0A4B4FD72D1E} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
 		{9AE636F2-988C-42E4-9B40-FF1F7177DF91} = {01167D3C-49C4-4CDE-9787-C176D139ACDD}
+		{1E26E439-C9E1-4DBE-A783-D5FE31FCDD6F} = {76F58DC4-19F1-43EF-A6E2-EC1CC8395AC5}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {03AD8E21-7507-4E68-A4E9-F4A7E7273164}

src/contrib/cluster/Akka.Cluster.Metrics/Akka.Cluster.Metrics.csproj

@@ -0,0 +1,43 @@
+<Project Sdk="Microsoft.NET.Sdk">
+    <Import Project="..\..\..\common.props" />
+
+    <PropertyGroup>
+        <AssemblyTitle>Akka.Cluster.Metrics</AssemblyTitle>
+        <Description>
+            The member nodes of the cluster can collect system health metrics and publish that to other cluster nodes 
+            and to the registered subscribers on the system event bus with the help of Cluster Metrics Extension.
+        </Description>
+        <VersionSuffix>beta</VersionSuffix>
+        <TargetFramework>$(NetStandardLibVersion)</TargetFramework>
+        <PackageTags>$(AkkaPackageTags);network;cluster;sharding</PackageTags>
+        <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Google.Protobuf" Version="3.11.2" />
+    </ItemGroup>
+
+    <ItemGroup Condition=" '$(TargetFramework)' == '$(NetFrameworkLibVersion)' ">
+        <Reference Include="System.Core" />
+        <Reference Include="System.Xml.Linq" />
+        <Reference Include="System.Data.DataSetExtensions" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\..\..\core\Akka.Cluster\Akka.Cluster.csproj" />
+      <ProjectReference Include="..\..\..\core\Akka\Akka.csproj" />
+    </ItemGroup>
+
+    <PropertyGroup Condition=" '$(TargetFramework)' == '$(NetFrameworkLibVersion)' ">
+        <DefineConstants>$(DefineConstants);SERIALIZATION</DefineConstants>
+    </PropertyGroup>
+
+    <PropertyGroup Condition=" '$(TargetFramework)' == '$(NetStandardLibVersion)' ">
+        <DefineConstants>$(DefineConstants);CORECLR</DefineConstants>
+    </PropertyGroup>
+
+    <PropertyGroup Condition=" '$(Configuration)' == 'Release' ">
+        <DefineConstants>$(DefineConstants);RELEASE</DefineConstants>
+    </PropertyGroup>
+
+</Project>

src/contrib/cluster/Akka.Cluster.Metrics/ClusterMetrics.cs

@@ -0,0 +1,124 @@
+// //-----------------------------------------------------------------------
+// // <copyright file="ClusterMetricsExtension.cs" company="Akka.NET Project">
+// //     Copyright (C) 2009-2019 Lightbend Inc. <http://www.lightbend.com>
+// //     Copyright (C) 2013-2019 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// // </copyright>
+// //-----------------------------------------------------------------------
+
+using System;
+using Akka.Actor;
+using Akka.Annotations;
+using Akka.Cluster.Metrics.Configuration;
+using Akka.Cluster.Metrics.Events;
+using Akka.Cluster.Metrics.Helpers;
+using Akka.Cluster.Metrics.Serialization;
+
+namespace Akka.Cluster.Metrics
+{
+    /// <summary>
+    /// Cluster metrics extension.
+    ///
+    /// Cluster metrics is primarily for load-balancing of nodes. It controls metrics sampling
+    /// at a regular frequency, prepares highly variable data for further analysis by other entities,
+    /// and publishes the latest cluster metrics data around the node ring and local eventStream
+    /// to assist in determining the need to redirect traffic to the least-loaded nodes.
+    ///
+    /// Metrics sampling is delegated to the <see cref="IMetricsCollector"/>.
+    ///
+    /// Smoothing of the data for each monitored process is delegated to the
+    /// <see cref="NodeMetrics.Types.EWMA"/> for exponential weighted moving average.
+    /// </summary>
+    public class ClusterMetrics : IExtension
+    {
+        private readonly ExtendedActorSystem _system;
+
+        /// <summary>
+        /// Cluster metrics settings
+        /// </summary>
+        public ClusterMetricsSettings Settings { get; }
+
+        /// <summary>
+        /// Creates new <see cref="ClusterMetrics"/> for given actor system
+        /// </summary>
+        /// <param name="system"></param>
+        public ClusterMetrics(ExtendedActorSystem  system)
+        {
+            _system = system;
+            Settings = ClusterMetricsSettings.Create(system.Settings.Config);
+
+            Supervisor = _system.SystemActorOf(
+                Props.Create<ClusterMetricsSupervisor>().WithDispatcher(Settings.MetricsDispatcher).WithDeploy(Deploy.Local), 
+                Settings.SupervisorName);
+        }
+
+        /// <summary>
+        /// Creates new <see cref="ClusterMetrics"/> for given actor system
+        /// </summary>
+        public static ClusterMetrics Get(ActorSystem system)
+        {
+            return system.WithExtension<ClusterMetrics, ClusterMetricsExtensionProvider>();
+        }
+
+        /// <summary>
+        /// INTERNAL API
+        ///
+        /// Supervision strategy.
+        /// </summary>
+        [InternalApi]
+        public ClusterMetricsStrategy Strategy
+        {
+            get
+            {
+                return DynamicAccess.CreateInstanceFor<ClusterMetricsStrategy>(Settings.SupervisorStrategyProvider)
+                    .GetOrElse(() =>
+                    {
+                        _system.Log.Error(
+                            $"Configured strategy provider {Settings.SupervisorStrategyProvider} failed to load, " +
+                            $"using default {typeof(ClusterMetricsStrategy).Name}.");
+                        return new ClusterMetricsStrategy(Settings.SupervisorStrategyConfiguration);
+                    })
+                    .Get();
+            }
+        }
+
+        /// <summary>
+        /// Supervisor actor.
+        /// Accepts subtypes of <see cref="ClusterMetricsSupervisorMetadata.ICollectionControlMessage"/> to manage metrics collection at runtime
+        /// </summary>
+        public IActorRef Supervisor { get; }
+
+        /// <summary>
+        /// Subscribe user metrics listener actor unto <see cref="IClusterMetricsEvent"/>
+        /// events published by extension on the system event bus.
+        /// </summary>
+        public void Subscribe(IActorRef metricsListener)
+        {
+            _system.EventStream.Subscribe(metricsListener, typeof(IClusterMetricsEvent));
+        }
+
+        /// <summary>
+        /// Unubscribe user metrics listener actor from <see cref="IClusterMetricsEvent"/>
+        /// events published by extension on the system event bus.
+        /// </summary>
+        /// <param name="metricsListener"></param>
+        public void Unsubscribe(IActorRef metricsListener)
+        {
+            _system.EventStream.Unsubscribe(metricsListener, typeof(IClusterMetricsEvent));
+        }
+    }
+
+    /// <summary>
+    /// ClusterMetricsExtensionProvider
+    /// </summary>
+    public class ClusterMetricsExtensionProvider: ExtensionIdProvider<ClusterMetrics>
+    {
+        /// <summary>
+        /// Creates new <see cref="ClusterMetrics"/> for given actor system
+        /// </summary>
+        public override ClusterMetrics CreateExtension(ExtendedActorSystem system)
+        {
+            var extension = new ClusterMetrics(system);
+            return extension;
+        }
+    }
+}