[Processor] Add enrichment and validation for async processing #3532
Conversation
| for _, supportedKind := range triggerKindsSupportAsync { | ||
| if triggerKind == supportedKind { | ||
| return true | ||
| } | ||
| } |
There was a problem hiding this comment.
Why not:
return strings.Contains(triggerKind, triggerKindsSupportAsync)
There was a problem hiding this comment.
trigger is always the same (we don't have any versions there), so it should be fine.
pkg/platform/abstract/platform.go
Outdated
| } | ||
| if !functionconfig.TriggerKindSupportsAsync(triggerInstance.Kind) { |
There was a problem hiding this comment.
| } | |
| if !functionconfig.TriggerKindSupportsAsync(triggerInstance.Kind) { | |
| } | |
| // Async validations | |
| if !functionconfig.TriggerKindSupportsAsync(triggerInstance.Kind) { |
pkg/platform/abstract/platform.go
Outdated
| } | ||
| if !functionconfig.TriggerKindSupportsAsync(triggerInstance.Kind) { | ||
| return nuclio.NewErrBadRequest(fmt.Sprintf( | ||
| "Async processing mode is not supported for %s trigger kind", |
There was a problem hiding this comment.
Better put variables at the end so it's easier for searching later
| "Async processing mode is not supported for %s trigger kind", | |
| "Async processing mode is not supported for trigger kind - %s", |
pkg/platform/abstract/platform.go
Outdated
|
|
||
| if !functionconfig.RuntimeSupportsAsync(functionConfig.Spec.Runtime) { | ||
| return nuclio.NewErrBadRequest(fmt.Sprintf( | ||
| "Async processing mode is not supported for %s runtime", |
There was a problem hiding this comment.
| "Async processing mode is not supported for %s runtime", | |
| "Async processing mode is not supported for runtime - %s", |
pkg/platform/abstract/platform.go
Outdated
| if triggerInstance.AsyncConfig != nil { | ||
| return nuclio.NewErrBadRequest("AsyncConfig should be empty when working in `sync` trigger mode") | ||
| } |
There was a problem hiding this comment.
What do you think about warning and ignoring?
There was a problem hiding this comment.
@TomerShor both options are fine. The problem here is that user may think that only AsyncConfig must be configured, when there is also a mode to configure - that's something I am worried about here
There was a problem hiding this comment.
It made me think weather we really need a "Mode" property, or just async.enabled...
There was a problem hiding this comment.
@TomerShor not sure about enabled, disabled. As it would mean that it's something not default (and we want to have it as default in the future probably). However on the other hand, it would be more aligned with Batching (which will never be smth we enable by default in the contrast).
And from BE side it's easier to work with mode than with enable/disable
pkg/platform/abstract/platform.go
Outdated
| @@ -1783,17 +1821,17 @@ func (ap *Platform) enrichTriggers(ctx context.Context, functionConfig *function | |||
| return errors.Wrap(err, "Failed to enrich batch params") | |||
| } | |||
|
|
|||
| if err := ap.enrichProcessingMode(ctx, functionConfig); err != nil { | |||
There was a problem hiding this comment.
Both this function and enrichBatchParams iterate over all triggers, AND we iterate over them again 2 rows below.
Instead - Iterate once, and enrich batch and processing params, as well as name etc.
| type ConnectionCreationMode string | ||
|
|
||
| const ( | ||
| ConnectionCreationModeStatic ConnectionCreationMode = "static" | ||
| ConnectionCreationModeDynamic ConnectionCreationMode = "dynamic" |
There was a problem hiding this comment.
Is this a prep for lazy connection creation?
pkg/platform/abstract/platform.go
Outdated
| "trigger", triggerName, | ||
| "maxConnectionsNumber", functionconfig.DefaultMaxConnectionsNumber, | ||
| ) | ||
| triggerInstance.AsyncConfig.MaxConnectionsNumber = functionconfig.DefaultMaxConnectionsNumber |
There was a problem hiding this comment.
Is there a limit on this number? maybe we should set something
There was a problem hiding this comment.
@TomerShor the limit is defined on the python side already, we can't really manage it on client side. Python side won't open more connections than file descriptors available. So, we should fail from wrapper side in this case
43c4e52 to
9d4715e
Compare
…multisockets # Conflicts: # pkg/processor/runtime/python/test/python_test.go
pkg/functionconfig/types.go
Outdated
| @@ -95,7 +95,8 @@ type Trigger struct { | |||
| // General attributes | |||
| Attributes map[string]interface{} `json:"attributes,omitempty"` | |||
|
|
|||
| Mode TriggerWorkMode `json:"mode,omitempty"` | |||
| Mode TriggerWorkMode `json:"mode,omitempty"` | |||
| AsyncConfig *AsyncConfig `json:"asyncConfig,omitempty"` | |||
There was a problem hiding this comment.
This aligns better with the config's naming conventions (e.g. build and not buildConfig)
| AsyncConfig *AsyncConfig `json:"asyncConfig,omitempty"` | |
| AsyncConfig *AsyncConfig `json:"async,omitempty"` |
Adds enrichment and validation for async processing