Refactored general logging data. Added memory data to processing logs.

nomad/app/v1/routers/info.py

@@ -77,7 +77,7 @@ class InfoModel(BaseModel):
     statistics: StatisticsModel = Field(None, description='General NOMAD statistics')
     search_quantities: dict
     version: str
-    release: str
+    deployment: str
     git: GitInfoModel
     oasis: bool
 
@@ -134,7 +134,7 @@ async def get_info():
             if 'optimade' not in s.qualified_name
         },
         'version': config.meta.version,
-        'release': config.meta.release,
+        'deployment': config.meta.deployment,
         'git': {
             'ref': gitinfo.ref,
             'version': gitinfo.version,

nomad/cli/cli.py

@@ -51,9 +51,11 @@ class POPO(dict):
     'It uses a sub-command structure similar to the git command.'))
 @click.option('-v', '--verbose', help='sets log level to info', is_flag=True)
 @click.option('--debug', help='sets log level to debug', is_flag=True)
+@click.option('--log-label', type=str, help='Label applied to logg entries.')
 @click.pass_context
-def cli(ctx, verbose: bool, debug: bool):
+def cli(ctx, verbose: bool, debug: bool, log_label: str):
     config.meta.service = os.environ.get('NOMAD_SERVICE', 'cli')
+    config.meta.label = log_label
 
     if debug:
         config.console_log_level = logging.DEBUG

nomad/config.py

