Revert "Revert: Blueprint "bp/distributed-serialization""danube.1.RC1danube.1.0

This reverts commit 8b28c44a4be6beef8a71e672f543001bcc0a080f.
The issue has been fixed in upstream.

Change-Id: Id4ad8fb7b1ec887f8d8d1d5b18a9b33a73bda10c
Signed-off-by: Michael Polenchuk <mpolenchuk@mirantis.com>
(cherry picked from commit 4a360c5076f839c8b700638f9b85c3e630894826)

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