Commit 238d4a29 authored by Benoit Lavenier's avatar Benoit Lavenier

[fix] Add the endpoint API in peer id (use as unique key to store peer

[enh] Allow to store node software, in peer stats
[enh] PeerService: allow to fill "peer by currency" cache manually
parent b8f7f85c
Pipeline #4278 passed with stage
in 28 seconds
......@@ -275,7 +275,7 @@ public class Peer implements LocalEntity<String>, Serializable {
@JsonIgnore
public String computeKey() {
return Joiner.on('-').skipNulls().join(pubkey, dns, ipv4, ipv6, port, useSsl);
return Joiner.on('-').skipNulls().join(pubkey, dns, ipv4, ipv6, port, useSsl, api);
}
public String getApi() {
......@@ -403,11 +403,13 @@ public class Peer implements LocalEntity<String>, Serializable {
}
public static class Stats {
public static final String PROPERTY_SOFTWARE = "software";
public static final String PROPERTY_VERSION = "version";
public static final String PROPERTY_STATUS = "status";
public static final String PROPERTY_LAST_UP_TIME = "lastUpTime";
public static final String PROPERTY_UID = "uid";
private String software;
private String version;
private PeerStatus status = PeerStatus.UP; // default
private Integer blockNumber;
......@@ -446,6 +448,14 @@ public class Peer implements LocalEntity<String>, Serializable {
this.error = error;
}
public String getSoftware() {
return software;
}
public void setSoftware(String software) {
this.software = software;
}
public String getVersion() {
return version;
}
......
......@@ -51,4 +51,8 @@ public final class Peers {
return hasEndPointAPI(peer, EndpointApi.BASIC_MERKLED_API) ||
hasEndPointAPI(peer, EndpointApi.BMAS);
}
public static boolean hasEsCoreEndpoint(Peer peer) {
return hasEndPointAPI(peer, EndpointApi.ES_CORE_API);
}
}
......@@ -28,6 +28,7 @@ import org.duniter.core.client.model.local.Peer;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
......@@ -90,6 +91,7 @@ public interface NetworkService extends Service {
final Filter filter, final Sort sort, final boolean autoreconnect,
final ExecutorService executor);
CompletableFuture<List<Peer>> asyncRefreshPeers(final Peer mainPeer, final List<Peer> peers, final ExecutorService pool);
String getVersion(final Peer peer);
}
......@@ -63,6 +63,7 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
private final static String BMA_URL_STATUS = "/node/summary";
private final static String BMA_URL_BLOCKCHAIN_CURRENT = "/blockchain/current";
private final static String BMA_URL_BLOCKCHAIN_HARDSHIP = "/blockchain/hardship/";
private final static String ES_URL_BLOCKCHAIN_CURRENT = "/blockchain/current";
private NetworkRemoteService networkRemoteService;
private WotRemoteService wotRemoteService;
......@@ -185,9 +186,10 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
});
}
public CompletableFuture<Peer> asyncRefreshPeer(final Peer peer, final Map<String, String> memberUids, final ExecutorService pool) {
return CompletableFuture.supplyAsync(() -> fillVersion(peer), pool)
.thenApply(p -> Peers.hasBmaEndpoint(p) ? fillCurrentBlock(p) : p)
.thenApply(p -> fillCurrentBlock(p))
.exceptionally(throwable -> {
peer.getStats().setStatus(Peer.PeerStatus.DOWN);
if(!(throwable instanceof HttpConnectException)) {
......@@ -221,6 +223,18 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
});
}
public CompletableFuture<List<Peer>> asyncRefreshPeers(final Peer mainPeer, final List<Peer> peers, final ExecutorService pool) {
return CompletableFuture.supplyAsync(() -> wotRemoteService.getMembersUids(mainPeer), pool)
// Refresh all endpoints
.thenApply(memberUids ->
peers.stream().map(peer ->
asyncRefreshPeer(peer, memberUids, pool))
.collect(Collectors.toList())
)
.thenCompose(CompletableFutures::allOfToList);
}
public List<Peer> fillPeerStatsConsensus(final List<Peer> peers) {
final Map<String,Long> peerCountByBuid = peers.stream()
......@@ -433,7 +447,14 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
}
public String getVersion(final Peer peer) {
JsonNode json = get(peer, BMA_URL_STATUS);
return getVersion(getNodeSummary(peer));
}
public JsonNode getNodeSummary(final Peer peer) {
return get(peer, BMA_URL_STATUS);
}
public String getVersion(JsonNode json) {
json = json.get("duniter");
if (json.isMissingNode()) throw new TechnicalException(String.format("Invalid format of [%s] response", BMA_URL_STATUS));
json = json.get("version");
......@@ -441,6 +462,14 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
return json.asText();
}
public String getSoftware(JsonNode json) {
json = json.get("duniter");
if (json.isMissingNode()) throw new TechnicalException(String.format("Invalid format of [%s] response", BMA_URL_STATUS));
json = json.get("software");
if (json.isMissingNode()) throw new TechnicalException(String.format("No software attribute found in [%s] response", BMA_URL_STATUS));
return json.asText();
}
/* -- protected methods -- */
protected List<Peer> loadPeerLeafs(Peer peer, List<String> filterEndpoints) {
......@@ -561,28 +590,32 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
}
protected Peer fillVersion(final Peer peer) {
String version = getVersion(peer);
peer.getStats().setVersion(version);
if (!Peers.hasBmaEndpoint(peer) && !Peers.hasEsCoreEndpoint(peer)) return peer;
JsonNode summary = getNodeSummary(peer);
peer.getStats().setVersion(getVersion(summary));
peer.getStats().setSoftware(getSoftware(summary));
return peer;
}
protected Peer fillCurrentBlock(final Peer peer) {
JsonNode json = get(peer, BMA_URL_BLOCKCHAIN_CURRENT);
if (Peers.hasBmaEndpoint(peer) || Peers.hasEsCoreEndpoint(peer)) {
JsonNode json = get(peer, BMA_URL_BLOCKCHAIN_CURRENT);
String currency = json.has("currency") ? json.get("currency").asText() : null;
peer.setCurrency(currency);
String currency = json.has("currency") ? json.get("currency").asText() : null;
peer.setCurrency(currency);
Integer number = json.has("number") ? json.get("number").asInt() : null;
peer.getStats().setBlockNumber(number);
Integer number = json.has("number") ? json.get("number").asInt() : null;
peer.getStats().setBlockNumber(number);
String hash = json.has("hash") ? json.get("hash").asText() : null;
peer.getStats().setBlockHash(hash);
String hash = json.has("hash") ? json.get("hash").asText() : null;
peer.getStats().setBlockHash(hash);
Long medianTime = json.has("medianTime") ? json.get("medianTime").asLong() : null;
peer.getStats().setMedianTime(medianTime);
Long medianTime = json.has("medianTime") ? json.get("medianTime").asLong() : null;
peer.getStats().setMedianTime(medianTime);
if (log.isTraceEnabled()) {
log.trace(String.format("[%s] current block [%s-%s]", peer.toString(), number, hash));
if (log.isTraceEnabled()) {
log.trace(String.format("[%s] current block [%s-%s]", peer.toString(), number, hash));
}
}
return peer;
......
......@@ -41,6 +41,13 @@ public interface PeerService extends Service {
*/
Peer getActivePeerByCurrencyId(String currencyId);
/**
* Save the active (default) peer, for a given currency id
* @param currencyId
* @param peer
*/
void setCurrencyMainPeer(String currency, Peer peer);
/**
* Return a (cached) peer list, by currency id
* @param currencyId
......
......@@ -60,6 +60,7 @@ public class PeerServiceImpl implements PeerService, InitializingBean {
public PeerServiceImpl() {
super();
}
@Override
......@@ -68,6 +69,12 @@ public class PeerServiceImpl implements PeerService, InitializingBean {
this.peerDao = ServiceLocator.instance().getBean(PeerDao.class);
this.config = Configuration.instance();
this.cryptoService = ServiceLocator.instance().getCryptoService();
this.activePeerByCurrencyIdCache = new SimpleCache<String, Peer>() {
@Override
public Peer load(String currencyId) {
return loadDefaultPeer(currencyId);
}
};
}
@Override
......@@ -126,28 +133,14 @@ public class PeerServiceImpl implements PeerService, InitializingBean {
* @return
*/
public Peer getActivePeerByCurrencyId(String currencyId) {
// Check if cache as been loaded
if (activePeerByCurrencyIdCache == null) {
activePeerByCurrencyIdCache = new SimpleCache<String, Peer>() {
@Override
public Peer load(String currencyId) {
List<Peer> peers = peerDao.getPeersByCurrencyId(currencyId);
if (CollectionUtils.isEmpty(peers)) {
String currencyName = currencyService.getCurrencyNameById(currencyId);
throw new TechnicalException(String.format(
"No peers configure for currency [%s]",
currencyName != null ? currencyName : currencyId));
}
return peers.get(0);
}
};
}
return activePeerByCurrencyIdCache.get(currencyId);
}
@Override
public void setCurrencyMainPeer(String currencyId, Peer peer) {
activePeerByCurrencyIdCache.put(currencyId, peer);
}
/**
* Return a (cached) peer list, by currency id
* @param currencyId
......@@ -223,4 +216,16 @@ public class PeerServiceImpl implements PeerService, InitializingBean {
public boolean isExists(String currencyId, String peerId) {
return peerDao.isExists(currencyId, peerId);
}
protected Peer loadDefaultPeer(String currencyId) {
List<Peer> peers = peerDao.getPeersByCurrencyId(currencyId);
if (CollectionUtils.isEmpty(peers)) {
String currencyName = currencyService.getCurrencyNameById(currencyId);
throw new TechnicalException(String.format(
"No peers configure for currency [%s]",
currencyName != null ? currencyName : currencyId));
}
return peers.get(0);
}
}
......@@ -22,6 +22,9 @@ package org.duniter.core.util;
* #L%
*/
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
......@@ -95,4 +98,15 @@ public class CollectionUtils {
throw new IllegalArgumentException("Can extract singleton only when collection size == 1");
}
}
public static <C> Collection<C> union(Collection<C> c1, Collection<C> c2) {
if (isEmpty(c1)) return c2;
if (isEmpty(c2)) return c1;
ArrayList<C> result = Lists.newArrayList();
result.addAll(c1);
c2.forEach(i -> {
if (!result.contains(i)) result.add(i);
});
return result;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment