diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java index 75c1ad93c2..372790a3c0 100644 --- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java +++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java @@ -28,7 +28,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -36,7 +35,6 @@ import org.astraea.common.Utils; import org.astraea.common.admin.Admin; import org.astraea.common.admin.ClusterInfo; -import org.astraea.common.admin.NodeInfo; import org.astraea.common.admin.Replica; import org.astraea.common.balancer.AlgorithmConfig; import org.astraea.common.balancer.Balancer; @@ -48,8 +46,6 @@ import org.astraea.common.cost.HasMoveCost; import org.astraea.common.cost.MigrationCost; import org.astraea.common.json.TypeRef; -import org.astraea.common.metrics.MBeanClient; -import org.astraea.common.metrics.collector.MetricSensor; import org.astraea.common.metrics.collector.MetricStore; class BalancerHandler implements Handler, AutoCloseable { @@ -61,35 +57,12 @@ class BalancerHandler implements Handler, AutoCloseable { new ConcurrentHashMap<>(); private final Map> planExecutions = new ConcurrentHashMap<>(); - private final Collection sensors = new ConcurrentLinkedQueue<>(); - private final MetricStore metricStore; - BalancerHandler(Admin admin, Function jmxPortMapper) { + BalancerHandler(Admin admin, MetricStore metricStore) { this.admin = admin; this.balancerConsole = BalancerConsole.create(admin); - Supplier>> clientSupplier = - () -> - admin - .brokers() - .thenApply( - brokers -> - brokers.stream() - .collect( - Collectors.toUnmodifiableMap( - NodeInfo::id, - b -> MBeanClient.jndi(b.host(), jmxPortMapper.apply(b.id()))))); - this.metricStore = - MetricStore.builder() - .beanExpiration(Duration.ofSeconds(90)) - .localReceiver(clientSupplier) - .sensorsSupplier( - () -> - sensors.stream() - .collect( - Collectors.toUnmodifiableMap( - Function.identity(), ignored -> (id, ee) -> {}))) - .build(); + this.metricStore = metricStore; } @Override @@ -117,9 +90,6 @@ public CompletionStage post(Channel channel) { Utils.construct(request.balancerClasspath, Balancer.class, request.balancerConfig); synchronized (this) { var taskId = UUID.randomUUID().toString(); - request.algorithmConfig.clusterCostFunction().metricSensor().ifPresent(sensors::add); - request.algorithmConfig.moveCostFunction().metricSensor().ifPresent(sensors::add); - var task = balancerConsole .launchRebalancePlanGeneration() @@ -262,7 +232,6 @@ static PostRequestWrapper parsePostRequestWrapper( static class BalancerPostRequest implements Request { String balancer = GreedyBalancer.class.getName(); - Map balancerConfig = Map.of(); Map costConfig = Map.of(); Duration timeout = Duration.ofSeconds(3); diff --git a/app/src/main/java/org/astraea/app/web/SensorHandler.java b/app/src/main/java/org/astraea/app/web/SensorHandler.java new file mode 100644 index 0000000000..425d2f15cc --- /dev/null +++ b/app/src/main/java/org/astraea/app/web/SensorHandler.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.web; + +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.stream.Collectors; +import org.astraea.app.web.WebService.Sensors; +import org.astraea.common.Configuration; +import org.astraea.common.Utils; +import org.astraea.common.cost.CostFunction; +import org.astraea.common.json.TypeRef; + +public class SensorHandler implements Handler { + + private final Sensors sensors; + private static final Set DEFAULT_COSTS = + Set.of( + "org.astraea.common.cost.ReplicaLeaderCost", + "org.astraea.common.cost.NetworkIngressCost"); + + SensorHandler(Sensors sensors) { + this.sensors = sensors; + } + + @Override + public CompletionStage get(Channel channel) { + var costs = + sensors.metricSensors().isEmpty() + ? DEFAULT_COSTS + : sensors.metricSensors().stream() + .map(x -> x.getClass().getName()) + .collect(Collectors.toSet()); + return CompletableFuture.completedFuture(new Response(costs)); + } + + @Override + public CompletionStage post(Channel channel) { + var metricSensorPostRequest = channel.request(TypeRef.of(MetricSensorPostRequest.class)); + var costs = costs(metricSensorPostRequest.costs); + sensors.clearSensors(); + costs.forEach(costFunction -> costFunction.metricSensor().ifPresent(sensors::addSensors)); + return CompletableFuture.completedFuture(new Response(metricSensorPostRequest.costs)); + } + + private static Set costs(Set costs) { + if (costs.isEmpty()) throw new IllegalArgumentException("costs is not specified"); + return Utils.costFunctions(costs, CostFunction.class, Configuration.EMPTY); + } + + static class MetricSensorPostRequest implements Request { + Set costs = DEFAULT_COSTS; + } + + static class Response implements org.astraea.app.web.Response { + final Set costs; + + Response(Set costs) { + this.costs = costs; + } + } +} diff --git a/app/src/main/java/org/astraea/app/web/WebService.java b/app/src/main/java/org/astraea/app/web/WebService.java index 3fa2b56330..e56316ce53 100644 --- a/app/src/main/java/org/astraea/app/web/WebService.java +++ b/app/src/main/java/org/astraea/app/web/WebService.java @@ -21,22 +21,61 @@ import com.sun.net.httpserver.HttpServer; import java.net.InetSocketAddress; import java.time.Duration; +import java.util.Collection; import java.util.Map; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import java.util.function.Supplier; +import java.util.stream.Collectors; import org.astraea.app.argument.DurationField; import org.astraea.app.argument.IntegerMapField; import org.astraea.app.argument.NonNegativeIntegerField; import org.astraea.common.Utils; import org.astraea.common.admin.Admin; +import org.astraea.common.admin.NodeInfo; +import org.astraea.common.metrics.MBeanClient; +import org.astraea.common.metrics.collector.MetricSensor; +import org.astraea.common.metrics.collector.MetricStore; public class WebService implements AutoCloseable { private final HttpServer server; private final Admin admin; + private final Sensors sensors = new Sensors(); - public WebService(Admin admin, int port, Function brokerIdToJmxPort) { + public WebService( + Admin admin, + int port, + Function brokerIdToJmxPort, + Duration beanExpiration) { this.admin = admin; + Supplier>> clientSupplier = + () -> + admin + .brokers() + .thenApply( + brokers -> + brokers.stream() + .collect( + Collectors.toUnmodifiableMap( + NodeInfo::id, + b -> + MBeanClient.jndi( + b.host(), brokerIdToJmxPort.apply(b.id()))))); + var metricStore = + MetricStore.builder() + .beanExpiration(beanExpiration) + .localReceiver(clientSupplier) + .sensorsSupplier( + () -> + sensors.metricSensors().stream() + .distinct() + .collect( + Collectors.toUnmodifiableMap( + Function.identity(), ignored -> (id, ee) -> {}))) + .build(); server = Utils.packException(() -> HttpServer.create(new InetSocketAddress(port), 0)); server.createContext("/topics", to(new TopicHandler(admin))); server.createContext("/groups", to(new GroupHandler(admin))); @@ -45,9 +84,10 @@ public WebService(Admin admin, int port, Function brokerIdToJm server.createContext("/quotas", to(new QuotaHandler(admin))); server.createContext("/transactions", to(new TransactionHandler(admin))); server.createContext("/beans", to(new BeanHandler(admin, brokerIdToJmxPort))); + server.createContext("/sensors", to(new SensorHandler(sensors))); server.createContext("/records", to(new RecordHandler(admin))); server.createContext("/reassignments", to(new ReassignmentHandler(admin))); - server.createContext("/balancer", to(new BalancerHandler(admin, brokerIdToJmxPort))); + server.createContext("/balancer", to(new BalancerHandler(admin, metricStore))); server.createContext("/throttles", to(new ThrottleHandler(admin))); server.start(); } @@ -64,7 +104,9 @@ public void close() { public static void main(String[] args) throws Exception { var arg = org.astraea.app.argument.Argument.parse(new Argument(), args); - try (var service = new WebService(Admin.of(arg.configs()), arg.port, arg::jmxPortMapping)) { + try (var service = + new WebService( + Admin.of(arg.configs()), arg.port, arg::jmxPortMapping, arg.beanExpiration)) { if (arg.ttl == null) { System.out.println("enter ctrl + c to terminate web service"); TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE); @@ -118,5 +160,32 @@ int jmxPortMapping(int brokerId) { validateWith = DurationField.class, converter = DurationField.class) Duration ttl = null; + + @Parameter( + names = {"--bean.expiration"}, + description = "Duration: the life of collected metrics", + validateWith = DurationField.class, + converter = DurationField.class) + Duration beanExpiration = Duration.ofHours(1); + } + + static class Sensors { + private final Collection sensors; + + Sensors() { + sensors = new ConcurrentLinkedQueue<>(); + } + + Collection metricSensors() { + return sensors; + } + + void clearSensors() { + sensors.clear(); + } + + void addSensors(MetricSensor metricSensor) { + sensors.add(metricSensor); + } } } diff --git a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java index 0aed00deaa..bb5e2d50b0 100644 --- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java +++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java @@ -47,6 +47,7 @@ import java.util.concurrent.atomic.LongAdder; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -76,7 +77,9 @@ import org.astraea.common.cost.ReplicaLeaderCost; import org.astraea.common.json.JsonConverter; import org.astraea.common.json.TypeRef; +import org.astraea.common.metrics.MBeanClient; import org.astraea.common.metrics.collector.MetricSensor; +import org.astraea.common.metrics.collector.MetricStore; import org.astraea.common.metrics.platform.HostMetrics; import org.astraea.common.metrics.platform.JvmMemory; import org.astraea.common.producer.Producer; @@ -92,7 +95,8 @@ public class BalancerHandlerTest { - private static final Service SERVICE = Service.builder().numberOfBrokers(3).build(); + private static final int numberOfBrokers = 3; + private static final Service SERVICE = Service.builder().numberOfBrokers(numberOfBrokers).build(); @AfterAll static void closeService() { @@ -115,8 +119,8 @@ static void closeService() { @Timeout(value = 60) void testReport() { var topics = createAndProduceTopic(3); - try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + try (var admin = Admin.of(SERVICE.bootstrapServers())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of())); // make sure all replicas have admin .clusterInfo(Set.copyOf(topics)) @@ -309,7 +313,7 @@ void testBestPlan() { void testMoveCost(String leaderLimit, String sizeLimit) { createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerHandler.BalancerPostRequest(); request.moveCosts = Set.of( @@ -356,7 +360,7 @@ void testMoveCost(String leaderLimit, String sizeLimit) { void testNoReport() { var topic = Utils.randomString(10); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { admin.creator().topic(topic).numberOfPartitions(1).run().toCompletableFuture().join(); Utils.sleep(Duration.ofSeconds(1)); var post = @@ -404,7 +408,7 @@ void testPut() { // arrange createAndProduceTopic(3, 10, (short) 2, false); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerHandler.BalancerPostRequest(); request.balancerConfig = Map.of("iteration", "100"); var progress = submitPlanGeneration(handler, request); @@ -433,7 +437,7 @@ void testPut() { void testBadPut() { createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { // no id offered Assertions.assertThrows( @@ -454,7 +458,7 @@ void testBadPut() { void testSubmitRebalancePlanThreadSafe() { var topic = Utils.randomString(); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { admin.creator().topic(topic).numberOfPartitions(30).run().toCompletableFuture().join(); Utils.sleep(Duration.ofSeconds(3)); admin @@ -502,7 +506,7 @@ void testSubmitRebalancePlanThreadSafe() { @Timeout(value = 60) void testRebalanceDetectOngoing() { try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { // create topic var theTopic = Utils.randomString(); admin.creator().topic(theTopic).numberOfPartitions(1).run().toCompletableFuture().join(); @@ -576,7 +580,7 @@ void testGenerationDetectOngoing() { .thenAnswer(invoke -> CompletableFuture.completedFuture(Set.of("A", "B", "C"))); Mockito.when(admin.clusterInfo(Mockito.any())) .thenAnswer(invoke -> CompletableFuture.completedFuture(clusterHasFuture)); - try (var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + try (var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var task0 = (BalancerHandler.PostPlanResponse) handler.post(defaultPostPlan).toCompletableFuture().join(); @@ -621,7 +625,7 @@ void testGenerationDetectOngoing() { void testPutSanityCheck() { var topic = createAndProduceTopic(1).iterator().next(); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerHandler.BalancerPostRequest(); request.balancerConfig = Map.of(BalancerConfigs.BALANCER_ALLOWED_TOPICS_REGEX, Pattern.quote(topic)); @@ -661,7 +665,7 @@ void testPutSanityCheck() { void testLookupRebalanceProgress() { createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var progress = submitPlanGeneration(handler, new BalancerPostRequest()); Assertions.assertEquals(Searched, progress.phase); @@ -715,7 +719,7 @@ void testLookupRebalanceProgress() { void testLookupBadExecutionProgress() { createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var post = Assertions.assertInstanceOf( BalancerHandler.PostPlanResponse.class, @@ -769,7 +773,7 @@ void testLookupBadExecutionProgress() { void testBadLookupRequest() { createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { Assertions.assertEquals( 404, handler.get(Channel.ofTarget("no such plan")).toCompletableFuture().join().code()); @@ -786,7 +790,7 @@ void testBadLookupRequest() { void testPutIdempotent() { var topics = createAndProduceTopic(3); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerHandler.BalancerPostRequest(); request.balancerConfig = Map.of( @@ -893,7 +897,7 @@ void testTimeout() { createAndProduceTopic(5); var costFunction = Collections.singleton(costWeight(TimeoutCost.class.getName(), 1)); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, (ignore) -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var channel = httpRequest(Map.of(TIMEOUT_KEY, "10", CLUSTER_COSTS_KEY, costFunction)); var post = (BalancerHandler.PostPlanResponse) handler.post(channel).toCompletableFuture().join(); @@ -911,8 +915,9 @@ void testTimeout() { @Test void testCostWithSensor() { var topics = createAndProduceTopic(3); + var function = List.of(costWeight(SensorAndCost.class.getName(), 1)); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, (ignore) -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, function))) { var invoked = new AtomicBoolean(); SensorAndCost.callback.set( (clusterBean) -> { @@ -927,7 +932,6 @@ void testCostWithSensor() { metrics.forEach(i -> Assertions.assertInstanceOf(JvmMemory.class, i)); invoked.set(true); }); - var function = List.of(costWeight(SensorAndCost.class.getName(), 1)); var request = new BalancerHandler.BalancerPostRequest(); request.timeout = Duration.ofSeconds(15); @@ -1073,7 +1077,7 @@ void testChangeOrder() { void testExecutorConfig() { var topic = createAndProduceTopic(1).iterator().next(); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerHandler.BalancerPostRequest(); request.balancerConfig = Map.of(BalancerConfigs.BALANCER_ALLOWED_TOPICS_REGEX, topic); var theProgress = submitPlanGeneration(handler, request); @@ -1107,7 +1111,7 @@ void testExecutorConfig() { void testBalancerConfig() { createAndProduceTopic(1); try (var admin = Admin.of(SERVICE.bootstrapServers()); - var handler = new BalancerHandler(admin, id -> SERVICE.jmxServiceURL().getPort())) { + var handler = new BalancerHandler(admin, metricStore(admin, List.of()))) { var request = new BalancerPostRequest(); request.balancer = SpyBalancer.class.getName(); request.balancerConfig = @@ -1350,4 +1354,35 @@ void testJsonToBalancerPostRequest() { Assertions.assertThrows(IllegalArgumentException.class, noCostRequest::clusterCost); } + + private MetricStore metricStore(Admin admin, List costWeights) { + Function brokerIdToJmxPort = (id) -> SERVICE.jmxServiceURL().getPort(); + Supplier>> clientSupplier = + () -> + admin + .brokers() + .thenApply( + brokers -> + brokers.stream() + .collect( + Collectors.toUnmodifiableMap( + NodeInfo::id, + b -> + MBeanClient.jndi( + b.host(), brokerIdToJmxPort.apply(b.id()))))); + var cw = costWeights.stream().map(x -> x.cost).collect(Collectors.toSet()); + var cf = Utils.costFunctions(cw, HasClusterCost.class, Configuration.EMPTY); + var metricSensors = cf.stream().map(c -> c.metricSensor().get()).collect(Collectors.toList()); + return MetricStore.builder() + .beanExpiration(Duration.ofMinutes(2)) + .localReceiver(clientSupplier) + .sensorsSupplier( + () -> + metricSensors.stream() + .distinct() + .collect( + Collectors.toUnmodifiableMap( + Function.identity(), ignored -> (id, ee) -> {}))) + .build(); + } } diff --git a/app/src/test/java/org/astraea/app/web/SensorHandlerTest.java b/app/src/test/java/org/astraea/app/web/SensorHandlerTest.java new file mode 100644 index 0000000000..bce5c14fdd --- /dev/null +++ b/app/src/test/java/org/astraea/app/web/SensorHandlerTest.java @@ -0,0 +1,62 @@ +/* + * 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.web; + +import java.time.Duration; +import org.astraea.common.Utils; +import org.astraea.common.admin.Admin; +import org.astraea.it.Service; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SensorHandlerTest { + private static final Service SERVICE = Service.builder().numberOfBrokers(3).build(); + + @Test + void testBeans() { + var topic = Utils.randomString(10); + try (var admin = Admin.of(SERVICE.bootstrapServers())) { + admin.creator().topic(topic).numberOfPartitions(10).run().toCompletableFuture().join(); + Utils.sleep(Duration.ofSeconds(2)); + var sensors = new WebService.Sensors(); + var defaultCostHandler = new SensorHandler(sensors); + var defaultCostResponse = + Assertions.assertInstanceOf( + SensorHandler.Response.class, + defaultCostHandler.get(Channel.EMPTY).toCompletableFuture().join()); + Assertions.assertEquals(2, defaultCostResponse.costs.size()); + + var changedCostResponse = + Assertions.assertInstanceOf( + SensorHandler.Response.class, + defaultCostHandler + .post( + Channel.builder() + .request("{\"costs\": [\"org.astraea.common.cost.ReplicaLeaderCost\"]}") + .build()) + .toCompletableFuture() + .join()); + Assertions.assertEquals(1, changedCostResponse.costs.size()); + + var changedCostGetResponse = + Assertions.assertInstanceOf( + SensorHandler.Response.class, + defaultCostHandler.get(Channel.EMPTY).toCompletableFuture().join()); + Assertions.assertEquals(1, changedCostGetResponse.costs.size()); + } + } +} diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java index d20ec12320..6aa023deff 100644 --- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java +++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java @@ -66,7 +66,10 @@ void testWithWebService() { try (var service = new WebService( - Admin.of(SERVICE.bootstrapServers()), 0, id -> SERVICE.jmxServiceURL().getPort())) { + Admin.of(SERVICE.bootstrapServers()), + 0, + id -> SERVICE.jmxServiceURL().getPort(), + Duration.ofMillis(5))) { Response response = HttpExecutor.builder() .build() diff --git a/app/src/test/java/org/astraea/app/web/WebServiceTest.java b/app/src/test/java/org/astraea/app/web/WebServiceTest.java index ac284c1713..d40700a0b8 100644 --- a/app/src/test/java/org/astraea/app/web/WebServiceTest.java +++ b/app/src/test/java/org/astraea/app/web/WebServiceTest.java @@ -16,6 +16,7 @@ */ package org.astraea.app.web; +import java.time.Duration; import java.util.concurrent.ThreadLocalRandom; import org.astraea.app.argument.Argument; import org.astraea.common.admin.Admin; @@ -39,7 +40,7 @@ void testArgument() { @Timeout(10) @Test void testClose() { - var web = new WebService(Mockito.mock(Admin.class), 0, id -> -1); + var web = new WebService(Mockito.mock(Admin.class), 0, id -> -1, Duration.ofMillis(5)); web.close(); } diff --git a/docs/web_server/README.md b/docs/web_server/README.md index 35052dba79..d311cc75f4 100644 --- a/docs/web_server/README.md +++ b/docs/web_server/README.md @@ -21,6 +21,7 @@ Astraea 建立了一套 Web Server 服務，使用者可以透過簡易好上手 - [/quotas](./web_api_quotas_chinese.md) - [/transactions](./web_api_transactions_chinese.md) - [/beans](./web_api_beans_chinese.md) +- [/metricSensors](./web_api_metricSensors_chinese.md) - [/reassignments](./web_api_reassignments_chinese.md) - [/records](./web_api_records_chinese.md) - [/balancer](./web_api_balancer_chinese.md) diff --git a/docs/web_server/web_api_metricSensors_chinese.md b/docs/web_server/web_api_metricSensors_chinese.md new file mode 100644 index 0000000000..9096dc2444 --- /dev/null +++ b/docs/web_server/web_api_metricSensors_chinese.md @@ -0,0 +1,60 @@ +/metricSensors +=== + +此api用來指定未來可能會想要使用的`Costfunction`之`MetricSensors`，主要功能如下 (相關討論請看[#1665](https://github.com/skiptests/astraea/pull/1665)) : + +- [指定MetricSensors](#指定-MetricSensors): 選擇要使用的`Costfunction`之`MetricSensor` +- [查詢已指定的 MetricSensors](#查詢已指定的-MetricSensors): 查看當前已指定之`MetricSensor`的`CostFunction` + +## 指定 MetricSensors +```shell +GET /sensors +``` + +cURL 範例 +```shell +curl -X POST http://localhost:8001/sensors \ + -H "Content-Type: application/json" \ + -d '{ + "costs": [ + "org.astraea.common.cost.ReplicaLeaderCost", + "org.astraea.common.cost.NetworkIngressCost", + "org.astraea.common.cost.NetworkEgressCost" + ] + }' +``` + +JSON Response 範例 +- `costs`: 目前已經註冊的`MetricSensors`之`Costfunction`，`MetricStore`會根據這些`MetricSensors`去撈取所需的metrics +```json +{ + "costs":[ + "org.astraea.common.cost.NetworkIngressCost", + "org.astraea.common.cost.ReplicaLeaderCost", + "org.astraea.common.cost.NetworkEgressCost" + ] +} +``` + +## 查詢已指定的 MetricSensors + +```shell +GET /metricSensors +``` + +cURL 範例 + +查詢已經註冊的`MetricSensors`之`Costfunction` +```shell +curl -X GET http://localhost:8001/sensors +``` + +JSON Response 範例 + ```json +{ + "costs":[ + "org.astraea.common.cost.NetworkCost$$Lambda$478/0x0000000840297840", + "org.astraea.common.cost.ReplicaLeaderCost$$Lambda$476/0x0000000840297040" + ] +} + ```