1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
|
# Copyright 2014 eBay Software Foundation
# 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.
from oslo_log import log as logging
from neutronclient.common import exceptions as neutron_exceptions
from novaclient import exceptions as nova_exceptions
from trove.cluster.tasks import ClusterTask
from trove.cluster.tasks import ClusterTasks
from trove.common import cfg
from trove.common import clients
from trove.common import exception
from trove.common.i18n import _
from trove.common.notification import (
DBaaSClusterAttachConfiguration,
DBaaSClusterDetachConfiguration,
DBaaSClusterGrow,
DBaaSClusterShrink,
DBaaSClusterResetStatus,
DBaaSClusterRestart)
from trove.common.notification import DBaaSClusterUpgrade
from trove.common.notification import DBaaSInstanceAttachConfiguration
from trove.common.notification import DBaaSInstanceDetachConfiguration
from trove.common.notification import EndNotification
from trove.common.notification import StartNotification
from trove.common import server_group as srv_grp
from trove.common.strategies.cluster import strategy
from trove.common import utils
from trove.configuration import models as config_models
from trove.datastore import models as datastore_models
from trove.db import models as dbmodels
from trove.instance import models as inst_models
from trove.instance.tasks import InstanceTasks
from trove.taskmanager import api as task_api
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
def persisted_models():
return {
'clusters': DBCluster,
}
class DBCluster(dbmodels.DatabaseModelBase):
_data_fields = ['created', 'updated', 'name', 'task_id',
'tenant_id', 'datastore_version_id', 'deleted',
'deleted_at', 'configuration_id']
_table_name = 'clusters'
def __init__(self, task_status, **kwargs):
"""
Creates a new persistable entity of the cluster.
:param task_status: the current task of the cluster.
:type task_status: trove.cluster.tasks.ClusterTask
"""
kwargs["task_id"] = task_status.code
kwargs["deleted"] = False
super(DBCluster, self).__init__(**kwargs)
self.task_status = task_status
def _validate(self, errors):
if ClusterTask.from_code(self.task_id) is None:
errors['task_id'] = "Not valid."
if self.task_status is None:
errors['task_status'] = "Cannot be None."
@property
def task_status(self):
return ClusterTask.from_code(self.task_id)
@task_status.setter
def task_status(self, task_status):
self.task_id = task_status.code
class Cluster(object):
DEFAULT_LIMIT = CONF.clusters_page_size
def __init__(self, context, db_info, datastore=None,
datastore_version=None):
self.context = context
self.db_info = db_info
self.ds = datastore
self.ds_version = datastore_version
if self.ds_version is None:
self.ds_version = (datastore_models.DatastoreVersion.
load_by_uuid(self.db_info.datastore_version_id))
if self.ds is None:
self.ds = (datastore_models.Datastore.
load(self.ds_version.datastore_id))
self._db_instances = None
self._server_group = None
self._server_group_loaded = False
self._locality = None
@classmethod
def get_guest(cls, instance):
return clients.create_guest_client(instance.context,
instance.db_info.id,
instance.datastore_version.manager)
@classmethod
def load_all(cls, context, tenant_id):
db_infos = DBCluster.find_all(tenant_id=tenant_id,
deleted=False)
limit = utils.pagination_limit(context.limit, Cluster.DEFAULT_LIMIT)
data_view = DBCluster.find_by_pagination('clusters', db_infos, "foo",
limit=limit,
marker=context.marker)
next_marker = data_view.next_page_marker
ret = [cls(context, db_info) for db_info in data_view.collection]
return ret, next_marker
@classmethod
def load(cls, context, cluster_id, clazz=None):
try:
db_info = DBCluster.find_by(context=context, id=cluster_id,
deleted=False)
except exception.ModelNotFoundError:
raise exception.ClusterNotFound(cluster=cluster_id)
if not clazz:
ds_version = (datastore_models.DatastoreVersion.
load_by_uuid(db_info.datastore_version_id))
manager = ds_version.manager
clazz = strategy.load_api_strategy(manager).cluster_class
return clazz(context, db_info)
def update_db(self, **values):
self.db_info = DBCluster.find_by(id=self.id, deleted=False)
for key in values:
setattr(self.db_info, key, values[key])
self.db_info.save()
def reset_task(self):
LOG.info("Setting task to NONE on cluster %s", self.id)
self.update_db(task_status=ClusterTasks.NONE)
def reset_status(self):
LOG.info("Resetting status to NONE on cluster %s", self.id)
self.reset_task()
instances = inst_models.DBInstance.find_all(cluster_id=self.id,
deleted=False).all()
for inst in instances:
instance = inst_models.load_any_instance(self.context, inst.id)
instance.reset_status()
@property
def id(self):
return self.db_info.id
@property
def created(self):
return self.db_info.created
@property
def updated(self):
return self.db_info.updated
@property
def name(self):
return self.db_info.name
@property
def task_id(self):
return self.db_info.task_status.code
@property
def task_name(self):
return self.db_info.task_status.name
@property
def task_description(self):
return self.db_info.task_status.description
@property
def tenant_id(self):
return self.db_info.tenant_id
@property
def datastore(self):
return self.ds
@property
def datastore_version(self):
return self.ds_version
@property
def deleted(self):
return self.db_info.deleted
@property
def deleted_at(self):
return self.db_info.deleted_at
@property
def configuration_id(self):
return self.db_info.configuration_id
@property
def db_instances(self):
"""DBInstance objects are persistent, therefore cacheable."""
if not self._db_instances:
self._db_instances = inst_models.DBInstance.find_all(
cluster_id=self.id, deleted=False).all()
return self._db_instances
@property
def instances(self):
return inst_models.Instances.load_all_by_cluster_id(self.context,
self.db_info.id)
@property
def instances_without_server(self):
return inst_models.Instances.load_all_by_cluster_id(
self.context, self.db_info.id, load_servers=False)
@property
def server_group(self):
# The server group could be empty, so we need a flag to cache it
if not self._server_group_loaded and self.instances:
self._server_group = None
# Not all the instances may have the server group loaded, so
# check them all
for instance in self.instances:
if instance.server_group:
self._server_group = instance.server_group
break
self._server_group_loaded = True
return self._server_group
@property
def locality(self):
if not self._locality:
if self.server_group:
self._locality = srv_grp.ServerGroup.get_locality(
self._server_group)
return self._locality
@locality.setter
def locality(self, value):
"""This is to facilitate the fact that the server group may not be
set up before the create command returns.
"""
self._locality = value
@classmethod
def create(cls, context, name, datastore, datastore_version,
instances, extended_properties, locality, configuration):
locality = srv_grp.ServerGroup.build_scheduler_hint(
context, locality, name)
api_strategy = strategy.load_api_strategy(datastore_version.manager)
return api_strategy.cluster_class.create(context, name, datastore,
datastore_version, instances,
extended_properties,
locality, configuration)
def validate_cluster_available(self, valid_states=[ClusterTasks.NONE]):
if self.db_info.task_status not in valid_states:
log_fmt = ("This action cannot be performed on the cluster while "
"the current cluster task is '%s'.")
exc_fmt = _("This action cannot be performed on the cluster while "
"the current cluster task is '%s'.")
LOG.error(log_fmt, self.db_info.task_status.name)
raise exception.UnprocessableEntity(
exc_fmt % self.db_info.task_status.name)
def delete(self):
self.validate_cluster_available([ClusterTasks.NONE,
ClusterTasks.DELETING])
db_insts = inst_models.DBInstance.find_all(cluster_id=self.id,
deleted=False).all()
self.update_db(task_status=ClusterTasks.DELETING)
# we force the server-group delete here since we need to load the
# group while the instances still exist. Also, since the instances
# take a while to be removed they might not all be gone even if we
# do it after the delete.
srv_grp.ServerGroup.delete(self.context, self.server_group, force=True)
for db_inst in db_insts:
instance = inst_models.load_any_instance(self.context, db_inst.id)
instance.delete()
task_api.API(self.context).delete_cluster(self.id)
def action(self, context, req, action, param):
if action == 'grow':
context.notification = DBaaSClusterGrow(context, request=req)
with StartNotification(context, cluster_id=self.id):
instances = []
for node in param:
instance = {
'flavor_id': utils.get_id_from_href(node['flavorRef'])
}
if 'name' in node:
instance['name'] = node['name']
if 'volume' in node:
instance['volume_size'] = int(node['volume']['size'])
if 'modules' in node:
instance['modules'] = node['modules']
if 'nics' in node:
instance['nics'] = node['nics']
if 'availability_zone' in node:
instance['availability_zone'] = (
node['availability_zone'])
if 'type' in node:
instance_type = node['type']
if isinstance(instance_type, str):
instance_type = instance_type.split(',')
instance['instance_type'] = instance_type
instances.append(instance)
return self.grow(instances)
elif action == 'shrink':
context.notification = DBaaSClusterShrink(context, request=req)
instance_ids = [instance['id'] for instance in param]
with StartNotification(context, cluster_id=self.id,
instance_ids=instance_ids):
instance_ids = [instance['id'] for instance in param]
return self.shrink(instance_ids)
elif action == "reset-status":
context.notification = DBaaSClusterResetStatus(context,
request=req)
with StartNotification(context, cluster_id=self.id):
return self.reset_status()
elif action == 'restart':
context.notification = DBaaSClusterRestart(context, request=req)
with StartNotification(context, cluster_id=self.id):
return self.restart()
elif action == 'upgrade':
context.notification = DBaaSClusterUpgrade(context, request=req)
dv_id = param['datastore_version']
dv = datastore_models.DatastoreVersion.load(self.datastore, dv_id)
with StartNotification(context, cluster_id=self.id,
datastore_version=dv.id):
self.upgrade(dv)
self.update_db(datastore_version_id=dv.id)
elif action == 'configuration_attach':
configuration_id = param['configuration_id']
context.notification = DBaaSClusterAttachConfiguration(context,
request=req)
with StartNotification(context, cluster_id=self.id,
configuration_id=configuration_id):
return self.configuration_attach(configuration_id)
elif action == 'configuration_detach':
context.notification = DBaaSClusterDetachConfiguration(context,
request=req)
with StartNotification(context, cluster_id=self.id):
return self.configuration_detach()
else:
raise exception.BadRequest(_("Action %s not supported") % action)
def grow(self, instances):
raise exception.BadRequest(_("Action 'grow' not supported"))
def shrink(self, instance_ids):
raise exception.BadRequest(_("Action 'shrink' not supported"))
def rolling_restart(self):
self.validate_cluster_available()
self.db_info.update(task_status=ClusterTasks.RESTARTING_CLUSTER)
try:
cluster_id = self.db_info.id
task_api.load(self.context, self.ds_version.manager
).restart_cluster(cluster_id)
except Exception:
self.db_info.update(task_status=ClusterTasks.NONE)
raise
return self.__class__(self.context, self.db_info,
self.ds, self.ds_version)
def rolling_upgrade(self, datastore_version):
"""Upgrades a cluster to a new datastore version."""
LOG.debug("Upgrading cluster %s.", self.id)
self.validate_cluster_available()
self.db_info.update(task_status=ClusterTasks.UPGRADING_CLUSTER)
try:
cluster_id = self.db_info.id
ds_ver_id = datastore_version.id
task_api.load(self.context, self.ds_version.manager
).upgrade_cluster(cluster_id, ds_ver_id)
except Exception:
self.db_info.update(task_status=ClusterTasks.NONE)
raise
return self.__class__(self.context, self.db_info,
self.ds, self.ds_version)
def restart(self):
raise exception.BadRequest(_("Action 'restart' not supported"))
def upgrade(self, datastore_version):
raise exception.BadRequest(_("Action 'upgrade' not supported"))
def configuration_attach(self, configuration_id):
raise exception.BadRequest(
_("Action 'configuration_attach' not supported"))
def rolling_configuration_update(self, configuration_id,
apply_on_all=True):
cluster_notification = self.context.notification
request_info = cluster_notification.serialize(self.context)
self.validate_cluster_available()
self.db_info.update(task_status=ClusterTasks.UPDATING_CLUSTER)
try:
configuration = config_models.Configuration.find(
self.context, configuration_id, self.datastore_version.id)
instances = [inst_models.Instance.load(self.context, instance.id)
for instance in self.instances]
LOG.debug("Persisting changes on cluster nodes.")
# Allow re-applying the same configuration (e.g. on configuration
# updates).
for instance in instances:
if not (instance.configuration and
instance.configuration.id != configuration_id):
self.context.notification = (
DBaaSInstanceAttachConfiguration(self.context,
**request_info))
with StartNotification(self.context,
instance_id=instance.id,
configuration_id=configuration_id):
with EndNotification(self.context):
instance.save_configuration(configuration)
else:
LOG.debug(
"Node '%(inst_id)s' already has the configuration "
"'%(conf_id)s' attached.",
{'inst_id': instance.id,
'conf_id': instance.configuration.id})
# Configuration has been persisted to all instances.
# The cluster is in a consistent state with all nodes
# requiring restart.
# We therefore assign the configuration group ID now.
# The configuration can be safely detached at this point.
self.update_db(configuration_id=configuration_id)
LOG.debug("Applying runtime configuration changes.")
if instances[0].apply_configuration(configuration):
LOG.debug(
"Runtime changes have been applied successfully to the "
"first node.")
remaining_nodes = instances[1:]
if apply_on_all:
LOG.debug(
"Applying the changes to the remaining nodes.")
for instance in remaining_nodes:
instance.apply_configuration(configuration)
else:
LOG.debug(
"Releasing restart-required task on the remaining "
"nodes.")
for instance in remaining_nodes:
instance.update_db(task_status=InstanceTasks.NONE)
finally:
self.update_db(task_status=ClusterTasks.NONE)
return self.__class__(self.context, self.db_info,
self.ds, self.ds_version)
def configuration_detach(self):
raise exception.BadRequest(
_("Action 'configuration_detach' not supported"))
def rolling_configuration_remove(self, apply_on_all=True):
cluster_notification = self.context.notification
request_info = cluster_notification.serialize(self.context)
self.validate_cluster_available()
self.db_info.update(task_status=ClusterTasks.UPDATING_CLUSTER)
try:
instances = [inst_models.Instance.load(self.context, instance.id)
for instance in self.instances]
LOG.debug("Removing changes from cluster nodes.")
for instance in instances:
if instance.configuration:
self.context.notification = (
DBaaSInstanceDetachConfiguration(self.context,
**request_info))
with StartNotification(self.context,
instance_id=instance.id):
with EndNotification(self.context):
instance.delete_configuration()
else:
LOG.debug(
"Node '%s' has no configuration attached.",
instance.id)
# The cluster is in a consistent state with all nodes
# requiring restart.
# New configuration can be safely attached at this point.
configuration_id = self.configuration_id
self.update_db(configuration_id=None)
LOG.debug("Applying runtime configuration changes.")
if instances[0].reset_configuration(configuration_id):
LOG.debug(
"Runtime changes have been applied successfully to the "
"first node.")
remaining_nodes = instances[1:]
if apply_on_all:
LOG.debug(
"Applying the changes to the remaining nodes.")
for instance in remaining_nodes:
instance.reset_configuration(configuration_id)
else:
LOG.debug(
"Releasing restart-required task on the remaining "
"nodes.")
for instance in remaining_nodes:
instance.update_db(task_status=InstanceTasks.NONE)
finally:
self.update_db(task_status=ClusterTasks.NONE)
return self.__class__(self.context, self.db_info,
self.ds, self.ds_version)
@staticmethod
def load_instance(context, cluster_id, instance_id):
return inst_models.load_instance_with_info(
inst_models.DetailInstance, context, instance_id, cluster_id)
@staticmethod
def manager_from_cluster_id(context, cluster_id):
db_info = DBCluster.find_by(context=context, id=cluster_id,
deleted=False)
ds_version = (datastore_models.DatastoreVersion.
load_by_uuid(db_info.datastore_version_id))
return ds_version.manager
def is_cluster_deleting(context, cluster_id):
cluster = Cluster.load(context, cluster_id)
return (cluster.db_info.task_status == ClusterTasks.DELETING or
cluster.db_info.task_status == ClusterTasks.SHRINKING_CLUSTER)
def validate_instance_flavors(context, instances,
volume_enabled, ephemeral_enabled):
"""Validate flavors for given instance definitions."""
nova_cli_cache = dict()
for instance in instances:
region_name = instance.get('region_name')
flavor_id = instance['flavor_id']
try:
if region_name in nova_cli_cache:
nova_client = nova_cli_cache[region_name]
else:
nova_client = clients.create_nova_client(
context, region_name)
nova_cli_cache[region_name] = nova_client
flavor = nova_client.flavors.get(flavor_id)
if (not volume_enabled and
(ephemeral_enabled and flavor.ephemeral == 0)):
raise exception.LocalStorageNotSpecified(
flavor=flavor_id)
except nova_exceptions.NotFound:
raise exception.FlavorNotFound(uuid=flavor_id)
def get_required_volume_size(instances, volume_enabled):
"""Calculate the total Trove volume size for given instances."""
volume_sizes = [instance['volume_size'] for instance in instances
if instance.get('volume_size', None)]
if volume_enabled:
if len(volume_sizes) != len(instances):
raise exception.ClusterVolumeSizeRequired()
total_volume_size = 0
for volume_size in volume_sizes:
validate_volume_size(volume_size)
total_volume_size += volume_size
return total_volume_size
if len(volume_sizes) > 0:
raise exception.VolumeNotSupported()
return None
def assert_homogeneous_cluster(instances, required_flavor=None,
required_volume_size=None):
"""Verify that all instances have the same flavor and volume size
(volume size = 0 if there should be no Trove volumes).
"""
assert_same_instance_flavors(instances, required_flavor=required_flavor)
assert_same_instance_volumes(instances, required_size=required_volume_size)
def assert_same_instance_flavors(instances, required_flavor=None):
"""Verify that all instances have the same flavor.
:param required_flavor The flavor all instances should have or
None if no specific flavor is required.
:type required_flavor flavor_id
"""
flavors = {instance['flavor_id'] for instance in instances}
if len(flavors) != 1 or (required_flavor is not None and
required_flavor not in flavors):
raise exception.ClusterFlavorsNotEqual()
def assert_same_instance_volumes(instances, required_size=None):
"""Verify that all instances have the same volume size (size = 0 if there
is not a Trove volume for the instance).
:param required_size Size in GB all instance's volumes should
have or 0 if there should be no attached
volumes.
None if no particular size is required.
:type required_size int
"""
sizes = {instance.get('volume_size', 0) for instance in instances}
if len(sizes) != 1 or (required_size is not None and
required_size not in sizes):
raise exception.ClusterVolumeSizesNotEqual()
def validate_volume_size(size):
"""Verify the volume size is within the maximum limit for Trove volumes."""
if size is None:
raise exception.VolumeSizeNotSpecified()
max_size = CONF.max_accepted_volume_size
if int(size) > max_size:
msg = ("Volume 'size' cannot exceed maximum "
"of %d Gb, %s cannot be accepted."
% (max_size, size))
raise exception.VolumeQuotaExceeded(msg)
def validate_instance_nics(context, instances):
"""Checking networks are same for the cluster."""
instance_nics = []
for instance in instances:
nics = instance.get('nics')
if nics:
instance_nics.append(nics[0].get('net-id'))
if len(set(instance_nics)) > 1:
raise exception.ClusterNetworksNotEqual()
if not instance_nics:
return
instance_nic = instance_nics[0]
try:
neutron_client = clients.create_neutron_client(context)
neutron_client.find_resource('network', instance_nic)
except neutron_exceptions.NotFound:
raise exception.NetworkNotFound(uuid=instance_nic)
|