fix: add timeout for database query in workflow estimation to prevent blocking

Add timeout control for GetWorkflowForEstimator database query to prevent
workflow execution from being blocked when database is slow or locked.

- Add default 5 second timeout for database queries
- Configurable via WORKFLOW_ESTIMATION_DB_QUERY_TIMEOUT environment variable
- Return default estimator on timeout/error to ensure workflow continues
- Add warning logs for timeout and error cases
- Add documentation for new environment variable

Signed-off-by: shuangkun <[email protected]>

Author: shuangkun

docs/environment-variables.md

@@ -58,6 +58,7 @@ This document outlines environment variables that can be used to customize behav
 | `SEMAPHORE_NOTIFY_DELAY`                 | `time.Duration`     | `1s`                                                                                        | Tuning Delay when notifying semaphore waiters about availability in the semaphore                                                                                                                                                                                        |
 | `WATCH_CONTROLLER_SEMAPHORE_CONFIGMAPS` | `bool` | `true` | Whether to watch the Controller's ConfigMap and semaphore ConfigMaps for run-time changes. When disabled, the Controller will only read these ConfigMaps once and will have to be manually restarted to pick up new changes. |
 | `SKIP_WORKFLOW_DURATION_ESTIMATION` | `bool` | `false` | Whether to lookup resource usage from prior workflows to estimate usage for new workflows. |
+| `WORKFLOW_ESTIMATION_DB_QUERY_TIMEOUT` | `time.Duration` | `5s` | Timeout for database queries when estimating workflow duration. Prevents workflow execution from being blocked when database is slow or locked. |
 
 CLI parameters of the Controller can be specified as environment variables with the `ARGO_` prefix.
 For example:

workflow/controller/estimation/estimator_factory.go

@@ -3,6 +3,7 @@ package estimation
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
@@ -11,6 +12,7 @@ import (
 	"github.com/argoproj/argo-workflows/v3/persist/sqldb"
 	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v3/util/env"
+	"github.com/argoproj/argo-workflows/v3/util/logging"
 	"github.com/argoproj/argo-workflows/v3/workflow/common"
 	"github.com/argoproj/argo-workflows/v3/workflow/controller/indexes"
 	"github.com/argoproj/argo-workflows/v3/workflow/hydrator"
@@ -32,6 +34,9 @@ var _ EstimatorFactory = &estimatorFactory{}
 
 var (
 	skipWorkflowDurationEstimation = env.LookupEnvStringOr("SKIP_WORKFLOW_DURATION_ESTIMATION", "false")
+	// Default timeout for database queries to prevent blocking workflow execution
+	// Can be overridden by WORKFLOW_ESTIMATION_DB_QUERY_TIMEOUT environment variable
+	defaultEstimationDBQueryTimeout = 5 * time.Second
 )
 
 func NewEstimatorFactory(ctx context.Context, wfInformer cache.SharedIndexInformer, hydrator hydrator.Interface, wfArchive sqldb.WorkflowArchive) EstimatorFactory {
@@ -84,9 +89,35 @@ func (f *estimatorFactory) NewEstimator(ctx context.Context, wf *wfv1.Workflow)
 			if err != nil {
 				return defaultEstimator, fmt.Errorf("failed to parse selector to requirements: %v", err)
 			}
-			baselineWF, err := f.wfArchive.GetWorkflowForEstimator(ctx, wf.Namespace, requirements)
+			// Add timeout to database query to prevent blocking workflow execution
+			// if database is slow or locked
+			queryTimeout := defaultEstimationDBQueryTimeout
+			if ctx != nil {
+				if timeout := env.LookupEnvDurationOr(ctx, "WORKFLOW_ESTIMATION_DB_QUERY_TIMEOUT", defaultEstimationDBQueryTimeout); timeout > 0 {
+					queryTimeout = timeout
+				}
+			}
+			var baseCtx context.Context
+			if ctx != nil {
+				baseCtx = ctx
+			} else {
+				baseCtx = context.Background()
+			}
+			queryCtx, cancel := context.WithTimeout(baseCtx, queryTimeout)
+			defer cancel()
+
+			baselineWF, err := f.wfArchive.GetWorkflowForEstimator(queryCtx, wf.Namespace, requirements)
 			if err != nil {
-				return defaultEstimator, fmt.Errorf("failed to get archived workflow for estimator: %v", err)
+				// Log the error but return default estimator to not block workflow execution
+				if ctx != nil {
+					logger := logging.RequireLoggerFromContext(ctx)
+					if queryCtx.Err() == context.DeadlineExceeded {
+						logger.WithError(err).WithField("timeout", queryTimeout).Warn(ctx, "database query for workflow estimation timed out, using default estimator")
+					} else {
+						logger.WithError(err).Warn(ctx, "failed to get archived workflow for estimator, using default estimator")
+					}
+				}
+				return defaultEstimator, nil
 			}
 			return &estimator{wf, baselineWF}, nil
 		}