Timeout fix and remove logging (#317)

Author: 1vn

pkg/workloads/cortex/onnx_serve/api.py

@@ -66,10 +66,8 @@
 }
 
 
-def prediction_failed(sample, reason=None):
-    message = "prediction failed for sample: {}".format(util.pp_str_flat(sample))
-    if reason:
-        message += " ({})".format(reason)
+def prediction_failed(reason):
+    message = "prediction failed: " + reason
 
     logger.error(message)
     return message, status.HTTP_406_NOT_ACCEPTABLE
@@ -113,7 +111,7 @@ def convert_to_onnx_input(sample, input_metadata_list):
             try:
                 input_dict[input_metadata.name] = transform_to_numpy(sample, input_metadata)
             except CortexException as e:
-                e.wrap("key {}".format(input_metadata.name))
+                e.wrap('key "{}"'.format(input_metadata.name))
                 raise
     else:
         for input_metadata in input_metadata_list:
@@ -130,7 +128,7 @@ def convert_to_onnx_input(sample, input_metadata_list):
             try:
                 input_dict[input_metadata.name] = transform_to_numpy(sample, input_metadata)
             except CortexException as e:
-                e.wrap("key {}".format(input_metadata.name))
+                e.wrap('key "{}"'.format(input_metadata.name))
                 raise
     return input_dict
 
@@ -151,24 +149,18 @@ def predict(app_name, api_name):
     response = {}
 
     if not util.is_dict(payload) or "samples" not in payload:
-        util.log_pretty_flat(payload, logging_func=logger.error)
-        return prediction_failed(payload, "top level `samples` key not found in request")
+        return prediction_failed('top level "samples" key not found in request')
 
     predictions = []
     samples = payload["samples"]
     if not util.is_list(samples):
-        util.log_pretty_flat(samples, logging_func=logger.error)
-        return prediction_failed(
-            payload, "expected the value of key `samples` to be a list of json objects"
-        )
+        return prediction_failed('expected the value of key "samples" to be a list of json objects')
 
     for i, sample in enumerate(payload["samples"]):
         try:
-            logger.info("sample: " + util.pp_str_flat(sample))
             prepared_sample = sample
             if request_handler is not None and util.has_function(request_handler, "pre_inference"):
                 prepared_sample = request_handler.pre_inference(sample, input_metadata)
-                logger.info("pre_inference: " + util.pp_str_flat(prepared_sample))
 
             inference_input = convert_to_onnx_input(prepared_sample, input_metadata)
             model_outputs = sess.run([], inference_input)
@@ -179,10 +171,8 @@ def predict(app_name, api_name):
                 else:
                     result.append(model_output)
 
-            logger.info("inference: " + util.pp_str_flat(result))
             if request_handler is not None and util.has_function(request_handler, "post_inference"):
                 result = request_handler.post_inference(result, output_metadata)
-                logger.info("post_inference: " + util.pp_str_flat(result))
 
             prediction = {"prediction": result}
         except CortexException as e:
@@ -191,12 +181,12 @@ def predict(app_name, api_name):
             logger.exception(
                 "An error occurred, see `cx logs -v api {}` for more details.".format(api["name"])
             )
-            return prediction_failed(sample, str(e))
+            return prediction_failed(str(e))
         except Exception as e:
             logger.exception(
                 "An error occurred, see `cx logs -v api {}` for more details.".format(api["name"])
             )
-            return prediction_failed(sample, str(e))
+            return prediction_failed(str(e))
 
         predictions.append(prediction)
 

pkg/workloads/cortex/tf_api/api.py

@@ -136,9 +136,13 @@ def create_prediction_request(transformed_sample):
         shape = [1]
         if util.is_list(value):
             shape = [len(value)]
-        tensor_proto = tf.make_tensor_proto([value], dtype=data_type, shape=shape)
-        prediction_request.inputs[column_name].CopyFrom(tensor_proto)
-
+        try:
+            tensor_proto = tf.make_tensor_proto([value], dtype=data_type, shape=shape)
+            prediction_request.inputs[column_name].CopyFrom(tensor_proto)
+        except Exception as e:
+            raise UserException(
+                'key "{}"'.format(column_name), "expected shape {}".format(shape)
+            ) from e
     return prediction_request
 
 
@@ -160,8 +164,15 @@ def create_raw_prediction_request(sample):
             shape = [1]
             value = [value]
         sig_type = signature_def[signature_key]["inputs"][column_name]["dtype"]