@@ -309,8 +309,8 @@ datacite = NomadConfig(
 meta = NomadConfig(
     version='1.0.0',
     commit=gitinfo.commit,
-    release='devel',
-    deployment='standard',
+    deployment='devel',
+    label=None,
     default_domain='dft',
     service='unknown nomad service',
     name='novel materials discovery (NOMAD)',

nomad/metainfo/elasticsearch_extension.py

@@ -900,7 +900,7 @@ def index_entries(entries: List, refresh: bool = False):
             except Exception as e:
                 logger.error('could not create entry index doc', calc_id=entry['entry_id'], exc_info=e)
 
-        timer_kwargs = {}
+        timer_kwargs: Dict[str, Any] = {}
         try:
             import json
             timer_kwargs['size'] = len(json.dumps(actions_and_docs))
@@ -1119,7 +1119,7 @@ def update_materials(entries: List, refresh: bool = False):
         all_n_entries += material_doc['n_entries']
 
     # Execute the created actions in bulk.
-    timer_kwargs = {}
+    timer_kwargs: Dict[str, Any] = {}
     try:
         import json
         timer_kwargs['size'] = len(json.dumps(_actions_and_docs_bulks))

nomad/processing/base.py

@@ -520,6 +520,10 @@ def proc_task(task, cls_name, self_id, func_attr, process_args, process_kwargs):
     might happen in sharded, distributed mongo setups where the object might not
     have yet been propagated and therefore appear missing.
     '''
+    if '_meta_label' in process_kwargs:
+        config.meta.label = process_kwargs['_meta_label']
+        del(process_kwargs['_meta_label'])
+
     self = unwarp_task(task, cls_name, self_id)
 
     logger = self.get_logger()
@@ -545,7 +549,7 @@ def proc_task(task, cls_name, self_id, func_attr, process_args, process_kwargs):
     # call the process function
     try:
         os.chdir(config.fs.working_directory)
-        with utils.timer(logger, 'process executed on worker'):
+        with utils.timer(logger, 'process executed on worker', log_memory=True):
             # Actually call the process function
             self.process_status = ProcessStatus.RUNNING
             self.last_status_message = 'Started process: ' + func_attr
@@ -610,6 +614,8 @@ def process(func):
         self.process_status = ProcessStatus.PENDING
         self.save()
 
+        kwargs['_meta_label'] = config.meta.label
+
         self._run_process(func, args, kwargs)
 
     setattr(wrapper, '__process_unwrapped', func)

nomad/processing/data.py

@@ -87,7 +87,7 @@ def _pack_log_event(logger, method_name, event_dict):
         log_data.update(**{
             key: value
             for key, value in getattr(logger, '_context', {}).items()
-            if key not in ['service', 'release', 'upload_id', 'calc_id', 'mainfile', 'process_status']})
+            if key not in ['service', 'deployment', 'upload_id', 'calc_id', 'mainfile', 'process_status']})
         log_data.update(logger=logger.name)
 
         return log_data

nomad/utils/__init__.py

@@ -52,6 +52,8 @@ import collections
 import logging
 import inspect
 import orjson
+import resource
+import os
 
 from nomad import config
 
@@ -214,7 +216,7 @@ def lnr(logger, event, **kwargs):
 
 
 @contextmanager
-def timer(logger, event, method='info', lnr_event: str = None, **kwargs):
+def timer(logger, event, method='info', lnr_event: str = None, log_memory: bool = False, **kwargs):
     '''
     A context manager that takes execution time and produces a log entry with said time.
 
@@ -223,22 +225,36 @@ def timer(logger, event, method='info', lnr_event: str = None, **kwargs):
         event: The log message/event.
         method: The log method that should be used. Must be a valid logger method name.
             Default is 'info'.
+        log_memory: Log process memory usage before and after.
         **kwargs: Additional logger data that is passed to the log entry.
 
     Returns:
         The method yields a dictionary that can be used to add further log data.
     '''
+    kwargs = dict(kwargs)
     start = time.time()
+    if log_memory:
+        rss_before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+        kwargs['pid'] = os.getpid()
+        kwargs['exec_rss_before'] = rss_before
 
     try:
         yield kwargs
     except Exception as e:
         if lnr_event is not None:
             stop = time.time()
+            if log_memory:
+                rss_after = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+                kwargs['exec_rss_after'] = rss_after
+                kwargs['exec_rss_delta'] = rss_before - rss_after
             logger.error(lnr_event, exc_info=e, exec_time=stop - start, **kwargs)
         raise e
     finally:
         stop = time.time()
+        if log_memory:
+            rss_after = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+            kwargs['exec_rss_after'] = rss_after
+            kwargs['exec_rss_delta'] = rss_before - rss_after
 
     if logger is None:
         print(event, stop - start)

nomad/utils/structlogging.py

@@ -143,11 +143,12 @@ class LogstashFormatter(logstash.formatter.LogstashFormatterBase):
 
             # Nomad specific
             'nomad.service': config.meta.service,
-            'nomad.release': config.meta.release,
+            'nomad.deployment': config.meta.deployment,
             'nomad.version': config.meta.version,
-            'nomad.commit': config.meta.commit,
-            'nomad.deployment': config.meta.deployment
+            'nomad.commit': config.meta.commit
         }
+        if config.meta.label:
+            message['nomad.label'] = config.meta.label
 
         if record.name.startswith('nomad'):
             for key, value in structlog.items():
@@ -233,7 +234,7 @@ class ConsoleFormatter(LogstashFormatter):
                 print_key = key[6:]
             else:
                 print_key = key
-            if not cls.short_format or print_key not in ['release', 'service']:
+            if not cls.short_format or print_key not in ['deployment', 'service']:
                 out.write('\n  - %s: %s' % (print_key, str(message_dict.get(key, None))))
         return out.getvalue()
 
@@ -245,7 +246,7 @@ def add_logstash_handler(logger):
 
     if logstash_handler is None:
         logstash_handler = LogstashHandler()
-        logstash_handler.formatter = LogstashFormatter(tags=['nomad', config.meta.release])
+        logstash_handler.formatter = LogstashFormatter(tags=['nomad', config.meta.deployment])
         logstash_handler.setLevel(config.logstash.level)
         logger.addHandler(logstash_handler)
 

ops/docker-compose/nomad-oasis/README.md

@@ -196,7 +196,7 @@ keycloak:
   oasis: true
 
 meta:
-  release: 'oasis'
+  deployment: 'oasis'
   deployment_id: '<your-host>'
   maintainer_email: '<oasis admin email>'