Task/WG-156 improve session handling in long running tasks (#144)

* Remove incorrect logging statements

* Refactor point cloud processing to use different db sessions

* Improve logging

* Close session quicker

* Fix success/failure notification blocks after conversion

* Fix linting error

geoapi/tasks/external_data.py

@@ -154,9 +154,7 @@ def import_point_clouds_from_agave(userId: int, files, pointCloudId: int):
         point_cloud.task = task
         session.add(point_cloud)
         session.add(task)
-        logger.error("committing")
         session.commit()
-        logger.error("done committing")
 
         new_asset_files = []
         failed_message = None
@@ -214,17 +212,23 @@ def import_point_clouds_from_agave(userId: int, files, pointCloudId: int):
                                     "success",
                                     "Running potree converter (for point cloud {}).".format(pointCloudId))
 
-        try:
-            convert_to_potree(pointCloudId)
+    try:
+        # use potree converter to convert las to web-friendly format
+        # this operation is memory-intensive and time-consuming.
+        convert_to_potree(pointCloudId)
+        with create_task_session() as session:
+            user = session.query(User).get(userId)
+            logger.info(f"point cloud:{pointCloudId} conversion completed for user:{user.username} and files:{files}")
             NotificationsService.create(session,
                                         user,
                                         "success",
                                         "Completed potree converter (for point cloud {}).".format(pointCloudId))
-        except:  # noqa: E722
-            logger.exception("point cloud:{} conversion failed for user:{}".format(pointCloudId, user.username))
+    except:  # noqa: E722
+        with create_task_session() as session:
+            user = session.query(User).get(userId)
+            logger.exception(f"point cloud:{pointCloudId} conversion failed for user:{user.username} and files:{files}")
             _update_point_cloud_task(session, pointCloudId, description="", status="FAILED")
             NotificationsService.create(session, user, "error", "Processing failed for point cloud ({})!".format(pointCloudId))
-            return
 
 
 @app.task(rate_limit="5/s")

geoapi/tasks/lidar.py

@@ -71,6 +71,9 @@ def on_failure(self, exc, task_id, args, kwargs, einfo):
 def convert_to_potree(self, pointCloudId: int) -> None:
     """
     Use the potree converter to convert a LAS/LAZ file to potree format
+
+    Note: this operation is memory-intensive and time-consuming.  Large LAS files (>8 Gb) can use >50gb of memory.
+
     :param pointCloudId: int
     :return: None
     """
@@ -79,32 +82,35 @@ def convert_to_potree(self, pointCloudId: int) -> None:
 
     with create_task_session() as session:
         point_cloud = PointCloudService.get(session, pointCloudId)
-
+        conversion_parameters = point_cloud.conversion_parameters
         path_to_original_point_clouds = get_asset_path(point_cloud.path, PointCloudService.ORIGINAL_FILES_DIR)
         path_temp_processed_point_cloud_path = get_asset_path(point_cloud.path, PointCloudService.PROCESSED_DIR)
 
-        input_files = [get_asset_path(path_to_original_point_clouds, file)
-                       for file in os.listdir(path_to_original_point_clouds)
-                       if pathlib.Path(file).suffix.lstrip('.').lower() in PointCloudService.LIDAR_FILE_EXTENSIONS]
-
-        outline = get_bounding_box_2d(input_files)
-
-        command = [
-            "PotreeConverter",
-            "--verbose",
-            "-i",
-            path_to_original_point_clouds,
-            "-o",
-            path_temp_processed_point_cloud_path,
-            "--overwrite",
-            "--generate-page",
-            "index"
-        ]
-        if point_cloud.conversion_parameters:
-            command.extend(point_cloud.conversion_parameters.split())
-        logger.info("Processing point cloud (#{}):  {}".format(pointCloudId, " ".join(command)))
-        subprocess.run(command, check=True, capture_output=True, text=True)
+    input_files = [get_asset_path(path_to_original_point_clouds, file)
+                   for file in os.listdir(path_to_original_point_clouds)
+                   if pathlib.Path(file).suffix.lstrip('.').lower() in PointCloudService.LIDAR_FILE_EXTENSIONS]
+
+    outline = get_bounding_box_2d(input_files)
+
+    command = [
+        "PotreeConverter",
+        "--verbose",
+        "-i",
+        path_to_original_point_clouds,
+        "-o",
+        path_temp_processed_point_cloud_path,
+        "--overwrite",
+        "--generate-page",
+        "index"
+    ]
+    if conversion_parameters:
+        command.extend(conversion_parameters.split())
+    logger.info("Processing point cloud (#{}):  {}".format(pointCloudId, " ".join(command)))
+
+    subprocess.run(command, check=True, capture_output=True, text=True)
 
+    with create_task_session() as session:
+        point_cloud = PointCloudService.get(session, pointCloudId)
         # Create preview viewer html (with no menu and now nsf logo)
         with open(os.path.join(path_temp_processed_point_cloud_path, "preview.html"), 'w+') as preview:
             with open(os.path.join(path_temp_processed_point_cloud_path, "index.html"), 'r') as viewer:
@@ -113,32 +119,33 @@ def convert_to_potree(self, pointCloudId: int) -> None:
                 content = content.replace("viewer.toggleSidebar()", "$('.potree_menu_toggle').hide()")
                 preview.write(content)
 
-        if point_cloud.feature_id:
-            feature = point_cloud.feature
-        else:
-            feature = Feature()
-            feature.project_id = point_cloud.project_id
-
-            asset_uuid = uuid.uuid4()
-            base_filepath = make_project_asset_dir(point_cloud.project_id)
-            asset_path = os.path.join(base_filepath, str(asset_uuid))
-            fa = FeatureAsset(
-                uuid=asset_uuid,
-                asset_type="point_cloud",
-                path=get_asset_relative_path(asset_path),
-                display_path=point_cloud.description,
-                feature=feature
-            )
-            feature.assets.append(fa)
-            point_cloud.feature = feature
-
-        feature.the_geom = from_shape(geometries.convert_3D_2D(outline), srid=4326)
-        point_cloud.task.status = "FINISHED"
-        point_cloud.task.description = ""
-
-        point_cloud_asset_path = get_asset_path(feature.assets[0].path)
-        shutil.rmtree(point_cloud_asset_path, ignore_errors=True)
-        shutil.move(path_temp_processed_point_cloud_path, point_cloud_asset_path)
-        session.add(point_cloud)
-        session.add(feature)
-        session.commit()
+            if point_cloud.feature_id:
+                feature = point_cloud.feature
+            else:
+                feature = Feature()
+                feature.project_id = point_cloud.project_id
+
+                asset_uuid = uuid.uuid4()
+                base_filepath = make_project_asset_dir(point_cloud.project_id)
+                asset_path = os.path.join(base_filepath, str(asset_uuid))
+                fa = FeatureAsset(
+                    uuid=asset_uuid,
+                    asset_type="point_cloud",
+                    path=get_asset_relative_path(asset_path),
+                    display_path=point_cloud.description,
+                    feature=feature
+                )
+                feature.assets.append(fa)
+                point_cloud.feature = feature
+
+            feature.the_geom = from_shape(geometries.convert_3D_2D(outline), srid=4326)
+            point_cloud.task.status = "FINISHED"
+            point_cloud.task.description = ""
+
+            point_cloud_asset_path = get_asset_path(feature.assets[0].path)
+            session.add(point_cloud)
+            session.add(feature)
+            session.commit()
+
+            shutil.rmtree(point_cloud_asset_path, ignore_errors=True)
+            shutil.move(path_temp_processed_point_cloud_path, point_cloud_asset_path)