Merge pull request #229 from jina-ai/fix-flow-cli-226

fix(cli): add cli to load directly from yaml

docs/chapters/envs.rst

@@ -81,6 +81,17 @@ Here is the list of environment variables that ``jina`` respects during runtime.
 
     :default: ``INFO``
 
+.. confval:: JINA_LOG_SSE
+
+    Turn on the server side event logging for all pea, pods, executors running in the context.
+
+    :default: unset
+
+.. confval:: JINA_LOG_PROFILING
+
+    Turn on the server side event logging for profiling
+
+    :default: unset
 
 .. confval:: JINA_LOG_LONG
 

jina/executors/__init__.py

@@ -305,9 +305,8 @@ def save(self, filename: str = None) -> bool:
         :return: successfully persisted or not
         """
         if not self.is_updated:
-            self.logger.info('no update since %s, will not save. '
-                             'If you really want to save it, call "touch()" before "save()" to force saving'
-                             % self._last_snapshot_ts)
+            self.logger.info(f'no update since {self._last_snapshot_ts:%Y-%m-%d %H:%M:%S%z}, will not save. '
+                             'If you really want to save it, call "touch()" before "save()" to force saving')
             return False
 
         self.is_updated = False
@@ -376,10 +375,9 @@ def load_config(cls: Type[AnyExecutor], filename: Union[str, TextIO], separated_
                     mod = tmp['metas']['py_modules']
 
                     if isinstance(mod, str):
-                        if not os.path.isabs(mod):
-                            mod = os.path.join(os.path.dirname(filename), mod)
-                        PathImporter.add_modules(mod)
-                    elif isinstance(mod, list):
+                        mod = [mod]
+
+                    if isinstance(mod, list):
                         mod = [m if os.path.isabs(m) else os.path.join(os.path.dirname(filename), m) for m in mod]
                         PathImporter.add_modules(*mod)
                     else:

jina/flow/__init__.py

@@ -81,23 +81,28 @@ def __init__(self, args: 'argparse.Namespace' = None, **kwargs):
         as the head and tail routers are removed.
         """
         self.logger = get_logger(self.__class__.__name__)
+        self._pod_nodes = OrderedDict()  # type: Dict[str, 'FlowPod']
+        self._build_level = FlowBuildLevel.EMPTY
+        self._pod_name_counter = 0
+        self._last_changed_pod = []
+
+        self.update_args(args, **kwargs)
+
+        if not self.args.no_gateway:
+            self._add_gateway()
+
+    def update_args(self, args, **kwargs):
         from ..main.parser import set_flow_parser
         _flow_parser = set_flow_parser()
         if args is None:
             from ..helper import get_parsed_args
             _, args, _ = get_parsed_args(kwargs, _flow_parser)
 
         self.args = args
-        if self.args.logserver and 'log_sse' not in kwargs:
+        if kwargs and self.args.logserver and 'log_sse' not in kwargs:
             kwargs['log_sse'] = True
         self._common_kwargs = kwargs
-        self._pod_nodes = OrderedDict()  # type: Dict[str, 'FlowPod']
-        self._build_level = FlowBuildLevel.EMPTY
-        self._pod_name_counter = 0
-        self._last_changed_pod = []
-        if not self.args.no_gateway:
-            self._add_gateway()
-        self._kwargs = get_non_defaults_args(args, _flow_parser)
+        self._kwargs = get_non_defaults_args(args, _flow_parser)  #: for yaml dump
 
     @classmethod
     def to_yaml(cls, representer, data):

jina/helper.py

@@ -429,6 +429,10 @@ def get_parsed_args(kwargs, parser):
     return args, p_args, unknown_args
 
 
-def get_non_defaults_args(args, parser) -> Dict:
-    _defaults = parser.parse_args([])
-    return {k: v for k, v in vars(args).items() if getattr(_defaults, k) != v}
+def get_non_defaults_args(args, parser, taboo=(None,)) -> Dict:
+    non_defaults = {}
+    _defaults = vars(parser.parse_args([]))
+    for k, v in vars(args).items():
+        if k in _defaults and k not in taboo and _defaults[k] != v:
+            non_defaults[k] = v
+    return non_defaults

