fix: improve duration executor order and use timeout handler for checks (#479)

* fix: improve duration executor order and use timeout handler for checks

* fix: improve duration executor and add ci-load to config

Author: gacevicljubisa

cmd/beekeeper/cmd/check.go

@@ -29,71 +29,70 @@ func (c *command) initCheckCmd() error {
 		Short: "runs integration tests on a Bee cluster",
 		Long:  `runs integration tests on a Bee cluster.`,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			ctx, cancel := context.WithTimeout(cmd.Context(), c.globalConfig.GetDuration(optionNameTimeout))
-			defer cancel()
-
-			checks := c.globalConfig.GetStringSlice(optionNameChecks)
-			if len(checks) == 0 {
-				return fmt.Errorf("no checks provided")
-			}
-
-			clusterName := c.globalConfig.GetString(optionNameClusterName)
-			if clusterName == "" {
-				return errMissingClusterName
-			}
-
-			// set cluster config
-			cfgCluster, ok := c.config.Clusters[clusterName]
-			if !ok {
-				return fmt.Errorf("cluster %s not defined", clusterName)
-			}
-
-			cluster, err := c.setupCluster(ctx, clusterName, c.globalConfig.GetBool(optionNameCreateCluster))
-			if err != nil {
-				return fmt.Errorf("cluster setup: %w", err)
-			}
-
-			var (
-				metricsPusher  *push.Pusher
-				metricsEnabled = c.globalConfig.GetBool(optionNameMetricsEnabled)
-				cleanup        func()
-			)
-
-			if metricsEnabled {
-				metricsPusher, cleanup = newMetricsPusher(c.globalConfig.GetString(optionNameMetricsPusherAddress), cfgCluster.GetNamespace(), c.log)
-				// cleanup executes when the calling context terminates
-				defer cleanup()
-			}
-
-			// logger metrics
-			if l, ok := c.log.(metrics.Reporter); ok && metricsEnabled {
-				metrics.RegisterCollectors(metricsPusher, l.Report()...)
-			}
-
-			// tracing
-			tracingEndpoint := c.globalConfig.GetString(optionNameTracingEndpoint)
-			if c.globalConfig.IsSet(optionNameTracingHost) && c.globalConfig.IsSet(optionNameTracingPort) {
-				tracingEndpoint = strings.Join([]string{c.globalConfig.GetString(optionNameTracingHost), c.globalConfig.GetString(optionNameTracingPort)}, ":")
-			}
-			tracer, tracerCloser, err := tracing.NewTracer(&tracing.Options{
-				Enabled:     c.globalConfig.GetBool(optionNameTracingEnabled),
-				Endpoint:    tracingEndpoint,
-				ServiceName: c.globalConfig.GetString(optionNameTracingServiceName),
+			return c.withTimeoutHandler(cmd, func(ctx context.Context) error {
+				checks := c.globalConfig.GetStringSlice(optionNameChecks)
+				if len(checks) == 0 {
+					return fmt.Errorf("no checks provided")
+				}
+
+				clusterName := c.globalConfig.GetString(optionNameClusterName)
+				if clusterName == "" {
+					return errMissingClusterName
+				}
+
+				// set cluster config
+				cfgCluster, ok := c.config.Clusters[clusterName]
+				if !ok {
+					return fmt.Errorf("cluster %s not defined", clusterName)
+				}
+
+				cluster, err := c.setupCluster(ctx, clusterName, c.globalConfig.GetBool(optionNameCreateCluster))
+				if err != nil {
+					return fmt.Errorf("cluster setup: %w", err)
+				}
+
+				var (
+					metricsPusher  *push.Pusher
+					metricsEnabled = c.globalConfig.GetBool(optionNameMetricsEnabled)
+					cleanup        func()
+				)
+
+				if metricsEnabled {
+					metricsPusher, cleanup = newMetricsPusher(c.globalConfig.GetString(optionNameMetricsPusherAddress), cfgCluster.GetNamespace(), c.log)
+					// cleanup executes when the calling context terminates
+					defer cleanup()
+				}
+
+				// logger metrics
+				if l, ok := c.log.(metrics.Reporter); ok && metricsEnabled {
+					metrics.RegisterCollectors(metricsPusher, l.Report()...)
+				}
+
+				// tracing
+				tracingEndpoint := c.globalConfig.GetString(optionNameTracingEndpoint)
+				if c.globalConfig.IsSet(optionNameTracingHost) && c.globalConfig.IsSet(optionNameTracingPort) {
+					tracingEndpoint = strings.Join([]string{c.globalConfig.GetString(optionNameTracingHost), c.globalConfig.GetString(optionNameTracingPort)}, ":")
+				}
+				tracer, tracerCloser, err := tracing.NewTracer(&tracing.Options{
+					Enabled:     c.globalConfig.GetBool(optionNameTracingEnabled),
+					Endpoint:    tracingEndpoint,
+					ServiceName: c.globalConfig.GetString(optionNameTracingServiceName),
+				})
+				if err != nil {
+					return fmt.Errorf("tracer: %w", err)
+				}
+				defer tracerCloser.Close()
+
+				// set global config
+				checkGlobalConfig := config.CheckGlobalConfig{
+					Seed:    c.globalConfig.GetInt64(optionNameSeed),
+					GethURL: c.globalConfig.GetString(optionNameGethURL),
+				}
+
+				checkRunner := check.NewCheckRunner(checkGlobalConfig, c.config.Checks, cluster, metricsPusher, tracer, c.log)
+
+				return checkRunner.Run(ctx, checks)
 			})
-			if err != nil {
-				return fmt.Errorf("tracer: %w", err)
-			}
-			defer tracerCloser.Close()
-
-			// set global config
-			checkGlobalConfig := config.CheckGlobalConfig{
-				Seed:    c.globalConfig.GetInt64(optionNameSeed),
-				GethURL: c.globalConfig.GetString(optionNameGethURL),
-			}
-
-			checkRunner := check.NewCheckRunner(checkGlobalConfig, c.config.Checks, cluster, metricsPusher, tracer, c.log)
-
-			return checkRunner.Run(ctx, checks)
 		},
 		PreRunE: c.preRunE,
 	}

config/local.yaml

@@ -293,6 +293,23 @@ checks:
       duration: 10m
     timeout: 11m
     type: smoke
+  ci-load:
+    options:
+      content-size: 50000
+      postage-ttl: 24h
+      postage-depth: 21
+      postage-label: test-label
+      duration: 10m
+      uploader-count: 2
+      downloader-count: 0
+      max-committed-depth: 2
+      committed-depth-check-wait: 5m
+      upload-groups:
+        - bee
+      download-groups:
+        - light
+    timeout: 30m
+    type: load
   ci-soc:
     options:
       postage-ttl: 24h

pkg/scheduler/duration.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/ethersphere/beekeeper/pkg/logging"
@@ -23,23 +24,29 @@ func NewDurationExecutor(duration time.Duration, log logging.Logger) *DurationEx
 
 // Run executes the given task and waits for the specified duration before stopping.
 func (de *DurationExecutor) Run(ctx context.Context, task func(ctx context.Context) error) error {
+	if task == nil {
+		return errors.New("task cannot be nil")
+	}
+
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	doneCh := make(chan error, 1)
-	defer close(doneCh)
 
 	go func() {
-		doneCh <- task(ctx)
+		select {
+		case <-ctx.Done():
+		case doneCh <- task(ctx):
+		}
 	}()
 
 	select {
-	case err := <-doneCh:
-		return err
+	case <-ctx.Done():
+		return ctx.Err()
 	case <-time.After(de.duration):
 		de.log.Infof("Duration of %s expired, stopping executor", de.duration)
 		return nil
-	case <-ctx.Done():
-		return ctx.Err()
+	case err := <-doneCh:
+		return err
 	}
 }