Fixed app base timeout problems for stuck functions
This commit is contained in:
+93
-32
@@ -18,6 +18,10 @@ import urllib.parse
|
|||||||
import jinja2
|
import jinja2
|
||||||
import datetime
|
import datetime
|
||||||
import dateutil
|
import dateutil
|
||||||
|
|
||||||
|
import threading
|
||||||
|
import concurrent.futures
|
||||||
|
|
||||||
from io import StringIO as StringBuffer
|
from io import StringIO as StringBuffer
|
||||||
from io import BytesIO
|
from io import BytesIO
|
||||||
from liquid import Liquid, defaults
|
from liquid import Liquid, defaults
|
||||||
@@ -267,6 +271,7 @@ def split(base, sep):
|
|||||||
###
|
###
|
||||||
###
|
###
|
||||||
|
|
||||||
|
|
||||||
class AppBase:
|
class AppBase:
|
||||||
__version__ = None
|
__version__ = None
|
||||||
app_name = None
|
app_name = None
|
||||||
@@ -396,7 +401,7 @@ class AppBase:
|
|||||||
try:
|
try:
|
||||||
new_input = input_data.split()
|
new_input = input_data.split()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.logger.info(f"[ERROR] Failed to run magic parser during split (1): {e}")
|
self.logger.info(f"[ERROR] Failed to run parser during split (1): {e}")
|
||||||
return input_data
|
return input_data
|
||||||
|
|
||||||
# Won't ever touch this one?
|
# Won't ever touch this one?
|
||||||
@@ -452,9 +457,11 @@ class AppBase:
|
|||||||
else:
|
else:
|
||||||
self.logger.warning(f"[WARNING] Magic output not defined.")
|
self.logger.warning(f"[WARNING] Magic output not defined.")
|
||||||
except KeyError as e:
|
except KeyError as e:
|
||||||
self.logger.warning(f"[DEBUG] Failed to run magic autoparser (send result) - keyerror: {e}")
|
#self.logger.warning(f"[DEBUG] Failed to run magic autoparser (send result) - keyerror: {e}")
|
||||||
|
pass
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.logger.warning(f"[DEBUG] Failed to run magic autoparser (send result): {e}")
|
#self.logger.warning(f"[DEBUG] Failed to run magic autoparser (send result): {e}")
|
||||||
|
pass
|
||||||
|
|
||||||
# Try it with some magic
|
# Try it with some magic
|
||||||
|
|
||||||
@@ -471,7 +478,7 @@ class AppBase:
|
|||||||
self.logger.info(f"[INFO] URL FOR RESULT (URL): {url}")
|
self.logger.info(f"[INFO] URL FOR RESULT (URL): {url}")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
log_contents = "disabled: add env SHUFFLE_LOGS_DISABLED=true to Orborus to re-enable logs for apps"
|
log_contents = "disabled: add env SHUFFLE_LOGS_DISABLED=true to Orborus to re-enable logs for apps. Can not be enabled natively in Cloud except in Hybrid mode."
|
||||||
if not os.getenv("SHUFFLE_LOGS_DISABLED") == "true":
|
if not os.getenv("SHUFFLE_LOGS_DISABLED") == "true":
|
||||||
log_contents = self.log_capture_string.getvalue()
|
log_contents = self.log_capture_string.getvalue()
|
||||||
|
|
||||||
@@ -1025,11 +1032,9 @@ class AppBase:
|
|||||||
for subparams in param_multiplier:
|
for subparams in param_multiplier:
|
||||||
#self.logger.info(f"SUBPARAMS IN MULTI: {subparams}")
|
#self.logger.info(f"SUBPARAMS IN MULTI: {subparams}")
|
||||||
try:
|
try:
|
||||||
#tmp = await func(**subparams)
|
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
#tmp = await func(**subparams)
|
|
||||||
tmp = func(**subparams)
|
tmp = func(**subparams)
|
||||||
break
|
break
|
||||||
except TypeError as e:
|
except TypeError as e:
|
||||||
@@ -1042,7 +1047,7 @@ class AppBase:
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
del subparams[field]
|
del subparams[field]
|
||||||
self.logger.info("Removed field invalid field %s" % field)
|
self.logger.info("Removed invalid field %s (1)" % field)
|
||||||
except KeyError:
|
except KeyError:
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
@@ -1488,7 +1493,7 @@ class AppBase:
|
|||||||
# If found, we get the full results list from backend
|
# If found, we get the full results list from backend
|
||||||
fullexecution = {}
|
fullexecution = {}
|
||||||
if isinstance(self.full_execution, str) and len(self.full_execution) == 0:
|
if isinstance(self.full_execution, str) and len(self.full_execution) == 0:
|
||||||
self.logger.info("[DEBUG] NO EXECUTION - LOADING!")
|
#self.logger.info("[DEBUG] NO EXECUTION - LOADING!")
|
||||||
try:
|
try:
|
||||||
failed = False
|
failed = False
|
||||||
rettext = ""
|
rettext = ""
|
||||||
@@ -1525,6 +1530,7 @@ class AppBase:
|
|||||||
|
|
||||||
rettext = ret.text
|
rettext = ret.text
|
||||||
failed = True
|
failed = True
|
||||||
|
time.sleep(8)
|
||||||
break
|
break
|
||||||
|
|
||||||
if failed:
|
if failed:
|
||||||
@@ -1832,7 +1838,7 @@ class AppBase:
|
|||||||
else:
|
else:
|
||||||
parse_string = inner_result
|
parse_string = inner_result
|
||||||
|
|
||||||
print("PARSE STRING: %s" % parse_string)
|
#print("PARSE STRING: %s" % parse_string)
|
||||||
return parse_string, True
|
return parse_string, True
|
||||||
|
|
||||||
# Looks for parantheses to grab special cases within a string, e.g:
|
# Looks for parantheses to grab special cases within a string, e.g:
|
||||||
@@ -1971,7 +1977,7 @@ class AppBase:
|
|||||||
if isinstance(seconditem, int):
|
if isinstance(seconditem, int):
|
||||||
seconditem = str(seconditem)
|
seconditem = str(seconditem)
|
||||||
|
|
||||||
print("[DEBUG] ACTUAL PARSED: %s" % actualitem)
|
#print("[DEBUG] ACTUAL PARSED: %s" % actualitem)
|
||||||
|
|
||||||
# Means it's a single item -> continue
|
# Means it's a single item -> continue
|
||||||
if seconditem == "":
|
if seconditem == "":
|
||||||
@@ -2093,13 +2099,13 @@ class AppBase:
|
|||||||
actionname_lower = parsersplit[0][1:].lower()
|
actionname_lower = parsersplit[0][1:].lower()
|
||||||
|
|
||||||
#Actionname: Start_node
|
#Actionname: Start_node
|
||||||
print(f"\n[INFO] Actionname: {actionname_lower}")
|
#print(f"\n[INFO] Actionname: {actionname_lower}")
|
||||||
|
|
||||||
# 1. Find the action
|
# 1. Find the action
|
||||||
baseresult = ""
|
baseresult = ""
|
||||||
|
|
||||||
appendresult = ""
|
appendresult = ""
|
||||||
print("[INFO] Parsersplit length: %d" % len(parsersplit))
|
#print("[INFO] Parsersplit length: %d" % len(parsersplit))
|
||||||
if (actionname_lower.startswith("exec ") or actionname_lower.startswith("webhook ") or actionname_lower.startswith("schedule ") or actionname_lower.startswith("userinput ") or actionname_lower.startswith("email_trigger ") or actionname_lower.startswith("trigger ")) and len(parsersplit) == 1:
|
if (actionname_lower.startswith("exec ") or actionname_lower.startswith("webhook ") or actionname_lower.startswith("schedule ") or actionname_lower.startswith("userinput ") or actionname_lower.startswith("email_trigger ") or actionname_lower.startswith("trigger ")) and len(parsersplit) == 1:
|
||||||
record = False
|
record = False
|
||||||
for char in actionname_lower:
|
for char in actionname_lower:
|
||||||
@@ -2142,7 +2148,6 @@ class AppBase:
|
|||||||
print("[DEBUG] No results to get values from.")
|
print("[DEBUG] No results to get values from.")
|
||||||
baseresult = "$" + parsersplit[0][1:]
|
baseresult = "$" + parsersplit[0][1:]
|
||||||
|
|
||||||
print("[DEBUG] BEFORE VARIABLES!")
|
|
||||||
if len(baseresult) == 0:
|
if len(baseresult) == 0:
|
||||||
try:
|
try:
|
||||||
for variable in execution_data["workflow"]["workflow_variables"]:
|
for variable in execution_data["workflow"]["workflow_variables"]:
|
||||||
@@ -2153,13 +2158,12 @@ class AppBase:
|
|||||||
break
|
break
|
||||||
|
|
||||||
except KeyError as e:
|
except KeyError as e:
|
||||||
print("[INFO] KeyError wf variables: %s" % e)
|
#print("[INFO] KeyError wf variables: %s" % e)
|
||||||
pass
|
pass
|
||||||
except TypeError as e:
|
except TypeError as e:
|
||||||
print("[INFO] TypeError wf variables: %s" % e)
|
#print("[INFO] TypeError wf variables: %s" % e)
|
||||||
pass
|
pass
|
||||||
|
|
||||||
print("[DEBUG] BEFORE EXECUTION VAR")
|
|
||||||
if len(baseresult) == 0:
|
if len(baseresult) == 0:
|
||||||
try:
|
try:
|
||||||
for variable in execution_data["execution_variables"]:
|
for variable in execution_data["execution_variables"]:
|
||||||
@@ -2177,14 +2181,14 @@ class AppBase:
|
|||||||
except KeyError as error:
|
except KeyError as error:
|
||||||
print(f"[DEBUG] KeyError in JSON: {error}")
|
print(f"[DEBUG] KeyError in JSON: {error}")
|
||||||
|
|
||||||
print(f"[INFO] After first trycatch. Baseresult")#, baseresult)
|
#print(f"[INFO] After first trycatch. Baseresult")#, baseresult)
|
||||||
|
|
||||||
# 2. Find the JSON data
|
# 2. Find the JSON data
|
||||||
# Returns if there isn't any JSON in the base ($nodename)
|
# Returns if there isn't any JSON in the base ($nodename)
|
||||||
if len(baseresult) == 0:
|
if len(baseresult) == 0:
|
||||||
return ""+appendresult, False
|
return ""+appendresult, False
|
||||||
|
|
||||||
print("[INFO] After second return")
|
#print("[INFO] After second return")
|
||||||
# Returns if the result is JUST something like $nodename, not $nodename.value
|
# Returns if the result is JUST something like $nodename, not $nodename.value
|
||||||
if len(parsersplit) == 1:
|
if len(parsersplit) == 1:
|
||||||
returndata = str(baseresult)+str(appendresult)
|
returndata = str(baseresult)+str(appendresult)
|
||||||
@@ -2195,7 +2199,7 @@ class AppBase:
|
|||||||
baseresult = baseresult.replace(" False", " false,")
|
baseresult = baseresult.replace(" False", " false,")
|
||||||
|
|
||||||
# Tries to actually read it as JSON with some stupid formatting
|
# Tries to actually read it as JSON with some stupid formatting
|
||||||
print("[INFO] After third parser return - Formatted")#, baseresult)
|
#print("[INFO] After third parser return - Formatted")#, baseresult)
|
||||||
basejson = {}
|
basejson = {}
|
||||||
try:
|
try:
|
||||||
basejson = json.loads(baseresult)
|
basejson = json.loads(baseresult)
|
||||||
@@ -2219,7 +2223,6 @@ class AppBase:
|
|||||||
print("[WARNING] Parseditem issue: %s" % e)
|
print("[WARNING] Parseditem issue: %s" % e)
|
||||||
pass
|
pass
|
||||||
|
|
||||||
print("[DEBUG] DATA: (%s) %s" % (type(data), data))
|
|
||||||
if is_loop:
|
if is_loop:
|
||||||
print("[DEBUG] DATA IS A LOOP - SHOULD WRAP")
|
print("[DEBUG] DATA IS A LOOP - SHOULD WRAP")
|
||||||
if parsersplit[-1] == "#":
|
if parsersplit[-1] == "#":
|
||||||
@@ -2231,7 +2234,6 @@ class AppBase:
|
|||||||
parseditem = "${%s%s}$" % (parsersplit[-1], json.dumps(data))
|
parseditem = "${%s%s}$" % (parsersplit[-1], json.dumps(data))
|
||||||
|
|
||||||
|
|
||||||
print("[DEBUG] Before last return with %s" % appendresult)
|
|
||||||
returndata = str(parseditem)+str(appendresult)
|
returndata = str(parseditem)+str(appendresult)
|
||||||
|
|
||||||
# New in 0.8.97: Don't return items without lists
|
# New in 0.8.97: Don't return items without lists
|
||||||
@@ -2558,7 +2560,7 @@ class AppBase:
|
|||||||
# Basic fix in case variant isn't set
|
# Basic fix in case variant isn't set
|
||||||
# Variant is ALWAYS STATIC_VALUE from mid 2021~
|
# Variant is ALWAYS STATIC_VALUE from mid 2021~
|
||||||
try:
|
try:
|
||||||
self.logger.info(f"[DEBUG] Parameter '{paramname}' of length {len(parameter['value'])}")
|
#self.logger.info(f"[DEBUG] Parameter '{paramname}' of length {len(parameter['value'])}")
|
||||||
parameter["variant"] = parameter["variant"]
|
parameter["variant"] = parameter["variant"]
|
||||||
except:
|
except:
|
||||||
parameter["variant"] = "STATIC_VALUE"
|
parameter["variant"] = "STATIC_VALUE"
|
||||||
@@ -2928,7 +2930,7 @@ class AppBase:
|
|||||||
# Checks whether conditions are met, otherwise set
|
# Checks whether conditions are met, otherwise set
|
||||||
branchcheck, tmpresult = check_branch_conditions(action, fullexecution, self)
|
branchcheck, tmpresult = check_branch_conditions(action, fullexecution, self)
|
||||||
if isinstance(tmpresult, object) or isinstance(tmpresult, list) or isinstance(tmpresult, dict):
|
if isinstance(tmpresult, object) or isinstance(tmpresult, list) or isinstance(tmpresult, dict):
|
||||||
self.logger.info("[DEBUG] Fixing branch return as object -> string")
|
#self.logger.info("[DEBUG] Fixing branch return as object -> string")
|
||||||
try:
|
try:
|
||||||
#tmpresult = tmpresult.replace("'", "\"")
|
#tmpresult = tmpresult.replace("'", "\"")
|
||||||
tmpresult = json.dumps(tmpresult)
|
tmpresult = json.dumps(tmpresult)
|
||||||
@@ -3064,7 +3066,6 @@ class AppBase:
|
|||||||
#self.logger.info(action["parameters"])
|
#self.logger.info(action["parameters"])
|
||||||
|
|
||||||
# This seems redundant now
|
# This seems redundant now
|
||||||
self.logger.info("[DEBUG] Pre parameters")
|
|
||||||
for parameter in newparams:
|
for parameter in newparams:
|
||||||
action["parameters"].append(parameter)
|
action["parameters"].append(parameter)
|
||||||
|
|
||||||
@@ -3086,7 +3087,6 @@ class AppBase:
|
|||||||
|
|
||||||
# Multi_parameter has the data for each. variable
|
# Multi_parameter has the data for each. variable
|
||||||
minlength = 0
|
minlength = 0
|
||||||
self.logger.info("[DEBUG] Pre-loading parameters")
|
|
||||||
multi_parameters = json.loads(json.dumps(params))
|
multi_parameters = json.loads(json.dumps(params))
|
||||||
multiexecution = False
|
multiexecution = False
|
||||||
multi_execution_lists = []
|
multi_execution_lists = []
|
||||||
@@ -3344,9 +3344,6 @@ class AppBase:
|
|||||||
# This part has fucked over so many random JSON usages because of weird paranthesis parsing
|
# This part has fucked over so many random JSON usages because of weird paranthesis parsing
|
||||||
|
|
||||||
value = parse_wrapper_start(value, self)
|
value = parse_wrapper_start(value, self)
|
||||||
#self.logger.info("[DEBUG] Post return: %s" % value)
|
|
||||||
|
|
||||||
#self.logger.info("POST data value: %s" % value)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if str(value).startswith("b'") and str(value).endswith("'"):
|
if str(value).startswith("b'") and str(value).endswith("'"):
|
||||||
@@ -3479,8 +3476,73 @@ class AppBase:
|
|||||||
break
|
break
|
||||||
|
|
||||||
try:
|
try:
|
||||||
newres = func(**params)
|
#try:
|
||||||
|
# Individual functions shouldn't take longer than this
|
||||||
|
# This is an attempt to make timeouts occur less, incentivizing users to make use efficient API's
|
||||||
|
# PS: Not implemented for lists - only single actions as of May 2023
|
||||||
|
timeout = 30
|
||||||
|
|
||||||
|
# Check if current app is Shuffle Tools, then set to 55 due to certain actions being slow (ioc parser..)
|
||||||
|
#uu In general, this should be disabled for onprem
|
||||||
|
if self.action["app_name"].lower() == "shuffle tools":
|
||||||
|
timeout = 55
|
||||||
|
|
||||||
|
timeout = 30
|
||||||
|
|
||||||
|
try:
|
||||||
|
executor = concurrent.futures.ThreadPoolExecutor()
|
||||||
|
future = executor.submit(func, **params)
|
||||||
|
newres = future.result(timeout)
|
||||||
|
|
||||||
|
if not future.done():
|
||||||
|
# The future is still running, so we need to cancel it
|
||||||
|
future.cancel()
|
||||||
|
newres = json.dumps({
|
||||||
|
"success": False,
|
||||||
|
"reason": "Timeout error within %d seconds. This happens if we can't reach or use the API you're trying to use within the time limit." % timeout,
|
||||||
|
"exception": str(e),
|
||||||
|
})
|
||||||
|
|
||||||
|
else:
|
||||||
|
# The future is done, so we can just get the result from newres :)
|
||||||
|
#newres = future.result()
|
||||||
|
#print("Future is done!")
|
||||||
|
pass
|
||||||
|
|
||||||
|
except concurrent.futures.TimeoutError as e:
|
||||||
|
newres = json.dumps({
|
||||||
|
"success": False,
|
||||||
|
"reason": "Timeout error within %d seconds (2). This happens if we can't reach or use the API you're trying to use within the time limit" % timeout
|
||||||
|
})
|
||||||
|
|
||||||
break
|
break
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
#thread = threading.Thread(target=func, args=(**params,))
|
||||||
|
#thread.start()
|
||||||
|
|
||||||
|
#thread.join(timeout)
|
||||||
|
|
||||||
|
#if thread.is_alive():
|
||||||
|
# # The thread is still running, so we need to stop it
|
||||||
|
# # You can handle this as needed, such as raising an exception
|
||||||
|
# timeout_handler()
|
||||||
|
|
||||||
|
|
||||||
|
#with Timeout(timeout):
|
||||||
|
# newres = func(**params)
|
||||||
|
# break
|
||||||
|
#except Timeout.Timeout as e:
|
||||||
|
# self.logger.info(f"[DEBUG] Timeout error: {e}")
|
||||||
|
# newres = json.dumps({
|
||||||
|
# "success": False,
|
||||||
|
# "reason": "Timeout error within %d seconds. This typically happens if we can't reach the API you're trying to reach." % timeout,
|
||||||
|
# "exception": str(e),
|
||||||
|
# })
|
||||||
|
|
||||||
|
# break
|
||||||
|
|
||||||
except TypeError as e:
|
except TypeError as e:
|
||||||
newres = ""
|
newres = ""
|
||||||
self.logger.info(f"[DEBUG] Got exec type error: {e}")
|
self.logger.info(f"[DEBUG] Got exec type error: {e}")
|
||||||
@@ -3512,7 +3574,7 @@ class AppBase:
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
del params[field]
|
del params[field]
|
||||||
self.logger.info("[WARNING] Removed field invalid field %s" % field)
|
self.logger.info("[WARNING] Removed invalid field %s (2)" % field)
|
||||||
except KeyError:
|
except KeyError:
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
@@ -3605,7 +3667,7 @@ class AppBase:
|
|||||||
result += "Failed autocasting. Can't handle %s type from function. Must be string" % type(newres)
|
result += "Failed autocasting. Can't handle %s type from function. Must be string" % type(newres)
|
||||||
self.logger.info("Can't handle type %s value from function" % (type(newres)))
|
self.logger.info("Can't handle type %s value from function" % (type(newres)))
|
||||||
|
|
||||||
self.logger.info("[INFO] POST NEWRES RESULT!")#, result)
|
#self.logger.info("[INFO] POST NEWRES RESULT!")#, result)
|
||||||
else:
|
else:
|
||||||
#self.logger.info("[INFO] APP_SDK DONE: Starting MULTI execution (length: %d) with values %s" % (minlength, multi_parameters))
|
#self.logger.info("[INFO] APP_SDK DONE: Starting MULTI execution (length: %d) with values %s" % (minlength, multi_parameters))
|
||||||
# 1. Use number of executions based on the arrays being similar
|
# 1. Use number of executions based on the arrays being similar
|
||||||
@@ -3776,7 +3838,6 @@ class AppBase:
|
|||||||
#logger.info(f"[DEBUG] Datatype: {type(requestdata)}: {requestdata}")
|
#logger.info(f"[DEBUG] Datatype: {type(requestdata)}: {requestdata}")
|
||||||
|
|
||||||
# Remaking class for each request
|
# Remaking class for each request
|
||||||
#print(f"APP: {app}")
|
|
||||||
|
|
||||||
app = cls(redis=None, logger=logger, console_logger=logger)
|
app = cls(redis=None, logger=logger, console_logger=logger)
|
||||||
extra_info = ""
|
extra_info = ""
|
||||||
|
|||||||
Reference in New Issue
Block a user