Bring back reverted logger code (#440)

Author: nfx

src/databricks/labs/ucx/framework/tasks.py

@@ -92,6 +92,9 @@ def trigger(*argv):
         raise KeyError(msg)
 
     task_name = args.get("task", "not specified")
+    # `{{parent_run_id}}` is the run of entire workflow, whereas `{{run_id}}` is the run of a task
+    workflow_run_id = args.get("parent_run_id", "unknown_run_id")
+    job_id = args.get("job_id")
     if task_name not in _TASKS:
         msg = f'task "{task_name}" not found. Valid tasks are: {", ".join(_TASKS.keys())}'
         raise KeyError(msg)
@@ -101,9 +104,49 @@ def trigger(*argv):
     current_task = _TASKS[task_name]
     print(current_task.doc)
 
-    _install()
-
-    cfg = WorkspaceConfig.from_file(Path(args["config"]))
-    logging.getLogger("databricks").setLevel(cfg.log_level)
-
-    current_task.fn(cfg)
+    config_path = Path(args["config"])
+    cfg = WorkspaceConfig.from_file(config_path)
+
+    # see https://docs.python.org/3/howto/logging-cookbook.html
+    databricks_logger = logging.getLogger("databricks")
+    databricks_logger.setLevel(logging.DEBUG)
+
+    ucx_logger = logging.getLogger("databricks.labs.ucx")
+    ucx_logger.setLevel(logging.DEBUG)
+
+    log_path = config_path.parent / "logs" / current_task.workflow / f"run-{workflow_run_id}"
+    log_path.mkdir(parents=True, exist_ok=True)
+
+    log_file = log_path / f"{task_name}.log"
+    file_handler = logging.FileHandler(log_file.as_posix())
+    log_format = "%(asctime)s %(levelname)s [%(name)s] {%(threadName)s} %(message)s"
+    log_formatter = logging.Formatter(fmt=log_format, datefmt="%H:%M:%S")
+    file_handler.setFormatter(log_formatter)
+    file_handler.setLevel(logging.DEBUG)
+
+    console_handler = _install(cfg.log_level)
+    databricks_logger.removeHandler(console_handler)
+    databricks_logger.addHandler(file_handler)
+
+    ucx_logger.info(f"See debug logs at {log_file}")
+
+    log_readme = log_path.joinpath("README.md")
+    if not log_readme.exists():
+        # this may race when run from multiple tasks, but let's accept the risk for now.
+        with log_readme.open(mode="w") as f:
+            f.write(f"# Logs for the UCX {current_task.workflow} workflow\n")
+            f.write("This folder contains UCX log files.\n\n")
+            f.write(f"See the [{current_task.workflow} job](/#job/{job_id}) and ")
+            f.write(f"[run #{workflow_run_id}](/#job/{job_id}/run/{workflow_run_id})\n")
+
+    try:
+        current_task.fn(cfg)
+    except BaseException as error:
+        log_file_for_cli = str(log_file).lstrip("/Workspace")
+        cli_command = f"databricks workspace export /{log_file_for_cli}"
+        ucx_logger.error(f"Task crashed. Execute `{cli_command}` locally to troubleshoot with more details. {error}")
+        databricks_logger.debug("Task crash details", exc_info=error)
+        file_handler.flush()
+        raise
+    finally:
+        file_handler.close()