Add release flag to rayvens init (#65)

* Add release flag to rayvens init.

* Clean-up.

Author: doru1004

rayvens/api.py

@@ -322,6 +322,7 @@ def _eval(context, f, data):
 
 def init(mode=os.getenv('RAYVENS_MODE', 'auto'),
          transport=os.getenv('RAYVENS_TRANSPORT', 'auto'),
+         release=os.getenv('RAYVENS_RELEASE', False),
          check_port=True):
     modes = ['auto', 'local', 'mixed', 'operator']
     transports = ['auto', 'http', 'kafka', 'ray-serve']
@@ -337,19 +338,19 @@ def init(mode=os.getenv('RAYVENS_MODE', 'auto'),
 
     if mode in ['auto', 'local']:
         if transport in ['auto', 'http']:
-            _global_camel = start_http(mode, check_port)
+            _global_camel = start_http(mode, check_port, release)
         elif transport == 'kafka':
-            _global_camel = start_kafka(mode, check_port)
+            _global_camel = start_kafka(mode, check_port, release)
         else:
             raise RuntimeError(
                 f'{transport} transport unsupported for mode {mode}.')
     elif mode in ['mixed', 'operator']:
         if transport in ['auto', 'http']:
-            _global_camel = start_operator_http(mode, check_port)
+            _global_camel = start_operator_http(mode, check_port, release)
         elif transport == 'kafka':
-            _global_camel = start_operator_kafka(mode, check_port)
+            _global_camel = start_operator_kafka(mode, check_port, release)
         elif transport in ['ray-serve']:
-            _global_camel = start_operator_ray_serve(mode, check_port)
+            _global_camel = start_operator_ray_serve(mode, check_port, release)
         else:
             raise RuntimeError(
                 f'{transport} transport unsupported for mode {mode}.')

rayvens/core/common.py

@@ -37,7 +37,7 @@ def __init__(self, data, headers={}):
         self.headers = headers
 
 
-def get_run_mode(camel_mode, check_port):
+def get_run_mode(camel_mode, check_port, release):
     if camel_mode == 'auto' or camel_mode == 'local':
         mode.run_mode = RayvensMode.LOCAL
     elif camel_mode == 'mixed':
@@ -47,6 +47,7 @@ def get_run_mode(camel_mode, check_port):
     else:
         raise RuntimeError("Unsupported camel mode.")
     mode.check_port = check_port
+    mode.release = release
     return mode
 
 

rayvens/core/invocation.py

@@ -153,6 +153,12 @@ def _check_kamel_output(self,
                             end_condition,
                             with_output=False,
                             with_timeout=False):
+        # Decide whether to print logs or not. If this is release mode
+        # then no output will be printed:
+        with_output_final = with_output
+        if self.mode.release:
+            with_output_final = False
+
         # Implicit 5 minute timout in hundredths of a second:
         timout_duration = 5 * 60 * 100
         if with_timeout:
@@ -175,10 +181,11 @@ def _check_kamel_output(self,
             if with_timeout:
                 output = utils.print_log_from_queue(self.subprocess_name,
                                                     reading_thread.queue,
-                                                    with_output)
+                                                    with_output_final)
             else:
                 output = utils.print_log_from_subprocess(
-                    self.subprocess_name, self.process.stdout, with_output)
+                    self.subprocess_name, self.process.stdout,
+                    with_output_final)
 
             # Use the Kamel output to decide when Kamel instance is
             # ready to receive requests.

rayvens/core/kafka.py

@@ -20,8 +20,8 @@
 from rayvens.core.integration import Integration
 
 
-def start(camel_mode, check_port):
-    return Camel(get_run_mode(camel_mode, check_port))
+def start(camel_mode, check_port, release):
+    return Camel(get_run_mode(camel_mode, check_port, release))
 
 
 class Camel:

rayvens/core/local.py

@@ -19,8 +19,8 @@
 from rayvens.core.common import get_run_mode, send_to, recv_from, await_start
 
 
-def start(camel_mode, check_port):
-    return Camel(get_run_mode(camel_mode, check_port))
+def start(camel_mode, check_port, release):
+    return Camel(get_run_mode(camel_mode, check_port, release))
 
 
 class Camel:

rayvens/core/mode.py

@@ -53,6 +53,7 @@ def __init__(self, run_mode=RayvensMode.LOCAL):
         self.namespace = "ray"
         self.transport = None
         self.check_port = True
+        self.release = False
 
     def server_address(self, integration):
         return self._get_server_address(integration.integration_name,

rayvens/core/operator.py

@@ -20,8 +20,8 @@
 from rayvens.core.integration import Integration
 
 
-def start(camel_mode, check_port):
-    return Camel(get_run_mode(camel_mode, check_port))
+def start(camel_mode, check_port, release):
+    return Camel(get_run_mode(camel_mode, check_port, release))
 
 
 class Camel:

rayvens/core/operator_kafka.py

@@ -20,8 +20,8 @@
 from rayvens.core.integration import Integration
 
 
-def start(camel_mode, check_port):
-    return Camel(get_run_mode(camel_mode, check_port))
+def start(camel_mode, check_port, release):
+    return Camel(get_run_mode(camel_mode, check_port, release))
 
 
 class Camel:

rayvens/core/ray_serve.py

@@ -24,8 +24,8 @@
 from rayvens.core.integration import Integration
 
 
-def start(camel_mode, check_port):
-    return Camel(common.get_run_mode(camel_mode, check_port))
+def start(camel_mode, check_port, release):
+    return Camel(common.get_run_mode(camel_mode, check_port, release))
 
 
 class Camel: