ethstats/src/main/java/org/apache/tuweni/ethstats/EthStatsReporter.java - incubator-tuweni - Git at Google

 /*
  * 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.tuweni.ethstats;

 import org.apache.tuweni.eth.EthJsonModule;
 import org.apache.tuweni.units.bigints.UInt256;

 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;

 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import io.vertx.core.Future;
 import io.vertx.core.MultiMap;
 import io.vertx.core.TimeoutStream;
 import io.vertx.core.Vertx;
 import io.vertx.core.WorkerExecutor;
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.http.WebSocket;
 import org.logl.Logger;

 /**
  * ETHNetStats reporting service.
  * <p>
  * This service connects to a running ethnetstats service and reports.
  * <p>
  * If the service is not available, the reporter will keep trying to connect periodically. The service will report
  * statistics over time.
  */
 public final class EthStatsReporter {

   private final static ObjectMapper mapper = new ObjectMapper();

   static {
     mapper.registerModule(new EthJsonModule());
   }

   private final static long DELAY = 5000;
   private final static long REPORTING_PERIOD = 1000;
   private final static long PING_PERIOD = 15000;


   private final String id;
   private final Vertx vertx;
   private final List<URI> ethstatsServerURIs;
   private final Logger logger;
   private final AtomicBoolean started = new AtomicBoolean(false);
   private final AtomicBoolean waitingOnPong = new AtomicBoolean(false);
   private final NodeInfo nodeInfo;
   private final String secret;
   private final AtomicReference<Integer> newTxCount = new AtomicReference<>();
   private final Consumer<List<UInt256>> historyRequester;

   private WorkerExecutor executor;
   private HttpClient client;
   private AtomicReference<BlockStats> newHead = new AtomicReference<>();
   private AtomicReference<NodeStats> newNodeStats = new AtomicReference<>();
   private AtomicReference<List<BlockStats>> newHistory = new AtomicReference<>();

   /**
    * Default constructor.
    *
    * @param vertx a Vert.x instance, externally managed.
    * @param logger a logger
    * @param ethstatsServerURIs the URIs to connect to eth-netstats, such as ws://www.ethnetstats.org:3000/api. URIs are
    *        tried in sequence, and the first one to work is used.
    * @param secret the secret to use when we connect to eth-netstats
    * @param name the name of the node to be reported in the UI
    * @param node the node name to be reported in the UI
    * @param port the devp2p port exposed by this node
    * @param network the network id
    * @param protocol the version of the devp2p eth subprotocol, such as eth/63
    * @param os the operating system on which the node runs
    * @param osVer the version of the OS on which the node runs
    * @param historyRequester a hook for ethstats to request block information by number.
    */
   public EthStatsReporter(
       Vertx vertx,
       Logger logger,
       List<URI> ethstatsServerURIs,
       String secret,
       String name,
       String node,
       int port,
       String network,
       String protocol,
       String os,
       String osVer,
       Consumer<List<UInt256>> historyRequester) {
     this.id = UUID.randomUUID().toString();
     this.vertx = vertx;
     this.logger = logger;
     this.ethstatsServerURIs = ethstatsServerURIs;
     this.secret = secret;
     this.nodeInfo = new NodeInfo(name, node, port, network, protocol, os, osVer);
     this.historyRequester = historyRequester;
   }

   public void start() {
     if (started.compareAndSet(false, true)) {
       executor = vertx.createSharedWorkerExecutor("ethnetstats");
       client = vertx.createHttpClient(new HttpClientOptions().setLogActivity(true));
       startInternal();
     }
   }

   public void stop() {
     if (started.compareAndSet(true, false)) {
       logger.debug("Stopping the service");
       executor.close();
     }
   }

   public void sendNewHead(BlockStats newBlockStats) {
     newHead.set(newBlockStats);
   }

   public void sendNewPendingTransactionCount(int txCount) {
     newTxCount.set(txCount);
   }

   public void sendNewNodeStats(NodeStats nodeStats) {
     newNodeStats.set(nodeStats);
   }

   public void sendHistoryResponse(List<BlockStats> blocks) {
     newHistory.set(blocks);
   }

   private void startInternal() {
     AtomicBoolean connectedOK = new AtomicBoolean(false);
     for (URI uri : ethstatsServerURIs) {
       executor.executeBlocking((Future<Boolean> handler) -> connect(handler, uri), result -> {
         logger.debug("Attempting to connect", result.cause());
         connectedOK.set(!result.failed() && result.result());
       });
       if (connectedOK.get()) {
         break;
       }
     }
     if (!connectedOK.get() && started.get()) {
       attemptConnect(null);
     }
   }

   private void attemptConnect(Void aVoid) {
     vertx.setTimer(DELAY, handler -> this.startInternal());
   }

   private void connect(Future<Boolean> result, URI uri) {
     client.websocket(
         uri.getPort(),
         uri.getHost(),
         uri.toString(),
         MultiMap.caseInsensitiveMultiMap().add("origin", "http://localhost"),
         ws -> {
           ws.closeHandler(this::attemptConnect);
           ws.exceptionHandler(e -> {
             logger.debug("Error while communicating with ethnetstats", e);

           });
           ws.textMessageHandler(message -> {
             try {
               JsonNode node = mapper.readTree(message);
               JsonNode emitEvent = node.get("emit");
               if (emitEvent.isArray()) {
                 String eventValue = emitEvent.get(0).textValue();
                 if (!result.isComplete()) {
                   if (!"ready".equals(eventValue)) {
                     logger.warn(message);
                     result.complete(false);
                   } else {
                     logger.debug("Connected OK! {}", message);
                     result.complete(true);

                     // we are connected and now sending information
                     reportPeriodically(ws);
                     writePing(ws);
                     report(ws);
                   }
                 } else {
                   handleEmitEvent((ArrayNode) emitEvent, ws);
                 }
               } else {
                 logger.warn(message);
                 result.complete(false);
               }
             } catch (IOException e) {
               throw new UncheckedIOException(e);
             }
           });

           writeCommand(ws, "hello", new AuthMessage(nodeInfo, id, secret));
         },
         result::fail);
   }

   private void handleEmitEvent(ArrayNode event, WebSocket ws) {
     String command = event.get(0).textValue();
     switch (command) {
       case "node-pong":
         logger.debug("Received a pong {}", event.get(1));
         if (!waitingOnPong.compareAndSet(true, false)) {
           logger.warn("Received pong when we didn't expect one");
         } else {
           long start = event.get(1).get("clientTime").longValue();
           long latency = (Instant.now().toEpochMilli() - start) / (2 * 1000);
           writeCommand(ws, "latency", "latency", latency);
         }
         break;
       case "history":
         logger.debug("History request {}", event.get(1));
         requestHistory(event.get(1));
         break;
       default:
         logger.warn("Unexpected message {}", command);

     }
   }

   private void requestHistory(JsonNode list) {
     historyRequester.accept(null);
   }

   private void writePing(WebSocket ws) {
     waitingOnPong.set(true);
     writeCommand(ws, "node-ping", "clientTime", Instant.now().toEpochMilli());
   }

   private void reportPeriodically(WebSocket ws) {
     TimeoutStream reportingStream = vertx.periodicStream(REPORTING_PERIOD).handler(ev -> {
       report(ws);
     });
     TimeoutStream pingStream = vertx.periodicStream(PING_PERIOD).handler(ev -> {
       writePing(ws);
     });
     ws.closeHandler(h -> {
       reportingStream.cancel();
       pingStream.cancel();
       attemptConnect(null);
     });
   }

   private void report(WebSocket ws) {
     BlockStats head = newHead.getAndSet(null);
     if (head != null) {
       writeCommand(ws, "block", "block", head);
     }
     Integer count = newTxCount.getAndSet(null);
     if (count != null) {
       writeCommand(ws, "pending", "stats", Collections.singletonMap("pending", count));
     }
     NodeStats nodeStats = newNodeStats.getAndSet(null);
     if (nodeStats != null) {
       writeCommand(ws, "stats", "stats", nodeStats);
     }
     List<BlockStats> newBlocks = newHistory.getAndSet(null);
     if (newBlocks != null && !newBlocks.isEmpty()) {
       writeCommand(ws, "history", "history", newBlocks);
     }
   }

   private void writeCommand(WebSocket ws, String command, Object payload) {
     try {
       String message =
           mapper.writer().writeValueAsString(Collections.singletonMap("emit", Arrays.asList(command, payload)));
       logger.debug("Sending {} message {}", command, message);
       ws.writeTextMessage(message);
     } catch (JsonProcessingException e) {
       throw new UncheckedIOException(e);
     }
   }

   private void writeCommand(WebSocket ws, String command, String key, Object payload) {
     Map<String, Object> body = new HashMap<>();
     body.put("id", id);
     body.put(key, payload);
     writeCommand(ws, command, body);
   }
 }
	/*
	* 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.tuweni.ethstats;

	import org.apache.tuweni.eth.EthJsonModule;
	import org.apache.tuweni.units.bigints.UInt256;

	import java.io.IOException;
	import java.io.UncheckedIOException;
	import java.net.URI;
	import java.time.Instant;
	import java.util.Arrays;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.UUID;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicReference;
	import java.util.function.Consumer;

	import com.fasterxml.jackson.core.JsonProcessingException;
	import com.fasterxml.jackson.databind.JsonNode;
	import com.fasterxml.jackson.databind.ObjectMapper;
	import com.fasterxml.jackson.databind.node.ArrayNode;
	import io.vertx.core.Future;
	import io.vertx.core.MultiMap;
	import io.vertx.core.TimeoutStream;
	import io.vertx.core.Vertx;
	import io.vertx.core.WorkerExecutor;
	import io.vertx.core.http.HttpClient;
	import io.vertx.core.http.HttpClientOptions;
	import io.vertx.core.http.WebSocket;
	import org.logl.Logger;

	/**
	* ETHNetStats reporting service.
	* <p>
	* This service connects to a running ethnetstats service and reports.
	* <p>
	* If the service is not available, the reporter will keep trying to connect periodically. The service will report
	* statistics over time.
	*/
	public final class EthStatsReporter {

	private final static ObjectMapper mapper = new ObjectMapper();

	static {
	mapper.registerModule(new EthJsonModule());
	}

	private final static long DELAY = 5000;
	private final static long REPORTING_PERIOD = 1000;
	private final static long PING_PERIOD = 15000;


	private final String id;
	private final Vertx vertx;
	private final List<URI> ethstatsServerURIs;
	private final Logger logger;
	private final AtomicBoolean started = new AtomicBoolean(false);
	private final AtomicBoolean waitingOnPong = new AtomicBoolean(false);
	private final NodeInfo nodeInfo;
	private final String secret;
	private final AtomicReference<Integer> newTxCount = new AtomicReference<>();
	private final Consumer<List<UInt256>> historyRequester;

	private WorkerExecutor executor;
	private HttpClient client;
	private AtomicReference<BlockStats> newHead = new AtomicReference<>();
	private AtomicReference<NodeStats> newNodeStats = new AtomicReference<>();
	private AtomicReference<List<BlockStats>> newHistory = new AtomicReference<>();

	/**
	* Default constructor.
	*
	* @param vertx a Vert.x instance, externally managed.
	* @param logger a logger
	* @param ethstatsServerURIs the URIs to connect to eth-netstats, such as ws://www.ethnetstats.org:3000/api. URIs are
	* tried in sequence, and the first one to work is used.
	* @param secret the secret to use when we connect to eth-netstats
	* @param name the name of the node to be reported in the UI
	* @param node the node name to be reported in the UI
	* @param port the devp2p port exposed by this node
	* @param network the network id
	* @param protocol the version of the devp2p eth subprotocol, such as eth/63
	* @param os the operating system on which the node runs
	* @param osVer the version of the OS on which the node runs
	* @param historyRequester a hook for ethstats to request block information by number.
	*/
	public EthStatsReporter(
	Vertx vertx,
	Logger logger,
	List<URI> ethstatsServerURIs,
	String secret,
	String name,
	String node,
	int port,
	String network,
	String protocol,
	String os,
	String osVer,
	Consumer<List<UInt256>> historyRequester) {
	this.id = UUID.randomUUID().toString();
	this.vertx = vertx;
	this.logger = logger;
	this.ethstatsServerURIs = ethstatsServerURIs;
	this.secret = secret;
	this.nodeInfo = new NodeInfo(name, node, port, network, protocol, os, osVer);
	this.historyRequester = historyRequester;
	}

	public void start() {
	if (started.compareAndSet(false, true)) {
	executor = vertx.createSharedWorkerExecutor("ethnetstats");
	client = vertx.createHttpClient(new HttpClientOptions().setLogActivity(true));
	startInternal();
	}
	}

	public void stop() {
	if (started.compareAndSet(true, false)) {
	logger.debug("Stopping the service");
	executor.close();
	}
	}

	public void sendNewHead(BlockStats newBlockStats) {
	newHead.set(newBlockStats);
	}

	public void sendNewPendingTransactionCount(int txCount) {
	newTxCount.set(txCount);
	}

	public void sendNewNodeStats(NodeStats nodeStats) {
	newNodeStats.set(nodeStats);
	}

	public void sendHistoryResponse(List<BlockStats> blocks) {
	newHistory.set(blocks);
	}

	private void startInternal() {
	AtomicBoolean connectedOK = new AtomicBoolean(false);
	for (URI uri : ethstatsServerURIs) {
	executor.executeBlocking((Future<Boolean> handler) -> connect(handler, uri), result -> {
	logger.debug("Attempting to connect", result.cause());
	connectedOK.set(!result.failed() && result.result());
	});
	if (connectedOK.get()) {
	break;
	}
	}
	if (!connectedOK.get() && started.get()) {
	attemptConnect(null);
	}
	}

	private void attemptConnect(Void aVoid) {
	vertx.setTimer(DELAY, handler -> this.startInternal());
	}

	private void connect(Future<Boolean> result, URI uri) {
	client.websocket(
	uri.getPort(),
	uri.getHost(),
	uri.toString(),
	MultiMap.caseInsensitiveMultiMap().add("origin", "http://localhost"),
	ws -> {
	ws.closeHandler(this::attemptConnect);
	ws.exceptionHandler(e -> {
	logger.debug("Error while communicating with ethnetstats", e);

	});
	ws.textMessageHandler(message -> {
	try {
	JsonNode node = mapper.readTree(message);
	JsonNode emitEvent = node.get("emit");
	if (emitEvent.isArray()) {
	String eventValue = emitEvent.get(0).textValue();
	if (!result.isComplete()) {
	if (!"ready".equals(eventValue)) {
	logger.warn(message);
	result.complete(false);
	} else {
	logger.debug("Connected OK! {}", message);
	result.complete(true);

	// we are connected and now sending information
	reportPeriodically(ws);
	writePing(ws);
	report(ws);
	}
	} else {
	handleEmitEvent((ArrayNode) emitEvent, ws);
	}
	} else {
	logger.warn(message);
	result.complete(false);
	}
	} catch (IOException e) {
	throw new UncheckedIOException(e);
	}
	});

	writeCommand(ws, "hello", new AuthMessage(nodeInfo, id, secret));
	},
	result::fail);
	}

	private void handleEmitEvent(ArrayNode event, WebSocket ws) {
	String command = event.get(0).textValue();
	switch (command) {
	case "node-pong":
	logger.debug("Received a pong {}", event.get(1));
	if (!waitingOnPong.compareAndSet(true, false)) {
	logger.warn("Received pong when we didn't expect one");
	} else {
	long start = event.get(1).get("clientTime").longValue();
	long latency = (Instant.now().toEpochMilli() - start) / (2 * 1000);
	writeCommand(ws, "latency", "latency", latency);
	}
	break;
	case "history":
	logger.debug("History request {}", event.get(1));
	requestHistory(event.get(1));
	break;
	default:
	logger.warn("Unexpected message {}", command);

	}
	}

	private void requestHistory(JsonNode list) {
	historyRequester.accept(null);
	}

	private void writePing(WebSocket ws) {
	waitingOnPong.set(true);
	writeCommand(ws, "node-ping", "clientTime", Instant.now().toEpochMilli());
	}

	private void reportPeriodically(WebSocket ws) {
	TimeoutStream reportingStream = vertx.periodicStream(REPORTING_PERIOD).handler(ev -> {
	report(ws);
	});
	TimeoutStream pingStream = vertx.periodicStream(PING_PERIOD).handler(ev -> {
	writePing(ws);
	});
	ws.closeHandler(h -> {
	reportingStream.cancel();
	pingStream.cancel();
	attemptConnect(null);
	});
	}

	private void report(WebSocket ws) {
	BlockStats head = newHead.getAndSet(null);
	if (head != null) {
	writeCommand(ws, "block", "block", head);
	}
	Integer count = newTxCount.getAndSet(null);
	if (count != null) {
	writeCommand(ws, "pending", "stats", Collections.singletonMap("pending", count));
	}
	NodeStats nodeStats = newNodeStats.getAndSet(null);
	if (nodeStats != null) {
	writeCommand(ws, "stats", "stats", nodeStats);
	}
	List<BlockStats> newBlocks = newHistory.getAndSet(null);
	if (newBlocks != null && !newBlocks.isEmpty()) {
	writeCommand(ws, "history", "history", newBlocks);
	}
	}

	private void writeCommand(WebSocket ws, String command, Object payload) {
	try {
	String message =
	mapper.writer().writeValueAsString(Collections.singletonMap("emit", Arrays.asList(command, payload)));
	logger.debug("Sending {} message {}", command, message);
	ws.writeTextMessage(message);
	} catch (JsonProcessingException e) {
	throw new UncheckedIOException(e);
	}
	}

	private void writeCommand(WebSocket ws, String command, String key, Object payload) {
	Map<String, Object> body = new HashMap<>();
	body.put("id", id);
	body.put(key, payload);
	writeCommand(ws, command, body);
	}
	}