Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add health page #1450

Merged
merged 2 commits into from
Jan 20, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -98,10 +98,10 @@ static TopicChecker skewPartition(double factor) {
var min =
clusterInfo.replicaLeaders(topic).stream()
.mapToLong(Replica::size)
.max();
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bug fix ...

.min();
return max.isPresent()
&& min.isPresent()
&& ((double) min.getAsLong() / max.getAsLong() >= factor);
&& ((double) min.getAsLong() / max.getAsLong() < factor);
})
.collect(Collectors.toSet()));
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -115,8 +115,16 @@ void testNoData() {

@Test
void testSkewPartition() {
var singlePartitionTopic = Utils.randomString();
var topic = Utils.randomString();
try (var admin = Admin.of(service.bootstrapServers())) {
admin
.creator()
.topic(singlePartitionTopic)
.numberOfPartitions(1)
.run()
.toCompletableFuture()
.join();
admin.creator().topic(topic).numberOfPartitions(2).run().toCompletableFuture().join();
Utils.sleep(Duration.ofSeconds(2));

Expand All @@ -128,15 +136,19 @@ void testSkewPartition() {
IntStream.range(0, 100)
.forEach(
ignored ->
producer
.send(
producer.send(
List.of(
Record.builder()
.topic(singlePartitionTopic)
.value("1".getBytes())
.partition(0)
.build(),
Record.builder()
.topic(topic)
.value("1".getBytes())
.partition(0)
.build())
.toCompletableFuture()
.join());
.build())));
producer.flush();
}

Assertions.assertEquals(
Expand Down
6 changes: 3 additions & 3 deletions gui/src/main/java/org/astraea/gui/Main.java
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,12 @@
import org.astraea.common.MapUtils;
import org.astraea.gui.pane.Slide;
import org.astraea.gui.tab.AboutNode;
import org.astraea.gui.tab.BalancerNode;
import org.astraea.gui.tab.BrokerNode;
import org.astraea.gui.tab.ClientNode;
import org.astraea.gui.tab.ConnectorNode;
import org.astraea.gui.tab.QuotaNode;
import org.astraea.gui.tab.SettingNode;
import org.astraea.gui.tab.health.HealthNode;
import org.astraea.gui.tab.topic.TopicNode;

/**
Expand Down Expand Up @@ -69,8 +69,8 @@ public void start(Stage stage) {
ConnectorNode.of(context),
"quota",
QuotaNode.of(context),
"balancer",
BalancerNode.of(context),
"health",
HealthNode.of(context),
"about",
AboutNode.of(context)))
.node(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.astraea.gui.tab;
package org.astraea.gui.tab.health;

import java.time.Duration;
import java.util.Arrays;
Expand Down Expand Up @@ -56,7 +56,7 @@
import org.astraea.gui.text.EditableText;
import org.astraea.gui.text.TextInput;

public class BalancerNode {
class BalancerNode {

static final AtomicReference<Balancer.Solution> LAST_PLAN = new AtomicReference<>();
static final String TOPIC_NAME_KEY = "topic";
Expand Down
145 changes: 145 additions & 0 deletions gui/src/main/java/org/astraea/gui/tab/health/HealthNode.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
/*
* 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.gui.tab.health;

import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javafx.geometry.Side;
import javafx.scene.Node;
import org.astraea.common.FutureUtils;
import org.astraea.common.MapUtils;
import org.astraea.common.admin.Admin;
import org.astraea.common.admin.Topic;
import org.astraea.common.admin.TopicChecker;
import org.astraea.common.admin.TopicConfigs;
import org.astraea.gui.Context;
import org.astraea.gui.pane.PaneBuilder;
import org.astraea.gui.pane.Slide;

public class HealthNode {

public static Node healthNode(Context context) {
return PaneBuilder.of()
.firstPart(
null,
List.of(),
"CHECK",
(argument, logger) ->
FutureUtils.combine(
badTopics(context.admin()),
unavailablePartitions(context.admin()),
(topics, partitions) -> {
var result = new LinkedHashMap<String, List<Map<String, Object>>>();
result.put("topic", topics);
result.put("partition", partitions);
return result;
}))
.build();
}

static CompletionStage<List<Map<String, Object>>> badTopics(Admin admin) {
return FutureUtils.combine(
admin.topicNames(List.of(TopicChecker.NO_DATA)),
admin.topicNames(List.of(TopicChecker.NO_CONSUMER_GROUP)),
admin.topicNames(
List.of(TopicChecker.noWriteAfter(Duration.ofHours(1), Duration.ofSeconds(1)))),
admin.topicNames(List.of(TopicChecker.skewPartition(0.5))),
(noDataTopics, noConsumerTopics, noWriteTopics, skewTopics) ->
Stream.of(
noDataTopics.stream(),
noConsumerTopics.stream(),
noWriteTopics.stream(),
skewTopics.stream())
.flatMap(s -> s)
.distinct()
.map(
name -> {
var r = new LinkedHashMap<String, Object>();
r.put("topic", name);
r.put("empty", noDataTopics.contains(name));
r.put("no consumer group", noConsumerTopics.contains(name));
r.put("no write (1 hour)", noWriteTopics.contains(name));
r.put("unbalanced", skewTopics.contains(name));
return r;
})
.collect(Collectors.toList()));
}

static CompletionStage<List<Map<String, Object>>> unavailablePartitions(Admin admin) {
return admin
.topicNames(true)
.thenCompose(
names ->
FutureUtils.combine(
admin.topics(names),
admin.partitions(names),
(topics, partitions) -> {
var minInSync =
topics.stream()
.collect(
Collectors.toMap(
Topic::name,
t ->
t.config()
.value(TopicConfigs.MIN_IN_SYNC_REPLICAS_CONFIG)
.map(Integer::parseInt)
.orElse(1)));

return partitions.stream()
.filter(
p ->
p.isr().size() < minInSync.getOrDefault(p.topic(), 1)
|| p.leader().isEmpty())
.map(
p -> {
var r = new LinkedHashMap<String, Object>();
r.put("topic", p.topic());
r.put("partition", p.partition());
r.put(
"leader",
p.leader().map(n -> String.valueOf(n.id())).orElse("null"));
r.put(
"in-sync replicas",
p.isr().stream()
.map(n -> String.valueOf(n.id()))
.collect(Collectors.joining(",")));
r.put(
TopicConfigs.MIN_IN_SYNC_REPLICAS_CONFIG,
minInSync.getOrDefault(p.topic(), 1));
r.put("readable", p.leader().isPresent());
r.put(
"writable",
p.leader().isPresent()
&& p.isr().size() >= minInSync.getOrDefault(p.topic(), 1));
return (Map<String, Object>) r;
})
.collect(Collectors.toList());
}));
}

public static Node of(Context context) {
return Slide.of(
Side.TOP,
MapUtils.of("basic", healthNode(context), "balancer", BalancerNode.of(context)))
.node();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@
import org.astraea.gui.text.EditableText;
import org.astraea.gui.text.TextInput;

public class PartitionNode {
class PartitionNode {

private static final String TOPIC_NAME_KEY = "topic";
private static final String PARTITION_KEY = "partition";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@
import org.astraea.gui.text.EditableText;
import org.astraea.gui.text.TextInput;

public class ReplicaNode {
class ReplicaNode {

static final String TOPIC_NAME_KEY = "topic";
static final String PARTITION_KEY = "partition";
Expand Down
81 changes: 0 additions & 81 deletions gui/src/main/java/org/astraea/gui/tab/topic/TopicNode.java
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import org.astraea.common.admin.NodeInfo;
import org.astraea.common.admin.Partition;
import org.astraea.common.admin.ProducerState;
import org.astraea.common.admin.Topic;
import org.astraea.common.admin.TopicConfigs;
import org.astraea.common.metrics.broker.HasRate;
import org.astraea.common.metrics.broker.ServerMetrics;
Expand Down Expand Up @@ -401,85 +400,6 @@ private static List<Map<String, Object>> basicResult(
.collect(Collectors.toList());
}

static List<Map<String, Object>> emptyTopics(List<Partition> partitions) {
return partitions.stream()
.collect(
Collectors.groupingBy(
Partition::topic,
Collectors.mapping(Partition::latestOffset, Collectors.reducing(0L, Long::sum))))
.entrySet()
.stream()
.filter(e -> e.getValue() <= 0)
.map(
e ->
(Map<String, Object>)
MapUtils.of("topic", (Object) e.getKey(), "records", e.getValue()))
.collect(Collectors.toList());
}

static List<Map<String, Object>> unavailablePartitions(
List<Topic> topics, List<Partition> partitions) {
var minInSync =
topics.stream()
.collect(
Collectors.toMap(
Topic::name,
t ->
t.config()
.value(TopicConfigs.MIN_IN_SYNC_REPLICAS_CONFIG)
.map(Integer::parseInt)
.orElse(1)));

return partitions.stream()
.filter(p -> p.isr().size() < minInSync.getOrDefault(p.topic(), 1) || p.leader().isEmpty())
.map(
p -> {
var r = new LinkedHashMap<String, Object>();
r.put("topic", p.topic());
r.put("partition", p.partition());
r.put("leader", p.leader().map(n -> String.valueOf(n.id())).orElse("null"));
r.put(
"in-sync replicas",
p.isr().stream()
.map(n -> String.valueOf(n.id()))
.collect(Collectors.joining(",")));
r.put(TopicConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, minInSync.getOrDefault(p.topic(), 1));
r.put("readable", p.leader().isPresent());
r.put(
"writable",
p.leader().isPresent() && p.isr().size() >= minInSync.getOrDefault(p.topic(), 1));
return (Map<String, Object>) r;
})
.collect(Collectors.toList());
}

public static Node healthNode(Context context) {
return PaneBuilder.of()
.firstPart(
null,
List.of(),
"CHECK",
(argument, logger) ->
context
.admin()
.topicNames(true)
.thenCompose(
names ->
FutureUtils.combine(
context.admin().topics(names),
context.admin().partitions(names),
(topics, partitions) -> {
var result =
new LinkedHashMap<String, List<Map<String, Object>>>();
result.put("empty topics", emptyTopics(partitions));
result.put(
"unavailable partitions",
unavailablePartitions(topics, partitions));
return result;
})))
.build();
}

public static Node of(Context context) {
return Slide.of(
Side.TOP,
Expand All @@ -489,7 +409,6 @@ public static Node of(Context context) {
"replica", ReplicaNode.of(context),
"config", configNode(context),
"metrics", metricsNode(context),
"health", healthNode(context),
"create", createNode(context)))
.node();
}
Expand Down
6 changes: 3 additions & 3 deletions gui/src/main/java/org/astraea/gui/table/TableViewer.java
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ private void refresh() {
allFilteredData =
allData.entrySet().stream()
.collect(
Collectors.toMap(
MapUtils.toLinkedHashMap(
Map.Entry::getKey,
e ->
e.getValue().stream()
Expand All @@ -174,9 +174,9 @@ private void refresh() {
.collect(Collectors.toUnmodifiableList())));

var allTables =
allData.keySet().stream()
allFilteredData.keySet().stream()
.collect(
Collectors.toMap(
MapUtils.toLinkedHashMap(
Function.identity(),
name -> {
var table = new TableView<Map<String, Object>>();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.astraea.gui.tab;
package org.astraea.gui.tab.health;

import java.time.Duration;
import java.util.List;
Expand Down
Loading