Add adaptive concurrency parameter for queue-worker

chart/openfaas/README.md

@@ -581,6 +581,7 @@ See [values.yaml](./values.yaml) for detailed configuration.
 | `jetstreamQueueWorker.consumer.pullMaxMessages` | PullMaxMessages limits the number of messages to be buffered per consumer. Leave empty to use optimized default for the selected queue mode | `` |
 | `jetstreamQueueWorker.logs.debug` | Log debug messages | `false` |
 | `jetstreamQueueWorker.logs.format` | Set the log format, supports `console` or `json` | `console` |
+| `jetstreamQueueWorker.adaptiveConcurrency` | Enable adaptive concurrency limiting for functions. This setting only takes effect when `jetstreamQueueWorker.mode` is set to `function`. | `true` |
 | `nats.channel` | The name of the NATS Streaming channel or NATS JetStream stream to use for asynchronous function invocations | `faas-request` |
 | `nats.external.clusterName` | The name of the externally-managed NATS Streaming server | `""` |
 | `nats.external.enabled` | Whether to use an externally-managed NATS Streaming server | `false` |

chart/openfaas/templates/queueworker-dep.yaml

@@ -99,6 +99,8 @@ spec:
         - name: ack_wait
           value: "{{ .Values.queueWorker.ackWait }}"
 
+        - name: limiter_enabled
+          value: "{{ .Values.jetstreamQueueWorker.adaptiveConcurrency }}"
         - name: upstream_timeout
           value: "{{ .Values.gateway.upstreamTimeout }}"
         - name: "max_retry_attempts"

chart/openfaas/values.yaml

@@ -222,6 +222,9 @@ faasnetes:
 jetstreamQueueWorker:
   image: ghcr.io/openfaasltd/jetstream-queue-worker:0.4.12
   mode: "static"
+  # Enable adaptive concurrency limiting for functions.
+  # This setting only takes effect when jetstreamQueueWorker.mode is set to "function".
+  adaptiveConcurrency: true
   consumer:
     inactiveThreshold: "30s"
 

chart/queue-worker/README.md

@@ -38,7 +38,7 @@ nats:
     name: slow-fns
   consumer:
     durableName: slow-fns-workers
-upstreamTimeout: 15m  
+upstreamTimeout: 15m 
 ```
 
 Then pass `-f ./values-slow-fns.yaml` to the `helm upgrade --install` command instead of the `--set` flags.
@@ -101,6 +101,7 @@ helm upgrade --install \
 | `queueName` | Name of the queue if you want it to be different to the stream name | `""` - when empty, defaults to `nats.stream.name` |
 | `mode` | Queue operation mode: `static` (OpenFaaS Standard) or `function` (requires OpenFaaS for Enterprises) | `static` |
 | `maxInflight` | Control the concurrent invocations | `1` |
+| `adaptiveConcurrency` | Enable adaptive concurrency limiting for functions. This setting only takes effect when `mode` is set to `function`. | `true` |
 | `queuePartitions` | Number of queue partitions | `1` |
 | `partition` | Queue partition number this queue should subscribe to | `0` |
 | `consumer.inactiveThreshold` | If a function is inactive (has no invocations) for longer than this threshold its consumer will be removed to save resources | `30s` |

chart/queue-worker/templates/deployment.yaml

@@ -82,6 +82,8 @@ spec:
             - name: "ack_wait"
               value: "{{ .Values.nats.consumer.ackWait }}"
 
+            - name: "limiter_enabled"
+              value: "{{ .Values.adaptiveConcurrency }}"
             - name: "upstream_timeout"
               value: "{{ .Values.upstreamTimeout }}"
             - name: "max_retry_attempts"

chart/queue-worker/values.yaml

@@ -13,6 +13,9 @@ replicas: 1
 queueName: ""
 mode: static
 maxInflight: 1
+# Enable adaptive concurrency limiting for functions.
+# This setting only takes effect when mode is set to "function".
+adaptiveConcurrency: true
 
 queuePartitions: 1
 partition: 0