diff options
Diffstat (limited to 'build/f_repos/patch/fuel-web/0002-Revert-Distributed-serialization-implementation.patch')
| -rw-r--r-- | build/f_repos/patch/fuel-web/0002-Revert-Distributed-serialization-implementation.patch | 1006 |
1 files changed, 0 insertions, 1006 deletions
diff --git a/build/f_repos/patch/fuel-web/0002-Revert-Distributed-serialization-implementation.patch b/build/f_repos/patch/fuel-web/0002-Revert-Distributed-serialization-implementation.patch deleted file mode 100644 index 152197e74..000000000 --- a/build/f_repos/patch/fuel-web/0002-Revert-Distributed-serialization-implementation.patch +++ /dev/null @@ -1,1006 +0,0 @@ -From: Alexandru Avadanii <Alexandru.Avadanii@enea.com> -Date: Sun, 26 Mar 2017 19:16:47 +0200 -Subject: [PATCH] Revert "Distributed serialization implementation" - -This reverts commit f3c440f76961ef161ab6650b3a41a3c775073ba0. ---- - nailgun/nailgun/consts.py | 4 - - nailgun/nailgun/fixtures/openstack.yaml | 49 --- - nailgun/nailgun/lcm/task_serializer.py | 15 +- - nailgun/nailgun/lcm/transaction_serializer.py | 317 +------------------ - .../nailgun/orchestrator/deployment_serializers.py | 1 - - nailgun/nailgun/settings.py | 22 +- - nailgun/nailgun/settings.yaml | 9 - - .../fuel_statistics/installation_info.py | 10 - - .../integration/test_cluster_changes_handler.py | 3 - - .../test/unit/test_lcm_transaction_serializer.py | 351 +-------------------- - nailgun/requirements.txt | 2 - - 11 files changed, 14 insertions(+), 769 deletions(-) - -diff --git a/nailgun/nailgun/consts.py b/nailgun/nailgun/consts.py -index 1a2f09f..5490d6c 100644 ---- a/nailgun/nailgun/consts.py -+++ b/nailgun/nailgun/consts.py -@@ -528,7 +528,3 @@ HYPERVISORS = Enum( - ) - - DPDK_DRIVER_IN_SRIOV_CASE = 'vfio-pci' -- --SERIALIZATION_POLICY = Enum( -- 'distributed' --) -diff --git a/nailgun/nailgun/fixtures/openstack.yaml b/nailgun/nailgun/fixtures/openstack.yaml -index 5557710..8b4d763 100644 ---- a/nailgun/nailgun/fixtures/openstack.yaml -+++ b/nailgun/nailgun/fixtures/openstack.yaml -@@ -1114,55 +1114,6 @@ - group: "security" - weight: 20 - type: "radio" -- serialization_policy: -- value: "default" -- values: -- - data: "default" -- label: "Default serialization" -- description: "Run serialization on the master node only" -- - data: "distributed" -- label: "Distributed serialization" -- description: "Run serialization on the master and environment nodes. Nodes for serialization are selected only form that environment for wich serialization is performing." -- label: "Serialization policy" -- group: "general" -- weight: 30 -- type: "radio" -- ds_use_discover: -- group: "general" -- label: "Use discovered nodes as workers for serialization" -- type: "checkbox" -- value: true -- weight: 31 -- restrictions: -- - condition: "settings:common.serialization_policy.value != 'distributed'" -- action: "hide" -- ds_use_provisioned: -- group: "general" -- label: "Use provisioned nodes as workers for serialization" -- type: "checkbox" -- value: true -- weight: 32 -- restrictions: -- - condition: "settings:common.serialization_policy.value != 'distributed'" -- action: "hide" -- ds_use_error: -- group: "general" -- label: "Use nodes in error state as workers for serialization" -- type: "checkbox" -- value: true -- weight: 33 -- restrictions: -- - condition: "settings:common.serialization_policy.value != 'distributed'" -- action: "hide" -- ds_use_ready: -- group: "general" -- label: "Use ready nodes as workers for serialization" -- type: "checkbox" -- value: false -- weight: 34 -- restrictions: -- - condition: "settings:common.serialization_policy.value != 'distributed'" -- action: "hide" - public_network_assignment: - metadata: - weight: 10 -diff --git a/nailgun/nailgun/lcm/task_serializer.py b/nailgun/nailgun/lcm/task_serializer.py -index 0d70f36..c004115 100644 ---- a/nailgun/nailgun/lcm/task_serializer.py -+++ b/nailgun/nailgun/lcm/task_serializer.py -@@ -110,8 +110,6 @@ class Context(object): - return evaluate - - def get_formatter_context(self, node_id): -- # TODO(akislitsky) remove formatter context from the -- # tasks serialization workflow - data = self._transaction.get_new_data(node_id) - return { - 'CLUSTER_ID': data.get('cluster', {}).get('id'), -@@ -149,14 +147,9 @@ class DeploymentTaskSerializer(object): - :return: the result - """ - -- def serialize(self, node_id, formatter_context=None): -- """Serialize task in expected by orchestrator format -+ def serialize(self, node_id): -+ """Serialize task in expected by orchestrator format. - -- If serialization is performed on the remote worker -- we should pass formatter_context parameter with values -- from the master node settings -- -- :param formatter_context: formatter context - :param node_id: the target node_id - """ - -@@ -164,12 +157,10 @@ class DeploymentTaskSerializer(object): - "serialize task %s for node %s", - self.task_template['id'], node_id - ) -- formatter_context = formatter_context \ -- or self.context.get_formatter_context(node_id) - task = utils.traverse( - self.task_template, - utils.text_format_safe, -- formatter_context, -+ self.context.get_formatter_context(node_id), - { - 'yaql_exp': self.context.get_yaql_interpreter( - node_id, self.task_template['id']) -diff --git a/nailgun/nailgun/lcm/transaction_serializer.py b/nailgun/nailgun/lcm/transaction_serializer.py -index 32953c1..c02146b 100644 ---- a/nailgun/nailgun/lcm/transaction_serializer.py -+++ b/nailgun/nailgun/lcm/transaction_serializer.py -@@ -14,21 +14,13 @@ - # License for the specific language governing permissions and limitations - # under the License. - --import datetime -+from distutils.version import StrictVersion - import multiprocessing --import os --from Queue import Queue --import shutil --import tempfile - --import distributed --from distutils.version import StrictVersion - import six --import toolz - - from nailgun import consts - from nailgun import errors --from nailgun.lcm.task_serializer import Context - from nailgun.lcm.task_serializer import TasksSerializersFactory - from nailgun.logger import logger - from nailgun.settings import settings -@@ -136,308 +128,7 @@ class MultiProcessingConcurrencyPolicy(object): - pool.join() - - --def _distributed_serialize_tasks_for_node(formatter_contexts_idx, -- node_and_tasks, scattered_data): -- """Remote serialization call for DistributedProcessingPolicy -- -- Code of the function is copied to the workers and executed there, thus -- we are including all required imports inside the function. -- -- :param formatter_contexts_idx: dict of formatter contexts with node_id -- value as key -- :param node_and_tasks: list of node_id, task_data tuples -- :param scattered_data: feature object, that points to data copied to -- workers -- :return: [(node_id, serialized), error] -- """ -- -- try: -- factory = TasksSerializersFactory(scattered_data['context']) -- -- # Restoring settings -- settings.config = scattered_data['settings_config'] -- for k in formatter_contexts_idx: -- formatter_contexts_idx[k]['SETTINGS'] = settings -- -- except Exception as e: -- logger.exception("Distributed serialization failed") -- return [((None, None), e)] -- -- result = [] -- -- for node_and_task in node_and_tasks: -- -- node_id = None -- try: -- node_id, task = node_and_task -- -- logger.debug("Starting distributed node %s task %s serialization", -- node_id, task['id']) -- -- formatter_context = formatter_contexts_idx[node_id] -- -- serializer = factory.create_serializer(task) -- serialized = serializer.serialize( -- node_id, formatter_context=formatter_context) -- -- logger.debug("Distributed node %s task %s serialization " -- "result: %s", node_id, task['id'], serialized) -- -- result.append(((node_id, serialized), None)) -- except Exception as e: -- logger.exception("Distributed serialization failed") -- result.append(((node_id, None), e)) -- break -- -- logger.debug("Processed tasks count: %s", len(result)) -- return result -- -- --class DistributedProcessingPolicy(object): -- -- def __init__(self): -- self.sent_jobs = Queue() -- self.sent_jobs_count = 0 -- -- def _consume_jobs(self, chunk_size=None): -- """Consumes jobs -- -- If chunk_size is set function consumes specified number of -- Finished tasks or less if sent_jobs_ids queue became empty. -- If chunk_size is None function consumes jobs until -- sent_jobs_ids queue became empty. -- Jobs with statuses Cancelled, Abandoned, Terminated will be -- resent and their ids added to sent_jobs_ids queue -- -- :param chunk_size: size of consuming chunk -- :return: generator on job results -- """ -- logger.debug("Consuming jobs started") -- -- jobs_to_consume = [] -- while not self.sent_jobs.empty(): -- job = self.sent_jobs.get() -- jobs_to_consume.append(job) -- -- if chunk_size is not None: -- chunk_size -= 1 -- if chunk_size <= 0: -- break -- -- for ready_job in distributed.as_completed(jobs_to_consume): -- results = ready_job.result() -- self.sent_jobs_count -= 1 -- -- for result in results: -- (node_id, serialized), exc = result -- logger.debug("Got ready task for node %s, serialized: %s, " -- "error: %s", node_id, serialized, exc) -- if exc is not None: -- raise exc -- yield node_id, serialized -- -- logger.debug("Consuming jobs finished") -- -- def _get_formatter_context(self, task_context, formatter_contexts_idx, -- node_id): -- try: -- return formatter_contexts_idx[node_id] -- except KeyError: -- pass -- -- logger.debug("Calculating formatter context for node %s", node_id) -- formatter_context = task_context.get_formatter_context( -- node_id) -- # Settings file is already sent to the workers -- formatter_context.pop('SETTINGS', None) -- formatter_contexts_idx[node_id] = formatter_context -- -- return formatter_context -- -- def _upload_nailgun_code(self, job_cluster): -- """Creates zip of current nailgun code and uploads it to workers -- -- TODO(akislitsky): add workers scope when it will be implemented -- in the distributed library -- -- :param job_cluster: distributed.Client -- """ -- logger.debug("Compressing nailgun code") -- file_dir = os.path.dirname(__file__) -- nailgun_root_dir = os.path.realpath(os.path.join(file_dir, '..', '..')) -- archive = os.path.join(tempfile.gettempdir(), 'nailgun') -- result = shutil.make_archive(archive, 'zip', nailgun_root_dir, -- 'nailgun') -- logger.debug("Nailgun code saved to: %s", result) -- -- logger.debug("Uploading nailgun archive %s to workers", result) -- job_cluster.upload_file(result) -- -- def _scatter_data(self, job_cluster, context, workers): -- logger.debug("Scattering data to workers started") -- shared_data = {'context': context, 'settings_config': settings.config} -- scattered = job_cluster.scatter(shared_data, broadcast=True, -- workers=workers) -- # Waiting data is scattered to workers -- distributed.wait(scattered.values()) -- logger.debug("Scattering data to workers finished") -- -- return scattered -- -- def _get_allowed_nodes_statuses(self, context): -- """Extracts node statuses that allows distributed serialization""" -- common = context.new.get('common', {}) -- cluster = common.get('cluster', {}) -- logger.debug("Getting allowed nodes statuses to use as serialization " -- "workers for cluster %s", cluster.get('id')) -- check_fields = { -- 'ds_use_ready': consts.NODE_STATUSES.ready, -- 'ds_use_provisioned': consts.NODE_STATUSES.provisioned, -- 'ds_use_discover': consts.NODE_STATUSES.discover, -- 'ds_use_error': consts.NODE_STATUSES.error -- } -- statuses = set() -- for field, node_status in check_fields.items(): -- if common.get(field): -- statuses.add(node_status) -- -- logger.debug("Allowed nodes statuses to use as serialization workers " -- "for cluster %s are: %s", cluster.get('id'), statuses) -- return statuses -- -- def _get_allowed_nodes_ips(self, context): -- """Filters online nodes from cluster by their status -- -- In the cluster settings we select nodes statuses allowed for -- using in the distributed serialization. Accordingly to selected -- statuses nodes are going to be filtered. -- -- :param context: TransactionContext -- :return: set of allowed nodes ips -- """ -- ips = set() -- allowed_statuses = self._get_allowed_nodes_statuses(context) -- for node in six.itervalues(context.new.get('nodes', {})): -- if node.get('status') in allowed_statuses: -- ips.add(node.get('ip')) -- ips.add(settings.MASTER_IP) -- return ips -- -- def _get_allowed_workers(self, job_cluster, allowed_ips): -- """Calculates workers addresses for distributed serialization -- -- Only workers that placed on the allowed nodes must be selected -- for the serialization. -- -- :param job_cluster: distributed.Client -- :param allowed_ips: allowed for serialization nodes ips -- :return: list of workers addresses in format 'ip:port' -- """ -- logger.debug("Getting allowed workers") -- workers = {} -- -- # Worker has address like tcp://ip:port -- info = job_cluster.scheduler_info() -- for worker_addr in six.iterkeys(info['workers']): -- ip_port = worker_addr.split('//')[1] -- ip = ip_port.split(':')[0] -- if ip not in allowed_ips: -- continue -- try: -- pool = workers[ip] -- pool.add(ip_port) -- except KeyError: -- workers[ip] = set([ip_port]) -- -- return list(toolz.itertoolz.concat(six.itervalues(workers))) -- -- def execute(self, context, _, tasks): -- """Executes task serialization on distributed nodes -- -- :param context: the transaction context -- :param _: serializers factory -- :param tasks: the tasks to serialize -- :return sequence of serialized tasks -- """ -- logger.debug("Performing distributed tasks processing") -- sched_address = '{0}:{1}'.format(settings.MASTER_IP, -- settings.LCM_DS_JOB_SHEDULER_PORT) -- job_cluster = distributed.Client(sched_address) -- -- allowed_ips = self._get_allowed_nodes_ips(context) -- workers = self._get_allowed_workers(job_cluster, allowed_ips) -- logger.debug("Allowed workers list for serialization: %s", workers) -- workers_ips = set([ip_port.split(':')[0] for ip_port in workers]) -- logger.debug("Allowed workers ips list for serialization: %s", -- workers_ips) -- -- task_context = Context(context) -- formatter_contexts_idx = {} -- workers_num = len(workers) -- max_jobs_in_queue = workers_num * settings.LCM_DS_NODE_LOAD_COEFF -- logger.debug("Max jobs allowed in queue: %s", max_jobs_in_queue) -- -- start = datetime.datetime.utcnow() -- tasks_count = 0 -- -- try: -- self._upload_nailgun_code(job_cluster) -- scattered = self._scatter_data(job_cluster, context, workers) -- -- for tasks_chunk in toolz.partition_all( -- settings.LCM_DS_TASKS_PER_JOB, tasks): -- -- formatter_contexts_for_tasks = {} -- -- # Collecting required contexts for tasks -- for task in tasks_chunk: -- node_id, task_data = task -- formatter_context = self._get_formatter_context( -- task_context, formatter_contexts_idx, node_id) -- if node_id not in formatter_contexts_for_tasks: -- formatter_contexts_for_tasks[node_id] = \ -- formatter_context -- -- logger.debug("Submitting job for tasks chunk: %s", tasks_chunk) -- job = job_cluster.submit( -- _distributed_serialize_tasks_for_node, -- formatter_contexts_for_tasks, -- tasks_chunk, -- scattered, -- workers=workers_ips -- ) -- -- self.sent_jobs.put(job) -- self.sent_jobs_count += 1 -- -- # We are limit the max number of tasks by the number of nodes -- # which are used in the serialization -- if self.sent_jobs_count >= max_jobs_in_queue: -- for result in self._consume_jobs(chunk_size=workers_num): -- tasks_count += 1 -- yield result -- -- # We have no tasks any more but have unconsumed jobs -- for result in self._consume_jobs(): -- tasks_count += 1 -- yield result -- finally: -- end = datetime.datetime.utcnow() -- logger.debug("Distributed tasks processing finished. " -- "Total time: %s. Tasks processed: %s", -- end - start, tasks_count) -- job_cluster.shutdown() -- -- --def is_distributed_processing_enabled(context): -- common = context.new.get('common', {}) -- return common.get('serialization_policy') == \ -- consts.SERIALIZATION_POLICY.distributed -- -- --def get_processing_policy(context): -- if is_distributed_processing_enabled(context): -- return DistributedProcessingPolicy() -+def get_concurrency_policy(): - cpu_num = settings.LCM_SERIALIZERS_CONCURRENCY_FACTOR - if not cpu_num: - try: -@@ -471,7 +162,7 @@ class TransactionSerializer(object): - # ids of nodes in this group and how many nodes in this group can fail - # and deployment will not be interrupted - self.fault_tolerance_groups = [] -- self.processing_policy = get_processing_policy(context) -+ self.concurrency_policy = get_concurrency_policy() - - @classmethod - def serialize(cls, context, tasks, resolver): -@@ -525,7 +216,7 @@ class TransactionSerializer(object): - :param tasks: the deployment tasks - :return the mapping tasks per node - """ -- serialized = self.processing_policy.execute( -+ serialized = self.concurrency_policy.execute( - self.context, - self.serializer_factory_class, - self.expand_tasks(tasks) -diff --git a/nailgun/nailgun/orchestrator/deployment_serializers.py b/nailgun/nailgun/orchestrator/deployment_serializers.py -index 99d52a2..9473733 100644 ---- a/nailgun/nailgun/orchestrator/deployment_serializers.py -+++ b/nailgun/nailgun/orchestrator/deployment_serializers.py -@@ -221,7 +221,6 @@ class DeploymentMultinodeSerializer(object): - 'role': role, - 'vms_conf': node.vms_conf, - 'fail_if_error': role in self.critical_roles, -- 'ip': node.ip, - # TODO(eli): need to remove, requried for the fake thread only - 'online': node.online, - } -diff --git a/nailgun/nailgun/settings.py b/nailgun/nailgun/settings.py -index 1f9222b..eee08c9 100644 ---- a/nailgun/nailgun/settings.py -+++ b/nailgun/nailgun/settings.py -@@ -38,21 +38,7 @@ class NailgunSettings(object): - if test_config: - settings_files.append(test_config) - -- # If settings.yaml doesn't exist we should have default -- # config structure. Nailgun without settings is used -- # when we distribute source code to the workers for -- # distributed serialization -- self.config = { -- 'VERSION': {}, -- 'DATABASE': { -- 'engine': 'postgresql', -- 'name': '', -- 'host': '', -- 'port': '0', -- 'user': '', -- 'passwd': '' -- } -- } -+ self.config = {} - for sf in settings_files: - try: - logger.debug("Trying to read config file %s" % sf) -@@ -61,9 +47,9 @@ class NailgunSettings(object): - logger.error("Error while reading config file %s: %s" % - (sf, str(e))) - -- self.config['VERSION']['api'] = self.config.get('API') -+ self.config['VERSION']['api'] = self.config['API'] - self.config['VERSION']['feature_groups'] = \ -- self.config.get('FEATURE_GROUPS') -+ self.config['FEATURE_GROUPS'] - - fuel_release = self.get_file_content(consts.FUEL_RELEASE_FILE) - if fuel_release: -@@ -75,7 +61,7 @@ class NailgunSettings(object): - self.config['VERSION']['openstack_version'] = \ - fuel_openstack_version - -- if int(self.config.get("DEVELOPMENT", 0)): -+ if int(self.config.get("DEVELOPMENT")): - logger.info("DEVELOPMENT MODE ON:") - here = os.path.abspath( - os.path.join(os.path.dirname(__file__), '..') -diff --git a/nailgun/nailgun/settings.yaml b/nailgun/nailgun/settings.yaml -index c2faaa8..92b3518 100644 ---- a/nailgun/nailgun/settings.yaml -+++ b/nailgun/nailgun/settings.yaml -@@ -177,15 +177,6 @@ YAQL_MEMORY_QUOTA: 104857600 - - LCM_CHECK_TASK_VERSION: False - --# Coefficient for calculation max jobs queue length. If jobs number reaches the --# len(nodes) * load_coef we stop generate and start consume of jobs. --LCM_DS_NODE_LOAD_COEFF: 2 --# Port of dask-scheduler on the master node --LCM_DS_JOB_SHEDULER_PORT: 8002 --# Size of tasks chunk sending to the distributed worker --LCM_DS_TASKS_PER_JOB: 100 -- -- - DPDK_MAX_CPUS_PER_NIC: 4 - - TRUNCATE_LOG_ENTRIES: 100 -diff --git a/nailgun/nailgun/statistics/fuel_statistics/installation_info.py b/nailgun/nailgun/statistics/fuel_statistics/installation_info.py -index d024cdd..482d8c5 100644 ---- a/nailgun/nailgun/statistics/fuel_statistics/installation_info.py -+++ b/nailgun/nailgun/statistics/fuel_statistics/installation_info.py -@@ -56,16 +56,6 @@ class InstallationInfo(object): - 'propagate_task_deploy', None), - WhiteListRule(('common', 'security_groups', 'value'), - 'security_groups', None), -- WhiteListRule(('common', 'serialization_policy', 'value'), -- 'serialization_policy', None), -- WhiteListRule(('common', 'ds_use_discover', 'value'), -- 'ds_use_discover', None), -- WhiteListRule(('common', 'ds_use_provisioned', 'value'), -- 'ds_use_provisioned', None), -- WhiteListRule(('common', 'ds_use_ready', 'value'), -- 'ds_use_ready', None), -- WhiteListRule(('common', 'ds_use_error', 'value'), -- 'ds_use_error', None), - WhiteListRule(('corosync', 'verified', 'value'), - 'corosync_verified', None), - -diff --git a/nailgun/nailgun/test/integration/test_cluster_changes_handler.py b/nailgun/nailgun/test/integration/test_cluster_changes_handler.py -index b8f4e98..a8661ab 100644 ---- a/nailgun/nailgun/test/integration/test_cluster_changes_handler.py -+++ b/nailgun/nailgun/test/integration/test_cluster_changes_handler.py -@@ -187,7 +187,6 @@ class TestHandlers(BaseIntegrationTest): - 'fail_if_error': is_critical, - 'vms_conf': [], - 'fqdn': 'node-%d.%s' % (node.id, settings.DNS_DOMAIN), -- 'ip': node.ip, - - 'network_data': { - 'eth1': { -@@ -604,7 +603,6 @@ class TestHandlers(BaseIntegrationTest): - 'online': node.online, - 'fail_if_error': is_critical, - 'fqdn': 'node-%d.%s' % (node.id, settings.DNS_DOMAIN), -- 'ip': node.ip, - 'priority': 100, - 'vms_conf': [], - 'network_scheme': { -@@ -1098,7 +1096,6 @@ class TestHandlers(BaseIntegrationTest): - 'fail_if_error': is_critical, - 'fqdn': 'node-%d.%s' % (node.id, settings.DNS_DOMAIN), - 'priority': 100, -- 'ip': node.ip, - 'vms_conf': [], - - 'network_scheme': { -diff --git a/nailgun/nailgun/test/unit/test_lcm_transaction_serializer.py b/nailgun/nailgun/test/unit/test_lcm_transaction_serializer.py -index 9751d55..6450e19 100644 ---- a/nailgun/nailgun/test/unit/test_lcm_transaction_serializer.py -+++ b/nailgun/nailgun/test/unit/test_lcm_transaction_serializer.py -@@ -14,24 +14,20 @@ - # License for the specific language governing permissions and limitations - # under the License. - --import copy --import exceptions - import mock - import multiprocessing.dummy - - from nailgun import consts - from nailgun import errors - from nailgun import lcm --from nailgun.lcm import TransactionContext --from nailgun.settings import settings --from nailgun.test.base import BaseTestCase - from nailgun.utils.resolvers import TagResolver - -+from nailgun.test.base import BaseUnitTest - --class TestTransactionSerializer(BaseTestCase): -+ -+class TestTransactionSerializer(BaseUnitTest): - @classmethod - def setUpClass(cls): -- super(TestTransactionSerializer, cls).setUpClass() - cls.tasks = [ - { - 'id': 'task1', 'roles': ['controller'], -@@ -466,344 +462,3 @@ class TestTransactionSerializer(BaseTestCase): - 9, - lcm.TransactionSerializer.calculate_fault_tolerance('-1 ', 10) - ) -- -- def _get_context_for_distributed_serialization(self): -- new = copy.deepcopy(self.context.new) -- new['common']['serialization_policy'] = \ -- consts.SERIALIZATION_POLICY.distributed -- return TransactionContext(new) -- -- @mock.patch('nailgun.lcm.transaction_serializer.distributed.wait') -- @mock.patch('nailgun.lcm.transaction_serializer.distributed.as_completed') -- def test_distributed_serialization(self, _, as_completed): -- context = self._get_context_for_distributed_serialization() -- -- with mock.patch( -- 'nailgun.lcm.transaction_serializer.distributed.Client' -- ) as job_cluster: -- job = mock.Mock() -- job.result.return_value = [ -- (('1', {"id": "task1", "type": "skipped"}), None) -- ] -- -- submit = mock.Mock() -- submit.return_value = job -- -- as_completed.return_value = [job] -- -- job_cluster.return_value.submit = submit -- job_cluster.return_value.scheduler_info.return_value = \ -- {'workers': {'tcp://worker': {}}} -- -- lcm.TransactionSerializer.serialize( -- context, self.tasks, self.resolver) -- self.assertTrue(submit.called) -- # 4 controller task + 1 compute + 1 cinder -- self.assertTrue(6, submit.call_count) -- -- @mock.patch('nailgun.lcm.transaction_serializer.distributed.wait') -- @mock.patch('nailgun.lcm.transaction_serializer.distributed.as_completed') -- @mock.patch('nailgun.lcm.transaction_serializer.' -- 'DistributedProcessingPolicy._get_formatter_context') -- def test_distributed_serialization_workers_scope(self, formatter_context, -- as_completed, _): -- context = self._get_context_for_distributed_serialization() -- -- node_id = '1' -- task = { -- 'id': 'task1', 'roles': ['controller'], -- 'type': 'puppet', 'version': '2.0.0' -- } -- -- with mock.patch( -- 'nailgun.lcm.transaction_serializer.distributed.Client' -- ) as job_cluster: -- -- # Mocking job processing -- job = mock.Mock() -- job.result.return_value = [((node_id, task), None)] -- -- submit = mock.Mock() -- submit.return_value = job -- -- as_completed.return_value = [job] -- -- scatter = mock.Mock() -- job_cluster.return_value.scatter = scatter -- -- job_cluster.return_value.scatter.return_value = {} -- job_cluster.return_value.submit = submit -- -- formatter_context.return_value = {node_id: {}} -- -- # Configuring available workers -- job_cluster.return_value.scheduler_info.return_value = \ -- { -- 'workers': { -- 'tcp://{0}'.format(settings.MASTER_IP): {}, -- 'tcp://192.168.0.1:33334': {}, -- 'tcp://127.0.0.2:33335': {}, -- } -- } -- -- # Performing serialization -- lcm.TransactionSerializer.serialize( -- context, [task], self.resolver -- ) -- -- # Checking data is scattered only to expected workers -- scatter.assert_called_once() -- scatter.assert_called_with( -- {'context': context, 'settings_config': settings.config}, -- broadcast=True, -- workers=[settings.MASTER_IP] -- ) -- -- # Checking submit job only to expected workers -- submit.assert_called_once() -- serializer = lcm.transaction_serializer -- submit.assert_called_with( -- serializer._distributed_serialize_tasks_for_node, -- {node_id: formatter_context()}, -- ((node_id, task),), -- job_cluster().scatter(), -- workers=set([settings.MASTER_IP]) -- ) -- -- def test_distributed_serialization_get_allowed_nodes_ips(self): -- policy = lcm.transaction_serializer.DistributedProcessingPolicy() -- -- context_data = { -- 'common': { -- 'serialization_policy': -- consts.SERIALIZATION_POLICY.distributed, -- 'ds_use_error': True, -- 'ds_use_provisioned': True, -- 'ds_use_discover': True, -- 'ds_use_ready': False -- }, -- 'nodes': { -- '1': {'status': consts.NODE_STATUSES.error, -- 'ip': '10.20.0.3'}, -- '2': {'status': consts.NODE_STATUSES.provisioned, -- 'ip': '10.20.0.4'}, -- '3': {'status': consts.NODE_STATUSES.discover, -- 'ip': '10.20.0.5'}, -- '4': {'status': consts.NODE_STATUSES.ready, -- 'ip': '10.20.0.6'}, -- } -- } -- -- actual = policy._get_allowed_nodes_ips( -- TransactionContext(context_data)) -- self.assertItemsEqual( -- [settings.MASTER_IP, '10.20.0.3', '10.20.0.4', '10.20.0.5'], -- actual -- ) -- -- def test_distributed_serialization_get_allowed_nodes_statuses(self): -- policy = lcm.transaction_serializer.DistributedProcessingPolicy() -- context_data = {} -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- self.assertItemsEqual([], actual) -- -- context_data['common'] = { -- 'ds_use_discover': False, -- 'ds_use_provisioned': False, -- 'ds_use_error': False, -- 'ds_use_ready': False -- } -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- self.assertItemsEqual([], actual) -- -- context_data['common']['ds_use_discover'] = True -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- expected = [consts.NODE_STATUSES.discover] -- self.assertItemsEqual(expected, actual) -- -- context_data['common']['ds_use_provisioned'] = True -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- expected = [consts.NODE_STATUSES.discover, -- consts.NODE_STATUSES.provisioned] -- self.assertItemsEqual(expected, actual) -- -- context_data['common']['ds_use_error'] = True -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- expected = [consts.NODE_STATUSES.discover, -- consts.NODE_STATUSES.provisioned, -- consts.NODE_STATUSES.error] -- self.assertItemsEqual(expected, actual) -- -- context_data['common']['ds_use_ready'] = True -- actual = policy._get_allowed_nodes_statuses( -- TransactionContext(context_data)) -- expected = [consts.NODE_STATUSES.discover, -- consts.NODE_STATUSES.provisioned, -- consts.NODE_STATUSES.error, -- consts.NODE_STATUSES.ready] -- self.assertItemsEqual(expected, actual) -- -- def test_distributed_serialization_get_allowed_workers(self): -- policy = lcm.transaction_serializer.DistributedProcessingPolicy() -- -- with mock.patch( -- 'nailgun.lcm.transaction_serializer.distributed.Client' -- ) as job_cluster: -- job_cluster.scheduler_info.return_value = \ -- {'workers': { -- 'tcp://10.20.0.2:1': {}, -- 'tcp://10.20.0.2:2': {}, -- 'tcp://10.20.0.3:1': {}, -- 'tcp://10.20.0.3:2': {}, -- 'tcp://10.20.0.3:3': {}, -- 'tcp://10.20.0.4:1': {}, -- 'tcp://10.20.0.5:1': {} -- }} -- allowed_ips = set(['10.20.0.2', '10.20.0.3', '10.20.0.5']) -- -- expected = ['10.20.0.2:1', '10.20.0.2:2', '10.20.0.3:1', -- '10.20.0.3:2', '10.20.0.3:3', '10.20.0.5:1'] -- actual = policy._get_allowed_workers(job_cluster, allowed_ips) -- self.assertItemsEqual(expected, actual) -- -- def test_distributed_serialization_serialize_task(self): -- task = { -- 'id': 'task1', 'roles': ['controller'], -- 'type': 'puppet', 'version': '2.0.0', -- 'parameters': { -- 'master_ip': '{MN_IP}', -- 'host': {'yaql_exp': '$.public_ssl.hostname'}, -- 'attr': {'yaql_exp': '$node.attributes.a_str'} -- } -- } -- -- formatter_contexts_idx = { -- '1': {'MN_IP': '10.0.0.1'}, -- '2': {} -- } -- scattered_data = { -- 'settings_config': settings.config, -- 'context': self.context -- } -- -- serializer = lcm.transaction_serializer -- actual = serializer._distributed_serialize_tasks_for_node( -- formatter_contexts_idx, [('1', task), ('2', task)], scattered_data) -- -- expected = [ -- ( -- ( -- '1', -- { -- 'id': 'task1', -- 'type': 'puppet', -- 'parameters': { -- 'cwd': '/', -- 'master_ip': '10.0.0.1', -- 'host': 'localhost', -- 'attr': 'text1' -- }, -- 'fail_on_error': True -- } -- ), -- None -- ), -- ( -- ( -- '2', -- { -- 'id': 'task1', -- 'type': 'puppet', -- 'parameters': { -- 'cwd': '/', -- 'master_ip': '{MN_IP}', -- 'host': 'localhost', -- 'attr': 'text2' -- }, -- 'fail_on_error': True -- } -- ), -- None -- ) -- ] -- -- self.assertItemsEqual(expected, actual) -- -- def test_distributed_serialization_serialize_task_failure(self): -- task = { -- 'id': 'task1', 'roles': ['controller'], -- 'type': 'puppet', 'version': '2.0.0', -- 'parameters': { -- 'fake': {'yaql_exp': '$.some.fake_param'} -- } -- } -- -- formatter_contexts_idx = {'2': {}} -- scattered_data = { -- 'settings_config': settings.config, -- 'context': self.context -- } -- -- serializer = lcm.transaction_serializer -- result = serializer._distributed_serialize_tasks_for_node( -- formatter_contexts_idx, [('2', task)], scattered_data) -- (_, __), err = result[0] -- self.assertIsInstance(err, exceptions.KeyError) -- -- --class TestConcurrencyPolicy(BaseTestCase): -- -- @mock.patch( -- 'nailgun.lcm.transaction_serializer.multiprocessing.cpu_count', -- return_value=1 -- ) -- def test_one_cpu(self, cpu_count): -- policy = lcm.transaction_serializer.get_processing_policy( -- lcm.TransactionContext({})) -- self.assertIsInstance( -- policy, -- lcm.transaction_serializer.SingleWorkerConcurrencyPolicy -- ) -- self.assertTrue(cpu_count.is_called) -- -- @mock.patch( -- 'nailgun.lcm.transaction_serializer.multiprocessing.cpu_count', -- return_value=0 -- ) -- def test_zero_cpu(self, cpu_count): -- policy = lcm.transaction_serializer.get_processing_policy( -- lcm.TransactionContext({})) -- self.assertIsInstance( -- policy, -- lcm.transaction_serializer.SingleWorkerConcurrencyPolicy -- ) -- self.assertTrue(cpu_count.is_called) -- -- @mock.patch( -- 'nailgun.lcm.transaction_serializer.multiprocessing.cpu_count', -- side_effect=NotImplementedError -- ) -- def test_cpu_count_not_implemented(self, cpu_count): -- policy = lcm.transaction_serializer.get_processing_policy( -- lcm.TransactionContext({})) -- self.assertIsInstance( -- policy, -- lcm.transaction_serializer.SingleWorkerConcurrencyPolicy -- ) -- self.assertTrue(cpu_count.is_called) -- -- def test_distributed_serialization_enabled_in_cluster(self): -- context_data = {'common': { -- 'serialization_policy': consts.SERIALIZATION_POLICY.distributed -- }} -- policy = lcm.transaction_serializer.get_processing_policy( -- lcm.TransactionContext(context_data)) -- self.assertIsInstance( -- policy, -- lcm.transaction_serializer.DistributedProcessingPolicy -- ) -diff --git a/nailgun/requirements.txt b/nailgun/requirements.txt -index c702e8a..96bed25 100644 ---- a/nailgun/requirements.txt -+++ b/nailgun/requirements.txt -@@ -47,5 +47,3 @@ stevedore>=1.5.0 - # See: https://bugs.launchpad.net/fuel/+bug/1519727 - setuptools<=18.5 - yaql>=1.0.0 --# Distributed nodes serialization --distributed==1.16.0 |