Add readiness probe to k8s job in KubernetesV2 backend

e2e/test/cases/kafka/kafka-spec.ts

@@ -51,7 +51,7 @@ describe('kafka', () => {
             terasliceHarness.waitForExStatus(sender, 'completed')
         ]);
 
-        await terasliceHarness.waitForIndexCount(specIndex, total);
+        await terasliceHarness.waitForIndexCount(specIndex, total, 45 * 1000);
         await reader.stop();
 
         await terasliceHarness.waitForExStatus(reader, 'stopped');

packages/teraslice-messaging/package.json

@@ -1,7 +1,7 @@
 {
     "name": "@terascope/teraslice-messaging",
     "displayName": "Teraslice Messaging",
-    "version": "1.2.0",
+    "version": "1.3.0",
     "description": "An internal teraslice messaging library using socket.io",
     "homepage": "https://github.com/terascope/teraslice/tree/master/packages/teraslice-messaging#readme",
     "bugs": {

packages/teraslice-messaging/src/execution-controller/interfaces.ts

@@ -1,4 +1,5 @@
 import { Logger } from '@terascope/utils';
+import http from 'node:http';
 
 export interface ClientOptions {
     executionControllerUrl: string;
@@ -16,9 +17,13 @@ export interface ServerOptions {
     workerDisconnectTimeout: number;
     networkLatencyBuffer?: number;
     actionTimeout: number;
+    requestListener?: RequestListener;
     logger?: Logger;
 }
 
+export interface RequestListener {
+    (request: http.IncomingMessage, response: http.ServerResponse): void;
+}
 export interface Worker {
     workerId: string;
 }

packages/teraslice-messaging/src/execution-controller/server.ts

@@ -8,10 +8,12 @@ const { Available, Unavailable } = core.ClientState;
 export class Server extends core.Server {
     private _activeWorkers: i.ActiveWorkers;
     queue: Queue<EnqueuedWorker>;
+    executionReady: boolean;
 
     constructor(opts: i.ServerOptions) {
         const {
-            port, actionTimeout, networkLatencyBuffer, workerDisconnectTimeout, logger
+            port, actionTimeout, networkLatencyBuffer,
+            workerDisconnectTimeout, logger, requestListener
         } = opts;
 
         if (!isNumber(workerDisconnectTimeout)) {
@@ -21,14 +23,16 @@ export class Server extends core.Server {
         super({
             port,
             actionTimeout,
+            requestListener,
             networkLatencyBuffer,
             clientDisconnectTimeout: workerDisconnectTimeout,
             serverName: 'ExecutionController',
-            logger,
+            logger
         });
 
         this.queue = new Queue();
         this._activeWorkers = {};
+        this.executionReady = false;
     }
 
     async start(): Promise<void> {

packages/teraslice/package.json

@@ -40,7 +40,7 @@
     "dependencies": {
         "@terascope/elasticsearch-api": "^4.0.0",
         "@terascope/job-components": "^1.2.0",
-        "@terascope/teraslice-messaging": "^1.2.0",
+        "@terascope/teraslice-messaging": "^1.3.0",
         "@terascope/types": "^1.0.0",
         "@terascope/utils": "^1.0.0",
         "async-mutex": "^0.5.0",

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetes/index.ts

@@ -132,6 +132,7 @@ export class KubernetesClusterBackend {
 
         const pod = await this.k8s.waitForSelectedPod(
             `${controllerLabel}=${controllerUid}`,
+            'pod-status',
             undefined,
             this.context.sysconfig.teraslice.slicer_timeout
         );
@@ -162,6 +163,14 @@ export class KubernetesClusterBackend {
         // @ts-expect-error
         execution.k8sUid = jobs.items[0].metadata.uid;
 
+        /// Wait for ex readiness probe to return 'Ready'
+        await this.k8s.waitForSelectedPod(
+            selector,
+            'readiness-probe',
+            undefined,
+            this.context.sysconfig.teraslice.slicer_timeout
+        );
+
         const kr = new K8sResource(
             'deployments',
             'worker',

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetes/jobs/execution_controller.hbs

@@ -30,6 +30,17 @@
                     "image": "{{dockerImage}}",
                     "name": "{{name}}",
                     "ports": [{ "containerPort": 45680 }],
+                    "readinessProbe": {
+                        "httpGet": {
+                            "path": "/health",
+                            "port": 45680
+                        },
+                        "initialDelaySeconds": 5,
+                        "periodSeconds": 10,
+                        "timeoutSeconds": 5,
+                        "successThreshold": 1,
+                        "failureThreshold": 3
+                    },
                     "volumeMounts": [{
                         "mountPath": "/app/config",
                         "name": "config"

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetes/k8s.ts

@@ -87,14 +87,15 @@ export class K8s {
      * NOTE: If your selector will return multiple pods, this method probably
      * won't work for you.
      * @param {String} selector kubernetes selector, like 'controller-uid=XXX'
+     * @param {String} statusType type of status to check, either pod status or readiness probe
      * @param {String} ns       namespace to search, this will override the default
      * @param {Number} timeout  time, in ms, to wait for pod to start
      * @return {Object}         pod
      *
      * TODO: Should this use the cluster state that gets polled periodically,
      * rather than making it's own k8s API calls
      */
-    async waitForSelectedPod(selector: string, ns?: string, timeout = 10000) {
+    async waitForSelectedPod(selector: string, statusType: string, ns?: string, timeout = 10000) {
         const namespace = ns || this.defaultNamespace;
         let now = Date.now();
         const end = now + timeout;
@@ -112,7 +113,20 @@ export class K8s {
             }
 
             if (typeof pod !== 'undefined' && pod) {
-                if (get(pod, 'status.phase') === 'Running') return pod;
+                if (statusType === 'readiness-probe') {
+                    if (pod.status?.conditions) {
+                        for (const condition of pod.status.conditions) {
+                            if (
+                                condition.type === 'ContainersReady'
+                                && condition.status === 'True'
+                            ) {
+                                return pod;
+                            }
+                        }
+                    }
+                } else if (statusType === 'pod-status') {
+                    if (get(pod, 'status.phase') === 'Running') return pod;
+                }
             }
             if (now > end) throw new Error(`Timeout waiting for pod matching: ${selector}`);
             this.logger.debug(`waiting for pod matching: ${selector}`);

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetesV2/index.ts

@@ -150,10 +150,9 @@ export class KubernetesClusterBackendV2 {
 
         const controllerUid = jobResult.spec?.selector?.matchLabels?.[controllerLabel];
 
-        // Right now this is waiting for the selected pod to come up in a "running"
-        // state. It may be better to check for a readiness probe instead
         const pod = await this.k8s.waitForSelectedPod(
             `${controllerLabel}=${controllerUid}`,
+            'pod-status',
             undefined,
             this.context.sysconfig.teraslice.slicer_timeout
         );
@@ -191,6 +190,14 @@ export class KubernetesClusterBackendV2 {
         // @ts-expect-error
         execution.k8sUid = jobs.items[0].metadata.uid;
 
+        /// Wait for ex readiness probe to return 'Ready'
+        await this.k8s.waitForSelectedPod(
+            selector,
+            'readiness-probe',
+            undefined,
+            this.context.sysconfig.teraslice.slicer_timeout
+        );
+
         const kr = new K8sResource(
             'deployments',
             'worker',

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetesV2/jobs/execution_controller.hbs

@@ -30,6 +30,17 @@
                     "image": "{{dockerImage}}",
                     "name": "{{name}}",
                     "ports": [{ "containerPort": 45680 }],
+                    "readinessProbe": {
+                        "httpGet": {
+                            "path": "/health",
+                            "port": 45680
+                        },
+                        "initialDelaySeconds": 5,
+                        "periodSeconds": 10,
+                        "timeoutSeconds": 5,
+                        "successThreshold": 1,
+                        "failureThreshold": 3
+                    },
                     "volumeMounts": [{
                         "mountPath": "/app/config",
                         "name": "config"

packages/teraslice/src/lib/cluster/services/cluster/backends/kubernetesV2/k8s.ts

@@ -87,7 +87,7 @@ export class K8s {
      * TODO: Should this use the cluster state that gets polled periodically,
      * rather than making it's own k8s API calls
      */
-    async waitForSelectedPod(selector: string, ns?: string, timeout = 10000) {
+    async waitForSelectedPod(selector: string, statusType: string, ns?: string, timeout = 10000) {
         const namespace = ns || this.defaultNamespace;
         let now = Date.now();
         const end = now + timeout;
@@ -104,7 +104,20 @@ export class K8s {
             }
 
             if (pod) {
-                if (get(pod, 'status.phase') === 'Running') return pod;
+                if (statusType === 'readiness-probe') {
+                    if (pod.status?.conditions) {
+                        for (const condition of pod.status.conditions) {
+                            if (
+                                condition.type === 'ContainersReady'
+                                && condition.status === 'True'
+                            ) {
+                                return pod;
+                            }
+                        }
+                    }
+                } else if (statusType === 'pod-status') {
+                    if (get(pod, 'status.phase') === 'Running') return pod;
+                }
             }
             if (now > end) throw new Error(`Timeout waiting for pod matching: ${selector}`);
             this.logger.debug(`waiting for pod matching: ${selector}`);

packages/teraslice/src/lib/workers/execution-controller/index.ts

@@ -1,5 +1,6 @@
 import ms from 'ms';
 import _ from 'lodash';
+import http from 'node:http';
 import {
     formatURL, ExecutionController as ExController, ClusterMaster
 } from '@terascope/teraslice-messaging';
@@ -79,10 +80,10 @@ export class ExecutionController {
         const workerDisconnectTimeout = get(config, 'worker_disconnect_timeout');
         const nodeDisconnectTimeout = get(config, 'node_disconnect_timeout');
         const shutdownTimeout = get(config, 'shutdown_timeout');
-
         this.server = new ExController.Server({
             port: slicerPort,
             networkLatencyBuffer,
+            requestListener: this.requestListener.bind(this),
             actionTimeout,
             workerDisconnectTimeout,
             logger
@@ -311,6 +312,8 @@ export class ExecutionController {
         this.logger.info(`execution: ${this.exId} initialized execution_controller`);
 
         this.isInitialized = true;
+        /// This will change the  '/ready' endpoint to Ready
+        this.server.executionReady = true;
     }
 
     async run() {
@@ -1176,4 +1179,19 @@ export class ExecutionController {
             );
         }
     }
+
+    requestListener(req: http.IncomingMessage, res: http.ServerResponse) {
+        if (req.url === '/health') {
+            if (this.server.executionReady) {
+                res.writeHead(200);
+                res.end('Ready');
+            } else {
+                res.writeHead(503);
+                res.end('Service Unavailable');
+            }
+        } else {
+            res.writeHead(501);
+            res.end('Not Implemented');
+        }
+    }
 }