jina/logging/base.py

@@ -193,7 +193,7 @@ def get_logger(context: str, context_len: int = 15,
         h.setFormatter(ColorFormatter(fmt_str))
         logger.addHandler(h)
 
-    if log_profile:
+    if ('JINA_LOG_PROFILING' in os.environ) or log_profile:
         h = logging.FileHandler('jina-profile-%s.json' % __uptime__, delay=True)
         h.setLevel(verbose_level.value)
         h.setFormatter(ProfileFormatter(timed_fmt_str))
@@ -204,7 +204,7 @@ def get_logger(context: str, context_len: int = 15,
         h.setFormatter(JsonFormatter(timed_fmt_str))
         logger.addHandler(h)
 
-    if log_sse:
+    if ('JINA_LOG_SSE' in os.environ) or log_sse:
         h = QueueHandler(__sse_queue__)
         h.setLevel(verbose_level.value)
         h.setFormatter(JsonFormatter(timed_fmt_str))

jina/main/api.py

@@ -46,5 +46,18 @@ def client(args):
 
 
 def flow(args):
-    from ..flow.cli import FlowCLI
-    FlowCLI(args)
+    """Start a Flow from a YAML file"""
+    from ..flow import Flow
+    if args.yaml_path:
+        from threading import Event
+        f = Flow.load_config(args.yaml_path)
+        f.update_args(args)
+        with f.build():
+
+            try:
+                Event().wait()
+            except KeyboardInterrupt:
+                pass
+    else:
+        from jina.logging import default_logger
+        default_logger.critical('start a flow from CLI requires a valid "--yaml-path"')

jina/main/parser.py

@@ -19,7 +19,7 @@ def set_base_parser():
         'Code': ('🧑‍💻', 'https://github.com/jina-ai/jina/'),
         'Jina Hub': ('📦', 'https://github.com/jina-ai/jina-hub/'),
         'Home': ('🌐', 'https://jina.ai'),
-        'Join Us': ('🙌', 'hr@jina.ai')
+        'Hiring': ('🙌', 'hr@jina.ai')
     }
     url_str = '\n'.join(f'{v[0]} {k:10.10} {colored(v[1], "cyan", attrs=["underline"])}' for k, v in urls.items())
 
@@ -116,6 +116,9 @@ def set_pea_parser(parser=None):
                           '> a supported executor\'s class name, '
                           '> one of "_clear", "_route", "_forward", "_logroute", "_merge" '
                           '> the content of YAML config (must starts with "!")')  # pod(no use) -> pea
+    gp0.add_argument('--py-modules', type=str, nargs='*',
+                     help='the customized python modules need to be imported before loading the'
+                          ' executor')
 
     gp1 = add_arg_group(parser, 'pea container arguments')
     gp1.add_argument('--image', type=str,
@@ -201,6 +204,8 @@ def set_pea_parser(parser=None):
                      help='turn on remote logging')
     gp7.add_argument('--log-profile', action='store_true', default=False,
                      help='turn on the profiling logger')
+    gp7.add_argument('--override-exec-log', action='store_true', default=False,
+                     help='turn on to allow the override of the executor logger by the pea logger')
     _set_grpc_parser(parser)
     return parser
 

jina/peapods/container.py

@@ -3,7 +3,7 @@
 
 from .pea import BasePea
 from .. import __ready_msg__
-from ..helper import valid_yaml_path, kwargs2list
+from ..helper import valid_yaml_path, kwargs2list, get_non_defaults_args
 from ..logging import get_logger
 
 
@@ -17,17 +17,12 @@ def post_init(self):
         import docker
         self._client = docker.from_env()
 
-        from ..main.parser import set_pea_parser
-        _defaults = vars(set_pea_parser().parse_args([]))
-        non_defaults = {}
         # the image arg should be ignored otherwise it keeps using ContainerPea in the container
         # basically all args in BasePea-docker arg group should be ignored.
         # this prevent setting containerPea twice
