project init

.gitignore

@@ -0,0 +1,6 @@
+.idea
+*.iml
+tmp
+settings.dev.yml
+go.sum
+*.log

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Jason
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,51 @@
+# eth-stats
+可用于实时监控分布自不同区域的ethereum节点，操作简单，一目了然。  
+该项目可监控基于ethereum的大多数项目，包括L2、L3等等，具体细节，就需要自行探索了
+
+## 功能
+1. 每个节点名称不得重复
+2. 支持实时上传节点信息
+3. 节点异常时，实时邮件反馈
+4. 定时邮件发送节点简报
+5. server和client强稳定性，运行期间，除非强制杀进程或者bug，否则程序不会因为任何逻辑问题停止运行，降低了运维复杂度
+6. 可通过命令行传入参或者通过配置文件启动`client、server`，不建议同时使用两种方式，选择其中一种即可
+7. 本项目没有前端页面，主要是不会用前端语言，也设计不了。。。本项目在server/app/service/api中提供了socket数据出口，只要前端使用socket调用，即可渲染在前端。
+   1. 前端通过socket的emit可读取：`stats 节点信息`、`latency 延迟`、`node-ping ping`三类数据
+
+## 使用方式
+分为客户端和服务器端，客户端安装在每台需要监控的节点上，服务器端找台有ip的稳定机子部署就行
+
+### client
+```shell
+cd client
+go build -o client ./client.go 
+CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o client client.go
+
+# 配置方式启动
+client start -c config/setting.yml
+
+# 命令行方式启动
+./client start --name test --secret 123456 --server-url 链地址，如：ws://127.0.0.1:30303
+```
+
+### server
+```shell
+cd server
+go build server.go -o server
+CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o server server.go
+
+# 配置方式启动
+server start -c config/setting.yml
+
+# 命令行方式启动
+./server start --name ethereum-server --secret 123456 --host 0.0.0.0 --port 3000 --email-subject-prefix ethereum --email-host 邮箱服务地址 --email-port 465 --email-username 发件邮箱账户 --email-password 邮箱密钥 --email-from 发件邮箱账户--email-to 收件邮箱账户(多个逗号隔开)
+```
+
+## 参考
+[1] [goerli-ethstats-server](https://github.com/goerli/ethstats-server)  
+[2] [goerli-ethstats-client](https://github.com/goerli/ethstats-client)  
+[3] [AvileneRausch2001-ethstats](https://github.com/AvileneRausch2001/ethstats)  
+[4] [AvileneRausch2001-ethstats](https://github.com/AvileneRausch2001/ethstats)  
+[5] [maticnetwork-ethstats-backend](https://github.com/maticnetwork/ethstats-backend)  
+
+

client/app/app.go

@@ -0,0 +1,270 @@
+package app
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"ethstats/client/app/model"
+	"ethstats/client/config"
+	"ethstats/common/util/connutil"
+	"github.com/bitxx/ethutil"
+	"github.com/bitxx/logger/logbase"
+	"os"
+	"os/signal"
+	"runtime"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type App struct {
+	node    model.Node
+	readyCh chan struct{}
+	pongCh  chan struct{}
+	logger  *logbase.Helper
+}
+
+func NewApp() *App {
+	node := model.Node{
+		Id:         config.ApplicationConfig.Name,
+		Name:       config.ApplicationConfig.Name,
+		Contact:    config.ApplicationConfig.Contract,
+		ChainPort:  config.ChainConfig.Port,
+		OSPlatform: runtime.GOARCH,
+		OS:         runtime.GOOS,
+		Client:     config.ApplicationConfig.Version,
+	}
+
+	return &App{
+		node:    node,
+		readyCh: make(chan struct{}),
+		pongCh:  make(chan struct{}),
+		logger:  logbase.NewHelper(logbase.DefaultLogger),
+	}
+}
+
+func (a *App) Start() {
+	// logbase.NewHelper(core.Runtime.GetLogger())
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, os.Interrupt)
+
+	var err error
+	isInterrupt := false
+
+	conn := &connutil.ConnWrapper{}
+	readTicker := time.NewTimer(0)
+	latencyTicker := time.NewTimer(0)
+
+	defer func() {
+		a.close(conn, readTicker, latencyTicker)
+		// if not interrupt,restart the client
+		if !isInterrupt {
+			time.Sleep(5 * time.Second)
+			a.Start()
+		}
+	}()
+
+	conn, err = connutil.NewDialConn(config.ApplicationConfig.ServerUrl)
+	if err != nil {
+		a.logger.Warn("dial error: ", err)
+		return
+	}
+
+	for {
+		select {
+		case <-readTicker.C:
+			//after customer,change the time
+			readTicker.Reset(10 * time.Second)
+			latencyTicker.Reset(2 * time.Second)
+
+			//login
+			login := map[string][]interface{}{
+				"emit": {"hello", map[string]string{
+					"id":     a.node.Name,
+					"secret": config.ApplicationConfig.Secret,
+				}},
+			}
+			err = conn.WriteJSON(login)
+			if err != nil {
+				return
+			}
+
+			//read info
+			go a.readLoop(conn)
+
+			select {
+			case <-latencyTicker.C:
+				if err = a.reportLatency(conn); err != nil {
+					a.logger.Warn("requested latency report failed: ", err)
+				}
+			case <-a.readyCh:
+				//登录成功，上传数据
+				if err = a.reportStats(conn); err != nil {
+					a.logger.Warn("stats info report failed: ", err)
+				}
+			}
+		case <-interrupt:
+			a.close(conn, readTicker, latencyTicker)
+			isInterrupt = true
+			return
+		}
+	}
+}
+
+func (a *App) readLoop(conn *connutil.ConnWrapper) {
+	for {
+		blob := json.RawMessage{}
+		if err := conn.ReadJSON(&blob); err != nil {
+			a.logger.Warn("received and decode message error: ", err)
+			return
+		}
+		// If the network packet is a system ping, respond to it directly
+		var ping string
+		if err := json.Unmarshal(blob, &ping); err == nil && strings.HasPrefix(ping, "primus::ping::") {
+			if err := conn.WriteJSON(strings.Replace(ping, "ping", "pong", -1)); err != nil {
+				a.logger.Warn("failed to respond to system ping message: ", err)
+				return
+			}
+			continue
+		}
+		// Not a system ping, try to decode an actual state message
+		var msg map[string][]interface{}
+		if err := json.Unmarshal(blob, &msg); err != nil {
+			a.logger.Warn("failed to decode message: ", err)
+			return
+		}
+
+		if len(msg["emit"]) == 0 {
+			a.logger.Warn("received message invalid: ", msg)
+			return
+		}
+		msgType, ok := msg["emit"][0].(string)
+		if !ok {
+			a.logger.Warn("received invalid message type: ", msg["emit"][0])
+			return
+		}
+		a.logger.Trace("received message type: ", msgType)
+
+		switch msgType {
+		case "ready":
+			//只有接收到了ready信息，才初始化获取数据
+			a.logger.Info("connect success!")
+			a.readyCh <- struct{}{}
+		case "un-authorization":
+			if len(msg["emit"]) >= 2 {
+				if errMsg, ok := msg["emit"][1].(string); ok {
+					a.logger.Warn(errMsg)
+				}
+			}
+			return
+		case "node-pong":
+			a.pongCh <- struct{}{}
+		}
+
+	}
+}
+
+func (a *App) reportLatency(conn *connutil.ConnWrapper) error {
+	start := time.Now()
+
+	ping := map[string][]interface{}{
+		"emit": {"node-ping", map[string]string{
+			"id":         config.ApplicationConfig.Name,
+			"clientTime": start.String(),
+		}},
+	}
+
+	if err := conn.WriteJSON(ping); err != nil {
+		return err
+	}
+	// Wait for the pong request to arrive back
+	select {
+	case <-a.pongCh:
+		// Pong delivered, report the latency
+	case <-time.After(10 * time.Second):
+		// MsgPing timeout, abort
+		return errors.New("ping timed out")
+	}
+	latency := strconv.Itoa(int((time.Since(start) / time.Duration(2)).Nanoseconds() / 1000000))
+
+	// Send back the measured latency
+	a.logger.Trace("sending measured latency: ", latency)
+
+	stats := map[string][]interface{}{
+		"emit": {"latency", map[string]string{
+			"id":      config.ApplicationConfig.Name,
+			"latency": latency,
+		}},
+	}
+	return conn.WriteJSON(stats)
+}
+
+func (a *App) reportStats(conn *connutil.ConnWrapper) error {
+	ethClient := ethutil.NewEthClient(config.ChainConfig.Url, config.ChainConfig.Timeout)
+	chain, err := ethClient.Chain()
+	if err != nil {
+		return err
+	}
+	c := chain.RemoteRpcClient
+	// peer count
+	peerCount, _ := c.PeerCount(context.Background())
+
+	// is active
+	active := false
+	if peerCount > 0 {
+		active = true
+	}
+
+	// gas price
+	gasPrice, _ := c.SuggestGasPrice(context.Background())
+
+	// is syncing
+	process, err := c.SyncProgress(context.Background())
+	syncing := false
+	if err == nil && process != nil {
+		progress := process.CurrentBlock - process.StartingBlock
+		total := process.HighestBlock - process.StartingBlock
+		if progress/total < 1 {
+			syncing = true
+		}
+	}
+
+	// latest block
+	latestBlock, err := c.BlockByNumber(context.Background(), nil)
+	block := model.Block{}
+	if err == nil {
+		block.Number = latestBlock.NumberU64()
+		block.Hash = latestBlock.Hash().String()
+		block.Difficulty = latestBlock.Difficulty().Uint64()
+		block.Time = latestBlock.Time()
+		//block.Transactions = latestBlock.Transactions()
+		//block.Uncles = latestBlock.Uncles()
+	}
+	pendingCount, _ := c.PendingTransactionCount(context.Background())
+
+	stats := model.Stats{
+		NodeInfo:  a.node,
+		Active:    active,
+		PeerCount: peerCount,
+		Pending:   pendingCount,
+		GasPrice:  gasPrice.Int64(),
+		Syncing:   syncing,
+		Block:     &block,
+	}
+	report := map[string][]interface{}{
+		"emit": {"stats", stats},
+	}
+	return conn.WriteJSON(report)
+}
+
+func (a *App) close(conn *connutil.ConnWrapper, readTicker, latencyTicker *time.Timer) {
+	if conn != nil {
+		_ = conn.Close()
+	}
+	if readTicker != nil {
+		_ = readTicker.Stop()
+	}
+	if latencyTicker != nil {
+		_ = latencyTicker.Stop()
+	}
+}