openstack/qinling
Code Issues Proposed changes

Support to update function code

Browse Source 
Change-Id: If7698350925119140b46cf319ad74f3e063ef0a6
Closes-Bug: #1733477
This commit is contained in:
Lingxian Kong 2017-11-21 17:15:45 +13:00
parent b2d82ee744
commit 9f5b474b6f
7 changed files with 124 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
example/functions/python/openstack/send_zaqar_message.py
View File

@ -1,13 +1,27 @@
# Copyright 2017 Catalyst IT Limited
#
# 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 os
import requests
from zaqarclient.queues import client
def _send_message(z_client, queue_name, status, message=''):
def _send_message(z_client, queue_name, status, server=''):
queue_name = queue_name or 'test_queue'
queue = z_client.queue(queue_name)
queue.post({"body": {'status': status, 'message': message}})
queue.post({"body": {'status': status, 'server': server}})
print 'message posted.'
@ -24,14 +38,14 @@ def check_and_trigger(context, **kwargs):
with open(file_name, 'r+') as f:
count = int(f.readline())
count += 1
if count >= 3:
if count == 3:
# Send message and stop trigger after 3 checks
z_client = client.Client(
session=context['os_session'],
version=2,
)
_send_message(z_client, kwargs.get('queue'), r.status_code,
'Service Not Available!')
count = 0
'api1.production.catalyst.co.nz')
f.seek(0)
f.write(str(count))

96
qinling/api/controllers/v1/function.py
View File

@ -31,6 +31,7 @@ from qinling.db import api as db_api
from qinling import exceptions as exc
from qinling import rpc
from qinling.storage import base as storage_base
from qinling.utils import constants
from qinling.utils.openstack import keystone as keystone_util
from qinling.utils.openstack import swift as swift_util
from qinling.utils import rest_utils
@ -40,7 +41,7 @@ CONF = cfg.CONF
POST_REQUIRED = set(['code'])
CODE_SOURCE = set(['package', 'swift', 'image'])
UPDATE_ALLOWED = set(['name', 'description', 'entry'])
UPDATE_ALLOWED = set(['name', 'description', 'code', 'package', 'entry'])
class FunctionsController(rest.RestController):
@ -56,6 +57,15 @@ class FunctionsController(rest.RestController):
super(FunctionsController, self).__init__(*args, **kwargs)
def _check_swift(self, container, object):
# Auth needs to be enabled because qinling needs to check swift
# object using user's credential.
if not CONF.pecan.auth_enable:
raise exc.InputException('Swift object not supported.')
if not swift_util.check_object(container, object):
raise exc.InputException('Object does not exist in Swift.')
@rest_utils.wrap_pecan_controller_exception
@pecan.expose()
def get(self, id):
@ -121,7 +131,7 @@ class FunctionsController(rest.RestController):
', '.join(CODE_SOURCE)
)
if source != 'image':
if source != constants.IMAGE_FUNCTION:
if not kwargs.get('runtime_id'):
raise exc.InputException('"runtime_id" must be specified.')
@ -132,20 +142,13 @@ class FunctionsController(rest.RestController):
)
store = False
if values['code']['source'] == 'package':
if values['code']['source'] == constants.PACKAGE_FUNCTION:
store = True
data = kwargs['package'].file.read()
elif values['code']['source'] == 'swift':
# Auth needs to be enabled because qinling needs to check swift
# object using user's credential.
if not CONF.pecan.auth_enable:
raise exc.InputException('Swift object not supported.')
container = values['code']['swift'].get('container')
object = values['code']['swift'].get('object')
if not swift_util.check_object(container, object):
raise exc.InputException('Object does not exist in Swift.')
elif values['code']['source'] == constants.SWIFT_FUNCTION:
swift_info = values['code'].get('swift', {})
self._check_swift(swift_info.get('container'),
swift_info.get('object'))
if cfg.CONF.pecan.auth_enable:
try:
@ -208,33 +211,70 @@ class FunctionsController(rest.RestController):
# This will also delete function service mapping as well.
db_api.delete_function(id)
@rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(
resources.Function,
types.uuid,
body=resources.Function
)
def put(self, id, func):
@rest_utils.wrap_pecan_controller_exception
@pecan.expose('json')
def put(self, id, **kwargs):
"""Update function.
Currently, we only support update name, description, entry.
- Function can not being used by job.
- Function can not being executed.
- (TODO)Function status should be changed so no execution will create
when function is updating.
"""
values = {}
for key in UPDATE_ALLOWED:
if func.to_dict().get(key) is not None:
values.update({key: func.to_dict()[key]})
if kwargs.get(key) is not None:
values.update({key: kwargs[key]})
LOG.info('Update resource, params: %s', values,
resource={'type': self.type, 'id': id})
with db_api.transaction():
ctx = context.get_ctx()
if set(values.keys()).issubset(set(['name', 'description'])):
func_db = db_api.update_function(id, values)
if 'entry' in values:
# Update entry will delete allocated resources in orchestrator.
else:
source = values.get('code', {}).get('source')
with db_api.transaction():
pre_func = db_api.get_function(id)
if len(pre_func.jobs) > 0:
raise exc.NotAllowedException(
'The function is still associated with running job(s).'
)
pre_source = pre_func.code['source']
if source and source != pre_source:
raise exc.InputException(
"The function code type can not be changed."
)
if source == constants.IMAGE_FUNCTION:
raise exc.InputException(
"The image type function code can not be changed."
)
if (pre_source == constants.PACKAGE_FUNCTION and
values.get('package') is not None):
# Update the package data.
data = values['package'].file.read()
self.storage_provider.store(
ctx.projectid,
id,
data
)
values.pop('package')
if pre_source == constants.SWIFT_FUNCTION:
swift_info = values['code'].get('swift', {})
self._check_swift(swift_info.get('container'),
swift_info.get('object'))
# Delete allocated resources in orchestrator.
db_api.delete_function_service_mapping(id)
self.engine_client.delete_function(id)
return resources.Function.from_dict(func_db.to_dict())
func_db = db_api.update_function(id, values)
pecan.response.status = 200
return resources.Function.from_dict(func_db.to_dict()).to_dict()
@rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(

1
qinling/context.py
View File

@ -106,6 +106,7 @@ class Context(oslo_context.RequestContext):
{
'is_trust_scoped': self.is_trust_scoped,
'trust_id': self.trust_id,
'auth_token': self.auth_token,
}
)

