Merge lp:~jameinel/maas/tag-updating into lp:maas/trunk

[John A Meinel]

Add a task for the provisioning_server that can update tags using the APIs we just added.

This allows us to farm out all the work for processing 100,000 tags into the provisioning_servers, which should allow us to consistently evaluate a Tag definition in under 10s, even as we scale.

Add a task for the provisioning_server that can update tags using the APIs we just added.

This uses a lot of mocking to make sure the calls are correct, but I'll still need to add testing on the other side that end-to-end actually works.

I can wait to land this until the next patch is complete if so desired, but I would like to get feedback on the mock work that was done here.

[lp:~jameinel/maas/tag-updating]

1141. By John A Meinel on 2012-10-04

Gavin confirmed that you just pass it as a python list.

1142. By John A Meinel on 2012-10-04

Revert the changes to the acceptance Makefile, as it was probably specific for Jelmer's machine.

[Martin Packman (gz)]

-include /etc/lsb-release
+DISTRIB_CODENAME=unstable

This seems unrelated, probably wants reverting.

+ # XXX: Check the response code before we parse the content

A helper in client that raised an exception on a non-2XX response would be good, can be added later.

The rest of the stuff I understand looks good. :)

review: Approve

[lp:~jameinel/maas/tag-updating]

1143. By John A Meinel on 2012-10-04

Simplify to a single upload_node_tags call.

The bandwidth required should be reasonable.

