Add adaptive concurrency parameter for queue-worker

Allow users to disable the new adaptive concurrency feature for the
queue-worker.

Signed-off-by: Han Verstraete (OpenFaaS Ltd) <[email protected]>

Author: welteki

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 based on 429 response. 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: adaptive_concurrency
+          value: "{{ .Values.jetstreamQueueWorker.adaptiveConcurrency }}"
         - name: upstream_timeout
           value: "{{ .Values.gateway.upstreamTimeout }}"
         - name: "max_retry_attempts"

chart/openfaas/values.yaml

@@ -1,37 +1,37 @@
 ## For OpenFaaS Standard and OpenFaaS for Enterprises users, see:
 # https://github.com/openfaas/faas-netes/blob/master/chart/openfaas/values-pro.yaml
 
-functionNamespace: openfaas-fn  # Default namespace for functions
+functionNamespace: openfaas-fn # Default namespace for functions
 
 # Contact us via https://www.openfaas.com/support to purchase a license
 openfaasPro: false
 
 # Contact us via https://www.openfaas.com/support to purchase a license
 oem: false
 
-httpProbe: true               # Setting to true will use HTTP for readiness and liveness probe on the OpenFaaS core components
+httpProbe: true # Setting to true will use HTTP for readiness and liveness probe on the OpenFaaS core components
 
 # set clusterRole: true for:
 ## CPU/RAM metrics in OpenFaaS API
 ## Multiple-namespace support
 clusterRole: false
 
-createCRDs: true              # Creates the Function/Profile CRDs, set to false if you are managing CRDs in another way
+createCRDs: true # Creates the Function/Profile CRDs, set to false if you are managing CRDs in another way
 
 # basic_auth must never be disabled, and is required for all OpenFaaS components.
 # There is no good reason to disable this, and it causes a severe security risk.
 # The configuration option remains for backwards-compatibility.
-basic_auth: true              # Authentication for core components, always set to true
+basic_auth: true # Authentication for core components, always set to true
 
-rbac: true                    # Kubernetes RBAC, no good reason to disable this
-generateBasicAuth: true       # Set to false if applying credentials separately from the chart, otherwise set to true
+rbac: true # Kubernetes RBAC, no good reason to disable this
+generateBasicAuth: true # Set to false if applying credentials separately from the chart, otherwise set to true
 
 # Define a securityContext for the containers deployed by this helm chart.
 securityContext: {}
 
 exposeServices: true
-serviceType: NodePort        # serviceType for OpenFaaS gateway
-async: true                  # No known reason to disable this, kept for legacy reasons
+serviceType: NodePort # serviceType for OpenFaaS gateway
+async: true # No known reason to disable this, kept for legacy reasons
 
 # create pod security policies for OpenFaaS control plane
 # https://kubernetes.io/docs/concepts/policy/pod-security-policy/
@@ -45,22 +45,22 @@ openfaasImagePullPolicy: "IfNotPresent"
 #
 # kubectl create secret generic -n openfaas ca-bundle --from-file=ca.crt=./ca.crt
 #
-caBundleSecretName: ""       # Name of the secret containing the CA bundle for the OpenFaaS gateway
+caBundleSecretName: "" # Name of the secret containing the CA bundle for the OpenFaaS gateway
 
 functions:
-  imagePullPolicy: "Always"    # Image pull policy for deployed functions, for OpenFaaS Pro you can also set: IfNotPresent and Never.
-  httpProbe: true              # Setting to true will use HTTP for readiness and liveness probe on function pods
-  setNonRootUser: false        # It's recommended to set this to "true", but test your images before committing to it
+  imagePullPolicy: "Always" # Image pull policy for deployed functions, for OpenFaaS Pro you can also set: IfNotPresent and Never.
+  httpProbe: true # Setting to true will use HTTP for readiness and liveness probe on function pods
+  setNonRootUser: false # It's recommended to set this to "true", but test your images before committing to it
   readinessProbe:
     initialDelaySeconds: 0
-    timeoutSeconds: 1           # Tuned-in to run checks early and quickly to support fast cold-start from zero replicas
-    periodSeconds: 2            # Reduce to 1 for a faster cold-start, increase higher for lower-CPU usage
+    timeoutSeconds: 1 # Tuned-in to run checks early and quickly to support fast cold-start from zero replicas
+    periodSeconds: 2 # Reduce to 1 for a faster cold-start, increase higher for lower-CPU usage
     successThreshold: 1
     failureThreshold: 3
   livenessProbe:
     initialDelaySeconds: 0
     timeoutSeconds: 1
-    periodSeconds: 2           # Reduce to 1 for a faster cold-start, increase higher for lower-CPU usage
+    periodSeconds: 2 # Reduce to 1 for a faster cold-start, increase higher for lower-CPU usage
     failureThreshold: 3
 
 gatewayPro:
@@ -71,7 +71,7 @@ gateway:
   image: ghcr.io/openfaas/gateway:0.27.13
   readTimeout: "1m05s"
   writeTimeout: "1m05s"
-  upstreamTimeout: "1m"  # Must be smaller than read/write_timeout
+  upstreamTimeout: "1m" # Must be smaller than read/write_timeout
   replicas: 1
   scaleFromZero: true
   # change the port when creating multiple releases in the same baremetal cluster
@@ -222,6 +222,9 @@ faasnetes:
 jetstreamQueueWorker:
   image: ghcr.io/openfaasltd/jetstream-queue-worker:0.4.7
   mode: "static"
+  # Enable adaptive concurrency limiting for functions based on 429 response.
+  # This setting only takes effect when jetstreamQueueWorker.mode is set to "function".
+  adaptiveConcurrency: true
   consumer:
     inactiveThreshold: "30s"
 
@@ -474,20 +477,20 @@ ingress:
 
   # Used to create Ingress record (should be used with exposeServices: false).
   hosts:
-  - host: gateway.openfaas.local  # Replace with gateway.example.com if public-facing
-    http:
-      paths:
-      - path: /
-        pathType: Prefix
-        backend:
-          service:
-            name: gateway
-            port:
-              number: 8080
+    - host: gateway.openfaas.local # Replace with gateway.example.com if public-facing
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: gateway
+                port:
+                  number: 8080
 
   # annotations:
-    # This annotation is deprecated, use IngressClass in the spec instead
-    # kubernetes.io/ingress.class: nginx
+  # This annotation is deprecated, use IngressClass in the spec instead
+  # kubernetes.io/ingress.class: nginx
   tls:
   # Secrets must be manually created in the namespace.
 

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 based on 429 response. 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: "adaptive_concurrency"
+              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 based on 429 response.
+# This setting only takes effect when mode is set to "function".
+adaptiveConcurrency: true
 
 queuePartitions: 1
 partition: 0