Keda servicebus：由 Keda 扩展的作业中的某些 Pod 处理的所有消息。

huangapple
go评论108阅读模式
英文:

Keda servicebus: All the messages getting processed by some pods of keda scaledJob

问题

I am using Azure Kubernetes Service (AKS) platform and used KEDA "ScaledJob" for long-running jobs. In this, Azure Service Bus Queue trigger has been used to auto-trigger Jobs. Now while I am adding messages in Azure Service Bus, KEDA will auto-trigger the job and create a node/pod as per configuration. But in this case, all the messages are getting picked up and processed by some pods. It is expected that each scaled-up pod processes a single message and terminates.

Following is my YAML file:

  1. apiVersion: keda.sh/v1alpha1
  2. kind: ScaledJob
  3. metadata:
  4.   name: {{ .Chart.Name }}
  5. spec:
  6.   jobTargetRef:
  7.     backoffLimit: 4
  8.     parallelism: 1
  9.     completions: 1
  10.     activeDeadlineSeconds: 300
  11.     template:
  12.       spec:
  13.         imagePullSecrets:
  14.           - name: {{ .Values.image.imagePullSecrets }}
  15.         terminationGracePeriodSeconds: 30
  16.         dnsPolicy: ClusterFirst
  17.         volumes:
  18.           - name: azure
  19.             azureFile:
  20.               shareName: sharenameone
  21.               secretName: secret-sharenameone
  22.               readOnly: true
  23.           - name: one-storage
  24.             emptyDir: {}
  25.           - name: "logging-volume-file"
  26.             persistentVolumeClaim:
  27.               claimName: "azure-file-logging"
  28.         initContainers:
  29.           - name: test-java-init
  30.             image: {{ .Values.global.imageRegistryURI }}/{{ .Values.image.javaInitImage.name}}:{{ .Values.image.javaInitImage.tag }}
  31.             imagePullPolicy: {{ .Values.image.pullPolicy }}
  32.             securityContext:
  33.               readOnlyRootFilesystem: true
  34.             resources:
  35.               requests:
  36.                 cpu: 100m
  37.                 memory: 300Mi
  38.               limits:
  39.                 cpu: 200m
  40.                 memory: 400Mi
  41.             volumeMounts:
  42.               - name: azure
  43.                 mountPath: /mnt/azure
  44.               - name: one-storage
  45.                 mountPath: /certs
  46.         containers:
  47.           - name: {{ .Chart.Name }}
  48.             image: {{ .Values.global.imageRegistryURI }}/tests/{{ .Chart.Name }}:{{ .Values.version }}
  49.             imagePullPolicy: {{ .Values.image.pullPolicy }}
  50.             env:
  51.               {{- include "chart.envVars" . | nindent 14 }}
  52.               - name: JAVA_OPTS
  53.                 value: >-
  54.                                         {{ .Values.application.javaOpts }}
  55.               - name: application_name
  56.                 value: "test_application"
  57.               - name: queueName
  58.                 value: "test-queue-name"
  59.               - name: servicebusconnstrenv
  60.                 valueFrom:
  61.                   secretKeyRef:
  62.                     name: secrets-service-bus
  63.                     key: service_bus_conn_str
  64.             volumeMounts:
  65.               - name: cert-storage
  66.                 mountPath: /certs
  67.               - name: "logging-volume-azure-file"
  68.                 mountPath: "/mnt/logging"
  69.             resources:
  70.               {{- toYaml .Values.resources | nindent 14 }}
  71.   pollingInterval: 30
  72.   maxReplicaCount: 5
  73.   successfulJobsHistoryLimit: 5
  74.   failedJobsHistoryLimit: 20
  75.   triggers:
  76.   - type: azure-servicebus
  77.     metadata:
  78.       queueName: "test-queue-name"
  79.       connectionFromEnv: servicebusconnstrenv
  80.       messageCount: "1"

And this is my Azure Function listener:

  1. @FunctionName("TestServiceBusTrigger")
  2. public void TestServiceBusTriggerHandler(
  3.     @ServiceBusQueueTrigger(
  4.             name = "msg",
  5.             queueName = "%TEST_QUEUE_NAME%",
  6.             connection = "ServiceBusConnectionString")
  7.         final String inputMessage,
  8.     final ExecutionContext context) {
  10.   final java.util.logging.Logger contextLogger = context.getLogger();
  11.   System.setProperty("javax.net.ssl.trustStore", "/certs/cacerts");
  13.   try {
  14.     // all the processing goes here 
  15.   } catch (Exception e) {
  16.    //Exception handling
  17.   }
  18. }

What configurations need to be added so that each scaled-up pod processes a single message and terminates?

英文:

I am using Azure Kubernetes Service(AKS) platform and used KEDA "ScaledJob" for long running job. In this, Azure Service Bus Queue trigger has been used to auto trigger Jobs. Now while I am adding messages in Azure Service Bus, KEDA will auto trigger job and create node/pod as per configuration. but in this case, all the messages are getting picked up and processed by some pods. it is expected that each scaled up pod processes single message and terminates.

