28861127500

Committed 07 Jul 2026 11:00AM UTC coverage: 55.685% (+1.3%) from 54.375%

Build Type

push

github

Committed by Ziroli

Commit Message

feat: cli *check* verifies more issues

Nanopub issues checked by the NanopubVerifier before publication are now also checked with the cli command "check". To see the detailed list of issues of a nanopub use "-v".

Coverage Stats

1469 of 3444 branches covered (42.65%)

Branch coverage included in aggregate %.

5966 of 9908 relevant lines covered (60.21%)

8.44 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.39

src/main/java/org/nanopub/extra/services/QueryCall.java

package org.nanopub.extra.services;

import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.util.EntityUtils;
import org.nanopub.NanopubUtils;
import org.nanopub.vocabulary.NPS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Second-generation query API call.
 */
public class QueryCall {

    private static final int DEFAULT_PARALLEL_CALL_COUNT = 2;

    /**
     * System property setting how many query API instances to call in parallel.
     * Must be {@code >= 1}; defaults to {@value #DEFAULT_PARALLEL_CALL_COUNT}.
     * Env var {@code NANOPUB_QUERY_PARALLEL_CALL_COUNT} also accepted.
     */
    public static final String PARALLEL_CALL_COUNT_PROPERTY = "nanopub.query.parallel-call-count";

    /**
     * Environment variable equivalent of {@link #PARALLEL_CALL_COUNT_PROPERTY}.
     */
    public static final String PARALLEL_CALL_COUNT_ENV = "NANOPUB_QUERY_PARALLEL_CALL_COUNT";

    private static int maxRetryCount = 3;
    private static final Logger logger = LoggerFactory.getLogger(QueryCall.class);

    /**
     * Returns the number of query API instances to call in parallel, resolved
     * (in order) from {@link #PARALLEL_CALL_COUNT_PROPERTY},
     * {@link #PARALLEL_CALL_COUNT_ENV}, or the default of
     * {@value #DEFAULT_PARALLEL_CALL_COUNT}. Invalid values are ignored.
     *
     * @return the parallel call count (always {@code >= 1})
     */
    public static int getParallelCallCount() {
        String value = System.getProperty(PARALLEL_CALL_COUNT_PROPERTY);
        if (value == null || value.isEmpty()) {
            value = System.getenv(PARALLEL_CALL_COUNT_ENV);
        }
        if (value != null && !value.trim().isEmpty()) {
            try {
                int n = Integer.parseInt(value.trim());
                if (n >= 1) {
                    return n;
                }
                logger.warn("Ignoring {}={}: must be >= 1; falling back to {}", PARALLEL_CALL_COUNT_PROPERTY, value, DEFAULT_PARALLEL_CALL_COUNT);
            } catch (NumberFormatException ex) {
                logger.warn("Ignoring {}={}: not an integer", PARALLEL_CALL_COUNT_PROPERTY, value);
            }
        }
        return DEFAULT_PARALLEL_CALL_COUNT;
    }

    /**
     * HTTP response header carrying the query instance's sync state.
     * See nanopub-query's {@code StatusController}.
     */
    public static final String QUERY_STATUS_HEADER = "Nanopub-Query-Status";

    /**
     * System property setting the cool-down (in seconds) before a query instance
     * evicted for non-ready status is re-considered. Default
     * {@value #DEFAULT_EVICTION_COOLDOWN_SECONDS}. Env var
     * {@code NANOPUB_QUERY_EVICTION_COOLDOWN_SECONDS} also accepted.
     */
    public static final String EVICTION_COOLDOWN_PROPERTY = "nanopub.query.eviction-cooldown-seconds";

    /**
     * Environment variable equivalent of {@link #EVICTION_COOLDOWN_PROPERTY}.
     */
    public static final String EVICTION_COOLDOWN_ENV = "NANOPUB_QUERY_EVICTION_COOLDOWN_SECONDS";

    private static final int DEFAULT_EVICTION_COOLDOWN_SECONDS = 300;

    private static final ConcurrentMap<String, Long> evictedUntil = new ConcurrentHashMap<>();

