1631 lines
58 KiB
Python
1631 lines
58 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2013 Mirantis, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
"""
|
|
Node-related objects and collections
|
|
"""
|
|
|
|
import collections
|
|
import copy
|
|
from datetime import datetime
|
|
import itertools
|
|
import math
|
|
import operator
|
|
import traceback
|
|
|
|
from netaddr import IPAddress
|
|
from netaddr import IPNetwork
|
|
from oslo_serialization import jsonutils
|
|
import six
|
|
import sqlalchemy as sa
|
|
|
|
from nailgun import consts
|
|
from nailgun.db import db
|
|
from nailgun.db.sqlalchemy import models
|
|
from nailgun import errors
|
|
from nailgun.extensions import fire_callback_on_node_collection_delete
|
|
from nailgun.extensions import fire_callback_on_node_create
|
|
from nailgun.extensions import fire_callback_on_node_delete
|
|
from nailgun.extensions import fire_callback_on_node_reset
|
|
from nailgun.extensions import fire_callback_on_node_update
|
|
from nailgun.extensions import fire_callback_on_remove_node_from_cluster
|
|
from nailgun.extensions.network_manager.template import NetworkTemplate
|
|
from nailgun.extensions.network_manager import utils as network_utils
|
|
from nailgun.logger import logger
|
|
from nailgun.objects import Bond
|
|
from nailgun.objects import Cluster
|
|
from nailgun.objects import NailgunCollection
|
|
from nailgun.objects import NailgunObject
|
|
from nailgun.objects import NetworkGroup
|
|
from nailgun.objects import NIC
|
|
from nailgun.objects import Notification
|
|
from nailgun.objects import Release
|
|
from nailgun.objects.serializers.node import NodeSerializer
|
|
from nailgun.plugins.manager import PluginManager
|
|
from nailgun.policy import cpu_distribution
|
|
from nailgun.policy import hugepages_distribution
|
|
from nailgun.settings import settings
|
|
from nailgun import utils
|
|
|
|
|
|
class Node(NailgunObject):
|
|
"""Node object"""
|
|
|
|
#: SQLAlchemy model for Node
|
|
model = models.Node
|
|
|
|
#: Serializer for Node
|
|
serializer = NodeSerializer
|
|
|
|
@classmethod
|
|
def get_status(cls, instance):
|
|
"""Get node state which is calculated from current state."""
|
|
# return transition statuses as is
|
|
if instance.status in (
|
|
consts.NODE_STATUSES.deploying,
|
|
consts.NODE_STATUSES.provisioning,
|
|
consts.NODE_STATUSES.removing
|
|
):
|
|
return instance.status
|
|
# if progress means that node in progress state,
|
|
# to avoid population of new status use deploying
|
|
if 0 < instance.progress < 100:
|
|
if instance.pending_deletion:
|
|
return consts.NODE_STATUSES.removing
|
|
if instance.pending_addition:
|
|
return consts.NODE_STATUSES.provisioning
|
|
return consts.NODE_STATUSES.deploying
|
|
|
|
return instance.status
|
|
|
|
@classmethod
|
|
def delete(cls, instance):
|
|
fire_callback_on_node_delete(instance)
|
|
super(Node, cls).delete(instance)
|
|
|
|
@classmethod
|
|
def set_netgroups_ids(cls, instance, netgroups_id_mapping):
|
|
"""Set IP network group assignment based on provided mapping.
|
|
|
|
:param instance: Node instance
|
|
:param netgroups_id_mapping: dict of network IDs to network IDs
|
|
:returns: None
|
|
"""
|
|
ip_addrs = db().query(models.IPAddr).filter(
|
|
models.IPAddr.node == instance.id)
|
|
for ip_addr in ip_addrs:
|
|
ip_addr.network = netgroups_id_mapping[ip_addr.network]
|
|
|
|
@classmethod
|
|
def set_nic_assignment_netgroups_ids(cls, instance, netgroups_id_mapping):
|
|
"""Set network group mapping for NIC assignments.
|
|
|
|
:param instance: Node instance
|
|
:param netgroups_id_mapping: dict of network IDs to network IDs
|
|
:returns: None
|
|
"""
|
|
nic_assignments = db.query(models.NetworkNICAssignment).\
|
|
join(models.NodeNICInterface).\
|
|
filter(models.NodeNICInterface.node_id == instance.id)
|
|
for nic_assignment in nic_assignments:
|
|
nic_assignment.network_id = \
|
|
netgroups_id_mapping[nic_assignment.network_id]
|
|
|
|
@classmethod
|
|
def clear_bonds(cls, instance):
|
|
instance.bond_interfaces = []
|
|
|
|
@classmethod
|
|
def set_bond_assignment_netgroups_ids(cls, instance, netgroups_id_mapping):
|
|
"""Set network group mapping for bond assignments.
|
|
|
|
:param instance: Node instance
|
|
:param netgroups_id_mapping: dict
|
|
:returns: None
|
|
"""
|
|
bond_assignments = db.query(models.NetworkBondAssignment).\
|
|
join(models.NodeBondInterface).\
|
|
filter(models.NodeBondInterface.node_id == instance.id)
|
|
for bond_assignment in bond_assignments:
|
|
bond_assignment.network_id = \
|
|
netgroups_id_mapping[bond_assignment.network_id]
|
|
|
|
@classmethod
|
|
def get_interface_by_mac_or_name(cls, instance, mac=None, name=None):
|
|
# try to get interface by mac address
|
|
interface = next((
|
|
n for n in instance.nic_interfaces
|
|
if network_utils.is_same_mac(n.mac, mac)), None)
|
|
|
|
# try to get interface instance by interface name. this protects
|
|
# us from loosing nodes when some NICs was replaced with a new one
|
|
interface = interface or next((
|
|
n for n in instance.nic_interfaces if n.name == name), None)
|
|
return interface
|
|
|
|
@classmethod
|
|
def get_all_node_ips(cls):
|
|
"""Get all Admin IPs assigned to nodes.
|
|
|
|
:returns: List of single element tuples with IP address
|
|
"""
|
|
return db().query(models.Node.ip)
|
|
|
|
@classmethod
|
|
def get_nodes_by_role(cls, cluster_ids, role_name):
|
|
return db().query(models.Node).filter(
|
|
models.Node.cluster_id.in_(cluster_ids)
|
|
).filter(sa.or_(
|
|
models.Node.roles.any(role_name),
|
|
models.Node.pending_roles.any(role_name)
|
|
))
|
|
|
|
@classmethod
|
|
def get_by_mac_or_uid(cls, mac=None, node_uid=None):
|
|
"""Get Node instance by MAC or ID.
|
|
|
|
:param mac: MAC address as string
|
|
:param node_uid: Node ID
|
|
:returns: Node instance
|
|
"""
|
|
node = None
|
|
if not mac and not node_uid:
|
|
return node
|
|
|
|
q = db().query(cls.model)
|
|
if mac:
|
|
node = q.filter_by(mac=mac.lower()).first()
|
|
else:
|
|
node = q.get(node_uid)
|
|
return node
|
|
|
|
@classmethod
|
|
def get_by_hostname(cls, hostname, cluster_id):
|
|
"""Get Node instance by hostname.
|
|
|
|
:param hostname: hostname as string
|
|
:param cluster_id: Node will be searched \
|
|
only within the cluster with this ID.
|
|
:returns: Node instance
|
|
"""
|
|
|
|
if not hostname:
|
|
return None
|
|
|
|
q = db().query(cls.model).filter_by(
|
|
hostname=hostname, cluster_id=cluster_id)
|
|
return q.first()
|
|
|
|
@classmethod
|
|
def get_by_meta(cls, meta):
|
|
"""Search for instance using mac, node id or interfaces.
|
|
|
|
:param meta: dict with nodes metadata
|
|
:returns: Node instance
|
|
"""
|
|
node = cls.get_by_mac_or_uid(
|
|
mac=meta.get('mac'), node_uid=meta.get('id'))
|
|
|
|
if not node:
|
|
can_search_by_ifaces = all([
|
|
meta.get('meta'), meta['meta'].get('interfaces')])
|
|
|
|
if can_search_by_ifaces:
|
|
node = cls.search_by_interfaces(meta['meta']['interfaces'])
|
|
|
|
return node
|
|
|
|
@classmethod
|
|
def search_by_interfaces(cls, interfaces):
|
|
"""Search for instance using MACs on interfaces.
|
|
|
|
:param interfaces: dict of Node interfaces
|
|
:returns: Node instance
|
|
"""
|
|
return db().query(cls.model).join(
|
|
models.NodeNICInterface,
|
|
cls.model.nic_interfaces
|
|
).filter(
|
|
models.NodeNICInterface.mac.in_(
|
|
[n["mac"].lower() for n in interfaces]
|
|
)
|
|
).first()
|
|
|
|
@classmethod
|
|
def should_have_public_with_ip(cls, instance, roles_metadata=None):
|
|
"""Returns True if node should have IP belonging to Public network.
|
|
|
|
:param instance: Node DB instance
|
|
:returns: True when node has Public network
|
|
"""
|
|
if Cluster.should_assign_public_to_all_nodes(instance.cluster):
|
|
return True
|
|
|
|
roles = itertools.chain(instance.roles, instance.pending_roles)
|
|
if not roles_metadata:
|
|
roles_metadata = Cluster.get_roles(instance.cluster)
|
|
|
|
for role in roles:
|
|
if roles_metadata.get(role, {}).get('public_ip_required'):
|
|
return True
|
|
|
|
return False
|
|
|
|
@classmethod
|
|
def should_have_public(cls, instance, roles_metadata=None):
|
|
"""Determine whether this node should be connected to Public network,
|
|
|
|
no matter with or without an IP address assigned from that network
|
|
|
|
For example Neutron DVR does require Public network access on compute
|
|
nodes, but does not require IP address assigned to external bridge.
|
|
|
|
:param instance: Node DB instance
|
|
:returns: True when node has Public network
|
|
"""
|
|
roles_metadata = roles_metadata or Cluster.get_roles(instance.cluster)
|
|
if cls.should_have_public_with_ip(instance, roles_metadata):
|
|
return True
|
|
|
|
dvr_enabled = Cluster.neutron_dvr_enabled(instance.cluster)
|
|
if dvr_enabled:
|
|
roles = itertools.chain(instance.roles, instance.pending_roles)
|
|
|
|
for role in roles:
|
|
if roles_metadata.get(role, {}).get('public_for_dvr_required'):
|
|
return True
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def get_interfaces_without_bonds_slaves(node):
|
|
ifaces = set(node.interfaces)
|
|
for bond in node.bond_interfaces:
|
|
ifaces ^= set(bond.slaves)
|
|
return sorted(ifaces, key=operator.attrgetter('name'))
|
|
|
|
@classmethod
|
|
def create(cls, data):
|
|
"""Create Node instance with specified parameters in DB.
|
|
|
|
This includes:
|
|
|
|
* generating its name by MAC (if name is not specified in data)
|
|
* adding node to Cluster (if cluster_id is not None in data) \
|
|
(see :func:`add_into_cluster`) with specified roles \
|
|
(see :func:`update_roles` and :func:`update_pending_roles`)
|
|
* creating interfaces for Node in DB (see :func:`update_interfaces`)
|
|
* creating default Node attributes (see :func:`create_attributes`)
|
|
* creating Notification about newly discovered Node \
|
|
(see :func:`create_discover_notification`)
|
|
|
|
:param data: dictionary of key-value pairs as object fields
|
|
:returns: Node instance
|
|
"""
|
|
if "name" not in data:
|
|
data["name"] = "Untitled ({0})".format(
|
|
data['mac'][-5:].lower()
|
|
)
|
|
data["timestamp"] = datetime.now()
|
|
data.pop("id", None)
|
|
|
|
# TODO(enchantner): fix this temporary hack in clients
|
|
if "cluster_id" not in data and "cluster" in data:
|
|
cluster_id = data.pop("cluster", None)
|
|
data["cluster_id"] = cluster_id
|
|
|
|
roles = data.pop("roles", None)
|
|
pending_roles = data.pop("pending_roles", None)
|
|
primary_tags = data.pop("primary_tags", None)
|
|
|
|
new_node_meta = data.pop("meta", {})
|
|
new_node_cluster_id = data.pop("cluster_id", None)
|
|
new_node = super(Node, cls).create(data)
|
|
new_node.create_meta(new_node_meta)
|
|
|
|
if 'hostname' not in data:
|
|
new_node.hostname = \
|
|
cls.get_unique_hostname(new_node, new_node_cluster_id)
|
|
db().flush()
|
|
|
|
# Add interfaces for node from 'meta'.
|
|
if new_node.meta and new_node.meta.get('interfaces'):
|
|
cls.update_interfaces(new_node)
|
|
|
|
# role cannot be assigned if cluster_id is not set
|
|
if new_node_cluster_id:
|
|
new_node.cluster_id = new_node_cluster_id
|
|
# updating roles
|
|
if roles is not None:
|
|
cls.update_roles(new_node, roles)
|
|
if pending_roles is not None:
|
|
cls.update_pending_roles(new_node, pending_roles)
|
|
if primary_tags is not None:
|
|
cls.update_primary_tags(new_node, primary_tags)
|
|
|
|
# adding node into cluster
|
|
if new_node_cluster_id:
|
|
cls.add_into_cluster(new_node, new_node_cluster_id)
|
|
|
|
# creating attributes
|
|
cls.create_discover_notification(new_node)
|
|
|
|
if new_node.ip:
|
|
cls.check_ip_belongs_to_any_admin_network(new_node)
|
|
|
|
fire_callback_on_node_create(new_node)
|
|
|
|
return new_node
|
|
|
|
@classmethod
|
|
def set_error_status_and_file_notification(cls, instance, etype, emessage):
|
|
instance.status = consts.NODE_STATUSES.error
|
|
instance.error_type = etype
|
|
instance.error_msg = emessage
|
|
db().flush()
|
|
Notification.create({
|
|
"topic": consts.NOTIFICATION_TOPICS.error,
|
|
"message": instance.error_msg,
|
|
"node_id": instance.id
|
|
})
|
|
|
|
@classmethod
|
|
def check_ip_belongs_to_any_admin_network(cls, instance, new_ip=None):
|
|
"""Checks that node's IP belongs to any of Admin networks IP ranges.
|
|
|
|
Node can be inside or out of a cluster. Set node to error and file a
|
|
notification if node's IP does not belong to any of Admin networks.
|
|
|
|
:param instance: node instance
|
|
:param new_ip: new IP for a node (got from Nailgun agent)
|
|
:return: True if IP belongs to any of Admin networks
|
|
"""
|
|
ip = new_ip or instance.ip
|
|
admin_ranges_db = db().query(
|
|
models.IPAddrRange.first,
|
|
models.IPAddrRange.last
|
|
).join(
|
|
models.NetworkGroup
|
|
).filter(
|
|
models.NetworkGroup.name == consts.NETWORKS.fuelweb_admin
|
|
)
|
|
nm = Cluster.get_network_manager(instance.cluster)
|
|
match = nm.check_ips_belong_to_ranges([ip], admin_ranges_db)
|
|
if not match:
|
|
cls.set_error_status_and_file_notification(
|
|
instance,
|
|
consts.NODE_ERRORS.discover,
|
|
"Node '{0}' has IP '{1}' that does not match any Admin "
|
|
"network".format(instance.hostname, ip)
|
|
)
|
|
return match
|
|
|
|
@classmethod
|
|
def check_ip_belongs_to_own_admin_network(cls, instance, new_ip=None):
|
|
"""Checks that node's IP belongs to node's Admin network IP ranges.
|
|
|
|
Node should be inside a cluster. Set node to error and file a
|
|
notification if node's IP does not belong to its Admin network.
|
|
|
|
:param instance: node instance
|
|
:param new_ip: new IP for a node (got from Nailgun agent)
|
|
:return: True if IP belongs to node's Admin network
|
|
"""
|
|
ip = new_ip or instance.ip
|
|
nm = Cluster.get_network_manager(instance.cluster)
|
|
admin_ng = NetworkGroup.get_admin_network_group(instance)
|
|
match = nm.is_same_network(ip, admin_ng.cidr)
|
|
if not match:
|
|
cls.set_error_status_and_file_notification(
|
|
instance,
|
|
consts.NODE_ERRORS.discover,
|
|
"Node '{0}' has IP '{1}' that does not match its own Admin "
|
|
"network '{2}'".format(instance.hostname, ip, admin_ng.cidr)
|
|
)
|
|
return match
|
|
|
|
@classmethod
|
|
def assign_group(cls, instance):
|
|
if instance.group_id is None and instance.ip:
|
|
query = db().query(models.NetworkGroup.cidr,
|
|
models.NetworkGroup.group_id).join(
|
|
models.NodeGroup.networks
|
|
).filter(
|
|
models.NetworkGroup.name == "fuelweb_admin",
|
|
models.NetworkGroup.group_id is not None,
|
|
# Only node group of the same cluster can be selected.
|
|
models.NodeGroup.cluster_id == instance.cluster_id
|
|
)
|
|
ip = IPAddress(instance.ip)
|
|
for cidr, group_id in query:
|
|
if ip in IPNetwork(cidr):
|
|
instance.group_id = group_id
|
|
break
|
|
|
|
if not instance.group_id:
|
|
instance.group_id = Cluster.get_default_group(instance.cluster).id
|
|
|
|
db().add(instance)
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def is_interfaces_configuration_locked(cls, instance, is_agent=False):
|
|
"""Returns true if update of network configuration is not allowed.
|
|
|
|
Configuration update is allowed in 'discover', 'stopped',
|
|
'ready' and discovery 'error'.
|
|
|
|
We are allowing configuration of the 'stopped' and 'ready' node to be
|
|
updated by API but not nailgun-agent.
|
|
|
|
:param instance: node instance
|
|
:type instance: models.Node
|
|
:param is_agent: is nailgun-agent
|
|
:type is_agent: bool
|
|
:return: is locked
|
|
:rtype: bool
|
|
"""
|
|
if is_agent:
|
|
unlocked_cluster_statuses = (
|
|
consts.NODE_STATUSES.discover,
|
|
consts.NODE_STATUSES.error,
|
|
)
|
|
unlocked_node_error_types = (
|
|
consts.NODE_ERRORS.discover,
|
|
)
|
|
else:
|
|
unlocked_cluster_statuses = (
|
|
consts.NODE_STATUSES.discover,
|
|
consts.NODE_STATUSES.error,
|
|
consts.NODE_STATUSES.stopped,
|
|
consts.NODE_STATUSES.ready
|
|
)
|
|
unlocked_node_error_types = (
|
|
consts.NODE_ERRORS.discover,
|
|
consts.NODE_ERRORS.deploy
|
|
)
|
|
|
|
return instance.status not in unlocked_cluster_statuses or (
|
|
instance.status == consts.NODE_STATUSES.error and
|
|
instance.error_type not in unlocked_node_error_types
|
|
)
|
|
|
|
@classmethod
|
|
def hardware_info_locked(cls, instance):
|
|
"""Returns true if update of hardware information is not allowed.
|
|
|
|
It is not allowed during provision/deployment, after
|
|
successful provision/deployment and during node removal.
|
|
"""
|
|
return instance.status not in (
|
|
consts.NODE_STATUSES.discover,
|
|
consts.NODE_STATUSES.error,
|
|
)
|
|
|
|
@classmethod
|
|
def update_interfaces(cls, instance):
|
|
"""Update interfaces for Node instance using Cluster
|
|
|
|
network manager (see :func:`get_network_manager`)
|
|
|
|
:param instance: Node instance
|
|
:returns: None
|
|
"""
|
|
try:
|
|
network_manager = Cluster.get_network_manager(instance.cluster)
|
|
network_manager.update_interfaces_info(instance)
|
|
db().refresh(instance)
|
|
except errors.InvalidInterfacesInfo as exc:
|
|
logger.warning(
|
|
"Failed to update interfaces for node '%s' - invalid info "
|
|
"in meta: %s", instance.human_readable_name, exc.message
|
|
)
|
|
logger.warning(traceback.format_exc())
|
|
|
|
@classmethod
|
|
def create_discover_notification(cls, instance):
|
|
"""Create notification about discovering new Node.
|
|
|
|
:param instance: Node instance
|
|
:returns: None
|
|
"""
|
|
try:
|
|
# we use multiplier of 1024 because there are no problems here
|
|
# with unfair size calculation
|
|
ram = str(round(float(
|
|
instance.meta['memory']['total']) / 1073741824, 1)) + " GB RAM"
|
|
except Exception:
|
|
logger.warning(traceback.format_exc())
|
|
ram = "unknown RAM"
|
|
|
|
try:
|
|
# we use multiplier of 1000 because disk vendors specify HDD size
|
|
# in terms of decimal capacity. Sources:
|
|
# http://knowledge.seagate.com/articles/en_US/FAQ/172191en
|
|
# http://physics.nist.gov/cuu/Units/binary.html
|
|
hd_size = round(
|
|
float(
|
|
sum(
|
|
[d["size"] for d in instance.meta["disks"]]
|
|
) / 1000000000
|
|
),
|
|
1
|
|
)
|
|
# if HDD > 100 GB we show it's size in TB
|
|
if hd_size > 100:
|
|
hd_size = str(hd_size / 1000) + " TB HDD"
|
|
else:
|
|
hd_size = str(hd_size) + " GB HDD"
|
|
except Exception:
|
|
logger.warning(traceback.format_exc())
|
|
hd_size = "unknown HDD"
|
|
|
|
cores = str(instance.meta.get('cpu', {}).get('total', "unknown"))
|
|
|
|
Notification.create({
|
|
"topic": "discover",
|
|
"message": u"New node is discovered: "
|
|
u"{0} CPUs / {1} / {2} ".format(cores, ram, hd_size),
|
|
"node_id": instance.id
|
|
})
|
|
|
|
@classmethod
|
|
def update(cls, instance, data):
|
|
"""Update Node instance with specified parameters in DB.
|
|
|
|
This includes:
|
|
|
|
* adding node to Cluster (if cluster_id is not None in data) \
|
|
(see :func:`add_into_cluster`)
|
|
* updating roles for Node if it belongs to Cluster \
|
|
(see :func:`update_roles` and :func:`update_pending_roles`)
|
|
* removing node from Cluster (if cluster_id is None in data) \
|
|
(see :func:`remove_from_cluster`)
|
|
* updating interfaces for Node in DB (see :func:`update_interfaces`)
|
|
* creating default Node attributes (see :func:`create_attributes`)
|
|
|
|
:param data: dictionary of key-value pairs as object fields
|
|
:returns: Node instance
|
|
"""
|
|
data.pop("id", None)
|
|
data.pop("network_data", None)
|
|
|
|
roles = data.pop("roles", None)
|
|
pending_roles = data.pop("pending_roles", None)
|
|
new_meta = data.pop("meta", None)
|
|
|
|
disks_changed = None
|
|
if new_meta and "disks" in new_meta and "disks" in instance.meta:
|
|
key = operator.itemgetter("name")
|
|
|
|
new_disks = sorted(new_meta["disks"], key=key)
|
|
old_disks = sorted(instance.meta["disks"], key=key)
|
|
|
|
disks_changed = (new_disks != old_disks)
|
|
|
|
# TODO(enchantner): fix this temporary hack in clients
|
|
if "cluster_id" not in data and "cluster" in data:
|
|
cluster_id = data.pop("cluster", None)
|
|
data["cluster_id"] = cluster_id
|
|
|
|
if new_meta:
|
|
instance.update_meta(new_meta)
|
|
# The call to update_interfaces will execute a select query for
|
|
# the current instance. This appears to overwrite the object in the
|
|
# current session and we lose the meta changes.
|
|
db().flush()
|
|
is_agent = bool(data.pop('is_agent', None))
|
|
if cls.is_interfaces_configuration_locked(instance, is_agent):
|
|
logger.debug("Interfaces are locked for update on node %s",
|
|
instance.human_readable_name)
|
|
else:
|
|
instance.ip = data.pop("ip", None) or instance.ip
|
|
instance.mac = data.pop("mac", None) or instance.mac
|
|
db().flush()
|
|
cls.update_interfaces(instance)
|
|
|
|
cluster_changed = False
|
|
add_to_cluster = False
|
|
if "cluster_id" in data:
|
|
new_cluster_id = data.pop("cluster_id")
|
|
if instance.cluster_id:
|
|
if new_cluster_id is None:
|
|
# removing node from cluster
|
|
cluster_changed = True
|
|
cls.remove_from_cluster(instance)
|
|
elif new_cluster_id != instance.cluster_id:
|
|
# changing node cluster to another
|
|
# (is currently not allowed)
|
|
raise errors.CannotUpdate(
|
|
u"Changing cluster on the fly is not allowed"
|
|
)
|
|
else:
|
|
if new_cluster_id is not None:
|
|
# assigning node to cluster
|
|
cluster_changed = True
|
|
add_to_cluster = True
|
|
instance.cluster_id = new_cluster_id
|
|
|
|
if "group_id" in data:
|
|
new_group_id = data.pop("group_id")
|
|
if instance.group_id != new_group_id:
|
|
nm = Cluster.get_network_manager(instance.cluster)
|
|
nm.clear_assigned_networks(instance)
|
|
nm.clear_bond_configuration(instance)
|
|
instance.group_id = new_group_id
|
|
add_to_cluster = True
|
|
|
|
# calculating flags
|
|
roles_changed = (
|
|
roles is not None and set(roles) != set(instance.roles)
|
|
)
|
|
pending_roles_changed = (
|
|
pending_roles is not None and
|
|
set(pending_roles) != set(instance.pending_roles)
|
|
)
|
|
|
|
super(Node, cls).update(instance, data)
|
|
|
|
# we need to update relationships, if fk's are changed
|
|
db.refresh(instance)
|
|
|
|
if roles_changed:
|
|
cls.update_roles(instance, roles)
|
|
if pending_roles_changed:
|
|
cls.update_pending_roles(instance, pending_roles)
|
|
|
|
if add_to_cluster:
|
|
cls.add_into_cluster(instance, instance.cluster_id)
|
|
|
|
if any((
|
|
roles_changed,
|
|
pending_roles_changed,
|
|
cluster_changed,
|
|
disks_changed,
|
|
)) and instance.status not in (
|
|
consts.NODE_STATUSES.provisioning,
|
|
consts.NODE_STATUSES.deploying
|
|
):
|
|
# TODO(eli): we somehow should move this
|
|
# condition into extension, in order to do
|
|
# that probably we will have to create separate
|
|
# table to keep disks which were used to create
|
|
# volumes mapping.
|
|
# Should be solved as a part of blueprint
|
|
# https://blueprints.launchpad.net/fuel/+spec
|
|
# /volume-manager-refactoring
|
|
fire_callback_on_node_update(instance)
|
|
|
|
return instance
|
|
|
|
@classmethod
|
|
def reset_to_discover(cls, instance):
|
|
"""Flush database objects which is not consistent with actual node
|
|
|
|
configuration in the event of resetting node to discover state
|
|
|
|
:param instance: Node database object
|
|
:returns: None
|
|
"""
|
|
node_data = {
|
|
"status": consts.NODE_STATUSES.discover,
|
|
"pending_addition": True,
|
|
"pending_deletion": False,
|
|
"progress": 0
|
|
}
|
|
cls.update(instance, node_data)
|
|
cls.move_roles_to_pending_roles(instance)
|
|
# when node reseted to discover:
|
|
# - cobbler system is deleted
|
|
# - mac to ip mapping from dnsmasq.conf is deleted
|
|
# imho we need to revert node to original state, as it was
|
|
# added to cluster (without any additonal state in database)
|
|
fire_callback_on_node_reset(instance)
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def update_cluster_assignment(cls, instance, cluster,
|
|
roles, pending_roles):
|
|
"""Update assignment of the node to the other cluster.
|
|
|
|
This method primarily used by the cluster_upgrade extension for
|
|
reassigning of a node. Be careful to use it outside of this extension
|
|
because node still plugged to networks of a previous cluster.
|
|
|
|
:param instance: An instance of :class:`Node`.
|
|
:param cluster: An instance of :class:`Cluster`.
|
|
:param roles: A list of roles to be assigned to the node.
|
|
:param pending_roles: A list of pending roles to be assigned to
|
|
the node.
|
|
"""
|
|
instance.cluster_id = cluster.id
|
|
instance.group_id = None
|
|
instance.kernel_params = None
|
|
cls.update_roles(instance, roles)
|
|
cls.update_primary_tags(instance, [])
|
|
cls.update_pending_roles(instance, pending_roles)
|
|
cls.remove_replaced_params(instance)
|
|
cls.assign_group(instance)
|
|
cls.set_network_template(instance)
|
|
cls.set_default_attributes(instance)
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def update_by_agent(cls, instance, data):
|
|
"""Update Node instance with some specific cases for agent.
|
|
|
|
* don't update provisioning or error state back to discover
|
|
* don't update volume information if disks arrays is empty
|
|
|
|
:param data: dictionary of key-value pairs as object fields
|
|
:returns: Node instance
|
|
"""
|
|
# don't update provisioning and error back to discover
|
|
data_status = data.get('status')
|
|
if instance.status in ('provisioning', 'error'):
|
|
if data.get('status', 'discover') == 'discover':
|
|
logger.debug(
|
|
u"Node {0} has provisioning or error status - "
|
|
u"status not updated by agent".format(
|
|
instance.human_readable_name
|
|
)
|
|
)
|
|
|
|
data.pop('status', None)
|
|
|
|
meta = data.get('meta', {})
|
|
# don't update volume information, if agent has sent an empty array
|
|
if len(meta.get('disks', [])) == 0 and instance.meta.get('disks'):
|
|
|
|
logger.warning(
|
|
u'Node {0} has received an empty disks array - '
|
|
u'volume information will not be updated'.format(
|
|
instance.human_readable_name
|
|
)
|
|
)
|
|
meta['disks'] = instance.meta['disks']
|
|
|
|
# don't update volume information, if it is locked by node status
|
|
if 'disks' in meta and cls.hardware_info_locked(instance):
|
|
logger.debug("Volume information is locked for update on node %s",
|
|
instance.human_readable_name)
|
|
meta['disks'] = instance.meta['disks']
|
|
|
|
if not cls.is_interfaces_configuration_locked(instance, is_agent=True)\
|
|
and data.get('ip'):
|
|
if instance.cluster_id:
|
|
update_status = cls.check_ip_belongs_to_own_admin_network(
|
|
instance, data['ip'])
|
|
else:
|
|
update_status = cls.check_ip_belongs_to_any_admin_network(
|
|
instance, data['ip'])
|
|
if update_status:
|
|
if instance.status == consts.NODE_STATUSES.error and \
|
|
instance.error_type == consts.NODE_ERRORS.discover:
|
|
# accept the status from agent if the node had wrong IP
|
|
# previously
|
|
if data_status:
|
|
instance.status = data_status
|
|
else:
|
|
instance.status = consts.NODE_STATUSES.discover
|
|
|
|
if data_status != consts.NODE_STATUSES.error:
|
|
instance.error_type = None
|
|
else:
|
|
data.pop('status', None)
|
|
return cls.update(instance, data)
|
|
|
|
@classmethod
|
|
def update_roles(cls, instance, new_roles):
|
|
"""Update roles for Node instance.
|
|
|
|
Logs an error if node doesn't belong to Cluster
|
|
|
|
:param instance: Node instance
|
|
:param new_roles: list of new role names
|
|
:returns: None
|
|
"""
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to assign roles to node "
|
|
u"'{0}' which isn't added to cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
logger.debug(
|
|
u"Updating roles for node {0}: {1}".format(
|
|
instance.full_name,
|
|
new_roles))
|
|
|
|
instance.roles = new_roles
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def update_pending_roles(cls, instance, new_pending_roles):
|
|
"""Update pending_roles for Node instance.
|
|
|
|
Logs an error if node doesn't belong to Cluster
|
|
|
|
:param instance: Node instance
|
|
:param new_pending_roles: list of new pending role names
|
|
:returns: None
|
|
"""
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to assign pending roles to node "
|
|
u"'{0}' which isn't added to cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
logger.debug(
|
|
u"Updating pending roles for node {0}: {1}".format(
|
|
instance.full_name,
|
|
new_pending_roles))
|
|
|
|
if new_pending_roles == []:
|
|
# TODO(enchantner): research why the hell we need this
|
|
Cluster.clear_pending_changes(
|
|
instance.cluster,
|
|
node_id=instance.id
|
|
)
|
|
|
|
instance.pending_roles = new_pending_roles
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def update_primary_tags(cls, instance, new_primary_tags):
|
|
"""Update primary_tags for Node instance.
|
|
|
|
Logs an error if node doesn't belong to Cluster
|
|
|
|
:param instance: Node instance
|
|
:param new_primary_tags: list of new primary tag names
|
|
:returns: None
|
|
"""
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to assign primary tags to node "
|
|
u"'{0}' which isn't added to cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
assigned_tags = set(cls.get_tags(instance))
|
|
missing_tags = set(new_primary_tags) - set(assigned_tags)
|
|
if missing_tags:
|
|
logger.warning(
|
|
u"Could not mark node {0} as primary for {1} tags, "
|
|
u"because corresponding roles are not assigned.".format(
|
|
instance.full_name, missing_tags)
|
|
)
|
|
return
|
|
|
|
logger.debug(
|
|
u"Updating primary tags for node {0}: {1}".format(
|
|
instance.full_name,
|
|
new_primary_tags))
|
|
|
|
instance.primary_tags = new_primary_tags
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def add_into_cluster(cls, instance, cluster_id):
|
|
"""Adds Node to Cluster by its ID.
|
|
|
|
Also assigns networks by default for Node.
|
|
|
|
:param instance: Node instance
|
|
:param cluster_id: Cluster ID
|
|
:returns: None
|
|
"""
|
|
instance.cluster_id = cluster_id
|
|
|
|
cls.assign_group(instance)
|
|
network_manager = Cluster.get_network_manager(instance.cluster)
|
|
network_manager.assign_networks_by_default(instance)
|
|
cls.refresh_dpdk_properties(instance)
|
|
cls.add_pending_change(instance, consts.CLUSTER_CHANGES.interfaces)
|
|
cls.set_network_template(instance)
|
|
cls.set_default_attributes(instance)
|
|
cls.create_nic_attributes(instance)
|
|
|
|
@classmethod
|
|
def set_network_template(cls, instance):
|
|
template = instance.cluster.network_config.configuration_template
|
|
cls.apply_network_template(instance, template)
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def add_pending_change(cls, instance, change):
|
|
"""Add pending change into Cluster.
|
|
|
|
:param instance: Node instance
|
|
:param change: string value of cluster change
|
|
:returns: None
|
|
"""
|
|
if instance.cluster:
|
|
Cluster.add_pending_changes(
|
|
instance.cluster, change, node_id=instance.id
|
|
)
|
|
|
|
@classmethod
|
|
def get_admin_physical_iface(cls, instance):
|
|
"""Returns node's physical iface.
|
|
|
|
In case if we have bonded admin iface, first
|
|
of the bonded ifaces will be returned
|
|
|
|
:param instance: Node instance
|
|
:returns: interface instance
|
|
"""
|
|
admin_iface = Cluster.get_network_manager(instance.cluster) \
|
|
.get_admin_interface(instance)
|
|
|
|
if admin_iface.type != consts.NETWORK_INTERFACE_TYPES.bond:
|
|
return admin_iface
|
|
|
|
for slave in admin_iface.slaves:
|
|
if slave.pxe or slave.mac == instance.mac:
|
|
return slave
|
|
|
|
return admin_iface.slaves[-1]
|
|
|
|
@classmethod
|
|
def remove_from_cluster(cls, instance):
|
|
"""Remove Node from Cluster.
|
|
|
|
Also drops networks assignment for Node and clears both
|
|
roles and pending roles
|
|
|
|
:param instance: Node instance
|
|
:returns: None
|
|
"""
|
|
fire_callback_on_remove_node_from_cluster(instance)
|
|
|
|
if instance.cluster:
|
|
Cluster.clear_pending_changes(
|
|
instance.cluster,
|
|
node_id=instance.id
|
|
)
|
|
cls.update_roles(instance, [])
|
|
cls.update_pending_roles(instance, [])
|
|
cls.remove_replaced_params(instance)
|
|
instance.cluster_id = None
|
|
instance.group_id = None
|
|
instance.kernel_params = None
|
|
instance.primary_tags = []
|
|
instance.hostname = cls.default_slave_name(instance)
|
|
instance.attributes = {}
|
|
|
|
from nailgun.objects import OpenstackConfig
|
|
OpenstackConfig.disable_by_nodes([instance])
|
|
|
|
db().flush()
|
|
db().refresh(instance)
|
|
|
|
@classmethod
|
|
def move_roles_to_pending_roles(cls, instance):
|
|
"""Move roles to pending_roles"""
|
|
instance.pending_roles = instance.pending_roles + instance.roles
|
|
instance.roles = []
|
|
instance.primary_tags = []
|
|
db().flush()
|
|
|
|
@classmethod
|
|
def get_slave_name(cls, instance):
|
|
if not instance.hostname:
|
|
return cls.default_slave_name(instance)
|
|
return instance.hostname
|
|
|
|
@classmethod
|
|
def default_slave_name(cls, instance):
|
|
return cls.permanent_id(instance)
|
|
|
|
@classmethod
|
|
def permanent_id(cls, instance):
|
|
return u"node-{node_id}".format(node_id=instance.id)
|
|
|
|
@classmethod
|
|
def generate_fqdn_by_hostname(cls, hostname):
|
|
return u"{instance_name}.{dns_domain}" \
|
|
.format(instance_name=hostname,
|
|
dns_domain=settings.DNS_DOMAIN)
|
|
|
|
@classmethod
|
|
def get_node_fqdn(cls, instance):
|
|
return cls.generate_fqdn_by_hostname(instance.hostname)
|
|
|
|
@classmethod
|
|
def get_kernel_params(cls, instance):
|
|
"""Get kernel params.
|
|
|
|
Returns kernel_params if they were replaced by custom ones.
|
|
Otherwise assemble kernel_params from cluster default
|
|
and node specific params: hugepages, sriov, isolcpus.
|
|
"""
|
|
|
|
if instance.kernel_params:
|
|
return instance.kernel_params
|
|
|
|
kernel_params = Cluster.get_default_kernel_params(instance.cluster)
|
|
|
|
if Release.is_nfv_supported(instance.cluster.release):
|
|
# Add intel_iommu=on amd_iommu=on if SR-IOV is enabled on node
|
|
for nic in instance.nic_interfaces:
|
|
if NIC.is_sriov_enabled(nic):
|
|
if 'intel_iommu=' not in kernel_params:
|
|
kernel_params += ' intel_iommu=on'
|
|
if 'amd_iommu=' not in kernel_params:
|
|
kernel_params += ' amd_iommu=on'
|
|
break
|
|
|
|
if ('hugepages' not in kernel_params
|
|
and NodeAttributes.is_hugepages_enabled(instance)):
|
|
kernel_params += NodeAttributes.hugepages_kernel_opts(instance)
|
|
|
|
if NodeAttributes.is_cpu_pinning_enabled(instance):
|
|
isolated_cpus = NodeAttributes.distribute_node_cpus(
|
|
instance)['isolated_cpus']
|
|
if isolated_cpus and 'isolcpus' not in kernel_params:
|
|
kernel_params += " isolcpus={0}".format(
|
|
",".join(six.moves.map(str, isolated_cpus)))
|
|
|
|
return kernel_params
|
|
|
|
@classmethod
|
|
def remove_replaced_params(cls, instance):
|
|
instance.replaced_deployment_info = []
|
|
instance.replaced_provisioning_info = {}
|
|
instance.network_template = None
|
|
|
|
@classmethod
|
|
def get_tags(cls, instance):
|
|
"""Get node tags
|
|
|
|
Returns list of nodes tags based on roles assigned to it.
|
|
Primary tags are included into result as it is.
|
|
"""
|
|
|
|
roles = set(instance.roles + instance.pending_roles)
|
|
roles_meta = Cluster.get_roles(instance.cluster)
|
|
tags = ()
|
|
for role in roles:
|
|
tags = itertools.chain(tags, roles_meta[role].get('tags', [role]))
|
|
return tags
|
|
|
|
@classmethod
|
|
def all_tags(cls, instance):
|
|
"""Get node tags including primary-tags with prefix
|
|
|
|
Returns list of nodes tags based on roles assigned to it.
|
|
Primary tags are included into result with 'primary-' prefix.
|
|
This method is mostly used for node's tasks resolution.
|
|
"""
|
|
|
|
tags = set(cls.get_tags(instance))
|
|
tags -= set(instance.primary_tags)
|
|
primary_tags = set([
|
|
'primary-{0}'.format(tag) for tag in instance.primary_tags])
|
|
|
|
return sorted(tags | primary_tags)
|
|
|
|
@classmethod
|
|
def apply_network_template(cls, instance, template):
|
|
if template is None:
|
|
instance.network_template = None
|
|
return
|
|
|
|
template_body = template['adv_net_template']
|
|
# Get the correct nic_mapping for this node so we can
|
|
# dynamically replace any interface references in any
|
|
# template for this node.
|
|
from nailgun.objects import NodeGroup
|
|
node_group = NodeGroup.get_by_uid(instance.group_id).name
|
|
if node_group not in template_body:
|
|
node_group = 'default'
|
|
|
|
node_name = cls.get_slave_name(instance)
|
|
nic_mapping = template_body[node_group]['nic_mapping']
|
|
if node_name not in nic_mapping:
|
|
node_name = 'default'
|
|
|
|
nic_mapping = nic_mapping[node_name]
|
|
|
|
# Replace interface references and re-parse JSON
|
|
template_object = NetworkTemplate(jsonutils.dumps(template_body))
|
|
node_template = template_object.safe_substitute(nic_mapping)
|
|
parsed_template = jsonutils.loads(node_template)
|
|
|
|
output = parsed_template[node_group]
|
|
output['templates'] = output.pop('network_scheme')
|
|
output['roles'] = {}
|
|
output['endpoints'] = {}
|
|
for v in output['templates'].values():
|
|
for endpoint in v['endpoints']:
|
|
output['endpoints'][endpoint] = {}
|
|
for role, ep in v['roles'].items():
|
|
output['roles'][role] = ep
|
|
|
|
instance.network_template = output
|
|
db().flush()
|
|
|
|
if instance.cluster:
|
|
nm = Cluster.get_network_manager(instance.cluster)
|
|
nm.assign_networks_by_template(instance)
|
|
|
|
@classmethod
|
|
def get_unique_hostname(cls, node, cluster_id):
|
|
"""Generate default hostname.
|
|
|
|
Hostname is 'node-{id}' if it's not used or 'node-{uuid} otherwise.
|
|
It's needed for case when user have manually renamed any another node
|
|
to 'node-{id}'.
|
|
"""
|
|
hostname = cls.get_slave_name(node)
|
|
if cls.get_by_hostname(hostname, cluster_id):
|
|
hostname = 'node-{0}'.format(node.uuid)
|
|
return hostname
|
|
|
|
@classmethod
|
|
def get_restrictions_models(cls, instance):
|
|
"""Return models which are used in restrictions mechanism
|
|
|
|
:param instance: nailgun.db.sqlalchemy.models.Node instance
|
|
:return: dict with models
|
|
"""
|
|
models = {'node_attributes': cls.get_attributes(instance)}
|
|
models.update(Cluster.get_restrictions_models(instance.cluster))
|
|
return models
|
|
|
|
@classmethod
|
|
def reset_vms_created_state(cls, node):
|
|
if consts.VIRTUAL_NODE_TYPES.virt not in node.all_roles:
|
|
return
|
|
|
|
for vm in node.vms_conf:
|
|
vm['created'] = False
|
|
node.vms_conf.changed()
|
|
|
|
@classmethod
|
|
def get_attributes(cls, instance):
|
|
attributes = copy.deepcopy(instance.attributes)
|
|
attributes.update(PluginManager.get_plugin_node_attributes(instance))
|
|
return attributes
|
|
|
|
@classmethod
|
|
def update_attributes(cls, instance, attrs):
|
|
PluginManager.update_plugin_node_attributes(attrs)
|
|
instance.attributes = utils.dict_merge(instance.attributes, attrs)
|
|
|
|
@classmethod
|
|
def get_default_attributes(cls, instance):
|
|
"""Get default attributes for Node.
|
|
|
|
:param instance: Node instance
|
|
:type instance: models.Node
|
|
:returns: dict -- Dict object of Node attributes
|
|
"""
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to update attributes of node "
|
|
u"'{0}' which isn't added to any cluster".format(
|
|
instance.full_name))
|
|
return {}
|
|
|
|
cluster = instance.cluster
|
|
attributes = copy.deepcopy(instance.cluster.release.node_attributes)
|
|
attributes.update(NodeAttributes.get_default_hugepages(instance))
|
|
attributes.update(
|
|
PluginManager.get_plugins_node_default_attributes(cluster))
|
|
|
|
return attributes
|
|
|
|
@classmethod
|
|
def set_default_attributes(cls, instance):
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to get default attributes of node "
|
|
u"'{0}' which isn't added to any cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
instance.attributes = copy.deepcopy(
|
|
instance.cluster.release.node_attributes)
|
|
NodeAttributes.set_default_hugepages(instance)
|
|
PluginManager.add_plugin_attributes_for_node(instance)
|
|
|
|
@classmethod
|
|
def dpdk_enabled(cls, instance):
|
|
network_manager = Cluster.get_network_manager(instance.cluster)
|
|
if not hasattr(network_manager, 'dpdk_enabled_for_node'):
|
|
return False
|
|
return network_manager.dpdk_enabled_for_node(instance)
|
|
|
|
@classmethod
|
|
def sriov_enabled(cls, instance):
|
|
for iface in instance.nic_interfaces:
|
|
if NIC.is_sriov_enabled(iface):
|
|
return True
|
|
return False
|
|
|
|
@classmethod
|
|
def refresh_dpdk_properties(cls, instance):
|
|
if not instance.cluster:
|
|
dpdk_drivers = {}
|
|
else:
|
|
dpdk_drivers = Release.get_supported_dpdk_drivers(
|
|
instance.cluster.release)
|
|
for interface in instance.nic_interfaces:
|
|
NIC.refresh_interface_dpdk_properties(interface, dpdk_drivers)
|
|
for interface in instance.bond_interfaces:
|
|
Bond.refresh_interface_dpdk_properties(interface, dpdk_drivers)
|
|
|
|
@classmethod
|
|
def is_provisioned(cls, instance):
|
|
"""Checks that node has been provisioned already.
|
|
|
|
:param instance: the Node object
|
|
:returns: True if provisioned otherwise False
|
|
"""
|
|
already_provisioned_statuses = (
|
|
consts.NODE_STATUSES.deploying,
|
|
consts.NODE_STATUSES.ready,
|
|
consts.NODE_STATUSES.provisioned,
|
|
consts.NODE_STATUSES.stopped
|
|
)
|
|
return (
|
|
instance.status in already_provisioned_statuses or
|
|
(instance.status == consts.NODE_STATUSES.error and
|
|
instance.error_type == consts.NODE_ERRORS.deploy)
|
|
)
|
|
|
|
@classmethod
|
|
def dpdk_nics(cls, instance):
|
|
if not instance.cluster:
|
|
return []
|
|
nm = Cluster.get_network_manager(instance.cluster)
|
|
return nm.dpdk_nics(instance)
|
|
|
|
@classmethod
|
|
def get_bond_default_attributes(cls, instance):
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to get bond default attributes of node "
|
|
u"'{0}' which isn't added to any cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
return Bond.get_default_attributes(instance.cluster)
|
|
|
|
@classmethod
|
|
def create_nic_attributes(cls, instance):
|
|
if not instance.cluster_id:
|
|
logger.warning(
|
|
u"Attempting to create NIC attributes of node "
|
|
u"'{0}' which isn't added to any cluster".format(
|
|
instance.full_name))
|
|
return
|
|
|
|
for nic_interface in instance.nic_interfaces:
|
|
NIC.create_attributes(nic_interface)
|
|
|
|
|
|
class NodeCollection(NailgunCollection):
|
|
"""Node collection"""
|
|
|
|
#: Single Node object class
|
|
single = Node
|
|
|
|
@classmethod
|
|
def eager_nodes_handlers(cls, iterable):
|
|
"""Eager load objects instances that is used in nodes handler.
|
|
|
|
:param iterable: iterable (SQLAlchemy query)
|
|
:returns: iterable (SQLAlchemy query)
|
|
"""
|
|
options = (
|
|
sa.orm.joinedload('cluster'),
|
|
sa.orm.subqueryload_all('nic_interfaces.assigned_networks_list'),
|
|
sa.orm.subqueryload_all('bond_interfaces.assigned_networks_list'),
|
|
sa.orm.subqueryload_all('ip_addrs.network_data')
|
|
)
|
|
return cls.eager_base(iterable, options)
|
|
|
|
@classmethod
|
|
def lock_nodes(cls, instances):
|
|
"""Locking nodes instances, fetched before, but required to be locked
|
|
|
|
:param instances: list of nodes
|
|
:return: list of locked nodes
|
|
"""
|
|
instances_ids = [instance.id for instance in instances]
|
|
q = cls.filter_by_list(None, 'id', instances_ids, order_by='id')
|
|
return cls.lock_for_update(q).all()
|
|
|
|
@classmethod
|
|
def get_by_group_id(cls, group_id):
|
|
return cls.filter_by(None, group_id=group_id)
|
|
|
|
@classmethod
|
|
def get_by_ids(cls, ids, cluster_id=None):
|
|
query = db.query(models.Node).filter(models.Node.id.in_(ids))
|
|
|
|
if cluster_id is not None:
|
|
query = query.filter(models.Node.cluster_id == cluster_id)
|
|
|
|
return query.all()
|
|
|
|
@classmethod
|
|
def reset_network_template(cls, instances):
|
|
for instance in instances:
|
|
instance.network_template = None
|
|
|
|
@classmethod
|
|
def reset_attributes(cls, instances):
|
|
for instance in instances:
|
|
instance.attributes = {}
|
|
|
|
@classmethod
|
|
def delete_by_ids(cls, ids):
|
|
fire_callback_on_node_collection_delete(ids)
|
|
db.query(cls.single.model).filter(
|
|
cls.single.model.id.in_(ids)).delete(synchronize_session='fetch')
|
|
|
|
@classmethod
|
|
def discovery_node_ids(self):
|
|
"""Ids of nodes which belong to the cluster and have 'discovery' status
|
|
|
|
:returns: list of node ids
|
|
"""
|
|
q_discovery = db().query(
|
|
models.Node.id).filter_by(status=consts.NODE_STATUSES.discover)
|
|
|
|
return [_id for (_id,) in q_discovery]
|
|
|
|
|
|
class NodeAttributes(object):
|
|
|
|
@classmethod
|
|
def _safe_get_hugepages(cls, node, attributes=None):
|
|
if not attributes:
|
|
attributes = Node.get_attributes(node)
|
|
hugepages = attributes.get('hugepages', {})
|
|
|
|
return hugepages
|
|
|
|
@classmethod
|
|
def _safe_get_cpu_pinning(cls, node, attributes=None):
|
|
if not attributes:
|
|
attributes = Node.get_attributes(node)
|
|
cpu_pinning = attributes.get('cpu_pinning', {})
|
|
|
|
return cpu_pinning
|
|
|
|
@classmethod
|
|
def set_default_hugepages(cls, node):
|
|
hugepages = cls.get_default_hugepages(node)
|
|
if hugepages:
|
|
Node.update_attributes(node, hugepages)
|
|
|
|
@classmethod
|
|
def get_default_hugepages(cls, node):
|
|
hugepages = cls._safe_get_hugepages(node)
|
|
if not hugepages:
|
|
return {}
|
|
|
|
sizes = [x[0] for x in consts.HUGE_PAGES_SIZE_MAP]
|
|
|
|
for attrs in six.itervalues(hugepages):
|
|
if attrs.get('type') == 'custom_hugepages':
|
|
attrs['value'] = dict.fromkeys(sizes, 0)
|
|
return {'hugepages': hugepages}
|
|
|
|
@classmethod
|
|
def distribute_node_cpus(cls, node, attributes=None):
|
|
numa_nodes = node.meta['numa_topology']['numa_nodes']
|
|
components = cls.node_cpu_pinning_info(node, attributes)['components']
|
|
dpdk_nics = Node.dpdk_nics(node)
|
|
|
|
nics_numas = []
|
|
for nic in dpdk_nics:
|
|
# NIC may have numa_node equal to null, in that case
|
|
# we assume that it belongs to first NUMA
|
|
nics_numas.append(nic.meta.get('numa_node') or 0)
|
|
|
|
return cpu_distribution.distribute_node_cpus(
|
|
numa_nodes, components, nics_numas)
|
|
|
|
@classmethod
|
|
def node_cpu_pinning_info(cls, node, attributes=None):
|
|
cpu_pinning_attrs = cls._safe_get_cpu_pinning(
|
|
node, attributes=attributes)
|
|
|
|
total_required_cpus = 0
|
|
components = {}
|
|
|
|
for name, attrs in six.iteritems(cpu_pinning_attrs):
|
|
# skip meta
|
|
if 'value' in attrs:
|
|
required_cpus = attrs['value']
|
|
total_required_cpus += required_cpus
|
|
components[name] = {'name': name,
|
|
'required_cpus': required_cpus}
|
|
return {'total_required_cpus': total_required_cpus,
|
|
'components': components}
|
|
|
|
@staticmethod
|
|
def pages_per_numa_node(size):
|
|
"""Convert memory size to 2 MiB pages count.
|
|
|
|
:param size: memory size in MiBs
|
|
:returns: amount of 2MiB pages
|
|
"""
|
|
# for python3 compatibility we have to use int() and float()
|
|
return int(math.ceil(float(size) / 2))
|
|
|
|
@classmethod
|
|
def total_hugepages(cls, node, attributes=None):
|
|
"""Return total hugepages for the node
|
|
|
|
Iterate over hugepages attributes and sum them
|
|
according their type: custom_hugepages - contains
|
|
items (size: count), number - this is the number of
|
|
memory in MB which must be allocated as hugepages
|
|
on each NUMA node (default hugepages size 2M will
|
|
be used and count will be calculated according to
|
|
number of specified memory).
|
|
|
|
:return: Dictionary with (size: count) items
|
|
"""
|
|
|
|
hugepages = collections.defaultdict(int)
|
|
numa_count = len(node.meta['numa_topology']['numa_nodes'])
|
|
|
|
hugepages_attributes = cls._safe_get_hugepages(
|
|
node, attributes=attributes)
|
|
|
|
for name, attrs in six.iteritems(hugepages_attributes):
|
|
if attrs.get('type') == 'custom_hugepages':
|
|
value = attrs['value']
|
|
for size, count in six.iteritems(value):
|
|
hugepages[size] += int(count)
|
|
elif attrs.get('type') == 'number':
|
|
# type number means that value is the number of memory in MB
|
|
# per NUMA node which should be converted to pages count
|
|
count_per_numa_node = cls.pages_per_numa_node(attrs['value'])
|
|
hugepages[consts.DEFAULT_HUGEPAGE_SIZE] += (
|
|
count_per_numa_node * numa_count)
|
|
|
|
for size in list(hugepages):
|
|
if not hugepages[size]:
|
|
hugepages.pop(size)
|
|
|
|
return hugepages
|
|
|
|
@classmethod
|
|
def hugepages_kernel_opts(cls, node):
|
|
hugepages = cls.total_hugepages(node)
|
|
|
|
kernel_opts = ""
|
|
for size, human_size in consts.HUGE_PAGES_SIZE_MAP:
|
|
hugepage_size = hugepages.get(size, 0)
|
|
if hugepage_size:
|
|
# extend kernel params with lines for huge pages
|
|
# hugepagesz is the size (2M, 1G, etc.)
|
|
# hugepages is the number of pages for specific size
|
|
kernel_opts += " hugepagesz={0} hugepages={1}".format(
|
|
human_size, hugepage_size)
|
|
|
|
return kernel_opts
|
|
|
|
@classmethod
|
|
def is_hugepages_enabled(cls, node, attributes=None):
|
|
return (
|
|
cls.is_nova_hugepages_enabled(node, attributes=attributes)
|
|
or cls.is_dpdk_hugepages_enabled(node, attributes=attributes)
|
|
)
|
|
|
|
@classmethod
|
|
def is_cpu_pinning_enabled(cls, node, attributes=None):
|
|
return (
|
|
cls.is_nova_cpu_pinning_enabled(node, attributes=attributes)
|
|
or cls.is_dpdk_cpu_pinning_enabled(node, attributes=attributes)
|
|
)
|
|
|
|
@classmethod
|
|
def is_dpdk_cpu_pinning_enabled(cls, node, attributes=None):
|
|
cpu_pinning = cls._safe_get_cpu_pinning(node, attributes=attributes)
|
|
return 'dpdk' in cpu_pinning and bool(cpu_pinning['dpdk']['value'])
|
|
|
|
@classmethod
|
|
def is_nova_cpu_pinning_enabled(cls, node, attributes=None):
|
|
cpu_pinning = cls._safe_get_cpu_pinning(node, attributes=attributes)
|
|
return 'nova' in cpu_pinning and bool(cpu_pinning['nova']['value'])
|
|
|
|
@classmethod
|
|
def is_nova_hugepages_enabled(cls, node, attributes=None):
|
|
hugepages = cls._safe_get_hugepages(node, attributes=attributes)
|
|
return ('nova' in hugepages and
|
|
any(six.itervalues(hugepages['nova']['value'])))
|
|
|
|
@classmethod
|
|
def is_dpdk_hugepages_enabled(cls, node, attributes=None):
|
|
hugepages = cls._safe_get_hugepages(node, attributes=attributes)
|
|
return 'dpdk' in hugepages and bool(hugepages['dpdk']['value'])
|
|
|
|
@classmethod
|
|
def dpdk_hugepages_attrs(cls, node):
|
|
"""Return hugepages configuration for DPDK
|
|
|
|
DPDK hugepages configured as the number of memory in MB
|
|
per NUMA node. This configuration passed to deployment
|
|
as comma-separeted values. (e.g. 'ovs_socket_mem': N,N,N,N
|
|
where N is the specified number of memory and there are
|
|
4 NUMA nodes)
|
|
"""
|
|
hugepages = cls._safe_get_hugepages(node)
|
|
|
|
if 'dpdk' not in hugepages or not hugepages['dpdk']['value']:
|
|
return {}
|
|
|
|
dpdk_memory = hugepages['dpdk']['value']
|
|
numa_nodes_len = len(node.meta['numa_topology']['numa_nodes'])
|
|
|
|
return {
|
|
'ovs_socket_mem':
|
|
list(itertools.repeat(dpdk_memory, numa_nodes_len))}
|
|
|
|
@classmethod
|
|
def distribute_hugepages(cls, node, attributes=None):
|
|
hugepages = cls._safe_get_hugepages(
|
|
node, attributes=attributes)
|
|
topology = node.meta['numa_topology']
|
|
|
|
# split components to 2 groups:
|
|
# components that should have pages on all numa nodes (such as dpdk)
|
|
# and components that may have pages on any numa node
|
|
components = {'all': [], 'any': []}
|
|
|
|
for attrs in hugepages.values():
|
|
if attrs.get('type') == 'number':
|
|
# type number means size of memory in MiB to allocate with
|
|
# 2MiB pages, so we need to calculate pages count
|
|
pages_count = cls.pages_per_numa_node(attrs['value'])
|
|
components['all'].append(
|
|
{consts.DEFAULT_HUGEPAGE_SIZE: pages_count})
|
|
|
|
elif attrs.get('type') == 'custom_hugepages':
|
|
components['any'].append(attrs['value'])
|
|
|
|
def numa_sort_func():
|
|
# Huge Pages distributor uses greedy algorithm and
|
|
# it knows nothing about CPU distribution. Thus
|
|
# this function will sort NUMA nodes according to
|
|
# distribution. Hence Huge Pages distribution will
|
|
# allocate more Huge Pages on NUMAs with more CPUs
|
|
cpu_distribution = cls.distribute_node_cpus(node, attributes)
|
|
|
|
nova_cpus = set(cpu_distribution['components'].get('nova', []))
|
|
numa_values = collections.defaultdict(int)
|
|
for numa_node in topology['numa_nodes']:
|
|
for cpu in numa_node['cpus']:
|
|
if cpu in nova_cpus:
|
|
numa_values[numa_node['id']] += 1
|
|
|
|
return lambda numa_id: -numa_values[numa_id]
|
|
|
|
return hugepages_distribution.distribute_hugepages(
|
|
topology, components, numa_sort_func())
|