Repository: skiptests/astraea
Pull request #410 (2022-06-15 09:11:53)

Entropy needs to be calculated using raw metrics values
#393 這支PR有兩個改動
1.entropy使用metrics原始值進行計算
2.Ahp與entropy的合併方式改爲論文中提到的相乘的做法


wycccccc: 還需要增加一些測試，確認它有正常運作，完成後再麻煩學長review
wycccccc: 已添加測試，確保了entropy的部分有在正常運行，再麻煩學長review了。

Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
index eb82ef5f..e81dbd97 100644
--- a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
@@ -33,7 +33,8 @@ import org.astraea.app.metrics.collector.Fetcher;
 public class NeutralIntegratedCost implements HasBrokerCost {
   private final List<HasBrokerCost> metricsCost =
       List.of(new BrokerInputCost(), new BrokerOutputCost(), new CpuCost(), new MemoryCost());
-  private final Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
+  // Visible for test
+  Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
   private final AHPEmpowerment ahpEmpowerment = new AHPEmpowerment();
   private final WeightProvider weightProvider = WeightProvider.entropy(Normalizer.minMax(true));
 
@@ -49,58 +50,80 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           }
         });
 
-    metricsCost.forEach(
-        hasBrokerCost -> {
-          if (hasBrokerCost instanceof BrokerInputCost) {
-            hasBrokerCost
-                .brokerCost(clusterInfo)
-                .normalize(Normalizer.TScore())
-                .value()
-                .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
-          } else if (hasBrokerCost instanceof BrokerOutputCost) {
-            hasBrokerCost
-                .brokerCost(clusterInfo)
-                .normalize(Normalizer.TScore())
-                .value()
-                .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
-          } else if (hasBrokerCost instanceof CpuCost) {
-            hasBrokerCost
-                .brokerCost(clusterInfo)
-                .normalize(Normalizer.TScore())
-                .value()
-                .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
-          } else if (hasBrokerCost instanceof MemoryCost) {
-            hasBrokerCost
-                .brokerCost(clusterInfo)
-                .normalize(Normalizer.TScore())
-                .value()
-                .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
-          }
-        });
+    metricsCost.forEach(hasBrokerCost -> setBrokerMetrics(hasBrokerCost, clusterInfo));
 
+    var entropyEmpowerment = weight(weightProvider, brokersMetric);
+    var entropyEmpowermentSum =
+        entropyEmpowerment.entrySet().stream()
+            .mapToDouble(
+                entry -> entry.getValue() * ahpEmpowerment.empowerment().get(entry.getKey()))
+            .sum();
+    // The weight of each metric is obtained by combining Entropy and AHP.
     var integratedEmpowerment =
-        weight(weightProvider, brokersMetric).entrySet().stream()
+        entropyEmpowerment.entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        entry.getValue() * 0.5
-                            + ahpEmpowerment.empowerment().get(entry.getKey()) * 0.5));
+                        entry.getValue()
+                            * ahpEmpowerment.empowerment().get(entry.getKey())
+                            / entropyEmpowermentSum));
 
-    return () ->
+    var integratedScore =
         brokersMetric.entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        entry.getValue().inputScore
+                        entry.getValue().inputTScore
                                 * integratedEmpowerment.get(Metrics.inputThroughput.metricName)
-                            + entry.getValue().outputScore
+                            + entry.getValue().outputTScore
                                 * integratedEmpowerment.get(Metrics.outputThroughput.metricName)
-                            + entry.getValue().cpuScore
+                            + entry.getValue().cpuTScore
                                 * integratedEmpowerment.get(Metrics.cpu.metricName)
-                            + entry.getValue().memoryScore
+                            + entry.getValue().memoryTScore
                                 * integratedEmpowerment.get(Metrics.memory.metricName)));
+
+    return () -> integratedScore;
+  }
+
+  // Save the original value of each metric and the value of the metric after the TScore
+  // calculation.
+  // TODO Refactor
+  void setBrokerMetrics(HasBrokerCost hasBrokerCost, ClusterInfo clusterInfo) {
+    if (hasBrokerCost instanceof BrokerInputCost) {
+      var inputBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      inputBrokerCost
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
+      inputBrokerCost
+          .normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputTScore = value);
+    } else if (hasBrokerCost instanceof BrokerOutputCost) {
+      var outPutBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      outPutBrokerCost
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
+      outPutBrokerCost
+          .normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputTScore = value);
+    } else if (hasBrokerCost instanceof CpuCost) {
+      var CPUBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      CPUBrokerCost.value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
+      CPUBrokerCost.normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuTScore = value);
+    } else if (hasBrokerCost instanceof MemoryCost) {
+      var MemoryBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      MemoryBrokerCost.value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
+      MemoryBrokerCost.normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryTScore = value);
+    }
   }
 
   static Map<String, Double> weight(
@@ -146,6 +169,10 @@ public class NeutralIntegratedCost implements HasBrokerCost {
     double outputScore = 0.0;
     double cpuScore = 0.0;
     double memoryScore = 0.0;
+    double inputTScore = 0.0;
+    double outputTScore = 0.0;
+    double cpuTScore = 0.0;
+    double memoryTScore = 0.0;
 
     BrokerMetrics() {}
   }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
new file mode 100644
index 00000000..6820960f
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.cost;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class NeutralIntegratedCostTest {
+  @Test
+  void testSetBrokerMetrics() {
+    var neutralIntegratedCost = new NeutralIntegratedCost();
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    neutralIntegratedCost.brokersMetric =
+        new HashMap<>(
+            Map.of(
+                0,
+                new NeutralIntegratedCost.BrokerMetrics(),
+                1,
+                new NeutralIntegratedCost.BrokerMetrics(),
+                2,
+                new NeutralIntegratedCost.BrokerMetrics()));
+
+    HasBrokerCost input = Mockito.mock(BrokerInputCost.class);
+    HasBrokerCost output = Mockito.mock(BrokerOutputCost.class);
+    HasBrokerCost cpu = Mockito.mock(CpuCost.class);
+    HasBrokerCost memory = Mockito.mock(MemoryCost.class);
+
+    // Convert Map to BrokerCost, Map represents the score of each node metrics
+    var inputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 50.0, 1, 100.0, 2, 150.0)));
+    var outputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 500.0, 1, 1000.0, 2, 1500.0)));
+    var cpuBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.05, 1, 0.1, 2, 0.15)));
+    var memoryBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.1, 1, 0.3, 2, 0.5)));
+    Mockito.when(input.brokerCost(clusterInfo)).thenReturn(inputBrokerCost);
+    Mockito.when(output.brokerCost(clusterInfo)).thenReturn(outputBrokerCost);
+    Mockito.when(memory.brokerCost(clusterInfo)).thenReturn(memoryBrokerCost);
+    Mockito.when(cpu.brokerCost(clusterInfo)).thenReturn(cpuBrokerCost);
+    List<HasBrokerCost> metricsCost = List.of(input, output, cpu, memory);
+    metricsCost.forEach(
+        hasBrokerCost -> neutralIntegratedCost.setBrokerMetrics(hasBrokerCost, clusterInfo));
+
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).inputScore, 50.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).inputScore, 100.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).inputScore, 150.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).outputScore, 500.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).outputScore, 1000.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).outputScore, 1500.0);
+
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).cpuTScore, 0.38);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).cpuTScore, 0.5);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).cpuTScore, 0.62);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).memoryTScore, 0.38);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).memoryTScore, 0.5);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).memoryTScore, 0.62);
+  }
+
+  BrokerCost getBrokerCost(Map<Integer, Double> map) {
+    return () -> map;
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java b/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
index e5f0708f..8a6841b7 100644
--- a/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
+++ b/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
@@ -17,16 +17,86 @@
 package org.astraea.app.cost;
 
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
 public class WeightProviderTest {
+  @Test
+  void testPositiveEntropy() {
+    Normalizer normalizer = Normalizer.minMax(true);
+    var weightProvider = new WeightProvider.EntropyWeightProvider(normalizer);
+    var confusion =
+        IntStream.range(0, 1)
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    String::valueOf,
+                    ignored ->
+                        IntStream.range(0, 100)
+                            .mapToObj(i -> Math.random() * i)
+                            .collect(Collectors.toUnmodifiableList())));
+    confusion.put(
+        "1",
+        IntStream.range(0, 100)
+            .mapToObj(i -> 0.0 + i % 10 * 0.1)
+            .collect(Collectors.toUnmodifiableList()));
+    var entropy = weightProvider.weight(confusion);
+    // "0" range of 0 to 100 numbers * Math.random().
+    // "1" range of 0 to 1 numbers.The number is accurate to one decimal places.
+    // "0" is more confusing than "1".
+    Assertions.assertTrue(entropy.get("0") > entropy.get("1"));
+
+    var metrics1 = Math.random();
+    var metrics2 = Math.random();
+    var metrics3 = Math.random();
+    var metrics =
+        Map.of(
+            0,
+            List.of(metrics1, metrics2, metrics3),
+            1,
+            List.of(metrics1 * 100, metrics2 * 100, metrics3 * 100));
+    var sameEntropy = weightProvider.weight(metrics);
+    Assertions.assertEquals(
+        Math.round(sameEntropy.get(0) * 100) / 100, Math.round(sameEntropy.get(1) * 100) / 100);
+
+    var AABAndABB =
+        Map.of(
+            0,
+            List.of(metrics1 * 10, metrics2, metrics2),
+            1,
+            List.of(metrics1 * 10, metrics1 * 10, metrics2));
+    var ABEntropy = weightProvider.weight(AABAndABB);
+    Assertions.assertEquals(ABEntropy.get(0), ABEntropy.get(1));
+
+    var uniformDistribution =
+        IntStream.range(0, 1)
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    String::valueOf,
+                    ignored ->
+                        IntStream.range(0, 100)
+                            .mapToObj(i -> 1.0 + i % 10 * 0.1)
+                            .collect(Collectors.toUnmodifiableList())));
+    // Since entropy represents the degree of uncertainty, uniform distribution does not represent
+    // certainty in the concept of information entropy. On the contrary, because the uniform
+    // distribution means that it is possible at every point, the degree of uncertainty is greater
+    // and the weight given by the entropy method is also greater.
+    uniformDistribution.put(
+        "1",
+        IntStream.range(0, 100).mapToObj(i -> 0.0 + i).collect(Collectors.toUnmodifiableList()));
+
+    var uniformEntropy = weightProvider.weight(uniformDistribution);
+    Assertions.assertTrue(uniformEntropy.get("0") < uniformEntropy.get("1"));
+  }
 
   @ParameterizedTest
   @MethodSource("normalizers")
@@ -51,7 +121,6 @@ public class WeightProviderTest {
         weightProvider.entropies(raw).entrySet().stream()
             .sorted(Map.Entry.comparingByValue())
             .collect(Collectors.toUnmodifiableList());
-
     entropies.forEach(e -> Assertions.assertTrue(0 <= e.getValue() && e.getValue() <= 1));
 
     var weights =