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