Merge "Timeout support for api and engine"
This commit is contained in:
commit
e99d0272a1
@ -164,6 +164,9 @@ class FunctionsController(rest.RestController):
|
||||
'memory_size': kwargs.get(
|
||||
'memory_size', CONF.resource_limits.default_memory
|
||||
),
|
||||
'timeout': kwargs.get(
|
||||
'timeout', CONF.resource_limits.default_timeout
|
||||
),
|
||||
}
|
||||
|
||||
# Check cpu and memory_size values.
|
||||
|
@ -254,6 +254,11 @@ rlimits_opts = [
|
||||
default=134217728,
|
||||
help='Maximum memory resource(unit: bytes).'
|
||||
),
|
||||
cfg.IntOpt(
|
||||
'default_timeout',
|
||||
default=5,
|
||||
help='Default function execution timeout(unit: seconds)'
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
|
@ -172,7 +172,7 @@ class DefaultEngine(object):
|
||||
data = utils.get_request_data(
|
||||
CONF, function_id, function_version, execution_id,
|
||||
rlimit, input, function.entry, function.trust_id,
|
||||
self.qinling_endpoint
|
||||
self.qinling_endpoint, function.timeout
|
||||
)
|
||||
success, res = utils.url_request(
|
||||
self.session, func_url, body=data
|
||||
|
@ -76,7 +76,7 @@ def url_request(request_session, url, body=None):
|
||||
|
||||
|
||||
def get_request_data(conf, function_id, version, execution_id, rlimit, input,
|
||||
entry, trust_id, qinling_endpoint):
|
||||
entry, trust_id, qinling_endpoint, timeout):
|
||||
"""Prepare the request body should send to the worker."""
|
||||
ctx = context.get_ctx()
|
||||
|
||||
@ -103,6 +103,7 @@ def get_request_data(conf, function_id, version, execution_id, rlimit, input,
|
||||
'entry': entry,
|
||||
'download_url': download_url,
|
||||
'request_id': ctx.request_id,
|
||||
'timeout': timeout,
|
||||
}
|
||||
if conf.pecan.auth_enable:
|
||||
data.update(
|
||||
|
@ -484,7 +484,7 @@ class KubernetesManager(base.OrchestratorBase):
|
||||
|
||||
def run_execution(self, execution_id, function_id, version, rlimit=None,
|
||||
input=None, identifier=None, service_url=None,
|
||||
entry='main.main', trust_id=None):
|
||||
entry='main.main', trust_id=None, timeout=None):
|
||||
"""Run execution.
|
||||
|
||||
Return a tuple including the result and the output.
|
||||
@ -493,7 +493,7 @@ class KubernetesManager(base.OrchestratorBase):
|
||||
func_url = '%s/execute' % service_url
|
||||
data = utils.get_request_data(
|
||||
self.conf, function_id, version, execution_id, rlimit, input,
|
||||
entry, trust_id, self.qinling_endpoint
|
||||
entry, trust_id, self.qinling_endpoint, timeout
|
||||
)
|
||||
LOG.debug(
|
||||
'Invoke function %s(version %s), url: %s, data: %s',
|
||||
|
@ -63,6 +63,25 @@ class TestFunctionController(base.APITest):
|
||||
)
|
||||
self._assertDictContainsSubset(resp.json, body)
|
||||
|
||||
@mock.patch('qinling.storage.file_system.FileSystemStorage.store')
|
||||
def test_post_timeout(self, mock_store):
|
||||
mock_store.return_value = (True, 'fake_md5')
|
||||
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
body = {
|
||||
'runtime_id': self.runtime_id,
|
||||
'code': json.dumps({"source": "package"}),
|
||||
'timeout': 3
|
||||
}
|
||||
resp = self.app.post(
|
||||
'/v1/functions',
|
||||
params=body,
|
||||
upload_files=[('package', f.name, f.read())]
|
||||
)
|
||||
|
||||
self.assertEqual(201, resp.status_int)
|
||||
self.assertEqual(3, resp.json['timeout'])
|
||||
|
||||
@mock.patch("qinling.utils.openstack.keystone.create_trust")
|
||||
@mock.patch('qinling.utils.openstack.keystone.get_swiftclient')
|
||||
@mock.patch('qinling.context.AuthHook.before')
|
||||
@ -160,6 +179,7 @@ class TestFunctionController(base.APITest):
|
||||
"project_id": unit_base.DEFAULT_PROJECT_ID,
|
||||
"cpu": cfg.CONF.resource_limits.default_cpu,
|
||||
"memory_size": cfg.CONF.resource_limits.default_memory,
|
||||
"timeout": cfg.CONF.resource_limits.default_timeout,
|
||||
}
|
||||
|
||||
resp = self.app.get('/v1/functions/%s' % db_func.id)
|
||||
@ -176,6 +196,7 @@ class TestFunctionController(base.APITest):
|
||||
"project_id": unit_base.DEFAULT_PROJECT_ID,
|
||||
"cpu": cfg.CONF.resource_limits.default_cpu,
|
||||
"memory_size": cfg.CONF.resource_limits.default_memory,
|
||||
"timeout": cfg.CONF.resource_limits.default_timeout,
|
||||
}
|
||||
|
||||
resp = self.app.get('/v1/functions')
|
||||
|
@ -197,6 +197,7 @@ class DbTestCase(BaseTest):
|
||||
'project_id': DEFAULT_PROJECT_ID,
|
||||
'cpu': cfg.CONF.resource_limits.default_cpu,
|
||||
'memory_size': cfg.CONF.resource_limits.default_memory,
|
||||
'timeout': cfg.CONF.resource_limits.default_timeout,
|
||||
}
|
||||
)
|
||||
|
||||
|
@ -447,7 +447,7 @@ class TestDefaultEngine(base.DbTestCase):
|
||||
engine_utils_get_request_data_mock.assert_called_once_with(
|
||||
mock.ANY, function_id, 0, execution_id, self.rlimit,
|
||||
'input', function.entry, function.trust_id,
|
||||
self.qinling_endpoint)
|
||||
self.qinling_endpoint, function.timeout)
|
||||
engine_utils_url_request_mock.assert_called_once_with(
|
||||
self.default_engine.session, 'svc_url/execute', body='data')
|
||||
|
||||
|
@ -695,10 +695,11 @@ class TestKubernetesManager(base.DbTestCase):
|
||||
mock_request.return_value = (True, 'fake output')
|
||||
execution_id = common.generate_unicode_uuid()
|
||||
function_id = common.generate_unicode_uuid()
|
||||
timeout = 3
|
||||
|
||||
result, output = self.manager.run_execution(
|
||||
execution_id, function_id, 0, rlimit=self.rlimit,
|
||||
service_url='FAKE_URL'
|
||||
service_url='FAKE_URL', timeout=timeout
|
||||
)
|
||||
|
||||
download_url = ('http://127.0.0.1:7070/v1/functions/%s?download=true'
|
||||
@ -713,6 +714,7 @@ class TestKubernetesManager(base.DbTestCase):
|
||||
'entry': 'main.main',
|
||||
'download_url': download_url,
|
||||
'request_id': self.ctx.request_id,
|
||||
'timeout': timeout,
|
||||
}
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
|
@ -56,7 +56,7 @@ class QinlingClient(client_base.QinlingClientBase):
|
||||
return resp, body
|
||||
|
||||
def create_function(self, code, runtime_id, name='', package_data=None,
|
||||
entry=None):
|
||||
entry=None, timeout=None):
|
||||
"""Create function.
|
||||
|
||||
Tempest rest client doesn't support multipart upload, so use requests
|
||||
@ -67,7 +67,8 @@ class QinlingClient(client_base.QinlingClientBase):
|
||||
req_body = {
|
||||
'name': name,
|
||||
'runtime_id': runtime_id,
|
||||
'code': json.dumps(code)
|
||||
'code': json.dumps(code),
|
||||
'timeout': timeout
|
||||
}
|
||||
if entry:
|
||||
req_body['entry'] = entry
|
||||
|
@ -118,7 +118,8 @@ class BaseQinlingTest(test.BaseTestCase):
|
||||
self.addCleanup(os.remove, zip_file)
|
||||
return zip_file
|
||||
|
||||
def create_function(self, package_path=None, image=False, md5sum=None):
|
||||
def create_function(self, package_path=None, image=False,
|
||||
md5sum=None, timeout=None):
|
||||
function_name = data_utils.rand_name(
|
||||
'function',
|
||||
prefix=self.name_prefix
|
||||
@ -139,7 +140,8 @@ class BaseQinlingTest(test.BaseTestCase):
|
||||
self.runtime_id,
|
||||
name=function_name,
|
||||
package_data=package_data,
|
||||
entry='%s.main' % module_name
|
||||
entry='%s.main' % module_name,
|
||||
timeout=timeout
|
||||
)
|
||||
else:
|
||||
resp, body = self.client.create_function(
|
||||
|
@ -0,0 +1,5 @@
|
||||
---
|
||||
features:
|
||||
- Users can now specify a timeout value for their function to prevent the function from running
|
||||
indefinitely. When the specified timeout is reached, Qinling will terminate the function
|
||||
execution.
|
Loading…
Reference in New Issue
Block a user