-        tensor_proto = tf.make_tensor_proto(value, dtype=DTYPE_TO_TF_TYPE[sig_type], shape=shape)
-        prediction_request.inputs[column_name].CopyFrom(tensor_proto)
+        try:
+            tensor_proto = tf.make_tensor_proto(
+                value, dtype=DTYPE_TO_TF_TYPE[sig_type], shape=shape
+            )
+            prediction_request.inputs[column_name].CopyFrom(tensor_proto)
+        except Exception as e:
+            raise UserException(
+                'key "{}"'.format(column_name), "expected shape {}".format(shape)
+            ) from e
 
     return prediction_request
 
@@ -248,7 +259,7 @@ def create_get_model_metadata_request():
 
 def run_get_model_metadata():
     request = create_get_model_metadata_request()
-    resp = local_cache["stub"].GetModelMetadata(request, timeout=10.0)
+    resp = local_cache["stub"].GetModelMetadata(request, timeout=30.0)
     sigAny = resp.metadata["signature_def"]
     signature_def_map = get_model_metadata_pb2.SignatureDefMap()
     sigAny.Unpack(signature_def_map)
@@ -272,14 +283,11 @@ def run_predict(sample):
     ctx = local_cache["ctx"]
     request_handler = local_cache.get("request_handler")
 
-    logger.info("sample: " + util.pp_str_flat(sample))
-
     prepared_sample = sample
     if request_handler is not None and util.has_function(request_handler, "pre_inference"):
         prepared_sample = request_handler.pre_inference(
             sample, local_cache["metadata"]["signatureDef"]
         )
-        logger.info("pre_inference: " + util.pp_str_flat(prepared_sample))
 
     validate_sample(prepared_sample)
 
@@ -291,24 +299,18 @@ def run_predict(sample):
             )
 
         transformed_sample = transform_sample(prepared_sample)
-        logger.info("transformed_sample: " + util.pp_str_flat(transformed_sample))
-
         prediction_request = create_prediction_request(transformed_sample)
-        response_proto = local_cache["stub"].Predict(prediction_request, timeout=10.0)
+        response_proto = local_cache["stub"].Predict(prediction_request, timeout=100.0)
         result = parse_response_proto(response_proto)
 
         result["transformed_sample"] = transformed_sample
-        logger.info("inference: " + util.pp_str_flat(result))
     else:
         prediction_request = create_raw_prediction_request(prepared_sample)
-        response_proto = local_cache["stub"].Predict(prediction_request, timeout=10.0)
+        response_proto = local_cache["stub"].Predict(prediction_request, timeout=100.0)
         result = parse_response_proto_raw(response_proto)
 
-        logger.info("inference: " + util.pp_str_flat(result))
-
     if request_handler is not None and util.has_function(request_handler, "post_inference"):
         result = request_handler.post_inference(result, local_cache["metadata"]["signatureDef"])
-        logger.info("post_inference: " + util.pp_str_flat(result))
 
     return result
 
@@ -335,10 +337,8 @@ def validate_sample(sample):
                 raise UserException('missing key "{}"'.format(input_name))
 
 
-def prediction_failed(sample, reason=None):
-    message = "prediction failed for sample: {}".format(util.pp_str_flat(sample))
-    if reason:
-        message += " ({})".format(reason)
+def prediction_failed(reason):
+    message = "prediction failed: " + reason
 
     logger.error(message)
     return message, status.HTTP_406_NOT_ACCEPTABLE
@@ -363,16 +363,12 @@ def predict(deployment_name, api_name):
     response = {}
 
     if not util.is_dict(payload) or "samples" not in payload:
-        util.log_pretty_flat(payload, logging_func=logger.error)
-        return prediction_failed(payload, "top level `samples` key not found in request")
+        return prediction_failed('top level "samples" key not found in request')
 
     predictions = []
     samples = payload["samples"]
     if not util.is_list(samples):
-        util.log_pretty_flat(samples, logging_func=logger.error)
-        return prediction_failed(
-            payload, "expected the value of key `samples` to be a list of json objects"
-        )
+        return prediction_failed('expected the value of key "samples" to be a list of json objects')
 
     for i, sample in enumerate(payload["samples"]):
         try:
@@ -385,14 +381,14 @@ def predict(deployment_name, api_name):
                     api["name"]
                 )
             )
-            return prediction_failed(sample, str(e))
+            return prediction_failed(str(e))
         except Exception as e:
             logger.exception(
                 "An error occurred, see `cortex logs -v api {}` for more details.".format(
                     api["name"]
                 )
             )
-            return prediction_failed(sample, str(e))
+            return prediction_failed(str(e))
 
         predictions.append(result)