From 6a614171a81a77d54d4e1e90bd5d420a2b72a4f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kurk=C3=B3=20Mih=C3=A1ly?= Date: Fri, 2 Mar 2018 18:05:39 +0200 Subject: [PATCH] dashboard, ethdb: implement proper diskIO metering function --- dashboard/dashboard.go | 8 ++--- ethdb/database.go | 67 +++++++++++++++++++++++++++++++++++------- 2 files changed, 61 insertions(+), 14 deletions(-) diff --git a/dashboard/dashboard.go b/dashboard/dashboard.go index 2ca795187f..0193794778 100644 --- a/dashboard/dashboard.go +++ b/dashboard/dashboard.go @@ -281,8 +281,8 @@ func (db *Dashboard) collectData() { prevNetworkEgress = metrics.DefaultRegistry.Get("p2p/OutboundTraffic").(metrics.Meter).Count() prevProcessCPUTime = getProcessCPUTime() prevSystemCPUUsage = systemCPUUsage - prevDiskRead = metrics.DefaultRegistry.Get("eth/db/chaindata/compact/input").(metrics.Meter).Count() - prevDiskWrite = metrics.DefaultRegistry.Get("eth/db/chaindata/compact/output").(metrics.Meter).Count() + prevDiskRead = metrics.DefaultRegistry.Get("eth/db/chaindata/disk/read").(metrics.Meter).Count() + prevDiskWrite = metrics.DefaultRegistry.Get("eth/db/chaindata/disk/write").(metrics.Meter).Count() frequency = float64(db.config.Refresh / time.Second) numCPU = float64(runtime.NumCPU()) @@ -300,8 +300,8 @@ func (db *Dashboard) collectData() { curNetworkEgress = metrics.DefaultRegistry.Get("p2p/OutboundTraffic").(metrics.Meter).Count() curProcessCPUTime = getProcessCPUTime() curSystemCPUUsage = systemCPUUsage - curDiskRead = metrics.DefaultRegistry.Get("eth/db/chaindata/compact/input").(metrics.Meter).Count() - curDiskWrite = metrics.DefaultRegistry.Get("eth/db/chaindata/compact/output").(metrics.Meter).Count() + curDiskRead = metrics.DefaultRegistry.Get("eth/db/chaindata/disk/read").(metrics.Meter).Count() + curDiskWrite = metrics.DefaultRegistry.Get("eth/db/chaindata/disk/write").(metrics.Meter).Count() deltaNetworkIngress = float64(curNetworkIngress - prevNetworkIngress) deltaNetworkEgress = float64(curNetworkEgress - prevNetworkEgress) diff --git a/ethdb/database.go b/ethdb/database.go index 59d74f0833..32f26258ad 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -30,6 +30,7 @@ import ( "github.com/syndtr/goleveldb/leveldb/filter" "github.com/syndtr/goleveldb/leveldb/iterator" "github.com/syndtr/goleveldb/leveldb/opt" + "regexp" ) var OpenFileLimit = 64 @@ -47,6 +48,8 @@ type LDBDatabase struct { compTimeMeter metrics.Meter // Meter for measuring the total time spent in database compaction compReadMeter metrics.Meter // Meter for measuring the data read during compaction compWriteMeter metrics.Meter // Meter for measuring the data written during compaction + diskReadMeter metrics.Meter // Meter for measuring the effective amount of data read + diskWriteMeter metrics.Meter // Meter for measuring the effective amount of data written quitLock sync.Mutex // Mutex protecting the quit channel access quitChan chan chan error // Quit channel to stop the metrics collection before closing the database @@ -188,16 +191,67 @@ func (db *LDBDatabase) Meter(prefix string) { db.compTimeMeter = metrics.NewRegisteredMeter(prefix+"compact/time", nil) db.compReadMeter = metrics.NewRegisteredMeter(prefix+"compact/input", nil) db.compWriteMeter = metrics.NewRegisteredMeter(prefix+"compact/output", nil) + db.diskReadMeter = metrics.NewRegisteredMeter(prefix+"disk/read", nil) + db.diskWriteMeter = metrics.NewRegisteredMeter(prefix+"disk/write", nil) // Create a quit channel for the periodic collector and run it db.quitLock.Lock() db.quitChan = make(chan chan error) db.quitLock.Unlock() - go db.meter(3 * time.Second) + go db.meterDiskIO(3 * time.Second) + go db.meterCompaction(3 * time.Second) } -// meter periodically retrieves internal leveldb counters and reports them to +// meterDiskIO periodically retrieves internal leveldb counters and reports them to +// the metrics subsystem. +// +// This is how the iostats look like (currently): +// Read(MB): 3895.04860 Write(MB): 3654.64712 +func (db *LDBDatabase) meterDiskIO(refresh time.Duration) { + var prev, curr [2]float64 + spaceTruncater := regexp.MustCompile(`[ ]{2,}`) + + for { + ioStats, err := db.db.GetProperty("leveldb.iostats") + if err != nil { + db.log.Error("Failed to read database iostats", "err", err) + return + } + + parts := strings.Split(spaceTruncater.ReplaceAllString(ioStats, " "), " ") + if curr[0], err = strconv.ParseFloat(parts[1], 64); err != nil { + db.log.Error("Read entry parsing failed", "err", err) + return + } + if curr[1], err = strconv.ParseFloat(parts[3], 64); err != nil { + db.log.Error("Write entry parsing failed", "err", err) + return + } + if db.diskReadMeter != nil { + db.diskReadMeter.Mark(int64((curr[0] - prev[0]) * 1024 * 1024)) + } + if db.diskWriteMeter != nil { + db.diskWriteMeter.Mark(int64((curr[1] - prev[1]) * 1024 * 1024)) + } + fmt.Printf("Read: %9.5fMB Write: %9.5fMB / %v\n", curr[0]-prev[0], curr[1]-prev[1], refresh) + prev[0] = curr[0] + prev[1] = curr[1] + + // Sleep a bit, then repeat the iostats collection + select { + case errc := <-db.quitChan: + // Quit requesting, stop hammering the database + errc <- nil + return + + case <-time.After(refresh): + // Timeout, gather a new set of iostats + } + } +} + +// meterCompaction periodically retrieves internal leveldb counters and reports them to // the metrics subsystem. // // This is how a stats table look like (currently): @@ -208,7 +262,7 @@ func (db *LDBDatabase) Meter(prefix string) { // 1 | 85 | 109.27913 | 28.09293 | 213.92493 | 214.26294 // 2 | 523 | 1000.37159 | 7.26059 | 66.86342 | 66.77884 // 3 | 570 | 1113.18458 | 0.00000 | 0.00000 | 0.00000 -func (db *LDBDatabase) meter(refresh time.Duration) { +func (db *LDBDatabase) meterCompaction(refresh time.Duration) { // Create the counters to store current and previous values counters := make([][]float64, 2) for i := 0; i < 2; i++ { @@ -222,13 +276,6 @@ func (db *LDBDatabase) meter(refresh time.Duration) { db.log.Error("Failed to read database stats", "err", err) return } - iostats, err := db.db.GetProperty("leveldb.iostats") - if err != nil { - db.log.Error("Failed to read database iostats", "err", err) - return - } - fmt.Println(iostats) - // Find the compaction table, skip the header lines := strings.Split(stats, "\n") for len(lines) > 0 && strings.TrimSpace(lines[0]) != "Compactions" {