models.py revision eab66ce582bfe05076ff096c3a044d8f0497bbca
1import logging, os 2from datetime import datetime 3from django.db import models as dbmodels, connection 4import common 5from autotest_lib.frontend.afe import model_logic 6from autotest_lib.frontend import settings, thread_local 7from autotest_lib.client.common_lib import enum, host_protections, global_config 8from autotest_lib.client.common_lib import host_queue_entry_states 9 10# job options and user preferences 11RebootBefore = enum.Enum('Never', 'If dirty', 'Always') 12DEFAULT_REBOOT_BEFORE = RebootBefore.IF_DIRTY 13RebootAfter = enum.Enum('Never', 'If all tests passed', 'Always') 14DEFAULT_REBOOT_AFTER = RebootBefore.ALWAYS 15 16 17class AclAccessViolation(Exception): 18 """\ 19 Raised when an operation is attempted with proper permissions as 20 dictated by ACLs. 21 """ 22 23 24class AtomicGroup(model_logic.ModelWithInvalid, dbmodels.Model): 25 """\ 26 An atomic group defines a collection of hosts which must only be scheduled 27 all at once. Any host with a label having an atomic group will only be 28 scheduled for a job at the same time as other hosts sharing that label. 29 30 Required: 31 name: A name for this atomic group. ex: 'rack23' or 'funky_net' 32 max_number_of_machines: The maximum number of machines that will be 33 scheduled at once when scheduling jobs to this atomic group. 34 The job.synch_count is considered the minimum. 35 36 Optional: 37 description: Arbitrary text description of this group's purpose. 38 """ 39 name = dbmodels.CharField(max_length=255, unique=True) 40 description = dbmodels.TextField(blank=True) 41 # This magic value is the default to simplify the scheduler logic. 42 # It must be "large". The common use of atomic groups is to want all 43 # machines in the group to be used, limits on which subset used are 44 # often chosen via dependency labels. 45 INFINITE_MACHINES = 333333333 46 max_number_of_machines = dbmodels.IntegerField(default=INFINITE_MACHINES) 47 invalid = dbmodels.BooleanField(default=False, 48 editable=settings.FULL_ADMIN) 49 50 name_field = 'name' 51 objects = model_logic.ExtendedManager() 52 valid_objects = model_logic.ValidObjectsManager() 53 54 55 def enqueue_job(self, job, is_template=False): 56 """Enqueue a job on an associated atomic group of hosts.""" 57 queue_entry = HostQueueEntry.create(atomic_group=self, job=job, 58 is_template=is_template) 59 queue_entry.save() 60 61 62 def clean_object(self): 63 self.label_set.clear() 64 65 66 class Meta: 67 db_table = 'afe_atomic_groups' 68 69 70 def __unicode__(self): 71 return unicode(self.name) 72 73 74class Label(model_logic.ModelWithInvalid, dbmodels.Model): 75 """\ 76 Required: 77 name: label name 78 79 Optional: 80 kernel_config: URL/path to kernel config for jobs run on this label. 81 platform: If True, this is a platform label (defaults to False). 82 only_if_needed: If True, a Host with this label can only be used if that 83 label is requested by the job/test (either as the meta_host or 84 in the job_dependencies). 85 atomic_group: The atomic group associated with this label. 86 """ 87 name = dbmodels.CharField(max_length=255, unique=True) 88 kernel_config = dbmodels.CharField(max_length=255, blank=True) 89 platform = dbmodels.BooleanField(default=False) 90 invalid = dbmodels.BooleanField(default=False, 91 editable=settings.FULL_ADMIN) 92 only_if_needed = dbmodels.BooleanField(default=False) 93 94 name_field = 'name' 95 objects = model_logic.ExtendedManager() 96 valid_objects = model_logic.ValidObjectsManager() 97 atomic_group = dbmodels.ForeignKey(AtomicGroup, null=True, blank=True) 98 99 100 def clean_object(self): 101 self.host_set.clear() 102 self.test_set.clear() 103 104 105 def enqueue_job(self, job, atomic_group=None, is_template=False): 106 """Enqueue a job on any host of this label.""" 107 queue_entry = HostQueueEntry.create(meta_host=self, job=job, 108 is_template=is_template, 109 atomic_group=atomic_group) 110 queue_entry.save() 111 112 113 class Meta: 114 db_table = 'afe_labels' 115 116 def __unicode__(self): 117 return unicode(self.name) 118 119 120class User(dbmodels.Model, model_logic.ModelExtensions): 121 """\ 122 Required: 123 login :user login name 124 125 Optional: 126 access_level: 0=User (default), 1=Admin, 100=Root 127 """ 128 ACCESS_ROOT = 100 129 ACCESS_ADMIN = 1 130 ACCESS_USER = 0 131 132 login = dbmodels.CharField(max_length=255, unique=True) 133 access_level = dbmodels.IntegerField(default=ACCESS_USER, blank=True) 134 135 # user preferences 136 reboot_before = dbmodels.SmallIntegerField(choices=RebootBefore.choices(), 137 blank=True, 138 default=DEFAULT_REBOOT_BEFORE) 139 reboot_after = dbmodels.SmallIntegerField(choices=RebootAfter.choices(), 140 blank=True, 141 default=DEFAULT_REBOOT_AFTER) 142 show_experimental = dbmodels.BooleanField(default=False) 143 144 name_field = 'login' 145 objects = model_logic.ExtendedManager() 146 147 148 def save(self, *args, **kwargs): 149 # is this a new object being saved for the first time? 150 first_time = (self.id is None) 151 user = thread_local.get_user() 152 if user and not user.is_superuser() and user.login != self.login: 153 raise AclAccessViolation("You cannot modify user " + self.login) 154 super(User, self).save(*args, **kwargs) 155 if first_time: 156 everyone = AclGroup.objects.get(name='Everyone') 157 everyone.users.add(self) 158 159 160 def is_superuser(self): 161 return self.access_level >= self.ACCESS_ROOT 162 163 164 class Meta: 165 db_table = 'afe_users' 166 167 def __unicode__(self): 168 return unicode(self.login) 169 170 171class Host(model_logic.ModelWithInvalid, dbmodels.Model, 172 model_logic.ModelWithAttributes): 173 """\ 174 Required: 175 hostname 176 177 optional: 178 locked: if true, host is locked and will not be queued 179 180 Internal: 181 synch_id: currently unused 182 status: string describing status of host 183 invalid: true if the host has been deleted 184 protection: indicates what can be done to this host during repair 185 locked_by: user that locked the host, or null if the host is unlocked 186 lock_time: DateTime at which the host was locked 187 dirty: true if the host has been used without being rebooted 188 """ 189 Status = enum.Enum('Verifying', 'Running', 'Ready', 'Repairing', 190 'Repair Failed', 'Dead', 'Cleaning', 'Pending', 191 string_values=True) 192 193 hostname = dbmodels.CharField(max_length=255, unique=True) 194 labels = dbmodels.ManyToManyField(Label, blank=True, 195 db_table='afe_hosts_labels') 196 locked = dbmodels.BooleanField(default=False) 197 synch_id = dbmodels.IntegerField(blank=True, null=True, 198 editable=settings.FULL_ADMIN) 199 status = dbmodels.CharField(max_length=255, default=Status.READY, 200 choices=Status.choices(), 201 editable=settings.FULL_ADMIN) 202 invalid = dbmodels.BooleanField(default=False, 203 editable=settings.FULL_ADMIN) 204 protection = dbmodels.SmallIntegerField(null=False, blank=True, 205 choices=host_protections.choices, 206 default=host_protections.default) 207 locked_by = dbmodels.ForeignKey(User, null=True, blank=True, editable=False) 208 lock_time = dbmodels.DateTimeField(null=True, blank=True, editable=False) 209 dirty = dbmodels.BooleanField(default=True, editable=settings.FULL_ADMIN) 210 211 name_field = 'hostname' 212 objects = model_logic.ExtendedManager() 213 valid_objects = model_logic.ValidObjectsManager() 214 215 216 def __init__(self, *args, **kwargs): 217 super(Host, self).__init__(*args, **kwargs) 218 self._record_attributes(['status']) 219 220 221 @staticmethod 222 def create_one_time_host(hostname): 223 query = Host.objects.filter(hostname=hostname) 224 if query.count() == 0: 225 host = Host(hostname=hostname, invalid=True) 226 host.do_validate() 227 else: 228 host = query[0] 229 if not host.invalid: 230 raise model_logic.ValidationError({ 231 'hostname' : '%s already exists in the autotest DB. ' 232 'Select it rather than entering it as a one time ' 233 'host.' % hostname 234 }) 235 host.protection = host_protections.Protection.DO_NOT_REPAIR 236 host.locked = False 237 host.save() 238 host.clean_object() 239 return host 240 241 242 def resurrect_object(self, old_object): 243 super(Host, self).resurrect_object(old_object) 244 # invalid hosts can be in use by the scheduler (as one-time hosts), so 245 # don't change the status 246 self.status = old_object.status 247 248 249 def clean_object(self): 250 self.aclgroup_set.clear() 251 self.labels.clear() 252 253 254 def save(self, *args, **kwargs): 255 # extra spaces in the hostname can be a sneaky source of errors 256 self.hostname = self.hostname.strip() 257 # is this a new object being saved for the first time? 258 first_time = (self.id is None) 259 if not first_time: 260 AclGroup.check_for_acl_violation_hosts([self]) 261 if self.locked and not self.locked_by: 262 self.locked_by = thread_local.get_user() 263 self.lock_time = datetime.now() 264 self.dirty = True 265 elif not self.locked and self.locked_by: 266 self.locked_by = None 267 self.lock_time = None 268 super(Host, self).save(*args, **kwargs) 269 if first_time: 270 everyone = AclGroup.objects.get(name='Everyone') 271 everyone.hosts.add(self) 272 self._check_for_updated_attributes() 273 274 275 def delete(self): 276 AclGroup.check_for_acl_violation_hosts([self]) 277 for queue_entry in self.hostqueueentry_set.all(): 278 queue_entry.deleted = True 279 queue_entry.abort(thread_local.get_user()) 280 super(Host, self).delete() 281 282 283 def on_attribute_changed(self, attribute, old_value): 284 assert attribute == 'status' 285 logging.info(self.hostname + ' -> ' + self.status) 286 287 288 def enqueue_job(self, job, atomic_group=None, is_template=False): 289 """Enqueue a job on this host.""" 290 queue_entry = HostQueueEntry.create(host=self, job=job, 291 is_template=is_template, 292 atomic_group=atomic_group) 293 # allow recovery of dead hosts from the frontend 294 if not self.active_queue_entry() and self.is_dead(): 295 self.status = Host.Status.READY 296 self.save() 297 queue_entry.save() 298 299 block = IneligibleHostQueue(job=job, host=self) 300 block.save() 301 302 303 def platform(self): 304 # TODO(showard): slighly hacky? 305 platforms = self.labels.filter(platform=True) 306 if len(platforms) == 0: 307 return None 308 return platforms[0] 309 platform.short_description = 'Platform' 310 311 312 @classmethod 313 def check_no_platform(cls, hosts): 314 Host.objects.populate_relationships(hosts, Label, 'label_list') 315 errors = [] 316 for host in hosts: 317 platforms = [label.name for label in host.label_list 318 if label.platform] 319 if platforms: 320 # do a join, just in case this host has multiple platforms, 321 # we'll be able to see it 322 errors.append('Host %s already has a platform: %s' % ( 323 host.hostname, ', '.join(platforms))) 324 if errors: 325 raise model_logic.ValidationError({'labels': '; '.join(errors)}) 326 327 328 def is_dead(self): 329 return self.status == Host.Status.REPAIR_FAILED 330 331 332 def active_queue_entry(self): 333 active = list(self.hostqueueentry_set.filter(active=True)) 334 if not active: 335 return None 336 assert len(active) == 1, ('More than one active entry for ' 337 'host ' + self.hostname) 338 return active[0] 339 340 341 def _get_attribute_model_and_args(self, attribute): 342 return HostAttribute, dict(host=self, attribute=attribute) 343 344 345 class Meta: 346 db_table = 'afe_hosts' 347 348 def __unicode__(self): 349 return unicode(self.hostname) 350 351 352class HostAttribute(dbmodels.Model): 353 """Arbitrary keyvals associated with hosts.""" 354 host = dbmodels.ForeignKey(Host) 355 attribute = dbmodels.CharField(max_length=90) 356 value = dbmodels.CharField(max_length=300) 357 358 objects = model_logic.ExtendedManager() 359 360 class Meta: 361 db_table = 'afe_host_attributes' 362 363 364class Test(dbmodels.Model, model_logic.ModelExtensions): 365 """\ 366 Required: 367 author: author name 368 description: description of the test 369 name: test name 370 time: short, medium, long 371 test_class: This describes the class for your the test belongs in. 372 test_category: This describes the category for your tests 373 test_type: Client or Server 374 path: path to pass to run_test() 375 sync_count: is a number >=1 (1 being the default). If it's 1, then it's an 376 async job. If it's >1 it's sync job for that number of machines 377 i.e. if sync_count = 2 it is a sync job that requires two 378 machines. 379 Optional: 380 dependencies: What the test requires to run. Comma deliminated list 381 dependency_labels: many-to-many relationship with labels corresponding to 382 test dependencies. 383 experimental: If this is set to True production servers will ignore the test 384 run_verify: Whether or not the scheduler should run the verify stage 385 """ 386 TestTime = enum.Enum('SHORT', 'MEDIUM', 'LONG', start_value=1) 387 # TODO(showard) - this should be merged with Job.ControlType (but right 388 # now they use opposite values) 389 Types = enum.Enum('Client', 'Server', start_value=1) 390 391 name = dbmodels.CharField(max_length=255, unique=True) 392 author = dbmodels.CharField(max_length=255) 393 test_class = dbmodels.CharField(max_length=255) 394 test_category = dbmodels.CharField(max_length=255) 395 dependencies = dbmodels.CharField(max_length=255, blank=True) 396 description = dbmodels.TextField(blank=True) 397 experimental = dbmodels.BooleanField(default=True) 398 run_verify = dbmodels.BooleanField(default=True) 399 test_time = dbmodels.SmallIntegerField(choices=TestTime.choices(), 400 default=TestTime.MEDIUM) 401 test_type = dbmodels.SmallIntegerField(choices=Types.choices()) 402 sync_count = dbmodels.IntegerField(default=1) 403 path = dbmodels.CharField(max_length=255, unique=True) 404 405 dependency_labels = ( 406 dbmodels.ManyToManyField(Label, blank=True, 407 db_table='afe_autotests_dependency_labels')) 408 name_field = 'name' 409 objects = model_logic.ExtendedManager() 410 411 412 class Meta: 413 db_table = 'afe_autotests' 414 415 def __unicode__(self): 416 return unicode(self.name) 417 418 419class Profiler(dbmodels.Model, model_logic.ModelExtensions): 420 """\ 421 Required: 422 name: profiler name 423 test_type: Client or Server 424 425 Optional: 426 description: arbirary text description 427 """ 428 name = dbmodels.CharField(max_length=255, unique=True) 429 description = dbmodels.TextField(blank=True) 430 431 name_field = 'name' 432 objects = model_logic.ExtendedManager() 433 434 435 class Meta: 436 db_table = 'afe_profilers' 437 438 def __unicode__(self): 439 return unicode(self.name) 440 441 442class AclGroup(dbmodels.Model, model_logic.ModelExtensions): 443 """\ 444 Required: 445 name: name of ACL group 446 447 Optional: 448 description: arbitrary description of group 449 """ 450 name = dbmodels.CharField(max_length=255, unique=True) 451 description = dbmodels.CharField(max_length=255, blank=True) 452 users = dbmodels.ManyToManyField(User, blank=False, 453 db_table='afe_acl_groups_users') 454 hosts = dbmodels.ManyToManyField(Host, blank=True, 455 db_table='afe_acl_groups_hosts') 456 457 name_field = 'name' 458 objects = model_logic.ExtendedManager() 459 460 @staticmethod 461 def check_for_acl_violation_hosts(hosts): 462 user = thread_local.get_user() 463 if user.is_superuser(): 464 return 465 accessible_host_ids = set( 466 host.id for host in Host.objects.filter(aclgroup__users=user)) 467 for host in hosts: 468 # Check if the user has access to this host, 469 # but only if it is not a metahost or a one-time-host 470 no_access = (isinstance(host, Host) 471 and not host.invalid 472 and int(host.id) not in accessible_host_ids) 473 if no_access: 474 raise AclAccessViolation("%s does not have access to %s" % 475 (str(user), str(host))) 476 477 478 @staticmethod 479 def check_abort_permissions(queue_entries): 480 """ 481 look for queue entries that aren't abortable, meaning 482 * the job isn't owned by this user, and 483 * the machine isn't ACL-accessible, or 484 * the machine is in the "Everyone" ACL 485 """ 486 user = thread_local.get_user() 487 if user.is_superuser(): 488 return 489 not_owned = queue_entries.exclude(job__owner=user.login) 490 # I do this using ID sets instead of just Django filters because 491 # filtering on M2M dbmodels is broken in Django 0.96. It's better in 492 # 1.0. 493 # TODO: Use Django filters, now that we're using 1.0. 494 accessible_ids = set( 495 entry.id for entry 496 in not_owned.filter(host__aclgroup__users__login=user.login)) 497 public_ids = set(entry.id for entry 498 in not_owned.filter(host__aclgroup__name='Everyone')) 499 cannot_abort = [entry for entry in not_owned.select_related() 500 if entry.id not in accessible_ids 501 or entry.id in public_ids] 502 if len(cannot_abort) == 0: 503 return 504 entry_names = ', '.join('%s-%s/%s' % (entry.job.id, entry.job.owner, 505 entry.host_or_metahost_name()) 506 for entry in cannot_abort) 507 raise AclAccessViolation('You cannot abort the following job entries: ' 508 + entry_names) 509 510 511 def check_for_acl_violation_acl_group(self): 512 user = thread_local.get_user() 513 if user.is_superuser(): 514 return 515 if self.name == 'Everyone': 516 raise AclAccessViolation("You cannot modify 'Everyone'!") 517 if not user in self.users.all(): 518 raise AclAccessViolation("You do not have access to %s" 519 % self.name) 520 521 @staticmethod 522 def on_host_membership_change(): 523 everyone = AclGroup.objects.get(name='Everyone') 524 525 # find hosts that aren't in any ACL group and add them to Everyone 526 # TODO(showard): this is a bit of a hack, since the fact that this query 527 # works is kind of a coincidence of Django internals. This trick 528 # doesn't work in general (on all foreign key relationships). I'll 529 # replace it with a better technique when the need arises. 530 orphaned_hosts = Host.valid_objects.filter(aclgroup__id__isnull=True) 531 everyone.hosts.add(*orphaned_hosts.distinct()) 532 533 # find hosts in both Everyone and another ACL group, and remove them 534 # from Everyone 535 hosts_in_everyone = Host.valid_objects.filter(aclgroup__name='Everyone') 536 acled_hosts = set() 537 for host in hosts_in_everyone: 538 # Has an ACL group other than Everyone 539 if host.aclgroup_set.count() > 1: 540 acled_hosts.add(host) 541 everyone.hosts.remove(*acled_hosts) 542 543 544 def delete(self): 545 if (self.name == 'Everyone'): 546 raise AclAccessViolation("You cannot delete 'Everyone'!") 547 self.check_for_acl_violation_acl_group() 548 super(AclGroup, self).delete() 549 self.on_host_membership_change() 550 551 552 def add_current_user_if_empty(self): 553 if not self.users.count(): 554 self.users.add(thread_local.get_user()) 555 556 557 def perform_after_save(self, change): 558 if not change: 559 self.users.add(thread_local.get_user()) 560 self.add_current_user_if_empty() 561 self.on_host_membership_change() 562 563 564 def save(self, *args, **kwargs): 565 change = bool(self.id) 566 if change: 567 # Check the original object for an ACL violation 568 AclGroup.objects.get(id=self.id).check_for_acl_violation_acl_group() 569 super(AclGroup, self).save(*args, **kwargs) 570 self.perform_after_save(change) 571 572 573 class Meta: 574 db_table = 'afe_acl_groups' 575 576 def __unicode__(self): 577 return unicode(self.name) 578 579 580class JobManager(model_logic.ExtendedManager): 581 'Custom manager to provide efficient status counts querying.' 582 def get_status_counts(self, job_ids): 583 """\ 584 Returns a dictionary mapping the given job IDs to their status 585 count dictionaries. 586 """ 587 if not job_ids: 588 return {} 589 id_list = '(%s)' % ','.join(str(job_id) for job_id in job_ids) 590 cursor = connection.cursor() 591 cursor.execute(""" 592 SELECT job_id, status, aborted, complete, COUNT(*) 593 FROM afe_host_queue_entries 594 WHERE job_id IN %s 595 GROUP BY job_id, status, aborted, complete 596 """ % id_list) 597 all_job_counts = dict((job_id, {}) for job_id in job_ids) 598 for job_id, status, aborted, complete, count in cursor.fetchall(): 599 job_dict = all_job_counts[job_id] 600 full_status = HostQueueEntry.compute_full_status(status, aborted, 601 complete) 602 job_dict.setdefault(full_status, 0) 603 job_dict[full_status] += count 604 return all_job_counts 605 606 607class Job(dbmodels.Model, model_logic.ModelExtensions): 608 """\ 609 owner: username of job owner 610 name: job name (does not have to be unique) 611 priority: Low, Medium, High, Urgent (or 0-3) 612 control_file: contents of control file 613 control_type: Client or Server 614 created_on: date of job creation 615 submitted_on: date of job submission 616 synch_count: how many hosts should be used per autoserv execution 617 run_verify: Whether or not to run the verify phase 618 timeout: hours from queuing time until job times out 619 max_runtime_hrs: hours from job starting time until job times out 620 email_list: list of people to email on completion delimited by any of: 621 white space, ',', ':', ';' 622 dependency_labels: many-to-many relationship with labels corresponding to 623 job dependencies 624 reboot_before: Never, If dirty, or Always 625 reboot_after: Never, If all tests passed, or Always 626 parse_failed_repair: if True, a failed repair launched by this job will have 627 its results parsed as part of the job. 628 """ 629 DEFAULT_TIMEOUT = global_config.global_config.get_config_value( 630 'AUTOTEST_WEB', 'job_timeout_default', default=240) 631 DEFAULT_MAX_RUNTIME_HRS = global_config.global_config.get_config_value( 632 'AUTOTEST_WEB', 'job_max_runtime_hrs_default', default=72) 633 DEFAULT_PARSE_FAILED_REPAIR = global_config.global_config.get_config_value( 634 'AUTOTEST_WEB', 'parse_failed_repair_default', type=bool, 635 default=False) 636 637 Priority = enum.Enum('Low', 'Medium', 'High', 'Urgent') 638 ControlType = enum.Enum('Server', 'Client', start_value=1) 639 640 owner = dbmodels.CharField(max_length=255) 641 name = dbmodels.CharField(max_length=255) 642 priority = dbmodels.SmallIntegerField(choices=Priority.choices(), 643 blank=True, # to allow 0 644 default=Priority.MEDIUM) 645 control_file = dbmodels.TextField() 646 control_type = dbmodels.SmallIntegerField(choices=ControlType.choices(), 647 blank=True, # to allow 0 648 default=ControlType.CLIENT) 649 created_on = dbmodels.DateTimeField() 650 synch_count = dbmodels.IntegerField(null=True, default=1) 651 timeout = dbmodels.IntegerField(default=DEFAULT_TIMEOUT) 652 run_verify = dbmodels.BooleanField(default=True) 653 email_list = dbmodels.CharField(max_length=250, blank=True) 654 dependency_labels = ( 655 dbmodels.ManyToManyField(Label, blank=True, 656 db_table='afe_jobs_dependency_labels')) 657 reboot_before = dbmodels.SmallIntegerField(choices=RebootBefore.choices(), 658 blank=True, 659 default=DEFAULT_REBOOT_BEFORE) 660 reboot_after = dbmodels.SmallIntegerField(choices=RebootAfter.choices(), 661 blank=True, 662 default=DEFAULT_REBOOT_AFTER) 663 parse_failed_repair = dbmodels.BooleanField( 664 default=DEFAULT_PARSE_FAILED_REPAIR) 665 max_runtime_hrs = dbmodels.IntegerField(default=DEFAULT_MAX_RUNTIME_HRS) 666 667 668 # custom manager 669 objects = JobManager() 670 671 672 def is_server_job(self): 673 return self.control_type == self.ControlType.SERVER 674 675 676 @classmethod 677 def create(cls, owner, options, hosts): 678 """\ 679 Creates a job by taking some information (the listed args) 680 and filling in the rest of the necessary information. 681 """ 682 AclGroup.check_for_acl_violation_hosts(hosts) 683 job = cls.add_object( 684 owner=owner, 685 name=options['name'], 686 priority=options['priority'], 687 control_file=options['control_file'], 688 control_type=options['control_type'], 689 synch_count=options.get('synch_count'), 690 timeout=options.get('timeout'), 691 max_runtime_hrs=options.get('max_runtime_hrs'), 692 run_verify=options.get('run_verify'), 693 email_list=options.get('email_list'), 694 reboot_before=options.get('reboot_before'), 695 reboot_after=options.get('reboot_after'), 696 parse_failed_repair=options.get('parse_failed_repair'), 697 created_on=datetime.now()) 698 699 job.dependency_labels = options['dependencies'] 700 return job 701 702 703 def queue(self, hosts, atomic_group=None, is_template=False): 704 """Enqueue a job on the given hosts.""" 705 if not hosts: 706 if atomic_group: 707 # No hosts or labels are required to queue an atomic group 708 # Job. However, if they are given, we respect them below. 709 atomic_group.enqueue_job(self, is_template=is_template) 710 else: 711 # hostless job 712 entry = HostQueueEntry.create(job=self, is_template=is_template) 713 entry.save() 714 return 715 716 for host in hosts: 717 host.enqueue_job(self, atomic_group=atomic_group, 718 is_template=is_template) 719 720 721 def create_recurring_job(self, start_date, loop_period, loop_count, owner): 722 rec = RecurringRun(job=self, start_date=start_date, 723 loop_period=loop_period, 724 loop_count=loop_count, 725 owner=User.objects.get(login=owner)) 726 rec.save() 727 return rec.id 728 729 730 def user(self): 731 try: 732 return User.objects.get(login=self.owner) 733 except self.DoesNotExist: 734 return None 735 736 737 def abort(self, aborted_by): 738 for queue_entry in self.hostqueueentry_set.all(): 739 queue_entry.abort(aborted_by) 740 741 742 def tag(self): 743 return '%s-%s' % (self.id, self.owner) 744 745 746 class Meta: 747 db_table = 'afe_jobs' 748 749 def __unicode__(self): 750 return u'%s (%s-%s)' % (self.name, self.id, self.owner) 751 752 753class IneligibleHostQueue(dbmodels.Model, model_logic.ModelExtensions): 754 job = dbmodels.ForeignKey(Job) 755 host = dbmodels.ForeignKey(Host) 756 757 objects = model_logic.ExtendedManager() 758 759 class Meta: 760 db_table = 'afe_ineligible_host_queues' 761 762 763class HostQueueEntry(dbmodels.Model, model_logic.ModelExtensions): 764 Status = host_queue_entry_states.Status 765 ACTIVE_STATUSES = host_queue_entry_states.ACTIVE_STATUSES 766 COMPLETE_STATUSES = host_queue_entry_states.COMPLETE_STATUSES 767 768 job = dbmodels.ForeignKey(Job) 769 host = dbmodels.ForeignKey(Host, blank=True, null=True) 770 status = dbmodels.CharField(max_length=255) 771 meta_host = dbmodels.ForeignKey(Label, blank=True, null=True, 772 db_column='meta_host') 773 active = dbmodels.BooleanField(default=False) 774 complete = dbmodels.BooleanField(default=False) 775 deleted = dbmodels.BooleanField(default=False) 776 execution_subdir = dbmodels.CharField(max_length=255, blank=True, 777 default='') 778 # If atomic_group is set, this is a virtual HostQueueEntry that will 779 # be expanded into many actual hosts within the group at schedule time. 780 atomic_group = dbmodels.ForeignKey(AtomicGroup, blank=True, null=True) 781 aborted = dbmodels.BooleanField(default=False) 782 started_on = dbmodels.DateTimeField(null=True, blank=True) 783 784 objects = model_logic.ExtendedManager() 785 786 787 def __init__(self, *args, **kwargs): 788 super(HostQueueEntry, self).__init__(*args, **kwargs) 789 self._record_attributes(['status']) 790 791 792 @classmethod 793 def create(cls, job, host=None, meta_host=None, atomic_group=None, 794 is_template=False): 795 if is_template: 796 status = cls.Status.TEMPLATE 797 else: 798 status = cls.Status.QUEUED 799 800 return cls(job=job, host=host, meta_host=meta_host, 801 atomic_group=atomic_group, status=status) 802 803 804 def save(self, *args, **kwargs): 805 self._set_active_and_complete() 806 super(HostQueueEntry, self).save(*args, **kwargs) 807 self._check_for_updated_attributes() 808 809 810 def execution_path(self): 811 """ 812 Path to this entry's results (relative to the base results directory). 813 """ 814 return os.path.join(self.job.tag(), self.execution_subdir) 815 816 817 def host_or_metahost_name(self): 818 if self.host: 819 return self.host.hostname 820 elif self.meta_host: 821 return self.meta_host.name 822 else: 823 assert self.atomic_group, "no host, meta_host or atomic group!" 824 return self.atomic_group.name 825 826 827 def _set_active_and_complete(self): 828 if self.status in self.ACTIVE_STATUSES: 829 self.active, self.complete = True, False 830 elif self.status in self.COMPLETE_STATUSES: 831 self.active, self.complete = False, True 832 else: 833 self.active, self.complete = False, False 834 835 836 def on_attribute_changed(self, attribute, old_value): 837 assert attribute == 'status' 838 logging.info('%s/%d (%d) -> %s' % (self.host, self.job.id, self.id, 839 self.status)) 840 841 842 def is_meta_host_entry(self): 843 'True if this is a entry has a meta_host instead of a host.' 844 return self.host is None and self.meta_host is not None 845 846 847 def log_abort(self, user): 848 if user is None: 849 # automatic system abort (i.e. job timeout) 850 return 851 abort_log = AbortedHostQueueEntry(queue_entry=self, aborted_by=user) 852 abort_log.save() 853 854 855 def abort(self, user): 856 # this isn't completely immune to race conditions since it's not atomic, 857 # but it should be safe given the scheduler's behavior. 858 if not self.complete and not self.aborted: 859 self.log_abort(user) 860 self.aborted = True 861 self.save() 862 863 864 @classmethod 865 def compute_full_status(cls, status, aborted, complete): 866 if aborted and not complete: 867 return 'Aborted (%s)' % status 868 return status 869 870 871 def full_status(self): 872 return self.compute_full_status(self.status, self.aborted, 873 self.complete) 874 875 876 def _postprocess_object_dict(self, object_dict): 877 object_dict['full_status'] = self.full_status() 878 879 880 class Meta: 881 db_table = 'afe_host_queue_entries' 882 883 884 885 def __unicode__(self): 886 hostname = None 887 if self.host: 888 hostname = self.host.hostname 889 return u"%s/%d (%d)" % (hostname, self.job.id, self.id) 890 891 892class AbortedHostQueueEntry(dbmodels.Model, model_logic.ModelExtensions): 893 queue_entry = dbmodels.OneToOneField(HostQueueEntry, primary_key=True) 894 aborted_by = dbmodels.ForeignKey(User) 895 aborted_on = dbmodels.DateTimeField() 896 897 objects = model_logic.ExtendedManager() 898 899 900 def save(self, *args, **kwargs): 901 self.aborted_on = datetime.now() 902 super(AbortedHostQueueEntry, self).save(*args, **kwargs) 903 904 class Meta: 905 db_table = 'afe_aborted_host_queue_entries' 906 907 908class RecurringRun(dbmodels.Model, model_logic.ModelExtensions): 909 """\ 910 job: job to use as a template 911 owner: owner of the instantiated template 912 start_date: Run the job at scheduled date 913 loop_period: Re-run (loop) the job periodically 914 (in every loop_period seconds) 915 loop_count: Re-run (loop) count 916 """ 917 918 job = dbmodels.ForeignKey(Job) 919 owner = dbmodels.ForeignKey(User) 920 start_date = dbmodels.DateTimeField() 921 loop_period = dbmodels.IntegerField(blank=True) 922 loop_count = dbmodels.IntegerField(blank=True) 923 924 objects = model_logic.ExtendedManager() 925 926 class Meta: 927 db_table = 'afe_recurring_run' 928 929 def __unicode__(self): 930 return u'RecurringRun(job %s, start %s, period %s, count %s)' % ( 931 self.job.id, self.start_date, self.loop_period, self.loop_count) 932 933 934class SpecialTask(dbmodels.Model, model_logic.ModelExtensions): 935 """\ 936 Tasks to run on hosts at the next time they are in the Ready state. Use this 937 for high-priority tasks, such as forced repair or forced reinstall. 938 939 host: host to run this task on 940 task: special task to run 941 time_requested: date and time the request for this task was made 942 is_active: task is currently running 943 is_complete: task has finished running 944 time_started: date and time the task started 945 queue_entry: Host queue entry waiting on this task (or None, if task was not 946 started in preparation of a job) 947 """ 948 Task = enum.Enum('Verify', 'Cleanup', 'Repair', string_values=True) 949 950 host = dbmodels.ForeignKey(Host, blank=False, null=False) 951 task = dbmodels.CharField(max_length=64, choices=Task.choices(), 952 blank=False, null=False) 953 requested_by = dbmodels.ForeignKey(User, blank=True, null=True) 954 time_requested = dbmodels.DateTimeField(auto_now_add=True, blank=False, 955 null=False) 956 is_active = dbmodels.BooleanField(default=False, blank=False, null=False) 957 is_complete = dbmodels.BooleanField(default=False, blank=False, null=False) 958 time_started = dbmodels.DateTimeField(null=True, blank=True) 959 queue_entry = dbmodels.ForeignKey(HostQueueEntry, blank=True, null=True) 960 success = dbmodels.BooleanField(default=False, blank=False, null=False) 961 962 objects = model_logic.ExtendedManager() 963 964 965 def save(self, **kwargs): 966 if self.queue_entry: 967 self.requested_by = User.objects.get( 968 login=self.queue_entry.job.owner) 969 super(SpecialTask, self).save(**kwargs) 970 971 972 def execution_path(self): 973 """@see HostQueueEntry.execution_path()""" 974 return 'hosts/%s/%s-%s' % (self.host.hostname, self.id, 975 self.task.lower()) 976 977 978 # property to emulate HostQueueEntry.status 979 @property 980 def status(self): 981 """ 982 Return a host queue entry status appropriate for this task. Although 983 SpecialTasks are not HostQueueEntries, it is helpful to the user to 984 present similar statuses. 985 """ 986 if self.is_complete: 987 if self.success: 988 return HostQueueEntry.Status.COMPLETED 989 return HostQueueEntry.Status.FAILED 990 if self.is_active: 991 return HostQueueEntry.Status.RUNNING 992 return HostQueueEntry.Status.QUEUED 993 994 995 # property to emulate HostQueueEntry.started_on 996 @property 997 def started_on(self): 998 return self.time_started 999 1000 1001 @classmethod 1002 def schedule_special_task(cls, hosts, task): 1003 """ 1004 Schedules hosts for a special task, if the task is not already scheduled 1005 """ 1006 for host in hosts: 1007 if not SpecialTask.objects.filter(host__id=host.id, task=task, 1008 is_active=False, 1009 is_complete=False): 1010 special_task = SpecialTask(host=host, task=task, 1011 requested_by=thread_local.get_user()) 1012 special_task.save() 1013 1014 1015 def activate(self): 1016 """ 1017 Sets a task as active and sets the time started to the current time. 1018 """ 1019 logging.info('Starting: %s', self) 1020 self.is_active = True 1021 self.time_started = datetime.now() 1022 self.save() 1023 1024 1025 def finish(self, success): 1026 """ 1027 Sets a task as completed 1028 """ 1029 logging.info('Finished: %s', self) 1030 self.is_active = False 1031 self.is_complete = True 1032 self.success = success 1033 self.save() 1034 1035 1036 class Meta: 1037 db_table = 'afe_special_tasks' 1038 1039 1040 def __unicode__(self): 1041 result = u'Special Task %s (host %s, task %s, time %s)' % ( 1042 self.id, self.host, self.task, self.time_requested) 1043 if self.is_complete: 1044 result += u' (completed)' 1045 elif self.is_active: 1046 result += u' (active)' 1047 1048 return result 1049