feat: kurtosis run --non-blocking-tasks (#2153)

## Description:
Adds a flag `--non-blocking-tasks`. If set, task services from `run_sh`
and `run_python` are not removed before continuing with execution. The
effects is small over Docker, but this significantly reduces run times
over k8s because of the overhead of removing a pod (up to 30s).

For the `ethereum-package` over k8s with images cached, running with the
flag goes from 4min to 1min. However, task services will have to be
removed manually after execution or via a `kurtosis clean`.

## Is this change user facing?
YES

## References:
#2140

api/golang/core/lib/binding_constructors/binding_constructors.go

@@ -42,6 +42,7 @@ func NewRunStarlarkScriptArgs(
 	cloudInstanceId string,
 	cloudUserId string,
 	imageDownloadMode kurtosis_core_rpc_api_bindings.ImageDownloadMode,
+	nonBlockingMode bool,
 ) *kurtosis_core_rpc_api_bindings.RunStarlarkScriptArgs {
 	cloudInstanceIdCopy := new(string)
 	*cloudInstanceIdCopy = cloudInstanceId
@@ -61,6 +62,7 @@ func NewRunStarlarkScriptArgs(
 		CloudInstanceId:      cloudInstanceIdCopy,
 		CloudUserId:          cloudUserIdCopy,
 		ImageDownloadMode:    imageDownloadModeCopy,
+		NonBlockingMode:      &nonBlockingMode,
 	}
 }
 
@@ -75,6 +77,7 @@ func NewRunStarlarkPackageArgs(
 	cloudInstanceId string,
 	cloudUserId string,
 	imageDownloadMode kurtosis_core_rpc_api_bindings.ImageDownloadMode,
+	nonBlockingMode bool,
 ) *kurtosis_core_rpc_api_bindings.RunStarlarkPackageArgs {
 	parallelismCopy := new(int32)
 	*parallelismCopy = parallelism
@@ -98,6 +101,7 @@ func NewRunStarlarkPackageArgs(
 		CloudInstanceId:        cloudInstanceIdCopy,
 		CloudUserId:            cloudUserIdCopy,
 		ImageDownloadMode:      imageDownloadModeCopy,
+		NonBlockingMode:        &nonBlockingMode,
 	}
 }
 
@@ -112,6 +116,7 @@ func NewRunStarlarkRemotePackageArgs(
 	cloudInstanceId string,
 	cloudUserId string,
 	imageDownloadMode kurtosis_core_rpc_api_bindings.ImageDownloadMode,
+	nonBlockingMode bool,
 ) *kurtosis_core_rpc_api_bindings.RunStarlarkPackageArgs {
 	parallelismCopy := new(int32)
 	*parallelismCopy = parallelism
@@ -135,6 +140,7 @@ func NewRunStarlarkRemotePackageArgs(
 		CloudInstanceId:        cloudInstanceIdCopy,
 		CloudUserId:            cloudUserIdCopy,
 		ImageDownloadMode:      imageDownloadModeCopy,
+		NonBlockingMode:        &nonBlockingMode,
 	}
 }
 

api/golang/core/lib/enclaves/enclave_context.go

@@ -110,7 +110,17 @@ func (enclaveCtx *EnclaveContext) RunStarlarkScript(
 		return nil, nil, stacktrace.Propagate(err, "An error occurred when parsing YAML args for script '%v'", oldSerializedParams)
 	}
 	ctxWithCancel, cancelCtxFunc := context.WithCancel(ctx)
-	executeStartosisScriptArgs := binding_constructors.NewRunStarlarkScriptArgs(runConfig.MainFunctionName, serializedScript, serializedParams, runConfig.DryRun, runConfig.Parallelism, runConfig.ExperimentalFeatureFlags, runConfig.CloudInstanceId, runConfig.CloudUserId, runConfig.ImageDownload)
+	executeStartosisScriptArgs := binding_constructors.NewRunStarlarkScriptArgs(
+		runConfig.MainFunctionName,
+		serializedScript,
+		serializedParams,
+		runConfig.DryRun,
+		runConfig.Parallelism,
+		runConfig.ExperimentalFeatureFlags,
+		runConfig.CloudInstanceId,
+		runConfig.CloudUserId,
+		runConfig.ImageDownload,
+		runConfig.NonBlockingMode)
 	starlarkResponseLineChan := make(chan *kurtosis_core_rpc_api_bindings.StarlarkRunResponseLine)
 
 	stream, err := enclaveCtx.client.RunStarlarkScript(ctxWithCancel, executeStartosisScriptArgs)
@@ -172,7 +182,8 @@ func (enclaveCtx *EnclaveContext) RunStarlarkPackage(
 		runConfig.ExperimentalFeatureFlags,
 		runConfig.CloudInstanceId,
 		runConfig.CloudUserId,
-		runConfig.ImageDownload)
+		runConfig.ImageDownload,
+		runConfig.NonBlockingMode)
 	if err != nil {
 		return nil, nil, stacktrace.Propagate(err, "Error preparing package '%s' for execution", packageRootPath)
 	}