=== added file 'src/provisioningserver/tags.py'
--- src/provisioningserver/tags.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/tags.py	2012-10-04 09:47:27 +0000
@@ -0,0 +1,150 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Celery jobs for managing tags.
+
+"""
+
+import json
+from lxml import etree
+
+from apiclient.maas_client import (
+    MAASClient,
+    MAASDispatcher,
+    MAASOAuth,
+    )
+
+from provisioningserver.auth import (
+    get_recorded_api_credentials,
+    get_recorded_maas_url,
+    get_recorded_nodegroup_uuid,
+    )
+
+from provisioningserver.logging import task_logger
+
+
+DEFAULT_BATCH_SIZE = 100
+
+
+def get_cached_knowledge():
+    """Get all the information that we need to know, or raise an error.
+
+    :return: (client, nodegroup_uuid)
+    """
+    maas_url = get_recorded_maas_url()
+    if maas_url is None:
+        task_logger.error("Not updating tags: don't have API URL yet.")
+        return None, None
+    api_credentials = get_recorded_api_credentials()
+    if api_credentials is None:
+        task_logger.error("Not updating tags: don't have API key yet.")
+        return None, None
+    nodegroup_uuid = get_recorded_nodegroup_uuid()
+    if nodegroup_uuid is None:
+        task_logger.error("Not updating tags: don't have UUID yet.")
+        return None, None
+    client = MAASClient(MAASOAuth(*api_credentials), MAASDispatcher(),
+        maas_url)
+    return client, nodegroup_uuid
+
+
+def get_nodes_for_node_group(client, nodegroup_uuid):
+    """Retrieve the UUIDs of nodes in a particular group.
+
+    :param client: MAAS client instance
+    :param nodegroup_uuid: Node group for which to retrieve nodes
+    :return: List of UUIDs for nodes in nodegroup
+    """
+    path = 'api/1.0/nodegroup/%s/' % (nodegroup_uuid)
+    response = client.get(path, op='list_nodes')
+    # XXX: Check the response code before we parse the content
+    return json.loads(response.content)
+
+
+def get_hardware_details_for_nodes(client, nodegroup_uuid, system_ids):
+    """Retrieve the lshw output for a set of nodes.
+
+    :param client: MAAS client
+    :param system_ids: List of UUIDs of systems for which to fetch lshw data
+    :return: Dictionary mapping node UUIDs to lshw output
+    """
+    path = 'api/1.0/nodegroup/%s/' % (nodegroup_uuid,)
+    response = client.get(
+        path, op='node_hardware_details', system_ids=system_ids)
+    # XXX: Check the response code before we parse the content
+    return json.loads(response.content)
+
+
+def update_node_tags(client, tag_name, uuid, added, removed):
+    """Update the nodes relevant for a particular tag.
+
+    :param client: MAAS client
+    :param tag_name: Name of tag
+    :param uuid: NodeGroup uuid of this worker. Needed for security
+        permissions. (The nodegroup worker is only allowed to touch nodes in
+        its nodegroup, otherwise you need to be a superuser.)
+    :param added: Set of nodes to add
+    :param removed: Set of nodes to remove
+    """
+    path = 'api/1.0/tags/%s/' % (tag_name,)
+    response = client.post(path, op='update_nodes', add=added, remove=removed)
+    # XXX: Check the response code before we parse the content
+    return json.loads(response.content)
+
+
+def process_batch(xpath, hardware_details):
+    """Get the details for one batch, and process whether they match or not.
+    """
+    # Fetch node XML in batches
+    matched_nodes = []
+    unmatched_nodes = []
+    for system_id, hw_xml in hardware_details:
+        xml = etree.XML(hw_xml)
+        if xpath(xml):
+            matched_nodes.append(system_id)
+        else:
+            unmatched_nodes.append(system_id)
+    return matched_nodes, unmatched_nodes
+
+
+def process_all(client, tag_name, nodegroup_uuid, system_ids, xpath,
+                batch_size=None):
+    if batch_size is None:
+        batch_size = DEFAULT_BATCH_SIZE
+    all_matched = []
+    all_unmatched = []
+    for i in range(0, len(system_ids), batch_size):
+        selected_ids = system_ids[i:i + batch_size]
+        details = get_hardware_details_for_nodes(
+            client, nodegroup_uuid, selected_ids)
+        matched, unmatched = process_batch(xpath, details)
+        all_matched.extend(matched)
+        all_unmatched.extend(unmatched)
+    # Upload all updates for one nodegroup at one time. This should be no more
+    # than ~41*10,000 = 410kB. That should take <1s even on a 10Mbit network.
+    # This also allows us to track if a nodegroup has been processed in the DB,
+    # without having to add another API call.
+    update_node_tags(client, tag_name, nodegroup_uuid, all_matched,
+                     all_unmatched)
+
+
+def process_node_tags(tag_name, tag_definition, batch_size=None):
+    """Update the nodes for a new/changed tag definition.
+
+    :param tag_name: Name of the tag to update nodes for
+    :param tag_definition: Tag definition
+    :param batch_size: Size of batch
+    """
+    client, nodegroup_uuid = get_cached_knowledge()
+    if not all([client, nodegroup_uuid]):
+        task_logger.error('Unable to update tag: %s for definition %r'
+            ' please refresh secrets, then rebuild this tag'
+            % (tag_name, tag_definition))
+        return
+    # We evaluate this early, so we can fail before sending a bunch of data to
+    # the server
+    xpath = etree.XPath(tag_definition)
+    # Get nodes to process
+    system_ids = get_nodes_for_node_group(client, nodegroup_uuid)
+    process_all(client, tag_name, nodegroup_uuid, system_ids, xpath,
+                batch_size=batch_size)

=== modified file 'src/provisioningserver/tasks.py'
--- src/provisioningserver/tasks.py	2012-09-29 23:49:08 +0000
+++ src/provisioningserver/tasks.py	2012-10-04 09:47:27 +0000
@@ -30,7 +30,10 @@
 
 from celery.app import app_or_default
 from celery.task import task
-from provisioningserver import boot_images
+from provisioningserver import (
+    boot_images,
+    tags,
+    )
 from provisioningserver.auth import (
     record_api_credentials,
     record_maas_url,
@@ -335,3 +338,13 @@
 def report_boot_images():
     """For master worker only: report available netboot images."""
     boot_images.report_to_server()
+
+
+@task
+def update_node_tags(tag_name, tag_definition):
+    """Update the nodes for a new/changed tag definition.
+
+    :param tag_name: Name of the tag to update nodes for
+    :param tag_definition: Tag definition
+    """
+    tags.process_node_tags(tag_name, tag_definition)

=== modified file 'src/provisioningserver/testing/testcase.py'
--- src/provisioningserver/testing/testcase.py	2012-08-24 02:51:42 +0000
+++ src/provisioningserver/testing/testcase.py	2012-10-04 09:47:27 +0000
@@ -14,7 +14,14 @@
     'PservTestCase',
     ]
 
+from apiclient.testing.credentials import make_api_credentials
 from maastesting import testcase
+from maastesting.factory import factory
+from provisioningserver.auth import (
+    record_api_credentials,
+    record_maas_url,
+    record_nodegroup_uuid,
+    )
 from provisioningserver.testing.worker_cache import WorkerCacheFixture
 
 
@@ -23,3 +30,22 @@
     def setUp(self):
         super(PservTestCase, self).setUp()
         self.useFixture(WorkerCacheFixture())
+
+    def make_maas_url(self):
+        return 'http://127.0.0.1/%s' % factory.make_name('path')
+
+    def set_maas_url(self):
+        record_maas_url(self.make_maas_url())
+
+    def set_api_credentials(self):
+        record_api_credentials(':'.join(make_api_credentials()))
+
+    def set_node_group_uuid(self):
+        nodegroup_uuid = factory.make_name('nodegroupuuid')
+        record_nodegroup_uuid(nodegroup_uuid)
+
+    def set_secrets(self):
+        """Setup all secrets that we would get from refresh_secrets."""
+        self.set_maas_url()
+        self.set_api_credentials()
+        self.set_node_group_uuid()

=== modified file 'src/provisioningserver/tests/test_boot_images.py'
--- src/provisioningserver/tests/test_boot_images.py	2012-09-25 14:45:59 +0000
+++ src/provisioningserver/tests/test_boot_images.py	2012-10-04 09:47:27 +0000
@@ -15,14 +15,8 @@
 import json
 
 from apiclient.maas_client import MAASClient
-from apiclient.testing.credentials import make_api_credentials
-from maastesting.factory import factory
 from mock import Mock
 from provisioningserver import boot_images
-from provisioningserver.auth import (
-    record_api_credentials,
-    record_maas_url,
-    )
 from provisioningserver.pxe import tftppath
 from provisioningserver.testing.boot_images import make_boot_image_params
 from provisioningserver.testing.config import ConfigFixture
@@ -35,13 +29,6 @@
         super(TestBootImagesTasks, self).setUp()
         self.useFixture(ConfigFixture({'tftp': {'root': self.make_dir()}}))
 
-    def set_maas_url(self):
-        record_maas_url(
-            'http://127.0.0.1/%s' % factory.make_name('path'))
-
-    def set_api_credentials(self):
-        record_api_credentials(':'.join(make_api_credentials()))
-
     def test_sends_boot_images_to_server(self):
         self.set_maas_url()
         self.set_api_credentials()

=== added file 'src/provisioningserver/tests/test_tags.py'
--- src/provisioningserver/tests/test_tags.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/tests/test_tags.py	2012-10-04 09:47:27 +0000
@@ -0,0 +1,181 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for tag updating."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+__metaclass__ = type
+__all__ = []
+
+from apiclient.maas_client import MAASClient
+import httplib
+from lxml import etree
+from maastesting.factory import factory
+from maastesting.fakemethod import (
+    FakeMethod,
+    MultiFakeMethod,
+    )
+from mock import MagicMock
+from provisioningserver.auth import (
+    get_recorded_nodegroup_uuid,
+    )
+from provisioningserver.testing.testcase import PservTestCase
+from provisioningserver import tags
+
+
+class FakeResponse:
+
+    def __init__(self, status_code, content):
+        self.status_code = status_code
+        self.content = content
+
+
+class TestTagUpdating(PservTestCase):
+
+    def test_get_cached_knowledge_knows_nothing(self):
+        # If we haven't given it any secrets, we should get back nothing
+        self.assertEqual((None, None), tags.get_cached_knowledge())
+
+    def test_get_cached_knowledge_with_only_url(self):
+        self.set_maas_url()
+        self.assertEqual((None, None), tags.get_cached_knowledge())
+
+    def test_get_cached_knowledge_with_only_url_creds(self):
+        self.set_maas_url()
+        self.set_api_credentials()
+        self.assertEqual((None, None), tags.get_cached_knowledge())
+
+    def test_get_cached_knowledge_with_all_info(self):
+        self.set_maas_url()
+        self.set_api_credentials()
+        self.set_node_group_uuid()
+        client, uuid = tags.get_cached_knowledge()
+        self.assertIsNot(None, client)
+        self.assertIsInstance(client, MAASClient)
+        self.assertIsNot(None, uuid)
+        self.assertEqual(get_recorded_nodegroup_uuid(), uuid)
+
+    def fake_client(self):
+        return MAASClient(None, None, self.make_maas_url())
+
+    def fake_cached_knowledge(self):
+        nodegroup_uuid = factory.make_name('nodegroupuuid')
+        return self.fake_client(), nodegroup_uuid
+
+    def test_get_nodes_calls_correct_api_and_parses_result(self):
+        client, uuid = self.fake_cached_knowledge()
+        response = FakeResponse(httplib.OK, '["system-id1", "system-id2"]')
+        mock = MagicMock(return_value=response)
+        self.patch(client, 'get', mock)
+        result = tags.get_nodes_for_node_group(client, uuid)
+        self.assertEqual(['system-id1', 'system-id2'], result)
+        url = 'api/1.0/nodegroup/%s/' % (uuid,)
+        mock.assert_called_once_with(url, op='list_nodes')
+
+    def test_get_hardware_details_calls_correct_api_and_parses_result(self):
+        client, uuid = self.fake_cached_knowledge()
+        xml_data = "<test><data /></test>"
+        content = '[["system-id1", "%s"]]' % (xml_data,)
+        response = FakeResponse(httplib.OK, content)
+        mock = MagicMock(return_value=response)
+        self.patch(client, 'get', mock)
+        result = tags.get_hardware_details_for_nodes(
+            client, uuid, ['system-id1', 'system-id2'])
+        self.assertEqual([['system-id1', xml_data]], result)
+        url = 'api/1.0/nodegroup/%s/' % (uuid,)
+        mock.assert_called_once_with(
+            url, op='node_hardware_details',
+            system_ids=["system-id1", "system-id2"])
+
+    def test_update_node_tags_calls_correct_api_and_parses_result(self):
+        client, uuid = self.fake_cached_knowledge()
+        content = '{"added": 1, "removed": 2}'
+        response = FakeResponse(httplib.OK, content)
+        mock = MagicMock(return_value=response)
+        self.patch(client, 'post', mock)
+        name = factory.make_name('tag')
+        result = tags.update_node_tags(client, name, uuid,
+            ['add-system-id'], ['remove-1', 'remove-2'])
+        self.assertEqual({'added': 1, 'removed': 2}, result)
+        url = 'api/1.0/tags/%s/' % (name,)
+        mock.assert_called_once_with(
+            url, op='update_nodes',
+            add=['add-system-id'], remove=['remove-1', 'remove-2'])
+
+    def test_process_batch_evaluates_xpath(self):
+        # Yay, something that doesn't need patching...
+        xpath = etree.XPath('//node')
+        node_details = [['a', '<node />'],
+                        ['b', '<not-node />'],
+                        ['c', '<parent><node /></parent>'],
+                       ]
+        self.assertEqual(
+            (['a', 'c'], ['b']),
+            tags.process_batch(xpath, node_details))
+
+    def test_process_node_tags_no_secrets(self):
+        self.patch(MAASClient, 'get')
+        self.patch(MAASClient, 'post')
+        tag_name = factory.make_name('tag')
+        tags.process_node_tags(tag_name, '//node')
+        self.assertFalse(MAASClient.get.called)
+        self.assertFalse(MAASClient.post.called)
+
+    def test_process_node_tags_integration(self):
+        self.set_secrets()
+        get_nodes = FakeMethod(
+            result=FakeResponse(httplib.OK, '["system-id1", "system-id2"]'))
+        get_hw_details = FakeMethod(
+            result=FakeResponse(httplib.OK,
+                '[["system-id1", "<node />"], ["system-id2", "<no-node />"]]'))
+        get_fake = MultiFakeMethod([get_nodes, get_hw_details])
+        post_fake = FakeMethod(
+            result=FakeResponse(httplib.OK, '{"added": 1, "removed": 1}'))
+        self.patch(MAASClient, 'get', get_fake)
+        self.patch(MAASClient, 'post', post_fake)
+        tag_name = factory.make_name('tag')
+        nodegroup_uuid = get_recorded_nodegroup_uuid()
+        tags.process_node_tags(tag_name, '//node')
+        nodegroup_url = 'api/1.0/nodegroup/%s/' % (nodegroup_uuid,)
+        tag_url = 'api/1.0/tags/%s/' % (tag_name,)
+        self.assertEqual([((nodegroup_url,), {'op': 'list_nodes'})],
+                         get_nodes.calls)
+        self.assertEqual([((nodegroup_url,),
+                          {'op': 'node_hardware_details',
+                           'system_ids': ['system-id1', 'system-id2']})],
+                         get_hw_details.calls)
+        self.assertEqual([((tag_url,),
+                          {'op': 'update_nodes',
+                           'add': ['system-id1'],
+                           'remove': ['system-id2'],
+                          })], post_fake.calls)
+
+    def test_process_node_tags_requests_details_in_batches(self):
+        client = object()
+        uuid = factory.make_name('nodegroupuuid')
+        self.patch(
+            tags, 'get_cached_knowledge',
+            MagicMock(return_value=(client, uuid)))
+        self.patch(
+            tags, 'get_nodes_for_node_group',
+            MagicMock(return_value=['a', 'b', 'c']))
+        fake_first = FakeMethod(
+            result=[['a', '<node />'], ['b', '<not-node />']])
+        fake_second = FakeMethod(
+            result=[['c', '<parent><node /></parent>']])
+        self.patch(tags, 'get_hardware_details_for_nodes',
+            MultiFakeMethod([fake_first, fake_second]))
+        self.patch(tags, 'update_node_tags')
+        tag_name = factory.make_name('tag')
+        tags.process_node_tags(tag_name, '//node', batch_size=2)
+        tags.get_cached_knowledge.assert_called_once_with()
+        tags.get_nodes_for_node_group.assert_called_once_with(client, uuid)
+        self.assertEqual([((client, uuid, ['a', 'b']), {})], fake_first.calls)
+        self.assertEqual([((client, uuid, ['c']), {})], fake_second.calls)
+        tags.update_node_tags.assert_called_once_with(
+            client, tag_name, uuid, ['a', 'c'], ['b'])