集群指标扩展

依赖

为了使用集群指标扩展（Cluster Metrics Extension），你需要将以下依赖添加到你的项目中：

  1. <!-- Maven -->
  2. <dependency>
  3.   <groupId>com.typesafe.akka</groupId>
  4.   <artifactId>akka-cluster-metrics_2.12</artifactId>
  5.   <version>2.5.22</version>
  6. </dependency>
  8. <!-- Gradle -->
  9. dependencies {
  10.   compile group: 'com.typesafe.akka', name: 'akka-cluster-metrics_2.12', version: '2.5.22'
  11. }
  13. <!-- sbt -->
  14. libraryDependencies += "com.typesafe.akka" %% "akka-cluster-metrics" % "2.5.22"

并将以下配置添加到application.conf中：

  1. akka.extensions = [ "akka.cluster.metrics.ClusterMetricsExtension" ]

简介

集群的成员节点可以收集系统健康指标，并在集群指标扩展的帮助下将其发布到其他集群节点和系统事件总线上注册的订阅者。

集群指标信息主要用于负载均衡路由器（load-balancing routers），也可用于实现基于指标的高级节点生命周期，例如当 CPU 窃取时间过多时“节点让它崩溃”。

如果启用了该功能，状态为「WeaklyUp」的集群成员将参与集群指标收集和分发。

指标收集器

指标集合委托给akka.cluster.metrics.MetricsCollector.的实现。

不同的收集器（collector）实现提供发布到集群的不同指标子集。当未设置Sigar时，某些消息路由和让其崩溃功能可能无法工作。

集群指标扩展附带两个内置收集器实现：

  1. akka.cluster.metrics.SigarMetricsCollector，它要求提供Sigar，并且更丰富/更精确
  2. akka.cluster.metrics.JmxMetricsCollector，用作回退，不太丰富/精确

你也可以插入（plug-in）自己的指标收集器实现。

默认情况下，指标扩展将使用收集器提供程序回滚，并尝试按以下顺序加载它们：

  1. 配置的用户提供的收集器
  2. 内置的akka.cluster.metrics.SigarMetricsCollector
  3. 最后是akka.cluster.metrics.JmxMetricsCollector

指标事件

指标扩展定期地将集群指标的当前快照发布到节点系统事件总线。

发布间隔由akka.cluster.metrics.collector.sample-interval设置控制。

akka.cluster.metrics.ClusterMetricsChanged事件的有效负载将包含节点的最新指标，以及在收集器采样间隔期间接收到的其他群集成员节点指标流言。

你可以通过指标侦听器 Actor 订阅这些事件，以实现自定义节点生命周期：

  1. ClusterMetricsExtension.get(system).subscribe(metricsListenerActor);

Hyperic Sigar 配置

与可以从普通 JMX MBean 中检索到的指标相比，用户提供的指标收集器和内置的指标收集器都可以选择使用Hyperic Sigar来获取更广泛、更准确的指标范围。

Sigar使用的是本机 O/S 库，需要提供库，即在运行时将 O/S 本机库部署、提取和加载到 JVM 中。

用户可以通过以下方式之一提供Sigar类和本机库：

  1. 使用「Kamon sigar-loader」加载器用作用户项目的项目依赖项。指标扩展将根据需要在Kamon sigar provisioner的帮助下提取和加载Sigar库。
  2. 使用「Kamon sigar-loader」加载器作为 Java 代理：java -javaagent:/path/to/sigar-loader.jar。Kamon sigar loader代理将在 JVM 启动期间提取和加载Sigar库。
  3. 将sigar.jar放在classpath上，将 O/S 的Sigar本机库放在java.library.path上。用户需要手动管理项目依赖项和库部署。
  • 警告：当使用Kamon sigar loader并在同一主机上运行同一应用程序的多个实例时，必须确保将Sigar库提取到一个唯一的每个实例目录中。你可以使用akka.cluster.metrics.native-library-extract-folder配置设置控制提取目录。

