20.2.4b

Author: lightforever

mlcomp/__init__.py

@@ -18,6 +18,7 @@
 LOG_FOLDER = join(ROOT_FOLDER, 'logs')
 CONFIG_FOLDER = join(ROOT_FOLDER, 'configs')
 DB_FOLDER = join(ROOT_FOLDER, 'db')
+REPORT_FOLDER = join(ROOT_FOLDER, 'report')
 TMP_FOLDER = join(ROOT_FOLDER, 'tmp')
 
 os.makedirs(ROOT_FOLDER, exist_ok=True)
@@ -27,6 +28,7 @@
 os.makedirs(LOG_FOLDER, exist_ok=True)
 os.makedirs(CONFIG_FOLDER, exist_ok=True)
 os.makedirs(DB_FOLDER, exist_ok=True)
+os.makedirs(REPORT_FOLDER, exist_ok=True)
 os.makedirs(TMP_FOLDER, exist_ok=True)
 
 # copy conf files if they do not exist
@@ -118,5 +120,5 @@
     'FILE_LOG_LEVEL', 'DB_TYPE', 'SA_CONNECTION_STRING', 'FLASK_ENV',
     'DOCKER_MAIN', 'IP', 'PORT', 'LOG_NAME', 'WORKER_USAGE_INTERVAL',
     'FILE_SYNC_INTERVAL', 'INSTALL_DEPENDENCIES', 'SYNC_WITH_THIS_COMPUTER',
-    'CAN_PROCESS_TASKS', 'TMP_FOLDER', 'CONTOUR_FILE'
+    'CAN_PROCESS_TASKS', 'TMP_FOLDER', 'CONTOUR_FILE', 'REPORT_FOLDER'
 ]

mlcomp/__main__.py

@@ -7,10 +7,11 @@
 
 import torch
 
+from mlcomp.utils.io import yaml_load, yaml_dump
 from mlcomp.contrib.search.grid import grid_cells
 from mlcomp.migration.manage import migrate as _migrate
 from mlcomp import ROOT_FOLDER, IP, PORT, \
-    WORKER_INDEX, SYNC_WITH_THIS_COMPUTER, CAN_PROCESS_TASKS
+    WORKER_INDEX, SYNC_WITH_THIS_COMPUTER, CAN_PROCESS_TASKS, CONFIG_FOLDER
 from mlcomp.db.core import Session
 from mlcomp.db.enums import DagType, ComponentType, TaskStatus
 from mlcomp.db.models import Computer
@@ -19,12 +20,13 @@
     TaskProvider, \
     StepProvider, \
     ProjectProvider
+from mlcomp.report import create_report, check_statuses
 from mlcomp.utils.config import merge_dicts_smart, dict_from_list_str
-from mlcomp.utils.io import yaml_load, yaml_dump
 from mlcomp.utils.logging import create_logger
 from mlcomp.worker.sync import sync_directed
 from mlcomp.worker.tasks import execute_by_id
-from mlcomp.utils.misc import memory, disk, get_username
+from mlcomp.utils.misc import memory, disk, get_username, \
+    get_default_network_interface
 from mlcomp.server.back.create_dags import dag_standard, dag_pipe
 
 _session = Session.create_session(key=__name__)
@@ -105,19 +107,21 @@ def migrate():
 @click.option('--control_reqs', type=bool, default=True)
 @click.option('--params', multiple=True)
 def dag(config: str, control_reqs: bool, params):
+    check_statuses()
     _dag(config, control_reqs=control_reqs, params=params)
 
 
 @main.command()
-def analyze():
-    pass
+def report():
+    create_report()
 
 
 @main.command()
 @click.argument('config')
 @click.option('--debug', type=bool, default=True)
 @click.option('--params', multiple=True)
 def execute(config: str, debug: bool, params):
+    check_statuses()
     _create_computer()
 
     # Fail all InProgress Tasks
@@ -164,6 +168,7 @@ def execute(config: str, debug: bool, params):
     help='only copy files from all the others to the computer'
 )
 def sync(project: str, computer: str, only_from: bool, only_to: bool):
+    check_statuses()
     _create_computer()
 
     computer = computer or socket.gethostname()
@@ -193,9 +198,14 @@ def sync(project: str, computer: str, only_from: bool, only_to: bool):
 
 @main.command()
 def init():
