Use readiness probe in local mode. (#50)

Author: doru1004

rayvens/core/common.py

@@ -83,31 +83,30 @@ def _wait_for_ready_integration(mode, integration):
 # 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):
-    # 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.integration_name,
-                           "Installed features:")
-    integration_is_running = invocation is not None
-    if integration_is_running:
-        print(f'Integration {integration.integration_name} is running.')
-    else:
-        print('Integration did not start correctly.')
+    if not mode.is_local():
+        invocation = kamel.log(mode, integration.integration_name,
+                               "Installed features:")
+        integration_is_running = invocation is not None
+        if integration_is_running:
+            print(f'Integration {integration.integration_name} is running.')
+        else:
+            print('Integration did not start correctly.')
 
     # For kafka transport the health check cannot be performed.
     if mode.transport == 'kafka':
         return True
 
-    # Perform health check and wait for integration to be ready.
+    # Perform readiness check and wait for integration to be ready.
     healthy_integration = _wait_for_ready_integration(mode, integration)
 
     if healthy_integration:
-        print(f'Integration {integration.integration_name} is healthy.')
-        return integration_is_running
+        print(f'Integration {integration.integration_name} is ready.')
+        if not mode.is_local():
+            return integration_is_running
+        return True
 
-    print(f'Integration {integration.integration_name} is not healthy.')
+    print(f'Integration {integration.integration_name} cannot be ready.')
     return False
 
 

rayvens/core/local.py

@@ -16,7 +16,7 @@
 
 import rayvens.core.catalog as catalog
 from rayvens.core.integration import Integration
-from rayvens.core.common import get_run_mode, send_to, recv_from
+from rayvens.core.common import get_run_mode, send_to, recv_from, await_start
 
 
 def start(camel_mode, check_port):
@@ -37,6 +37,8 @@ def add_source(self, stream, config, source_name):
         integration.prepare_environment(self.mode)
         integration.invoke_local_run(self.mode, spec)
         send_to(stream.actor, self.mode.server_address(integration), route)
+        if not await_start(self.mode, integration):
+            raise RuntimeError('Could not start source')
         return integration
 
     def add_sink(self, stream, config, sink_name):
@@ -47,6 +49,8 @@ def add_sink(self, stream, config, sink_name):
         integration.invoke_local_run(self.mode, spec)
         recv_from(stream.actor, sink_name,
                   self.mode.server_address(integration), route)
+        if not await_start(self.mode, integration):
+            raise RuntimeError('Could not start source')
         return integration
 
     def disconnect(self, integration):