following is my yml file

  1. apiVersion: keda.sh/v1alpha1
  2. kind: ScaledJob
  3. metadata:
  4. name: {{ .Chart.Name }}
  5. spec:
  6. jobTargetRef:
  7. backoffLimit: 4
  8. parallelism: 1
  9. completions: 1
  10. activeDeadlineSeconds: 300
  11. template:
  12. spec:
  13. imagePullSecrets:
  14. - name: {{ .Values.image.imagePullSecrets }}
  15. terminationGracePeriodSeconds: 30
  16. dnsPolicy: ClusterFirst
  17. volumes:
  18. - name: azure
  19. azureFile:
  20. shareName: sharenameone
  21. secretName: secret-sharenameone
  22. readOnly: true
  23. - name: one-storage
  24. emptyDir: {}
  25. - name: "logging-volume-file"
  26. persistentVolumeClaim:
  27. claimName: "azure-file-logging"        
  28. initContainers:
  29. - name: test-java-init
  30. image: {{ .Values.global.imageRegistryURI }}/{{ .Values.image.javaInitImage.name}}:{{ .Values.image.javaInitImage.tag }}
  31. imagePullPolicy: {{ .Values.image.pullPolicy }}
  32. securityContext:
  33. readOnlyRootFilesystem: true
  34. resources:
  35. requests:
  36. cpu: 100m
  37. memory: 300Mi
  38. limits:
  39. cpu: 200m
  40. memory: 400Mi
  41. volumeMounts:
  42. - name: azure
  43. mountPath: /mnt/azure
  44. - name: one-storage
  45. mountPath: /certs
  46. containers:
  47. - name: {{ .Chart.Name }}
  48. image: {{ .Values.global.imageRegistryURI }}/tests/{{ .Chart.Name }}:{{ .Values.version }}
  49. imagePullPolicy: {{ .Values.image.pullPolicy }}
  50. env:
  51. {{- include "chart.envVars" . | nindent 14 }}
  52. - name: JAVA_OPTS
  53. value: >-
  54. {{ .Values.application.javaOpts }}          
  55. - name: application_name
  56. value: "test_application"
  57. - name: queueName
  58. value: "test-queue-name"
  59. - name: servicebusconnstrenv
  60. valueFrom: 
  61. secretKeyRef:
  62. name: secrets-service-bus
  63. key: service_bus_conn_str
  64. volumeMounts:
  65. - name: cert-storage
  66. mountPath: /certs
  67. - name: "logging-volume-azure-file"
  68. mountPath: "/mnt/logging"
  69. resources:
  70. {{- toYaml .Values.resources | nindent 14 }}                   
  71. pollingInterval: 30
  72. maxReplicaCount: 5
  73. successfulJobsHistoryLimit: 5
  74. failedJobsHistoryLimit: 20
  75. triggers:
  76. - type: azure-servicebus
  77. metadata:
  78. queueName: "test-queue-name"
  79. connectionFromEnv: servicebusconnstrenv
  80. messageCount: "1"

and this is my azure function listener

  1.  @FunctionName("TestServiceBusTrigger")
  2. public void TestServiceBusTriggerHandler(
  3. @ServiceBusQueueTrigger(
  4. name = "msg",
  5. queueName = "%TEST_QUEUE_NAME%",
  6. connection = "ServiceBusConnectionString")
  7. final String inputMessage,
  8. final ExecutionContext context) {
  9. final java.util.logging.Logger contextLogger = context.getLogger();
  10. System.setProperty("javax.net.ssl.trustStore", "/certs/cacerts");
  11. try {
  12. // all the processing goes here 
  13. } catch (Exception e) {
  14. //Exception handling
  15. }
  16. }

what configurations needs to be added, so that each scaled up pod processes single message and terminates?

答案1

得分: 1

这不是 Azure Functions 的设计方式，或者更准确地说，甚至不是一般情况下如何使用 KEDA 的方式。如果已经在运行的容器可以在分配后处理尽可能多的消息，将会更理想。

话虽如此，如果您的情景仍然需要这样做，您可以编写一个简单的脚本，直接使用 Azure Service Bus SDK 来获取一个消息，处理它，然后终止。

英文:

This is not how Azure Functions were designed, or rather even how KEDA should be used in general. It would be more ideal if the already running container processes as many messages as it can once provisioned.

That being said, if your scenario still requires this, you could write a simple script that directly uses the Azure Service Bus SDK to fetch just one message, process it, and terminate.

huangapple
  • 本文由 发表于 2023年4月10日 18:38:00
  • 转载请务必保留本文链接：https://go.coder-hub.com/75976349.html
  • azure
  • azureservicebus
  • keda
  • keda-scaledjob
  • kubernetes
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定