From d0f4a2548d8953a37cf4c1a44645d0457b423dc5 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Fri, 18 Sep 2026 18:31:56 +0000 Subject: [PATCH 1/4] feat(server): add a storage-aware GET /readiness endpoint Closes #3212. A Kubernetes readiness probe on /versions keeps a server in the Service while every graph request fails, because /versions answers 200 as long as the REST layer is up, even with no Store in the cluster (measured under #3132: ready=true with 0 Stores for 150+ s while every GET ended in a 500 after the 30 s request bound). GET /readiness answers 200 while this server can serve graph traffic and 503 otherwise, unauthenticated like /versions (whitelisted in both AuthenticationFilter and PathFilter, so an httpGet probe needs neither a credential nor a graphspace prefix), with a JSON body that carries no addresses. Ready means, from this server's own view: 1. its PD client answers (getActiveStores()), and 2. at least one Store that PD reports as active answers one cheap direct call (a node session and Table/EXISTS on the vertex table). The first Store that answers ends the probe: a rolling restart of the Stores never pulls the servers out of the Service, while zero Stores does. - hugegraph-hstore: HstoreStorageProbe (pure logic over an active-store lister and a store pinger, one shared time budget through an executor so a hung PD or Store yields "did not answer within N ms" instead of a hung probe) and a "storage_readiness" meta handler in HstoreStore - hugegraph-api: ReadinessAPI and StorageReadiness (first hstore graph is probed through HugeGraph.metadata(null, "storage_readiness", timeout), result cached for readiness.cache_ttl; servers without an hstore graph answer 200 with storage=embedded); the api module gains no dependency - ServerOptions: readiness.timeout (default 1000 ms) and readiness.cache_ttl (default 2000 ms) Tests: HstoreStorageProbeTest (9) and StorageReadinessTest (6, including the filter whitelists). Co-Authored-By: Claude Fable 5.1 --- .../api/filter/AuthenticationFilter.java | 1 + .../hugegraph/api/filter/PathFilter.java | 1 + .../hugegraph/api/profile/ReadinessAPI.java | 66 ++++ .../api/profile/StorageReadiness.java | 129 +++++++ .../hugegraph/config/ServerOptions.java | 22 ++ .../store/hstore/HstoreStorageProbe.java | 348 ++++++++++++++++++ .../backend/store/hstore/HstoreStore.java | 6 + .../store/hstore/HstoreStorageProbeTest.java | 264 +++++++++++++ .../apache/hugegraph/unit/UnitTestSuite.java | 2 + .../unit/core/StorageReadinessTest.java | 126 +++++++ 10 files changed, 965 insertions(+) create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java create mode 100644 hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java create mode 100644 hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/AuthenticationFilter.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/AuthenticationFilter.java index 96eb273be7..00b836e31d 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/AuthenticationFilter.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/AuthenticationFilter.java @@ -77,6 +77,7 @@ public class AuthenticationFilter implements ContainerRequestFilter, ContainerRe private static final AntPathMatcher MATCHER = new AntPathMatcher(); private static final Set FIXED_WHITE_API_SET = ImmutableSet.of( "versions", + "readiness", "openapi.json" ); /** Remove auth/login API from whitelist */ diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/PathFilter.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/PathFilter.java index 5e4dd5081c..016740d181 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/PathFilter.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/PathFilter.java @@ -57,6 +57,7 @@ public class PathFilter implements ContainerRequestFilter { "apis", "metrics", "versions", + "readiness", "health", "gremlin", "graphs/auth", diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java new file mode 100644 index 0000000000..af9bf9c8ce --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java @@ -0,0 +1,66 @@ +/* + * 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.apache.hugegraph.api.profile; + +import java.util.Map; + +import org.apache.hugegraph.api.API; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.util.JsonUtil; + +import com.codahale.metrics.annotation.Timed; + +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.annotation.security.PermitAll; +import jakarta.inject.Singleton; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.core.Context; +import jakarta.ws.rs.core.Response; + +/** + * Storage-aware readiness for Kubernetes and load balancers: 200 while this + * server can serve graph traffic, 503 while PD or every Store is unreachable + * from it. Unauthenticated, like /versions, so that an httpGet probe needs + * no credential; the body carries no addresses. + */ +@Path("readiness") +@Singleton +@Tag(name = "ReadinessAPI") +public class ReadinessAPI extends API { + + @GET + @Timed + @Produces(APPLICATION_JSON_WITH_CHARSET) + @PermitAll + public Response get(@Context GraphManager manager, @Context HugeConfig conf) { + Map body = StorageReadiness.check( + manager, conf.get(ServerOptions.READINESS_TIMEOUT), + conf.get(ServerOptions.READINESS_CACHE_TTL)); + Response.Status status = StorageReadiness.isReady(body) ? + Response.Status.OK : + Response.Status.SERVICE_UNAVAILABLE; + return Response.status(status) + .type(APPLICATION_JSON_WITH_CHARSET) + .entity(JsonUtil.toJson(body)) + .build(); + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java new file mode 100644 index 0000000000..415202d7b8 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java @@ -0,0 +1,129 @@ +/* + * 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.apache.hugegraph.api.profile; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.auth.HugeGraphAuthProxy; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; + +/** + * Whether this server can serve graph traffic, for a readiness probe. + * Graphs on an embedded backend are ready as soon as the REST layer answers. + * Graphs on hstore are probed through the backend's "storage_readiness" + * metadata (PD answers and one active Store answers). The storage is shared + * by every hstore graph of the process, so one graph is probed and the + * result is reused for a short TTL to keep repeated probes cheap. + */ +public final class StorageReadiness { + + public static final String STORAGE_READINESS_META = "storage_readiness"; + public static final String BACKEND_HSTORE = "hstore"; + + private static final Logger LOG = Log.logger(StorageReadiness.class); + + private static volatile Map lastResult; + private static volatile long lastCheckedAt; + + private StorageReadiness() { + } + + /** One storage probe with a time budget in ms. */ + public interface Probe { + + Map probe(long timeoutMs) throws Exception; + } + + public static Map check(GraphManager manager, + long timeoutMs, long cacheTtlMs) { + // The graphs are auth proxies and the probe request carries no user, + // so look the graph up and probe it as the internal admin, the way + // other internal paths do; the result carries no data or addresses + List> holder = new ArrayList<>(1); + HugeGraphAuthProxy.runAsAdmin(() -> { + HugeGraph graph = firstHstoreGraph(manager); + if (graph == null) { + Map body = new LinkedHashMap<>(); + body.put("ready", true); + body.put("storage", "embedded"); + body.put("reason", "no graph on a remote storage"); + holder.add(body); + return; + } + holder.add(check(t -> graph.metadata(null, STORAGE_READINESS_META, t), + timeoutMs, cacheTtlMs)); + }); + return holder.get(0); + } + + public static synchronized Map check(Probe probe, long timeoutMs, + long cacheTtlMs) { + long now = System.currentTimeMillis(); + Map cached = lastResult; + if (cached != null && now - lastCheckedAt < cacheTtlMs) { + Map body = new LinkedHashMap<>(cached); + body.put("cached", true); + return body; + } + Map body = new LinkedHashMap<>(); + body.put("ready", false); + body.put("storage", BACKEND_HSTORE); + try { + Map result = probe.probe(timeoutMs); + body.putAll(result); + } catch (Throwable e) { + LOG.debug("Storage readiness probe failed", e); + body.put("ready", false); + body.put("reason", "probe failed: " + e.getClass().getSimpleName() + + (e.getMessage() == null ? "" : ": " + e.getMessage())); + } + body.put("cached", false); + lastResult = body; + lastCheckedAt = System.currentTimeMillis(); + return new LinkedHashMap<>(body); + } + + public static boolean isReady(Map body) { + return Boolean.TRUE.equals(body.get("ready")); + } + + public static synchronized void resetCache() { + lastResult = null; + lastCheckedAt = 0L; + } + + private static HugeGraph firstHstoreGraph(GraphManager manager) { + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph != null && BACKEND_HSTORE.equals(graph.backend())) { + return graph; + } + } catch (Throwable e) { + LOG.debug("Skip graph {} while looking for a remote storage", name, e); + } + } + return null; + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java index d43b095b76..46d1c8c48b 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java @@ -211,6 +211,28 @@ public class ServerOptions extends OptionHolder { 300 ); + public static final ConfigOption READINESS_TIMEOUT = + new ConfigOption<>( + "readiness.timeout", + "The whole time budget in ms of one GET /readiness probe " + + "(PD call plus one cheap call to an active Store); a PD " + + "or Store that does not answer within it makes the " + + "server report not ready.", + rangeInt(100, 60000), + 1000 + ); + + public static final ConfigOption READINESS_CACHE_TTL = + new ConfigOption<>( + "readiness.cache_ttl", + "How many ms a GET /readiness result is reused before " + + "the storage is probed again, so that several probes " + + "(Kubernetes, load balancers) cost one PD and one Store " + + "call per interval; 0 probes on every request.", + rangeInt(0, 60000), + 2000 + ); + public static final ConfigOption SERVER_USE_K8S = new ConfigOption<>( "server.use_k8s", diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java new file mode 100644 index 0000000000..f1d77ee349 --- /dev/null +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java @@ -0,0 +1,348 @@ +/* + * 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.apache.hugegraph.backend.store.hstore; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.store.grpc.state.HgStoreStateGrpc; +import org.apache.hugegraph.store.grpc.state.SubStateReq; +import org.apache.hugegraph.util.E; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; + +/** + * Storage-aware readiness of this server, from this server's point of view: + * at least one Store answers a direct, local, read-only gRPC call + * (HgStoreState.getScanState, which reads the node's own scan-pool stats and + * never touches raft). The Store list comes from PD, but PD is refreshed in + * the background and the last known list is used right away, so a PD that is + * slow, restarting or down does not change the readiness of a server whose + * Stores still answer. Every wait is bounded by one shared time budget, so a + * hung PD or Store turns into "not ready" instead of a hung probe. The result + * carries no addresses, since it is served without authentication. + */ +public final class HstoreStorageProbe { + + public static final String META_STORAGE_READINESS = "storage_readiness"; + + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( + new ThreadFactory() { + private final AtomicInteger seq = new AtomicInteger(); + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "storage-readiness-" + this.seq.incrementAndGet()); + t.setDaemon(true); + return t; + } + }); + + private static final KnownStores KNOWN = new KnownStores(); + private static final Map CHANNELS = new ConcurrentHashMap<>(); + + private HstoreStorageProbe() { + } + + /** The active stores as PD sees them. */ + public interface StoreLister { + + List activeStores() throws Exception; + } + + /** One cheap call to one store; returning (any value) means it answered. */ + public interface StorePinger { + + void ping(Metapb.Store store, long timeoutMs) throws Exception; + } + + /** The last store list PD answered with, shared by consecutive probes. */ + public static final class KnownStores { + + private volatile List stores = Collections.emptyList(); + private volatile long at; + private volatile Boolean pdOk; + private volatile long pdAt; + + public List stores() { + return this.stores; + } + + public long ageMs() { + return this.at == 0L ? -1L : System.currentTimeMillis() - this.at; + } + + /** Outcome of the last finished PD refresh, null before the first one. */ + public Boolean pdOk() { + return this.pdOk; + } + + public long pdAgeMs() { + return this.pdAt == 0L ? -1L : System.currentTimeMillis() - this.pdAt; + } + + public void update(List stores) { + this.pdOk = true; + this.pdAt = System.currentTimeMillis(); + if (stores != null && !stores.isEmpty()) { + this.stores = Collections.unmodifiableList(new ArrayList<>(stores)); + this.at = this.pdAt; + } + } + + public void pdFailed() { + this.pdOk = false; + this.pdAt = System.currentTimeMillis(); + } + } + + public static final class Result { + + private final boolean ready; + private final String reason; + private final int activeStores; + private final Long answeredStore; + private final Boolean pdReachable; + private final long pdAgeMs; + private final long storesAgeMs; + private final long storeMillis; + + Result(boolean ready, String reason, int activeStores, Long answeredStore, + Boolean pdReachable, long pdAgeMs, long storesAgeMs, long storeMillis) { + this.ready = ready; + this.reason = reason; + this.activeStores = activeStores; + this.answeredStore = answeredStore; + this.pdReachable = pdReachable; + this.pdAgeMs = pdAgeMs; + this.storesAgeMs = storesAgeMs; + this.storeMillis = storeMillis; + } + + public boolean ready() { + return this.ready; + } + + public String reason() { + return this.reason; + } + + public int activeStores() { + return this.activeStores; + } + + public Long answeredStore() { + return this.answeredStore; + } + + public Boolean pdReachable() { + return this.pdReachable; + } + + public Map toMap() { + Map map = new LinkedHashMap<>(); + map.put("ready", this.ready); + map.put("reason", this.reason); + map.put("active_stores", this.activeStores); + map.put("answered_store", this.answeredStore); + map.put("pd_reachable", this.pdReachable); + map.put("pd_checked_age_ms", this.pdAgeMs); + map.put("stores_age_ms", this.storesAgeMs); + map.put("store_millis", this.storeMillis); + return map; + } + } + + /** + * Probe through the process-wide PD client and this probe's own plaintext + * channels to the stores (the store gRPC server takes no credentials). + * + * @param graphName the store-side graph name, kept for the meta handler + * @param timeoutMs the whole budget for PD plus stores + */ + public static Map probe(String graphName, long timeoutMs) { + PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); + if (pd == null) { + return new Result(false, "pd client not initialised", 0, null, + false, -1L, -1L, 0L).toMap(); + } + return probe(KNOWN, pd::getActiveStores, HstoreStorageProbe::pingScanState, + timeoutMs, EXECUTOR).toMap(); + } + + private static void pingScanState(Metapb.Store store, long timeoutMs) { + ManagedChannel channel = CHANNELS.computeIfAbsent(store.getAddress(), address -> { + return ManagedChannelBuilder.forTarget(address).usePlaintext().build(); + }); + HgStoreStateGrpc.newBlockingStub(channel) + .withDeadlineAfter(timeoutMs, TimeUnit.MILLISECONDS) + .getScanState(SubStateReq.getDefaultInstance()); + } + + public static Result probe(KnownStores known, StoreLister lister, StorePinger pinger, + long timeoutMs, ExecutorService executor) { + E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); + long deadline = System.currentTimeMillis() + timeoutMs; + + // Refresh the store list from PD in the background; whatever PD + // answers lands in `known` for this or the next probe + CompletableFuture> refresh = new CompletableFuture<>(); + executor.execute(() -> { + try { + List stores = lister.activeStores(); + known.update(stores); + refresh.complete(stores == null ? Collections.emptyList() : stores); + } catch (Throwable e) { + known.pdFailed(); + refresh.completeExceptionally(e); + } + }); + + List stores = known.stores(); + Boolean pdReachable = null; + if (stores.isEmpty()) { + // Nothing known yet (first probe after start): PD is the only source + try { + stores = await(refresh, deadline); + pdReachable = true; + } catch (TimeoutException e) { + return new Result(false, "no store list known and pd did not answer within " + + timeoutMs + " ms", 0, null, false, -1L, -1L, 0L); + } catch (Exception e) { + return new Result(false, "no store list known and pd failed: " + message(e), + 0, null, false, known.pdAgeMs(), -1L, 0L); + } + if (stores.isEmpty()) { + return new Result(false, "no active store registered in pd", + 0, null, true, known.pdAgeMs(), known.ageMs(), 0L); + } + } + + long storeStart = System.currentTimeMillis(); + // Ping every known store at once and take the first answer: a store + // whose connection hangs (a pod that just went away) must not eat the + // budget of the stores that are fine, or a rolling restart would + // flap the readiness of every server + CompletionService pings = new ExecutorCompletionService<>(executor); + List> futures = new ArrayList<>(stores.size()); + for (Metapb.Store store : stores) { + futures.add(pings.submit(() -> { + pinger.ping(store, Math.max(1L, deadline - System.currentTimeMillis())); + return store; + })); + } + List failures = new ArrayList<>(); + Result result = null; + try { + for (int done = 0; done < stores.size() && result == null; done++) { + long remaining = deadline - System.currentTimeMillis(); + Future first; + try { + first = remaining > 0 ? + pings.poll(remaining, TimeUnit.MILLISECONDS) : + pings.poll(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + failures.add("interrupted"); + break; + } + if (first == null) { + failures.add((stores.size() - done) + " store(s) did not answer within " + + timeoutMs + " ms"); + break; + } + try { + Metapb.Store store = first.get(); + result = new Result(true, "ok", stores.size(), store.getId(), + pdState(refresh, known, pdReachable), known.pdAgeMs(), + known.ageMs(), elapsed(storeStart)); + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + failures.add("a store failed: " + message(cause)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + failures.add("interrupted"); + break; + } + } + } finally { + for (Future f : futures) { + f.cancel(true); + } + } + if (result != null) { + return result; + } + return new Result(false, "none of " + stores.size() + " known store(s) answered: " + + String.join("; ", failures), + stores.size(), null, pdState(refresh, known, pdReachable), + known.pdAgeMs(), known.ageMs(), elapsed(storeStart)); + } + + /** + * The outcome of this probe's PD refresh when it already finished, else + * the outcome of the last finished one (null before any finished). + */ + private static Boolean pdState(CompletableFuture refresh, KnownStores known, + Boolean awaited) { + if (awaited != null) { + return awaited; + } + if (refresh.isDone()) { + return !refresh.isCompletedExceptionally(); + } + return known.pdOk(); + } + + private static T await(Future future, long deadline) throws Exception { + long remaining = Math.max(1L, deadline - System.currentTimeMillis()); + try { + return future.get(remaining, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + throw cause instanceof Exception ? (Exception) cause : new RuntimeException(cause); + } + } + + private static long elapsed(long since) { + return System.currentTimeMillis() - since; + } + + private static String message(Throwable e) { + String msg = e.getMessage(); + return e.getClass().getSimpleName() + (msg == null ? "" : ": " + msg); + } +} diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java index 6439096674..0797da85ee 100644 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java @@ -114,6 +114,12 @@ private void registerMetaHandlers() { HstoreMetrics metrics = new HstoreMetrics(dbsGet.get(), session); return metrics.metrics(); }); + this.registerMetaHandler(HstoreStorageProbe.META_STORAGE_READINESS, (session, meta, args) -> { + E.checkArgument(args.length == 1 && args[0] instanceof Number, + "Expect the timeout in ms as the only argument"); + return HstoreStorageProbe.probe(this.namespace + "/" + this.store, + ((Number) args[0]).longValue()); + }); this.registerMetaHandler("mode", (session, meta, args) -> { E.checkArgument(args.length == 1, "The args count of %s must be 1", meta); diff --git a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java new file mode 100644 index 0000000000..473f253fe5 --- /dev/null +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java @@ -0,0 +1,264 @@ +/* + * 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.apache.hugegraph.backend.store.hstore; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.KnownStores; +import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.Result; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Test; + +public class HstoreStorageProbeTest { + + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool(); + private static final long BUDGET = 500L; + + @AfterClass + public static void shutdown() { + EXECUTOR.shutdownNow(); + } + + private static Metapb.Store store(long id) { + return Metapb.Store.newBuilder().setId(id).setAddress("10.0.0." + id + ":8500") + .setState(Metapb.StoreState.Up).build(); + } + + private static List stores(long... ids) { + return Arrays.stream(ids).mapToObj(HstoreStorageProbeTest::store) + .collect(Collectors.toList()); + } + + private static KnownStores knowing(long... ids) { + KnownStores known = new KnownStores(); + known.update(stores(ids)); + return known; + } + + private static final HstoreStorageProbe.StorePinger ANSWERS = (store, timeout) -> { + }; + + @Test + public void testReadyWhenPdAndOneStoreAnswer() { + Result r = HstoreStorageProbe.probe(new KnownStores(), () -> stores(1L, 2L, 3L), + ANSWERS, BUDGET, EXECUTOR); + Assert.assertTrue(r.reason(), r.ready()); + Assert.assertEquals(3, r.activeStores()); + Assert.assertNotNull(r.answeredStore()); + Assert.assertEquals(Boolean.TRUE, r.pdReachable()); + Assert.assertEquals("ok", r.reason()); + } + + @Test + public void testFirstAnsweringStoreWinsAfterFailures() { + AtomicInteger pings = new AtomicInteger(); + HstoreStorageProbe.StorePinger onlyThird = (store, timeout) -> { + pings.incrementAndGet(); + if (store.getId() != 3L) { + throw new IllegalStateException("UNAVAILABLE"); + } + }; + Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + onlyThird, BUDGET, EXECUTOR); + Assert.assertTrue(r.reason(), r.ready()); + Assert.assertEquals(Long.valueOf(3L), r.answeredStore()); + // the pings run in parallel; the failing ones may or may not have run + Assert.assertTrue(pings.get() >= 1 && pings.get() <= 3); + } + + /** + * The store whose pod just went away hangs until its deadline; it must + * not eat the budget of a store that answers (otherwise a rolling restart + * would flap every server's readiness). + */ + @Test + public void testHungStoreDoesNotHideAnAnsweringOne() { + HstoreStorageProbe.StorePinger onlySecondAnswers = (store, timeout) -> { + if (store.getId() != 2L) { + Thread.sleep(10_000L); + } + }; + long start = System.currentTimeMillis(); + Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + onlySecondAnswers, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + Assert.assertTrue(r.reason(), r.ready()); + Assert.assertEquals(Long.valueOf(2L), r.answeredStore()); + Assert.assertTrue("took " + took, took < BUDGET); + } + + @Test + public void testFirstProbeWithoutKnownStoresNeedsPd() { + Result r = HstoreStorageProbe.probe(new KnownStores(), () -> { + throw new IllegalStateException("UNAVAILABLE: io exception"); + }, ANSWERS, BUDGET, EXECUTOR); + Assert.assertFalse(r.ready()); + Assert.assertTrue(r.reason(), r.reason().startsWith( + "no store list known and pd failed: IllegalStateException")); + Assert.assertEquals(Boolean.FALSE, r.pdReachable()); + } + + @Test + public void testFirstProbeWithHungPdStaysWithinBudget() { + long start = System.currentTimeMillis(); + Result r = HstoreStorageProbe.probe(new KnownStores(), () -> { + Thread.sleep(10_000L); + return stores(1L); + }, ANSWERS, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + Assert.assertFalse(r.ready()); + Assert.assertTrue(r.reason(), r.reason().contains("pd did not answer within")); + Assert.assertTrue("took " + took, took < BUDGET * 4); + } + + /** + * PD down, restarting or slow must not change the readiness of a server + * whose stores still answer: the last known list is used right away. + */ + @Test + public void testKnownStoresKeepTheServerReadyWhilePdIsDown() { + Result r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { + throw new IllegalStateException("PD unreachable"); + }, ANSWERS, BUDGET, EXECUTOR); + Assert.assertTrue(r.reason(), r.ready()); + Assert.assertEquals(2, r.activeStores()); + } + + @Test + public void testHungPdDoesNotDelayAProbeWithKnownStores() { + long start = System.currentTimeMillis(); + Result r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { + Thread.sleep(10_000L); + return stores(1L, 2L); + }, ANSWERS, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + Assert.assertTrue(r.reason(), r.ready()); + // the refresh is still pending, so the outcome of the last finished + // one (the seed) is reported + Assert.assertEquals(Boolean.TRUE, r.pdReachable()); + Assert.assertTrue("took " + took, took < BUDGET); + } + + @Test + public void testLastPdOutcomeIsReportedWhileTheRefreshIsPending() throws Exception { + KnownStores known = knowing(1L); + HstoreStorageProbe.probe(known, () -> { + throw new IllegalStateException("PD unreachable"); + }, ANSWERS, BUDGET, EXECUTOR); + for (int i = 0; i < 50 && known.pdOk() == null; i++) { + Thread.sleep(20L); + } + Assert.assertEquals(Boolean.FALSE, known.pdOk()); + Result r = HstoreStorageProbe.probe(known, () -> { + Thread.sleep(10_000L); + return stores(1L); + }, ANSWERS, BUDGET, EXECUTOR); + Assert.assertTrue(r.ready()); + Assert.assertEquals(Boolean.FALSE, r.pdReachable()); + Assert.assertTrue(r.toMap().containsKey("pd_checked_age_ms")); + } + + @Test + public void testPdAnswerUpdatesTheKnownStoresForTheNextProbe() throws Exception { + KnownStores known = knowing(1L); + HstoreStorageProbe.probe(known, () -> stores(1L, 2L, 3L), ANSWERS, BUDGET, EXECUTOR); + for (int i = 0; i < 50 && known.stores().size() != 3; i++) { + Thread.sleep(20L); + } + Assert.assertEquals(3, known.stores().size()); + Assert.assertTrue(known.ageMs() >= 0L); + } + + @Test + public void testEmptyPdAnswerIsNotReadyAndKeepsTheOldList() { + KnownStores fresh = new KnownStores(); + Result r = HstoreStorageProbe.probe(fresh, Collections::emptyList, ANSWERS, + BUDGET, EXECUTOR); + Assert.assertFalse(r.ready()); + Assert.assertEquals("no active store registered in pd", r.reason()); + KnownStores known = knowing(1L); + HstoreStorageProbe.probe(known, Collections::emptyList, ANSWERS, BUDGET, EXECUTOR); + Assert.assertEquals(1, known.stores().size()); + } + + @Test + public void testNotReadyWhenEveryStoreFails() { + HstoreStorageProbe.StorePinger refused = (store, timeout) -> { + throw new IllegalStateException("connection refused"); + }; + Result r = HstoreStorageProbe.probe(knowing(7L, 8L), () -> stores(7L, 8L), + refused, BUDGET, EXECUTOR); + Assert.assertFalse(r.ready()); + Assert.assertTrue(r.reason(), r.reason().startsWith("none of 2 known store(s) answered")); + Assert.assertTrue(r.reason(), r.reason().contains("connection refused")); + Assert.assertNull(r.answeredStore()); + } + + @Test + public void testHungStoresStayWithinTheBudget() { + HstoreStorageProbe.StorePinger hung = (store, timeout) -> { + Thread.sleep(10_000L); + }; + long start = System.currentTimeMillis(); + Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + hung, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + Assert.assertFalse(r.ready()); + Assert.assertTrue(r.reason(), r.reason().contains("did not answer within")); + Assert.assertTrue("took " + took, took < BUDGET * 4); + } + + @Test + public void testPingGetsTheRemainingBudget() { + List budgets = Collections.synchronizedList(new java.util.ArrayList<>()); + HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), (store, timeout) -> { + budgets.add(timeout); + }, BUDGET, EXECUTOR); + Assert.assertEquals(1, budgets.size()); + Assert.assertTrue(budgets.get(0) > 0L && budgets.get(0) <= BUDGET); + } + + @Test + public void testMapCarriesNoAddresses() { + Map map = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), + ANSWERS, BUDGET, EXECUTOR).toMap(); + Assert.assertEquals(true, map.get("ready")); + Assert.assertEquals(1, map.get("active_stores")); + Assert.assertEquals(1L, map.get("answered_store")); + Assert.assertTrue(map.containsKey("pd_reachable")); + Assert.assertTrue(map.containsKey("stores_age_ms")); + Assert.assertFalse(map.toString().contains("10.0.0.")); + } + + @Test + public void testRejectsNonPositiveBudget() { + Assert.assertThrows(IllegalArgumentException.class, () -> { + HstoreStorageProbe.probe(new KnownStores(), Collections::emptyList, ANSWERS, + 0L, EXECUTOR); + }); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index 0a9c621f90..e23d3f505b 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -52,6 +52,7 @@ import org.apache.hugegraph.unit.core.DataTypeTest; import org.apache.hugegraph.unit.core.GraphSpaceInfoLocaleTest; import org.apache.hugegraph.unit.core.GraphManagerStoresWaitTest; +import org.apache.hugegraph.unit.core.StorageReadinessTest; import org.apache.hugegraph.unit.core.DirectionsTest; import org.apache.hugegraph.unit.core.ExceptionTest; import org.apache.hugegraph.unit.core.GraphManagerAdminInitTest; @@ -141,6 +142,7 @@ DataTypeTest.class, GraphSpaceInfoLocaleTest.class, GraphManagerStoresWaitTest.class, + StorageReadinessTest.class, DirectionsTest.class, SerialEnumTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java new file mode 100644 index 0000000000..d4bda1dd50 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java @@ -0,0 +1,126 @@ +/* + * 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.apache.hugegraph.unit.core; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.api.filter.AuthenticationFilter; +import org.apache.hugegraph.api.filter.PathFilter; +import org.apache.hugegraph.api.profile.StorageReadiness; +import org.apache.hugegraph.testutil.Assert; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import jakarta.ws.rs.container.ContainerRequestContext; +import jakarta.ws.rs.core.UriInfo; + +public class StorageReadinessTest { + + @Before + @After + public void reset() { + StorageReadiness.resetCache(); + } + + private static Map result(boolean ready, String reason) { + Map map = new LinkedHashMap<>(); + map.put("ready", ready); + map.put("reason", reason); + map.put("active_stores", 3); + return map; + } + + @Test + public void testReadyBodyAndCacheReuse() { + AtomicInteger probes = new AtomicInteger(); + StorageReadiness.Probe probe = t -> { + probes.incrementAndGet(); + return result(true, "ok"); + }; + Map first = StorageReadiness.check(probe, 1000L, 60_000L); + Map second = StorageReadiness.check(probe, 1000L, 60_000L); + Assert.assertTrue(StorageReadiness.isReady(first)); + Assert.assertEquals("hstore", first.get("storage")); + Assert.assertEquals(false, first.get("cached")); + Assert.assertEquals(true, second.get("cached")); + Assert.assertEquals(1, probes.get()); + } + + @Test + public void testZeroTtlProbesEveryTime() { + AtomicInteger probes = new AtomicInteger(); + StorageReadiness.Probe probe = t -> { + probes.incrementAndGet(); + return result(false, "no active store registered in pd"); + }; + StorageReadiness.check(probe, 1000L, 0L); + Map body = StorageReadiness.check(probe, 1000L, 0L); + Assert.assertFalse(StorageReadiness.isReady(body)); + Assert.assertEquals(2, probes.get()); + Assert.assertEquals("no active store registered in pd", body.get("reason")); + } + + @Test + public void testProbeFailureIsNotReadyWithReason() { + StorageReadiness.Probe probe = t -> { + throw new IllegalStateException("The 'hugegraph' store of hstore has not been opened"); + }; + Map body = StorageReadiness.check(probe, 1000L, 0L); + Assert.assertFalse(StorageReadiness.isReady(body)); + Assert.assertContains("probe failed: IllegalStateException", (String) body.get("reason")); + Assert.assertContains("has not been opened", (String) body.get("reason")); + } + + @Test + public void testTimeoutIsPassedToTheProbe() { + StorageReadiness.Probe probe = t -> result(true, "budget " + t); + Map body = StorageReadiness.check(probe, 750L, 0L); + Assert.assertEquals("budget 750", body.get("reason")); + } + + @Test + public void testCachedCopyIsIsolated() { + StorageReadiness.Probe probe = t -> result(true, "ok"); + Map first = StorageReadiness.check(probe, 1000L, 60_000L); + first.put("ready", false); + Map second = StorageReadiness.check(probe, 1000L, 60_000L); + Assert.assertTrue(StorageReadiness.isReady(second)); + } + + /** + * A Kubernetes httpGet probe carries no credential and no graphspace, so + * the endpoint must pass both the graphspace path rewrite and the auth + * filter, exactly like /versions. + */ + @Test + public void testReadinessBypassesPathAndAuthFilters() { + Assert.assertTrue(PathFilter.isWhiteAPI("readiness")); + Assert.assertTrue(PathFilter.isWhiteAPI("versions")); + UriInfo uri = Mockito.mock(UriInfo.class); + Mockito.when(uri.getPath()).thenReturn("readiness"); + ContainerRequestContext ctx = Mockito.mock(ContainerRequestContext.class); + Mockito.when(ctx.getUriInfo()).thenReturn(uri); + Assert.assertTrue(AuthenticationFilter.isWhiteAPI(ctx)); + Mockito.when(uri.getPath()).thenReturn("readiness/"); + Assert.assertFalse(AuthenticationFilter.isWhiteAPI(ctx)); + } +} From c9e70b14b65a8ec1735fbc47e377b54004e324a8 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Sat, 19 Sep 2026 04:38:17 +0000 Subject: [PATCH 2/4] feat(server): keep addresses out of the readiness body, single-flight PD refresh, prune dead Store channels Review round 1 of #3221: - the reason field carries a fixed category (gRPC status code, 'pd unreachable' or the exception class) instead of raw exception text, which could name PD peers and Store hosts on an unauthenticated endpoint; the full messages go to the log - the background PD refresh is single-flight (KnownStores keeps the future in flight), so a hung PD parks one thread, not one per cache-missed probe - channels of addresses PD no longer lists are shut down after a successful listing - Javadoc and option descriptions say what the probe does: PD only until the first Store list is known, one call per known Store in parallel Co-Authored-By: Claude Fable 5.1 --- .../hugegraph/api/profile/ReadinessAPI.java | 9 +- .../api/profile/StorageReadiness.java | 13 +- .../hugegraph/config/ServerOptions.java | 14 ++- .../store/hstore/HstoreStorageProbe.java | 117 +++++++++++++---- .../store/hstore/HstoreStorageProbeTest.java | 119 +++++++++++++++++- .../unit/core/StorageReadinessTest.java | 5 +- 6 files changed, 232 insertions(+), 45 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java index af9bf9c8ce..3f5a985aff 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java @@ -37,10 +37,11 @@ import jakarta.ws.rs.core.Response; /** - * Storage-aware readiness for Kubernetes and load balancers: 200 while this - * server can serve graph traffic, 503 while PD or every Store is unreachable - * from it. Unauthenticated, like /versions, so that an httpGet probe needs - * no credential; the body carries no addresses. + * Storage-aware readiness for Kubernetes and load balancers: 200 while at + * least one known Store answers this server, 503 while none does (or, before + * any Store list is known, while PD does not answer). Unauthenticated, like + * /versions, so that an httpGet probe needs no credential; the body carries + * no addresses and no raw exception text. */ @Path("readiness") @Singleton diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java index 415202d7b8..b455be1d61 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java @@ -32,9 +32,11 @@ * Whether this server can serve graph traffic, for a readiness probe. * Graphs on an embedded backend are ready as soon as the REST layer answers. * Graphs on hstore are probed through the backend's "storage_readiness" - * metadata (PD answers and one active Store answers). The storage is shared - * by every hstore graph of the process, so one graph is probed and the - * result is reused for a short TTL to keep repeated probes cheap. + * metadata: at least one known Store answers a cheap direct call; PD is only + * needed until the first Store list is known. The storage is shared by every + * hstore graph of the process, so one graph is probed and the result is + * reused for a short TTL to keep repeated probes cheap. The body never + * carries raw exception text, since the endpoint is unauthenticated. */ public final class StorageReadiness { @@ -93,10 +95,9 @@ public static synchronized Map check(Probe probe, long timeoutMs Map result = probe.probe(timeoutMs); body.putAll(result); } catch (Throwable e) { - LOG.debug("Storage readiness probe failed", e); + LOG.warn("Storage readiness probe failed", e); body.put("ready", false); - body.put("reason", "probe failed: " + e.getClass().getSimpleName() + - (e.getMessage() == null ? "" : ": " + e.getMessage())); + body.put("reason", "probe failed: " + e.getClass().getSimpleName()); } body.put("cached", false); lastResult = body; diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java index 46d1c8c48b..4b1f40f946 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java @@ -214,10 +214,12 @@ public class ServerOptions extends OptionHolder { public static final ConfigOption READINESS_TIMEOUT = new ConfigOption<>( "readiness.timeout", - "The whole time budget in ms of one GET /readiness probe " + - "(PD call plus one cheap call to an active Store); a PD " + - "or Store that does not answer within it makes the " + - "server report not ready.", + "The whole time budget in ms of one GET /readiness probe: " + + "one cheap call to every known Store in parallel, the " + + "first answer wins; the PD call that refreshes the Store " + + "list runs in the background and is only waited for " + + "before the first list is known. No Store answering " + + "within the budget makes the server report not ready.", rangeInt(100, 60000), 1000 ); @@ -227,8 +229,8 @@ public class ServerOptions extends OptionHolder { "readiness.cache_ttl", "How many ms a GET /readiness result is reused before " + "the storage is probed again, so that several probes " + - "(Kubernetes, load balancers) cost one PD and one Store " + - "call per interval; 0 probes on every request.", + "(Kubernetes, load balancers) cost one round of Store " + + "calls per interval; 0 probes on every request.", rangeInt(0, 60000), 2000 ); diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java index f1d77ee349..390bbacc1e 100644 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java @@ -19,6 +19,8 @@ import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; +import java.util.Set; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -34,31 +36,40 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.grpc.Metapb; import org.apache.hugegraph.store.grpc.state.HgStoreStateGrpc; import org.apache.hugegraph.store.grpc.state.SubStateReq; import org.apache.hugegraph.util.E; +import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; +import io.grpc.StatusRuntimeException; /** * Storage-aware readiness of this server, from this server's point of view: * at least one Store answers a direct, local, read-only gRPC call * (HgStoreState.getScanState, which reads the node's own scan-pool stats and - * never touches raft). The Store list comes from PD, but PD is refreshed in - * the background and the last known list is used right away, so a PD that is - * slow, restarting or down does not change the readiness of a server whose - * Stores still answer. Every wait is bounded by one shared time budget, so a - * hung PD or Store turns into "not ready" instead of a hung probe. The result - * carries no addresses, since it is served without authentication. + * never touches raft). The Store list comes from PD, refreshed in the + * background (single-flight) while the last known list is used right away, so + * PD only matters until the first list is known: a PD that is slow, restarting + * or down afterwards does not change the readiness of a server whose Stores + * still answer. Every known Store is pinged in parallel and the first answer + * wins; every wait is bounded by one shared time budget. The result carries + * no addresses and no raw exception text, since it is served without + * authentication; the full messages go to the log. */ public final class HstoreStorageProbe { public static final String META_STORAGE_READINESS = "storage_readiness"; + private static final Logger LOG = Log.logger(HstoreStorageProbe.class); + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( new ThreadFactory() { private final AtomicInteger seq = new AtomicInteger(); @@ -96,6 +107,8 @@ public static final class KnownStores { private volatile long at; private volatile Boolean pdOk; private volatile long pdAt; + private final AtomicReference>> inFlight = + new AtomicReference<>(); public List stores() { return this.stores; @@ -127,6 +140,34 @@ public void pdFailed() { this.pdOk = false; this.pdAt = System.currentTimeMillis(); } + + /** + * The refresh in flight, or a new one started on `executor`: only one + * PD call runs at a time no matter how many probes miss the cache, + * so a hung PD parks one thread, not one per probe. + */ + CompletableFuture> refresh(StoreLister lister, + ExecutorService executor) { + CompletableFuture> running = this.inFlight.get(); + if (running != null && !running.isDone()) { + return running; + } + CompletableFuture> mine = new CompletableFuture<>(); + if (!this.inFlight.compareAndSet(running, mine)) { + return this.inFlight.get(); + } + executor.execute(() -> { + try { + List stores = lister.activeStores(); + this.update(stores); + mine.complete(stores == null ? Collections.emptyList() : stores); + } catch (Throwable e) { + this.pdFailed(); + mine.completeExceptionally(e); + } + }); + return mine; + } } public static final class Result { @@ -199,8 +240,30 @@ public static Map probe(String graphName, long timeoutMs) { return new Result(false, "pd client not initialised", 0, null, false, -1L, -1L, 0L).toMap(); } - return probe(KNOWN, pd::getActiveStores, HstoreStorageProbe::pingScanState, - timeoutMs, EXECUTOR).toMap(); + return probe(KNOWN, () -> { + List stores = pd.getActiveStores(); + pruneChannels(CHANNELS, stores); + return stores; + }, HstoreStorageProbe::pingScanState, timeoutMs, EXECUTOR).toMap(); + } + + /** Shut down the channels of addresses PD no longer lists (replaced Stores). */ + static void pruneChannels(Map channels, + List stores) { + if (stores == null) { + return; + } + Set live = new HashSet<>(); + for (Metapb.Store store : stores) { + live.add(store.getAddress()); + } + channels.entrySet().removeIf(e -> { + if (live.contains(e.getKey())) { + return false; + } + e.getValue().shutdownNow(); + return true; + }); } private static void pingScanState(Metapb.Store store, long timeoutMs) { @@ -217,19 +280,9 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); long deadline = System.currentTimeMillis() + timeoutMs; - // Refresh the store list from PD in the background; whatever PD - // answers lands in `known` for this or the next probe - CompletableFuture> refresh = new CompletableFuture<>(); - executor.execute(() -> { - try { - List stores = lister.activeStores(); - known.update(stores); - refresh.complete(stores == null ? Collections.emptyList() : stores); - } catch (Throwable e) { - known.pdFailed(); - refresh.completeExceptionally(e); - } - }); + // Refresh the store list from PD in the background (single-flight); + // whatever PD answers lands in `known` for this or the next probe + CompletableFuture> refresh = known.refresh(lister, executor); List stores = known.stores(); Boolean pdReachable = null; @@ -242,7 +295,8 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi return new Result(false, "no store list known and pd did not answer within " + timeoutMs + " ms", 0, null, false, -1L, -1L, 0L); } catch (Exception e) { - return new Result(false, "no store list known and pd failed: " + message(e), + LOG.warn("Storage readiness: no store list known and pd failed", e); + return new Result(false, "no store list known and pd failed: " + category(e), 0, null, false, known.pdAgeMs(), -1L, 0L); } if (stores.isEmpty()) { @@ -291,7 +345,8 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi known.ageMs(), elapsed(storeStart)); } catch (ExecutionException e) { Throwable cause = e.getCause() != null ? e.getCause() : e; - failures.add("a store failed: " + message(cause)); + LOG.debug("Storage readiness: a store ping failed", cause); + failures.add("a store failed: " + category(cause)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); failures.add("interrupted"); @@ -341,8 +396,18 @@ private static long elapsed(long since) { return System.currentTimeMillis() - since; } - private static String message(Throwable e) { - String msg = e.getMessage(); - return e.getClass().getSimpleName() + (msg == null ? "" : ": " + msg); + /** + * A fixed category for the unauthenticated body: the gRPC status code, "pd + * unreachable" or the exception class, never the message (it can carry PD + * peers and Store host names). + */ + static String category(Throwable e) { + if (e instanceof StatusRuntimeException) { + return ((StatusRuntimeException) e).getStatus().getCode().name(); + } + if (e instanceof PDException) { + return "pd unreachable"; + } + return e.getClass().getSimpleName(); } } diff --git a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java index 473f253fe5..2447944247 100644 --- a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java @@ -23,16 +23,24 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.KnownStores; import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.Result; +import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.grpc.Metapb; import org.junit.AfterClass; import org.junit.Assert; import org.junit.Test; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import io.grpc.Status; + public class HstoreStorageProbeTest { private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool(); @@ -214,7 +222,8 @@ public void testNotReadyWhenEveryStoreFails() { refused, BUDGET, EXECUTOR); Assert.assertFalse(r.ready()); Assert.assertTrue(r.reason(), r.reason().startsWith("none of 2 known store(s) answered")); - Assert.assertTrue(r.reason(), r.reason().contains("connection refused")); + Assert.assertTrue(r.reason(), r.reason().contains("a store failed: IllegalStateException")); + Assert.assertFalse(r.reason(), r.reason().contains("connection refused")); Assert.assertNull(r.answeredStore()); } @@ -261,4 +270,112 @@ public void testRejectsNonPositiveBudget() { 0L, EXECUTOR); }); } + + /** + * The body is served without authentication: PD peers from the PD client's + * "PD unreachable, pd.peers=..." and Store host names from gRPC's "Unable + * to resolve host ..." must not reach it, only a category. + */ + @Test + public void testReasonCarriesNoPdPeersNorStoreHosts() { + Result pd = HstoreStorageProbe.probe(new KnownStores(), () -> { + throw new PDException(1, "PD unreachable, pd.peers=pd-0.internal:8686,pd-1.internal:8686"); + }, ANSWERS, BUDGET, EXECUTOR); + Assert.assertFalse(pd.ready()); + Assert.assertEquals("no store list known and pd failed: pd unreachable", pd.reason()); + + HstoreStorageProbe.StorePinger unresolved = (store, timeout) -> { + throw Status.UNAVAILABLE.withDescription( + "Unable to resolve host store-0.hugegraph-store.svc").asRuntimeException(); + }; + Result st = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), unresolved, + BUDGET, EXECUTOR); + Assert.assertFalse(st.ready()); + Assert.assertTrue(st.reason(), st.reason().contains("a store failed: UNAVAILABLE")); + String all = pd.toMap().toString() + st.toMap().toString(); + Assert.assertFalse(all, all.contains("internal") || all.contains("svc") || + all.contains("8686")); + } + + /** A hung PD parks one refresh, not one per probe. */ + @Test + public void testRefreshIsSingleFlight() throws Exception { + AtomicInteger calls = new AtomicInteger(); + KnownStores known = knowing(1L); + HstoreStorageProbe.StoreLister hung = () -> { + calls.incrementAndGet(); + Thread.sleep(3_000L); + return stores(1L, 2L); + }; + for (int i = 0; i < 5; i++) { + Assert.assertTrue(HstoreStorageProbe.probe(known, hung, ANSWERS, BUDGET, + EXECUTOR).ready()); + } + Thread.sleep(200L); + Assert.assertEquals(1, calls.get()); + for (int i = 0; i < 40 && known.stores().size() != 2; i++) { + Thread.sleep(100L); + } + Assert.assertEquals(2, known.stores().size()); + HstoreStorageProbe.probe(known, hung, ANSWERS, BUDGET, EXECUTOR); + Thread.sleep(100L); + Assert.assertEquals("a finished refresh allows a new one", 2, calls.get()); + } + + private static final class FakeChannel extends ManagedChannel { + + boolean shut; + + @Override + public ManagedChannel shutdown() { + this.shut = true; + return this; + } + + @Override + public boolean isShutdown() { + return this.shut; + } + + @Override + public boolean isTerminated() { + return this.shut; + } + + @Override + public ManagedChannel shutdownNow() { + return this.shutdown(); + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return true; + } + + @Override + public ClientCall newCall(MethodDescriptor method, + CallOptions options) { + throw new UnsupportedOperationException(); + } + + @Override + public String authority() { + return "fake"; + } + } + + @Test + public void testChannelsOfReplacedStoresAreShutDown() { + Map channels = new java.util.concurrent.ConcurrentHashMap<>(); + FakeChannel kept = new FakeChannel(); + FakeChannel gone = new FakeChannel(); + channels.put("10.0.0.1:8500", kept); + channels.put("10.0.0.9:8500", gone); + HstoreStorageProbe.pruneChannels(channels, stores(1L, 2L)); + Assert.assertEquals(1, channels.size()); + Assert.assertFalse(kept.shut); + Assert.assertTrue(gone.shut); + HstoreStorageProbe.pruneChannels(channels, null); + Assert.assertEquals("a failed listing prunes nothing", 1, channels.size()); + } } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java index d4bda1dd50..64e6a0c126 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java @@ -86,8 +86,9 @@ public void testProbeFailureIsNotReadyWithReason() { }; Map body = StorageReadiness.check(probe, 1000L, 0L); Assert.assertFalse(StorageReadiness.isReady(body)); - Assert.assertContains("probe failed: IllegalStateException", (String) body.get("reason")); - Assert.assertContains("has not been opened", (String) body.get("reason")); + Assert.assertEquals("probe failed: IllegalStateException", body.get("reason")); + // the endpoint is unauthenticated: no raw message in the body + Assert.assertFalse(body.toString().contains("has not been opened")); } @Test From 213fd9c0ffe459df2f2bfa4b564dc2521cd601e9 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Sat, 19 Sep 2026 08:07:26 +0000 Subject: [PATCH 3/4] feat(server): simplify the readiness probe (BasicThreadFactory, no unused graphName, map result) Review round 2 of #3221: the executor uses commons-lang3's BasicThreadFactory like ExecutorUtil does, the unread graphName parameter and its concat at the HstoreStore call site are gone, and probe() builds the body map through one private helper instead of a Result holder whose only consumer was toMap(); the tests assert on map entries. Co-Authored-By: Claude Fable 5.1 --- .../store/hstore/HstoreStorageProbe.java | 109 +++++------------- .../backend/store/hstore/HstoreStore.java | 3 +- .../store/hstore/HstoreStorageProbeTest.java | 108 ++++++++--------- 3 files changed, 86 insertions(+), 134 deletions(-) diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java index 390bbacc1e..35d0773058 100644 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java @@ -32,12 +32,11 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.apache.hugegraph.pd.client.PDClient; import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.grpc.Metapb; @@ -71,16 +70,8 @@ public final class HstoreStorageProbe { private static final Logger LOG = Log.logger(HstoreStorageProbe.class); private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( - new ThreadFactory() { - private final AtomicInteger seq = new AtomicInteger(); - - @Override - public Thread newThread(Runnable r) { - Thread t = new Thread(r, "storage-readiness-" + this.seq.incrementAndGet()); - t.setDaemon(true); - return t; - } - }); + new BasicThreadFactory.Builder().namingPattern("storage-readiness-%d") + .daemon(true).build()); private static final KnownStores KNOWN = new KnownStores(); private static final Map CHANNELS = new ConcurrentHashMap<>(); @@ -170,81 +161,38 @@ CompletableFuture> refresh(StoreLister lister, } } - public static final class Result { - - private final boolean ready; - private final String reason; - private final int activeStores; - private final Long answeredStore; - private final Boolean pdReachable; - private final long pdAgeMs; - private final long storesAgeMs; - private final long storeMillis; - - Result(boolean ready, String reason, int activeStores, Long answeredStore, - Boolean pdReachable, long pdAgeMs, long storesAgeMs, long storeMillis) { - this.ready = ready; - this.reason = reason; - this.activeStores = activeStores; - this.answeredStore = answeredStore; - this.pdReachable = pdReachable; - this.pdAgeMs = pdAgeMs; - this.storesAgeMs = storesAgeMs; - this.storeMillis = storeMillis; - } - - public boolean ready() { - return this.ready; - } - - public String reason() { - return this.reason; - } - - public int activeStores() { - return this.activeStores; - } - - public Long answeredStore() { - return this.answeredStore; - } - - public Boolean pdReachable() { - return this.pdReachable; - } - - public Map toMap() { - Map map = new LinkedHashMap<>(); - map.put("ready", this.ready); - map.put("reason", this.reason); - map.put("active_stores", this.activeStores); - map.put("answered_store", this.answeredStore); - map.put("pd_reachable", this.pdReachable); - map.put("pd_checked_age_ms", this.pdAgeMs); - map.put("stores_age_ms", this.storesAgeMs); - map.put("store_millis", this.storeMillis); - return map; - } + /** The unauthenticated body: no addresses, no raw exception text. */ + private static Map result(boolean ready, String reason, int activeStores, + Long answeredStore, Boolean pdReachable, + long pdAgeMs, long storesAgeMs, long storeMillis) { + Map map = new LinkedHashMap<>(); + map.put("ready", ready); + map.put("reason", reason); + map.put("active_stores", activeStores); + map.put("answered_store", answeredStore); + map.put("pd_reachable", pdReachable); + map.put("pd_checked_age_ms", pdAgeMs); + map.put("stores_age_ms", storesAgeMs); + map.put("store_millis", storeMillis); + return map; } /** * Probe through the process-wide PD client and this probe's own plaintext * channels to the stores (the store gRPC server takes no credentials). * - * @param graphName the store-side graph name, kept for the meta handler * @param timeoutMs the whole budget for PD plus stores */ - public static Map probe(String graphName, long timeoutMs) { + public static Map probe(long timeoutMs) { PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); if (pd == null) { - return new Result(false, "pd client not initialised", 0, null, - false, -1L, -1L, 0L).toMap(); + return result(false, "pd client not initialised", 0, null, false, -1L, -1L, 0L); } return probe(KNOWN, () -> { List stores = pd.getActiveStores(); pruneChannels(CHANNELS, stores); return stores; - }, HstoreStorageProbe::pingScanState, timeoutMs, EXECUTOR).toMap(); + }, HstoreStorageProbe::pingScanState, timeoutMs, EXECUTOR); } /** Shut down the channels of addresses PD no longer lists (replaced Stores). */ @@ -275,8 +223,9 @@ private static void pingScanState(Metapb.Store store, long timeoutMs) { .getScanState(SubStateReq.getDefaultInstance()); } - public static Result probe(KnownStores known, StoreLister lister, StorePinger pinger, - long timeoutMs, ExecutorService executor) { + public static Map probe(KnownStores known, StoreLister lister, + StorePinger pinger, long timeoutMs, + ExecutorService executor) { E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); long deadline = System.currentTimeMillis() + timeoutMs; @@ -292,15 +241,15 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi stores = await(refresh, deadline); pdReachable = true; } catch (TimeoutException e) { - return new Result(false, "no store list known and pd did not answer within " + + return result(false, "no store list known and pd did not answer within " + timeoutMs + " ms", 0, null, false, -1L, -1L, 0L); } catch (Exception e) { LOG.warn("Storage readiness: no store list known and pd failed", e); - return new Result(false, "no store list known and pd failed: " + category(e), + return result(false, "no store list known and pd failed: " + category(e), 0, null, false, known.pdAgeMs(), -1L, 0L); } if (stores.isEmpty()) { - return new Result(false, "no active store registered in pd", + return result(false, "no active store registered in pd", 0, null, true, known.pdAgeMs(), known.ageMs(), 0L); } } @@ -319,7 +268,7 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi })); } List failures = new ArrayList<>(); - Result result = null; + Map result = null; try { for (int done = 0; done < stores.size() && result == null; done++) { long remaining = deadline - System.currentTimeMillis(); @@ -340,7 +289,7 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi } try { Metapb.Store store = first.get(); - result = new Result(true, "ok", stores.size(), store.getId(), + result = result(true, "ok", stores.size(), store.getId(), pdState(refresh, known, pdReachable), known.pdAgeMs(), known.ageMs(), elapsed(storeStart)); } catch (ExecutionException e) { @@ -361,7 +310,7 @@ public static Result probe(KnownStores known, StoreLister lister, StorePinger pi if (result != null) { return result; } - return new Result(false, "none of " + stores.size() + " known store(s) answered: " + + return result(false, "none of " + stores.size() + " known store(s) answered: " + String.join("; ", failures), stores.size(), null, pdState(refresh, known, pdReachable), known.pdAgeMs(), known.ageMs(), elapsed(storeStart)); diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java index 0797da85ee..08d87a2aae 100644 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStore.java @@ -117,8 +117,7 @@ private void registerMetaHandlers() { this.registerMetaHandler(HstoreStorageProbe.META_STORAGE_READINESS, (session, meta, args) -> { E.checkArgument(args.length == 1 && args[0] instanceof Number, "Expect the timeout in ms as the only argument"); - return HstoreStorageProbe.probe(this.namespace + "/" + this.store, - ((Number) args[0]).longValue()); + return HstoreStorageProbe.probe(((Number) args[0]).longValue()); }); this.registerMetaHandler("mode", (session, meta, args) -> { E.checkArgument(args.length == 1, diff --git a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java index 2447944247..33fe18b59d 100644 --- a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java @@ -28,7 +28,6 @@ import java.util.stream.Collectors; import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.KnownStores; -import org.apache.hugegraph.backend.store.hstore.HstoreStorageProbe.Result; import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.grpc.Metapb; import org.junit.AfterClass; @@ -70,15 +69,19 @@ private static KnownStores knowing(long... ids) { private static final HstoreStorageProbe.StorePinger ANSWERS = (store, timeout) -> { }; + private static String reason(Map body) { + return (String) body.get("reason"); + } + @Test public void testReadyWhenPdAndOneStoreAnswer() { - Result r = HstoreStorageProbe.probe(new KnownStores(), () -> stores(1L, 2L, 3L), + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> stores(1L, 2L, 3L), ANSWERS, BUDGET, EXECUTOR); - Assert.assertTrue(r.reason(), r.ready()); - Assert.assertEquals(3, r.activeStores()); - Assert.assertNotNull(r.answeredStore()); - Assert.assertEquals(Boolean.TRUE, r.pdReachable()); - Assert.assertEquals("ok", r.reason()); + Assert.assertTrue(reason(r), Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals(3, r.get("active_stores")); + Assert.assertNotNull(r.get("answered_store")); + Assert.assertEquals(Boolean.TRUE, r.get("pd_reachable")); + Assert.assertEquals("ok", reason(r)); } @Test @@ -90,10 +93,10 @@ public void testFirstAnsweringStoreWinsAfterFailures() { throw new IllegalStateException("UNAVAILABLE"); } }; - Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), onlyThird, BUDGET, EXECUTOR); - Assert.assertTrue(r.reason(), r.ready()); - Assert.assertEquals(Long.valueOf(3L), r.answeredStore()); + Assert.assertTrue(reason(r), Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals(3L, r.get("answered_store")); // the pings run in parallel; the failing ones may or may not have run Assert.assertTrue(pings.get() >= 1 && pings.get() <= 3); } @@ -111,35 +114,35 @@ public void testHungStoreDoesNotHideAnAnsweringOne() { } }; long start = System.currentTimeMillis(); - Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), onlySecondAnswers, BUDGET, EXECUTOR); long took = System.currentTimeMillis() - start; - Assert.assertTrue(r.reason(), r.ready()); - Assert.assertEquals(Long.valueOf(2L), r.answeredStore()); + Assert.assertTrue(reason(r), Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals(2L, r.get("answered_store")); Assert.assertTrue("took " + took, took < BUDGET); } @Test public void testFirstProbeWithoutKnownStoresNeedsPd() { - Result r = HstoreStorageProbe.probe(new KnownStores(), () -> { + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> { throw new IllegalStateException("UNAVAILABLE: io exception"); }, ANSWERS, BUDGET, EXECUTOR); - Assert.assertFalse(r.ready()); - Assert.assertTrue(r.reason(), r.reason().startsWith( + Assert.assertFalse(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertTrue(reason(r), reason(r).startsWith( "no store list known and pd failed: IllegalStateException")); - Assert.assertEquals(Boolean.FALSE, r.pdReachable()); + Assert.assertEquals(Boolean.FALSE, r.get("pd_reachable")); } @Test public void testFirstProbeWithHungPdStaysWithinBudget() { long start = System.currentTimeMillis(); - Result r = HstoreStorageProbe.probe(new KnownStores(), () -> { + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> { Thread.sleep(10_000L); return stores(1L); }, ANSWERS, BUDGET, EXECUTOR); long took = System.currentTimeMillis() - start; - Assert.assertFalse(r.ready()); - Assert.assertTrue(r.reason(), r.reason().contains("pd did not answer within")); + Assert.assertFalse(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertTrue(reason(r), reason(r).contains("pd did not answer within")); Assert.assertTrue("took " + took, took < BUDGET * 4); } @@ -149,25 +152,25 @@ public void testFirstProbeWithHungPdStaysWithinBudget() { */ @Test public void testKnownStoresKeepTheServerReadyWhilePdIsDown() { - Result r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { + Map r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { throw new IllegalStateException("PD unreachable"); }, ANSWERS, BUDGET, EXECUTOR); - Assert.assertTrue(r.reason(), r.ready()); - Assert.assertEquals(2, r.activeStores()); + Assert.assertTrue(reason(r), Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals(2, r.get("active_stores")); } @Test public void testHungPdDoesNotDelayAProbeWithKnownStores() { long start = System.currentTimeMillis(); - Result r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { + Map r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { Thread.sleep(10_000L); return stores(1L, 2L); }, ANSWERS, BUDGET, EXECUTOR); long took = System.currentTimeMillis() - start; - Assert.assertTrue(r.reason(), r.ready()); + Assert.assertTrue(reason(r), Boolean.TRUE.equals(r.get("ready"))); // the refresh is still pending, so the outcome of the last finished // one (the seed) is reported - Assert.assertEquals(Boolean.TRUE, r.pdReachable()); + Assert.assertEquals(Boolean.TRUE, r.get("pd_reachable")); Assert.assertTrue("took " + took, took < BUDGET); } @@ -181,13 +184,13 @@ public void testLastPdOutcomeIsReportedWhileTheRefreshIsPending() throws Excepti Thread.sleep(20L); } Assert.assertEquals(Boolean.FALSE, known.pdOk()); - Result r = HstoreStorageProbe.probe(known, () -> { + Map r = HstoreStorageProbe.probe(known, () -> { Thread.sleep(10_000L); return stores(1L); }, ANSWERS, BUDGET, EXECUTOR); - Assert.assertTrue(r.ready()); - Assert.assertEquals(Boolean.FALSE, r.pdReachable()); - Assert.assertTrue(r.toMap().containsKey("pd_checked_age_ms")); + Assert.assertTrue(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals(Boolean.FALSE, r.get("pd_reachable")); + Assert.assertTrue(r.containsKey("pd_checked_age_ms")); } @Test @@ -204,10 +207,10 @@ public void testPdAnswerUpdatesTheKnownStoresForTheNextProbe() throws Exception @Test public void testEmptyPdAnswerIsNotReadyAndKeepsTheOldList() { KnownStores fresh = new KnownStores(); - Result r = HstoreStorageProbe.probe(fresh, Collections::emptyList, ANSWERS, + Map r = HstoreStorageProbe.probe(fresh, Collections::emptyList, ANSWERS, BUDGET, EXECUTOR); - Assert.assertFalse(r.ready()); - Assert.assertEquals("no active store registered in pd", r.reason()); + Assert.assertFalse(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertEquals("no active store registered in pd", reason(r)); KnownStores known = knowing(1L); HstoreStorageProbe.probe(known, Collections::emptyList, ANSWERS, BUDGET, EXECUTOR); Assert.assertEquals(1, known.stores().size()); @@ -218,13 +221,13 @@ public void testNotReadyWhenEveryStoreFails() { HstoreStorageProbe.StorePinger refused = (store, timeout) -> { throw new IllegalStateException("connection refused"); }; - Result r = HstoreStorageProbe.probe(knowing(7L, 8L), () -> stores(7L, 8L), + Map r = HstoreStorageProbe.probe(knowing(7L, 8L), () -> stores(7L, 8L), refused, BUDGET, EXECUTOR); - Assert.assertFalse(r.ready()); - Assert.assertTrue(r.reason(), r.reason().startsWith("none of 2 known store(s) answered")); - Assert.assertTrue(r.reason(), r.reason().contains("a store failed: IllegalStateException")); - Assert.assertFalse(r.reason(), r.reason().contains("connection refused")); - Assert.assertNull(r.answeredStore()); + Assert.assertFalse(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertTrue(reason(r), reason(r).startsWith("none of 2 known store(s) answered")); + Assert.assertTrue(reason(r), reason(r).contains("a store failed: IllegalStateException")); + Assert.assertFalse(reason(r), reason(r).contains("connection refused")); + Assert.assertNull(r.get("answered_store")); } @Test @@ -233,11 +236,11 @@ public void testHungStoresStayWithinTheBudget() { Thread.sleep(10_000L); }; long start = System.currentTimeMillis(); - Result r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), hung, BUDGET, EXECUTOR); long took = System.currentTimeMillis() - start; - Assert.assertFalse(r.ready()); - Assert.assertTrue(r.reason(), r.reason().contains("did not answer within")); + Assert.assertFalse(Boolean.TRUE.equals(r.get("ready"))); + Assert.assertTrue(reason(r), reason(r).contains("did not answer within")); Assert.assertTrue("took " + took, took < BUDGET * 4); } @@ -254,7 +257,7 @@ public void testPingGetsTheRemainingBudget() { @Test public void testMapCarriesNoAddresses() { Map map = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), - ANSWERS, BUDGET, EXECUTOR).toMap(); + ANSWERS, BUDGET, EXECUTOR); Assert.assertEquals(true, map.get("ready")); Assert.assertEquals(1, map.get("active_stores")); Assert.assertEquals(1L, map.get("answered_store")); @@ -278,21 +281,21 @@ public void testRejectsNonPositiveBudget() { */ @Test public void testReasonCarriesNoPdPeersNorStoreHosts() { - Result pd = HstoreStorageProbe.probe(new KnownStores(), () -> { + Map pd = HstoreStorageProbe.probe(new KnownStores(), () -> { throw new PDException(1, "PD unreachable, pd.peers=pd-0.internal:8686,pd-1.internal:8686"); }, ANSWERS, BUDGET, EXECUTOR); - Assert.assertFalse(pd.ready()); - Assert.assertEquals("no store list known and pd failed: pd unreachable", pd.reason()); + Assert.assertFalse(Boolean.TRUE.equals(pd.get("ready"))); + Assert.assertEquals("no store list known and pd failed: pd unreachable", reason(pd)); HstoreStorageProbe.StorePinger unresolved = (store, timeout) -> { throw Status.UNAVAILABLE.withDescription( "Unable to resolve host store-0.hugegraph-store.svc").asRuntimeException(); }; - Result st = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), unresolved, + Map st = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), unresolved, BUDGET, EXECUTOR); - Assert.assertFalse(st.ready()); - Assert.assertTrue(st.reason(), st.reason().contains("a store failed: UNAVAILABLE")); - String all = pd.toMap().toString() + st.toMap().toString(); + Assert.assertFalse(Boolean.TRUE.equals(st.get("ready"))); + Assert.assertTrue(reason(st), reason(st).contains("a store failed: UNAVAILABLE")); + String all = pd.toString() + st.toString(); Assert.assertFalse(all, all.contains("internal") || all.contains("svc") || all.contains("8686")); } @@ -308,8 +311,9 @@ public void testRefreshIsSingleFlight() throws Exception { return stores(1L, 2L); }; for (int i = 0; i < 5; i++) { - Assert.assertTrue(HstoreStorageProbe.probe(known, hung, ANSWERS, BUDGET, - EXECUTOR).ready()); + Assert.assertEquals(Boolean.TRUE, HstoreStorageProbe.probe(known, hung, ANSWERS, + BUDGET, EXECUTOR) + .get("ready")); } Thread.sleep(200L); Assert.assertEquals(1, calls.get()); From 4a4aa1cc1177c68631f282f7b86e140718abb669 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Sat, 19 Sep 2026 15:31:05 +0000 Subject: [PATCH 4/4] feat(server): exempt /readiness from load shedding, keep channels on an empty PD answer, add ReadinessApiTest Review round 3 of #3221: - readiness joins LoadDetectFilter.WHITE_API_LIST next to versions, so a busy server still answers its readiness probe from the storage state instead of shedding it (LoadReleaseFilter reads the same list); test case in LoadDetectFilterTest - pruneChannels() ignores an empty PD answer, the same rule KnownStores.update() applies, so the channels the pings still use stay open while PD reports no active Store; the channel test covers it - ReadinessApiTest drives GET /readiness through the API suite, with and without credentials, on every backend of the suite Co-Authored-By: Claude Fable 5.1 --- .../api/filter/LoadDetectFilter.java | 3 +- .../store/hstore/HstoreStorageProbe.java | 9 ++- .../store/hstore/HstoreStorageProbeTest.java | 4 ++ .../apache/hugegraph/api/ApiTestSuite.java | 1 + .../hugegraph/api/ReadinessApiTest.java | 71 +++++++++++++++++++ .../unit/api/filter/LoadDetectFilterTest.java | 17 +++++ 6 files changed, 102 insertions(+), 3 deletions(-) create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ReadinessApiTest.java diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/LoadDetectFilter.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/LoadDetectFilter.java index 1df19f5e5c..390c2c9e80 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/LoadDetectFilter.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/LoadDetectFilter.java @@ -53,7 +53,8 @@ public class LoadDetectFilter implements ContainerRequestFilter { "", "apis", "metrics", - "versions" + "versions", + "readiness" ); // Call gc every 30+ seconds if memory is low and request frequently diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java index 35d0773058..f2fe3c3f58 100644 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java @@ -195,10 +195,15 @@ public static Map probe(long timeoutMs) { }, HstoreStorageProbe::pingScanState, timeoutMs, EXECUTOR); } - /** Shut down the channels of addresses PD no longer lists (replaced Stores). */ + /** + * Shut down the channels of addresses PD no longer lists (replaced + * Stores). An empty answer is ignored, the same rule KnownStores.update + * applies: the pings keep using the last known Stores, so their channels + * must stay open. + */ static void pruneChannels(Map channels, List stores) { - if (stores == null) { + if (stores == null || stores.isEmpty()) { return; } Set live = new HashSet<>(); diff --git a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java index 33fe18b59d..939433f49f 100644 --- a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java @@ -381,5 +381,9 @@ public void testChannelsOfReplacedStoresAreShutDown() { Assert.assertTrue(gone.shut); HstoreStorageProbe.pruneChannels(channels, null); Assert.assertEquals("a failed listing prunes nothing", 1, channels.size()); + HstoreStorageProbe.pruneChannels(channels, Collections.emptyList()); + Assert.assertEquals("an empty listing keeps the channels the pings still use", + 1, channels.size()); + Assert.assertFalse(kept.shut); } } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ApiTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ApiTestSuite.java index f4d74e49b3..a2ebf17926 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ApiTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ApiTestSuite.java @@ -35,6 +35,7 @@ TaskApiTest.class, GremlinApiTest.class, MetricsApiTest.class, + ReadinessApiTest.class, UserApiTest.class, LoginApiTest.class, ProjectApiTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ReadinessApiTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ReadinessApiTest.java new file mode 100644 index 0000000000..636c1f7fea --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/ReadinessApiTest.java @@ -0,0 +1,71 @@ +/* + * 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.apache.hugegraph.api; + +import java.util.Map; + +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.util.JsonUtil; +import org.junit.Test; + +import jakarta.ws.rs.client.ClientBuilder; +import jakarta.ws.rs.core.Response; + +/** + * The readiness endpoint answers 200 on a healthy server whatever the + * backend: "embedded" for the in-process backends of the API suite, "hstore" + * (with the storage fields) on the hstore job. + */ +public class ReadinessApiTest extends BaseApiTest { + + private static final String PATH = "/readiness"; + + @Test + public void testReadyOnAHealthyServer() { + Response r = client().get(PATH); + String result = assertResponseStatus(200, r); + Map body = JsonUtil.fromJson(result, Map.class); + Assert.assertEquals(true, body.get("ready")); + Assert.assertTrue(String.valueOf(body.get("storage")), + "embedded".equals(body.get("storage")) || + "hstore".equals(body.get("storage"))); + Assert.assertNotNull(body.get("reason")); + if ("hstore".equals(body.get("storage"))) { + Assert.assertEquals("ok", body.get("reason")); + Assert.assertTrue(((Number) body.get("active_stores")).intValue() >= 1); + Assert.assertNotNull(body.get("answered_store")); + Assert.assertTrue(body.containsKey("cached")); + } + } + + /** + * A Kubernetes httpGet probe carries no credential and no graphspace + * prefix, so the endpoint must answer without either. + */ + @Test + public void testReadyWithoutCredentials() { + Response r = ClientBuilder.newClient().target(BASE_URL + PATH) + .request().get(); + try { + Assert.assertEquals(200, r.getStatus()); + Assert.assertContains("\"ready\":true", r.readEntity(String.class)); + } finally { + r.close(); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/filter/LoadDetectFilterTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/filter/LoadDetectFilterTest.java index 5be5b64a92..1cfd22e757 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/filter/LoadDetectFilterTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/filter/LoadDetectFilterTest.java @@ -112,6 +112,23 @@ public void testFilter_WhiteListPathIgnored() { Assert.assertTrue(this.testAppender.events().isEmpty()); } + /** + * A readiness probe must answer from the storage state, not from the + * worker load: a Server that is merely busy is still ready, and shedding + * probes would pull every busy Server out of the Service during a spike. + */ + @Test + public void testFilter_ReadinessIgnoredLikeVersions() { + setupPath("readiness", List.of("readiness")); + this.setConfigProvider(createConfig(2, 0)); + this.workLoad.incrementAndGet(); + + this.loadDetectFilter.filter(this.requestContext); + + Assert.assertEquals(1, this.workLoad.get().get()); + Assert.assertTrue(this.testAppender.events().isEmpty()); + } + @Test public void testFilter_RejectsWhenWorkerLoadIsTooHigh() { setupPath("graphs/hugegraph/vertices",