OpenStack Compute (Nova)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

2112 lines
97 KiB

  1. # Copyright (c) 2014 Red Hat, Inc.
  2. # All Rights Reserved.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  5. # not use this file except in compliance with the License. You may obtain
  6. # a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13. # License for the specific language governing permissions and limitations
  14. # under the License.
  15. import collections
  16. import contextlib
  17. import copy
  18. import functools
  19. import random
  20. import re
  21. import retrying
  22. import time
  23. from keystoneauth1 import exceptions as ks_exc
  24. import os_traits
  25. from oslo_log import log as logging
  26. from oslo_middleware import request_id
  27. from oslo_utils import versionutils
  28. from nova.compute import provider_tree
  29. from nova.compute import utils as compute_utils
  30. import nova.conf
  31. from nova import exception
  32. from nova.i18n import _
  33. from nova import objects
  34. from nova import rc_fields as fields
  35. from nova.scheduler import utils as scheduler_utils
  36. from nova import utils
  37. CONF = nova.conf.CONF
  38. LOG = logging.getLogger(__name__)
  39. VCPU = fields.ResourceClass.VCPU
  40. MEMORY_MB = fields.ResourceClass.MEMORY_MB
  41. DISK_GB = fields.ResourceClass.DISK_GB
  42. _RE_INV_IN_USE = re.compile("Inventory for (.+) on resource provider "
  43. "(.+) in use")
  44. WARN_EVERY = 10
  45. PLACEMENT_CLIENT_SEMAPHORE = 'placement_client'
  46. CONSUMER_GENERATION_VERSION = '1.28'
  47. GRANULAR_AC_VERSION = '1.25'
  48. ALLOW_RESERVED_EQUAL_TOTAL_INVENTORY_VERSION = '1.26'
  49. POST_RPS_RETURNS_PAYLOAD_API_VERSION = '1.20'
  50. AGGREGATE_GENERATION_VERSION = '1.19'
  51. NESTED_PROVIDER_API_VERSION = '1.14'
  52. POST_ALLOCATIONS_API_VERSION = '1.13'
  53. AggInfo = collections.namedtuple('AggInfo', ['aggregates', 'generation'])
  54. TraitInfo = collections.namedtuple('TraitInfo', ['traits', 'generation'])
  55. def warn_limit(self, msg):
  56. if self._warn_count:
  57. self._warn_count -= 1
  58. else:
  59. self._warn_count = WARN_EVERY
  60. LOG.warning(msg)
  61. def safe_connect(f):
  62. @functools.wraps(f)
  63. def wrapper(self, *a, **k):
  64. try:
  65. return f(self, *a, **k)
  66. except ks_exc.EndpointNotFound:
  67. warn_limit(
  68. self, 'The placement API endpoint was not found.')
  69. # Reset client session so there is a new catalog, which
  70. # gets cached when keystone is first successfully contacted.
  71. self._client = self._create_client()
  72. except ks_exc.MissingAuthPlugin:
  73. warn_limit(
  74. self, 'No authentication information found for placement API.')
  75. except ks_exc.Unauthorized:
  76. warn_limit(
  77. self, 'Placement service credentials do not work.')
  78. except ks_exc.DiscoveryFailure:
  79. # TODO(_gryf): Looks like DiscoveryFailure is not the only missing
  80. # exception here. In Pike we should take care about keystoneauth1
  81. # failures handling globally.
  82. warn_limit(self,
  83. 'Discovering suitable URL for placement API failed.')
  84. except ks_exc.ConnectFailure:
  85. LOG.warning('Placement API service is not responding.')
  86. return wrapper
  87. class Retry(Exception):
  88. def __init__(self, operation, reason):
  89. self.operation = operation
  90. self.reason = reason
  91. def retries(f):
  92. """Decorator to retry a call three times if it raises Retry
  93. Note that this returns the actual value of the inner call on success
  94. or returns False if all the retries fail.
  95. """
  96. @functools.wraps(f)
  97. def wrapper(self, *a, **k):
  98. for retry in range(0, 4):
  99. try:
  100. sleep_time = random.uniform(0, retry * 2)
  101. time.sleep(sleep_time)
  102. return f(self, *a, **k)
  103. except Retry as e:
  104. LOG.debug(
  105. 'Unable to %(op)s because %(reason)s; retrying...',
  106. {'op': e.operation, 'reason': e.reason})
  107. LOG.error('Failed scheduler client operation %s: out of retries',
  108. f.__name__)
  109. return False
  110. return wrapper
  111. def _compute_node_to_inventory_dict(compute_node):
  112. """Given a supplied `objects.ComputeNode` object, return a dict, keyed
  113. by resource class, of various inventory information.
  114. :param compute_node: `objects.ComputeNode` object to translate
  115. """
  116. result = {}
  117. # NOTE(jaypipes): Ironic virt driver will return 0 values for vcpus,
  118. # memory_mb and disk_gb if the Ironic node is not available/operable
  119. if compute_node.vcpus > 0:
  120. result[VCPU] = {
  121. 'total': compute_node.vcpus,
  122. 'reserved': CONF.reserved_host_cpus,
  123. 'min_unit': 1,
  124. 'max_unit': compute_node.vcpus,
  125. 'step_size': 1,
  126. 'allocation_ratio': compute_node.cpu_allocation_ratio,
  127. }
  128. if compute_node.memory_mb > 0:
  129. result[MEMORY_MB] = {
  130. 'total': compute_node.memory_mb,
  131. 'reserved': CONF.reserved_host_memory_mb,
  132. 'min_unit': 1,
  133. 'max_unit': compute_node.memory_mb,
  134. 'step_size': 1,
  135. 'allocation_ratio': compute_node.ram_allocation_ratio,
  136. }
  137. if compute_node.local_gb > 0:
  138. # TODO(johngarbutt) We should either move to reserved_host_disk_gb
  139. # or start tracking DISK_MB.
  140. reserved_disk_gb = compute_utils.convert_mb_to_ceil_gb(
  141. CONF.reserved_host_disk_mb)
  142. result[DISK_GB] = {
  143. 'total': compute_node.local_gb,
  144. 'reserved': reserved_disk_gb,
  145. 'min_unit': 1,
  146. 'max_unit': compute_node.local_gb,
  147. 'step_size': 1,
  148. 'allocation_ratio': compute_node.disk_allocation_ratio,
  149. }
  150. return result
  151. def _instance_to_allocations_dict(instance):
  152. """Given an `objects.Instance` object, return a dict, keyed by resource
  153. class of the amount used by the instance.
  154. :param instance: `objects.Instance` object to translate
  155. """
  156. alloc_dict = scheduler_utils.resources_from_flavor(instance,
  157. instance.flavor)
  158. # Remove any zero allocations.
  159. return {key: val for key, val in alloc_dict.items() if val}
  160. def _move_operation_alloc_request(source_allocs, dest_alloc_req):
  161. """Given existing allocations for a source host and a new allocation
  162. request for a destination host, return a new allocation_request that
  163. contains resources claimed against both source and destination, accounting
  164. for shared providers.
  165. Also accounts for a resize to the same host where the source and dest
  166. compute node resource providers are going to be the same. In that case
  167. we sum the resource allocations for the single provider.
  168. :param source_allocs: Dict, keyed by resource provider UUID, of resources
  169. allocated on the source host
  170. :param dest_alloc_req: The allocation_request for resources against the
  171. destination host
  172. """
  173. LOG.debug("Doubling-up allocation_request for move operation.")
  174. # Remove any allocations against resource providers that are
  175. # already allocated against on the source host (like shared storage
  176. # providers)
  177. cur_rp_uuids = set(source_allocs.keys())
  178. new_rp_uuids = set(dest_alloc_req['allocations']) - cur_rp_uuids
  179. current_allocs = {
  180. cur_rp_uuid: {'resources': alloc['resources']}
  181. for cur_rp_uuid, alloc in source_allocs.items()
  182. }
  183. new_alloc_req = {'allocations': current_allocs}
  184. for rp_uuid in dest_alloc_req['allocations']:
  185. if rp_uuid in new_rp_uuids:
  186. new_alloc_req['allocations'][rp_uuid] = dest_alloc_req[
  187. 'allocations'][rp_uuid]
  188. elif not new_rp_uuids:
  189. # If there are no new_rp_uuids that means we're resizing to
  190. # the same host so we need to sum the allocations for
  191. # the compute node (and possibly shared providers) using both
  192. # the current and new allocations.
  193. # Note that we sum the allocations rather than take the max per
  194. # resource class between the current and new allocations because
  195. # the compute node/resource tracker is going to adjust for
  196. # decrementing any old allocations as necessary, the scheduler
  197. # shouldn't make assumptions about that.
  198. scheduler_utils.merge_resources(
  199. new_alloc_req['allocations'][rp_uuid]['resources'],
  200. dest_alloc_req['allocations'][rp_uuid]['resources'])
  201. LOG.debug("New allocation_request containing both source and "
  202. "destination hosts in move operation: %s", new_alloc_req)
  203. return new_alloc_req
  204. def _extract_inventory_in_use(body):
  205. """Given an HTTP response body, extract the resource classes that were
  206. still in use when we tried to delete inventory.
  207. :returns: String of resource classes or None if there was no InventoryInUse
  208. error in the response body.
  209. """
  210. match = _RE_INV_IN_USE.search(body)
  211. if match:
  212. return match.group(1)
  213. return None
  214. def get_placement_request_id(response):
  215. if response is not None:
  216. return response.headers.get(request_id.HTTP_RESP_HEADER_REQUEST_ID)
  217. class SchedulerReportClient(object):
  218. """Client class for updating the scheduler."""
  219. def __init__(self, adapter=None):
  220. """Initialize the report client.
  221. :param adapter: A prepared keystoneauth1 Adapter for API communication.
  222. If unspecified, one is created based on config options in the
  223. [placement] section.
  224. """
  225. self._adapter = adapter
  226. # An object that contains a nova-compute-side cache of resource
  227. # provider and inventory information
  228. self._provider_tree = provider_tree.ProviderTree()
  229. # Track the last time we updated providers' aggregates and traits
  230. self._association_refresh_time = {}
  231. self._client = self._create_client()
  232. # NOTE(danms): Keep track of how naggy we've been
  233. self._warn_count = 0
  234. @utils.synchronized(PLACEMENT_CLIENT_SEMAPHORE)
  235. def _create_client(self):
  236. """Create the HTTP session accessing the placement service."""
  237. # Flush provider tree and associations so we start from a clean slate.
  238. self._provider_tree = provider_tree.ProviderTree()
  239. self._association_refresh_time = {}
  240. client = self._adapter or utils.get_ksa_adapter('placement')
  241. # Set accept header on every request to ensure we notify placement
  242. # service of our response body media type preferences.
  243. client.additional_headers = {'accept': 'application/json'}
  244. return client
  245. def get(self, url, version=None, global_request_id=None):
  246. headers = ({request_id.INBOUND_HEADER: global_request_id}
  247. if global_request_id else {})
  248. return self._client.get(url, microversion=version, headers=headers)
  249. def post(self, url, data, version=None, global_request_id=None):
  250. headers = ({request_id.INBOUND_HEADER: global_request_id}
  251. if global_request_id else {})
  252. # NOTE(sdague): using json= instead of data= sets the
  253. # media type to application/json for us. Placement API is
  254. # more sensitive to this than other APIs in the OpenStack
  255. # ecosystem.
  256. return self._client.post(url, json=data, microversion=version,
  257. headers=headers)
  258. def put(self, url, data, version=None, global_request_id=None):
  259. # NOTE(sdague): using json= instead of data= sets the
  260. # media type to application/json for us. Placement API is
  261. # more sensitive to this than other APIs in the OpenStack
  262. # ecosystem.
  263. kwargs = {'microversion': version,
  264. 'headers': {request_id.INBOUND_HEADER:
  265. global_request_id} if global_request_id else {}}
  266. if data is not None:
  267. kwargs['json'] = data
  268. return self._client.put(url, **kwargs)
  269. def delete(self, url, version=None, global_request_id=None):
  270. headers = ({request_id.INBOUND_HEADER: global_request_id}
  271. if global_request_id else {})
  272. return self._client.delete(url, microversion=version, headers=headers)
  273. @safe_connect
  274. def get_allocation_candidates(self, context, resources):
  275. """Returns a tuple of (allocation_requests, provider_summaries,
  276. allocation_request_version).
  277. The allocation_requests are a collection of potential JSON objects that
  278. can be passed to the PUT /allocations/{consumer_uuid} Placement REST
  279. API to claim resources against one or more resource providers that meet
  280. the requested resource constraints.
  281. The provider summaries is a dict, keyed by resource provider UUID, of
  282. inventory and capacity information and traits for any resource
  283. provider involved in the allocation_requests.
  284. :returns: A tuple with a list of allocation_request dicts, a dict of
  285. provider information, and the microversion used to request
  286. this data from placement, or (None, None, None) if the
  287. request failed
  288. :param context: The security context
  289. :param nova.scheduler.utils.ResourceRequest resources:
  290. A ResourceRequest object representing the requested resources,
  291. traits, and aggregates from the request spec.
  292. Example member_of (aggregates) value in resources:
  293. [('foo', 'bar'), ('baz',)]
  294. translates to:
  295. "Candidates are in either 'foo' or 'bar', but definitely in 'baz'"
  296. """
  297. version = GRANULAR_AC_VERSION
  298. qparams = resources.to_querystring()
  299. url = "/allocation_candidates?%s" % qparams
  300. resp = self.get(url, version=version,
  301. global_request_id=context.global_id)
  302. if resp.status_code == 200:
  303. data = resp.json()
  304. return (data['allocation_requests'], data['provider_summaries'],
  305. version)
  306. args = {
  307. 'resource_request': str(resources),
  308. 'status_code': resp.status_code,
  309. 'err_text': resp.text,
  310. }
  311. msg = ("Failed to retrieve allocation candidates from placement "
  312. "API for filters: %(resource_request)s\n"
  313. "Got %(status_code)d: %(err_text)s.")
  314. LOG.error(msg, args)
  315. return None, None, None
  316. @safe_connect
  317. def _get_provider_aggregates(self, context, rp_uuid):
  318. """Queries the placement API for a resource provider's aggregates.
  319. :param rp_uuid: UUID of the resource provider to grab aggregates for.
  320. :return: A namedtuple comprising:
  321. * .aggregates: A set() of string aggregate UUIDs, which may
  322. be empty if the specified provider is associated with no
  323. aggregates.
  324. * .generation: The resource provider generation.
  325. :raise: ResourceProviderAggregateRetrievalFailed on errors. In
  326. particular, we raise this exception (as opposed to returning
  327. None or the empty set()) if the specified resource provider
  328. does not exist.
  329. """
  330. resp = self.get("/resource_providers/%s/aggregates" % rp_uuid,
  331. version=AGGREGATE_GENERATION_VERSION,
  332. global_request_id=context.global_id)
  333. if resp.status_code == 200:
  334. data = resp.json()
  335. return AggInfo(aggregates=set(data['aggregates']),
  336. generation=data['resource_provider_generation'])
  337. placement_req_id = get_placement_request_id(resp)
  338. msg = ("[%(placement_req_id)s] Failed to retrieve aggregates from "
  339. "placement API for resource provider with UUID %(uuid)s. "
  340. "Got %(status_code)d: %(err_text)s.")
  341. args = {
  342. 'placement_req_id': placement_req_id,
  343. 'uuid': rp_uuid,
  344. 'status_code': resp.status_code,
  345. 'err_text': resp.text,
  346. }
  347. LOG.error(msg, args)
  348. raise exception.ResourceProviderAggregateRetrievalFailed(uuid=rp_uuid)
  349. @safe_connect
  350. def _get_provider_traits(self, context, rp_uuid):
  351. """Queries the placement API for a resource provider's traits.
  352. :param context: The security context
  353. :param rp_uuid: UUID of the resource provider to grab traits for.
  354. :return: A namedtuple comprising:
  355. * .traits: A set() of string trait names, which may be
  356. empty if the specified provider has no traits.
  357. * .generation: The resource provider generation.
  358. :raise: ResourceProviderTraitRetrievalFailed on errors. In particular,
  359. we raise this exception (as opposed to returning None or the
  360. empty set()) if the specified resource provider does not exist.
  361. """
  362. resp = self.get("/resource_providers/%s/traits" % rp_uuid,
  363. version='1.6', global_request_id=context.global_id)
  364. if resp.status_code == 200:
  365. json = resp.json()
  366. return TraitInfo(traits=set(json['traits']),
  367. generation=json['resource_provider_generation'])
  368. placement_req_id = get_placement_request_id(resp)
  369. LOG.error(
  370. "[%(placement_req_id)s] Failed to retrieve traits from "
  371. "placement API for resource provider with UUID %(uuid)s. Got "
  372. "%(status_code)d: %(err_text)s.",
  373. {'placement_req_id': placement_req_id, 'uuid': rp_uuid,
  374. 'status_code': resp.status_code, 'err_text': resp.text})
  375. raise exception.ResourceProviderTraitRetrievalFailed(uuid=rp_uuid)
  376. @safe_connect
  377. def _get_resource_provider(self, context, uuid):
  378. """Queries the placement API for a resource provider record with the
  379. supplied UUID.
  380. :param context: The security context
  381. :param uuid: UUID identifier for the resource provider to look up
  382. :return: A dict of resource provider information if found or None if no
  383. such resource provider could be found.
  384. :raise: ResourceProviderRetrievalFailed on error.
  385. """
  386. resp = self.get("/resource_providers/%s" % uuid,
  387. version=NESTED_PROVIDER_API_VERSION,
  388. global_request_id=context.global_id)
  389. if resp.status_code == 200:
  390. data = resp.json()
  391. return data
  392. elif resp.status_code == 404:
  393. return None
  394. else:
  395. placement_req_id = get_placement_request_id(resp)
  396. msg = ("[%(placement_req_id)s] Failed to retrieve resource "
  397. "provider record from placement API for UUID %(uuid)s. Got "
  398. "%(status_code)d: %(err_text)s.")
  399. args = {
  400. 'uuid': uuid,
  401. 'status_code': resp.status_code,
  402. 'err_text': resp.text,
  403. 'placement_req_id': placement_req_id,
  404. }
  405. LOG.error(msg, args)
  406. raise exception.ResourceProviderRetrievalFailed(uuid=uuid)
  407. @safe_connect
  408. def _get_sharing_providers(self, context, agg_uuids):
  409. """Queries the placement API for a list of the resource providers
  410. associated with any of the specified aggregates and possessing the
  411. MISC_SHARES_VIA_AGGREGATE trait.
  412. :param context: The security context
  413. :param agg_uuids: Iterable of string UUIDs of aggregates to filter on.
  414. :return: A list of dicts of resource provider information, which may be
  415. empty if no provider exists with the specified UUID.
  416. :raise: ResourceProviderRetrievalFailed on error.
  417. """
  418. if not agg_uuids:
  419. return []
  420. aggs = ','.join(agg_uuids)
  421. url = "/resource_providers?member_of=in:%s&required=%s" % (
  422. aggs, os_traits.MISC_SHARES_VIA_AGGREGATE)
  423. resp = self.get(url, version='1.18',
  424. global_request_id=context.global_id)
  425. if resp.status_code == 200:
  426. return resp.json()['resource_providers']
  427. msg = _("[%(placement_req_id)s] Failed to retrieve sharing resource "
  428. "providers associated with the following aggregates from "
  429. "placement API: %(aggs)s. Got %(status_code)d: %(err_text)s.")
  430. args = {
  431. 'aggs': aggs,
  432. 'status_code': resp.status_code,
  433. 'err_text': resp.text,
  434. 'placement_req_id': get_placement_request_id(resp),
  435. }
  436. LOG.error(msg, args)
  437. raise exception.ResourceProviderRetrievalFailed(message=msg % args)
  438. @safe_connect
  439. def _get_providers_in_tree(self, context, uuid):
  440. """Queries the placement API for a list of the resource providers in
  441. the tree associated with the specified UUID.
  442. :param context: The security context
  443. :param uuid: UUID identifier for the resource provider to look up
  444. :return: A list of dicts of resource provider information, which may be
  445. empty if no provider exists with the specified UUID.
  446. :raise: ResourceProviderRetrievalFailed on error.
  447. """
  448. resp = self.get("/resource_providers?in_tree=%s" % uuid,
  449. version=NESTED_PROVIDER_API_VERSION,
  450. global_request_id=context.global_id)
  451. if resp.status_code == 200:
  452. return resp.json()['resource_providers']
  453. # Some unexpected error
  454. placement_req_id = get_placement_request_id(resp)
  455. msg = ("[%(placement_req_id)s] Failed to retrieve resource provider "
  456. "tree from placement API for UUID %(uuid)s. Got "
  457. "%(status_code)d: %(err_text)s.")
  458. args = {
  459. 'uuid': uuid,
  460. 'status_code': resp.status_code,
  461. 'err_text': resp.text,
  462. 'placement_req_id': placement_req_id,
  463. }
  464. LOG.error(msg, args)
  465. raise exception.ResourceProviderRetrievalFailed(uuid=uuid)
  466. @safe_connect
  467. def _create_resource_provider(self, context, uuid, name,
  468. parent_provider_uuid=None):
  469. """Calls the placement API to create a new resource provider record.
  470. :param context: The security context
  471. :param uuid: UUID of the new resource provider
  472. :param name: Name of the resource provider
  473. :param parent_provider_uuid: Optional UUID of the immediate parent
  474. :return: A dict of resource provider information object representing
  475. the newly-created resource provider.
  476. :raise: ResourceProviderCreationFailed or
  477. ResourceProviderRetrievalFailed on error.
  478. """
  479. url = "/resource_providers"
  480. payload = {
  481. 'uuid': uuid,
  482. 'name': name,
  483. }
  484. if parent_provider_uuid is not None:
  485. payload['parent_provider_uuid'] = parent_provider_uuid
  486. # Bug #1746075: First try the microversion that returns the new
  487. # provider's payload.
  488. resp = self.post(url, payload,
  489. version=POST_RPS_RETURNS_PAYLOAD_API_VERSION,
  490. global_request_id=context.global_id)
  491. placement_req_id = get_placement_request_id(resp)
  492. if resp:
  493. msg = ("[%(placement_req_id)s] Created resource provider record "
  494. "via placement API for resource provider with UUID "
  495. "%(uuid)s and name %(name)s.")
  496. args = {
  497. 'uuid': uuid,
  498. 'name': name,
  499. 'placement_req_id': placement_req_id,
  500. }
  501. LOG.info(msg, args)
  502. return resp.json()
  503. # TODO(efried): Push error codes from placement, and use 'em.
  504. name_conflict = 'Conflicting resource provider name:'
  505. if resp.status_code == 409 and name_conflict not in resp.text:
  506. # Another thread concurrently created a resource provider with the
  507. # same UUID. Log a warning and then just return the resource
  508. # provider object from _get_resource_provider()
  509. msg = ("[%(placement_req_id)s] Another thread already created a "
  510. "resource provider with the UUID %(uuid)s. Grabbing that "
  511. "record from the placement API.")
  512. args = {
  513. 'uuid': uuid,
  514. 'placement_req_id': placement_req_id,
  515. }
  516. LOG.info(msg, args)
  517. return self._get_resource_provider(context, uuid)
  518. # A provider with the same *name* already exists, or some other error.
  519. msg = ("[%(placement_req_id)s] Failed to create resource provider "
  520. "record in placement API for UUID %(uuid)s. Got "
  521. "%(status_code)d: %(err_text)s.")
  522. args = {
  523. 'uuid': uuid,
  524. 'status_code': resp.status_code,
  525. 'err_text': resp.text,
  526. 'placement_req_id': placement_req_id,
  527. }
  528. LOG.error(msg, args)
  529. raise exception.ResourceProviderCreationFailed(name=name)
  530. def _ensure_resource_provider(self, context, uuid, name=None,
  531. parent_provider_uuid=None):
  532. """Ensures that the placement API has a record of a resource provider
  533. with the supplied UUID. If not, creates the resource provider record in
  534. the placement API for the supplied UUID, passing in a name for the
  535. resource provider.
  536. If found or created, the provider's UUID is returned from this method.
  537. If the resource provider for the supplied uuid was not found and the
  538. resource provider record could not be created in the placement API, an
  539. exception is raised.
  540. If this method returns successfully, callers are assured that the
  541. placement API contains a record of the provider; and that the local
  542. cache of resource provider information contains a record of:
  543. - The specified provider
  544. - All providers in its tree
  545. - All providers associated via aggregate with all providers in said
  546. tree
  547. and for each of those providers:
  548. - The UUIDs of its aggregates
  549. - The trait strings associated with the provider
  550. Note that if the provider did not exist prior to this call, the above
  551. reduces to just the specified provider as a root, with no aggregates or
  552. traits.
  553. :param context: The security context
  554. :param uuid: UUID identifier for the resource provider to ensure exists
  555. :param name: Optional name for the resource provider if the record
  556. does not exist. If empty, the name is set to the UUID
  557. value
  558. :param parent_provider_uuid: Optional UUID of the immediate parent,
  559. which must have been previously _ensured.
  560. """
  561. # NOTE(efried): We currently have no code path where we need to set the
  562. # parent_provider_uuid on a previously-parent-less provider - so we do
  563. # NOT handle that scenario here.
  564. # TODO(efried): Reinstate this optimization if possible.
  565. # For now, this is removed due to the following:
  566. # - update_provider_tree adds a child with some bogus inventory (bad
  567. # resource class) or trait (invalid trait name).
  568. # - update_from_provider_tree creates the child in placement and adds
  569. # it to the cache, then attempts to add the bogus inventory/trait.
  570. # The latter fails, so update_from_provider_tree invalidates the
  571. # cache entry by removing the child from the cache.
  572. # - Ordinarily, we would rely on the code below (_get_providers_in_tree
  573. # and _provider_tree.populate_from_iterable) to restore the child to
  574. # the cache on the next iteration. BUT since the root is still
  575. # present in the cache, the commented-out block will cause that part
  576. # of this method to be skipped.
  577. # if self._provider_tree.exists(uuid):
  578. # # If we had the requested provider locally, refresh it and its
  579. # # descendants, but only if stale.
  580. # for u in self._provider_tree.get_provider_uuids(uuid):
  581. # self._refresh_associations(context, u, force=False)
  582. # return uuid
  583. # We don't have it locally; check placement or create it.
  584. created_rp = None
  585. rps_to_refresh = self._get_providers_in_tree(context, uuid)
  586. if not rps_to_refresh:
  587. created_rp = self._create_resource_provider(
  588. context, uuid, name or uuid,
  589. parent_provider_uuid=parent_provider_uuid)
  590. # If @safe_connect can't establish a connection to the placement
  591. # service, like if placement isn't running or nova-compute is
  592. # mis-configured for authentication, we'll get None back and need
  593. # to treat it like we couldn't create the provider (because we
  594. # couldn't).
  595. if created_rp is None:
  596. raise exception.ResourceProviderCreationFailed(
  597. name=name or uuid)
  598. # Don't add the created_rp to rps_to_refresh. Since we just
  599. # created it, it has no aggregates or traits.
  600. self._provider_tree.populate_from_iterable(
  601. rps_to_refresh or [created_rp])
  602. # At this point, the whole tree exists in the local cache.
  603. for rp_to_refresh in rps_to_refresh:
  604. # NOTE(efried): _refresh_associations doesn't refresh inventory
  605. # (yet) - see that method's docstring for the why.
  606. self._refresh_and_get_inventory(context, rp_to_refresh['uuid'])
  607. self._refresh_associations(context, rp_to_refresh['uuid'],
  608. force=True)
  609. return uuid
  610. @safe_connect
  611. def _delete_provider(self, rp_uuid, global_request_id=None):
  612. resp = self.delete('/resource_providers/%s' % rp_uuid,
  613. global_request_id=global_request_id)
  614. # Check for 404 since we don't need to warn/raise if we tried to delete
  615. # something which doesn"t actually exist.
  616. if resp or resp.status_code == 404:
  617. if resp:
  618. LOG.info("Deleted resource provider %s", rp_uuid)
  619. # clean the caches
  620. try:
  621. self._provider_tree.remove(rp_uuid)
  622. except ValueError:
  623. pass
  624. self._association_refresh_time.pop(rp_uuid, None)
  625. return
  626. msg = ("[%(placement_req_id)s] Failed to delete resource provider "
  627. "with UUID %(uuid)s from the placement API. Got "
  628. "%(status_code)d: %(err_text)s.")
  629. args = {
  630. 'placement_req_id': get_placement_request_id(resp),
  631. 'uuid': rp_uuid,
  632. 'status_code': resp.status_code,
  633. 'err_text': resp.text
  634. }
  635. LOG.error(msg, args)
  636. # On conflict, the caller may wish to delete allocations and
  637. # redrive. (Note that this is not the same as a
  638. # PlacementAPIConflict case.)
  639. if resp.status_code == 409:
  640. raise exception.ResourceProviderInUse()
  641. raise exception.ResourceProviderDeletionFailed(uuid=rp_uuid)
  642. def _get_inventory(self, context, rp_uuid):
  643. url = '/resource_providers/%s/inventories' % rp_uuid
  644. result = self.get(url, global_request_id=context.global_id)
  645. if not result:
  646. return None
  647. return result.json()
  648. def _refresh_and_get_inventory(self, context, rp_uuid):
  649. """Helper method that retrieves the current inventory for the supplied
  650. resource provider according to the placement API.
  651. If the cached generation of the resource provider is not the same as
  652. the generation returned from the placement API, we update the cached
  653. generation and attempt to update inventory if any exists, otherwise
  654. return empty inventories.
  655. """
  656. curr = self._get_inventory(context, rp_uuid)
  657. if curr is None:
  658. return None
  659. self._provider_tree.update_inventory(
  660. rp_uuid, curr['inventories'],
  661. generation=curr['resource_provider_generation'])
  662. return curr
  663. def _refresh_associations(self, context, rp_uuid, force=False,
  664. refresh_sharing=True):
  665. """Refresh aggregates, traits, and (optionally) aggregate-associated
  666. sharing providers for the specified resource provider uuid.
  667. Only refresh if there has been no refresh during the lifetime of
  668. this process, CONF.compute.resource_provider_association_refresh
  669. seconds have passed, or the force arg has been set to True.
  670. Note that we do *not* refresh inventories. The reason is largely
  671. historical: all code paths that get us here are doing inventory refresh
  672. themselves.
  673. :param context: The security context
  674. :param rp_uuid: UUID of the resource provider to check for fresh
  675. aggregates and traits
  676. :param force: If True, force the refresh
  677. :param refresh_sharing: If True, fetch all the providers associated
  678. by aggregate with the specified provider,
  679. including their traits and aggregates (but not
  680. *their* sharing providers).
  681. :raise: On various placement API errors, one of:
  682. - ResourceProviderAggregateRetrievalFailed
  683. - ResourceProviderTraitRetrievalFailed
  684. - ResourceProviderRetrievalFailed
  685. """
  686. if force or self._associations_stale(rp_uuid):
  687. # Refresh aggregates
  688. agg_info = self._get_provider_aggregates(context, rp_uuid)
  689. # If @safe_connect makes the above return None, this will raise
  690. # TypeError. Good.
  691. aggs, generation = agg_info.aggregates, agg_info.generation
  692. msg = ("Refreshing aggregate associations for resource provider "
  693. "%s, aggregates: %s")
  694. LOG.debug(msg, rp_uuid, ','.join(aggs or ['None']))
  695. # NOTE(efried): This will blow up if called for a RP that doesn't
  696. # exist in our _provider_tree.
  697. self._provider_tree.update_aggregates(
  698. rp_uuid, aggs, generation=generation)
  699. # Refresh traits
  700. trait_info = self._get_provider_traits(context, rp_uuid)
  701. # If @safe_connect makes the above return None, this will raise
  702. # TypeError. Good.
  703. traits, generation = trait_info.traits, trait_info.generation
  704. msg = ("Refreshing trait associations for resource provider %s, "
  705. "traits: %s")
  706. LOG.debug(msg, rp_uuid, ','.join(traits or ['None']))
  707. # NOTE(efried): This will blow up if called for a RP that doesn't
  708. # exist in our _provider_tree.
  709. self._provider_tree.update_traits(
  710. rp_uuid, traits, generation=generation)
  711. if refresh_sharing:
  712. # Refresh providers associated by aggregate
  713. for rp in self._get_sharing_providers(context, aggs):
  714. if not self._provider_tree.exists(rp['uuid']):
  715. # NOTE(efried): Right now sharing providers are always
  716. # treated as roots. This is deliberate. From the
  717. # context of this compute's RP, it doesn't matter if a
  718. # sharing RP is part of a tree.
  719. self._provider_tree.new_root(
  720. rp['name'], rp['uuid'],
  721. generation=rp['generation'])
  722. # Now we have to (populate or) refresh that guy's traits
  723. # and aggregates (but not *his* aggregate-associated
  724. # providers). No need to override force=True for newly-
  725. # added providers - the missing timestamp will always
  726. # trigger them to refresh.
  727. self._refresh_associations(context, rp['uuid'],
  728. force=force,
  729. refresh_sharing=False)
  730. self._association_refresh_time[rp_uuid] = time.time()
  731. def _associations_stale(self, uuid):
  732. """Respond True if aggregates and traits have not been refreshed
  733. "recently".
  734. Associations are stale if association_refresh_time for this uuid is not
  735. set or is more than CONF.compute.resource_provider_association_refresh
  736. seconds ago.
  737. """
  738. refresh_time = self._association_refresh_time.get(uuid, 0)
  739. return ((time.time() - refresh_time) >
  740. CONF.compute.resource_provider_association_refresh)
  741. def _update_inventory_attempt(self, context, rp_uuid, inv_data):
  742. """Update the inventory for this resource provider if needed.
  743. :param context: The security context
  744. :param rp_uuid: The resource provider UUID for the operation
  745. :param inv_data: The new inventory for the resource provider
  746. :returns: True if the inventory was updated (or did not need to be),
  747. False otherwise.
  748. """
  749. # TODO(jaypipes): Should we really be calling the placement API to get
  750. # the current inventory for every resource provider each and every time
  751. # update_resource_stats() is called? :(
  752. curr = self._refresh_and_get_inventory(context, rp_uuid)
  753. if curr is None:
  754. return False
  755. cur_gen = curr['resource_provider_generation']
  756. # Check to see if we need to update placement's view
  757. if not self._provider_tree.has_inventory_changed(rp_uuid, inv_data):
  758. return True
  759. payload = {
  760. 'resource_provider_generation': cur_gen,
  761. 'inventories': inv_data,
  762. }
  763. url = '/resource_providers/%s/inventories' % rp_uuid
  764. # NOTE(vdrok): in microversion 1.26 it is allowed to have inventory
  765. # records with reserved value equal to total
  766. version = ALLOW_RESERVED_EQUAL_TOTAL_INVENTORY_VERSION
  767. result = self.put(url, payload, version=version,
  768. global_request_id=context.global_id)
  769. if result.status_code == 409:
  770. LOG.info('[%(placement_req_id)s] Inventory update conflict for '
  771. '%(resource_provider_uuid)s with generation ID '
  772. '%(generation)s',
  773. {'placement_req_id': get_placement_request_id(result),
  774. 'resource_provider_uuid': rp_uuid,
  775. 'generation': cur_gen})
  776. # NOTE(jaypipes): There may be cases when we try to set a
  777. # provider's inventory that results in attempting to delete an
  778. # inventory record for a resource class that has an active
  779. # allocation. We need to catch this particular case and raise an
  780. # exception here instead of returning False, since we should not
  781. # re-try the operation in this case.
  782. #
  783. # A use case for where this can occur is the following:
  784. #
  785. # 1) Provider created for each Ironic baremetal node in Newton
  786. # 2) Inventory records for baremetal node created for VCPU,
  787. # MEMORY_MB and DISK_GB
  788. # 3) A Nova instance consumes the baremetal node and allocation
  789. # records are created for VCPU, MEMORY_MB and DISK_GB matching
  790. # the total amount of those resource on the baremetal node.
  791. # 3) Upgrade to Ocata and now resource tracker wants to set the
  792. # provider's inventory to a single record of resource class
  793. # CUSTOM_IRON_SILVER (or whatever the Ironic node's
  794. # "resource_class" attribute is)
  795. # 4) Scheduler report client sends the inventory list containing a
  796. # single CUSTOM_IRON_SILVER record and placement service
  797. # attempts to delete the inventory records for VCPU, MEMORY_MB
  798. # and DISK_GB. An exception is raised from the placement service
  799. # because allocation records exist for those resource classes,
  800. # and a 409 Conflict is returned to the compute node. We need to
  801. # trigger a delete of the old allocation records and then set
  802. # the new inventory, and then set the allocation record to the
  803. # new CUSTOM_IRON_SILVER record.
  804. rc = _extract_inventory_in_use(result.text)
  805. if rc is not None:
  806. raise exception.InventoryInUse(
  807. resource_classes=rc,
  808. resource_provider=rp_uuid,
  809. )
  810. # Invalidate our cache and re-fetch the resource provider
  811. # to be sure to get the latest generation.
  812. self._provider_tree.remove(rp_uuid)
  813. # NOTE(jaypipes): We don't need to pass a name parameter to
  814. # _ensure_resource_provider() because we know the resource provider
  815. # record already exists. We're just reloading the record here.
  816. self._ensure_resource_provider(context, rp_uuid)
  817. return False
  818. elif not result:
  819. placement_req_id = get_placement_request_id(result)
  820. LOG.warning('[%(placement_req_id)s] Failed to update inventory '
  821. 'for resource provider %(uuid)s: %(status)i %(text)s',
  822. {'placement_req_id': placement_req_id,
  823. 'uuid': rp_uuid,
  824. 'status': result.status_code,
  825. 'text': result.text})
  826. # log the body at debug level
  827. LOG.debug('[%(placement_req_id)s] Failed inventory update request '
  828. 'for resource provider %(uuid)s with body: %(payload)s',
  829. {'placement_req_id': placement_req_id,
  830. 'uuid': rp_uuid,
  831. 'payload': payload})
  832. return False
  833. if result.status_code != 200:
  834. placement_req_id = get_placement_request_id(result)
  835. LOG.info('[%(placement_req_id)s] Received unexpected response '
  836. 'code %(code)i while trying to update inventory for '
  837. 'resource provider %(uuid)s: %(text)s',
  838. {'placement_req_id': placement_req_id,
  839. 'uuid': rp_uuid,
  840. 'code': result.status_code,
  841. 'text': result.text})
  842. return False
  843. # Update our view of the generation for next time
  844. updated_inventories_result = result.json()
  845. new_gen = updated_inventories_result['resource_provider_generation']
  846. self._provider_tree.update_inventory(rp_uuid, inv_data,
  847. generation=new_gen)
  848. LOG.debug('Updated inventory for %s at generation %i',
  849. rp_uuid, new_gen)
  850. return True
  851. @safe_connect
  852. def _update_inventory(self, context, rp_uuid, inv_data):
  853. for attempt in (1, 2, 3):
  854. if not self._provider_tree.exists(rp_uuid):
  855. # NOTE(danms): Either we failed to fetch/create the RP
  856. # on our first attempt, or a previous attempt had to
  857. # invalidate the cache, and we were unable to refresh
  858. # it. Bail and try again next time.
  859. LOG.warning('Unable to refresh my resource provider record')
  860. return False
  861. if self._update_inventory_attempt(context, rp_uuid, inv_data):
  862. return True
  863. time.sleep(1)
  864. return False
  865. def get_provider_tree_and_ensure_root(self, context, rp_uuid, name=None,
  866. parent_provider_uuid=None):
  867. """Returns a fresh ProviderTree representing all providers which are in
  868. the same tree or in the same aggregate as the specified provider,
  869. including their aggregates, traits, and inventories.
  870. If the specified provider does not exist, it is created with the
  871. specified UUID, name, and parent provider (which *must* already exist).
  872. :param context: The security context
  873. :param rp_uuid: UUID of the resource provider for which to populate the
  874. tree. (This doesn't need to be the UUID of the root.)
  875. :param name: Optional name for the resource provider if the record
  876. does not exist. If empty, the name is set to the UUID
  877. value
  878. :param parent_provider_uuid: Optional UUID of the immediate parent,
  879. which must have been previously _ensured.
  880. :return: A new ProviderTree object.
  881. """
  882. # TODO(efried): We would like to have the caller handle create-and/or-
  883. # cache-if-not-already, but the resource tracker is currently
  884. # structured to handle initialization and update in a single path. At
  885. # some point this should be refactored, and this method can *just*
  886. # return a deep copy of the local _provider_tree cache.
  887. # (Re)populate the local ProviderTree
  888. self._ensure_resource_provider(
  889. context, rp_uuid, name=name,
  890. parent_provider_uuid=parent_provider_uuid)
  891. # Ensure inventories are up to date (for *all* cached RPs)
  892. for uuid in self._provider_tree.get_provider_uuids():
  893. self._refresh_and_get_inventory(context, uuid)
  894. # Return a *copy* of the tree.
  895. return copy.deepcopy(self._provider_tree)
  896. def set_inventory_for_provider(self, context, rp_uuid, rp_name, inv_data,
  897. parent_provider_uuid=None):
  898. """Given the UUID of a provider, set the inventory records for the
  899. provider to the supplied dict of resources.
  900. :param context: The security context
  901. :param rp_uuid: UUID of the resource provider to set inventory for
  902. :param rp_name: Name of the resource provider in case we need to create
  903. a record for it in the placement API
  904. :param inv_data: Dict, keyed by resource class name, of inventory data
  905. to set against the provider
  906. :param parent_provider_uuid:
  907. If the provider is not a root, this is required, and represents
  908. the UUID of the immediate parent, which is a provider for which
  909. this method has already been invoked.
  910. :raises: exc.InvalidResourceClass if a supplied custom resource class
  911. name does not meet the placement API's format requirements.
  912. """
  913. self._ensure_resource_provider(
  914. context, rp_uuid, rp_name,
  915. parent_provider_uuid=parent_provider_uuid)
  916. # Auto-create custom resource classes coming from a virt driver
  917. self._ensure_resource_classes(context, set(inv_data))
  918. # NOTE(efried): Do not use the DELETE API introduced in microversion
  919. # 1.5, even if the new inventory is empty. It provides no way of
  920. # sending the generation down, so no way to trigger/detect a conflict
  921. # if an out-of-band update occurs between when we GET the latest and
  922. # when we invoke the DELETE. See bug #1746374.
  923. self._update_inventory(context, rp_uuid, inv_data)
  924. def _set_inventory_for_provider(self, context, rp_uuid, inv_data):
  925. """Given the UUID of a provider, set the inventory records for the
  926. provider to the supplied dict of resources.
  927. Compare and contrast with set_inventory_for_provider above. This one
  928. is specially formulated for use by update_from_provider_tree. Like the
  929. other method, we DO need to _ensure_resource_class - i.e. automatically
  930. create new resource classes specified in the inv_data. However, UNLIKE
  931. the other method:
  932. - We don't use the DELETE API when inventory is empty, because that guy
  933. doesn't return content, and we need to update the cached provider
  934. tree with the new generation.
  935. - We raise exceptions (rather than returning a boolean) which are
  936. handled in a consistent fashion by update_from_provider_tree.
  937. - We don't invalidate the cache on failure. That's controlled at a
  938. broader scope (based on errors from ANY of the set_*_for_provider
  939. methods, etc.) by update_from_provider_tree.
  940. - We don't retry. In this code path, retries happen at the level of
  941. the resource tracker on the next iteration.
  942. - We take advantage of the cache and no-op if inv_data isn't different
  943. from what we have locally. This is an optimization, not essential.
  944. - We don't _ensure_resource_provider or refresh_and_get_inventory,
  945. because that's already been done in the code paths leading up to
  946. update_from_provider_tree (by get_provider_tree). This is an
  947. optimization, not essential.
  948. In short, this version is more in the spirit of set_traits_for_provider
  949. and set_aggregates_for_provider.
  950. :param context: The security context
  951. :param rp_uuid: The UUID of the provider whose inventory is to be
  952. updated.
  953. :param inv_data: Dict, keyed by resource class name, of inventory data
  954. to set for the provider. Use None or the empty dict
  955. to remove all inventory for the provider.
  956. :raises: InventoryInUse if inv_data indicates removal of inventory in a
  957. resource class which has active allocations for this provider.
  958. :raises: InvalidResourceClass if inv_data contains a resource class
  959. which cannot be created.
  960. :raises: ResourceProviderUpdateConflict if the provider's generation
  961. doesn't match the generation in the cache. Callers may choose
  962. to retrieve the provider and its associations afresh and
  963. redrive this operation.
  964. :raises: ResourceProviderUpdateFailed on any other placement API
  965. failure.
  966. """
  967. # TODO(efried): Consolidate/refactor to one set_inventory_for_provider.
  968. # NOTE(efried): This is here because _ensure_resource_class already has
  969. # @safe_connect, so we don't want to decorate this whole method with it
  970. @safe_connect
  971. def do_put(url, payload):
  972. # NOTE(vdrok): in microversion 1.26 it is allowed to have inventory
  973. # records with reserved value equal to total
  974. return self.put(
  975. url, payload, global_request_id=context.global_id,
  976. version=ALLOW_RESERVED_EQUAL_TOTAL_INVENTORY_VERSION)
  977. # If not different from what we've got, short out
  978. if not self._provider_tree.has_inventory_changed(rp_uuid, inv_data):
  979. return
  980. # Ensure non-standard resource classes exist, creating them if needed.
  981. self._ensure_resource_classes(context, set(inv_data))
  982. url = '/resource_providers/%s/inventories' % rp_uuid
  983. inv_data = inv_data or {}
  984. generation = self._provider_tree.data(rp_uuid).generation
  985. payload = {
  986. 'resource_provider_generation': generation,
  987. 'inventories': inv_data,
  988. }
  989. resp = do_put(url, payload)
  990. if resp.status_code == 200:
  991. json = resp.json()
  992. self._provider_tree.update_inventory(
  993. rp_uuid, json['inventories'],
  994. generation=json['resource_provider_generation'])
  995. return
  996. # Some error occurred; log it
  997. msg = ("[%(placement_req_id)s] Failed to update inventory to "
  998. "[%(inv_data)s] for resource provider with UUID %(uuid)s. Got "
  999. "%(status_code)d: %(err_text)s")
  1000. args = {
  1001. 'placement_req_id': get_placement_request_id(resp),
  1002. 'uuid': rp_uuid,
  1003. 'inv_data': str(inv_data),
  1004. 'status_code': resp.status_code,
  1005. 'err_text': resp.text,
  1006. }
  1007. LOG.error(msg, args)
  1008. if resp.status_code == 409:
  1009. # If a conflict attempting to remove inventory in a resource class
  1010. # with active allocations, raise InventoryInUse
  1011. rc = _extract_inventory_in_use(resp.text)
  1012. if rc is not None:
  1013. raise exception.InventoryInUse(
  1014. resource_classes=rc,
  1015. resource_provider=rp_uuid,
  1016. )
  1017. # Other conflicts are generation mismatch: raise conflict exception
  1018. raise exception.ResourceProviderUpdateConflict(
  1019. uuid=rp_uuid, generation=generation, error=resp.text)
  1020. # Otherwise, raise generic exception
  1021. raise exception.ResourceProviderUpdateFailed(url=url, error=resp.text)
  1022. @safe_connect
  1023. def _ensure_traits(self, context, traits):
  1024. """Make sure all specified traits exist in the placement service.
  1025. :param context: The security context
  1026. :param traits: Iterable of trait strings to ensure exist.
  1027. :raises: TraitCreationFailed if traits contains a trait that did not
  1028. exist in placement, and couldn't be created. When this
  1029. exception is raised, it is possible that *some* of the
  1030. requested traits were created.
  1031. :raises: TraitRetrievalFailed if the initial query of existing traits
  1032. was unsuccessful. In this scenario, it is guaranteed that
  1033. no traits were created.
  1034. """
  1035. if not traits:
  1036. return
  1037. # Query for all the requested traits. Whichever ones we *don't* get
  1038. # back, we need to create.
  1039. # NOTE(efried): We don't attempt to filter based on our local idea of
  1040. # standard traits, which may not be in sync with what the placement
  1041. # service knows. If the caller tries to ensure a nonexistent
  1042. # "standard" trait, they deserve the TraitCreationFailed exception
  1043. # they'll get.
  1044. resp = self.get('/traits?name=in:' + ','.join(traits), version='1.6',
  1045. global_request_id=context.global_id)
  1046. if resp.status_code == 200:
  1047. traits_to_create = set(traits) - set(resp.json()['traits'])
  1048. # Might be neat to have a batch create. But creating multiple
  1049. # traits will generally happen once, at initial startup, if at all.
  1050. for trait in traits_to_create:
  1051. resp = self.put('/traits/' + trait, None, version='1.6',
  1052. global_request_id=context.global_id)
  1053. if not resp:
  1054. raise exception.TraitCreationFailed(name=trait,
  1055. error=resp.text)
  1056. return
  1057. # The initial GET failed
  1058. msg = ("[%(placement_req_id)s] Failed to retrieve the list of traits. "
  1059. "Got %(status_code)d: %(err_text)s")
  1060. args = {
  1061. 'placement_req_id': get_placement_request_id(resp),
  1062. 'status_code': resp.status_code,
  1063. 'err_text': resp.text,
  1064. }
  1065. LOG.error(msg, args)
  1066. raise exception.TraitRetrievalFailed(error=resp.text)
  1067. @safe_connect
  1068. def set_traits_for_provider(self, context, rp_uuid, traits):
  1069. """Replace a provider's traits with those specified.
  1070. The provider must exist - this method does not attempt to create it.
  1071. :param context: The security context
  1072. :param rp_uuid: The UUID of the provider whose traits are to be updated
  1073. :param traits: Iterable of traits to set on the provider
  1074. :raises: ResourceProviderUpdateConflict if the provider's generation
  1075. doesn't match the generation in the cache. Callers may choose
  1076. to retrieve the provider and its associations afresh and
  1077. redrive this operation.
  1078. :raises: ResourceProviderUpdateFailed on any other placement API
  1079. failure.
  1080. :raises: TraitCreationFailed if traits contains a trait that did not
  1081. exist in placement, and couldn't be created.
  1082. :raises: TraitRetrievalFailed if the initial query of existing traits
  1083. was unsuccessful.
  1084. """
  1085. # If not different from what we've got, short out
  1086. if not self._provider_tree.have_traits_changed(rp_uuid, traits):
  1087. return
  1088. self._ensure_traits(context, traits)
  1089. url = '/resource_providers/%s/traits' % rp_uuid
  1090. # NOTE(efried): Don't use the DELETE API when traits is empty, because
  1091. # that guy doesn't return content, and we need to update the cached
  1092. # provider tree with the new generation.
  1093. traits = list(traits) if traits else []
  1094. generation = self._provider_tree.data(rp_uuid).generation
  1095. payload = {
  1096. 'resource_provider_generation': generation,
  1097. 'traits': traits,
  1098. }
  1099. resp = self.put(url, payload, version='1.6',
  1100. global_request_id=context.global_id)
  1101. if resp.status_code == 200:
  1102. json = resp.json()
  1103. self._provider_tree.update_traits(
  1104. rp_uuid, json['traits'],
  1105. generation=json['resource_provider_generation'])
  1106. return
  1107. # Some error occurred; log it
  1108. msg = ("[%(placement_req_id)s] Failed to update traits to "
  1109. "[%(traits)s] for resource provider with UUID %(uuid)s. Got "
  1110. "%(status_code)d: %(err_text)s")
  1111. args = {
  1112. 'placement_req_id': get_placement_request_id(resp),
  1113. 'uuid': rp_uuid,
  1114. 'traits': ','.join(traits),
  1115. 'status_code': resp.status_code,
  1116. 'err_text': resp.text,
  1117. }
  1118. LOG.error(msg, args)
  1119. # If a conflict, raise special conflict exception
  1120. if resp.status_code == 409:
  1121. raise exception.ResourceProviderUpdateConflict(
  1122. uuid=rp_uuid, generation=generation, error=resp.text)
  1123. # Otherwise, raise generic exception
  1124. raise exception.ResourceProviderUpdateFailed(url=url, error=resp.text)
  1125. @safe_connect
  1126. def set_aggregates_for_provider(self, context, rp_uuid, aggregates,
  1127. use_cache=True, generation=None):
  1128. """Replace a provider's aggregates with those specified.
  1129. The provider must exist - this method does not attempt to create it.
  1130. :param context: The security context
  1131. :param rp_uuid: The UUID of the provider whose aggregates are to be
  1132. updated.
  1133. :param aggregates: Iterable of aggregates to set on the provider.
  1134. :param use_cache: If False, indicates not to update the cache of
  1135. resource providers.
  1136. :param generation: Resource provider generation. Required if use_cache
  1137. is False.
  1138. :raises: ResourceProviderUpdateConflict if the provider's generation
  1139. doesn't match the generation in the cache. Callers may choose
  1140. to retrieve the provider and its associations afresh and
  1141. redrive this operation.
  1142. :raises: ResourceProviderUpdateFailed on any other placement API
  1143. failure.
  1144. """
  1145. # If a generation is specified, it trumps whatever's in the cache.
  1146. # Otherwise...
  1147. if generation is None:
  1148. if use_cache:
  1149. generation = self._provider_tree.data(rp_uuid).generation
  1150. else:
  1151. # Either cache or generation is required
  1152. raise ValueError(
  1153. _("generation is required with use_cache=False"))
  1154. # Check whether aggregates need updating. We can only do this if we
  1155. # have a cache entry with a matching generation.
  1156. try:
  1157. if (self._provider_tree.data(rp_uuid).generation == generation
  1158. and not self._provider_tree.have_aggregates_changed(
  1159. rp_uuid, aggregates)):
  1160. return
  1161. except ValueError:
  1162. # Not found in the cache; proceed
  1163. pass
  1164. url = '/resource_providers/%s/aggregates' % rp_uuid
  1165. aggregates = list(aggregates) if aggregates else []
  1166. payload = {'aggregates': aggregates,
  1167. 'resource_provider_generation': generation}
  1168. resp = self.put(url, payload, version=AGGREGATE_GENERATION_VERSION,
  1169. global_request_id=context.global_id)
  1170. if resp.status_code == 200:
  1171. # Try to update the cache regardless. If use_cache=False, ignore
  1172. # any failures.
  1173. try:
  1174. data = resp.json()
  1175. self._provider_tree.update_aggregates(
  1176. rp_uuid, data['aggregates'],
  1177. generation=data['resource_provider_generation'])
  1178. except ValueError:
  1179. if use_cache:
  1180. # The entry should've been there
  1181. raise
  1182. return
  1183. # Some error occurred; log it
  1184. msg = ("[%(placement_req_id)s] Failed to update aggregates to "
  1185. "[%(aggs)s] for resource provider with UUID %(uuid)s. Got "
  1186. "%(status_code)d: %(err_text)s")
  1187. args = {
  1188. 'placement_req_id': get_placement_request_id(resp),
  1189. 'uuid': rp_uuid,
  1190. 'aggs': ','.join(aggregates),
  1191. 'status_code': resp.status_code,
  1192. 'err_text': resp.text,
  1193. }
  1194. # If a conflict, invalidate the cache and raise special exception
  1195. if resp.status_code == 409:
  1196. # No reason to condition cache invalidation on use_cache - if we
  1197. # got a 409, the cache entry is still bogus if it exists; and the
  1198. # below is a no-op if it doesn't.
  1199. try:
  1200. self._provider_tree.remove(rp_uuid)
  1201. except ValueError:
  1202. pass
  1203. self._association_refresh_time.pop(rp_uuid, None)
  1204. LOG.warning(msg, args)
  1205. raise exception.ResourceProviderUpdateConflict(
  1206. uuid=rp_uuid, generation=generation, error=resp.text)
  1207. # Otherwise, raise generic exception
  1208. LOG.error(msg, args)
  1209. raise exception.ResourceProviderUpdateFailed(url=url, error=resp.text)
  1210. @safe_connect
  1211. def _ensure_resource_classes(self, context, names):
  1212. """Make sure resource classes exist.
  1213. :param context: The security context
  1214. :param names: Iterable of string names of the resource classes to
  1215. check/create. Must not be None.
  1216. :raises: exception.InvalidResourceClass if an attempt is made to create
  1217. an invalid resource class.
  1218. """
  1219. # Placement API version that supports PUT /resource_classes/CUSTOM_*
  1220. # to create (or validate the existence of) a consumer-specified
  1221. # resource class.
  1222. version = '1.7'
  1223. to_ensure = set(n for n in names
  1224. if n.startswith(fields.ResourceClass.CUSTOM_NAMESPACE))
  1225. for name in to_ensure:
  1226. # no payload on the put request
  1227. resp = self.put(
  1228. "/resource_classes/%s" % name, None, version=version,
  1229. global_request_id=context.global_id)
  1230. if not resp:
  1231. msg = ("Failed to ensure resource class record with placement "
  1232. "API for resource class %(rc_name)s. Got "
  1233. "%(status_code)d: %(err_text)s.")
  1234. args = {
  1235. 'rc_name': name,
  1236. 'status_code': resp.status_code,
  1237. 'err_text': resp.text,
  1238. }
  1239. LOG.error(msg, args)
  1240. raise exception.InvalidResourceClass(resource_class=name)
  1241. def update_compute_node(self, context, compute_node):
  1242. """Creates or updates stats for the supplied compute node.
  1243. :param context: The security context
  1244. :param compute_node: updated nova.objects.ComputeNode to report
  1245. :raises `exception.InventoryInUse` if the compute node has had changes
  1246. to its inventory but there are still active allocations for
  1247. resource classes that would be deleted by an update to the
  1248. placement API.
  1249. """
  1250. self._ensure_resource_provider(context, compute_node.uuid,
  1251. compute_node.hypervisor_hostname)
  1252. inv_data = _compute_node_to_inventory_dict(compute_node)
  1253. # NOTE(efried): Do not use the DELETE API introduced in microversion
  1254. # 1.5, even if the new inventory is empty. It provides no way of
  1255. # sending the generation down, so no way to trigger/detect a conflict
  1256. # if an out-of-band update occurs between when we GET the latest and
  1257. # when we invoke the DELETE. See bug #1746374.
  1258. self._update_inventory(context, compute_node.uuid, inv_data)
  1259. def update_from_provider_tree(self, context, new_tree):
  1260. """Flush changes from a specified ProviderTree back to placement.
  1261. The specified ProviderTree is compared against the local cache. Any
  1262. changes are flushed back to the placement service. Upon successful
  1263. completion, the local cache should reflect the specified ProviderTree.
  1264. This method is best-effort and not atomic. When exceptions are raised,
  1265. it is possible that some of the changes have been flushed back, leaving
  1266. the placement database in an inconsistent state. This should be
  1267. recoverable through subsequent calls.
  1268. :param context: The security context
  1269. :param new_tree: A ProviderTree instance representing the desired state
  1270. of providers in placement.
  1271. :raises: ResourceProviderSyncFailed if any errors were encountered
  1272. attempting to perform the necessary API operations.
  1273. """
  1274. # NOTE(efried): We currently do not handle the "rename" case. This is
  1275. # where new_tree contains a provider named Y whose UUID already exists
  1276. # but is named X. Today the only way the consumer could accomplish
  1277. # this is by deleting the provider and recreating it with the new name.
  1278. @contextlib.contextmanager
  1279. def catch_all(rp_uuid):
  1280. """Convert all "expected" exceptions from placement API helpers to
  1281. True or False. Saves having to do try/except for every helper call
  1282. below.
  1283. """
  1284. class Status(object):
  1285. success = True
  1286. s = Status()
  1287. # TODO(efried): Make a base exception class from which all these
  1288. # can inherit.
  1289. helper_exceptions = (
  1290. exception.InvalidResourceClass,
  1291. exception.InventoryInUse,
  1292. exception.ResourceProviderAggregateRetrievalFailed,
  1293. exception.ResourceProviderDeletionFailed,
  1294. exception.ResourceProviderInUse,
  1295. exception.ResourceProviderRetrievalFailed,
  1296. exception.ResourceProviderTraitRetrievalFailed,
  1297. exception.ResourceProviderUpdateConflict,
  1298. exception.ResourceProviderUpdateFailed,
  1299. exception.TraitCreationFailed,
  1300. exception.TraitRetrievalFailed,
  1301. )
  1302. try:
  1303. yield s
  1304. except helper_exceptions:
  1305. s.success = False
  1306. # Invalidate the caches
  1307. try:
  1308. self._provider_tree.remove(rp_uuid)
  1309. except ValueError:
  1310. pass
  1311. self._association_refresh_time.pop(rp_uuid, None)
  1312. # Overall indicator of success. Will be set to False on any exception.
  1313. success = True
  1314. # Helper methods herein will be updating the local cache (this is
  1315. # intentional) so we need to grab up front any data we need to operate
  1316. # on in its "original" form.
  1317. old_tree = self._provider_tree
  1318. old_uuids = old_tree.get_provider_uuids()
  1319. new_uuids = new_tree.get_provider_uuids()
  1320. # Do provider deletion first, since it has the best chance of failing
  1321. # for non-generation-conflict reasons (i.e. allocations).
  1322. uuids_to_remove = set(old_uuids) - set(new_uuids)
  1323. # We have to do deletions in bottom-up order, so we don't error
  1324. # attempting to delete a parent who still has children.
  1325. for uuid in reversed(old_uuids):
  1326. if uuid not in uuids_to_remove:
  1327. continue
  1328. with catch_all(uuid) as status:
  1329. self._delete_provider(uuid)
  1330. success = success and status.success
  1331. # Now create (or load) any "new" providers
  1332. uuids_to_add = set(new_uuids) - set(old_uuids)
  1333. # We have to do additions in top-down order, so we don't error
  1334. # attempting to create a child before its parent exists.
  1335. for uuid in new_uuids:
  1336. if uuid not in uuids_to_add:
  1337. continue
  1338. provider = new_tree.data(uuid)
  1339. with catch_all(uuid) as status:
  1340. self._ensure_resource_provider(
  1341. context, uuid, name=provider.name,
  1342. parent_provider_uuid=provider.parent_uuid)
  1343. success = success and status.success
  1344. # At this point the local cache should have all the same providers as
  1345. # new_tree. Whether we added them or not, walk through and diff/flush
  1346. # inventories, traits, and aggregates as necessary (the helper methods
  1347. # are set up to check and short out when the relevant property does not
  1348. # differ from what's in the cache).
  1349. # If we encounter any error and remove a provider from the cache, all
  1350. # its descendants are also removed, and set_*_for_provider methods on
  1351. # it wouldn't be able to get started. Walking the tree in bottom-up
  1352. # order ensures we at least try to process all of the providers.
  1353. for uuid in reversed(new_uuids):
  1354. pd = new_tree.data(uuid)
  1355. with catch_all(pd.uuid) as status:
  1356. self._set_inventory_for_provider(
  1357. context, pd.uuid, pd.inventory)
  1358. self.set_aggregates_for_provider(
  1359. context, pd.uuid, pd.aggregates)
  1360. self.set_traits_for_provider(context, pd.uuid, pd.traits)
  1361. success = success and status.success
  1362. if not success:
  1363. raise exception.ResourceProviderSyncFailed()
  1364. @safe_connect
  1365. def get_allocations_for_consumer(self, context, consumer,
  1366. include_generation=False):
  1367. """Makes a GET /allocations/{consumer} call to Placement.
  1368. :param context: The nova.context.RequestContext auth context
  1369. :param consumer: UUID of the consumer resource
  1370. :param include_generation: True if the response should be the
  1371. full allocations response including ``consumer_generation`` (new
  1372. in microversion 1.28), False if only the "allocations" dict from
  1373. the response body should be returned.
  1374. :returns: dict, see ``include_generation`` for details on format;
  1375. returns None if unable to connect to Placement (see safe_connect)
  1376. """
  1377. url = '/allocations/%s' % consumer
  1378. resp = self.get(
  1379. url, version=CONSUMER_GENERATION_VERSION,
  1380. global_request_id=context.global_id)
  1381. if not resp:
  1382. return {}
  1383. else:
  1384. # TODO(efried): refactor all callers to accept the whole response
  1385. # so we can get rid of this condition
  1386. if include_generation:
  1387. return resp.json()
  1388. return resp.json()['allocations']
  1389. def get_allocations_for_consumer_by_provider(self, context, rp_uuid,
  1390. consumer):
  1391. # NOTE(cdent): This trims to just the allocations being
  1392. # used on this resource provider. In the future when there
  1393. # are shared resources there might be other providers.
  1394. allocations = self.get_allocations_for_consumer(context, consumer)
  1395. if allocations is None:
  1396. # safe_connect can return None on 404
  1397. allocations = {}
  1398. return allocations.get(
  1399. rp_uuid, {}).get('resources', {})
  1400. def _allocate_for_instance(self, context, rp_uuid, instance):
  1401. my_allocations = _instance_to_allocations_dict(instance)
  1402. current_allocations = self.get_allocations_for_consumer_by_provider(
  1403. context, rp_uuid, instance.uuid)
  1404. if current_allocations == my_allocations:
  1405. allocstr = ','.join(['%s=%s' % (k, v)
  1406. for k, v in my_allocations.items()])
  1407. LOG.debug('Instance %(uuid)s allocations are unchanged: %(alloc)s',
  1408. {'uuid': instance.uuid, 'alloc': allocstr})
  1409. return
  1410. LOG.debug('Sending allocation for instance %s',
  1411. my_allocations,
  1412. instance=instance)
  1413. res = self.put_allocations(context, rp_uuid, instance.uuid,
  1414. my_allocations, instance.project_id,
  1415. instance.user_id)
  1416. if res:
  1417. LOG.info('Submitted allocation for instance', instance=instance)
  1418. # NOTE(jaypipes): Currently, this method is ONLY used in two places:
  1419. # 1. By the scheduler to allocate resources on the selected destination
  1420. # hosts.
  1421. # 2. By the conductor LiveMigrationTask to allocate resources on a forced
  1422. # destination host. This is a short-term fix for Pike which should be
  1423. # replaced in Queens by conductor calling the scheduler in the force
  1424. # host case.
  1425. # This method should not be called by the resource tracker; instead, the
  1426. # _allocate_for_instance() method is used which does not perform any
  1427. # checking that a move operation is in place.
  1428. @safe_connect
  1429. @retries
  1430. def claim_resources(self, context, consumer_uuid, alloc_request,
  1431. project_id, user_id, allocation_request_version=None):
  1432. """Creates allocation records for the supplied instance UUID against
  1433. the supplied resource providers.
  1434. We check to see if resources have already been claimed for this
  1435. consumer. If so, we assume that a move operation is underway and the
  1436. scheduler is attempting to claim resources against the new (destination
  1437. host). In order to prevent compute nodes currently performing move
  1438. operations from being scheduled to improperly, we create a "doubled-up"
  1439. allocation that consumes resources on *both* the source and the
  1440. destination host during the move operation. When the move operation
  1441. completes, the destination host (via _allocate_for_instance()) will
  1442. end up setting allocations for the instance only on the destination
  1443. host thereby freeing up resources on the source host appropriately.
  1444. :param context: The security context
  1445. :param consumer_uuid: The instance's UUID.
  1446. :param alloc_request: The JSON body of the request to make to the
  1447. placement's PUT /allocations API
  1448. :param project_id: The project_id associated with the allocations.
  1449. :param user_id: The user_id associated with the allocations.
  1450. :param allocation_request_version: The microversion used to request the
  1451. allocations.
  1452. :returns: True if the allocations were created, False otherwise.
  1453. """
  1454. # Older clients might not send the allocation_request_version, so
  1455. # default to 1.10.
  1456. # TODO(alex_xu): In the rocky, all the client should send the
  1457. # allocation_request_version. So remove this default value.
  1458. allocation_request_version = allocation_request_version or '1.10'
  1459. # Ensure we don't change the supplied alloc request since it's used in
  1460. # a loop within the scheduler against multiple instance claims
  1461. ar = copy.deepcopy(alloc_request)
  1462. # If the allocation_request_version less than 1.12, then convert the
  1463. # allocation array format to the dict format. This conversion can be
  1464. # remove in Rocky release.
  1465. if versionutils.convert_version_to_tuple(
  1466. allocation_request_version) < (1, 12):
  1467. ar = {
  1468. 'allocations': {
  1469. alloc['resource_provider']['uuid']: {
  1470. 'resources': alloc['resources']
  1471. } for alloc in ar['allocations']
  1472. }
  1473. }
  1474. allocation_request_version = '1.12'
  1475. url = '/allocations/%s' % consumer_uuid
  1476. payload = ar
  1477. # We first need to determine if this is a move operation and if so
  1478. # create the "doubled-up" allocation that exists for the duration of
  1479. # the move operation against both the source and destination hosts
  1480. r = self.get(url, global_request_id=context.global_id)
  1481. if r.status_code == 200:
  1482. current_allocs = r.json()['allocations']
  1483. if current_allocs:
  1484. payload = _move_operation_alloc_request(current_allocs, ar)
  1485. payload['project_id'] = project_id
  1486. payload['user_id'] = user_id
  1487. r = self.put(url, payload, version=allocation_request_version,
  1488. global_request_id=context.global_id)
  1489. if r.status_code != 204:
  1490. # NOTE(jaypipes): Yes, it sucks doing string comparison like this
  1491. # but we have no error codes, only error messages.
  1492. if 'concurrently updated' in r.text:
  1493. reason = ('another process changed the resource providers '
  1494. 'involved in our attempt to put allocations for '
  1495. 'consumer %s' % consumer_uuid)
  1496. raise Retry('claim_resources', reason)
  1497. else:
  1498. LOG.warning(
  1499. 'Unable to submit allocation for instance '
  1500. '%(uuid)s (%(code)i %(text)s)',
  1501. {'uuid': consumer_uuid,
  1502. 'code': r.status_code,
  1503. 'text': r.text})
  1504. return r.status_code == 204
  1505. @safe_connect
  1506. def remove_provider_from_instance_allocation(self, context, consumer_uuid,
  1507. rp_uuid, user_id, project_id,
  1508. resources):
  1509. """Grabs an allocation for a particular consumer UUID, strips parts of
  1510. the allocation that refer to a supplied resource provider UUID, and
  1511. then PUTs the resulting allocation back to the placement API for the
  1512. consumer.
  1513. This is used to reconcile the "doubled-up" allocation that the
  1514. scheduler constructs when claiming resources against the destination
  1515. host during a move operation.
  1516. If the move was between hosts, the entire allocation for rp_uuid will
  1517. be dropped. If the move is a resize on the same host, then we will
  1518. subtract resources from the single allocation to ensure we do not
  1519. exceed the reserved or max_unit amounts for the resource on the host.
  1520. :param context: The security context
  1521. :param consumer_uuid: The instance/consumer UUID
  1522. :param rp_uuid: The UUID of the provider whose resources we wish to
  1523. remove from the consumer's allocation
  1524. :param user_id: The instance's user
  1525. :param project_id: The instance's project
  1526. :param resources: The resources to be dropped from the allocation
  1527. """
  1528. url = '/allocations/%s' % consumer_uuid
  1529. # Grab the "doubled-up" allocation that we will manipulate
  1530. r = self.get(url, global_request_id=context.global_id)
  1531. if r.status_code != 200:
  1532. LOG.warning("Failed to retrieve allocations for %s. Got HTTP %s",
  1533. consumer_uuid, r.status_code)
  1534. return False
  1535. current_allocs = r.json()['allocations']
  1536. if not current_allocs:
  1537. LOG.error("Expected to find current allocations for %s, but "
  1538. "found none.", consumer_uuid)
  1539. return False
  1540. # If the host isn't in the current allocation for the instance, don't
  1541. # do anything
  1542. if rp_uuid not in current_allocs:
  1543. LOG.warning("Expected to find allocations referencing resource "
  1544. "provider %s for %s, but found none.",
  1545. rp_uuid, consumer_uuid)
  1546. return True
  1547. compute_providers = [uuid for uuid, alloc in current_allocs.items()
  1548. if 'VCPU' in alloc['resources']]
  1549. LOG.debug('Current allocations for instance: %s', current_allocs,
  1550. instance_uuid=consumer_uuid)
  1551. LOG.debug('Instance %s has resources on %i compute nodes',
  1552. consumer_uuid, len(compute_providers))
  1553. new_allocs = [
  1554. {
  1555. 'resource_provider': {
  1556. 'uuid': alloc_rp_uuid,
  1557. },
  1558. 'resources': alloc['resources'],
  1559. }
  1560. for alloc_rp_uuid, alloc in current_allocs.items()
  1561. if alloc_rp_uuid != rp_uuid
  1562. ]
  1563. if len(compute_providers) == 1:
  1564. # NOTE(danms): We are in a resize to same host scenario. Since we
  1565. # are the only provider then we need to merge back in the doubled
  1566. # allocation with our part subtracted
  1567. peer_alloc = {
  1568. 'resource_provider': {
  1569. 'uuid': rp_uuid,
  1570. },
  1571. 'resources': current_allocs[rp_uuid]['resources']
  1572. }
  1573. LOG.debug('Original resources from same-host '
  1574. 'allocation: %s', peer_alloc['resources'])
  1575. scheduler_utils.merge_resources(peer_alloc['resources'],
  1576. resources, -1)
  1577. LOG.debug('Subtracting old resources from same-host '
  1578. 'allocation: %s', peer_alloc['resources'])
  1579. new_allocs.append(peer_alloc)
  1580. payload = {'allocations': new_allocs}
  1581. payload['project_id'] = project_id
  1582. payload['user_id'] = user_id
  1583. LOG.debug("Sending updated allocation %s for instance %s after "
  1584. "removing resources for %s.",
  1585. new_allocs, consumer_uuid, rp_uuid)
  1586. r = self.put(url, payload, version='1.10',
  1587. global_request_id=context.global_id)
  1588. if r.status_code != 204:
  1589. LOG.warning("Failed to save allocation for %s. Got HTTP %s: %s",
  1590. consumer_uuid, r.status_code, r.text)
  1591. return r.status_code == 204
  1592. @safe_connect
  1593. @retries
  1594. def set_and_clear_allocations(self, context, rp_uuid, consumer_uuid,
  1595. alloc_data, project_id, user_id,
  1596. consumer_to_clear=None):
  1597. """Create allocation records for the supplied consumer UUID while
  1598. simultaneously clearing any allocations identified by the uuid
  1599. in consumer_to_clear, for example a migration uuid when moving an
  1600. instance to another host. This is for atomically managing so-called
  1601. "doubled" migration records.
  1602. :note Currently we only allocate against a single resource provider.
  1603. Once shared storage and things like NUMA allocations are a
  1604. reality, this will change to allocate against multiple providers.
  1605. :param context: The security context
  1606. :param rp_uuid: The UUID of the resource provider to allocate against.
  1607. :param consumer_uuid: The consumer UUID for which allocations are
  1608. being set.
  1609. :param alloc_data: Dict, keyed by resource class, of amounts to
  1610. consume.
  1611. :param project_id: The project_id associated with the allocations.
  1612. :param user_id: The user_id associated with the allocations.
  1613. :param consumer_to_clear: A UUID identifying allocations for a
  1614. consumer that should be cleared.
  1615. :returns: True if the allocations were created, False otherwise.
  1616. :raises: Retry if the operation should be retried due to a concurrent
  1617. update.
  1618. """
  1619. # FIXME(cdent): Fair amount of duplicate with put in here, but now
  1620. # just working things through.
  1621. payload = {
  1622. consumer_uuid: {
  1623. 'allocations': {
  1624. rp_uuid: {
  1625. 'resources': alloc_data
  1626. }
  1627. },
  1628. 'project_id': project_id,
  1629. 'user_id': user_id,
  1630. }
  1631. }
  1632. if consumer_to_clear:
  1633. payload[consumer_to_clear] = {
  1634. 'allocations': {},
  1635. 'project_id': project_id,
  1636. 'user_id': user_id,
  1637. }
  1638. r = self.post('/allocations', payload,
  1639. version=POST_ALLOCATIONS_API_VERSION,
  1640. global_request_id=context.global_id)
  1641. if r.status_code != 204:
  1642. # NOTE(jaypipes): Yes, it sucks doing string comparison like this
  1643. # but we have no error codes, only error messages.
  1644. if 'concurrently updated' in r.text:
  1645. reason = ('another process changed the resource providers '
  1646. 'involved in our attempt to post allocations for '
  1647. 'consumer %s' % consumer_uuid)
  1648. raise Retry('set_and_clear_allocations', reason)
  1649. else:
  1650. LOG.warning(
  1651. 'Unable to post allocations for instance '
  1652. '%(uuid)s (%(code)i %(text)s)',
  1653. {'uuid': consumer_uuid,
  1654. 'code': r.status_code,
  1655. 'text': r.text})
  1656. return r.status_code == 204
  1657. @safe_connect
  1658. @retries
  1659. def put_allocations(self, context, rp_uuid, consumer_uuid, alloc_data,
  1660. project_id, user_id):
  1661. """Creates allocation records for the supplied instance UUID against
  1662. the supplied resource provider.
  1663. :note Currently we only allocate against a single resource provider.
  1664. Once shared storage and things like NUMA allocations are a
  1665. reality, this will change to allocate against multiple providers.
  1666. :param context: The security context
  1667. :param rp_uuid: The UUID of the resource provider to allocate against.
  1668. :param consumer_uuid: The instance's UUID.
  1669. :param alloc_data: Dict, keyed by resource class, of amounts to
  1670. consume.
  1671. :param project_id: The project_id associated with the allocations.
  1672. :param user_id: The user_id associated with the allocations.
  1673. :returns: True if the allocations were created, False otherwise.
  1674. :raises: Retry if the operation should be retried due to a concurrent
  1675. update.
  1676. """
  1677. payload = {
  1678. 'allocations': [
  1679. {
  1680. 'resource_provider': {
  1681. 'uuid': rp_uuid,
  1682. },
  1683. 'resources': alloc_data,
  1684. },
  1685. ],
  1686. 'project_id': project_id,
  1687. 'user_id': user_id,
  1688. }
  1689. url = '/allocations/%s' % consumer_uuid
  1690. r = self.put(url, payload, version='1.8',
  1691. global_request_id=context.global_id)
  1692. if r.status_code != 204:
  1693. # NOTE(jaypipes): Yes, it sucks doing string comparison like this
  1694. # but we have no error codes, only error messages.
  1695. if 'concurrently updated' in r.text:
  1696. reason = ('another process changed the resource providers '
  1697. 'involved in our attempt to put allocations for '
  1698. 'consumer %s' % consumer_uuid)
  1699. raise Retry('put_allocations', reason)
  1700. else:
  1701. LOG.warning(
  1702. 'Unable to submit allocation for instance '
  1703. '%(uuid)s (%(code)i %(text)s)',
  1704. {'uuid': consumer_uuid,
  1705. 'code': r.status_code,
  1706. 'text': r.text})
  1707. return r.status_code == 204
  1708. @safe_connect
  1709. def delete_allocation_for_instance(self, context, uuid):
  1710. url = '/allocations/%s' % uuid
  1711. r = self.delete(url, global_request_id=context.global_id)
  1712. if r:
  1713. LOG.info('Deleted allocation for instance %s', uuid)
  1714. return True
  1715. else:
  1716. # Check for 404 since we don't need to log a warning if we tried to
  1717. # delete something which doesn't actually exist.
  1718. if r.status_code != 404:
  1719. LOG.warning('Unable to delete allocation for instance '
  1720. '%(uuid)s: (%(code)i %(text)s)',
  1721. {'uuid': uuid,
  1722. 'code': r.status_code,
  1723. 'text': r.text})
  1724. return False
  1725. def update_instance_allocation(self, context, compute_node, instance,
  1726. sign):
  1727. if sign > 0:
  1728. self._allocate_for_instance(context, compute_node.uuid, instance)
  1729. else:
  1730. self.delete_allocation_for_instance(context, instance.uuid)
  1731. @safe_connect
  1732. def get_allocations_for_resource_provider(self, context, rp_uuid):
  1733. url = '/resource_providers/%s/allocations' % rp_uuid
  1734. resp = self.get(url, global_request_id=context.global_id)
  1735. if not resp:
  1736. # NOTE(gibi): The request failed with an error response. Rather
  1737. # than return an empty dict, which is possible if there are no
  1738. # allocations against the given provider, return None to indicate
  1739. # a failure - like in the @safe_connect decorator.
  1740. return None
  1741. else:
  1742. return resp.json()['allocations']
  1743. def delete_resource_provider(self, context, compute_node, cascade=False):
  1744. """Deletes the ResourceProvider record for the compute_node.
  1745. :param context: The security context
  1746. :param compute_node: The nova.objects.ComputeNode object that is the
  1747. resource provider being deleted.
  1748. :param cascade: Boolean value that, when True, will first delete any
  1749. associated Allocation and Inventory records for the
  1750. compute node
  1751. """
  1752. nodename = compute_node.hypervisor_hostname
  1753. host = compute_node.host
  1754. rp_uuid = compute_node.uuid
  1755. if cascade:
  1756. # Delete any allocations for this resource provider.
  1757. # Since allocations are by consumer, we get the consumers on this
  1758. # host, which are its instances.
  1759. instances = objects.InstanceList.get_by_host_and_node(context,
  1760. host, nodename)
  1761. for instance in instances:
  1762. self.delete_allocation_for_instance(context, instance.uuid)
  1763. try:
  1764. self._delete_provider(rp_uuid, global_request_id=context.global_id)
  1765. except (exception.ResourceProviderInUse,
  1766. exception.ResourceProviderDeletionFailed):
  1767. # TODO(efried): Raise these. Right now this is being left a no-op
  1768. # for backward compatibility.
  1769. pass
  1770. @safe_connect
  1771. def _get_provider_by_name(self, context, name):
  1772. """Queries the placement API for resource provider information matching
  1773. a supplied name.
  1774. :param context: The security context
  1775. :param name: Name of the resource provider to look up
  1776. :return: A dict of resource provider information including the
  1777. provider's UUID and generation
  1778. :raises: `exception.ResourceProviderNotFound` when no such provider was
  1779. found
  1780. """
  1781. resp = self.get("/resource_providers?name=%s" % name,
  1782. global_request_id=context.global_id)
  1783. if resp.status_code == 200:
  1784. data = resp.json()
  1785. records = data['resource_providers']
  1786. num_recs = len(records)
  1787. if num_recs == 1:
  1788. return records[0]
  1789. elif num_recs > 1:
  1790. msg = ("Found multiple resource provider records for resource "
  1791. "provider name %(rp_name)s: %(rp_uuids)s. "
  1792. "This should not happen.")
  1793. LOG.warning(msg, {
  1794. 'rp_name': name,
  1795. 'rp_uuids': ','.join([r['uuid'] for r in records])
  1796. })
  1797. elif resp.status_code != 404:
  1798. msg = ("Failed to retrieve resource provider information by name "
  1799. "for resource provider %s. Got %d: %s")
  1800. LOG.warning(msg, name, resp.status_code, resp.text)
  1801. raise exception.ResourceProviderNotFound(name_or_uuid=name)
  1802. @retrying.retry(stop_max_attempt_number=4,
  1803. retry_on_exception=lambda e: isinstance(
  1804. e, exception.ResourceProviderUpdateConflict))
  1805. def aggregate_add_host(self, context, agg_uuid, host_name):
  1806. """Looks up a resource provider by the supplied host name, and adds the
  1807. aggregate with supplied UUID to that resource provider.
  1808. :note: This method does NOT use the cached provider tree. It is only
  1809. called from the Compute API when a nova host aggregate is
  1810. modified
  1811. :param context: The security context
  1812. :param agg_uuid: UUID of the aggregate being modified
  1813. :param host_name: Name of the nova-compute service worker to look up a
  1814. resource provider for
  1815. :raises: `exceptions.ResourceProviderNotFound` if no resource provider
  1816. matching the host name could be found from the placement API
  1817. :raises: `exception.ResourceProviderAggregateRetrievalFailed` when
  1818. failing to get a provider's existing aggregates
  1819. :raises: `exception.ResourceProviderUpdateFailed` if there was a
  1820. failure attempting to save the provider aggregates
  1821. :raises: `exception.ResourceProviderUpdateConflict` if a concurrent
  1822. update to the provider was detected.
  1823. """
  1824. rp = self._get_provider_by_name(context, host_name)
  1825. # NOTE(jaypipes): Unfortunately, due to @safe_connect,
  1826. # _get_provider_by_name() can return None. If that happens, raise an
  1827. # error so we can trap for it in the Nova API code and ignore in Rocky,
  1828. # blow up in Stein.
  1829. if rp is None:
  1830. raise exception.PlacementAPIConnectFailure()
  1831. rp_uuid = rp['uuid']
  1832. # Now attempt to add the aggregate to the resource provider. We don't
  1833. # want to overwrite any other aggregates the provider may be associated
  1834. # with, however, so we first grab the list of aggregates for this
  1835. # provider and add the aggregate to the list of aggregates it already
  1836. # has
  1837. agg_info = self._get_provider_aggregates(context, rp_uuid)
  1838. # @safe_connect can make the above return None
  1839. if agg_info is None:
  1840. raise exception.PlacementAPIConnectFailure()
  1841. existing_aggs, gen = agg_info.aggregates, agg_info.generation
  1842. if agg_uuid in existing_aggs:
  1843. return
  1844. new_aggs = existing_aggs | set([agg_uuid])
  1845. self.set_aggregates_for_provider(
  1846. context, rp_uuid, new_aggs, use_cache=False, generation=gen)
  1847. @retrying.retry(stop_max_attempt_number=4,
  1848. retry_on_exception=lambda e: isinstance(
  1849. e, exception.ResourceProviderUpdateConflict))
  1850. def aggregate_remove_host(self, context, agg_uuid, host_name):
  1851. """Looks up a resource provider by the supplied host name, and removes
  1852. the aggregate with supplied UUID from that resource provider.
  1853. :note: This method does NOT use the cached provider tree. It is only
  1854. called from the Compute API when a nova host aggregate is
  1855. modified
  1856. :param context: The security context
  1857. :param agg_uuid: UUID of the aggregate being modified
  1858. :param host_name: Name of the nova-compute service worker to look up a
  1859. resource provider for
  1860. :raises: `exceptions.ResourceProviderNotFound` if no resource provider
  1861. matching the host name could be found from the placement API
  1862. :raises: `exception.ResourceProviderAggregateRetrievalFailed` when
  1863. failing to get a provider's existing aggregates
  1864. :raises: `exception.ResourceProviderUpdateFailed` if there was a
  1865. failure attempting to save the provider aggregates
  1866. :raises: `exception.ResourceProviderUpdateConflict` if a concurrent
  1867. update to the provider was detected.
  1868. """
  1869. rp = self._get_provider_by_name(context, host_name)
  1870. # NOTE(jaypipes): Unfortunately, due to @safe_connect,
  1871. # _get_provider_by_name() can return None. If that happens, raise an
  1872. # error so we can trap for it in the Nova API code and ignore in Rocky,
  1873. # blow up in Stein.
  1874. if rp is None:
  1875. raise exception.PlacementAPIConnectFailure()
  1876. rp_uuid = rp['uuid']
  1877. # Now attempt to remove the aggregate from the resource provider. We
  1878. # don't want to overwrite any other aggregates the provider may be
  1879. # associated with, however, so we first grab the list of aggregates for
  1880. # this provider and remove the aggregate from the list of aggregates it
  1881. # already has
  1882. agg_info = self._get_provider_aggregates(context, rp_uuid)
  1883. # @safe_connect can make the above return None
  1884. if agg_info is None:
  1885. raise exception.PlacementAPIConnectFailure()
  1886. existing_aggs, gen = agg_info.aggregates, agg_info.generation
  1887. if agg_uuid not in existing_aggs:
  1888. return
  1889. new_aggs = existing_aggs - set([agg_uuid])
  1890. self.set_aggregates_for_provider(
  1891. context, rp_uuid, new_aggs, use_cache=False, generation=gen)