Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
whqwjb
go-ethereum
提交
ee445a2b
G
go-ethereum
项目概览
whqwjb
/
go-ethereum
与 Fork 源项目一致
从无法访问的项目Fork
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
G
go-ethereum
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
ee445a2b
编写于
12月 12, 2016
作者:
P
Péter Szilágyi
提交者:
GitHub
12月 12, 2016
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #3425 from karalabe/netstats-time-fixup
netstats: time and block history
上级
4f9ccdd7
b2c226cb
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
197 addition
and
38 deletion
+197
-38
ethstats/ethstats.go
ethstats/ethstats.go
+197
-38
未找到文件。
ethstats/ethstats.go
浏览文件 @
ee445a2b
...
@@ -25,6 +25,7 @@ import (
...
@@ -25,6 +25,7 @@ import (
"regexp"
"regexp"
"runtime"
"runtime"
"strconv"
"strconv"
"strings"
"time"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common"
...
@@ -41,6 +42,10 @@ import (
...
@@ -41,6 +42,10 @@ import (
"golang.org/x/net/websocket"
"golang.org/x/net/websocket"
)
)
// historyUpdateRange is the number of blocks a node should report upon login or
// history request.
const
historyUpdateRange
=
50
// Service implements an Ethereum netstats reporting daemon that pushes local
// Service implements an Ethereum netstats reporting daemon that pushes local
// chain statistics up to a monitoring server.
// chain statistics up to a monitoring server.
type
Service
struct
{
type
Service
struct
{
...
@@ -53,6 +58,9 @@ type Service struct {
...
@@ -53,6 +58,9 @@ type Service struct {
node
string
// Name of the node to display on the monitoring page
node
string
// Name of the node to display on the monitoring page
pass
string
// Password to authorize access to the monitoring page
pass
string
// Password to authorize access to the monitoring page
host
string
// Remote address of the monitoring service
host
string
// Remote address of the monitoring service
pongCh
chan
struct
{}
// Pong notifications are fed into this channel
histCh
chan
[]
uint64
// History request block numbers are fed into this channel
}
}
// New returns a monitoring service ready for stats reporting.
// New returns a monitoring service ready for stats reporting.
...
@@ -65,11 +73,13 @@ func New(url string, ethServ *eth.Ethereum, lesServ *les.LightEthereum) (*Servic
...
@@ -65,11 +73,13 @@ func New(url string, ethServ *eth.Ethereum, lesServ *les.LightEthereum) (*Servic
}
}
// Assemble and return the stats service
// Assemble and return the stats service
return
&
Service
{
return
&
Service
{
eth
:
ethServ
,
eth
:
ethServ
,
les
:
lesServ
,
les
:
lesServ
,
node
:
parts
[
1
],
node
:
parts
[
1
],
pass
:
parts
[
3
],
pass
:
parts
[
3
],
host
:
parts
[
4
],
host
:
parts
[
4
],
pongCh
:
make
(
chan
struct
{}),
histCh
:
make
(
chan
[]
uint64
,
1
),
},
nil
},
nil
}
}
...
@@ -115,7 +125,11 @@ func (s *Service) loop() {
...
@@ -115,7 +125,11 @@ func (s *Service) loop() {
// Loop reporting until termination
// Loop reporting until termination
for
{
for
{
// Establish a websocket connection to the server and authenticate the node
// Establish a websocket connection to the server and authenticate the node
conn
,
err
:=
websocket
.
Dial
(
fmt
.
Sprintf
(
"wss://%s/api"
,
s
.
host
),
""
,
"http://localhost/"
)
url
:=
fmt
.
Sprintf
(
"%s/api"
,
s
.
host
)
if
!
strings
.
Contains
(
url
,
"://"
)
{
url
=
"wss://"
+
url
}
conn
,
err
:=
websocket
.
Dial
(
url
,
""
,
"http://localhost/"
)
if
err
!=
nil
{
if
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Stats server unreachable: %v"
,
err
)
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Stats server unreachable: %v"
,
err
)
time
.
Sleep
(
10
*
time
.
Second
)
time
.
Sleep
(
10
*
time
.
Second
)
...
@@ -130,22 +144,34 @@ func (s *Service) loop() {
...
@@ -130,22 +144,34 @@ func (s *Service) loop() {
time
.
Sleep
(
10
*
time
.
Second
)
time
.
Sleep
(
10
*
time
.
Second
)
continue
continue
}
}
if
err
=
s
.
report
(
in
,
out
);
err
!=
nil
{
go
s
.
readLoop
(
conn
,
in
)
// Send the initial stats so our node looks decent from the get go
if
err
=
s
.
report
(
out
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Initial stats report failed: %v"
,
err
)
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Initial stats report failed: %v"
,
err
)
conn
.
Close
()
conn
.
Close
()
continue
continue
}
}
if
err
=
s
.
reportHistory
(
out
,
nil
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"History report failed: %v"
,
err
)
conn
.
Close
()
continue
}
// Keep sending status updates until the connection breaks
// Keep sending status updates until the connection breaks
fullReport
:=
time
.
NewTicker
(
15
*
time
.
Second
)
fullReport
:=
time
.
NewTicker
(
15
*
time
.
Second
)
for
err
==
nil
{
for
err
==
nil
{
select
{
select
{
case
<-
fullReport
.
C
:
case
<-
fullReport
.
C
:
if
err
=
s
.
report
(
in
,
out
);
err
!=
nil
{
if
err
=
s
.
report
(
out
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Full stats report failed: %v"
,
err
)
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Full stats report failed: %v"
,
err
)
}
}
case
head
:=
<-
headSub
.
Chan
()
:
case
list
:=
<-
s
.
histCh
:
if
head
==
nil
{
// node stopped
if
err
=
s
.
reportHistory
(
out
,
list
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Block history report failed: %v"
,
err
)
}
case
head
,
ok
:=
<-
headSub
.
Chan
()
:
if
!
ok
{
// node stopped
conn
.
Close
()
conn
.
Close
()
return
return
}
}
...
@@ -155,8 +181,8 @@ func (s *Service) loop() {
...
@@ -155,8 +181,8 @@ func (s *Service) loop() {
if
err
=
s
.
reportPending
(
out
);
err
!=
nil
{
if
err
=
s
.
reportPending
(
out
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Post-block transaction stats report failed: %v"
,
err
)
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Post-block transaction stats report failed: %v"
,
err
)
}
}
case
ev
:=
<-
txSub
.
Chan
()
:
case
_
,
ok
:=
<-
txSub
.
Chan
()
:
if
ev
==
nil
{
// node stopped
if
!
ok
{
// node stopped
conn
.
Close
()
conn
.
Close
()
return
return
}
}
...
@@ -178,6 +204,76 @@ func (s *Service) loop() {
...
@@ -178,6 +204,76 @@ func (s *Service) loop() {
}
}
}
}
// readLoop loops as long as the connection is alive and retrieves data packets
// from the network socket. If any of them match an active request, it forwards
// it, if they themselves are requests it initiates a reply, and lastly it drops
// unknown packets.
func
(
s
*
Service
)
readLoop
(
conn
*
websocket
.
Conn
,
in
*
json
.
Decoder
)
{
// If the read loop exists, close the connection
defer
conn
.
Close
()
for
{
// Retrieve the next generic network packet and bail out on error
var
msg
map
[
string
][]
interface
{}
if
err
:=
in
.
Decode
(
&
msg
);
err
!=
nil
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Failed to decode stats server message: %v"
,
err
)
return
}
if
len
(
msg
[
"emit"
])
==
0
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Stats server sent non-broadcast: %v"
,
msg
)
return
}
command
,
ok
:=
msg
[
"emit"
][
0
]
.
(
string
)
if
!
ok
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Invalid stats server message type: %v"
,
msg
[
"emit"
][
0
])
return
}
// If the message is a ping reply, deliver (someone must be listening!)
if
len
(
msg
[
"emit"
])
==
2
&&
command
==
"node-pong"
{
select
{
case
s
.
pongCh
<-
struct
{}{}
:
// Pong delivered, continue listening
continue
default
:
// Ping routine dead, abort
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Stats server pinger seems to have died"
)
return
}
}
// If the message is a history request, forward to the event processor
if
len
(
msg
[
"emit"
])
==
2
&&
command
==
"history"
{
// Make sure the request is valid and doesn't crash us
request
,
ok
:=
msg
[
"emit"
][
1
]
.
(
map
[
string
]
interface
{})
if
!
ok
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Invalid history request: %v"
,
msg
[
"emit"
][
1
])
return
}
list
,
ok
:=
request
[
"list"
]
.
([]
interface
{})
if
!
ok
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Invalid history block list: %v"
,
request
[
"list"
])
return
}
// Convert the block number list to an integer list
numbers
:=
make
([]
uint64
,
len
(
list
))
for
i
,
num
:=
range
list
{
n
,
ok
:=
num
.
(
float64
)
if
!
ok
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Invalid history block number: %v"
,
num
)
return
}
numbers
[
i
]
=
uint64
(
n
)
}
select
{
case
s
.
histCh
<-
numbers
:
continue
default
:
}
}
// Report anything else and continue
glog
.
V
(
logger
.
Info
)
.
Infof
(
"Unknown stats message: %v"
,
msg
)
}
}
// nodeInfo is the collection of metainformation about a node that is displayed
// nodeInfo is the collection of metainformation about a node that is displayed
// on the monitoring page.
// on the monitoring page.
type
nodeInfo
struct
{
type
nodeInfo
struct
{
...
@@ -190,6 +286,7 @@ type nodeInfo struct {
...
@@ -190,6 +286,7 @@ type nodeInfo struct {
Os
string
`json:"os"`
Os
string
`json:"os"`
OsVer
string
`json:"os_v"`
OsVer
string
`json:"os_v"`
Client
string
`json:"client"`
Client
string
`json:"client"`
History
bool
`json:"canUpdateHistory"`
}
}
// authMsg is the authentication infos needed to login to a monitoring server.
// authMsg is the authentication infos needed to login to a monitoring server.
...
@@ -224,6 +321,7 @@ func (s *Service) login(in *json.Decoder, out *json.Encoder) error {
...
@@ -224,6 +321,7 @@ func (s *Service) login(in *json.Decoder, out *json.Encoder) error {
Os
:
runtime
.
GOOS
,
Os
:
runtime
.
GOOS
,
OsVer
:
runtime
.
GOARCH
,
OsVer
:
runtime
.
GOARCH
,
Client
:
"0.1.1"
,
Client
:
"0.1.1"
,
History
:
true
,
},
},
Secret
:
s
.
pass
,
Secret
:
s
.
pass
,
}
}
...
@@ -244,8 +342,8 @@ func (s *Service) login(in *json.Decoder, out *json.Encoder) error {
...
@@ -244,8 +342,8 @@ func (s *Service) login(in *json.Decoder, out *json.Encoder) error {
// report collects all possible data to report and send it to the stats server.
// report collects all possible data to report and send it to the stats server.
// This should only be used on reconnects or rarely to avoid overloading the
// This should only be used on reconnects or rarely to avoid overloading the
// server. Use the individual methods for reporting subscribed events.
// server. Use the individual methods for reporting subscribed events.
func
(
s
*
Service
)
report
(
in
*
json
.
Decoder
,
out
*
json
.
Encoder
)
error
{
func
(
s
*
Service
)
report
(
out
*
json
.
Encoder
)
error
{
if
err
:=
s
.
reportLatency
(
in
,
out
);
err
!=
nil
{
if
err
:=
s
.
reportLatency
(
out
);
err
!=
nil
{
return
err
return
err
}
}
if
err
:=
s
.
reportBlock
(
out
,
nil
);
err
!=
nil
{
if
err
:=
s
.
reportBlock
(
out
,
nil
);
err
!=
nil
{
...
@@ -262,7 +360,7 @@ func (s *Service) report(in *json.Decoder, out *json.Encoder) error {
...
@@ -262,7 +360,7 @@ func (s *Service) report(in *json.Decoder, out *json.Encoder) error {
// reportLatency sends a ping request to the server, measures the RTT time and
// reportLatency sends a ping request to the server, measures the RTT time and
// finally sends a latency update.
// finally sends a latency update.
func
(
s
*
Service
)
reportLatency
(
in
*
json
.
Decoder
,
out
*
json
.
Encoder
)
error
{
func
(
s
*
Service
)
reportLatency
(
out
*
json
.
Encoder
)
error
{
// Send the current time to the ethstats server
// Send the current time to the ethstats server
start
:=
time
.
Now
()
start
:=
time
.
Now
()
...
@@ -276,9 +374,12 @@ func (s *Service) reportLatency(in *json.Decoder, out *json.Encoder) error {
...
@@ -276,9 +374,12 @@ func (s *Service) reportLatency(in *json.Decoder, out *json.Encoder) error {
return
err
return
err
}
}
// Wait for the pong request to arrive back
// Wait for the pong request to arrive back
var
pong
map
[
string
][]
interface
{}
select
{
if
err
:=
in
.
Decode
(
&
pong
);
err
!=
nil
||
len
(
pong
[
"emit"
])
!=
2
||
pong
[
"emit"
][
0
]
.
(
string
)
!=
"node-pong"
{
case
<-
s
.
pongCh
:
return
errors
.
New
(
"unexpected ping reply"
)
// Pong delivered, report the latency
case
<-
time
.
After
(
3
*
time
.
Second
)
:
// Ping timeout, abort
return
errors
.
New
(
"ping timed out"
)
}
}
// Send back the measured latency
// Send back the measured latency
latency
:=
map
[
string
][]
interface
{}{
latency
:=
map
[
string
][]
interface
{}{
...
@@ -297,6 +398,7 @@ func (s *Service) reportLatency(in *json.Decoder, out *json.Encoder) error {
...
@@ -297,6 +398,7 @@ func (s *Service) reportLatency(in *json.Decoder, out *json.Encoder) error {
type
blockStats
struct
{
type
blockStats
struct
{
Number
*
big
.
Int
`json:"number"`
Number
*
big
.
Int
`json:"number"`
Hash
common
.
Hash
`json:"hash"`
Hash
common
.
Hash
`json:"hash"`
Timestamp
*
big
.
Int
`json:"timestamp"`
Miner
common
.
Address
`json:"miner"`
Miner
common
.
Address
`json:"miner"`
GasUsed
*
big
.
Int
`json:"gasUsed"`
GasUsed
*
big
.
Int
`json:"gasUsed"`
GasLimit
*
big
.
Int
`json:"gasLimit"`
GasLimit
*
big
.
Int
`json:"gasLimit"`
...
@@ -330,9 +432,26 @@ func (s uncleStats) MarshalJSON() ([]byte, error) {
...
@@ -330,9 +432,26 @@ func (s uncleStats) MarshalJSON() ([]byte, error) {
// reportBlock retrieves the current chain head and repors it to the stats server.
// reportBlock retrieves the current chain head and repors it to the stats server.
func
(
s
*
Service
)
reportBlock
(
out
*
json
.
Encoder
,
block
*
types
.
Block
)
error
{
func
(
s
*
Service
)
reportBlock
(
out
*
json
.
Encoder
,
block
*
types
.
Block
)
error
{
// Gather the head block infos from the local blockchain
// Assemble the block stats report and send it to the server
stats
:=
map
[
string
]
interface
{}{
"id"
:
s
.
node
,
"block"
:
s
.
assembleBlockStats
(
block
),
}
report
:=
map
[
string
][]
interface
{}{
"emit"
:
[]
interface
{}{
"block"
,
stats
},
}
if
err
:=
out
.
Encode
(
report
);
err
!=
nil
{
return
err
}
return
nil
}
// assembleBlockStats retrieves any required metadata to report a single block
// and assembles the block stats. If block is nil, the current head is processed.
func
(
s
*
Service
)
assembleBlockStats
(
block
*
types
.
Block
)
*
blockStats
{
// Gather the block infos from the local blockchain
var
(
var
(
head
*
types
.
Header
head
er
*
types
.
Header
td
*
big
.
Int
td
*
big
.
Int
txs
[]
*
types
.
Transaction
txs
[]
*
types
.
Transaction
uncles
[]
*
types
.
Header
uncles
[]
*
types
.
Header
...
@@ -342,37 +461,77 @@ func (s *Service) reportBlock(out *json.Encoder, block *types.Block) error {
...
@@ -342,37 +461,77 @@ func (s *Service) reportBlock(out *json.Encoder, block *types.Block) error {
if
block
==
nil
{
if
block
==
nil
{
block
=
s
.
eth
.
BlockChain
()
.
CurrentBlock
()
block
=
s
.
eth
.
BlockChain
()
.
CurrentBlock
()
}
}
head
=
block
.
Header
()
head
er
=
block
.
Header
()
td
=
s
.
eth
.
BlockChain
()
.
GetTd
(
head
.
Hash
(),
head
.
Number
.
Uint64
())
td
=
s
.
eth
.
BlockChain
()
.
GetTd
(
head
er
.
Hash
(),
header
.
Number
.
Uint64
())
txs
=
block
.
Transactions
()
txs
=
block
.
Transactions
()
uncles
=
block
.
Uncles
()
uncles
=
block
.
Uncles
()
}
else
{
}
else
{
// Light nodes would need on-demand lookups for transactions/uncles, skip
// Light nodes would need on-demand lookups for transactions/uncles, skip
if
block
!=
nil
{
if
block
!=
nil
{
head
=
block
.
Header
()
header
=
block
.
Header
()
}
else
{
header
=
s
.
les
.
BlockChain
()
.
CurrentHeader
()
}
td
=
s
.
les
.
BlockChain
()
.
GetTd
(
header
.
Hash
(),
header
.
Number
.
Uint64
())
}
// Assemble and return the block stats
return
&
blockStats
{
Number
:
header
.
Number
,
Hash
:
header
.
Hash
(),
Timestamp
:
header
.
Time
,
Miner
:
header
.
Coinbase
,
GasUsed
:
new
(
big
.
Int
)
.
Set
(
header
.
GasUsed
),
GasLimit
:
new
(
big
.
Int
)
.
Set
(
header
.
GasLimit
),
Diff
:
header
.
Difficulty
.
String
(),
TotalDiff
:
td
.
String
(),
Txs
:
txs
,
Uncles
:
uncles
,
}
}
// reportHistory retrieves the most recent batch of blocks and reports it to the
// stats server.
func
(
s
*
Service
)
reportHistory
(
out
*
json
.
Encoder
,
list
[]
uint64
)
error
{
// Figure out the indexes that need reporting
indexes
:=
make
([]
uint64
,
0
,
historyUpdateRange
)
if
len
(
list
)
>
0
{
// Specific indexes requested, send them back in particular
for
_
,
idx
:=
range
list
{
indexes
=
append
(
indexes
,
idx
)
}
}
else
{
// No indexes requested, send back the top ones
var
head
*
types
.
Header
if
s
.
eth
!=
nil
{
head
=
s
.
eth
.
BlockChain
()
.
CurrentHeader
()
}
else
{
}
else
{
head
=
s
.
les
.
BlockChain
()
.
CurrentHeader
()
head
=
s
.
les
.
BlockChain
()
.
CurrentHeader
()
}
}
td
=
s
.
les
.
BlockChain
()
.
GetTd
(
head
.
Hash
(),
head
.
Number
.
Uint64
())
start
:=
head
.
Number
.
Int64
()
-
historyUpdateRange
if
start
<
0
{
start
=
0
}
for
i
:=
uint64
(
start
);
i
<=
head
.
Number
.
Uint64
();
i
++
{
indexes
=
append
(
indexes
,
i
)
}
}
}
// Assemble the block stats report and send it to the server
// Gather the batch of blocks to report
history
:=
make
([]
*
blockStats
,
len
(
indexes
))
for
i
,
number
:=
range
indexes
{
if
s
.
eth
!=
nil
{
history
[
i
]
=
s
.
assembleBlockStats
(
s
.
eth
.
BlockChain
()
.
GetBlockByNumber
(
number
))
}
else
{
history
[
i
]
=
s
.
assembleBlockStats
(
types
.
NewBlockWithHeader
(
s
.
les
.
BlockChain
()
.
GetHeaderByNumber
(
number
)))
}
}
// Assemble the history report and send it to the server
stats
:=
map
[
string
]
interface
{}{
stats
:=
map
[
string
]
interface
{}{
"id"
:
s
.
node
,
"id"
:
s
.
node
,
"block"
:
&
blockStats
{
"history"
:
history
,
Number
:
head
.
Number
,
Hash
:
head
.
Hash
(),
Miner
:
head
.
Coinbase
,
GasUsed
:
new
(
big
.
Int
)
.
Set
(
head
.
GasUsed
),
GasLimit
:
new
(
big
.
Int
)
.
Set
(
head
.
GasLimit
),
Diff
:
head
.
Difficulty
.
String
(),
TotalDiff
:
td
.
String
(),
Txs
:
txs
,
Uncles
:
uncles
,
},
}
}
report
:=
map
[
string
][]
interface
{}{
report
:=
map
[
string
][]
interface
{}{
"emit"
:
[]
interface
{}{
"
block
"
,
stats
},
"emit"
:
[]
interface
{}{
"
history
"
,
stats
},
}
}
if
err
:=
out
.
Encode
(
report
);
err
!=
nil
{
if
err
:=
out
.
Encode
(
report
);
err
!=
nil
{
return
err
return
err
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录