@@ -297,7 +308,7 @@ func (enclaveCtx *EnclaveContext) RunStarlarkRemotePackage(
 	}()
 
 	starlarkResponseLineChan := make(chan *kurtosis_core_rpc_api_bindings.StarlarkRunResponseLine)
-	executeStartosisScriptArgs := binding_constructors.NewRunStarlarkRemotePackageArgs(packageId, runConfig.RelativePathToMainFile, runConfig.MainFunctionName, serializedParams, runConfig.DryRun, runConfig.Parallelism, runConfig.ExperimentalFeatureFlags, runConfig.CloudInstanceId, runConfig.CloudUserId, runConfig.ImageDownload)
+	executeStartosisScriptArgs := binding_constructors.NewRunStarlarkRemotePackageArgs(packageId, runConfig.RelativePathToMainFile, runConfig.MainFunctionName, serializedParams, runConfig.DryRun, runConfig.Parallelism, runConfig.ExperimentalFeatureFlags, runConfig.CloudInstanceId, runConfig.CloudUserId, runConfig.ImageDownload, runConfig.NonBlockingMode)
 
 	stream, err := enclaveCtx.client.RunStarlarkPackage(ctxWithCancel, executeStartosisScriptArgs)
 	if err != nil {
@@ -605,9 +616,21 @@ func (enclaveCtx *EnclaveContext) assembleRunStartosisPackageArg(
 	cloudInstanceId string,
 	cloudUserId string,
 	imageDownloadMode kurtosis_core_rpc_api_bindings.ImageDownloadMode,
+	nonBlockingMode bool,
 ) (*kurtosis_core_rpc_api_bindings.RunStarlarkPackageArgs, error) {
 
-	return binding_constructors.NewRunStarlarkPackageArgs(packageName, relativePathToMainFile, mainFunctionName, serializedParams, dryRun, parallelism, experimentalFeatures, cloudInstanceId, cloudUserId, imageDownloadMode), nil
+	return binding_constructors.NewRunStarlarkPackageArgs(
+		packageName,
+		relativePathToMainFile,
+		mainFunctionName,
+		serializedParams,
+		dryRun,
+		parallelism,
+		experimentalFeatures,
+		cloudInstanceId,
+		cloudUserId,
+		imageDownloadMode,
+		nonBlockingMode), nil
 }
 
 func (enclaveCtx *EnclaveContext) uploadStarlarkPackage(packageId string, packageRootPath string) error {

api/golang/core/lib/starlark_run_config/starlark_run_config.go

@@ -11,6 +11,7 @@ const (
 	defaultCloudInstanceId        = ""
 	defaultCloudUserId            = ""
 	defaultImageDownload          = kurtosis_core_rpc_api_bindings.ImageDownloadMode_missing
+	defaultNonBlockingMode        = false
 )
 
 var defaultExperimentalFeatureFlags = []kurtosis_core_rpc_api_bindings.KurtosisFeatureFlag(nil)
@@ -25,6 +26,7 @@ type StarlarkRunConfig struct {
 	CloudInstanceId          string
 	CloudUserId              string
 	ImageDownload            kurtosis_core_rpc_api_bindings.ImageDownloadMode
+	NonBlockingMode          bool
 }
 
 type starlarkRunConfigOption func(*StarlarkRunConfig)
@@ -40,6 +42,7 @@ func NewRunStarlarkConfig(opts ...starlarkRunConfigOption) *StarlarkRunConfig {
 		CloudInstanceId:          defaultCloudInstanceId,
 		CloudUserId:              defaultCloudUserId,
 		ImageDownload:            defaultImageDownload,
+		NonBlockingMode:          defaultNonBlockingMode,
 	}
 
 	for _, opt := range opts {
@@ -104,3 +107,9 @@ func WithImageDownloadMode(imageDownloadMode kurtosis_core_rpc_api_bindings.Imag
 		config.ImageDownload = imageDownloadMode
 	}
 }
+
+func WithNonBlockingMode(nonBlockingMode bool) starlarkRunConfigOption {
+	return func(config *StarlarkRunConfig) {
+		config.NonBlockingMode = nonBlockingMode
+	}
+}