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/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-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..3f5a985aff --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java @@ -0,0 +1,67 @@ +/* + * 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 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 +@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..b455be1d61 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java @@ -0,0 +1,130 @@ +/* + * 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: 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 { + + 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.warn("Storage readiness probe failed", e); + body.put("ready", false); + body.put("reason", "probe failed: " + e.getClass().getSimpleName()); + } + 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..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 @@ -211,6 +211,30 @@ 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: " + + "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 + ); + + 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 round of Store " + + "calls 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..f2fe3c3f58 --- /dev/null +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java @@ -0,0 +1,367 @@ +/* + * 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.HashSet; +import java.util.Set; +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.TimeUnit; +import java.util.concurrent.TimeoutException; +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; +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, 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 BasicThreadFactory.Builder().namingPattern("storage-readiness-%d") + .daemon(true).build()); + + 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; + private final AtomicReference>> inFlight = + new AtomicReference<>(); + + 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(); + } + + /** + * 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; + } + } + + /** 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 timeoutMs the whole budget for PD plus stores + */ + public static Map probe(long timeoutMs) { + PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); + if (pd == null) { + 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); + } + + /** + * 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 || stores.isEmpty()) { + 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) { + 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 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; + + // 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; + 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 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 result(false, "no store list known and pd failed: " + category(e), + 0, null, false, known.pdAgeMs(), -1L, 0L); + } + if (stores.isEmpty()) { + return 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<>(); + Map 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 = 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; + 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"); + break; + } + } + } finally { + for (Future f : futures) { + f.cancel(true); + } + } + if (result != null) { + return result; + } + 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)); + } + + /** + * 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; + } + + /** + * 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/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..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 @@ -114,6 +114,11 @@ 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(((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..939433f49f --- /dev/null +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbeTest.java @@ -0,0 +1,389 @@ +/* + * 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.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.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(); + 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) -> { + }; + + private static String reason(Map body) { + return (String) body.get("reason"); + } + + @Test + public void testReadyWhenPdAndOneStoreAnswer() { + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> stores(1L, 2L, 3L), + ANSWERS, BUDGET, EXECUTOR); + 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 + public void testFirstAnsweringStoreWinsAfterFailures() { + AtomicInteger pings = new AtomicInteger(); + HstoreStorageProbe.StorePinger onlyThird = (store, timeout) -> { + pings.incrementAndGet(); + if (store.getId() != 3L) { + throw new IllegalStateException("UNAVAILABLE"); + } + }; + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + onlyThird, BUDGET, EXECUTOR); + 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); + } + + /** + * 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(); + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + onlySecondAnswers, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + 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() { + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> { + throw new IllegalStateException("UNAVAILABLE: io exception"); + }, ANSWERS, BUDGET, EXECUTOR); + 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.get("pd_reachable")); + } + + @Test + public void testFirstProbeWithHungPdStaysWithinBudget() { + long start = System.currentTimeMillis(); + Map r = HstoreStorageProbe.probe(new KnownStores(), () -> { + Thread.sleep(10_000L); + return stores(1L); + }, ANSWERS, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + 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); + } + + /** + * 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() { + Map r = HstoreStorageProbe.probe(knowing(1L, 2L), () -> { + throw new IllegalStateException("PD unreachable"); + }, ANSWERS, BUDGET, EXECUTOR); + 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(); + 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(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.get("pd_reachable")); + 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()); + Map r = HstoreStorageProbe.probe(known, () -> { + Thread.sleep(10_000L); + return stores(1L); + }, ANSWERS, BUDGET, EXECUTOR); + 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 + 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(); + Map r = HstoreStorageProbe.probe(fresh, Collections::emptyList, ANSWERS, + BUDGET, EXECUTOR); + 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()); + } + + @Test + public void testNotReadyWhenEveryStoreFails() { + HstoreStorageProbe.StorePinger refused = (store, timeout) -> { + throw new IllegalStateException("connection refused"); + }; + Map r = HstoreStorageProbe.probe(knowing(7L, 8L), () -> stores(7L, 8L), + refused, BUDGET, EXECUTOR); + 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 + public void testHungStoresStayWithinTheBudget() { + HstoreStorageProbe.StorePinger hung = (store, timeout) -> { + Thread.sleep(10_000L); + }; + long start = System.currentTimeMillis(); + Map r = HstoreStorageProbe.probe(knowing(1L, 2L, 3L), () -> stores(1L, 2L, 3L), + hung, BUDGET, EXECUTOR); + long took = System.currentTimeMillis() - start; + 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); + } + + @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); + 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); + }); + } + + /** + * 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() { + 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(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(); + }; + Map st = HstoreStorageProbe.probe(knowing(1L), () -> stores(1L), unresolved, + BUDGET, EXECUTOR); + 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")); + } + + /** 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.assertEquals(Boolean.TRUE, HstoreStorageProbe.probe(known, hung, ANSWERS, + BUDGET, EXECUTOR) + .get("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()); + 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/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/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", 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..64e6a0c126 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/StorageReadinessTest.java @@ -0,0 +1,127 @@ +/* + * 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.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 + 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)); + } +}