Add readiness probe to await start (#16)

* Attempt to add rediness probe.

* Fix probe.

* Factor out readiness check.

Author: doru1004

examples/slack/slack_operator_mode.py

@@ -55,8 +55,6 @@
 # Wait for sink to reach running state.
 ray.get(sink)
 
-time.sleep(15)
-
 # Sends message to all sinks attached to this stream.
 stream << f'Sending message to Slack sink in run mode {run_mode}.'
 

rayvens/core/common.py

@@ -19,6 +19,7 @@
 import time
 import threading
 import os
+import json
 from pathlib import Path
 from confluent_kafka import Consumer, Producer
 from rayvens.core import kamel
@@ -37,21 +38,51 @@ def get_run_mode(camel_mode):
     return mode
 
 
+def _wait_for_ready_integration(mode, integration):
+    server_address = mode.server_address(integration)
+    health_check_address = f"{server_address}/q/health"
+    while True:
+        response = requests.get(health_check_address, timeout=(5, None))
+        json_response = json.loads(response.content)
+        all_routes_are_up = True
+        for check in json_response['checks']:
+            if check['name'] == 'camel-readiness-checks' and check[
+                    'status'] == 'UP':
+                data = check['data']
+                if data['context'] == 'UP':
+                    # Ensure all routes are up.
+                    route_index = 1
+                    route = f'route:route{route_index}'
+                    while route in data:
+                        if data[route] != 'UP':
+                            all_routes_are_up = False
+                            break
+                        route_index += 1
+                        route = f'route:route{route_index}'
+        if all_routes_are_up:
+            break
+        time.sleep(1)
+
+
 # Wait for an integration to reach its running state and not only that but
 # also be in a state where it can immediately execute incoming requests.
-def await_start(mode, integration_name):
+def await_start(mode, integration):
     # Only needed when operator is used.
     if mode.is_local():
         return True
 
     # Check logs of the integration to make sure it was installed properly.
-    invocation = kamel.log(mode, integration_name, "Installed features:")
+    invocation = kamel.log(mode, integration.integration_name,
+                           "Installed features:")
     integration_is_running = invocation is not None
     if integration_is_running:
-        print(f'Integration {integration_name} is running.')
+        print(f'Integration {integration.integration_name} is running.')
     else:
         print('Integration did not start correctly.')
 
+    # Perform health check and wait for integration to be ready.
+    _wait_for_ready_integration(mode, integration)
+
     return integration_is_running
 
 

rayvens/core/kamel.py

@@ -128,6 +128,10 @@ def run(integration_content,
     command.append("-n")
     command.append(mode.namespace)
 
+    # Include health check:
+    command.append("-d")
+    command.append("camel:camel-quarkus-microprofile-health")
+
     for envVar in envVars:
         if envVar not in os.environ:
             raise RuntimeError("Variable %s not set in current environment" %
@@ -193,6 +197,10 @@ def local_run(integration_content,
         command.append('--property')
         command.append(f'quarkus.http.port={port}')
 
+    # Include health check:
+    command.append("-d")
+    command.append("camel:camel-quarkus-microprofile-health")
+
     return kamel_utils.invoke_kamel_command(
         command,
         mode,

rayvens/core/operator.py

@@ -49,7 +49,7 @@ def add_source(self, stream, source, source_name):
         # Set up source for the HTTP connector case.
         send_to(stream.actor, self.mode.server_address(integration), route)
 
-        if not await_start(self.mode, integration.integration_name):
+        if not await_start(self.mode, integration):
             raise RuntimeError('Could not start source')
         return integration
 
@@ -71,7 +71,7 @@ def add_sink(self, stream, sink, sink_name):
         stream.actor.send_to.remote(helper, sink_name)
 
         # Wait for integration to finish.
-        if not await_start(self.mode, integration.integration_name):
+        if not await_start(self.mode, integration):
             raise RuntimeError('Could not start sink')
 
         return integration

rayvens/core/ray_serve.py

@@ -74,7 +74,7 @@ def add_source(self, stream, source, source_name):
         # Start running the source integration.
         integration.invoke_run(self.mode, integration_content)
 
-        if not await_start(self.mode, integration.integration_name):
+        if not await_start(self.mode, integration):
             raise RuntimeError('Could not start source')
         return integration
 
@@ -111,7 +111,7 @@ def add_sink(self, stream, sink, sink_name):
         stream.actor.send_to.remote(helper, sink_name)
 
         # Wait for integration to finish.
-        if not await_start(self.mode, integration_name):
+        if not await_start(self.mode, integration):
             raise RuntimeError('Could not start sink')
 
         return integration