SebastianGruza commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4052318708
########## 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: Done in c9e70b14, thanks, both paths were real: with no Store list yet the `PDException` from `newBlockingStub()` carried the whole `pd.peers`, and with every Store down gRPC carried `Unable to resolve host …` (it was in my own Stores→0 samples, I had not connected the dots). `reason` now carries a fixed category only: the gRPC status code for `StatusRuntimeException`, `pd unreachable` for `PDException`, the class name otherwise (`HstoreStorageProbe.category()`); the full messages go to the log (WARN for PD, DEBUG for Store pings). `StorageReadiness.check()` likewise: class name only, message to the log. `testReasonCarriesNoPdPeersNorStoreHosts` feeds a `PDException` with `pd.peers=pd-0.internal:8686,…` and an `UNAVAILABLE: Unable to resolve host store-0…svc` and asserts neither map contains `internal`, `svc` or `8686`; `StorageReadinessTest` checks the same for a probe exception. On k3s with this build the reasons read `none of 3 known store(s) answered: a store failed: U NAVAILABLE; a store failed: UNAVAILABLE; a store failed: UNAVAILABLE`, and a grep of every sampled body from `stores-zero` and `pd-zero` for `.svc`, `8686` and `hugegraph-store-` finds 0 occurrences. ########## 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: Done in c9e70b14. The refresh is single-flight: `KnownStores` keeps an `AtomicReference` to the in-flight `CompletableFuture`, `refresh()` returns the running one while it is not `isDone()` and only starts a new one after it finished (CAS on the reference, so two racing probes also share one). A hung PD therefore parks one thread per process regardless of `readiness.cache_ttl`, including 0. `testRefreshIsSingleFlight`: a lister sleeping 3 s, five probes in a row → one call, all five ready from the known list; after it completes the next probe starts the second. The queueing on the `synchronized` `newBlockingStub()` next to graph traffic is thereby bounded to that one thread as well. ########## 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: Done in c9e70b14. The Javadoc of `ReadinessAPI`, `StorageReadiness` and `HstoreStorageProbe` and the descriptions of `readiness.timeout` and `readiness.cache_ttl` now say what the code does: PD only matters until the first Store list is known, afterwards it is refreshed in the background; the cost of a probe is one cheap call to every known Store in parallel, first answer wins; 503 only when none answers within the budget (or, before the first list, when PD does not answer). ########## 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: Done in c9e70b14. After every successful listing from PD, `pruneChannels()` shuts down (`shutdownNow`) and removes the channels of addresses no longer listed; a failed listing touches nothing. Wired into the lister lambda of `probe(graphName, timeoutMs)`, so the pure `probe(...)` stays stateless. `testChannelsOfReplacedStoresAreShutDown` uses a small `ManagedChannel` subclass (the hstore module has no Mockito): the channel of a listed address stays open, the replaced one is shut down and removed, a `null` listing removes 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]
