Merge pull request #111 from ethpandaops/feat/refactor-disk-usage

feat(disk): Refactor disk usage calc

example_config.yaml

@@ -13,5 +13,6 @@ execution:
     - "txpool"
 diskUsage:
   enabled: false
+  interval: 60m
   directories:
   - /data/ethereum

pkg/exporter/config.go

@@ -1,5 +1,11 @@
 package exporter
 
+import (
+	"time"
+
+	"github.com/ethpandaops/beacon/pkg/human"
+)
+
 // Config holds the configuration for the ethereum sync status tool.
 type Config struct {
 	// Execution is the execution node to use.
@@ -29,8 +35,9 @@ type ExecutionNode struct {
 
 // DiskUsage configures the exporter to expose disk usage stats for these directories.
 type DiskUsage struct {
-	Enabled     bool     `yaml:"enabled"`
-	Directories []string `yaml:"directories"`
+	Enabled     bool           `yaml:"enabled"`
+	Directories []string       `yaml:"directories"`
+	Interval    human.Duration `yaml:"interval"`
 }
 
 // PairConfig holds the config for a Pair of Execution and Consensus Clients
@@ -55,6 +62,9 @@ func DefaultConfig() *Config {
 		DiskUsage: DiskUsage{
 			Enabled:     false,
 			Directories: []string{},
+			Interval: human.Duration{
+				Duration: 60 * time.Minute,
+			},
 		},
 		Pair: PairConfig{
 			Enabled: true,

pkg/exporter/disk/disk.go

@@ -3,6 +3,7 @@ package disk
 import (
 	"context"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -20,24 +21,33 @@ type diskUsage struct {
 	log         logrus.FieldLogger
 	metrics     Metrics
 	directories []string
+	interval    time.Duration
 }
 
 // NewUsage returns a new DiskUsage instance.
-func NewUsage(ctx context.Context, log logrus.FieldLogger, namespace string, directories []string) (UsageMetrics, error) {
+func NewUsage(ctx context.Context, log logrus.FieldLogger, namespace string, directories []string, interval time.Duration) (UsageMetrics, error) {
 	return &diskUsage{
 		log:         log,
 		metrics:     NewMetrics(log, namespace),
 		directories: directories,
+		interval:    interval,
 	}, nil
 }
 
 func (d *diskUsage) StartAsync(ctx context.Context) {
+	d.log.WithField("directories", d.directories).Info("Starting disk usage metrics...")
+
+	_, err := d.GetUsage(ctx, d.directories)
+	if err != nil {
+		d.log.WithError(err).Error("Failed to get disk usage")
+	}
+
 	go func() {
 		for {
 			select {
 			case <-ctx.Done():
 				return
-			case <-time.After(time.Second * 60):
+			case <-time.After(d.interval):
 				if _, err := d.GetUsage(ctx, d.directories); err != nil {
 					d.log.WithError(err).Error("Failed to get disk usage")
 				}
@@ -51,15 +61,17 @@ func (d *diskUsage) GetUsage(ctx context.Context, directories []string) ([]Usage
 	var diskUsed []Usage
 
 	for _, directory := range directories {
-		info, err := os.Lstat(directory)
+		_, err := os.Lstat(directory)
 		if err != nil {
 			d.log.WithField("directory", directory).Warn("Directory does not exist")
+
 			continue
 		}
 
-		used, err := getDiskUsed(directory, info)
+		used, err := getDiskUsed(directory)
 		if err != nil {
 			d.log.WithField("directory", directory).WithError(err).Error("Failed to get usage")
+
 			continue
 		}
 
@@ -76,33 +88,19 @@ func (d *diskUsage) GetUsage(ctx context.Context, directories []string) ([]Usage
 	return diskUsed, nil
 }
 
-func getDiskUsed(currentPath string, info os.FileInfo) (int64, error) {
-	size := info.Size()
-	if !info.IsDir() {
-		return size, nil
-	}
-
-	dir, err := os.Open(currentPath)
-
-	if err != nil {
-		return size, err
-	}
-	defer dir.Close()
-
-	files, err := dir.Readdir(-1)
-	if err != nil {
-		return size, err
-	}
-
-	for _, file := range files {
-		if file.Name() == "." || file.Name() == ".." {
-			continue
+func getDiskUsed(path string) (int64, error) {
+	var size int64
+	err := filepath.Walk(path, func(_ string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
 		}
 
-		s, _ := getDiskUsed(currentPath+"/"+file.Name(), file)
+		if !info.IsDir() {
+			size += info.Size()
+		}
 
-		size += s
-	}
+		return err
+	})
 
-	return size, nil
+	return size, err
 }

pkg/exporter/exporter.go

@@ -53,7 +53,14 @@ func (e *exporter) Init(ctx context.Context) error {
 	if e.config.Execution.Enabled {
 		e.log.WithField("modules", strings.Join(e.config.Execution.Modules, ", ")).Info("Initializing execution...")
 
-		executionNode, err := execution.NewExecutionNode(ctx, e.log.WithField("exporter", "execution"), fmt.Sprintf("%s_exe", e.namespace), e.config.Execution.Name, e.config.Execution.URL, e.config.Execution.Modules)
+		executionNode, err := execution.NewExecutionNode(
+			ctx,
+			e.log.WithField("exporter", "execution"),
+			fmt.Sprintf("%s_exe", e.namespace),
+			e.config.Execution.Name,
+			e.config.Execution.URL,
+			e.config.Execution.Modules,
+		)
 		if err != nil {
 			return err
 		}
@@ -68,7 +75,18 @@ func (e *exporter) Init(ctx context.Context) error {
 	if e.config.DiskUsage.Enabled {
 		e.log.Info("Initializing disk usage...")
 
-		diskUsage, err := disk.NewUsage(ctx, e.log.WithField("exporter", "disk"), fmt.Sprintf("%s_disk", e.namespace), e.config.DiskUsage.Directories)
+		interval := e.config.DiskUsage.Interval.Duration
+		if interval == 0 {
+			interval = 60 * time.Minute
+		}
+
+		diskUsage, err := disk.NewUsage(
+			ctx,
+			e.log.WithField("exporter", "disk"),
+			fmt.Sprintf("%s_disk", e.namespace),
+			e.config.DiskUsage.Directories,
+			interval,
+		)
 		if err != nil {
 			return err
 		}
@@ -130,7 +148,7 @@ func (e *exporter) Serve(ctx context.Context, port int) error {
 	return nil
 }
 
-func (e *exporter) bootstrapConsensusClients(ctx context.Context) error {
+func (e *exporter) bootstrapConsensusClients(_ context.Context) error {
 	opts := *beacon.DefaultOptions().
 		EnableDefaultBeaconSubscription().
 		EnablePrometheusMetrics()