Merge "Limit resource consumption of function"

qinling_tempest_plugin/functions/test_python_file_limit.py

@@ -0,0 +1,38 @@
+# Copyright 2017 Catalyst IT Ltd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+import resource
+
+
+def main(number=1024, **kwargs):
+    for name, desc in [
+        ('RLIMIT_NOFILE', 'number of open files'),
+    ]:
+        limit_num = getattr(resource, name)
+        soft, hard = resource.getrlimit(limit_num)
+        print('Maximum %-25s (%-15s) : %20s %20s' % (desc, name, soft, hard))
+
+    files = []
+
+    try:
+        for i in range(0, number):
+            files.append(_create_file(i))
+    finally:
+        for f in files:
+            f.close()
+
+
+def _create_file(index):
+    f = open('file_%s' % index, 'w')
+    return f

qinling_tempest_plugin/functions/test_python_process_limit.py

@@ -0,0 +1,43 @@
+# Copyright 2017 Catalyst IT Ltd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+from multiprocessing import Process
+import resource
+import time
+
+
+def main(number=128, **kwargs):
+    for name, desc in [
+        ('RLIMIT_NPROC', 'number of processes'),
+    ]:
+        limit_num = getattr(resource, name)
+        soft, hard = resource.getrlimit(limit_num)
+        print('Maximum %-25s (%-15s) : %20s %20s' % (desc, name, soft, hard))
+
+    processes = []
+
+    for i in range(0, number):
+        p = Process(
+            target=_sleep,
+            args=(i,)
+        )
+        p.start()
+        processes.append(p)
+
+    for p in processes:
+        p.join()
+
+
+def _sleep(index):
+    time.sleep(10)

qinling_tempest_plugin/tests/api/test_executions.py

@@ -12,9 +12,11 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 import os
+import pkg_resources
 import tempfile
 import zipfile
 
+from oslo_serialization import jsonutils
 from tempest.lib.common.utils import data_utils
 from tempest.lib import decorators
 
@@ -48,19 +50,10 @@ class ExecutionsTest(base.BaseQinlingTest):
 
         super(ExecutionsTest, cls).resource_cleanup()
 
-    def setUp(self):
-        super(ExecutionsTest, self).setUp()
-
-        # Wait until runtime is available
-        self.await_runtime_available(self.runtime_id)
-
-        python_file_path = os.path.abspath(
-            os.path.join(
-                os.path.dirname(__file__),
-                os.pardir,
-                os.pardir,
-                'functions/python_test.py'
-            )
+    def _create_function(self, name='python_test.py'):
+        python_file_path = pkg_resources.resource_filename(
+            'qinling_tempest_plugin',
+            "functions/%s" % name
         )
         base_name, extention = os.path.splitext(python_file_path)
         self.base_name = os.path.basename(base_name)
@@ -96,9 +89,13 @@ class ExecutionsTest(base.BaseQinlingTest):
 
         self.addCleanup(self.client.delete_resource, 'functions',
                         self.function_id, ignore_notfound=True)
+        self.addCleanup(os.remove, self.python_zip_file)
 
     @decorators.idempotent_id('2a93fab0-2dae-4748-b0d4-f06b735ff451')
     def test_crud_execution(self):
+        self.await_runtime_available(self.runtime_id)
+        self._create_function()
+
         resp, body = self.client.create_execution(self.function_id,
                                                   input={'name': 'Qinling'})
 
@@ -125,6 +122,9 @@ class ExecutionsTest(base.BaseQinlingTest):
 
     @decorators.idempotent_id('8096cc52-64d2-4660-a657-9ac0bdd743ae')
     def test_execution_async(self):
+        self.await_runtime_available(self.runtime_id)
+        self._create_function()
+
         resp, body = self.client.create_execution(self.function_id, sync=False)
 
         self.assertEqual(201, resp.status)
@@ -138,6 +138,9 @@ class ExecutionsTest(base.BaseQinlingTest):
 
     @decorators.idempotent_id('6cb47b1d-a8c6-48f2-a92f-c4f613c33d1c')
     def test_execution_log(self):
+        self.await_runtime_available(self.runtime_id)
+        self._create_function()
+
         resp, body = self.client.create_execution(self.function_id,
                                                   input={'name': 'OpenStack'})
 
@@ -153,3 +156,31 @@ class ExecutionsTest(base.BaseQinlingTest):
 
         self.assertEqual(200, resp.status)
         self.assertIn('Hello, OpenStack', body)
+
+    def test_python_execution_file_limit(self):
+        self.await_runtime_available(self.runtime_id)
+        self._create_function(name='test_python_file_limit.py')
+
+        resp, body = self.client.create_execution(self.function_id)
+
+        self.assertEqual(201, resp.status)
+        self.assertEqual('failed', body['status'])
+
+        output = jsonutils.loads(body['output'])
+        self.assertIn(
+            'Too many open files', output['output']
+        )
+
+    def test_python_execution_process_number(self):
+        self.await_runtime_available(self.runtime_id)
+        self._create_function(name='test_python_process_limit.py')
+
+        resp, body = self.client.create_execution(self.function_id)
+
+        self.assertEqual(201, resp.status)
+        self.assertEqual('failed', body['status'])
+
+        output = jsonutils.loads(body['output'])
+        self.assertIn(
+            'too much resource consumption', output['output']
+        )

runtimes/python2/Dockerfile

@@ -1,13 +1,17 @@
 FROM phusion/baseimage:0.9.22
-MAINTAINER lingxian.kong@gmail.com
+MAINTAINER anlin.kong@gmail.com
 
-RUN apt-get update
-RUN apt-get -y install python-dev python-setuptools libffi-dev libxslt1-dev libxml2-dev libyaml-dev libssl-dev python-pip
-RUN pip install -U pip setuptools
+USER root
+RUN useradd -Ms /bin/bash qinling
+
+RUN apt-get update && \
+    apt-get -y install python-dev python-setuptools libffi-dev libxslt1-dev libxml2-dev libyaml-dev libssl-dev python-pip && \
+    pip install -U pip setuptools
 
 COPY . /app
 WORKDIR /app
-RUN pip install -r requirements.txt
+RUN pip install -r requirements.txt && \
+    chmod 0750 custom-entrypoint.sh && \
+    chown -R qinling:qinling /app
 
-ENTRYPOINT ["python", "-u"]
-CMD ["server.py"]
+CMD ["/bin/bash", "custom-entrypoint.sh"]

runtimes/python2/custom-entrypoint.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+# This is expected to run as root for setting the ulimits
+
+set -e
+
+# ensure increased ulimits - for nofile - for the runtime containers
+# the limit on the number of files that a single process can have open at a time
+ulimit -n 1024
+
+# ensure increased ulimits - for nproc - for the runtime containers
+# the limit on the number of processes
+ulimit -u 128
+
+# ensure increased ulimits - for file size - for the runtime containers
+# the limit on the total file size that a single process can create, 30M
+ulimit -f 61440
+
+/sbin/setuser qinling python -u server.py

runtimes/python2/requirements.txt

@@ -7,3 +7,4 @@ python-zaqarclient>=1.0.0  # Apache-2.0
 python-octaviaclient>=1.0.0  # Apache-2.0
 python-mistralclient>=3.1.0  # Apache-2.0
 keystoneauth1>=2.21.0  # Apache-2.0
+openstacksdk>=0.9.19

runtimes/python2/server.py

@@ -88,6 +88,12 @@ def download():
     return 'success'
 
 
+def _print_trace():
+    exc_type, exc_value, exc_traceback = sys.exc_info()
+    lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
+    print(''.join(line for line in lines))
+
+
 def _invoke_function(execution_id, zip_file, module_name, method, input,
                      return_dict):
     """Thie function is supposed to be running in a child process."""
@@ -102,13 +108,13 @@ def _invoke_function(execution_id, zip_file, module_name, method, input,
         return_dict['result'] = func(**input)
         return_dict['success'] = True
     except Exception as e:
+        if isinstance(e, OSError) and 'Resource' in str(e):
+            sys.exit(1)
+
         return_dict['result'] = str(e)
         return_dict['success'] = False
 
-        # Print stacktrace
-        exc_type, exc_value, exc_traceback = sys.exc_info()
-        lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
-        print(''.join(line for line in lines))
+        _print_trace()
     finally:
         print('Finished execution: %s' % execution_id)
 
@@ -119,8 +125,10 @@ def execute():
 
     Several things need to handle in this function:
     - Save the function log
-    - Capture the function exception
-    - Deal with process execution error
+    - Capture the function internal exception
+    - Deal with process execution error (The process may be killed for some
+      reason, e.g. unlimited memory allocation)
+    - Deal with os error for process (e.g. Resource temporarily unavailable)
     """
 
     global zip_file
@@ -170,12 +178,15 @@ def execute():
     p.join()
 
     duration = round(time.time() - start, 3)
-    output = return_dict.get('result')
 
-    # Process was killed unexpectedly.
+    # Process was killed unexpectedly or finished with error.
     if p.exitcode != 0:
         output = "Function execution failed because of too much resource " \
                  "consumption."
+        success = False
+    else:
+        output = return_dict.get('result')
+        success = return_dict['success']
 
     # Execution log
     with open('%s.out' % execution_id) as f:
@@ -188,7 +199,7 @@ def execute():
                 'output': output,
                 'duration': duration,
                 'logs': logs,
-                'success': return_dict['success']
+                'success': success
             }
         ),
         status=200,