bitflicker64 commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4051992606
########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/ReadinessAPI.java: ########## @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.api.profile; + +import java.util.Map; + +import org.apache.hugegraph.api.API; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.util.JsonUtil; + +import com.codahale.metrics.annotation.Timed; + +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.annotation.security.PermitAll; +import jakarta.inject.Singleton; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.core.Context; +import jakarta.ws.rs.core.Response; + +/** + * Storage-aware readiness for Kubernetes and load balancers: 200 while this + * server can serve graph traffic, 503 while PD or every Store is unreachable Review Comment: 🧹 This Javadoc says 503 while PD is unreachable. So do the `StorageReadiness` Javadoc ("PD answers and one active Store answers"), the `HstoreStorageProbe` class Javadoc ("a hung PD or Store turns into not ready") and the `readiness.timeout` description. The probe deliberately stays ready with PD down once a Store list is known (`testKnownStoresKeepTheServerReadyWhilePdIsDown`). Both option descriptions also say one Store call per probe, while it pings every known Store in parallel. Please reword these to match: PD only matters until the first Store list is known, and the Store cost is one call per known Store. ########## hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java: ########## @@ -0,0 +1,348 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.backend.store.hstore; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.store.grpc.state.HgStoreStateGrpc; +import org.apache.hugegraph.store.grpc.state.SubStateReq; +import org.apache.hugegraph.util.E; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; + +/** + * Storage-aware readiness of this server, from this server's point of view: + * at least one Store answers a direct, local, read-only gRPC call + * (HgStoreState.getScanState, which reads the node's own scan-pool stats and + * never touches raft). The Store list comes from PD, but PD is refreshed in + * the background and the last known list is used right away, so a PD that is + * slow, restarting or down does not change the readiness of a server whose + * Stores still answer. Every wait is bounded by one shared time budget, so a + * hung PD or Store turns into "not ready" instead of a hung probe. The result + * carries no addresses, since it is served without authentication. + */ +public final class HstoreStorageProbe { + + public static final String META_STORAGE_READINESS = "storage_readiness"; + + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( + new ThreadFactory() { + private final AtomicInteger seq = new AtomicInteger(); + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "storage-readiness-" + this.seq.incrementAndGet()); + t.setDaemon(true); + return t; + } + }); + + private static final KnownStores KNOWN = new KnownStores(); + private static final Map<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; + + 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(); + } + } + + public static final class Result { + + private final boolean ready; + private final String reason; + private final int activeStores; + private final Long answeredStore; + private final Boolean pdReachable; + private final long pdAgeMs; + private final long storesAgeMs; + private final long storeMillis; + + Result(boolean ready, String reason, int activeStores, Long answeredStore, + Boolean pdReachable, long pdAgeMs, long storesAgeMs, long storeMillis) { + this.ready = ready; + this.reason = reason; + this.activeStores = activeStores; + this.answeredStore = answeredStore; + this.pdReachable = pdReachable; + this.pdAgeMs = pdAgeMs; + this.storesAgeMs = storesAgeMs; + this.storeMillis = storeMillis; + } + + public boolean ready() { + return this.ready; + } + + public String reason() { + return this.reason; + } + + public int activeStores() { + return this.activeStores; + } + + public Long answeredStore() { + return this.answeredStore; + } + + public Boolean pdReachable() { + return this.pdReachable; + } + + public Map<String, Object> toMap() { + Map<String, Object> map = new LinkedHashMap<>(); + map.put("ready", this.ready); + map.put("reason", this.reason); + map.put("active_stores", this.activeStores); + map.put("answered_store", this.answeredStore); + map.put("pd_reachable", this.pdReachable); + map.put("pd_checked_age_ms", this.pdAgeMs); + map.put("stores_age_ms", this.storesAgeMs); + map.put("store_millis", this.storeMillis); + return map; + } + } + + /** + * Probe through the process-wide PD client and this probe's own plaintext + * channels to the stores (the store gRPC server takes no credentials). + * + * @param graphName the store-side graph name, kept for the meta handler + * @param timeoutMs the whole budget for PD plus stores + */ + public static Map<String, Object> probe(String graphName, long timeoutMs) { + PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); + if (pd == null) { + return new Result(false, "pd client not initialised", 0, null, + false, -1L, -1L, 0L).toMap(); + } + return probe(KNOWN, pd::getActiveStores, HstoreStorageProbe::pingScanState, + timeoutMs, EXECUTOR).toMap(); + } + + private static void pingScanState(Metapb.Store store, long timeoutMs) { + ManagedChannel channel = CHANNELS.computeIfAbsent(store.getAddress(), address -> { + return ManagedChannelBuilder.forTarget(address).usePlaintext().build(); + }); + HgStoreStateGrpc.newBlockingStub(channel) + .withDeadlineAfter(timeoutMs, TimeUnit.MILLISECONDS) + .getScanState(SubStateReq.getDefaultInstance()); + } + + public static Result probe(KnownStores known, StoreLister lister, StorePinger pinger, + long timeoutMs, ExecutorService executor) { + E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); + long deadline = System.currentTimeMillis() + timeoutMs; + + // Refresh the store list from PD in the background; whatever PD + // answers lands in `known` for this or the next probe + CompletableFuture<List<Metapb.Store>> refresh = new CompletableFuture<>(); + executor.execute(() -> { Review Comment: ⚠️ Every cache-missed probe submits a new PD refresh here, even while the previous one is still running, and `EXECUTOR` is an unbounded cached pool. `getActiveStores()` is bounded only by the PD client deadline, 60 s by default (`PDConfig.grpcTimeOut`; `HstoreSessionsImpl.initStoreNode` does not override it). If PD hangs instead of refusing connections, the probe still returns in milliseconds from the known Stores, so each probe parks one more thread on PD: up to about 30 at the default 2 s TTL, and one per request with `readiness.cache_ttl=0`, which the option allows, on an endpoint anyone can call. Without a live stub they also queue on the `synchronized` `PDClient.newBlockingStub()` alongside graph traffic. Could the refresh be single-flight, e.g. keep the in-flight `CompletableFuture` and skip submitting until it completes? ########## hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java: ########## @@ -0,0 +1,348 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.backend.store.hstore; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.store.grpc.state.HgStoreStateGrpc; +import org.apache.hugegraph.store.grpc.state.SubStateReq; +import org.apache.hugegraph.util.E; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; + +/** + * Storage-aware readiness of this server, from this server's point of view: + * at least one Store answers a direct, local, read-only gRPC call + * (HgStoreState.getScanState, which reads the node's own scan-pool stats and + * never touches raft). The Store list comes from PD, but PD is refreshed in + * the background and the last known list is used right away, so a PD that is + * slow, restarting or down does not change the readiness of a server whose + * Stores still answer. Every wait is bounded by one shared time budget, so a + * hung PD or Store turns into "not ready" instead of a hung probe. The result + * carries no addresses, since it is served without authentication. + */ +public final class HstoreStorageProbe { + + public static final String META_STORAGE_READINESS = "storage_readiness"; + + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( + new ThreadFactory() { + private final AtomicInteger seq = new AtomicInteger(); + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "storage-readiness-" + this.seq.incrementAndGet()); + t.setDaemon(true); + return t; + } + }); + + private static final KnownStores KNOWN = new KnownStores(); + private static final Map<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; + + 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(); + } + } + + public static final class Result { + + private final boolean ready; + private final String reason; + private final int activeStores; + private final Long answeredStore; + private final Boolean pdReachable; + private final long pdAgeMs; + private final long storesAgeMs; + private final long storeMillis; + + Result(boolean ready, String reason, int activeStores, Long answeredStore, + Boolean pdReachable, long pdAgeMs, long storesAgeMs, long storeMillis) { + this.ready = ready; + this.reason = reason; + this.activeStores = activeStores; + this.answeredStore = answeredStore; + this.pdReachable = pdReachable; + this.pdAgeMs = pdAgeMs; + this.storesAgeMs = storesAgeMs; + this.storeMillis = storeMillis; + } + + public boolean ready() { + return this.ready; + } + + public String reason() { + return this.reason; + } + + public int activeStores() { + return this.activeStores; + } + + public Long answeredStore() { + return this.answeredStore; + } + + public Boolean pdReachable() { + return this.pdReachable; + } + + public Map<String, Object> toMap() { + Map<String, Object> map = new LinkedHashMap<>(); + map.put("ready", this.ready); + map.put("reason", this.reason); + map.put("active_stores", this.activeStores); + map.put("answered_store", this.answeredStore); + map.put("pd_reachable", this.pdReachable); + map.put("pd_checked_age_ms", this.pdAgeMs); + map.put("stores_age_ms", this.storesAgeMs); + map.put("store_millis", this.storeMillis); + return map; + } + } + + /** + * Probe through the process-wide PD client and this probe's own plaintext + * channels to the stores (the store gRPC server takes no credentials). + * + * @param graphName the store-side graph name, kept for the meta handler + * @param timeoutMs the whole budget for PD plus stores + */ + public static Map<String, Object> probe(String graphName, long timeoutMs) { + PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); + if (pd == null) { + return new Result(false, "pd client not initialised", 0, null, + false, -1L, -1L, 0L).toMap(); + } + return probe(KNOWN, pd::getActiveStores, HstoreStorageProbe::pingScanState, + timeoutMs, EXECUTOR).toMap(); + } + + private static void pingScanState(Metapb.Store store, long timeoutMs) { + ManagedChannel channel = CHANNELS.computeIfAbsent(store.getAddress(), address -> { + return ManagedChannelBuilder.forTarget(address).usePlaintext().build(); + }); + HgStoreStateGrpc.newBlockingStub(channel) + .withDeadlineAfter(timeoutMs, TimeUnit.MILLISECONDS) + .getScanState(SubStateReq.getDefaultInstance()); + } + + public static Result probe(KnownStores known, StoreLister lister, StorePinger pinger, + long timeoutMs, ExecutorService executor) { + E.checkArgument(timeoutMs > 0, "The probe timeout must be > 0, but got %s", timeoutMs); + long deadline = System.currentTimeMillis() + timeoutMs; + + // Refresh the store list from PD in the background; whatever PD + // answers lands in `known` for this or the next probe + CompletableFuture<List<Metapb.Store>> refresh = new CompletableFuture<>(); + executor.execute(() -> { + try { + List<Metapb.Store> stores = lister.activeStores(); + known.update(stores); + refresh.complete(stores == null ? Collections.emptyList() : stores); + } catch (Throwable e) { + known.pdFailed(); + refresh.completeExceptionally(e); + } + }); + + 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 new Result(false, "no store list known and pd did not answer within " + + timeoutMs + " ms", 0, null, false, -1L, -1L, 0L); + } catch (Exception e) { + return new Result(false, "no store list known and pd failed: " + message(e), + 0, null, false, known.pdAgeMs(), -1L, 0L); + } + if (stores.isEmpty()) { + return new Result(false, "no active store registered in pd", + 0, null, true, known.pdAgeMs(), known.ageMs(), 0L); + } + } + + long storeStart = System.currentTimeMillis(); + // Ping every known store at once and take the first answer: a store + // whose connection hangs (a pod that just went away) must not eat the + // budget of the stores that are fine, or a rolling restart would + // flap the readiness of every server + CompletionService<Metapb.Store> pings = new ExecutorCompletionService<>(executor); + List<Future<Metapb.Store>> 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<String> failures = new ArrayList<>(); + Result result = null; + try { + for (int done = 0; done < stores.size() && result == null; done++) { + long remaining = deadline - System.currentTimeMillis(); + Future<Metapb.Store> first; + try { + first = remaining > 0 ? + pings.poll(remaining, TimeUnit.MILLISECONDS) : + pings.poll(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + failures.add("interrupted"); + break; + } + if (first == null) { + failures.add((stores.size() - done) + " store(s) did not answer within " + + timeoutMs + " ms"); + break; + } + try { + Metapb.Store store = first.get(); + result = new Result(true, "ok", stores.size(), store.getId(), + pdState(refresh, known, pdReachable), known.pdAgeMs(), + known.ageMs(), elapsed(storeStart)); + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + failures.add("a store failed: " + message(cause)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + failures.add("interrupted"); + break; + } + } + } finally { + for (Future<Metapb.Store> f : futures) { + f.cancel(true); + } + } + if (result != null) { + return result; + } + return new Result(false, "none of " + stores.size() + " known store(s) answered: " + + String.join("; ", failures), + stores.size(), null, pdState(refresh, known, pdReachable), + known.pdAgeMs(), known.ageMs(), elapsed(storeStart)); + } + + /** + * The outcome of this probe's PD refresh when it already finished, else + * the outcome of the last finished one (null before any finished). + */ + private static Boolean pdState(CompletableFuture<?> refresh, KnownStores known, + Boolean awaited) { + if (awaited != null) { + return awaited; + } + if (refresh.isDone()) { + return !refresh.isCompletedExceptionally(); + } + return known.pdOk(); + } + + private static <T> T await(Future<T> future, long deadline) throws Exception { + long remaining = Math.max(1L, deadline - System.currentTimeMillis()); + try { + return future.get(remaining, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + throw cause instanceof Exception ? (Exception) cause : new RuntimeException(cause); + } + } + + private static long elapsed(long since) { + return System.currentTimeMillis() - since; + } + + private static String message(Throwable e) { Review Comment: ⚠️ `/readiness` is unauthenticated and the PR says the body carries no addresses, but `reason` embeds raw exception messages. Two paths: - No Store list known yet and the PD client has no live stub (never connected, or reset by `closeStub` after a watch error) with peers refusing fast: `await()` unwraps the `PDException` from `newBlockingStub()`, whose message is `"PD unreachable, pd.peers=" + config.getServerHost()` (PDClient.java:142-143), so every PD peer is listed. - Every known Store fails and one failed name resolution: `UNAVAILABLE: Unable to resolve host <host>` (grpc-core 1.47 `DnsNameResolver`), if Stores register by DNS name. `StorageReadiness.check` also appends `e.getMessage()`; `testMapCarriesNoAddresses` covers only the ready path. Could `reason` use a fixed category (`pd unreachable`, the gRPC status code or exception class) and log the full message instead? ########## hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreStorageProbe.java: ########## @@ -0,0 +1,348 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.backend.store.hstore; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.store.grpc.state.HgStoreStateGrpc; +import org.apache.hugegraph.store.grpc.state.SubStateReq; +import org.apache.hugegraph.util.E; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; + +/** + * Storage-aware readiness of this server, from this server's point of view: + * at least one Store answers a direct, local, read-only gRPC call + * (HgStoreState.getScanState, which reads the node's own scan-pool stats and + * never touches raft). The Store list comes from PD, but PD is refreshed in + * the background and the last known list is used right away, so a PD that is + * slow, restarting or down does not change the readiness of a server whose + * Stores still answer. Every wait is bounded by one shared time budget, so a + * hung PD or Store turns into "not ready" instead of a hung probe. The result + * carries no addresses, since it is served without authentication. + */ +public final class HstoreStorageProbe { + + public static final String META_STORAGE_READINESS = "storage_readiness"; + + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool( + new ThreadFactory() { + private final AtomicInteger seq = new AtomicInteger(); + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, "storage-readiness-" + this.seq.incrementAndGet()); + t.setDaemon(true); + return t; + } + }); + + private static final KnownStores KNOWN = new KnownStores(); + private static final Map<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; + + 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(); + } + } + + public static final class Result { + + private final boolean ready; + private final String reason; + private final int activeStores; + private final Long answeredStore; + private final Boolean pdReachable; + private final long pdAgeMs; + private final long storesAgeMs; + private final long storeMillis; + + Result(boolean ready, String reason, int activeStores, Long answeredStore, + Boolean pdReachable, long pdAgeMs, long storesAgeMs, long storeMillis) { + this.ready = ready; + this.reason = reason; + this.activeStores = activeStores; + this.answeredStore = answeredStore; + this.pdReachable = pdReachable; + this.pdAgeMs = pdAgeMs; + this.storesAgeMs = storesAgeMs; + this.storeMillis = storeMillis; + } + + public boolean ready() { + return this.ready; + } + + public String reason() { + return this.reason; + } + + public int activeStores() { + return this.activeStores; + } + + public Long answeredStore() { + return this.answeredStore; + } + + public Boolean pdReachable() { + return this.pdReachable; + } + + public Map<String, Object> toMap() { + Map<String, Object> map = new LinkedHashMap<>(); + map.put("ready", this.ready); + map.put("reason", this.reason); + map.put("active_stores", this.activeStores); + map.put("answered_store", this.answeredStore); + map.put("pd_reachable", this.pdReachable); + map.put("pd_checked_age_ms", this.pdAgeMs); + map.put("stores_age_ms", this.storesAgeMs); + map.put("store_millis", this.storeMillis); + return map; + } + } + + /** + * Probe through the process-wide PD client and this probe's own plaintext + * channels to the stores (the store gRPC server takes no credentials). + * + * @param graphName the store-side graph name, kept for the meta handler + * @param timeoutMs the whole budget for PD plus stores + */ + public static Map<String, Object> probe(String graphName, long timeoutMs) { + PDClient pd = HstoreSessionsImpl.getDefaultPdClient(); + if (pd == null) { + return new Result(false, "pd client not initialised", 0, null, + false, -1L, -1L, 0L).toMap(); + } + return probe(KNOWN, pd::getActiveStores, HstoreStorageProbe::pingScanState, + timeoutMs, EXECUTOR).toMap(); + } + + private static void pingScanState(Metapb.Store store, long timeoutMs) { + ManagedChannel channel = CHANNELS.computeIfAbsent(store.getAddress(), address -> { Review Comment: 🧹 `CHANNELS` keeps one plaintext channel per Store address and nothing ever removes or shuts one down. When a Store is replaced or comes back under a different address, the old channel stays allocated for the life of the process. Could channels whose address is no longer in the latest PD list be shut down after a successful refresh in `probe`? -- 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]