    /**
     * Returns the eviction cool-down in milliseconds, resolved from
     * {@link #EVICTION_COOLDOWN_PROPERTY}, {@link #EVICTION_COOLDOWN_ENV},
     * or the default of {@value #DEFAULT_EVICTION_COOLDOWN_SECONDS} seconds.
     */
    public static long getEvictionCooldownMillis() {
        String value = System.getProperty(EVICTION_COOLDOWN_PROPERTY);
        if (value == null || value.isEmpty()) {
            value = System.getenv(EVICTION_COOLDOWN_ENV);
        }
        if (value != null && !value.trim().isEmpty()) {
            try {
                long n = Long.parseLong(value.trim());
                if (n >= 0) {
                    return n * 1000L;
                }
                logger.warn("Ignoring {}={}: must be >= 0", EVICTION_COOLDOWN_PROPERTY, value);
            } catch (NumberFormatException ex) {
                logger.warn("Ignoring {}={}: not a number", EVICTION_COOLDOWN_PROPERTY, value);
            }
        }
        return DEFAULT_EVICTION_COOLDOWN_SECONDS * 1000L;
    }

    /**
     * Returns true if the response's {@link #QUERY_STATUS_HEADER} signals a
     * fully-synced state ({@code READY} or {@code LOADING_UPDATES}). Missing
     * header is treated as ready for backwards compatibility with older
     * query instances.
     */
    static boolean isReadyStatus(HttpResponse resp) {
        Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);
        if (h == null) {
            return true;
        }
        String v = h.getValue();
        if (v == null || v.isEmpty()) {
            return true;
        }
        String upper = v.toUpperCase(Locale.ROOT);
        return upper.equals("READY") || upper.equals("LOADING_UPDATES");
    }

    private static void evict(String apiUrl, String reason) {
        long until = System.currentTimeMillis() + getEvictionCooldownMillis();
        evictedUntil.put(apiUrl, until);
        logger.warn("Evicting Nanopub Query instance {} for {}s (reason: {}); re-eligible at {}", apiUrl, getEvictionCooldownMillis() / 1000, reason, new Date(until));
    }

    private static List<String> filterEvicted(List<String> instances) {
        long now = System.currentTimeMillis();
        List<String> result = new ArrayList<>(instances.size());
        for (String url : instances) {
            Long until = evictedUntil.get(url);
            if (until == null || until <= now) {
                result.add(url);
            }
        }
        return result;
    }

    /**
     * Run a query call with the given query ID and parameters.
     *
     * @param queryRef the reference to the query to run
     * @return the HTTP response from the query API
     * @throws APINotReachableException       if the API is not reachable after retries
     * @throws NotEnoughAPIInstancesException if there are not enough API instances available
     */
    public static HttpResponse run(QueryRef queryRef) throws APINotReachableException, NotEnoughAPIInstancesException {
        int retryCount = 0;
        while (retryCount < maxRetryCount) {
            QueryCall apiCall = new QueryCall(queryRef);
            apiCall.run();
            while (!apiCall.calls.isEmpty() && apiCall.resp == null) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
            if (apiCall.resp != null) {
                return apiCall.resp;
            }
            retryCount = retryCount + 1;
        }
        throw new APINotReachableException("Giving up contacting API: " + queryRef.getQueryId());
    }

    /**
     * System property naming a whitespace-separated list of query API instance URLs.
     * When set, this overrides discovery via the nanopub setting (env var
     * {@code NANOPUB_QUERY_INSTANCES} also accepted).
     */
    public static final String QUERY_INSTANCES_PROPERTY = "nanopub.query.instances";

    /**
     * Environment variable equivalent of {@link #QUERY_INSTANCES_PROPERTY}.
     */
    public static final String QUERY_INSTANCES_ENV = "NANOPUB_QUERY_INSTANCES";

    private static List<String> checkedApiInstances;

    /**
     * Returns the list of available query API instances that are currently accessible.
     * <p>
     * Sources, in order of priority:
     * <ol>
     *   <li>{@code nanopub.query.instances} system property / {@code NANOPUB_QUERY_INSTANCES} env var
     *       (whitespace-separated URLs).</li>
     *   <li>The active {@link org.nanopub.extra.setting.NanopubSetting}'s service intro collection,
     *       filtered to services of type {@link NPS#NANOPUB_QUERY_1_1}.</li>
     * </ol>
     * Each candidate is liveness-checked via an HTTP GET to its root URL.
     *
     * @return a list of accessible query API instances
     */
    public static synchronized List<String> getApiInstances() throws NotEnoughAPIInstancesException {
        List<String> candidates = resolveCandidateInstances();
        if (candidates.isEmpty()) {
            throw new NotEnoughAPIInstancesException("No query API instances configured or discoverable");
        }
        if (checkedApiInstances == null) {
            checkedApiInstances = new ArrayList<>();
        }
        long now = System.currentTimeMillis();
        boolean anyNewAdmitted = false;
        for (String a : candidates) {
            if (checkedApiInstances.contains(a)) {
                continue;
            }
            Long until = evictedUntil.get(a);
            if (until != null && until > now) {
                continue;
            }
            try {
                logger.info("Checking API instance: {}", a);
                HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(a));
                if (!wasSuccessful(resp)) {
                    EntityUtils.consumeQuietly(resp.getEntity());
                    logger.warn("Nanopub Query instance not accessible (HTTP {}): {}", resp.getStatusLine().getStatusCode(), a);
                    evict(a, "not accessible");
                } else if (!isReadyStatus(resp)) {
                    Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);
                    String status = h == null ? "missing" : h.getValue();
                    EntityUtils.consumeQuietly(resp.getEntity());
                    logger.warn("Nanopub Query instance not ready (status={}): {}; evicting", status, a);
                    evict(a, "status " + status);
                } else {
                    EntityUtils.consumeQuietly(resp.getEntity());
                    logger.info("Nanopub Query instance admitted: {}", a);
                    checkedApiInstances.add(a);
                    anyNewAdmitted = true;
                }
            } catch (IOException ex) {
                logger.warn("Nanopub Query instance not accessible ({}): {}", ex.getMessage(), a);
                evict(a, "not accessible");
            }
        }
        if (anyNewAdmitted) {
            logger.debug("{} accessible Nanopub Query instance(s) in pool", checkedApiInstances.size());
        }
        if (checkedApiInstances.isEmpty()) {
            throw new NotEnoughAPIInstancesException("No healthy Nanopub Query instances available");
        }
        if (anyNewAdmitted && checkedApiInstances.size() == 1) {
            logger.warn("Only one healthy Nanopub Query instance available; no failover.");
        }
        return checkedApiInstances;
    }

    private static List<String> resolveCandidateInstances() {
        String override = System.getProperty(QUERY_INSTANCES_PROPERTY);
        if (override == null || override.isEmpty()) {
            override = System.getenv(QUERY_INSTANCES_ENV);
        }
        if (override != null && !override.trim().isEmpty()) {
            List<String> list = new ArrayList<>();
            for (String url : override.trim().split("\\s+")) list.add(url);
            logger.info("Using {} query API instance(s) from override", list.size());
            return list;
        }
        List<String> fromSetting = ServiceLookup.getServices(NPS.NANOPUB_QUERY_1_1);
        logger.info("Discovered {} query API instance(s) from setting", fromSetting.size());
        return new ArrayList<>(fromSetting);
    }

    private QueryRef queryRef;
    private List<String> apisToCall = new ArrayList<>();
    private List<Call> calls = new ArrayList<>();

    private HttpResponse resp;

    private QueryCall(QueryRef queryRef) {
        this.queryRef = queryRef;
        logger.debug("Preparing query call: {}", queryRef);
    }

    private void run() throws NotEnoughAPIInstancesException {
        List<String> candidates = filterEvicted(getApiInstances());
        if (candidates.isEmpty()) {
            logger.warn("All {} Nanopub Query instance(s) currently evicted; cannot dispatch call for {}", getApiInstances().size(), queryRef.getQueryId());
            throw new NotEnoughAPIInstancesException(
                    "All Nanopub Query instances are currently evicted (loading/resetting); try again later");
        }
        List<String> apiInstancesToTry = new LinkedList<>(candidates);
        int parallelCallCount = getParallelCallCount();
        while (!apiInstancesToTry.isEmpty() && apisToCall.size() < parallelCallCount) {
            int randomIndex = (int) ((Math.random() * apiInstancesToTry.size()));
            String apiUrl = apiInstancesToTry.get(randomIndex);
            apisToCall.add(apiUrl);
            logger.info("Dispatching to instance {}/{}: {}", apisToCall.size(), parallelCallCount, apiUrl);
            apiInstancesToTry.remove(randomIndex);
        }
        for (String api : apisToCall) {
            Call call = new Call(api);
            calls.add(call);
            new Thread(call).start();
        }
    }

    private synchronized void finished(Call call, HttpResponse resp, String apiUrl) {
        if (this.resp != null) { // result already in
            EntityUtils.consumeQuietly(resp.getEntity());
            return;
        }
        long len = resp.getEntity().getContentLength();
        String sizeStr = len >= 0 ? len + " bytes" : "unknown (chunked/no Content-Length)";
        logger.info("Response received from {} for query {} ({})", apiUrl, queryRef, sizeStr);
        this.resp = resp;

        for (Call c : calls) {
            if (c != call) {
                c.abort();
            }
        }
    }

    private static boolean wasSuccessful(HttpResponse resp) {
        if (resp == null || resp.getEntity() == null) {
            return false;
        }
        int c = resp.getStatusLine().getStatusCode();
        return c >= 200 && c < 300;
    }

    private static boolean wasSuccessfulNonempty(HttpResponse resp) {
        if (!wasSuccessful(resp)) {
            return false;
        }
        // TODO Make sure we always return proper error codes, and then this shouldn't be necessary:
        if (resp.getHeaders("Content-Length").length > 0 && resp.getEntity().getContentLength() < 0) {
            return false;
        }
        return true;
    }


    private class Call implements Runnable {

        private String apiUrl;
        private HttpGet get;

        public Call(String apiUrl) {
            this.apiUrl = apiUrl;
        }

        public void run() {
            get = new HttpGet(apiUrl + "api/" + queryRef.getAsUrlString());
            get.setHeader("Accept", "text/csv, text/turtle;q=0.9, application/ld+json;q=0.8");
            HttpResponse resp = null;
            try {
                resp = NanopubUtils.getHttpClient().execute(get);
                if (!wasSuccessfulNonempty(resp)) {
                    throw new IOException(resp.getStatusLine().toString());
                }
                if (!isReadyStatus(resp)) {
                    Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);
                    String status = h == null ? "missing" : h.getValue();
                    evict(apiUrl, "status " + status);
                    EntityUtils.consumeQuietly(resp.getEntity());
                } else {
                    finished(this, resp, apiUrl);
                }
            } catch (Exception ex) {
                if (resp != null) {
                    EntityUtils.consumeQuietly(resp.getEntity());
                }
                logger.warn("Request to {} failed for query {} — {}: {}", apiUrl, queryRef, ex.getClass().getSimpleName(), ex.getMessage());
            }
            calls.remove(this);
        }

        private void abort() {
            if (get == null) {
                return;
            }
            if (get.isAborted()) {
                return;
            }
            get.abort();
        }

    }

}

