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