-    # already done by importing mlcomp
-    # that is needed to import it
-    pass
+    env_path = join(CONFIG_FOLDER, '.env')
+    lines = open(env_path).readlines()
+    for i in range(len(lines)):
+        if 'NCCL_SOCKET_IFNAME' in lines[i]:
+            interface = get_default_network_interface()
+            if interface:
+                lines[i] = f'NCCL_SOCKET_IFNAME={interface}\n'
+    open(env_path, 'w').writelines(lines)
 
 
 if __name__ == '__main__':

mlcomp/__version__.py

@@ -1 +1 @@
-__version__ = '20.2.4a'
+__version__ = '20.2.4b'

mlcomp/db/providers/dag.py

@@ -215,5 +215,8 @@ def remove_all(self, ids: List[int]):
             synchronize_session=False)
         self.commit()
 
+    def count(self):
+        return self.query(Dag).count()
+
 
 __all__ = ['DagProvider']

mlcomp/db/providers/log.py

@@ -59,19 +59,24 @@ def get(self, filter: dict, options: PaginatorOptions):
                 'component': to_snake(ComponentType(log.component).name),
                 'computer': log.computer,
                 'step': self.to_dict(step) if step else None,
-                'task': self.to_dict(task, rules=('-additional_info', ))
+                'task': self.to_dict(task, rules=('-additional_info',))
                 if task else None
             }
             data.append(item)
 
         return {'total': total, 'data': data}
 
-    def last(self, count: int, dag: int = None, task: int = None):
+    def last(self, count: int, dag: int = None, task: int = None,
+             levels=None, components=None):
         query = self.query(Log, Task.id).outerjoin(Task)
         if dag is not None:
             query = query.filter(Task.dag == dag)
         if task is not None:
             query = query.filter(Task.id == task)
+        if levels is not None:
+            query = query.filter(Log.level.in_(levels))
+        if components is not None:
+            query = query.filter(Log.component.in_(components))
         return query.order_by(Log.id.desc()).limit(count).all()
 
 

mlcomp/report.py

@@ -0,0 +1,171 @@
+import shutil
+import traceback
+from glob import glob
+from os import makedirs
+from os.path import dirname, join, basename, exists
+
+import pandas as pd
+import migrate.versioning.api as api
+from mlcomp.utils.io import zip_folder
+
+from mlcomp.db.enums import LogStatus, ComponentType
+
+from mlcomp import SA_CONNECTION_STRING, REPORT_FOLDER, LOG_FOLDER, DB_TYPE, \
+    CONFIG_FOLDER, ROOT_FOLDER, DATA_FOLDER, MODEL_FOLDER, TASK_FOLDER, \
+    DB_FOLDER, TMP_FOLDER
+from mlcomp.utils.misc import now, to_snake
+
+from mlcomp.db.providers import DagProvider, LogProvider
+
+
+def statuses(folder: str = None):
+    rows = []
+
+    folder_status = 'OK'
+    folder_comment = ''
+
+    folders = [
+        ROOT_FOLDER,
+        DATA_FOLDER,
+        MODEL_FOLDER,
+        TASK_FOLDER,
+        LOG_FOLDER,
+        CONFIG_FOLDER,
+        DB_FOLDER,
+        REPORT_FOLDER,
+        TMP_FOLDER
+    ]
+    for f in folders:
+        if not exists(f):
+            folder_status = 'ERROR'
+            folder_comment = f'folder {f} does not exist'
+
+    files = [
+        join(CONFIG_FOLDER, '.env')
+    ]
+    for f in files:
+        if not exists(f):
+            folder_status = 'ERROR'
+            folder_comment = f'file {f} does not exist'
+
+    rows.append({
+        'name': 'Folders',
+        'status': folder_status,
+        'comment': folder_comment
+    })
+
+    database_status = 'OK'
+    database_comment = f'DB_TYPE = {DB_TYPE}'
+    try:
+        provider = DagProvider()
+        provider.count()
+    except Exception:
+        database_status = 'ERROR'
+        database_comment += ' ' + traceback.format_exc()
+
+    rows.append({
+        'name': 'Database',
+        'status': database_status,
+        'comment': database_comment
+    })
+
+    if database_status == 'OK':
+        migrate_status = 'OK'
+
+        repository_folder = join(dirname(__file__), 'migration')
+        repository_version = api.version(repository_folder)
+
+        db_version = api.db_version(SA_CONNECTION_STRING, repository_folder)
+
+        if db_version != repository_version:
+            migrate_status = 'ERROR'
+            migrate_comment = f'Repository version = {repository_version} ' \
+                              f'Db version = {db_version}'
+        else:
+            migrate_comment = f'version: {db_version}'
+
+        rows.append({
+            'name': 'Migrate',
+            'status': migrate_status,
+            'comment': migrate_comment
+        })
+
+    df = pd.DataFrame(rows)
+
+    if folder is not None:
+        print('Statuses:')
+        print(df)
+
+        df.to_csv(join(folder, 'statuses.csv'), index=False)
+
+    return df
+
+
+def check_statuses():
+    stats = statuses()
+    failed = stats[stats['status'] != 'OK']
+    if failed.shape[0] > 0:
+        print('There are errors in statuses')
+        for row in stats.itertuples():
+            print(f'name: {row.name} status'
+                  f' {row.status} comment {row.comment}')
+
+        import time
+        time.sleep(0.01)
+
+        raise Exception('There are errors in statuses. '
+                        'Please check them above')
+
+
+def logs(folder: str = None):
+    log_provider = LogProvider()
+    errors = log_provider.last(count=1000, levels=[LogStatus.Error.value])
+    service_components = [ComponentType.Supervisor.value,
+                          ComponentType.API.value,
+                          ComponentType.WorkerSupervisor.value]
+    services = log_provider.last(count=1000, components=service_components)
+    logs = errors + services
+
+    rows = []
+    for l, _ in logs:
+        rows.append({
+            'status': to_snake(LogStatus(l.level).name),
+            'component': to_snake(ComponentType(l.component).name),
+            'time': l.time,
+            'message': l.message,
+        })
+    df = pd.DataFrame(rows)
+    df.to_csv(join(folder, 'logs_db.csv'), index=False)
+
+    if folder is not None:
+        for file in glob(join(LOG_FOLDER, '*')):
+            shutil.copy(file, join(folder, basename(file)))
+        print('logs formed')
+    return df
+
+
+def create_report():
+    print('*** Report Start ***')
+    print()
+
+    folder = join(REPORT_FOLDER, f'{now()}'.split('.')[0])
+    makedirs(folder, exist_ok=True)
+
+    statuses(folder)
+
+    print()
+
+    logs(folder)
+
+    print()
+
+    zip_path = folder + '.zip'
+    zip_folder(folder, dst=zip_path)
+
+    print('Report path', zip_path)
+    print()
+
+    print('*** Report End ***')
+
+
+__all__ = ['check_statuses', 'create_report']

