Create 2nd thread to handle USM REST API requests
This change is to create 2nd thread to provide concurrent service. In a different commit [1], the haproxy is to be configured to distribute the slow requests to the 2nd thread, and the fast requests to the primiary thread. TCs: passed: concurrent keystone requests of "software upload/ deploy precheck/deploy start" and "software list/deploy show/ deploy host-list" passed: keystone authenticated "software deploy precheck" request completed. Story: 2010676 Task: 49647 [1] https://review.opendev.org/c/starlingx/stx-puppet/+/910644 Change-Id: I0e8e8ac1b5177f1bbf40e047335c075b0a471fc1 Signed-off-by: Bin Qian <bin.qian@windriver.com>
This commit is contained in:
parent
1fb44f6b28
commit
b700e3f4a1
|
@ -1,4 +1,4 @@
|
||||||
# Copyright (c) 2023 Wind River Systems, Inc.
|
# Copyright (c) 2023-2024 Wind River Systems, Inc.
|
||||||
#
|
#
|
||||||
# SPDX-License-Identifier: Apache-2.0
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
#
|
#
|
||||||
|
@ -11,7 +11,10 @@ API_SERVICE_OPTS = [
|
||||||
help='IP for the authenticated Unified Software Management API server to bind to'),
|
help='IP for the authenticated Unified Software Management API server to bind to'),
|
||||||
cfg.IntOpt('auth_api_port',
|
cfg.IntOpt('auth_api_port',
|
||||||
default=5497,
|
default=5497,
|
||||||
help='The port for the authenticated Unified Software Management API server'),
|
help='The port for the authenticated Unified Software Management API server for GET operations'),
|
||||||
|
cfg.IntOpt('auth_api_alt_port',
|
||||||
|
default=5499,
|
||||||
|
help='The port for the authenticated Unified Software Management API server for update and slow operations'),
|
||||||
cfg.IntOpt('api_limit_max',
|
cfg.IntOpt('api_limit_max',
|
||||||
default=1000,
|
default=1000,
|
||||||
help='the maximum number of items returned in a single '
|
help='the maximum number of items returned in a single '
|
||||||
|
|
|
@ -1,16 +1,10 @@
|
||||||
"""
|
"""
|
||||||
Copyright (c) 2023 Wind River Systems, Inc.
|
Copyright (c) 2023-2024 Wind River Systems, Inc.
|
||||||
|
|
||||||
SPDX-License-Identifier: Apache-2.0
|
SPDX-License-Identifier: Apache-2.0
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# Server Specific Configurations
|
|
||||||
server = {
|
|
||||||
'port': '5497',
|
|
||||||
'host': '0.0.0.0'
|
|
||||||
}
|
|
||||||
|
|
||||||
# Pecan Application Configurations
|
# Pecan Application Configurations
|
||||||
app = {
|
app = {
|
||||||
'root': 'software.api.controllers.root.RootController',
|
'root': 'software.api.controllers.root.RootController',
|
||||||
|
|
|
@ -1,14 +1,18 @@
|
||||||
"""
|
"""
|
||||||
Copyright (c) 2023 Wind River Systems, Inc.
|
Copyright (c) 2023-2024 Wind River Systems, Inc.
|
||||||
|
|
||||||
SPDX-License-Identifier: Apache-2.0
|
SPDX-License-Identifier: Apache-2.0
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import threading
|
||||||
from software.software_entities import DeployHandler
|
from software.software_entities import DeployHandler
|
||||||
from software.software_entities import DeployHostHandler
|
from software.software_entities import DeployHostHandler
|
||||||
from software.constants import DEPLOY_STATES
|
from software.constants import DEPLOY_STATES
|
||||||
|
|
||||||
|
LOG = logging.getLogger('main_logger')
|
||||||
|
|
||||||
|
|
||||||
def get_instance():
|
def get_instance():
|
||||||
"""Return a Software API instance."""
|
"""Return a Software API instance."""
|
||||||
|
@ -17,6 +21,7 @@ def get_instance():
|
||||||
|
|
||||||
class SoftwareAPI:
|
class SoftwareAPI:
|
||||||
_instance = None
|
_instance = None
|
||||||
|
_lock = threading.RLock()
|
||||||
|
|
||||||
def __new__(cls):
|
def __new__(cls):
|
||||||
if cls._instance is None:
|
if cls._instance is None:
|
||||||
|
@ -28,28 +33,74 @@ class SoftwareAPI:
|
||||||
self.deploy_host_handler = DeployHostHandler()
|
self.deploy_host_handler = DeployHostHandler()
|
||||||
|
|
||||||
def create_deploy(self, from_release, to_release, reboot_required: bool):
|
def create_deploy(self, from_release, to_release, reboot_required: bool):
|
||||||
|
self.begin_update()
|
||||||
self.deploy_handler.create(from_release, to_release, reboot_required)
|
self.deploy_handler.create(from_release, to_release, reboot_required)
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def get_deploy(self):
|
def get_deploy(self):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
return self.deploy_handler.query()
|
return self.deploy_handler.query()
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def update_deploy(self, state: DEPLOY_STATES):
|
def update_deploy(self, state: DEPLOY_STATES):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
self.deploy_handler.update(state)
|
self.deploy_handler.update(state)
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def delete_deploy(self):
|
def delete_deploy(self):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
self.deploy_handler.delete()
|
self.deploy_handler.delete()
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def create_deploy_host(self, hostname):
|
def create_deploy_host(self, hostname):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
self.deploy_host_handler.create(hostname)
|
self.deploy_host_handler.create(hostname)
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def get_deploy_host(self):
|
def get_deploy_host(self):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
return self.deploy_host_handler.query_all()
|
return self.deploy_host_handler.query_all()
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def update_deploy_host(self, hostname, state):
|
def update_deploy_host(self, hostname, state):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
return self.deploy_host_handler.update(hostname, state)
|
return self.deploy_host_handler.update(hostname, state)
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def delete_deploy_host(self, hostname):
|
def delete_deploy_host(self, hostname):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
self.deploy_host_handler.delete(hostname)
|
self.deploy_host_handler.delete(hostname)
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
def delete_deploy_host_all(self):
|
def delete_deploy_host_all(self):
|
||||||
|
self.begin_update()
|
||||||
|
try:
|
||||||
self.deploy_host_handler.delete_all()
|
self.deploy_host_handler.delete_all()
|
||||||
|
finally:
|
||||||
|
self.end_update()
|
||||||
|
|
||||||
|
def begin_update(self):
|
||||||
|
tid = threading.get_native_id()
|
||||||
|
msg = f"{tid} is to acquire lock."
|
||||||
|
LOG.info(msg)
|
||||||
|
SoftwareAPI._lock.acquire()
|
||||||
|
|
||||||
|
def end_update(self):
|
||||||
|
SoftwareAPI._lock.release()
|
||||||
|
tid = threading.get_native_id()
|
||||||
|
msg = f"{tid} released lock."
|
||||||
|
LOG.info(msg)
|
||||||
|
|
|
@ -2267,10 +2267,15 @@ class PatchController(PatchService):
|
||||||
|
|
||||||
if self._deploy_upgrade_start(to_release):
|
if self._deploy_upgrade_start(to_release):
|
||||||
collect_current_load_for_hosts()
|
collect_current_load_for_hosts()
|
||||||
|
self.db_api_instance.begin_update()
|
||||||
|
try:
|
||||||
self.update_and_sync_deploy_state(self.db_api_instance.create_deploy,
|
self.update_and_sync_deploy_state(self.db_api_instance.create_deploy,
|
||||||
SW_VERSION, to_release, True)
|
SW_VERSION, to_release, True)
|
||||||
self.update_and_sync_deploy_state(self.db_api_instance.update_deploy,
|
self.update_and_sync_deploy_state(self.db_api_instance.update_deploy,
|
||||||
DEPLOY_STATES.START)
|
DEPLOY_STATES.START)
|
||||||
|
finally:
|
||||||
|
self.db_api_instance.end_update()
|
||||||
|
|
||||||
sw_rel = self.release_collection.get_release_by_id(deployment)
|
sw_rel = self.release_collection.get_release_by_id(deployment)
|
||||||
if sw_rel is None:
|
if sw_rel is None:
|
||||||
raise InternalError("%s cannot be found" % to_release)
|
raise InternalError("%s cannot be found" % to_release)
|
||||||
|
@ -2931,7 +2936,6 @@ class PatchController(PatchService):
|
||||||
return query_hosts
|
return query_hosts
|
||||||
return deploy_host_list
|
return deploy_host_list
|
||||||
|
|
||||||
|
|
||||||
def update_and_sync_deploy_state(self, func, *args, **kwargs):
|
def update_and_sync_deploy_state(self, func, *args, **kwargs):
|
||||||
"""
|
"""
|
||||||
:param func: SoftwareApi method
|
:param func: SoftwareApi method
|
||||||
|
@ -3001,14 +3005,14 @@ class PatchControllerApiThread(threading.Thread):
|
||||||
|
|
||||||
|
|
||||||
class PatchControllerAuthApiThread(threading.Thread):
|
class PatchControllerAuthApiThread(threading.Thread):
|
||||||
def __init__(self):
|
def __init__(self, port):
|
||||||
threading.Thread.__init__(self)
|
threading.Thread.__init__(self)
|
||||||
# LOG.info ("Initializing Authenticated API thread")
|
# LOG.info ("Initializing Authenticated API thread")
|
||||||
self.wsgi = None
|
self.wsgi = None
|
||||||
|
self.port = port
|
||||||
|
|
||||||
def run(self):
|
def run(self):
|
||||||
host = CONF.auth_api_bind_ip
|
host = CONF.auth_api_bind_ip
|
||||||
port = CONF.auth_api_port
|
|
||||||
if host is None:
|
if host is None:
|
||||||
host = utils.get_versioned_address_all()
|
host = utils.get_versioned_address_all()
|
||||||
try:
|
try:
|
||||||
|
@ -3027,7 +3031,7 @@ class PatchControllerAuthApiThread(threading.Thread):
|
||||||
|
|
||||||
server_class.address_family = utils.get_management_family()
|
server_class.address_family = utils.get_management_family()
|
||||||
self.wsgi = simple_server.make_server(
|
self.wsgi = simple_server.make_server(
|
||||||
host, port,
|
host, self.port,
|
||||||
auth_app.VersionSelectorApplication(),
|
auth_app.VersionSelectorApplication(),
|
||||||
server_class=server_class)
|
server_class=server_class)
|
||||||
|
|
||||||
|
@ -3286,11 +3290,13 @@ def main():
|
||||||
|
|
||||||
LOG.info("launching")
|
LOG.info("launching")
|
||||||
api_thread = PatchControllerApiThread()
|
api_thread = PatchControllerApiThread()
|
||||||
auth_api_thread = PatchControllerAuthApiThread()
|
auth_api_thread = PatchControllerAuthApiThread(CONF.auth_api_port)
|
||||||
|
auth_api_alt_thread = PatchControllerAuthApiThread(CONF.auth_api_alt_port)
|
||||||
main_thread = PatchControllerMainThread()
|
main_thread = PatchControllerMainThread()
|
||||||
|
|
||||||
api_thread.start()
|
api_thread.start()
|
||||||
auth_api_thread.start()
|
auth_api_thread.start()
|
||||||
|
auth_api_alt_thread.start()
|
||||||
main_thread.start()
|
main_thread.start()
|
||||||
|
|
||||||
thread_death.wait()
|
thread_death.wait()
|
||||||
|
@ -3299,4 +3305,5 @@ def main():
|
||||||
|
|
||||||
api_thread.join()
|
api_thread.join()
|
||||||
auth_api_thread.join()
|
auth_api_thread.join()
|
||||||
|
auth_api_alt_thread.join()
|
||||||
main_thread.join()
|
main_thread.join()
|
||||||
|
|
Loading…
Reference in New Issue