-        taboo = {'image', 'entrypoint', 'volumes', 'pull_latest'}
-
-        for k, v in vars(self.args).items():
-            if k in _defaults and k not in taboo and _defaults[k] != v:
-                non_defaults[k] = v
+        from ..main.parser import set_pea_parser
+        non_defaults = get_non_defaults_args(self.args, set_pea_parser(),
+                                             taboo={'image', 'entrypoint', 'volumes', 'pull_latest'})
 
         if self.args.pull_latest:
             self._client.images.pull(self.args.image)

jina/peapods/pea.py

@@ -220,6 +220,8 @@ def load_executor(self):
             try:
                 self.executor = BaseExecutor.load_config(self.args.yaml_path,
                                                          self.args.separated_workspace, self.args.replica_id)
+                if self.args.override_exec_log:
+                    self.executor.logger = self.logger
                 self.executor.attach(pea=self)
                 # self.logger = get_logger('%s(%s)' % (self.name, self.executor.name), **vars(self.args))
             except FileNotFoundError:
@@ -296,6 +298,7 @@ def msg_callback(self, msg: 'jina_pb2.Message') -> Optional['jina_pb2.Message']:
 
     def loop_body(self):
         """The body of the request loop """
+        self.load_plugins()
         self.load_executor()
         self.zmqlet = Zmqlet(self.args, logger=self.logger)
         self.set_ready()
@@ -313,6 +316,11 @@ def loop_body(self):
             # t_loop_end = time.perf_counter()
             # self.logger.info(f'handle {(t_callback - t_loop_start) / (t_loop_end - t_loop_start):2.2f}')
 
+    def load_plugins(self):
+        if self.args.py_modules:
+            from ..helper import PathImporter
+            PathImporter.add_modules(*self.args.py_modules)
+
     def loop_teardown(self):
         """Stop the request loop """
         if hasattr(self, 'executor'):

jina/peapods/pod.py

@@ -30,12 +30,8 @@ def __init__(self, args: Union['argparse.Namespace', Dict]):
         self.is_tail_router = False
         self.deducted_head = None
         self.deducted_tail = None
+        self._args = args
         self.peas_args = self._parse_args(args)
-        self.sentinel_threads = []
-        if isinstance(args, argparse.Namespace) and getattr(args, 'shutdown_idle', False):
-            self.sentinel_threads.append(Thread(target=self.close_if_idle,
-                                                name='sentinel-shutdown-idle',
-                                                daemon=True))
 
     @property
     def is_idle(self) -> bool:
@@ -163,6 +159,11 @@ def set_runtime(self, runtime: str):
             # s.host_out = __default_host__
 
     def start_sentinels(self):
+        self.sentinel_threads = []
+        if isinstance(self._args, argparse.Namespace) and getattr(self._args, 'shutdown_idle', False):
+            self.sentinel_threads.append(Thread(target=self.close_if_idle,
+                                                name='sentinel-shutdown-idle',
+                                                daemon=True))
         for t in self.sentinel_threads:
             t.start()
 

tests/yaml/slow-flow.yml

@@ -0,0 +1,12 @@
+!Flow
+with:
+  runtime: process
+  logserver: True
+  logserver_config: test-server-config.yml
+  max_idle_time: 5
+  shutdown_idle: true
+pods:
+  sw:
+    yaml_path: SlowWorker
+    replicas: 5
+    py_modules: ./yaml/slow-worker.py

tests/yaml/slow-worker.py

@@ -0,0 +1,19 @@
+import os
+import time
+
+from jina.executors.crafters import BaseDocCrafter
+
+
+class SlowWorker(BaseDocCrafter):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # half of worker is slow
+        self.is_slow = os.getpid() % 10 != 0
+        self.logger.warning('im a slow worker')
+
+    def craft(self, doc_id, *args, **kwargs):
+        if self.is_slow:
+            self.logger.warning('slowly doing')
+            time.sleep(2)
+        return {'doc_id': doc_id}