imbajin commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4081389643
########## 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<String, Object> lastResult; + private static volatile long lastCheckedAt; + + 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) { + // 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 = firstHstoreGraph(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(t -> graph.metadata(null, STORAGE_READINESS_META, t), + timeoutMs, cacheTtlMs)); + }); + return holder.get(0); + } + + public static synchronized Map<String, Object> check(Probe probe, long timeoutMs, Review Comment: ⚠️ This synchronized method holds the class monitor while the Store probe performs network I/O for up to readiness.timeout. With the supported readiness.cache_ttl=0 setting, concurrent requests each trigger another probe and queue behind the previous call, so request latency can exceed the configured timeout by multiple probes and occupy REST workers. Please share an in-flight probe or bound each caller's wait without holding the monitor during network I/O. ########## 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<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); + 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<String, ManagedChannel> channels, + List<Metapb.Store> stores) { + if (stores == null || stores.isEmpty()) { + return; + } + Set<String> 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()); Review Comment: ⚠️ A successful getScanState response is treated as proof that this Server can serve graph traffic, but this RPC only reads local scan-pool statistics and never checks Raft or graph-data access. A Store whose status RPC remains healthy while its Raft or partition path is unavailable will still make /readiness return 200, so Kubernetes can keep routing graph requests to an unusable instance. Please validate the graph-serving path, or use a Store health signal that covers it, before reporting ready. -- 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]
