[project] Remove unused internal memprofiling, profiling, metrics service (#2679)
* [project] Remove unused internal memprofiling, profiling * [internal] Unused attack package * [internal] More dead metrics related codepull/2685/head
Edgar Aroutiounian
5 years ago
committed by
GitHub
parent
7da8ec0a02
commit
45135b21d6
@ -1,236 +0,0 @@ |
||||
package metrics |
||||
|
||||
import ( |
||||
"fmt" |
||||
"math" |
||||
"math/big" |
||||
"strconv" |
||||
|
||||
"github.com/ethereum/go-ethereum/rpc" |
||||
msg_pb "github.com/harmony-one/harmony/api/proto/message" |
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
"github.com/harmony-one/harmony/p2p" |
||||
"github.com/prometheus/client_golang/prometheus" |
||||
"github.com/prometheus/client_golang/prometheus/push" |
||||
) |
||||
|
||||
// Constants for metrics service.
|
||||
const ( |
||||
BalanceScale int = 18 |
||||
BalancePrecision int = 13 |
||||
ConnectionsNumberPush int = 0 |
||||
BlockHeightPush int = 1 |
||||
NodeBalancePush int = 2 |
||||
LastConsensusPush int = 3 |
||||
BlockRewardPush int = 4 |
||||
TxPoolPush int = 5 |
||||
IsLeaderPush int = 6 |
||||
metricsServicePortDifference = 2000 |
||||
) |
||||
|
||||
// Service is the struct for metrics service.
|
||||
type Service struct { |
||||
BLSPublicKey string |
||||
IP string |
||||
Port string |
||||
PushgatewayIP string |
||||
PushgatewayPort string |
||||
storage *Storage |
||||
pusher *push.Pusher |
||||
messageChan chan *msg_pb.Message |
||||
} |
||||
|
||||
// init vars for prometheus
|
||||
var ( |
||||
curTxPoolSize = uint64(0) |
||||
curBlockHeight = uint64(0) |
||||
curBlocks = uint64(0) |
||||
curBalance = big.NewInt(0) |
||||
curConnectionsNumber = 0 |
||||
curIsLeader = false |
||||
lastBlockReward = big.NewInt(0) |
||||
lastConsensusTime = int64(0) |
||||
metricsPush = make(chan int) |
||||
blockHeightGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "block_height", |
||||
Help: "Get current block height.", |
||||
}) |
||||
txPoolGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "tx_pool_size", |
||||
Help: "Get current tx pool size.", |
||||
}) |
||||
isLeaderGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "is_leader", |
||||
Help: "Is node a leader now.", |
||||
}) |
||||
blocksAcceptedGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "blocks_accepted", |
||||
Help: "Get accepted blocks.", |
||||
}) |
||||
connectionsNumberGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "connections_number", |
||||
Help: "Get current connections number for a node.", |
||||
}) |
||||
nodeBalanceGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "node_balance", |
||||
Help: "Get current node balance.", |
||||
}) |
||||
lastConsensusGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "last_consensus", |
||||
Help: "Get last consensus time.", |
||||
}) |
||||
blockRewardGauge = prometheus.NewGauge(prometheus.GaugeOpts{ |
||||
Name: "block_reward", |
||||
Help: "Get last block reward.", |
||||
}) |
||||
) |
||||
|
||||
// New returns metrics service.
|
||||
func New(selfPeer *p2p.Peer, blsPublicKey, pushgatewayIP, pushgatewayPort string) *Service { |
||||
return &Service{ |
||||
BLSPublicKey: blsPublicKey, |
||||
IP: selfPeer.IP, |
||||
Port: selfPeer.Port, |
||||
PushgatewayIP: pushgatewayIP, |
||||
PushgatewayPort: pushgatewayPort, |
||||
} |
||||
} |
||||
|
||||
// StartService starts metrics service.
|
||||
func (s *Service) StartService() { |
||||
utils.Logger().Info().Msg("Starting metrics service.") |
||||
s.Run() |
||||
} |
||||
|
||||
// StopService shutdowns metrics service.
|
||||
func (s *Service) StopService() { |
||||
utils.Logger().Info().Msg("Shutting down metrics service.") |
||||
metricsPush <- -1 |
||||
} |
||||
|
||||
// GetMetricsServicePort returns the port serving metrics service dashboard. This port is metricsServicePortDifference less than the node port.
|
||||
func GetMetricsServicePort(nodePort string) string { |
||||
if port, err := strconv.Atoi(nodePort); err == nil { |
||||
return fmt.Sprintf("%d", port-metricsServicePortDifference) |
||||
} |
||||
utils.Logger().Error().Msg("Error on parsing.") |
||||
return "" |
||||
} |
||||
|
||||
// Run is to run http serving metrics service.
|
||||
func (s *Service) Run() { |
||||
// Init local storage for metrics.
|
||||
s.storage = GetStorageInstance(s.IP, s.Port, true) |
||||
registry := prometheus.NewRegistry() |
||||
registry.MustRegister(blockHeightGauge, connectionsNumberGauge, nodeBalanceGauge, lastConsensusGauge, blockRewardGauge, blocksAcceptedGauge, txPoolGauge, isLeaderGauge) |
||||
|
||||
s.pusher = push.New("http://"+s.PushgatewayIP+":"+s.PushgatewayPort, "node_metrics").Gatherer(registry).Grouping("instance", s.IP+":"+s.Port).Grouping("bls_key", s.BLSPublicKey) |
||||
go s.PushMetrics() |
||||
} |
||||
|
||||
// FormatBalance formats big.Int balance with precision.
|
||||
func FormatBalance(balance *big.Int) float64 { |
||||
scaledBalance := new(big.Float).Quo(new(big.Float).SetInt(balance), new(big.Float).SetFloat64(math.Pow10(BalanceScale))) |
||||
floatBalance, _ := scaledBalance.Float64() |
||||
return floatBalance |
||||
} |
||||
|
||||
// UpdateBlockHeight updates block height.
|
||||
func UpdateBlockHeight(blockHeight uint64) { |
||||
blockHeightGauge.Set(float64(blockHeight)) |
||||
blocksAcceptedGauge.Set(float64(blockHeight) - float64(curBlockHeight)) |
||||
curBlockHeight = blockHeight |
||||
metricsPush <- BlockHeightPush |
||||
} |
||||
|
||||
// UpdateNodeBalance updates node balance.
|
||||
func UpdateNodeBalance(balance *big.Int) { |
||||
nodeBalanceGauge.Set(FormatBalance(balance)) |
||||
curBalance = balance |
||||
metricsPush <- NodeBalancePush |
||||
} |
||||
|
||||
// UpdateTxPoolSize updates tx pool size.
|
||||
func UpdateTxPoolSize(txPoolSize uint64) { |
||||
txPoolGauge.Set(float64(txPoolSize)) |
||||
curTxPoolSize = txPoolSize |
||||
metricsPush <- TxPoolPush |
||||
} |
||||
|
||||
// UpdateBlockReward updates block reward.
|
||||
func UpdateBlockReward(blockReward *big.Int) { |
||||
blockRewardGauge.Set(FormatBalance(blockReward)) |
||||
lastBlockReward = blockReward |
||||
metricsPush <- BlockRewardPush |
||||
} |
||||
|
||||
// UpdateLastConsensus updates last consensus time.
|
||||
func UpdateLastConsensus(consensusTime int64) { |
||||
lastConsensusGauge.Set(float64(consensusTime)) |
||||
lastConsensusTime = consensusTime |
||||
metricsPush <- LastConsensusPush |
||||
} |
||||
|
||||
// UpdateConnectionsNumber updates connections number.
|
||||
func UpdateConnectionsNumber(connectionsNumber int) { |
||||
connectionsNumberGauge.Set(float64(connectionsNumber)) |
||||
curConnectionsNumber = connectionsNumber |
||||
metricsPush <- ConnectionsNumberPush |
||||
} |
||||
|
||||
// UpdateIsLeader updates if node is a leader.
|
||||
func UpdateIsLeader(isLeader bool) { |
||||
if isLeader { |
||||
isLeaderGauge.Set(1.0) |
||||
} else { |
||||
isLeaderGauge.Set(0.0) |
||||
} |
||||
curIsLeader = isLeader |
||||
metricsPush <- IsLeaderPush |
||||
} |
||||
|
||||
// PushMetrics pushes metrics updates to prometheus pushgateway.
|
||||
func (s *Service) PushMetrics() { |
||||
for metricType := range metricsPush { |
||||
if metricType == -1 { |
||||
break |
||||
} |
||||
if err := s.pusher.Add(); err != nil { |
||||
utils.Logger().Error().Err(err).Msg("Could not push to a prometheus pushgateway.") |
||||
// No dump for now, not necessarily for metrics and consumes memory, doesn't restore from db anyway.
|
||||
/* switch metricType { |
||||
case ConnectionsNumberPush: |
||||
s.storage.Dump(curConnectionsNumber, ConnectionsNumberPrefix) |
||||
case BlockHeightPush: |
||||
s.storage.Dump(curBlockHeight, BlockHeightPrefix) |
||||
s.storage.Dump(curBlocks, BlocksPrefix) |
||||
case BlockRewardPush: |
||||
s.storage.Dump(lastBlockReward, BlockHeightPrefix) |
||||
case NodeBalancePush: |
||||
s.storage.Dump(curBalance, BalancePrefix) |
||||
case LastConsensusPush: |
||||
s.storage.Dump(lastConsensusTime, ConsensusTimePrefix) |
||||
case TxPoolPush: |
||||
s.storage.Dump(curTxPoolSize, TxPoolPrefix) |
||||
case IsLeaderPush: |
||||
s.storage.Dump(curIsLeader, IsLeaderPrefix) |
||||
}*/ |
||||
} |
||||
} |
||||
return |
||||
} |
||||
|
||||
// NotifyService notify service
|
||||
func (s *Service) NotifyService(params map[string]interface{}) { |
||||
return |
||||
} |
||||
|
||||
// SetMessageChan sets up message channel to service.
|
||||
func (s *Service) SetMessageChan(messageChan chan *msg_pb.Message) { |
||||
s.messageChan = messageChan |
||||
} |
||||
|
||||
// APIs for the services.
|
||||
func (s *Service) APIs() []rpc.API { |
||||
return nil |
||||
} |
||||
@ -1,102 +0,0 @@ |
||||
package metrics |
||||
|
||||
import ( |
||||
"fmt" |
||||
"os" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/ethereum/go-ethereum/ethdb" |
||||
"github.com/ethereum/go-ethereum/rlp" |
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
) |
||||
|
||||
// Constants for storage.
|
||||
const ( |
||||
BalancePrefix = "bap" |
||||
BlockHeightPrefix = "bhp" |
||||
BlocksPrefix = "bp" |
||||
BlockRewardPrefix = "brp" |
||||
ConnectionsNumberPrefix = "cnp" |
||||
ConsensusTimePrefix = "ltp" |
||||
IsLeaderPrefix = "ilp" |
||||
TxPoolPrefix = "tpp" |
||||
) |
||||
|
||||
// GetKey returns key by prefix and pushed time momemnt.
|
||||
func GetKey(prefix string, moment int64) string { |
||||
return fmt.Sprintf("%s_%d", prefix, moment) |
||||
} |
||||
|
||||
// storage instance
|
||||
var storage *Storage |
||||
var onceMetrics sync.Once |
||||
|
||||
// Storage storage dump the block info into leveldb.
|
||||
type Storage struct { |
||||
db *ethdb.LDBDatabase |
||||
} |
||||
|
||||
// GetStorageInstance returns attack model by using singleton pattern.
|
||||
func GetStorageInstance(ip, port string, remove bool) *Storage { |
||||
onceMetrics.Do(func() { |
||||
storage = &Storage{} |
||||
storage.Init(ip, port, remove) |
||||
}) |
||||
return storage |
||||
} |
||||
|
||||
// Init initializes storage.
|
||||
func (storage *Storage) Init(ip, port string, remove bool) { |
||||
dbFileName := "/tmp/db_metrics_" + ip + "_" + port |
||||
var err error |
||||
if remove { |
||||
var err = os.RemoveAll(dbFileName) |
||||
if err != nil { |
||||
utils.Logger().Error().Err(err).Msg("Failed to remove existing database files.") |
||||
} |
||||
} |
||||
if storage.db, err = ethdb.NewLDBDatabase(dbFileName, 0, 0); err != nil { |
||||
utils.Logger().Error().Err(err).Msg("Failed to create new database.") |
||||
} |
||||
} |
||||
|
||||
// GetDB returns the LDBDatabase of the storage.
|
||||
func (storage *Storage) GetDB() *ethdb.LDBDatabase { |
||||
return storage.db |
||||
} |
||||
|
||||
// Dump data into lvdb by value and prefix.
|
||||
func (storage *Storage) Dump(value interface{}, prefix string) error { |
||||
currentTime := time.Now().UnixNano() |
||||
utils.Logger().Info().Msgf("Store %s %v at time %d", prefix, value, currentTime) |
||||
batch := storage.db.NewBatch() |
||||
// Update database.
|
||||
if err := batch.Put([]byte(GetKey(prefix, currentTime)), []byte(fmt.Sprintf("%v", value.(interface{})))); err != nil { |
||||
utils.Logger().Warn().Err(err).Msgf("Cannot batch %s.", prefix) |
||||
return err |
||||
} |
||||
if err := batch.Write(); err != nil { |
||||
utils.Logger().Warn().Err(err).Msg("Cannot write batch.") |
||||
return err |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
// Read returns data list of a particular metric by since, until, prefix, interface.
|
||||
func (storage *Storage) Read(since, until int64, prefix string, varType interface{}) []interface{} { |
||||
dataList := make([]interface{}, 0) |
||||
for i := since; i <= until; i++ { |
||||
data, err := storage.db.Get([]byte(GetKey(prefix, i))) |
||||
if err != nil { |
||||
continue |
||||
} |
||||
decodedData := varType |
||||
if rlp.DecodeBytes(data, decodedData) != nil { |
||||
utils.Logger().Error().Msg("Error on getting data from db.") |
||||
os.Exit(1) |
||||
} |
||||
dataList = append(dataList, decodedData) |
||||
} |
||||
return dataList |
||||
} |
||||
@ -1,118 +0,0 @@ |
||||
package attack |
||||
|
||||
import ( |
||||
"math/rand" |
||||
"os" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
) |
||||
|
||||
// Constants used for attack model.
|
||||
const ( |
||||
DroppingTickDuration = 2 * time.Second |
||||
HitRate = 10 |
||||
DelayResponseDuration = 10 * time.Second |
||||
ViewIDThresholdMin = 10 |
||||
ViewIDThresholdMax = 100 |
||||
) |
||||
|
||||
// Type is the type of attack model.
|
||||
type Type byte |
||||
|
||||
// Constants of different attack models.
|
||||
const ( |
||||
KilledItself Type = iota |
||||
DelayResponse |
||||
IncorrectResponse |
||||
) |
||||
|
||||
// Model contains different models of attacking.
|
||||
type Model struct { |
||||
AttackEnabled bool |
||||
attackType Type |
||||
ViewIDThreshold uint64 |
||||
readyByConsensusThreshold bool |
||||
} |
||||
|
||||
var attackModel *Model |
||||
var once sync.Once |
||||
|
||||
// GetInstance returns attack model by using singleton pattern.
|
||||
func GetInstance() *Model { |
||||
once.Do(func() { |
||||
attackModel = &Model{} |
||||
attackModel.Init() |
||||
}) |
||||
return attackModel |
||||
} |
||||
|
||||
// Init initializes attack model.
|
||||
func (attack *Model) Init() { |
||||
attack.AttackEnabled = false |
||||
attack.readyByConsensusThreshold = false |
||||
} |
||||
|
||||
// SetAttackEnabled sets attack model enabled.
|
||||
func (attack *Model) SetAttackEnabled(AttackEnabled bool) { |
||||
attack.AttackEnabled = AttackEnabled |
||||
if AttackEnabled { |
||||
attack.attackType = Type(rand.Intn(3)) |
||||
attack.ViewIDThreshold = uint64(ViewIDThresholdMin + rand.Intn(ViewIDThresholdMax-ViewIDThresholdMin)) |
||||
} |
||||
} |
||||
|
||||
// Run runs enabled attacks.
|
||||
func (attack *Model) Run() { |
||||
attack.NodeKilledByItSelf() |
||||
attack.DelayResponse() |
||||
} |
||||
|
||||
// NodeKilledByItSelf runs killing itself attack
|
||||
func (attack *Model) NodeKilledByItSelf() { |
||||
if !attack.AttackEnabled || attack.attackType != KilledItself || !attack.readyByConsensusThreshold { |
||||
return |
||||
} |
||||
|
||||
if rand.Intn(HitRate) == 0 { |
||||
utils.Logger().Debug(). |
||||
Int("PID", os.Getpid()). |
||||
Msg("******************Killing myself******************") |
||||
os.Exit(1) |
||||
} |
||||
} |
||||
|
||||
// DelayResponse does attack by delaying response.
|
||||
func (attack *Model) DelayResponse() { |
||||
if !attack.AttackEnabled || attack.attackType != DelayResponse || !attack.readyByConsensusThreshold { |
||||
return |
||||
} |
||||
if rand.Intn(HitRate) == 0 { |
||||
utils.Logger().Debug(). |
||||
Int("PID", os.Getpid()). |
||||
Msg("******************Model: DelayResponse******************") |
||||
time.Sleep(DelayResponseDuration) |
||||
} |
||||
} |
||||
|
||||
// IncorrectResponse returns if the attack model enable incorrect responding.
|
||||
func (attack *Model) IncorrectResponse() bool { |
||||
if !attack.AttackEnabled || attack.attackType != IncorrectResponse || !attack.readyByConsensusThreshold { |
||||
return false |
||||
} |
||||
if rand.Intn(HitRate) == 0 { |
||||
utils.Logger().Debug(). |
||||
Int("PID", os.Getpid()). |
||||
Msg("******************Model: IncorrectResponse******************") |
||||
return true |
||||
} |
||||
return false |
||||
} |
||||
|
||||
// UpdateConsensusReady enables an attack type given the current viewID.
|
||||
func (attack *Model) UpdateConsensusReady(viewID uint64) { |
||||
if viewID > attack.ViewIDThreshold { |
||||
attack.readyByConsensusThreshold = true |
||||
} |
||||
} |
||||
@ -1,26 +0,0 @@ |
||||
package attack |
||||
|
||||
import ( |
||||
"testing" |
||||
|
||||
"github.com/stretchr/testify/assert" |
||||
) |
||||
|
||||
// Simple test for IncorrectResponse
|
||||
func TestIncorrectResponse(t *testing.T) { |
||||
GetInstance().SetAttackEnabled(false) |
||||
assert.False(t, GetInstance().IncorrectResponse(), "error") |
||||
GetInstance().SetAttackEnabled(true) |
||||
} |
||||
|
||||
// Simple test for UpdateConsensusReady
|
||||
func TestUpdateConsensusReady(t *testing.T) { |
||||
model := GetInstance() |
||||
model.NodeKilledByItSelf() |
||||
|
||||
model.UpdateConsensusReady(model.ViewIDThreshold - 1) |
||||
model.DelayResponse() |
||||
|
||||
model.UpdateConsensusReady(model.ViewIDThreshold + 1) |
||||
model.DelayResponse() |
||||
} |
||||
@ -1,136 +0,0 @@ |
||||
package memprofiling |
||||
|
||||
import ( |
||||
"fmt" |
||||
"net/http" |
||||
"reflect" |
||||
"runtime" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/fjl/memsize" |
||||
"github.com/fjl/memsize/memsizeui" |
||||
nodeconfig "github.com/harmony-one/harmony/internal/configs/node" |
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
) |
||||
|
||||
// Constants for mem profiling.
|
||||
const ( |
||||
MemProfilingPortDiff = 1000 |
||||
// Constants of for scanning mem size.
|
||||
memSizeScanTime = 30 * time.Second |
||||
// Run garbage collector every 30 minutes.
|
||||
gcTime = 10 * time.Minute |
||||
// Print out memstat every memStatTime.
|
||||
memStatTime = 300 * time.Second |
||||
) |
||||
|
||||
// MemProfiling is the struct to watch objects for memprofiling.
|
||||
type MemProfiling struct { |
||||
h *memsizeui.Handler |
||||
s *http.Server |
||||
observedObject map[string]interface{} |
||||
mu sync.Mutex |
||||
} |
||||
|
||||
// New returns MemProfiling object.
|
||||
func New() *MemProfiling { |
||||
return &MemProfiling{ |
||||
observedObject: map[string]interface{}{}, |
||||
h: new(memsizeui.Handler), |
||||
} |
||||
} |
||||
|
||||
// Config configures mem profiling.
|
||||
func (m *MemProfiling) Config() { |
||||
m.s = &http.Server{ |
||||
Addr: fmt.Sprintf("%s:%s", nodeconfig.GetDefaultConfig().IP, utils.GetPortFromDiff(nodeconfig.GetDefaultConfig().Port, MemProfilingPortDiff)), |
||||
Handler: m.h, |
||||
} |
||||
utils.Logger().Info(). |
||||
Str("port", utils.GetPortFromDiff(nodeconfig.GetDefaultConfig().Port, MemProfilingPortDiff)). |
||||
Msgf("running mem profiling") |
||||
} |
||||
|
||||
// Add adds variables to watch for profiling.
|
||||
func (m *MemProfiling) Add(name string, v interface{}) { |
||||
m.mu.Lock() |
||||
defer m.mu.Unlock() |
||||
if v != nil { |
||||
rv := reflect.ValueOf(v) |
||||
if !(rv.Kind() != reflect.Ptr || rv.IsNil()) { |
||||
m.h.Add(name, v) |
||||
m.observedObject[name] = v |
||||
} |
||||
} |
||||
} |
||||
|
||||
// Start starts profiling server.
|
||||
func (m *MemProfiling) Start() { |
||||
go m.s.ListenAndServe() |
||||
m.PeriodicallyScanMemSize() |
||||
} |
||||
|
||||
// Stop stops mem profiling.
|
||||
func (m *MemProfiling) Stop() { |
||||
m.s.Shutdown(nil) |
||||
} |
||||
|
||||
// PeriodicallyScanMemSize scans memsize of the observed objects every 30 seconds.
|
||||
func (m *MemProfiling) PeriodicallyScanMemSize() { |
||||
go func() { |
||||
// TODO ek – infinite loop; add shutdown/cleanup logic
|
||||
for { |
||||
select { |
||||
case <-time.After(memSizeScanTime): |
||||
m.mu.Lock() |
||||
m := GetMemProfiling() |
||||
for k, v := range m.observedObject { |
||||
s := memsize.Scan(v) |
||||
r := s.Report() |
||||
utils.Logger().Info().Msgf("memsize report for %s:\n %s", k, r) |
||||
} |
||||
m.mu.Unlock() |
||||
} |
||||
} |
||||
}() |
||||
} |
||||
|
||||
// MaybeCallGCPeriodically runs GC manually every gcTime minutes. This is one of the options to mitigate the OOM issue.
|
||||
func MaybeCallGCPeriodically() { |
||||
go func() { |
||||
// TODO ek – infinite loop; add shutdown/cleanup logic
|
||||
for { |
||||
select { |
||||
case <-time.After(gcTime): |
||||
PrintMemUsage("Memory stats before GC") |
||||
runtime.GC() |
||||
PrintMemUsage("Memory stats after GC") |
||||
} |
||||
} |
||||
}() |
||||
go func() { |
||||
// TODO ek – infinite loop; add shutdown/cleanup logic
|
||||
for { |
||||
select { |
||||
case <-time.After(memStatTime): |
||||
PrintMemUsage("Memory stats") |
||||
} |
||||
} |
||||
}() |
||||
} |
||||
|
||||
// PrintMemUsage prints memory usage.
|
||||
func PrintMemUsage(msg string) { |
||||
var m runtime.MemStats |
||||
runtime.ReadMemStats(&m) |
||||
utils.Logger().Info(). |
||||
Uint64("alloc", bToMb(m.Alloc)). |
||||
Uint64("totalalloc", bToMb(m.TotalAlloc)). |
||||
Uint64("sys", bToMb(m.Sys)). |
||||
Uint32("numgc", m.NumGC) |
||||
} |
||||
|
||||
func bToMb(b uint64) uint64 { |
||||
return b / 1024 / 1024 |
||||
} |
||||
@ -1,14 +0,0 @@ |
||||
package memprofiling |
||||
|
||||
import "sync" |
||||
|
||||
var singleton *MemProfiling |
||||
var once sync.Once |
||||
|
||||
// GetMemProfiling returns a pointer of MemProfiling.
|
||||
func GetMemProfiling() *MemProfiling { |
||||
once.Do(func() { |
||||
singleton = New() |
||||
}) |
||||
return singleton |
||||
} |
||||
@ -1,96 +0,0 @@ |
||||
package profiler |
||||
|
||||
import ( |
||||
"bytes" |
||||
"encoding/json" |
||||
"fmt" |
||||
"net/http" |
||||
"os" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
"github.com/shirou/gopsutil/process" |
||||
) |
||||
|
||||
// Profiler is the profiler data structure.
|
||||
type Profiler struct { |
||||
pid int32 |
||||
shardID uint32 |
||||
MetricsReportURL string |
||||
// Internal
|
||||
proc *process.Process |
||||
} |
||||
|
||||
var singleton *Profiler |
||||
var once sync.Once |
||||
|
||||
// GetProfiler returns a pointer of Profiler.
|
||||
// TODO: This should be a New method.
|
||||
func GetProfiler() *Profiler { |
||||
once.Do(func() { |
||||
singleton = &Profiler{} |
||||
}) |
||||
return singleton |
||||
} |
||||
|
||||
// Config configurates Profiler.
|
||||
func (profiler *Profiler) Config(shardID uint32, metricsReportURL string) { |
||||
profiler.pid = int32(os.Getpid()) |
||||
profiler.shardID = shardID |
||||
profiler.MetricsReportURL = metricsReportURL |
||||
} |
||||
|
||||
// LogMemory logs memory.
|
||||
func (profiler *Profiler) LogMemory() { |
||||
// TODO ek – infinite loop; add shutdown/cleanup logic
|
||||
for { |
||||
// log mem usage
|
||||
info, _ := profiler.proc.MemoryInfo() |
||||
memMap, _ := profiler.proc.MemoryMaps(false) |
||||
loggedMemMap := "" |
||||
for _, mems := range *memMap { |
||||
loggedMemMap = fmt.Sprintf("%v; %v", loggedMemMap, mems) |
||||
} |
||||
utils.Logger().Info(). |
||||
Str("info", info.String()). |
||||
Str("map", loggedMemMap). |
||||
Uint32("shardID", profiler.shardID). |
||||
Msg("Mem Report") |
||||
|
||||
time.Sleep(3 * time.Second) |
||||
} |
||||
} |
||||
|
||||
// LogCPU logs CPU metrics.
|
||||
func (profiler *Profiler) LogCPU() { |
||||
// TODO ek – infinite loop; add shutdown/cleanup logic
|
||||
for { |
||||
// log cpu usage
|
||||
percent, _ := profiler.proc.CPUPercent() |
||||
times, _ := profiler.proc.Times() |
||||
utils.Logger().Info(). |
||||
Float64("percent", percent). |
||||
Str("times", times.String()). |
||||
Uint32("shardID", profiler.shardID). |
||||
Msg("CPU Report") |
||||
|
||||
time.Sleep(3 * time.Second) |
||||
} |
||||
} |
||||
|
||||
// LogMetrics logs metrics.
|
||||
func (profiler *Profiler) LogMetrics(metrics map[string]interface{}) { |
||||
jsonValue, _ := json.Marshal(metrics) |
||||
rsp, err := http.Post(profiler.MetricsReportURL, "application/json", bytes.NewBuffer(jsonValue)) |
||||
if err == nil { |
||||
defer rsp.Body.Close() |
||||
} |
||||
} |
||||
|
||||
// Start starts profiling.
|
||||
func (profiler *Profiler) Start() { |
||||
profiler.proc, _ = process.NewProcess(profiler.pid) |
||||
go profiler.LogCPU() |
||||
go profiler.LogMemory() |
||||
} |
||||
@ -1,6 +0,0 @@ |
||||
package utils |
||||
|
||||
// BToMb ...
|
||||
func BToMb(b uint64) uint64 { |
||||
return b / 1024 / 1024 |
||||
} |
||||
@ -1,19 +0,0 @@ |
||||
package utils |
||||
|
||||
import ( |
||||
"testing" |
||||
|
||||
"github.com/stretchr/testify/assert" |
||||
) |
||||
|
||||
// Test for BToMb.
|
||||
func TestBToMb(t *testing.T) { |
||||
a := uint64(1024*1024 + 1) |
||||
assert.Equal(t, BToMb(a), uint64(1), "should be equal to 1") |
||||
|
||||
a = uint64(1024*1024 - 1) |
||||
assert.Equal(t, BToMb(a), uint64(0), "should be equal to 0") |
||||
|
||||
a = uint64(1024 * 1024) |
||||
assert.Equal(t, BToMb(a), uint64(1), "should be equal to 0") |
||||
} |
||||
@ -1,92 +0,0 @@ |
||||
package node |
||||
|
||||
import ( |
||||
"time" |
||||
|
||||
metrics "github.com/harmony-one/harmony/api/service/metrics" |
||||
"github.com/harmony-one/harmony/internal/utils" |
||||
) |
||||
|
||||
// UpdateBlockHeightForMetrics updates block height for metrics service.
|
||||
func (node *Node) UpdateBlockHeightForMetrics(prevBlockHeight uint64) uint64 { |
||||
curBlock := node.Blockchain().CurrentBlock() |
||||
curBlockHeight := curBlock.NumberU64() |
||||
if curBlockHeight == prevBlockHeight { |
||||
return prevBlockHeight |
||||
} |
||||
utils.Logger().Info().Msgf("Updating metrics block height %d", curBlockHeight) |
||||
metrics.UpdateBlockHeight(curBlockHeight) |
||||
blockReward := node.Consensus.GetBlockReward() |
||||
if blockReward != nil { |
||||
utils.Logger().Info().Msgf("Updating metrics block reward %d", blockReward.Uint64()) |
||||
metrics.UpdateBlockReward(blockReward) |
||||
} |
||||
return curBlockHeight |
||||
} |
||||
|
||||
// UpdateConnectionsNumberForMetrics uppdates connections number for metrics service.
|
||||
func (node *Node) UpdateConnectionsNumberForMetrics(prevNumPeers int) int { |
||||
curNumPeers := node.numPeers |
||||
if curNumPeers == prevNumPeers { |
||||
return prevNumPeers |
||||
} |
||||
utils.Logger().Info().Msgf("Updating metrics connections number %d", curNumPeers) |
||||
metrics.UpdateConnectionsNumber(curNumPeers) |
||||
return curNumPeers |
||||
} |
||||
|
||||
// UpdateTxPoolSizeForMetrics updates tx pool size for metrics service.
|
||||
func (node *Node) UpdateTxPoolSizeForMetrics(txPoolSize uint64) { |
||||
utils.Logger().Info().Msgf("Updating metrics tx pool size %d", txPoolSize) |
||||
metrics.UpdateTxPoolSize(txPoolSize) |
||||
} |
||||
|
||||
// UpdateBalanceForMetrics uppdates node balance for metrics service.
|
||||
func (node *Node) UpdateBalanceForMetrics() { |
||||
for _, addr := range node.Consensus.SelfAddresses { |
||||
curBalance, err := node.GetBalanceOfAddress(addr) |
||||
if err != nil { |
||||
return |
||||
} |
||||
utils.Logger().Info().Msgf("Updating metrics node balance %d", curBalance.Uint64()) |
||||
metrics.UpdateNodeBalance(curBalance) |
||||
} |
||||
} |
||||
|
||||
// UpdateLastConsensusTimeForMetrics uppdates last consensus reached time for metrics service.
|
||||
func (node *Node) UpdateLastConsensusTimeForMetrics(prevLastConsensusTime int64) int64 { |
||||
lastConsensusTime := node.lastConsensusTime |
||||
if lastConsensusTime == prevLastConsensusTime { |
||||
return prevLastConsensusTime |
||||
} |
||||
utils.Logger().Info().Msgf("Updating metrics last consensus time reached %d", lastConsensusTime) |
||||
metrics.UpdateLastConsensus(lastConsensusTime) |
||||
return lastConsensusTime |
||||
} |
||||
|
||||
// UpdateIsLeaderForMetrics updates if node is a leader now for metrics serivce.
|
||||
func (node *Node) UpdateIsLeaderForMetrics() { |
||||
if node.Consensus.LeaderPubKey.SerializeToHexStr() == node.Consensus.PubKey.SerializeToHexStr() { |
||||
utils.Logger().Info().Msgf("Node %s is a leader now", node.Consensus.PubKey.SerializeToHexStr()) |
||||
metrics.UpdateIsLeader(true) |
||||
} else { |
||||
utils.Logger().Info().Msgf("Node %s is not a leader now", node.Consensus.PubKey.SerializeToHexStr()) |
||||
metrics.UpdateIsLeader(false) |
||||
} |
||||
} |
||||
|
||||
// CollectMetrics collects metrics: block height, connections number, node balance, block reward, last consensus, accepted blocks.
|
||||
func (node *Node) CollectMetrics() { |
||||
utils.Logger().Info().Msg("[Metrics Service] Update metrics") |
||||
prevNumPeers := 0 |
||||
prevBlockHeight := uint64(0) |
||||
prevLastConsensusTime := int64(0) |
||||
for range time.Tick(100 * time.Millisecond) { |
||||
prevBlockHeight = node.UpdateBlockHeightForMetrics(prevBlockHeight) |
||||
prevNumPeers = node.UpdateConnectionsNumberForMetrics(prevNumPeers) |
||||
prevLastConsensusTime = node.UpdateLastConsensusTimeForMetrics(prevLastConsensusTime) |
||||
node.UpdateBalanceForMetrics() |
||||
node.UpdateTxPoolSizeForMetrics(node.TxPool.GetTxPoolSize()) |
||||
node.UpdateIsLeaderForMetrics() |
||||
} |
||||
} |
||||
Loading…
Reference in new issue