imbajin commented on code in PR #3221: URL: https://github.com/apache/hugegraph/pull/3221#discussion_r4158750802
########## hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/StorageReadiness.java: ########## @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.api.profile; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.auth.HugeGraphAuthProxy; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; + +import com.google.common.collect.ImmutableSet; + +/** + * Whether this server can serve graph traffic, for a readiness probe. + * Graphs on an embedded backend are ready as soon as the REST layer answers. + * Graphs on a remote backend are probed through the backend's + * "storage_readiness" metadata: on hstore at least one known Store answers a + * cheap direct call (PD is only needed until the first Store list is known), + * on hbase the cluster answers an admin call about one of the graph's tables. The storage is shared by every + * hstore graph of the process, so one graph is probed and the result is + * reused for a short TTL to keep repeated probes cheap. The body never + * carries raw exception text, since the endpoint is unauthenticated. + */ +public final class StorageReadiness { + + public static final String STORAGE_READINESS_META = "storage_readiness"; + public static final String BACKEND_HSTORE = "hstore"; + public static final String BACKEND_HBASE = "hbase"; + /** Backends on a remote cluster, whose availability the probe checks. */ + public static final Set<String> REMOTE_BACKENDS = ImmutableSet.of(BACKEND_HSTORE, + BACKEND_HBASE); + + private static final Logger LOG = Log.logger(StorageReadiness.class); + + /** A probe result with the time it was taken: the TTL check and the body come from one probe. */ + private static final class Cached { + + private final Map<String, Object> result; + private final long at; + + private Cached(Map<String, Object> result, long at) { + this.result = result; + this.at = at; + } + } + + private static final AtomicReference<Cached> LAST = new AtomicReference<>(); + private static final AtomicReference<CompletableFuture<Map<String, Object>>> IN_FLIGHT = + new AtomicReference<>(); + private static final AtomicInteger WAITERS = new AtomicInteger(); + + private StorageReadiness() { + } + + /** One storage probe with a time budget in ms. */ + public interface Probe { + + Map<String, Object> probe(long timeoutMs) throws Exception; + } + + public static Map<String, Object> check(GraphManager manager, long timeoutMs, + long cacheTtlMs, int maxWaiters) { + // The graphs are auth proxies and the probe request carries no user, + // so look the graph up and probe it as the internal admin, the way + // other internal paths do; the result carries no data or addresses + List<Map<String, Object>> holder = new ArrayList<>(1); + HugeGraphAuthProxy.runAsAdmin(() -> { + HugeGraph graph = firstRemoteGraph(manager); + if (graph == null) { + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", true); + body.put("storage", "embedded"); + body.put("reason", "no graph on a remote storage"); + holder.add(body); + return; + } + holder.add(check(graph.backend(), + t -> graph.metadata(null, STORAGE_READINESS_META, t), + timeoutMs, cacheTtlMs, maxWaiters)); + }); + return holder.get(0); + } + + public static Map<String, Object> check(Probe probe, long timeoutMs, long cacheTtlMs) { + return check(BACKEND_HSTORE, probe, timeoutMs, cacheTtlMs, Integer.MAX_VALUE); + } + + /** + * At most one probe runs at a time and every caller gets its answer: + * the first caller runs the probe on its own thread, concurrent callers + * wait for that result, each bounded by its own timeout, and at most + * {@code maxWaiters} of them wait at once: the rest get an immediate + * not-ready, so a burst of probes during slow storage cannot hold the + * REST worker pool (the endpoint is unauthenticated and outside the + * load-shedding filter). No monitor is held during the probe. + */ + public static Map<String, Object> check(String storage, Probe probe, long timeoutMs, + long cacheTtlMs, int maxWaiters) { + Cached last = LAST.get(); + if (last != null && System.currentTimeMillis() - last.at < cacheTtlMs) { + Map<String, Object> body = new LinkedHashMap<>(last.result); + body.put("cached", true); + return body; + } + CompletableFuture<Map<String, Object>> mine = new CompletableFuture<>(); + CompletableFuture<Map<String, Object>> running = IN_FLIGHT.get(); + if (running == null && IN_FLIGHT.compareAndSet(null, mine)) { + Map<String, Object> body; + try { + body = probeOnce(storage, probe, timeoutMs); + LAST.set(new Cached(body, System.currentTimeMillis())); + } finally { + IN_FLIGHT.set(null); + } + mine.complete(body); + return new LinkedHashMap<>(body); + } + if (running == null) { + running = IN_FLIGHT.get(); + } + if (running == null) { + // The owner finished between our two reads: its result is cached + return check(storage, probe, timeoutMs, cacheTtlMs, maxWaiters); + } + if (WAITERS.incrementAndGet() > maxWaiters) { + WAITERS.decrementAndGet(); + return notReady(storage, "too many readiness callers waiting for the probe (" + + maxWaiters + ")"); + } + try { + Map<String, Object> body = new LinkedHashMap<>( + running.get(timeoutMs, TimeUnit.MILLISECONDS)); + body.put("shared", true); + return body; + } catch (TimeoutException e) { + return notReady(storage, "a probe is still running after " + timeoutMs + " ms"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return notReady(storage, "interrupted"); + } catch (ExecutionException e) { + return notReady(storage, "probe failed: " + + e.getCause().getClass().getSimpleName()); + } finally { + WAITERS.decrementAndGet(); + } + } + + private static Map<String, Object> probeOnce(String storage, Probe probe, long timeoutMs) { + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", false); + body.put("storage", storage); + try { + Map<String, Object> result = probe.probe(timeoutMs); + body.putAll(result); + } catch (Throwable e) { + LOG.warn("Storage readiness probe failed", e); + body.put("ready", false); + body.put("reason", "probe failed: " + e.getClass().getSimpleName()); + } + body.put("cached", false); + return body; + } + + private static Map<String, Object> notReady(String storage, String reason) { + Map<String, Object> body = new LinkedHashMap<>(); + body.put("ready", false); + body.put("storage", storage); + body.put("reason", reason); + body.put("cached", false); + return body; + } + + public static boolean isReady(Map<String, Object> body) { + return Boolean.TRUE.equals(body.get("ready")); + } + + public static void resetCache() { + LAST.set(null); + IN_FLIGHT.set(null); + WAITERS.set(0); + } + + private static HugeGraph firstRemoteGraph(GraphManager manager) { + for (String name : manager.graphs()) { + try { + HugeGraph graph = manager.graph(name); + if (graph != null && REMOTE_BACKENDS.contains(graph.backend())) { Review Comment: 🟡 [P2] Define readiness for independently configured remote backends With two graphs backed by independent HBase clusters, this returns the first graph permanently: a healthy first cluster makes `/readiness` return `ready=true` while a failed second cluster is never probed. I reproduced this with the exact `StorageReadiness` source and a unit harness: healthy graph metadata called once, failed graph metadata never called, result ready=true. This is a unit-level reproduction, not cluster acceptance. Either probe each independent backend configuration within the common budget, or document that readiness covers only the selected first remote graph and does not certify other HBase/HStore configurations. The currently documented HStore status-ping limitation is a separate, accepted boundary. -- 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]
