Introduces the ZoneManager to the Scheduler which polls the child zones and caches their availability and capabilities.
This commit is contained in:
Sandy Walsh
@@ -356,3 +356,7 @@ DEFINE_string('host', socket.gethostname(),
|
||||
|
||||
DEFINE_string('node_availability_zone', 'nova',
|
||||
'availability zone of this node')
|
||||
|
||||
DEFINE_string('zone_name', 'nova', 'name of this zone')
|
||||
DEFINE_string('zone_capabilities', 'kypervisor:xenserver;os:linux',
|
||||
'Key/Value tags which represent capabilities of this zone')
|
||||
|
||||
49
nova/scheduler/api.py
Normal file
49
nova/scheduler/api.py
Normal file
@@ -0,0 +1,49 @@
|
||||
# Copyright (c) 2011 Openstack, LLC.
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Handles all requests relating to schedulers.
|
||||
"""
|
||||
|
||||
from nova import flags
|
||||
from nova import log as logging
|
||||
from nova import rpc
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
LOG = logging.getLogger('nova.scheduler.api')
|
||||
|
||||
|
||||
class API(object):
|
||||
"""API for interacting with the scheduler."""
|
||||
|
||||
def _call_scheduler(self, method, context, params=None):
|
||||
"""Generic handler for RPC calls to the scheduler.
|
||||
|
||||
:param params: Optional dictionary of arguments to be passed to the
|
||||
scheduler worker
|
||||
|
||||
:retval: Result returned by scheduler worker
|
||||
"""
|
||||
if not params:
|
||||
params = {}
|
||||
queue = FLAGS.scheduler_topic
|
||||
kwargs = {'method': method, 'args': params}
|
||||
return rpc.call(context, queue, kwargs)
|
||||
|
||||
def get_zone_list(self, context):
|
||||
items = self._call_scheduler('get_zone_list', context)
|
||||
for item in items:
|
||||
item['api_url'] = item['api_url'].replace('\\/', '/')
|
||||
return items
|
||||
143
nova/scheduler/zone_manager.py
Normal file
143
nova/scheduler/zone_manager.py
Normal file
@@ -0,0 +1,143 @@
|
||||
# Copyright (c) 2011 Openstack, LLC.
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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.
|
||||
|
||||
"""
|
||||
ZoneManager oversees all communications with child Zones.
|
||||
"""
|
||||
|
||||
import novaclient
|
||||
import thread
|
||||
import traceback
|
||||
|
||||
from datetime import datetime
|
||||
from eventlet import greenpool
|
||||
|
||||
from nova import db
|
||||
from nova import flags
|
||||
from nova import log as logging
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
flags.DEFINE_integer('zone_db_check_interval', 60,
|
||||
'Seconds between getting fresh zone info from db.')
|
||||
flags.DEFINE_integer('zone_failures_to_offline', 3,
|
||||
'Number of consecutive errors before marking zone offline')
|
||||
|
||||
|
||||
class ZoneState(object):
|
||||
"""Holds the state of all connected child zones."""
|
||||
def __init__(self):
|
||||
self.is_active = True
|
||||
self.name = None
|
||||
self.capabilities = None
|
||||
self.attempt = 0
|
||||
self.last_seen = datetime.min
|
||||
self.last_exception = None
|
||||
self.last_exception_time = None
|
||||
|
||||
def update_credentials(self, zone):
|
||||
"""Update zone credentials from db"""
|
||||
self.zone_id = zone.id
|
||||
self.api_url = zone.api_url
|
||||
self.username = zone.username
|
||||
self.password = zone.password
|
||||
|
||||
def update_metadata(self, zone_metadata):
|
||||
"""Update zone metadata after successful communications with
|
||||
child zone."""
|
||||
self.last_seen = datetime.now()
|
||||
self.attempt = 0
|
||||
self.name = zone_metadata["name"]
|
||||
self.capabilities = zone_metadata["capabilities"]
|
||||
self.is_active = True
|
||||
|
||||
def to_dict(self):
|
||||
return dict(name=self.name, capabilities=self.capabilities,
|
||||
is_active=self.is_active, api_url=self.api_url,
|
||||
id=self.zone_id)
|
||||
|
||||
def log_error(self, exception):
|
||||
"""Something went wrong. Check to see if zone should be
|
||||
marked as offline."""
|
||||
self.last_exception = exception
|
||||
self.last_exception_time = datetime.now()
|
||||
api_url = self.api_url
|
||||
logging.warning(_("'%(exception)s' error talking to "
|
||||
"zone %(api_url)s") % locals())
|
||||
|
||||
max_errors = FLAGS.zone_failures_to_offline
|
||||
self.attempt += 1
|
||||
if self.attempt >= max_errors:
|
||||
self.is_active = False
|
||||
logging.error(_("No answer from zone %(api_url)s "
|
||||
"after %(max_errors)d "
|
||||
"attempts. Marking inactive.") % locals())
|
||||
|
||||
|
||||
def _call_novaclient(zone):
|
||||
"""Call novaclient. Broken out for testing purposes."""
|
||||
client = novaclient.OpenStack(zone.username, zone.password, zone.api_url)
|
||||
return client.zones.info()._info
|
||||
|
||||
|
||||
def _poll_zone(zone):
|
||||
"""Eventlet worker to poll a zone."""
|
||||
logging.debug(_("Polling zone: %s") % zone.api_url)
|
||||
try:
|
||||
zone.update_metadata(_call_novaclient(zone))
|
||||
except Exception, e:
|
||||
zone.log_error(traceback.format_exc())
|
||||
|
||||
|
||||
class ZoneManager(object):
|
||||
"""Keeps the zone states updated."""
|
||||
def __init__(self):
|
||||
self.last_zone_db_check = datetime.min
|
||||
self.zone_states = {}
|
||||
self.green_pool = greenpool.GreenPool()
|
||||
|
||||
def get_zone_list(self):
|
||||
"""Return the list of zones we know about."""
|
||||
return [zone.to_dict() for zone in self.zone_states.values()]
|
||||
|
||||
def _refresh_from_db(self, context):
|
||||
"""Make our zone state map match the db."""
|
||||
# Add/update existing zones ...
|
||||
zones = db.zone_get_all(context)
|
||||
existing = self.zone_states.keys()
|
||||
db_keys = []
|
||||
for zone in zones:
|
||||
db_keys.append(zone.id)
|
||||
if zone.id not in existing:
|
||||
self.zone_states[zone.id] = ZoneState()
|
||||
self.zone_states[zone.id].update_credentials(zone)
|
||||
|
||||
# Cleanup zones removed from db ...
|
||||
keys = self.zone_states.keys() # since we're deleting
|
||||
for zone_id in keys:
|
||||
if zone_id not in db_keys:
|
||||
del self.zone_states[zone_id]
|
||||
|
||||
def _poll_zones(self, context):
|
||||
"""Try to connect to each child zone and get update."""
|
||||
self.green_pool.imap(_poll_zone, self.zone_states.values())
|
||||
|
||||
def ping(self, context=None):
|
||||
"""Ping should be called periodically to update zone status."""
|
||||
diff = datetime.now() - self.last_zone_db_check
|
||||
if diff.seconds >= FLAGS.zone_db_check_interval:
|
||||
logging.debug(_("Updating zone cache from db."))
|
||||
self.last_zone_db_check = datetime.now()
|
||||
self._refresh_from_db(context)
|
||||
self._poll_zones(context)
|
||||
172
nova/tests/test_zones.py
Normal file
172
nova/tests/test_zones.py
Normal file
@@ -0,0 +1,172 @@
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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.
|
||||
"""
|
||||
Tests For ZoneManager
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import mox
|
||||
import novaclient
|
||||
|
||||
from nova import context
|
||||
from nova import db
|
||||
from nova import flags
|
||||
from nova import service
|
||||
from nova import test
|
||||
from nova import rpc
|
||||
from nova import utils
|
||||
from nova.auth import manager as auth_manager
|
||||
from nova.scheduler import zone_manager
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
|
||||
|
||||
class FakeZone:
|
||||
"""Represents a fake zone from the db"""
|
||||
def __init__(self, *args, **kwargs):
|
||||
for k, v in kwargs.iteritems():
|
||||
setattr(self, k, v)
|
||||
|
||||
|
||||
def exploding_novaclient(zone):
|
||||
"""Used when we want to simulate a novaclient call failing."""
|
||||
raise Exception("kaboom")
|
||||
|
||||
|
||||
class ZoneManagerTestCase(test.TestCase):
|
||||
"""Test case for zone manager"""
|
||||
def test_ping(self):
|
||||
zm = zone_manager.ZoneManager()
|
||||
self.mox.StubOutWithMock(zm, '_refresh_from_db')
|
||||
self.mox.StubOutWithMock(zm, '_poll_zones')
|
||||
zm._refresh_from_db(mox.IgnoreArg())
|
||||
zm._poll_zones(mox.IgnoreArg())
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zm.ping(None)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_refresh_from_db_new(self):
|
||||
zm = zone_manager.ZoneManager()
|
||||
|
||||
self.mox.StubOutWithMock(db, 'zone_get_all')
|
||||
db.zone_get_all(mox.IgnoreArg()).AndReturn([
|
||||
FakeZone(id=1, api_url='http://foo.com', username='user1',
|
||||
password='pass1'),
|
||||
])
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 0)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zm._refresh_from_db(None)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
self.assertEquals(zm.zone_states[1].username, 'user1')
|
||||
|
||||
def test_refresh_from_db_replace_existing(self):
|
||||
zm = zone_manager.ZoneManager()
|
||||
zone_state = zone_manager.ZoneState()
|
||||
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
|
||||
username='user1', password='pass1'))
|
||||
zm.zone_states[1] = zone_state
|
||||
|
||||
self.mox.StubOutWithMock(db, 'zone_get_all')
|
||||
db.zone_get_all(mox.IgnoreArg()).AndReturn([
|
||||
FakeZone(id=1, api_url='http://foo.com', username='user2',
|
||||
password='pass2'),
|
||||
])
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zm._refresh_from_db(None)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
self.assertEquals(zm.zone_states[1].username, 'user2')
|
||||
|
||||
def test_refresh_from_db_missing(self):
|
||||
zm = zone_manager.ZoneManager()
|
||||
zone_state = zone_manager.ZoneState()
|
||||
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
|
||||
username='user1', password='pass1'))
|
||||
zm.zone_states[1] = zone_state
|
||||
|
||||
self.mox.StubOutWithMock(db, 'zone_get_all')
|
||||
db.zone_get_all(mox.IgnoreArg()).AndReturn([])
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zm._refresh_from_db(None)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 0)
|
||||
|
||||
def test_refresh_from_db_add_and_delete(self):
|
||||
zm = zone_manager.ZoneManager()
|
||||
zone_state = zone_manager.ZoneState()
|
||||
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
|
||||
username='user1', password='pass1'))
|
||||
zm.zone_states[1] = zone_state
|
||||
|
||||
self.mox.StubOutWithMock(db, 'zone_get_all')
|
||||
|
||||
db.zone_get_all(mox.IgnoreArg()).AndReturn([
|
||||
FakeZone(id=2, api_url='http://foo.com', username='user2',
|
||||
password='pass2'),
|
||||
])
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zm._refresh_from_db(None)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
self.assertEquals(len(zm.zone_states), 1)
|
||||
self.assertEquals(zm.zone_states[2].username, 'user2')
|
||||
|
||||
def test_poll_zone(self):
|
||||
self.mox.StubOutWithMock(zone_manager, '_call_novaclient')
|
||||
zone_manager._call_novaclient(mox.IgnoreArg()).AndReturn(
|
||||
dict(name='zohan', capabilities='hairdresser'))
|
||||
|
||||
zone_state = zone_manager.ZoneState()
|
||||
zone_state.update_credentials(FakeZone(id=2,
|
||||
api_url='http://foo.com', username='user2',
|
||||
password='pass2'))
|
||||
zone_state.attempt = 1
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zone_manager._poll_zone(zone_state)
|
||||
self.mox.VerifyAll()
|
||||
self.assertEquals(zone_state.attempt, 0)
|
||||
self.assertEquals(zone_state.name, 'zohan')
|
||||
|
||||
def test_poll_zone_fails(self):
|
||||
self.stubs.Set(zone_manager, "_call_novaclient", exploding_novaclient)
|
||||
|
||||
zone_state = zone_manager.ZoneState()
|
||||
zone_state.update_credentials(FakeZone(id=2,
|
||||
api_url='http://foo.com', username='user2',
|
||||
password='pass2'))
|
||||
zone_state.attempt = FLAGS.zone_failures_to_offline - 1
|
||||
|
||||
self.mox.ReplayAll()
|
||||
zone_manager._poll_zone(zone_state)
|
||||
self.mox.VerifyAll()
|
||||
self.assertEquals(zone_state.attempt, 3)
|
||||
self.assertFalse(zone_state.is_active)
|
||||
self.assertEquals(zone_state.name, None)
|
||||
Reference in New Issue
Block a user