SebastianGruza commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4183134525
########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java: ########## @@ -0,0 +1,342 @@ +/* + * 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.HashSet; +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.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +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(); + /** Probes of independent backend configurations run side by side here. */ + private static final ExecutorService PROBES = + Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "storage-readiness-probe"); + t.setDaemon(true); + return t; + }); + + 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 graphs up and probe them 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(() -> { + List<RemoteGraph> remotes = remoteGraphs(manager); + if (remotes.isEmpty()) { + 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; + } + String storage = remotes.stream().map(r -> r.backend).distinct() + .collect(Collectors.joining(",")); + holder.add(check(storage, t -> probeAll(remotes, t), timeoutMs, cacheTtlMs, maxWaiters)); + }); + return holder.get(0); + } + + /** One graph per distinct remote backend configuration, in graph order; public for the unit test. */ + public static final class RemoteGraph { + + final String name; + final String backend; + final Probe probe; + + public RemoteGraph(String name, String backend, Probe probe) { + this.name = name; + this.backend = backend; + this.probe = probe; + } + } + + /** + * Readiness covers every remote backend configuration this server + * uses: graphs that share a configuration (the same PD peers, 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. + */ + static List<RemoteGraph> remoteGraphs(GraphManager manager) { + List<RemoteGraph> out = new ArrayList<>(); + Set<String> seen = new HashSet<>(); + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph == null || !REMOTE_BACKENDS.contains(graph.backend())) { + continue; + } + String key = graph.backend() + "|" + configKey(graph); + if (!seen.add(key)) { + continue; + } + out.add(new RemoteGraph(name, graph.backend(), + t -> graph.metadata(null, STORAGE_READINESS_META, t))); + } catch (Throwable e) { + LOG.debug("Skip graph {} while looking for remote storages", name, e); + } + } + return out; + } + + private static String configKey(HugeGraph graph) { + try { + if (BACKEND_HSTORE.equals(graph.backend())) { + return String.valueOf(graph.configuration().getString("pd.peers")); + } + return graph.configuration().getString("hbase.hosts") + "/" + + graph.configuration().getString("hbase.namespace"); + } catch (Throwable e) { + return graph.name(); + } + } + + /** Every configuration's probe in parallel on the shared budget; the body merges them. */ + public static Map<String, Object> probeAll(List<RemoteGraph> remotes, long timeoutMs) + throws Exception { + if (remotes.size() == 1) { + Map<String, Object> body = new LinkedHashMap<>(remotes.get(0).probe.probe(timeoutMs)); + body.put("probes", List.of(probeEntry(remotes.get(0), body))); + return body; + } + long deadline = System.currentTimeMillis() + timeoutMs; + List<CompletableFuture<Map<String, Object>>> futures = new ArrayList<>(); + for (RemoteGraph r : remotes) { + futures.add(CompletableFuture.supplyAsync(() -> { Review Comment: Right. Every probe on the pool runs inside `HugeGraphAuthProxy.runAsAdmin()` on its own thread. Test `testAsyncProbesCarryTheAdminContext`: three configurations, every probe sees the admin user's context. ########## hugegraph-server/hugegraph-hbase/src/main/java/org/apache/hugegraph/backend/store/hbase/HbaseStore.java: ########## @@ -87,6 +93,61 @@ 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 can serve this graph: one admin round trip + * (is the graph's first table enabled and available) 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) { + List<String> tables = this.tableNames(); + String table = tables.isEmpty() ? null : tables.get(0); Review Comment: Right. The probe now checks every table of the graph (vertices, edges, indexes, counters) one after another within the budget: `firstUnavailable(tables, check)` returns the number of the first table that is not enabled and available, and the reason reads "table N of the graph is not available". Test: four tables with the third disabled is not ready; an empty table list is not ready; a failing admin call surfaces as a failure. ########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java: ########## @@ -0,0 +1,342 @@ +/* + * 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.HashSet; +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.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +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(); + /** Probes of independent backend configurations run side by side here. */ + private static final ExecutorService PROBES = + Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "storage-readiness-probe"); + t.setDaemon(true); + return t; + }); + + 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 graphs up and probe them 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(() -> { + List<RemoteGraph> remotes = remoteGraphs(manager); + if (remotes.isEmpty()) { + 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; + } + String storage = remotes.stream().map(r -> r.backend).distinct() + .collect(Collectors.joining(",")); + holder.add(check(storage, t -> probeAll(remotes, t), timeoutMs, cacheTtlMs, maxWaiters)); + }); + return holder.get(0); + } + + /** One graph per distinct remote backend configuration, in graph order; public for the unit test. */ + public static final class RemoteGraph { + + final String name; + final String backend; + final Probe probe; + + public RemoteGraph(String name, String backend, Probe probe) { + this.name = name; + this.backend = backend; + this.probe = probe; + } + } + + /** + * Readiness covers every remote backend configuration this server + * uses: graphs that share a configuration (the same PD peers, 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. + */ + static List<RemoteGraph> remoteGraphs(GraphManager manager) { + List<RemoteGraph> out = new ArrayList<>(); + Set<String> seen = new HashSet<>(); + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph == null || !REMOTE_BACKENDS.contains(graph.backend())) { + continue; + } + String key = graph.backend() + "|" + configKey(graph); + if (!seen.add(key)) { + continue; + } + out.add(new RemoteGraph(name, graph.backend(), + t -> graph.metadata(null, STORAGE_READINESS_META, t))); + } catch (Throwable e) { + LOG.debug("Skip graph {} while looking for remote storages", name, e); + } + } + return out; + } + + private static String configKey(HugeGraph graph) { + try { + if (BACKEND_HSTORE.equals(graph.backend())) { + return String.valueOf(graph.configuration().getString("pd.peers")); + } + return graph.configuration().getString("hbase.hosts") + "/" + + graph.configuration().getString("hbase.namespace"); + } catch (Throwable e) { + return graph.name(); + } + } + + /** Every configuration's probe in parallel on the shared budget; the body merges them. */ + public static Map<String, Object> probeAll(List<RemoteGraph> remotes, long timeoutMs) + throws Exception { + if (remotes.size() == 1) { + Map<String, Object> body = new LinkedHashMap<>(remotes.get(0).probe.probe(timeoutMs)); + body.put("probes", List.of(probeEntry(remotes.get(0), body))); + return body; + } + long deadline = System.currentTimeMillis() + timeoutMs; + List<CompletableFuture<Map<String, Object>>> futures = new ArrayList<>(); + for (RemoteGraph r : remotes) { + futures.add(CompletableFuture.supplyAsync(() -> { + try { + return r.probe.probe(timeoutMs); + } catch (Exception e) { + Map<String, Object> failed = new LinkedHashMap<>(); + failed.put("ready", false); + failed.put("reason", "probe failed: " + e.getClass().getSimpleName()); + return failed; + } + }, PROBES)); + } + List<Map<String, Object>> entries = new ArrayList<>(); + Map<String, Object> first = null; + boolean ready = true; + String reason = "ok"; + for (int i = 0; i < remotes.size(); i++) { + Map<String, Object> result; + try { + long remaining = Math.max(1L, deadline - System.currentTimeMillis()); + result = futures.get(i).get(remaining, TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + futures.get(i).cancel(true); + result = new LinkedHashMap<>(); + result.put("ready", false); + result.put("reason", "did not answer within " + timeoutMs + " ms"); + } + if (first == null) { + first = result; + } + Map<String, Object> entry = probeEntry(remotes.get(i), result); + entries.add(entry); + if (!isReady(result)) { + if (ready) { + reason = remotes.get(i).backend + " of graph " + remotes.get(i).name + ": " + + result.get("reason"); + } + ready = false; + } + } + Map<String, Object> body = new LinkedHashMap<>(first); + body.put("ready", ready); + body.put("reason", reason); + body.put("probes", entries); + return body; + } + + private static Map<String, Object> probeEntry(RemoteGraph r, Map<String, Object> result) { + Map<String, Object> entry = new LinkedHashMap<>(); + entry.put("graph", r.name); Review Comment: Right. The `probes` entries carry the backend and the outcome only (no graph name), and a failure reason reads "hbase configuration 2 of 2: ...". The unit test asserts the body contains no graph name, and `ReadinessApiTest` asserts there is no `graph` field. ########## hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java: ########## @@ -0,0 +1,384 @@ +/* + * 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.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +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 static final Set<String> STALE_CHANNELS = ConcurrentHashMap.newKeySet(); + + 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, STALE_CHANNELS); + return stores; + }, HstoreStorageProbe::pingScanState, timeoutMs, EXECUTOR); + } + + /** + * Channels of stores that left the PD list are shut down only when they + * are absent from two consecutive listings: a probe that snapshotted the + * previous store list may still be pinging them (and a stale-list ping + * through a closed channel would be a false not-ready, cached for the + * TTL). `stale` remembers the addresses missing from the last listing. + */ + static void pruneChannels(Map<String, ManagedChannel> channels, + List<Metapb.Store> stores, Set<String> stale) { + if (stores == null || stores.isEmpty()) { + return; + } + Set<String> live = new HashSet<>(); + for (Metapb.Store store : stores) { + live.add(store.getAddress()); + } + stale.retainAll(channels.keySet()); + channels.entrySet().removeIf(e -> { + if (live.contains(e.getKey())) { + stale.remove(e.getKey()); + return false; + } + if (stale.add(e.getKey())) { + return false; // first listing without it: keep for the probes in flight + } + e.getValue().shutdownNow(); + stale.remove(e.getKey()); + 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<String, Object> 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<List<Metapb.Store>> refresh = known.refresh(lister, executor); + + List<Metapb.Store> 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) { Review Comment: Right: a separate catch, the interrupt flag restored, and the reason "interrupted while waiting for the store list". ########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java: ########## @@ -0,0 +1,342 @@ +/* + * 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.HashSet; +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.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +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(); + /** Probes of independent backend configurations run side by side here. */ + private static final ExecutorService PROBES = + Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "storage-readiness-probe"); + t.setDaemon(true); + return t; + }); + + 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 graphs up and probe them 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(() -> { + List<RemoteGraph> remotes = remoteGraphs(manager); + if (remotes.isEmpty()) { + 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; + } + String storage = remotes.stream().map(r -> r.backend).distinct() + .collect(Collectors.joining(",")); + holder.add(check(storage, t -> probeAll(remotes, t), timeoutMs, cacheTtlMs, maxWaiters)); + }); + return holder.get(0); + } + + /** One graph per distinct remote backend configuration, in graph order; public for the unit test. */ + public static final class RemoteGraph { + + final String name; + final String backend; + final Probe probe; + + public RemoteGraph(String name, String backend, Probe probe) { + this.name = name; + this.backend = backend; + this.probe = probe; + } + } + + /** + * Readiness covers every remote backend configuration this server + * uses: graphs that share a configuration (the same PD peers, 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. + */ + static List<RemoteGraph> remoteGraphs(GraphManager manager) { + List<RemoteGraph> out = new ArrayList<>(); + Set<String> seen = new HashSet<>(); + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph == null || !REMOTE_BACKENDS.contains(graph.backend())) { + continue; + } + String key = graph.backend() + "|" + configKey(graph); + if (!seen.add(key)) { + continue; + } + out.add(new RemoteGraph(name, graph.backend(), + t -> graph.metadata(null, STORAGE_READINESS_META, t))); Review Comment: Right, thanks for the reproduction on memory. The probe is now `probeOf(graph)`: `metadata()` in a try, `graph.tx().close()` in the finally when the transaction is open; this covers the request thread and the pool threads alike. -- 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]
