Osmos (#440)

* assign container rank via registration order

* use temp variable

* re-raise cancel errors

* re-throw cancelled exceptions

* fixed typo

---------

Co-authored-by: Joshua Stillerman <jas@mit.edu>

Author: jreadey

hsds/basenode.py

@@ -236,6 +236,8 @@ async def docker_update_dn_info(app):
         log.error("HEAD node seems to be down.")
         app["dn_urls"] = []
         app["dn_ids"] = []
+    except HTTPServiceUnavailable:
+        log.warn("Head ServiceUnavailable")
     except OSError:
         log.error("failed to register")
         app["dn_urls"] = []

hsds/datanode_lib.py

@@ -149,8 +149,7 @@ async def write_s3_obj(app, obj_id, bucket=None):
             bucket = domain_bucket
 
     if obj_id in pending_s3_write:
-        msg = f"write_s3_key - not expected for key {obj_id} to be in "
-        msg += "pending_s3_write map"
+        msg = f"write_s3_key - not expected for key {obj_id} to be in pending_s3_write map"
         log.error(msg)
         raise KeyError(msg)
 
@@ -172,12 +171,10 @@ async def write_s3_obj(app, obj_id, bucket=None):
         # timestamp is first element of two-tuple
         last_update_time = dirty_ids[obj_id][0]
     else:
-        msg = f"write_s3_obj - {obj_id} not in dirty_ids, "
-        msg += "assuming flush write"
+        msg = f"write_s3_obj - {obj_id} not in dirty_ids, assuming flush write"
         log.debug(msg)
     if last_update_time > now:
-        msg = f"last_update time {last_update_time} is in the future for "
-        msg += f"obj_id: {obj_id}"
+        msg = f"last_update time {last_update_time} is in the future for obj_id: {obj_id}"
         log.error(msg)
         raise ValueError(msg)
 
@@ -198,8 +195,7 @@ async def write_s3_obj(app, obj_id, bucket=None):
             dset_id = getDatasetId(obj_id)
             if dset_id in filter_map:
                 filter_ops = filter_map[dset_id]
-                msg = f"write_s3_obj: got filter_op: {filter_ops} "
-                msg += f"for dset: {dset_id}"
+                msg = f"write_s3_obj: got filter_op: {filter_ops} for dset: {dset_id}"
                 log.debug(msg)
             else:
                 filter_ops = None
@@ -237,13 +233,11 @@ async def write_s3_obj(app, obj_id, bucket=None):
             # meta data update
             # check for object in meta cache
             if obj_id not in meta_cache:
-                msg = f"write_s3_obj: expected to find obj_id: {obj_id} "
-                msg += "in meta cache"
+                msg = f"write_s3_obj: expected to find obj_id: {obj_id} in meta cache"
                 log.error(msg)
                 raise KeyError(f"{obj_id} not found in meta cache")
             if not meta_cache.isDirty(obj_id):
-                msg = f"write_s3_obj: expected meta cache obj {obj_id} "
-                msg == "to be dirty"
+                msg = f"write_s3_obj: expected meta cache obj {obj_id} to be dirty"
                 log.error(msg)
                 raise ValueError("bad dirty state for obj")
             obj_json = meta_cache[obj_id]
@@ -264,8 +258,7 @@ async def write_s3_obj(app, obj_id, bucket=None):
                 else:
                     timestamp = 0
                 if timestamp > last_update_time:
-                    msg = f"write_s3_obj: {obj_id} got updated while s3 "
-                    msg += "write was in progress"
+                    msg = f"write_s3_obj: {obj_id} got updated while s3 write was in progress"
                     log.info(msg)
                 else:
                     log.debug(f"write_s3obj: clear dirty for {obj_id} ")
@@ -279,11 +272,10 @@ async def write_s3_obj(app, obj_id, bucket=None):
 
     finally:
         # clear pending_s3_write item
-        log.debug(f"write_s3_obj finally block, success={success}")
+        log.debug(f"write_s3_obj {obj_id} finally block, success={success}")
         if obj_id in pending_s3_write:
             if pending_s3_write[obj_id] != now:
-                msg = "pending_s3_write timestamp got updated unexpectedly "
-                msg += f"for {obj_id}"
+                msg = f"pending_s3_write timestamp got updated unexpectedly for {obj_id}"
                 log.error(msg)
             del pending_s3_write[obj_id]
         # clear task
@@ -1259,10 +1251,9 @@ def callback(future):
 
         if obj_id in pending_s3_write:
             pending_time = s3sync_start - pending_s3_write[obj_id]
-            msg = f"s3sync - key {obj_id} has been pending for "
-            msg += f"{pending_time:.3f}"
+            msg = f"s3sync - key {obj_id} has been pending for {pending_time:.3f}"
             log.debug(msg)
-            if s3sync_start - pending_s3_write[obj_id] > s3_sync_task_timeout:
+            if pending_time > s3_sync_task_timeout:
                 msg = f"s3sync - obj {obj_id} has been in pending_s3_write "
                 msg += f"for {pending_time:.3f} seconds, restarting"
                 log.warn(msg)

hsds/util/fileClient.py

@@ -166,8 +166,8 @@ async def get_object(self, key, bucket=None, offset=0, length=-1):
         except CancelledError as cle:
             self._file_stats_increment("error_count")
             msg = f"CancelledError for get file obj {key}: {cle}"
-            log.error(msg)
-            raise HTTPInternalServerError()
+            log.warn(msg)
+            raise
         except Exception as e:
             self._file_stats_increment("error_count")
             msg = f"Unexpected Exception {type(e)} get get_object {key}: {e}"
@@ -227,8 +227,8 @@ async def put_object(self, key, data, bucket=None):
         except CancelledError as cle:
             # file_stats_increment(app, "error_count")
             msg = f"CancelledError for put file obj {key}: {cle}"
-            log.error(msg)
-            raise HTTPInternalServerError()
+            log.warn(msg)
+            raise
 
         except Exception as e:
             # file_stats_increment(app, "error_count")
@@ -274,8 +274,8 @@ async def delete_object(self, key, bucket=None):
         except CancelledError as cle:
             self._file_stats_increment("error_count")
             msg = f"CancelledError deleting file obj {key}: {cle}"
-            log.error(msg)
-            raise HTTPInternalServerError()
+            log.warn(msg)
+            raise
 
         except Exception as e:
             self._file_stats_increment("error_count")

hsds/util/idUtil.py

@@ -536,5 +536,9 @@ def getDataNodeUrl(app, obj_id):
         raise HTTPServiceUnavailable()
     dn_number = getObjPartition(obj_id, dn_node_count)
     url = dn_urls[dn_number]
+    if not url:
+        msg = "Service not ready (no DN url set)"
+        log.warn(msg)
+        raise HTTPServiceUnavailable()
     log.debug(f"got dn_url: {url} for obj_id: {obj_id}")
     return url