mlcomp/server/__main__.py

@@ -4,6 +4,7 @@
 import click
 
 from mlcomp import CONFIG_FOLDER, REDIS_PORT, REDIS_PASSWORD
+from mlcomp.report import check_statuses
 from mlcomp.server.back.app import start_server as _start_server
 from mlcomp.server.back.app import stop_server as _stop_server
 from mlcomp.utils.misc import kill_child_processes
@@ -19,6 +20,7 @@ def start_site():
     """
     Start only site
     """
+    check_statuses()
     _start_server()
 
 
@@ -27,6 +29,7 @@ def stop_site():
     """
     Stop site
     """
+    check_statuses()
     _stop_server()
 
 
@@ -45,6 +48,8 @@ def start(daemon: bool, debug: bool, workers: int, log_level: str):
 
     It starts: redis-server, site, worker_supervisor, workers
     """
+    check_statuses()
+
     # creating supervisord config
     supervisor_command = 'mlcomp-worker worker-supervisor'
     worker_command = 'mlcomp-worker worker'
@@ -93,6 +98,8 @@ def stop():
     """
     Stop supervisord started by start command
     """
+    check_statuses()
+
     lines = os.popen('ps -ef | grep supervisord').readlines()
     for line in lines:
         if 'mlcomp/configs/supervisord.conf' not in line:

mlcomp/utils/misc.py

@@ -57,7 +57,7 @@ def dict_func(objcts: List, func=np.mean):
 
 def get_pid(name):
     res = []
-    lines = check_output(["ps", '-ef']).decode().split('\n')
+    lines = check_output(['ps', '-ef']).decode().split('\n')
     header = lines[0].split()
     for line in lines[1:]:
         if line.strip() == '':
@@ -80,6 +80,20 @@ def get_pid(name):
     return res
 
 
+def get_default_network_interface():
+    lines = check_output(['route']).decode().split('\n')
+    header = lines[1].split()
+
+    for line in lines[2:]:
+        parts = line.split()
+        item = dict()
+        for h, p in zip(header, parts):
+            item[h] = p
+        if item['Destination'] == 'default' and 'Iface' in item:
+            return item['Iface']
+    return None
+
+
 def now():
     return datetime.utcnow()