3
qinling/engine/default_engine.py
View File

@ -19,6 +19,7 @@ import requests
from qinling.db import api as db_api
from qinling import status
from qinling.utils import common
from qinling.utils import constants
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
@ -119,7 +120,7 @@ class DefaultEngine(object):
identifier = None
labels = None
if source == 'image':
if source == constants.IMAGE_FUNCTION:
image = function.code['image']
identifier = ('%s-%s' %
(common.generate_unicode_uuid(dashed=False),

13
qinling/storage/file_system.py
View File

@ -33,6 +33,12 @@ class FileSystemStorage(base.PackageStorage):
fileutils.ensure_tree(CONF.storage.file_system_dir)
def store(self, project_id, function, data):
"""Store the function package data to local file system.
:param project_id: Project ID.
:param function: Function ID.
:param data: Package data.
"""
LOG.info(
'Store package, function: %s, project: %s', function, project_id
)
@ -46,10 +52,15 @@ class FileSystemStorage(base.PackageStorage):
if not zipfile.is_zipfile(func_zip):
fileutils.delete_if_exists(func_zip)
raise exc.InputException("Package is not a valid ZIP package.")
def retrieve(self, project_id, function):
"""Get function package data.
:param project_id: Project ID.
:param function: Function ID.
:return: File descriptor that needs to close outside.
"""
LOG.info(
'Get package data, function: %s, project: %s', function, project_id
)

22
qinling/tests/unit/api/controllers/v1/test_function.py
View File

@ -36,10 +36,6 @@ class TestFunctionController(base.APITest):
@mock.patch('qinling.storage.file_system.FileSystemStorage.store')
def test_post(self, mock_store):
class File(object):
def __init__(self, f):
self.file = f
with tempfile.NamedTemporaryFile() as f:
body = {
'name': self.rand_name('function', prefix=TEST_CASE_NAME),
@ -107,6 +103,24 @@ class TestFunctionController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertEqual('new_name', resp.json['name'])
@mock.patch('qinling.storage.file_system.FileSystemStorage.store')
@mock.patch('qinling.rpc.EngineClient.delete_function')
def test_put_package(self, mock_delete_func, mock_store):
db_func = self.create_function(
runtime_id=self.runtime_id, prefix=TEST_CASE_NAME
)
with tempfile.NamedTemporaryFile() as f:
resp = self.app.put(
'/v1/functions/%s' % db_func.id,
params={},
upload_files=[('package', f.name, f.read())]
)
self.assertEqual(200, resp.status_int)
self.assertEqual(1, mock_store.call_count)
mock_delete_func.assert_called_once_with(db_func.id)
@mock.patch('qinling.rpc.EngineClient.delete_function')
@mock.patch('qinling.storage.file_system.FileSystemStorage.delete')
def test_delete(self, mock_delete, mock_delete_func):

4
qinling/utils/constants.py
View File

@ -16,3 +16,7 @@ EXECUTION_BY_JOB = 'Created by Job %s'
PERIODIC_JOB_HANDLER = 'job_handler'
PERIODIC_FUNC_MAPPING_HANDLER = 'function_mapping_handler'
PACKAGE_FUNCTION = 'package'
SWIFT_FUNCTION = 'swift'
IMAGE_FUNCTION = 'image'