为了使用Sigar的功能，需要在用户项目中添加以下依赖项：

  1. <!-- Maven -->
  2. <dependency>
  3.   <groupId>io.kamon</groupId>
  4.   <artifactId>sigar-loader</artifactId>
  5.   <version>1.6.6-rev002</version>
  6. </dependency>
  8. <!-- Gradle -->
  9. dependencies {
  10.   compile group: 'io.kamon', name: 'sigar-loader', version: '1.6.6-rev002'
  11. }
  13. <!-- sbt -->
  14. libraryDependencies += "io.kamon" % "sigar-loader" % "1.6.6-rev002"

你可以从「Maven Central」中下载Kamon sigar loader的依赖包。

自适应负载平衡

AdaptiveLoadBalancingPool / AdaptiveLoadBalancingGroup根据集群指标数据对集群节点的消息执行负载平衡。它使用随机选择的路由，概率来自于相应节点的剩余容量。它可以配置为使用特定的MetricsSelector来产生概率，即a.k.a.权重：

  • heap / HeapMetricsSelector - 已用和最大 JVM 堆内存。基于剩余堆容量的权重；(max - used) / max
  • load / SystemLoadAverageMetricsSelector - 过去 1 分钟的系统平均负载，在 Linux 系统顶部可以找到相应的值。如果系统平均负载接近cpus/cores的数量，则系统可能接近瓶颈。基于剩余负载能力的权重；1 - (load / processors)
  • cpu / CpuMetricsSelector - 以百分比表示的 CPU 利用率，User + Sys + Nice + Wait之和。基于剩余 CPU 容量的权重；1 - utilization
  • mix / MixMetricsSelector - 组合堆、CPU 和负载。基于组合选择器剩余容量平均值的权重。
  • akka.cluster.metrics.MetricsSelector的任何自定义实现

使用「指数加权移动平均值」平滑收集的指标值。在「集群配置」中，你可以调整过去的数据相对于新数据的衰减速度。

让我们来看看这台正在运行的路由器。还有什么比计算阶乘（factorial）更苛刻的呢？

执行阶乘计算的后端工作程序：

  1. public class FactorialBackend extends AbstractActor {
  3.   @Override
  4.   public Receive createReceive() {
  5.     return receiveBuilder()
  6.         .match(
  7.             Integer.class,
  8.             n -> {
  9.               CompletableFuture<FactorialResult> result =
  10.                   CompletableFuture.supplyAsync(() -> factorial(n))
  11.                       .thenApply((factorial) -> new FactorialResult(n, factorial));
  13.               pipe(result, getContext().dispatcher()).to(getSender());
  14.             })
  15.         .build();
  16.   }
  18.   BigInteger factorial(int n) {
  19.     BigInteger acc = BigInteger.ONE;
  20.     for (int i = 1; i <= n; ++i) {
  21.       acc = acc.multiply(BigInteger.valueOf(i));
  22.     }
  23.     return acc;
  24.   }
  25. }

接收用户作业并通过路由器委派到后端的前端：

  1. public class FactorialFrontend extends AbstractActor {
  2.   final int upToN;
  3.   final boolean repeat;
  5.   LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  7.   ActorRef backend =
  8.       getContext().actorOf(FromConfig.getInstance().props(), "factorialBackendRouter");
  10.   public FactorialFrontend(int upToN, boolean repeat) {
  11.     this.upToN = upToN;
  12.     this.repeat = repeat;
  13.   }
  15.   @Override
  16.   public void preStart() {
  17.     sendJobs();
  18.     getContext().setReceiveTimeout(Duration.ofSeconds(10));
  19.   }
  21.   @Override
  22.   public Receive createReceive() {
  23.     return receiveBuilder()
  24.         .match(
  25.             FactorialResult.class,
  26.             result -> {
  27.               if (result.n == upToN) {
  28.                 log.debug("{}! = {}", result.n, result.factorial);
  29.                 if (repeat) sendJobs();
  30.                 else getContext().stop(getSelf());
  31.               }
  32.             })
  33.         .match(
  34.             ReceiveTimeout.class,
  35.             x -> {
  36.               log.info("Timeout");
  37.               sendJobs();
  38.             })
  39.         .build();
  40.   }
  42.   void sendJobs() {
  43.     log.info("Starting batch of factorials up to [{}]", upToN);
  44.     for (int n = 1; n <= upToN; n++) {
  45.       backend.tell(n, getSelf());
  46.     }
  47.   }
  48. }

如你所见，路由器的定义方式与其他路由器相同，在这种情况下，配置如下：

  1. akka.actor.deployment {
  2.   /factorialFrontend/factorialBackendRouter = {
  3.     # Router type provided by metrics extension.
  4.     router = cluster-metrics-adaptive-group
  5.     # Router parameter specific for metrics extension.
  6.     # metrics-selector = heap
  7.     # metrics-selector = load
  8.     # metrics-selector = cpu
  9.     metrics-selector = mix
  10.     #
  11.     routees.paths = ["/user/factorialBackend"]
  12.     cluster {
  13.       enabled = on
  14.       use-roles = ["backend"]
  15.       allow-local-routees = off
  16.     }
  17.   }
  18. }

只有router类型和metrics-selector参数特定于此路由器，其他事物的工作方式与其他路由器相同。

同样类型的路由器也可以在代码中定义：

  1. int totalInstances = 100;
  2. Iterable<String> routeesPaths = Arrays.asList("/user/factorialBackend", "");
  3. boolean allowLocalRoutees = true;
  4. Set<String> useRoles = new HashSet<>(Arrays.asList("backend"));
  5. ActorRef backend =
  6.     getContext()
  7.         .actorOf(
  8.             new ClusterRouterGroup(
  9.                     new AdaptiveLoadBalancingGroup(
  10.                         HeapMetricsSelector.getInstance(), Collections.<String>emptyList()),
  11.                     new ClusterRouterGroupSettings(
  12.                         totalInstances, routeesPaths, allowLocalRoutees, useRoles))
  13.                 .props(),
  14.             "factorialBackendRouter2");
  16. int totalInstances = 100;
  17. int maxInstancesPerNode = 3;
  18. boolean allowLocalRoutees = false;
  19. Set<String> useRoles = new HashSet<>(Arrays.asList("backend"));
  20. ActorRef backend =
  21.     getContext()
  22.         .actorOf(
  23.             new ClusterRouterPool(
  24.                     new AdaptiveLoadBalancingPool(
  25.                         SystemLoadAverageMetricsSelector.getInstance(), 0),
  26.                     new ClusterRouterPoolSettings(
  27.                         totalInstances, maxInstancesPerNode, allowLocalRoutees, useRoles))
  28.                 .props(Props.create(FactorialBackend.class)),
  29.             "factorialBackendRouter3");

运行自适应负载平衡示例最简单的方法下载「Akka Cluster Sample with Java」中的代码和教程。它包含有关如何运行自适应负载平衡示例的说明，此示例的源代码也可以在「 Akka Samples Repository」中找到。

订阅指标事件

可以直接订阅指标事件来实现其他功能。

  1. import akka.actor.AbstractActor;
  2. import akka.cluster.Cluster;
  3. import akka.cluster.ClusterEvent.CurrentClusterState;
  4. import akka.cluster.metrics.ClusterMetricsChanged;
  5. import akka.cluster.metrics.NodeMetrics;
  6. import akka.cluster.metrics.StandardMetrics;
  7. import akka.cluster.metrics.StandardMetrics.HeapMemory;
  8. import akka.cluster.metrics.StandardMetrics.Cpu;
  9. import akka.cluster.metrics.ClusterMetricsExtension;
  10. import akka.event.Logging;
  11. import akka.event.LoggingAdapter;
  13. public class MetricsListener extends AbstractActor {
  14.   LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  16.   Cluster cluster = Cluster.get(getContext().getSystem());
  18.   ClusterMetricsExtension extension = ClusterMetricsExtension.get(getContext().getSystem());
  20.   // Subscribe unto ClusterMetricsEvent events.
  21.   @Override
  22.   public void preStart() {
  23.     extension.subscribe(getSelf());
  24.   }
  26.   // Unsubscribe from ClusterMetricsEvent events.
  27.   @Override
  28.   public void postStop() {
  29.     extension.unsubscribe(getSelf());
  30.   }
  32.   @Override
  33.   public Receive createReceive() {
  34.     return receiveBuilder()
  35.         .match(
  36.             ClusterMetricsChanged.class,
  37.             clusterMetrics -> {
  38.               for (NodeMetrics nodeMetrics : clusterMetrics.getNodeMetrics()) {
  39.                 if (nodeMetrics.address().equals(cluster.selfAddress())) {
  40.                   logHeap(nodeMetrics);
  41.                   logCpu(nodeMetrics);
  42.                 }
  43.               }
  44.             })
  45.         .match(
  46.             CurrentClusterState.class,
  47.             message -> {
  48.               // Ignore.
  49.             })
  50.         .build();
  51.   }
  53.   void logHeap(NodeMetrics nodeMetrics) {
  54.     HeapMemory heap = StandardMetrics.extractHeapMemory(nodeMetrics);
  55.     if (heap != null) {
  56.       log.info("Used heap: {} MB", ((double) heap.used()) / 1024 / 1024);
  57.     }
  58.   }
  60.   void logCpu(NodeMetrics nodeMetrics) {
  61.     Cpu cpu = StandardMetrics.extractCpu(nodeMetrics);
  62.     if (cpu != null && cpu.systemLoadAverage().isDefined()) {
  63.       log.info("Load: {} ({} processors)", cpu.systemLoadAverage().get(), cpu.processors());
  64.     }
  65.   }
  66. }

自定义指标收集器

指标集合委托给akka.cluster.metrics.MetricsCollector的实现

你也可以插入自己的指标收集器，而不是内置的akka.cluster.metrics.SigarMetricsCollector或akka.cluster.metrics.JmxMetricsCollector。

看看这两个实现的灵感。

自定义指标收集器实现类必须在akka.cluster.metrics.collector.provider配置属性中指定。

配置