1	package org.nanopub.extra.services;
2
3	import org.apache.http.Header;
4	import org.apache.http.HttpResponse;
5	import org.apache.http.client.methods.HttpGet;
6	import org.apache.http.util.EntityUtils;
7	import org.nanopub.NanopubUtils;
8	import org.nanopub.vocabulary.NPS;
9	import org.slf4j.Logger;
10	import org.slf4j.LoggerFactory;
11
12	import java.io.IOException;
13	import java.util.*;
14	import java.util.concurrent.ConcurrentHashMap;
15	import java.util.concurrent.ConcurrentMap;
16
17	/**
18	* Second-generation query API call.
19	*/
20	public class QueryCall {
21
22	private static final int DEFAULT_PARALLEL_CALL_COUNT = 2;
23
24	/**
25	* System property setting how many query API instances to call in parallel.
26	* Must be {@code >= 1}; defaults to {@value #DEFAULT_PARALLEL_CALL_COUNT}.
27	* Env var {@code NANOPUB_QUERY_PARALLEL_CALL_COUNT} also accepted.
28	*/
29	public static final String PARALLEL_CALL_COUNT_PROPERTY = "nanopub.query.parallel-call-count";
30
31	/**
32	* Environment variable equivalent of {@link #PARALLEL_CALL_COUNT_PROPERTY}.
33	*/
34	public static final String PARALLEL_CALL_COUNT_ENV = "NANOPUB_QUERY_PARALLEL_CALL_COUNT";
35
36	private static int maxRetryCount = 3;	6✔
37	private static final Logger logger = LoggerFactory.getLogger(QueryCall.class);	9✔
38
39	/**
40	* Returns the number of query API instances to call in parallel, resolved
41	* (in order) from {@link #PARALLEL_CALL_COUNT_PROPERTY},
42	* {@link #PARALLEL_CALL_COUNT_ENV}, or the default of
43	* {@value #DEFAULT_PARALLEL_CALL_COUNT}. Invalid values are ignored.
44	*
45	* @return the parallel call count (always {@code >= 1})
46	*/
47	public static int getParallelCallCount() {
48	String value = System.getProperty(PARALLEL_CALL_COUNT_PROPERTY);	9✔
49	if (value == null \|\| value.isEmpty()) {	15!
50	value = System.getenv(PARALLEL_CALL_COUNT_ENV);	9✔
51	}
52	if (value != null && !value.trim().isEmpty()) {	18!
53	try {
54	int n = Integer.parseInt(value.trim());	12✔
55	if (n >= 1) {	9✔
56	return n;	6✔
57	}
58	logger.warn("Ignoring {}={}: must be >= 1; falling back to {}", PARALLEL_CALL_COUNT_PROPERTY, value, DEFAULT_PARALLEL_CALL_COUNT);	54✔
59	} catch (NumberFormatException ex) {	3✔
60	logger.warn("Ignoring {}={}: not an integer", PARALLEL_CALL_COUNT_PROPERTY, value);	15✔
61	}	3✔
62	}
63	return DEFAULT_PARALLEL_CALL_COUNT;	6✔
64	}
65
66	/**
67	* HTTP response header carrying the query instance's sync state.
68	* See nanopub-query's {@code StatusController}.
69	*/
70	public static final String QUERY_STATUS_HEADER = "Nanopub-Query-Status";
71
72	/**
73	* System property setting the cool-down (in seconds) before a query instance
74	* evicted for non-ready status is re-considered. Default
75	* {@value #DEFAULT_EVICTION_COOLDOWN_SECONDS}. Env var
76	* {@code NANOPUB_QUERY_EVICTION_COOLDOWN_SECONDS} also accepted.
77	*/
78	public static final String EVICTION_COOLDOWN_PROPERTY = "nanopub.query.eviction-cooldown-seconds";
79
80	/**
81	* Environment variable equivalent of {@link #EVICTION_COOLDOWN_PROPERTY}.
82	*/
83	public static final String EVICTION_COOLDOWN_ENV = "NANOPUB_QUERY_EVICTION_COOLDOWN_SECONDS";
84
85	private static final int DEFAULT_EVICTION_COOLDOWN_SECONDS = 300;
86
87	private static final ConcurrentMap<String, Long> evictedUntil = new ConcurrentHashMap<>();	15✔
88
89	/**
90	* Returns the eviction cool-down in milliseconds, resolved from
91	* {@link #EVICTION_COOLDOWN_PROPERTY}, {@link #EVICTION_COOLDOWN_ENV},
92	* or the default of {@value #DEFAULT_EVICTION_COOLDOWN_SECONDS} seconds.
93	*/
94	public static long getEvictionCooldownMillis() {
95	String value = System.getProperty(EVICTION_COOLDOWN_PROPERTY);	9✔
96	if (value == null \|\| value.isEmpty()) {	15!
97	value = System.getenv(EVICTION_COOLDOWN_ENV);	9✔
98	}
99	if (value != null && !value.trim().isEmpty()) {	18!
100	try {
101	long n = Long.parseLong(value.trim());	12✔
102	if (n >= 0) {	12✔
103	return n * 1000L;	12✔
104	}
105	logger.warn("Ignoring {}={}: must be >= 0", EVICTION_COOLDOWN_PROPERTY, value);	15✔
106	} catch (NumberFormatException ex) {	3✔
107	logger.warn("Ignoring {}={}: not a number", EVICTION_COOLDOWN_PROPERTY, value);	15✔
108	}	3✔
109	}
110	return DEFAULT_EVICTION_COOLDOWN_SECONDS * 1000L;	6✔
111	}
112
113	/**
114	* Returns true if the response's {@link #QUERY_STATUS_HEADER} signals a
115	* fully-synced state ({@code READY} or {@code LOADING_UPDATES}). Missing
116	* header is treated as ready for backwards compatibility with older
117	* query instances.
118	*/
119	static boolean isReadyStatus(HttpResponse resp) {
120	Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);	12✔
121	if (h == null) {	6✔
122	return true;	6✔
123	}
124	String v = h.getValue();	9✔
125	if (v == null \|\| v.isEmpty()) {	15!
126	return true;	×
127	}
128	String upper = v.toUpperCase(Locale.ROOT);	12✔
129	return upper.equals("READY") \|\| upper.equals("LOADING_UPDATES");	36✔
130	}
131
132	private static void evict(String apiUrl, String reason) {
133	long until = System.currentTimeMillis() + getEvictionCooldownMillis();	12✔
134	evictedUntil.put(apiUrl, until);	18✔
135	logger.warn("Evicting Nanopub Query instance {} for {}s (reason: {}); re-eligible at {}", apiUrl, getEvictionCooldownMillis() / 1000, reason, new Date(until));	81✔
136	}	3✔
137
138	private static List<String> filterEvicted(List<String> instances) {
139	long now = System.currentTimeMillis();	6✔
140	List<String> result = new ArrayList<>(instances.size());	18✔
141	for (String url : instances) {	30✔
142	Long until = evictedUntil.get(url);	15✔
143	if (until == null \|\| until <= now) {	6!
144	result.add(url);	12✔
145	}
146	}	3✔
147	return result;	6✔
148	}
149
150	/**
151	* Run a query call with the given query ID and parameters.
152	*
153	* @param queryRef the reference to the query to run
154	* @return the HTTP response from the query API
155	* @throws APINotReachableException if the API is not reachable after retries
156	* @throws NotEnoughAPIInstancesException if there are not enough API instances available
157	*/
158	public static HttpResponse run(QueryRef queryRef) throws APINotReachableException, NotEnoughAPIInstancesException {
159	int retryCount = 0;	6✔
160	while (retryCount < maxRetryCount) {	9!
161	QueryCall apiCall = new QueryCall(queryRef);	15✔
162	apiCall.run();	6✔
163	while (!apiCall.calls.isEmpty() && apiCall.resp == null) {	21!
164	try {
165	Thread.sleep(200);	6✔
166	} catch (InterruptedException ex) {	×
167	Thread.currentThread().interrupt();	×
168	}	3✔
169	}
170	if (apiCall.resp != null) {	9!
171	return apiCall.resp;	9✔
172	}
173	retryCount = retryCount + 1;	×
174	}	×
175	throw new APINotReachableException("Giving up contacting API: " + queryRef.getQueryId());	×
176	}
177
178	/**
179	* System property naming a whitespace-separated list of query API instance URLs.
180	* When set, this overrides discovery via the nanopub setting (env var
181	* {@code NANOPUB_QUERY_INSTANCES} also accepted).
182	*/
183	public static final String QUERY_INSTANCES_PROPERTY = "nanopub.query.instances";
184
185	/**
186	* Environment variable equivalent of {@link #QUERY_INSTANCES_PROPERTY}.
187	*/
188	public static final String QUERY_INSTANCES_ENV = "NANOPUB_QUERY_INSTANCES";
189
190	private static List<String> checkedApiInstances;
191
192	/**
193	* Returns the list of available query API instances that are currently accessible.
194	* <p>
195	* Sources, in order of priority:
196	* <ol>
197	* <li>{@code nanopub.query.instances} system property / {@code NANOPUB_QUERY_INSTANCES} env var
198	* (whitespace-separated URLs).</li>
199	* <li>The active {@link org.nanopub.extra.setting.NanopubSetting}'s service intro collection,
200	* filtered to services of type {@link NPS#NANOPUB_QUERY_1_1}.</li>
201	* </ol>
202	* Each candidate is liveness-checked via an HTTP GET to its root URL.
203	*
204	* @return a list of accessible query API instances
205	*/
206	public static synchronized List<String> getApiInstances() throws NotEnoughAPIInstancesException {
207	List<String> candidates = resolveCandidateInstances();	6✔
208	if (candidates.isEmpty()) {	9!
209	throw new NotEnoughAPIInstancesException("No query API instances configured or discoverable");	×
210	}
211	if (checkedApiInstances == null) {	6✔
212	checkedApiInstances = new ArrayList<>();	12✔
213	}
214	long now = System.currentTimeMillis();	6✔
215	boolean anyNewAdmitted = false;	6✔
216	for (String a : candidates) {	30✔
217	if (checkedApiInstances.contains(a)) {	12✔
218	continue;	3✔
219	}
220	Long until = evictedUntil.get(a);	15✔
221	if (until != null && until > now) {	21!
222	continue;	×
223	}
224	try {
225	logger.info("Checking API instance: {}", a);	12✔
226	HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(a));	21✔
227	if (!wasSuccessful(resp)) {	9✔
228	EntityUtils.consumeQuietly(resp.getEntity());	9✔
229	logger.warn("Nanopub Query instance not accessible (HTTP {}): {}", resp.getStatusLine().getStatusCode(), a);	24✔
230	evict(a, "not accessible");	12✔
231	} else if (!isReadyStatus(resp)) {	9✔
232	Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);	12✔
233	String status = h == null ? "missing" : h.getValue();	15!
234	EntityUtils.consumeQuietly(resp.getEntity());	9✔
235	logger.warn("Nanopub Query instance not ready (status={}): {}; evicting", status, a);	15✔
236	evict(a, "status " + status);	12✔
237	} else {	3✔
238	EntityUtils.consumeQuietly(resp.getEntity());	9✔
239	logger.info("Nanopub Query instance admitted: {}", a);	12✔
240	checkedApiInstances.add(a);	12✔
241	anyNewAdmitted = true;	6✔
242	}
243	} catch (IOException ex) {	×
244	logger.warn("Nanopub Query instance not accessible ({}): {}", ex.getMessage(), a);	×
245	evict(a, "not accessible");	×
246	}	3✔
247	}	3✔
248	if (anyNewAdmitted) {	6✔
249	logger.debug("{} accessible Nanopub Query instance(s) in pool", checkedApiInstances.size());	18✔
250	}
251	if (checkedApiInstances.isEmpty()) {	9✔
252	throw new NotEnoughAPIInstancesException("No healthy Nanopub Query instances available");	15✔
253	}
254	if (anyNewAdmitted && checkedApiInstances.size() == 1) {	18✔
255	logger.warn("Only one healthy Nanopub Query instance available; no failover.");	9✔
256	}
257	return checkedApiInstances;	6✔
258	}
259
260	private static List<String> resolveCandidateInstances() {
261	String override = System.getProperty(QUERY_INSTANCES_PROPERTY);	9✔
262	if (override == null \|\| override.isEmpty()) {	15!
263	override = System.getenv(QUERY_INSTANCES_ENV);	9✔
264	}
265	if (override != null && !override.trim().isEmpty()) {	18!
266	List<String> list = new ArrayList<>();	12✔
267	for (String url : override.trim().split("\\s+")) list.add(url);	69✔
268	logger.info("Using {} query API instance(s) from override", list.size());	18✔
269	return list;	6✔
270	}
271	List<String> fromSetting = ServiceLookup.getServices(NPS.NANOPUB_QUERY_1_1);	9✔
272	logger.info("Discovered {} query API instance(s) from setting", fromSetting.size());	18✔
273	return new ArrayList<>(fromSetting);	15✔
274	}
275
276	private QueryRef queryRef;
277	private List<String> apisToCall = new ArrayList<>();	15✔
278	private List<Call> calls = new ArrayList<>();	15✔
279
280	private HttpResponse resp;
281
282	private QueryCall(QueryRef queryRef) {	6✔
283	this.queryRef = queryRef;	9✔
284	logger.debug("Preparing query call: {}", queryRef);	12✔
285	}	3✔
286
287	private void run() throws NotEnoughAPIInstancesException {
288	List<String> candidates = filterEvicted(getApiInstances());	9✔
289	if (candidates.isEmpty()) {	9!
290	logger.warn("All {} Nanopub Query instance(s) currently evicted; cannot dispatch call for {}", getApiInstances().size(), queryRef.getQueryId());	×
291	throw new NotEnoughAPIInstancesException(	×
292	"All Nanopub Query instances are currently evicted (loading/resetting); try again later");
293	}
294	List<String> apiInstancesToTry = new LinkedList<>(candidates);	15✔
295	int parallelCallCount = getParallelCallCount();	6✔
296	while (!apiInstancesToTry.isEmpty() && apisToCall.size() < parallelCallCount) {	24!
297	int randomIndex = (int) ((Math.random() * apiInstancesToTry.size()));	21✔
298	String apiUrl = apiInstancesToTry.get(randomIndex);	15✔
299	apisToCall.add(apiUrl);	15✔
300	logger.info("Dispatching to instance {}/{}: {}", apisToCall.size(), parallelCallCount, apiUrl);	63✔
301	apiInstancesToTry.remove(randomIndex);	12✔
302	}	3✔
303	for (String api : apisToCall) {	33✔
304	Call call = new Call(api);	18✔
305	calls.add(call);	15✔
306	new Thread(call).start();	15✔
307	}	3✔
308	}	3✔
309
310	private synchronized void finished(Call call, HttpResponse resp, String apiUrl) {
311	if (this.resp != null) { // result already in	9!
312	EntityUtils.consumeQuietly(resp.getEntity());	×
313	return;	×
314	}
315	long len = resp.getEntity().getContentLength();	12✔
316	String sizeStr = len >= 0 ? len + " bytes" : "unknown (chunked/no Content-Length)";	24!
317	logger.info("Response received from {} for query {} ({})", apiUrl, queryRef, sizeStr);	54✔
318	this.resp = resp;	9✔
319
320	for (Call c : calls) {	33✔
321	if (c != call) {	9✔
322	c.abort();	6✔
323	}
324	}	3✔
325	}	3✔
326
327	private static boolean wasSuccessful(HttpResponse resp) {
328	if (resp == null \|\| resp.getEntity() == null) {	15!
329	return false;	×
330	}
331	int c = resp.getStatusLine().getStatusCode();	12✔
332	return c >= 200 && c < 300;	30!
333	}
334
335	private static boolean wasSuccessfulNonempty(HttpResponse resp) {
336	if (!wasSuccessful(resp)) {	9!
337	return false;	×
338	}
339	// TODO Make sure we always return proper error codes, and then this shouldn't be necessary:
340	if (resp.getHeaders("Content-Length").length > 0 && resp.getEntity().getContentLength() < 0) {	33!
341	return false;	×
342	}
343	return true;	6✔
344	}
345
346
347	private class Call implements Runnable {
348
349	private String apiUrl;
350	private HttpGet get;
351
352	public Call(String apiUrl) {	15✔
353	this.apiUrl = apiUrl;	9✔
354	}	3✔
355
356	public void run() {
357	get = new HttpGet(apiUrl + "api/" + queryRef.getAsUrlString());	36✔
358	get.setHeader("Accept", "text/csv, text/turtle;q=0.9, application/ld+json;q=0.8");	15✔
359	HttpResponse resp = null;	6✔
360	try {
361	resp = NanopubUtils.getHttpClient().execute(get);	15✔
362	if (!wasSuccessfulNonempty(resp)) {	9!
363	throw new IOException(resp.getStatusLine().toString());	×
364	}
365	if (!isReadyStatus(resp)) {	9!
366	Header h = resp.getFirstHeader(QUERY_STATUS_HEADER);	×
367	String status = h == null ? "missing" : h.getValue();	×
368	evict(apiUrl, "status " + status);	×
369	EntityUtils.consumeQuietly(resp.getEntity());	×
370	} else {	×
371	finished(this, resp, apiUrl);	21✔
372	}
373	} catch (Exception ex) {	3✔
374	if (resp != null) {	6!
375	EntityUtils.consumeQuietly(resp.getEntity());	×
376	}
377	logger.warn("Request to {} failed for query {} — {}: {}", apiUrl, queryRef, ex.getClass().getSimpleName(), ex.getMessage());	81✔
378	}	3✔
379	calls.remove(this);	18✔
380	}	3✔
381
382	private void abort() {
383	if (get == null) {	9!
384	return;	×
385	}
386	if (get.isAborted()) {	12!
387	return;	×
388	}
389	get.abort();	9✔
390	}	3✔
391
392	}
393
394	}

Nanopublication / nanopub-java / 28861127500

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous