-
Notifications
You must be signed in to change notification settings - Fork 231
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(issue1120): support slow broker detect in controller (#1122)
Signed-off-by: Shichao Nie <[email protected]>
- Loading branch information
Showing
26 changed files
with
662 additions
and
35 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
30 changes: 30 additions & 0 deletions
30
core/src/main/java/kafka/autobalancer/common/types/metrics/AbnormalLatency.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ CO.,LTD. | ||
| * | ||
| * Use of this software is governed by the Business Source License | ||
| * included in the file BSL.md | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package kafka.autobalancer.common.types.metrics; | ||
|
|
||
| import kafka.autobalancer.model.BrokerUpdater; | ||
| import kafka.autobalancer.model.Snapshot; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public class AbnormalLatency extends AbstractSimpleAbnormalMetric { | ||
|
|
||
| public AbnormalLatency(float abnormalLatency) { | ||
| super(abnormalLatency); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isAbnormalToPeer(Snapshot self, Map<BrokerUpdater.Broker, Snapshot> peers) { | ||
| // TODO: compare with peer | ||
| return false; | ||
| } | ||
| } |
33 changes: 33 additions & 0 deletions
33
core/src/main/java/kafka/autobalancer/common/types/metrics/AbnormalMetric.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ CO.,LTD. | ||
| * | ||
| * Use of this software is governed by the Business Source License | ||
| * included in the file BSL.md | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package kafka.autobalancer.common.types.metrics; | ||
|
|
||
| import kafka.autobalancer.model.BrokerUpdater; | ||
| import kafka.autobalancer.model.Snapshot; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public interface AbnormalMetric { | ||
| default boolean isAbnormal(Snapshot self, Map<BrokerUpdater.Broker, Snapshot> peers) { | ||
| if (self == null) { | ||
| return false; | ||
| } | ||
| if (isSelfAbnormal(self)) { | ||
| return true; | ||
| } | ||
| return isAbnormalToPeer(self, peers); | ||
| } | ||
|
|
||
| boolean isSelfAbnormal(Snapshot self); | ||
|
|
||
| boolean isAbnormalToPeer(Snapshot self, Map<BrokerUpdater.Broker, Snapshot> peers); | ||
| } |
30 changes: 30 additions & 0 deletions
30
core/src/main/java/kafka/autobalancer/common/types/metrics/AbnormalQueueSize.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ CO.,LTD. | ||
| * | ||
| * Use of this software is governed by the Business Source License | ||
| * included in the file BSL.md | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package kafka.autobalancer.common.types.metrics; | ||
|
|
||
| import kafka.autobalancer.model.BrokerUpdater; | ||
| import kafka.autobalancer.model.Snapshot; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public class AbnormalQueueSize extends AbstractSimpleAbnormalMetric { | ||
|
|
||
| public AbnormalQueueSize(int abnormalQueueSizeThreshold) { | ||
| super(abnormalQueueSizeThreshold); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isAbnormalToPeer(Snapshot self, Map<BrokerUpdater.Broker, Snapshot> peers) { | ||
| return false; | ||
| } | ||
|
|
||
| } |
28 changes: 28 additions & 0 deletions
28
core/src/main/java/kafka/autobalancer/common/types/metrics/AbstractSimpleAbnormalMetric.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ CO.,LTD. | ||
| * | ||
| * Use of this software is governed by the Business Source License | ||
| * included in the file BSL.md | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package kafka.autobalancer.common.types.metrics; | ||
|
|
||
| import kafka.autobalancer.model.Snapshot; | ||
|
|
||
| public abstract class AbstractSimpleAbnormalMetric implements AbnormalMetric { | ||
| protected final double abnormalThreshold; | ||
|
|
||
| public AbstractSimpleAbnormalMetric(double abnormalThreshold) { | ||
| this.abnormalThreshold = abnormalThreshold; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isSelfAbnormal(Snapshot self) { | ||
| // TODO: compare with historical data | ||
| return self.getLatest() > abnormalThreshold; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
108 changes: 108 additions & 0 deletions
108
core/src/main/java/kafka/autobalancer/metricsreporter/metric/BrokerMetrics.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ CO.,LTD. | ||
| * | ||
| * Use of this software is governed by the Business Source License | ||
| * included in the file BSL.md | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package kafka.autobalancer.metricsreporter.metric; | ||
|
|
||
| import kafka.autobalancer.common.types.MetricTypes; | ||
| import kafka.autobalancer.common.types.RawMetricTypes; | ||
| import kafka.autobalancer.metricsreporter.exception.UnknownVersionException; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Map; | ||
|
|
||
| public class BrokerMetrics extends AutoBalancerMetrics { | ||
|
|
||
| public BrokerMetrics(long time, int brokerId, String brokerRack) { | ||
| super(time, brokerId, brokerRack); | ||
| } | ||
|
|
||
| public BrokerMetrics(long time, int brokerId, String brokerRack, Map<Byte, Double> metricTypeValueMap) { | ||
| super(time, brokerId, brokerRack, metricTypeValueMap); | ||
| } | ||
|
|
||
| public static BrokerMetrics fromBuffer(ByteBuffer buffer) throws UnknownVersionException { | ||
| byte version = buffer.get(); | ||
| if (version > METRIC_VERSION) { | ||
| throw new UnknownVersionException("Cannot deserialize the topic metrics for version " + version + ". " | ||
| + "Current version is " + METRIC_VERSION); | ||
| } | ||
| long time = buffer.getLong(); | ||
| int brokerId = buffer.getInt(); | ||
| int brokerRackLength = buffer.getInt(); | ||
| String brokerRack = ""; | ||
| if (brokerRackLength > 0) { | ||
| brokerRack = new String(buffer.array(), buffer.arrayOffset() + buffer.position(), brokerRackLength, StandardCharsets.UTF_8); | ||
| buffer.position(buffer.position() + brokerRackLength); | ||
| } | ||
| Map<Byte, Double> metricsMap = parseMetricsMap(buffer); | ||
| return new BrokerMetrics(time, brokerId, brokerRack, metricsMap); | ||
| } | ||
|
|
||
| @Override | ||
| public AutoBalancerMetrics put(byte type, double value) { | ||
| if (!RawMetricTypes.BROKER_METRICS.contains(type)) { | ||
| throw new IllegalArgumentException("Cannot put non broker metric type " + type + " into a partition metric."); | ||
| } | ||
| return super.put(type, value); | ||
| } | ||
|
|
||
| @Override | ||
| public String key() { | ||
| return Integer.toString(brokerId()); | ||
| } | ||
|
|
||
| @Override | ||
| public byte metricType() { | ||
| return MetricTypes.BROKER_METRIC; | ||
| } | ||
|
|
||
| /** | ||
| * The buffer capacity is calculated as follows: | ||
| * <ul> | ||
| * <li>(headerPos + {@link Byte#BYTES}) - version</li> | ||
| * <li>{@link Long#BYTES} - time</li> | ||
| * <li>{@link Integer#BYTES} - broker id</li> | ||
| * <li>{@link Integer#BYTES} - broker rack length</li> | ||
| * <li>brokerRack.length - broker rack</li> | ||
| * <li>body length - metric-value body</li> | ||
| * | ||
| * </ul> | ||
| * | ||
| * @param headerPos Header position | ||
| * @return Byte buffer of the partition metric. | ||
| */ | ||
| @Override | ||
| public ByteBuffer toBuffer(int headerPos) { | ||
| byte[] brokerRackBytes = brokerRack().getBytes(StandardCharsets.UTF_8); | ||
| ByteBuffer buffer = ByteBuffer.allocate(headerPos + Byte.BYTES | ||
| + Long.BYTES | ||
| + Integer.BYTES | ||
| + Integer.BYTES | ||
| + brokerRackBytes.length | ||
| + bodySize()); | ||
| buffer.position(headerPos); | ||
| buffer.put(METRIC_VERSION); | ||
| buffer.putLong(time()); | ||
| buffer.putInt(brokerId()); | ||
| buffer.putInt(brokerRackBytes.length); | ||
| if (brokerRackBytes.length > 0) { | ||
| buffer.put(brokerRackBytes); | ||
| } | ||
| buffer = writeBody(buffer); | ||
| return buffer; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return String.format("[BrokerMetrics,BrokerId=%d,Time=%d,Key:Value=%s]", brokerId(), time(), buildKVString()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.