可以使用以下属性配置群集指标扩展：

  1. ##############################################
  2. # Akka Cluster Metrics Reference Config File #
  3. ##############################################
  5. # This is the reference config file that contains all the default settings.
  6. # Make your edits in your application.conf in order to override these settings.
  8. # Sigar provisioning:
  9. #
  10. #  User can provision sigar classes and native library in one of the following ways:
  11. # 
  12. #  1) Use https://github.com/kamon-io/sigar-loader Kamon sigar-loader as a project dependency for the user project.
  13. #  Metrics extension will extract and load sigar library on demand with help of Kamon sigar provisioner.
  14. # 
  15. #  2) Use https://github.com/kamon-io/sigar-loader Kamon sigar-loader as java agent: `java -javaagent:/path/to/sigar-loader.jar`
  16. #  Kamon sigar loader agent will extract and load sigar library during JVM start.
  17. # 
  18. #  3) Place `sigar.jar` on the `classpath` and sigar native library for the o/s on the `java.library.path`
  19. #  User is required to manage both project dependency and library deployment manually.
  21. # Cluster metrics extension.
  22. # Provides periodic statistics collection and publication throughout the cluster.
  23. akka.cluster.metrics {
  24.   # Full path of dispatcher configuration key.
  25.   # Use "" for default key `akka.actor.default-dispatcher`.
  26.   dispatcher = ""
  27.   # How long should any actor wait before starting the periodic tasks.
  28.   periodic-tasks-initial-delay = 1s
  29.   # Sigar native library extract location.
  30.   # Use per-application-instance scoped location, such as program working directory.
  31.   native-library-extract-folder = ${user.dir}"/native"
  32.   # Metrics supervisor actor.
  33.   supervisor {
  34.     # Actor name. Example name space: /system/cluster-metrics
  35.     name = "cluster-metrics"
  36.     # Supervision strategy.
  37.     strategy {
  38.       #
  39.       # FQCN of class providing `akka.actor.SupervisorStrategy`.
  40.       # Must have a constructor with signature `<init>(com.typesafe.config.Config)`.
  41.       # Default metrics strategy provider is a configurable extension of `OneForOneStrategy`.
  42.       provider = "akka.cluster.metrics.ClusterMetricsStrategy"
  43.       #
  44.       # Configuration of the default strategy provider.
  45.       # Replace with custom settings when overriding the provider.
  46.       configuration = {
  47.         # Log restart attempts.
  48.         loggingEnabled = true
  49.         # Child actor restart-on-failure window.
  50.         withinTimeRange = 3s
  51.         # Maximum number of restart attempts before child actor is stopped.
  52.         maxNrOfRetries = 3
  53.       }
  54.     }
  55.   }
  56.   # Metrics collector actor.
  57.   collector {
  58.     # Enable or disable metrics collector for load-balancing nodes.
  59.     # Metrics collection can also be controlled at runtime by sending control messages
  60.     # to /system/cluster-metrics actor: `akka.cluster.metrics.{CollectionStartMessage,CollectionStopMessage}`
  61.     enabled = on
  62.     # FQCN of the metrics collector implementation.
  63.     # It must implement `akka.cluster.metrics.MetricsCollector` and
  64.     # have public constructor with akka.actor.ActorSystem parameter.
  65.     # Will try to load in the following order of priority:
  66.     # 1) configured custom collector 2) internal `SigarMetricsCollector` 3) internal `JmxMetricsCollector`
  67.     provider = ""
  68.     # Try all 3 available collector providers, or else fail on the configured custom collector provider.
  69.     fallback = true
  70.     # How often metrics are sampled on a node.
  71.     # Shorter interval will collect the metrics more often.
  72.     # Also controls frequency of the metrics publication to the node system event bus.
  73.     sample-interval = 3s
  74.     # How often a node publishes metrics information to the other nodes in the cluster.
  75.     # Shorter interval will publish the metrics gossip more often.
  76.     gossip-interval = 3s
  77.     # How quickly the exponential weighting of past data is decayed compared to
  78.     # new data. Set lower to increase the bias toward newer values.
  79.     # The relevance of each data sample is halved for every passing half-life
  80.     # duration, i.e. after 4 times the half-life, a data sample’s relevance is
  81.     # reduced to 6% of its original relevance. The initial relevance of a data
  82.     # sample is given by 1 – 0.5 ^ (collect-interval / half-life).
  83.     moving-average-half-life = 12s
  84.   }
  85. }
  87. # Cluster metrics extension serializers and routers.
  88. akka.actor {
  89.   # Protobuf serializer for remote cluster metrics messages.
  90.   serializers {
  91.     akka-cluster-metrics = "akka.cluster.metrics.protobuf.MessageSerializer"
  92.   }
  93.   # Interface binding for remote cluster metrics messages.
  94.   serialization-bindings {
  95.     "akka.cluster.metrics.ClusterMetricsMessage" = akka-cluster-metrics
  96.     "akka.cluster.metrics.AdaptiveLoadBalancingPool" = akka-cluster-metrics
  97.     "akka.cluster.metrics.MixMetricsSelector" = akka-cluster-metrics
  98.     "akka.cluster.metrics.CpuMetricsSelector$" = akka-cluster-metrics
  99.     "akka.cluster.metrics.HeapMetricsSelector$" = akka-cluster-metrics
  100.     "akka.cluster.metrics.SystemLoadAverageMetricsSelector$" = akka-cluster-metrics
  101.   }
  102.   # Globally unique metrics extension serializer identifier.
  103.   serialization-identifiers {
  104.     "akka.cluster.metrics.protobuf.MessageSerializer" = 10
  105.   }
  106.   #  Provide routing of messages based on cluster metrics.
  107.   router.type-mapping {
  108.     cluster-metrics-adaptive-pool  = "akka.cluster.metrics.AdaptiveLoadBalancingPool"
  109.     cluster-metrics-adaptive-group = "akka.cluster.metrics.AdaptiveLoadBalancingGroup"
  110.   }
  111. }

英文原文链接：Cluster Metrics Extension.