SebastianGruza commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4165457821
########## 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<String, Object> 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"))); Review Comment: Right, thanks, I did not think it through for the hbase job. The test accepts `storage` embedded, hstore or hbase; for hbase it checks `reason == "ok"` and the presence of `hbase_millis`, for both remote backends the presence of the `probes` list. The class Javadoc lists the three jobs. ########## hugegraph-server/hugegraph-hbase/src/main/java/org/apache/hugegraph/backend/store/hbase/HbaseStore.java: ########## @@ -87,6 +92,56 @@ private void registerMetaHandlers() { HbaseMetrics metrics = new HbaseMetrics(this.sessions); return metrics.compact(this.tableNames()); }); + + this.registerMetaHandler(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 this.storageReadiness(((Number) args[0]).longValue()); + }); + } + + public static final String META_STORAGE_READINESS = "storage_readiness"; + + private static final ExecutorService READINESS_EXECUTOR = Executors.newCachedThreadPool(r -> { + Thread thread = new Thread(r, "hbase-readiness"); + thread.setDaemon(true); + return thread; + }); + + /** + * Whether the HBase cluster answers this server: one admin round trip + * (does the graph's first table exist) within the budget. The body + * carries no addresses and no raw exception text, since the readiness + * endpoint is unauthenticated. + */ + private Map<String, Object> storageReadiness(long timeoutMs) { + E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", false); + List<String> tables = this.tableNames(); + String table = tables.isEmpty() ? null : tables.get(0); + long start = System.currentTimeMillis(); + Future<Boolean> exists = READINESS_EXECUTOR.submit(() -> { + return table != null && this.sessions.existsTable(table); Review Comment: Right. New `HbaseSessions.tableAvailable(table)`: `tableExists && isTableEnabled && isTableAvailable` through `Admin`. The probe logic moved to a static `HbaseStore.readinessOf(Callable, timeout, executor)`, testable without a cluster; `HbaseReadinessTest` covers: available, existing but disabled (not ready, reason "the graph's first table is not available"), a failing admin call (no exception text in the body), and a hung call (bounded by the budget). ########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java: ########## @@ -0,0 +1,223 @@ +/* + * 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 java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +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.HugeGraph; +import org.apache.hugegraph.auth.HugeGraphAuthProxy; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; + +import com.google.common.collect.ImmutableSet; + +/** + * 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 a remote backend are probed through the backend's + * "storage_readiness" metadata: on hstore at least one known Store answers a + * cheap direct call (PD is only needed until the first Store list is known), + * on hbase the cluster answers an admin call about one of the graph's tables. 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"; + public static final String BACKEND_HBASE = "hbase"; + /** Backends on a remote cluster, whose availability the probe checks. */ + public static final Set<String> REMOTE_BACKENDS = ImmutableSet.of(BACKEND_HSTORE, + BACKEND_HBASE); + + private static final Logger LOG = Log.logger(StorageReadiness.class); + + /** A probe result with the time it was taken: the TTL check and the body come from one probe. */ + private static final class Cached { + + private final Map<String, Object> result; + private final long at; + + private Cached(Map<String, Object> result, long at) { + this.result = result; + this.at = at; + } + } + + private static final AtomicReference<Cached> LAST = new AtomicReference<>(); + private static final AtomicReference<CompletableFuture<Map<String, Object>>> IN_FLIGHT = + new AtomicReference<>(); + private static final AtomicInteger WAITERS = new AtomicInteger(); + + private StorageReadiness() { + } + + /** One storage probe with a time budget in ms. */ + public interface Probe { + + Map<String, Object> probe(long timeoutMs) throws Exception; + } + + public static Map<String, Object> check(GraphManager manager, long timeoutMs, + long cacheTtlMs, int maxWaiters) { + // 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<Map<String, Object>> holder = new ArrayList<>(1); + HugeGraphAuthProxy.runAsAdmin(() -> { + HugeGraph graph = firstRemoteGraph(manager); + if (graph == null) { + Map<String, Object> 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(graph.backend(), + t -> graph.metadata(null, STORAGE_READINESS_META, t), + timeoutMs, cacheTtlMs, maxWaiters)); + }); + return holder.get(0); + } + + public static Map<String, Object> check(Probe probe, long timeoutMs, long cacheTtlMs) { + return check(BACKEND_HSTORE, probe, timeoutMs, cacheTtlMs, Integer.MAX_VALUE); + } + + /** + * At most one probe runs at a time and every caller gets its answer: + * the first caller runs the probe on its own thread, concurrent callers + * wait for that result, each bounded by its own timeout, and at most + * {@code maxWaiters} of them wait at once: the rest get an immediate + * not-ready, so a burst of probes during slow storage cannot hold the + * REST worker pool (the endpoint is unauthenticated and outside the + * load-shedding filter). No monitor is held during the probe. + */ + public static Map<String, Object> check(String storage, Probe probe, long timeoutMs, + long cacheTtlMs, int maxWaiters) { + Cached last = LAST.get(); + if (last != null && System.currentTimeMillis() - last.at < cacheTtlMs) { + Map<String, Object> body = new LinkedHashMap<>(last.result); + body.put("cached", true); + return body; + } + CompletableFuture<Map<String, Object>> mine = new CompletableFuture<>(); + CompletableFuture<Map<String, Object>> running = IN_FLIGHT.get(); + if (running == null && IN_FLIGHT.compareAndSet(null, mine)) { + Map<String, Object> body; + try { + body = probeOnce(storage, probe, timeoutMs); + LAST.set(new Cached(body, System.currentTimeMillis())); + } finally { + IN_FLIGHT.set(null); + } + mine.complete(body); + return new LinkedHashMap<>(body); + } + if (running == null) { + running = IN_FLIGHT.get(); + } + if (running == null) { + // The owner finished between our two reads: its result is cached + return check(storage, probe, timeoutMs, cacheTtlMs, maxWaiters); + } + if (WAITERS.incrementAndGet() > maxWaiters) { + WAITERS.decrementAndGet(); + return notReady(storage, "too many readiness callers waiting for the probe (" + + maxWaiters + ")"); + } + try { + Map<String, Object> body = new LinkedHashMap<>( + running.get(timeoutMs, TimeUnit.MILLISECONDS)); + body.put("shared", true); + return body; + } catch (TimeoutException e) { + return notReady(storage, "a probe is still running after " + timeoutMs + " ms"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return notReady(storage, "interrupted"); + } catch (ExecutionException e) { + return notReady(storage, "probe failed: " + + e.getCause().getClass().getSimpleName()); + } finally { + WAITERS.decrementAndGet(); + } + } + + private static Map<String, Object> probeOnce(String storage, Probe probe, long timeoutMs) { + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", false); + body.put("storage", storage); + try { + Map<String, Object> 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); + return body; + } + + private static Map<String, Object> notReady(String storage, String reason) { + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", false); + body.put("storage", storage); + body.put("reason", reason); + body.put("cached", false); + return body; + } + + public static boolean isReady(Map<String, Object> body) { + return Boolean.TRUE.equals(body.get("ready")); + } + + public static void resetCache() { + LAST.set(null); + IN_FLIGHT.set(null); + WAITERS.set(0); + } + + private static HugeGraph firstRemoteGraph(GraphManager manager) { + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph != null && REMOTE_BACKENDS.contains(graph.backend())) { Review Comment: Right. Readiness now covers every distinct remote backend configuration: graphs with the same configuration (the same `pd.peers`, or the same `hbase.hosts` and namespace) share one probe, independent clusters are probed side by side within the common budget, and ready means all of them are ready. The body keeps the fields of the first probe, `storage` lists the backends, and a `probes` list carries one entry per configuration (graph, backend, ready, reason); on a failure `reason` names the configuration. Test `testEveryRemoteConfigurationIsProbed`: a healthy plus a failed configuration, a hung configuration bounded by the budget, a single configuration with the body unchanged. ########## hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java: ########## @@ -0,0 +1,376 @@ +/* + * 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. + * <p> + * Scope: ready means this server knows a Store list and reaches at least one + * Store over gRPC. A Store whose status RPC answers while its raft or + * partition path is broken is not detected. A read through the graph path + * was measured as the gate and rejected: a failed read invalidates the + * partition cache the queries share (with PD down the data plane then fails + * too), and one moving partition stalls every read through that client past + * the budget, so a rolling Store restart took every server out of the + * Service while the traffic was fine. + */ +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<String, ManagedChannel> CHANNELS = new ConcurrentHashMap<>(); + + private HstoreStorageProbe() { + } + + /** The active stores as PD sees them. */ + public interface StoreLister { + + List<Metapb.Store> 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<Metapb.Store> stores = Collections.emptyList(); + private volatile long at; + private volatile Boolean pdOk; + private volatile long pdAt; + private final AtomicReference<CompletableFuture<List<Metapb.Store>>> inFlight = + new AtomicReference<>(); + + public List<Metapb.Store> 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<Metapb.Store> 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<List<Metapb.Store>> refresh(StoreLister lister, + ExecutorService executor) { + CompletableFuture<List<Metapb.Store>> running = this.inFlight.get(); + if (running != null && !running.isDone()) { + return running; + } + CompletableFuture<List<Metapb.Store>> mine = new CompletableFuture<>(); + if (!this.inFlight.compareAndSet(running, mine)) { + return this.inFlight.get(); + } + executor.execute(() -> { + try { + List<Metapb.Store> 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<String, Object> result(boolean ready, String reason, int activeStores, + Long answeredStore, Boolean pdReachable, + long pdAgeMs, long storesAgeMs, long storeMillis) { + Map<String, Object> 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<String, Object> 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<Metapb.Store> stores = pd.getActiveStores(); + pruneChannels(CHANNELS, stores); Review Comment: Right. Channel shutdown is deferred: an address missing from a listing first goes into a stale set, its channel is shut down only when it is missing from two consecutive listings, and its return clears the state. A probe that snapshotted the previous list therefore still pings through open channels. `testChannelsOfReplacedStoresAreShutDown` rewritten for that scenario (the first listing keeps, a return forgets, two in a row shut down, an empty and a failed listing prune nothing). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
