Cleanup a bit for logging

Author: HansKallekleiv

backend/src/backend/auth/enforce_logged_in_middleware.py

@@ -32,7 +32,7 @@ def __init__(
         self._paths_redirected_to_login = paths_redirected_to_login or []
 
     async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
-        print("##################################### ensure_logged_in")
+        # print("##################################### ensure_logged_in")
 
         path_to_check = request.url.path
 
@@ -42,27 +42,27 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -
         if root_path:
             path_to_check = path_to_check.replace(root_path, "")
 
-        print(f"##### {request.url.path=}")
-        print(f"##### {root_path=}")
-        print(f"##### {path_to_check=}")
+        # print(f"##### {request.url.path=}")
+        # print(f"##### {root_path=}")
+        # print(f"##### {path_to_check=}")
 
         path_is_protected = True
 
         if path_to_check in ["/login", "/auth-callback"] + self._unprotected_paths:
             path_is_protected = False
-            print(f"##### path not protected: {path_to_check=}")
+            # print(f"##### path not protected: {path_to_check=}")
 
         if path_is_protected:
-            print(f"##### path requires login:  {path_to_check=}")
+            # print(f"##### path requires login:  {path_to_check=}")
             await starsessions.load_session(request)
 
             authenticated_user = AuthHelper.get_authenticated_user(request)
             is_logged_in = authenticated_user is not None
-            print(f"##### {is_logged_in=}")
+            # print(f"##### {is_logged_in=}")
 
             if not is_logged_in:
                 if path_to_check in self._paths_redirected_to_login:
-                    print("##### LOGGING IN USING REDIRECT")
+                    # print("##### LOGGING IN USING REDIRECT")
                     target_url_b64 = base64.urlsafe_b64encode(str(request.url).encode()).decode()
                     return RedirectResponse(f"{root_path}/login?redirect_url_after_login={target_url_b64}")
                 return PlainTextResponse("Not authorized yet, must log in", 401)

backend/src/backend/user_session/routers/surface/router.py

@@ -22,9 +22,13 @@
 from .test_go import go_get_surface_blobs
 from sumo.wrapper import SumoClient
 from fmu.sumo.explorer.objects import CaseCollection, Case, SurfaceCollection
+from src.services.utils.perf_timer import PerfTimer
+import base64
+from io import BytesIO
 
-router = APIRouter()
 LOGGER = logging.getLogger(__name__)
+router = APIRouter()
+
 
 cache = Cache(Cache.MEMORY, ttl=3600)
 # @router.post("/well_intersection_reals_from_user_session")
@@ -68,22 +72,17 @@ async def well_intersection_reals_from_user_session(
 ) -> List[schemas.SurfaceIntersectionPoints]:
     body = await request.json()
     polyline = schemas.FencePolyline(**body.get("polyline"))
-
+    timer = PerfTimer()
     # Config
     case_uuid = polyline.case_uuid
     snames = polyline.names
     sattr = polyline.attribute
     ensemble_name = polyline.ensemble_name
-    sumo_client = SumoClient(env="prod", token=authenticated_user.get_sumo_access_token(), interactive=False)
-    case_collection = CaseCollection(sumo_client).filter(uuid=case_uuid)
-    case = case_collection[0]
-    surface_collection = case.surfaces.filter(
-        iteration=ensemble_name,
-        name=snames,
-        tagname=sattr,
-        realization=polyline.realization_nums,
-    )
-    uuids = [surf.uuid for surf in surface_collection]
+    # Get uuids
+    uuids = get_uuids(case_uuid, ensemble_name, polyline, snames, sattr, authenticated_user.get_sumo_access_token())
+    elapsed_meta = timer.lap_ms()
+
+    # Check if cached
     uuids_to_download = []
     surfaces = []
     for uuid in uuids:
@@ -92,13 +91,22 @@ async def well_intersection_reals_from_user_session(
             uuids_to_download.append(uuid)
         else:
             surfaces.append(possible_surface)
-    downloaded_surface_dict = go_get_surface_blobs(
-        authenticated_user.get_sumo_access_token(), case_uuid, uuids_to_download
-    )
+    elapsed_cache = timer.lap_ms()
+
+    # Download remaining
+    data_map_b64 = go_get_surface_blobs(authenticated_user.get_sumo_access_token(), case_uuid, uuids_to_download)
+    elapsed_download = timer.lap_ms()
+
+    # Convert to xtgeo
+    downloaded_surface_dict = await b64_to_xtgeo(data_map_b64)
+    elapsed_xtgeo = timer.lap_ms()
+
+    # Add to cache
     for uuid, surface in downloaded_surface_dict.items():
         # await cache.set(f"{authenticated_user._user_id}-{uuid}", surface)
         surfaces.append(surface)
 
+    # Intersect
     fence_arr = np.array(
         [
             polyline.x_points,
@@ -107,8 +115,52 @@ async def well_intersection_reals_from_user_session(
             polyline.cum_length,
         ]
     ).T
+    intersections = await make_intersections(surfaces, fence_arr)
+    elapsed_intersect = timer.lap_ms()
 
-    def worker(surf):
+    result = [intersection.dict() for intersection in intersections]
+    elapsed_response_format = timer.lap_ms()
+    LOGGER.info(
+        f"Got intersected surface set from Sumo: {timer.elapsed_ms()}ms ("
+        f"meta={elapsed_meta}ms, "
+        f"cache={elapsed_cache}ms, "
+        f"download={elapsed_download}ms, "
+        f"xtgeo={elapsed_xtgeo}ms, "
+        f"intersect={elapsed_intersect}ms, "
+        f"response_format={elapsed_response_format}ms) ",
+        extra={
+            "meta": elapsed_meta,
+            "cache": elapsed_cache,
+            "download": elapsed_download,
+            "xtgeo": elapsed_xtgeo,
+            "intersect": elapsed_intersect,
+            "response_format": elapsed_response_format,
+        },
+    )
+    return ORJSONResponse(result)
+
+
+async def b64_to_xtgeo(data_map_b64):
+    def to_xtgeo(object_id, b64_blob):
+        bytestr = base64.b64decode(b64_blob)
+        xtgeo_surface = xtgeo.surface_from_file(BytesIO(bytestr), fformat="irap_binary")
+        return {object_id: xtgeo_surface}
+
+    loop = asyncio.get_running_loop()
+    with ThreadPoolExecutor() as executor:
+        tasks = [
+            loop.run_in_executor(executor, to_xtgeo, object_id, b64_blob)
+            for object_id, b64_blob in data_map_b64.items()
+        ]
+        results = await asyncio.gather(*tasks)
+    downloaded_surface_dict = {}
+    for result in results:
+        downloaded_surface_dict.update(result)
+    return downloaded_surface_dict
+
+
+async def make_intersections(surfaces, fence_arr):
+    def make_intersection(surf):
         line = surf.get_randomline(fence_arr)
         intersection = schemas.SurfaceIntersectionPoints(
             name=f"{surf.name}",
@@ -120,8 +172,19 @@ def worker(surf):
     loop = asyncio.get_running_loop()
 
     with ThreadPoolExecutor() as executor:
-        tasks = [loop.run_in_executor(executor, worker, surf) for surf in surfaces]
+        tasks = [loop.run_in_executor(executor, make_intersection, surf) for surf in surfaces]
         intersections = await asyncio.gather(*tasks)
+    return intersections
 
-    result = [intersection.dict() for intersection in intersections]
-    return ORJSONResponse(result)
+
+def get_uuids(case_uuid, ensemble_name, polyline, snames, sattr, bearer_token):
+    sumo_client = SumoClient(env="prod", token=bearer_token, interactive=False)
+    case_collection = CaseCollection(sumo_client).filter(uuid=case_uuid)
+    case = case_collection[0]
+    surface_collection = case.surfaces.filter(
+        iteration=ensemble_name,
+        name=snames,
+        tagname=sattr,
+        realization=polyline.realization_nums,
+    )
+    return [surf.uuid for surf in surface_collection]

backend/src/backend/user_session/routers/surface/test_go.py

@@ -7,9 +7,14 @@
 import xtgeo
 import base64
 from . import sumo
+import logging
+from src.services.utils.perf_timer import PerfTimer
+
+LOGGER = logging.getLogger(__name__)
 
 
 def go_get_surface_blobs(sumo_token: str, case_uuid: str, object_ids: list[str]) -> list[xtgeo.RegularSurface]:
+    timer = PerfTimer()
     GetZippedBlobs = golibrary.GetZippedBlobs
     GetZippedBlobs.restype = ctypes.c_void_p
     base_uri, auth_token = sumo.get_base_uri_and_auth_token_for_case(
@@ -27,14 +32,26 @@ def go_get_surface_blobs(sumo_token: str, case_uuid: str, object_ids: list[str])
         "operation": ["mean"],
         "env": "prod",
     }
+    elapsed_init = timer.lap_ms()
     res = GetZippedBlobs(json.dumps(new_request).encode("utf-8"))
+    elapsed_get = timer.lap_ms()
     res_string = ctypes.string_at(res).decode("ascii")
     data_map_b64 = json.loads(res_string)
-    print("Done downloading surfaces", flush=True)
+    elapsed_decode = timer.lap_ms()
     surfaces = {}
+    return data_map_b64
     for object_id, b64_blob in data_map_b64.items():
         bytestr = base64.b64decode(b64_blob)
         xtgeo_surface = xtgeo.surface_from_file(BytesIO(bytestr), fformat="irap_binary")
         surfaces[object_id] = xtgeo_surface
-    print("Done decoding surfaces", flush=True)
+    elapsed_xtgeo = timer.lap_ms()
+    LOGGER.info(
+        f"Got surface set from Sumo in: {timer.elapsed_ms()}ms ("
+        f"init go and query sumo={elapsed_init}ms, "
+        f"get={elapsed_get}ms, "
+        f"decode={elapsed_decode}ms, "
+        f"xtgeo={elapsed_xtgeo}ms) ",
+        extra={"init": elapsed_init, "get": elapsed_get, "decode": elapsed_decode, "xtgeo": elapsed_xtgeo},
+    )
+
     return surfaces

backend/src/config.py

@@ -23,7 +23,6 @@
     "smda": [SMDA_RESOURCE_SCOPE],
 }
 
-print(f"{RESOURCE_SCOPES_DICT=}")
 
 REDIS_USER_SESSION_URL = "redis://redis-user-session:6379"
 REDIS_CACHE_URL = "redis://redis-cache:6379"