CORE-17476 Modifying helm charts to create ClusterIP services for workers, passing to FlowWorker

applications/workers/release/flow-worker/src/main/kotlin/net/corda/applications/workers/flow/FlowWorker.kt

@@ -24,6 +24,7 @@ import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
 import org.slf4j.LoggerFactory
 import picocli.CommandLine.Mixin
+import picocli.CommandLine.Option
 
 /** The worker for handling flows. */
 @Suppress("Unused", "LongParameterList")
@@ -92,4 +93,7 @@ class FlowWorker @Activate constructor(
 private class FlowWorkerParams {
     @Mixin
     var defaultParams = DefaultWorkerParams()
-}
+
+    @Option(names = ["--endpoint"], description = ["Internal REST endpoints for Corda workers"], required = true)
+    val workerEndpoints: Map<String, String> = emptyMap()
+}

charts/corda-lib/templates/_helpers.tpl

@@ -614,3 +614,28 @@ data:
   {{ $keySecretKey }}: {{ $keySecretValue }}
   {{ $caSecretKey }}: {{ $caSecretValue }}
 {{- end }}
+
+{{/*
+The port which should be used to connect to Corda worker instances
+*/}}
+{{- define "corda.workerServicePort" -}}
+7000
+{{- end -}}
+
+{{/*
+Cluster IP service name
+*/}}
+{{- define "corda.workerInternalServiceName" -}}
+{{- printf "%s-internal-service" . -}}
+{{- end -}}
+
+{{/*
+Get the endpoint argument for a given worker
+*/}}
+{{- define "corda.getWorkerEndpoint" }}
+{{- $context := .context }}
+{{- $worker := .worker }}
+{{- $workerName := printf "%s-%s-worker" (include "corda.fullname" $context) (include "corda.workerTypeKebabCase" $worker) }}
+{{- $workerServiceName := include "corda.workerInternalServiceName" $workerName }}
+{{- printf "worker.http.endpoint.%s=%s:%s" $worker $workerServiceName (include "corda.workerServicePort" $) }}
+{{- end }}

charts/corda-lib/templates/_worker.tpl

@@ -77,6 +77,19 @@ spec:
     targetPort: http
 {{- end }}
 ---
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "corda.workerInternalServiceName" $workerName }}
+spec:
+  type: ClusterIP
+  selector:
+    app: {{ $workerName }}
+  ports:
+      - protocol: TCP
+        port: {{ include "corda.workerServicePort" . }}
+        targetPort: "monitor"
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -318,6 +331,12 @@ spec:
           {{- if $.Values.tracing.samplesPerSecond }}
           - "--trace-samples-per-second={{ $.Values.tracing.samplesPerSecond }}"
           {{- end }}
+          {{- if $optionalArgs.servicesAccessed }}
+          {{- range $worker := $optionalArgs.servicesAccessed }}
+          {{- $endpoint := include "corda.getWorkerEndpoint" (dict "context" $ "worker" $worker) }}
+          - --endpoint={{ $endpoint }}
+          {{- end }}
+          {{- end }}
           {{- range $i, $arg := $optionalArgs.additionalWorkerArgs }}
           - {{ $arg | quote }}
           {{- end }}
@@ -513,4 +532,4 @@ Worker affinity
 {{- $_ := set $affinity.podAntiAffinity "preferredDuringSchedulingIgnoredDuringExecution" ( append $affinity.podAntiAffinity.preferredDuringSchedulingIgnoredDuringExecution ( fromYaml ( include "corda.defaultAffinity" ( list ( add ( len $affinity.podAntiAffinity.preferredDuringSchedulingIgnoredDuringExecution ) 1 ) $worker ) ) ) ) }}
 affinity:
 {{- toYaml $affinity | nindent 2 }}
-{{- end }}
+{{- end }}

charts/corda/templates/workers.yaml

@@ -5,7 +5,7 @@
   ( dict "clusterDbAccess" true )
 ) }}
 {{- include "corda.worker" ( list $ .Values.workers.flow "flow"
-  ( dict "stateManagerDbAccess" true )
+  ( dict "stateManagerDbAccess" true "servicesAccessed" (list "crypto" "verification" "uniqueness" "persistence") )
 ) }}
 {{- include "corda.worker" ( list $ .Values.workers.flowMapper "flowMapper" ) }}
 {{- include "corda.worker" ( list $ .Values.workers.verification "verification" ) }}