commands/jobs.py

   1 #!/usr/bin/env python
   2 #-*- coding:utf-8 -*-
   3 #  Copyright (C) 2010-2013  CEA/DEN
   4 #
   5 #  This library is free software; you can redistribute it and/or
   6 #  modify it under the terms of the GNU Lesser General Public
   7 #  License as published by the Free Software Foundation; either
   8 #  version 2.1 of the License.
   9 #
  10 #  This library is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 #  Lesser General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU Lesser General Public
  16 #  License along with this library; if not, write to the Free Software
  17 #  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  18
  19 import os
  20 import sys
  21 import datetime
  22 import time
  23 import paramiko
  24
  25 import src
  26
  27
  28 parser = src.options.Options()
  29
  30 parser.add_option('j', 'jobs_config', 'string', 'jobs_cfg',
  31                   _('The name of the config file that contains'
  32                   ' the jobs configuration'))
  33 parser.add_option('o', 'only_jobs', 'list2', 'only_jobs',
  34                   _('The list of jobs to launch, by their name. '))
  35 parser.add_option('l', 'list', 'boolean', 'list',
  36                   _('list all available config files.'))
  37 parser.add_option('n', 'no_label', 'boolean', 'no_label',
  38                   _("do not print labels, Works only with --list."), False)
  39 parser.add_option('t', 'test_connection', 'boolean', 'test_connection',
  40                   _("Try to connect to the machines. Not executing the jobs."),
  41                   False)
  42 parser.add_option('p', 'publish', 'boolean', 'publish',
  43                   _("Generate an xml file that can be read in a browser to "
  44                     "display the jobs status."),
  45                   False)
  46
  47 class machine(object):
  48     '''Class to manage a ssh connection on a machine
  49     '''
  50     def __init__(self, host, user, port=22, passwd=None):
  51         self.host = host
  52         self.port = port
  53         self.user = user
  54         self.password = passwd
  55         self.ssh = paramiko.SSHClient()
  56         self._connection_successful = None
  57
  58     def connect(self, logger):
  59         '''Initiate the ssh connection to the remote machine
  60
  61         :param logger src.logger.Logger: The logger instance
  62         :return: Nothing
  63         :rtype: N\A
  64         '''
  65
  66         self._connection_successful = False
  67         self.ssh.load_system_host_keys()
  68         self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
  69         try:
  70             self.ssh.connect(self.host,
  71                              port=self.port,
  72                              username=self.user,
  73                              password = self.password)
  74         except paramiko.AuthenticationException:
  75             message = src.KO_STATUS + _(": authentication failed\n")
  76             logger.write( src.printcolors.printcError(message))
  77         except paramiko.BadHostKeyException:
  78             message = (src.KO_STATUS +
  79                        _(": the server's host key could not be verified\n"))
  80             logger.write( src.printcolors.printcError(message))
  81         except paramiko.SSHException:
  82             message = (src.KO_STATUS +
  83                     _(": error connecting or establishing an SSH session\n"))
  84             logger.write( src.printcolors.printcError(message))
  85         except:
  86             logger.write( src.printcolors.printcError(src.KO_STATUS + '\n'))
  87         else:
  88             self._connection_successful = True
  89             logger.write( src.printcolors.printcSuccess(src.OK_STATUS) + '\n')
  90
  91     def successfully_connected(self, logger):
  92         '''Verify if the connection to the remote machine has succeed
  93
  94         :param logger src.logger.Logger: The logger instance
  95         :return: True if the connection has succeed, False if not
  96         :rtype: bool
  97         '''
  98         if self._connection_successful == None:
  99             message = "Warning : trying to ask if the connection to "
 100             "(host: %s, port: %s, user: %s) is OK whereas there were"
 101             " no connection request" % \
 102             (machine.host, machine.port, machine.user)
 103             logger.write( src.printcolors.printcWarning(message))
 104         return self._connection_successful
 105
 106
 107     def close(self):
 108         '''Close the ssh connection
 109
 110         :rtype: N\A
 111         '''
 112         self.ssh.close()
 113
 114     def exec_command(self, command, logger):
 115         '''Execute the command on the remote machine
 116
 117         :param command str: The command to be run
 118         :param logger src.logger.Logger: The logger instance
 119         :return: the stdin, stdout, and stderr of the executing command,
 120                  as a 3-tuple
 121         :rtype: (paramiko.channel.ChannelFile, paramiko.channel.ChannelFile,
 122                 paramiko.channel.ChannelFile)
 123         '''
 124         try:
 125             # Does not wait the end of the command
 126             (stdin, stdout, stderr) = self.ssh.exec_command(command)
 127         except paramiko.SSHException:
 128             message = src.KO_STATUS + _(
 129                             ": the server failed to execute the command\n")
 130             logger.write( src.printcolors.printcError(message))
 131             return (None, None, None)
 132         except:
 133             logger.write( src.printcolors.printcError(src.KO_STATUS + '\n'))
 134             return (None, None, None)
 135         else:
 136             return (stdin, stdout, stderr)
 137
 138     def write_info(self, logger):
 139         '''Prints the informations relative to the machine in the logger
 140            (terminal traces and log file)
 141
 142         :param logger src.logger.Logger: The logger instance
 143         :return: Nothing
 144         :rtype: N\A
 145         '''
 146         logger.write("host : " + self.host + "\n")
 147         logger.write("port : " + str(self.port) + "\n")
 148         logger.write("user : " + str(self.user) + "\n")
 149         if self.successfully_connected(logger):
 150             status = src.OK_STATUS
 151         else:
 152             status = src.KO_STATUS
 153         logger.write("Connection : " + status + "\n\n")
 154
 155
 156 class job(object):
 157     '''Class to manage one job
 158     '''
 159     def __init__(self, name, machine, application, distribution, commands, timeout, logger, after=None):
 160
 161         self.name = name
 162         self.machine = machine
 163         self.after = after
 164         self.timeout = timeout
 165         self.application = application
 166         self.distribution = distribution
 167         self.logger = logger
 168
 169         self._T0 = -1
 170         self._Tf = -1
 171         self._has_begun = False
 172         self._has_finished = False
 173         self._has_timouted = False
 174         self._stdin = None # Store the command inputs field
 175         self._stdout = None # Store the command outputs field
 176         self._stderr = None # Store the command errors field
 177
 178         self.out = None # Contains something only if the job is finished
 179         self.err = None # Contains something only if the job is finished
 180
 181         self.commands = " ; ".join(commands)
 182
 183     def get_pids(self):
 184         pids = []
 185         for cmd in self.commands.split(" ; "):
 186             cmd_pid = 'ps aux | grep "' + cmd + '" | awk \'{print $2}\''
 187             (_, out_pid, _) = self.machine.exec_command(cmd_pid, self.logger)
 188             pids_cmd = out_pid.readlines()
 189             pids_cmd = [str(src.only_numbers(pid)) for pid in pids_cmd]
 190             pids+=pids_cmd
 191         return pids
 192
 193     def kill_remote_process(self):
 194         '''Kills the process on the remote machine.
 195
 196         :return: (the output of the kill, the error of the kill)
 197         :rtype: (str, str)
 198         '''
 199
 200         pids = self.get_pids()
 201         cmd_kill = " ; ".join([("kill -9 " + pid) for pid in pids])
 202         (_, out_kill, err_kill) = self.machine.exec_command(cmd_kill,
 203                                                             self.logger)
 204         return (out_kill, err_kill)
 205
 206     def has_begun(self):
 207         '''Returns True if the job has already begun
 208
 209         :return: True if the job has already begun
 210         :rtype: bool
 211         '''
 212         return self._has_begun
 213
 214     def has_finished(self):
 215         '''Returns True if the job has already finished
 216            (i.e. all the commands have been executed)
 217            If it is finished, the outputs are stored in the fields out and err.
 218
 219         :return: True if the job has already finished
 220         :rtype: bool
 221         '''
 222
 223         # If the method has already been called and returned True
 224         if self._has_finished:
 225             return True
 226
 227         # If the job has not begun yet
 228         if not self.has_begun():
 229             return False
 230
 231         if self._stdout.channel.closed:
 232             self._has_finished = True
 233             # And store the result outputs
 234             self.out = self._stdout.read()
 235             self.err = self._stderr.read()
 236             # And put end time
 237             self._Tf = time.time()
 238
 239         return self._has_finished
 240
 241     def is_running(self):
 242         '''Returns True if the job commands are running
 243
 244         :return: True if the job is running
 245         :rtype: bool
 246         '''
 247         return self.has_begun() and not self.has_finished()
 248
 249     def is_timeout(self):
 250         '''Returns True if the job commands has finished with timeout
 251
 252         :return: True if the job has finished with timeout
 253         :rtype: bool
 254         '''
 255         return self._has_timouted
 256
 257     def time_elapsed(self):
 258         if not self.has_begun():
 259             return -1
 260         T_now = time.time()
 261         return T_now - self._T0
 262
 263     def check_time(self):
 264         if not self.has_begun():
 265             return
 266         if self.time_elapsed() > self.timeout:
 267             self._has_finished = True
 268             self._has_timouted = True
 269             self._Tf = time.time()
 270             self.get_pids()
 271             (out_kill, _) = self.kill_remote_process()
 272             self.out = "TIMEOUT \n" + out_kill.read()
 273             self.err = "TIMEOUT : %s seconds elapsed\n" % str(self.timeout)
 274
 275     def total_duration(self):
 276         return self._Tf - self._T0
 277
 278     def run(self, logger):
 279         if self.has_begun():
 280             print("Warn the user that a job can only be launched one time")
 281             return
 282
 283         if not self.machine.successfully_connected(logger):
 284             self._has_finished = True
 285             self.out = "N\A"
 286             self.err = ("Connection to machine (host: %s, port: %s, user: %s) has failed"
 287                         % (self.machine.host, self.machine.port, self.machine.user))
 288         else:
 289             self._T0 = time.time()
 290             self._stdin, self._stdout, self._stderr = self.machine.exec_command(
 291                                                         self.commands, logger)
 292             if (self._stdin, self._stdout, self._stderr) == (None, None, None):
 293                 self._has_finished = True
 294                 self._Tf = time.time()
 295                 self.out = "N\A"
 296                 self.err = "The server failed to execute the command"
 297
 298         self._has_begun = True
 299
 300     def write_results(self, logger):
 301         logger.write("name : " + self.name + "\n")
 302         if self.after:
 303             logger.write("after : %s\n" % self.after)
 304         logger.write("Time elapsed : %4imin %2is \n" %
 305                      (self.total_duration()/60 , self.total_duration()%60))
 306         if self._T0 != -1:
 307             logger.write("Begin time : %s\n" %
 308                          time.strftime('%Y-%m-%d %H:%M:%S',
 309                                        time.localtime(self._T0)) )
 310         if self._Tf != -1:
 311             logger.write("End time   : %s\n\n" %
 312                          time.strftime('%Y-%m-%d %H:%M:%S',
 313                                        time.localtime(self._Tf)) )
 314
 315         machine_head = "Informations about connection :\n"
 316         underline = (len(machine_head) - 2) * "-"
 317         logger.write(src.printcolors.printcInfo(machine_head + underline + "\n"))
 318         self.machine.write_info(logger)
 319
 320         logger.write(src.printcolors.printcInfo("out : \n"))
 321         if self.out is None:
 322             logger.write("Unable to get output\n")
 323         else:
 324             logger.write(self.out + "\n")
 325         logger.write(src.printcolors.printcInfo("err : \n"))
 326         if self.err is None:
 327             logger.write("Unable to get error\n")
 328         else:
 329             logger.write(self.err + "\n")
 330
 331     def get_status(self):
 332         if not self.machine.successfully_connected(self.logger):
 333             return "SSH connection KO"
 334         if not self.has_begun():
 335             return "Not launched"
 336         if self.is_running():
 337             return "running since " + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self._T0))
 338         if self.has_finished():
 339             if self.is_timeout():
 340                 return "Timeout since " + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self._Tf))
 341             return "Finished since " + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self._Tf))
 342
 343 class Jobs(object):
 344     '''Class to manage the jobs to be run
 345     '''
 346     def __init__(self, runner, logger, config_jobs, lenght_columns = 20):
 347         # The jobs configuration
 348         self.cfg_jobs = config_jobs
 349         # The machine that will be used today
 350         self.lmachines = []
 351         # The list of machine (hosts, port) that will be used today
 352         # (a same host can have several machine instances since there
 353         # can be several ssh parameters)
 354         self.lhosts = []
 355         # The jobs to be launched today
 356         self.ljobs = []
 357         self.runner = runner
 358         self.logger = logger
 359         # The correlation dictionary between jobs and machines
 360         self.dic_job_machine = {}
 361         self.len_columns = lenght_columns
 362
 363         # the list of jobs that have not been run yet
 364         self._l_jobs_not_started = []
 365         # the list of jobs that have already ran
 366         self._l_jobs_finished = []
 367         # the list of jobs that are running
 368         self._l_jobs_running = []
 369
 370         self.determine_products_and_machines()
 371
 372     def define_job(self, job_def, machine):
 373         '''Takes a pyconf job definition and a machine (from class machine)
 374            and returns the job instance corresponding to the definition.
 375
 376         :param job_def src.config.Mapping: a job definition
 377         :param machine machine: the machine on which the job will run
 378         :return: The corresponding job in a job class instance
 379         :rtype: job
 380         '''
 381         name = job_def.name
 382         cmmnds = job_def.commands
 383         timeout = job_def.timeout
 384         after = None
 385         if 'after' in job_def:
 386             after = job_def.after
 387         application = None
 388         if 'application' in job_def:
 389             application = job_def.application
 390         distribution = None
 391         if 'distribution' in job_def:
 392             distribution = job_def.distribution
 393
 394         return job(name, machine, application, distribution, cmmnds, timeout, self.logger, after = after)
 395
 396     def determine_products_and_machines(self):
 397         '''Function that reads the pyconf jobs definition and instantiates all
 398            the machines and jobs to be done today.
 399
 400         :return: Nothing
 401         :rtype: N\A
 402         '''
 403         today = datetime.date.weekday(datetime.date.today())
 404         host_list = []
 405
 406         for job_def in self.cfg_jobs.jobs :
 407             if today in job_def.when:
 408                 if 'host' not in job_def:
 409                     host = self.runner.cfg.VARS.hostname
 410                 else:
 411                     host = job_def.host
 412
 413                 if 'port' not in job_def:
 414                     port = 22
 415                 else:
 416                     port = job_def.port
 417
 418                 if (host, port) not in host_list:
 419                     host_list.append((host, port))
 420
 421                 if 'user' not in job_def:
 422                     user = self.runner.cfg.VARS.user
 423                 else:
 424                     user = job_def.user
 425
 426                 if 'password' not in job_def:
 427                     passwd = None
 428                 else:
 429                     passwd = job_def.password
 430
 431                 a_machine = machine(host, user, port=port, passwd=passwd)
 432
 433                 self.lmachines.append(a_machine)
 434
 435                 a_job = self.define_job(job_def, a_machine)
 436
 437                 self.ljobs.append(a_job)
 438
 439                 self.dic_job_machine[a_job] = a_machine
 440
 441         self.lhosts = host_list
 442
 443     def ssh_connection_all_machines(self, pad=50):
 444         '''Function that do the ssh connection to every machine
 445            to be used today.
 446
 447         :return: Nothing
 448         :rtype: N\A
 449         '''
 450         self.logger.write(src.printcolors.printcInfo((
 451                         "Establishing connection with all the machines :\n")))
 452         for machine in self.lmachines:
 453             # little algorithm in order to display traces
 454             begin_line = ("(host: %s, port: %s, user: %s)" %
 455                           (machine.host, machine.port, machine.user))
 456             if pad - len(begin_line) < 0:
 457                 endline = " "
 458             else:
 459                 endline = (pad - len(begin_line)) * "." + " "
 460             self.logger.write( begin_line + endline )
 461             self.logger.flush()
 462             # the call to the method that initiate the ssh connection
 463             machine.connect(self.logger)
 464         self.logger.write("\n")
 465
 466
 467     def is_occupied(self, hostname):
 468         '''Function that returns True if a job is running on
 469            the machine defined by its host and its port.
 470
 471         :param hostname (str, int): the pair (host, port)
 472         :return: the job that is running on the host,
 473                 or false if there is no job running on the host.
 474         :rtype: job / bool
 475         '''
 476         host = hostname[0]
 477         port = hostname[1]
 478         for jb in self.dic_job_machine:
 479             if jb.machine.host == host and jb.machine.port == port:
 480                 if jb.is_running():
 481                     return jb
 482         return False
 483
 484     def update_jobs_states_list(self):
 485         '''Function that updates the lists that store the currently
 486            running jobs and the jobs that have already finished.
 487
 488         :return: Nothing.
 489         :rtype: N\A
 490         '''
 491         jobs_finished_list = []
 492         jobs_running_list = []
 493         for jb in self.dic_job_machine:
 494             if jb.is_running():
 495                 jobs_running_list.append(jb)
 496                 jb.check_time()
 497             if jb.has_finished():
 498                 jobs_finished_list.append(jb)
 499
 500         nb_job_finished_before = len(self._l_jobs_finished)
 501         self._l_jobs_finished = jobs_finished_list
 502         self._l_jobs_running = jobs_running_list
 503
 504         nb_job_finished_now = len(self._l_jobs_finished)
 505
 506         return nb_job_finished_now > nb_job_finished_before
 507
 508
 509     def findJobThatHasName(self, name):
 510         '''Returns the job by its name.
 511
 512         :param name str: a job name
 513         :return: the job that has the name.
 514         :rtype: job
 515         '''
 516         for jb in self.ljobs:
 517             if jb.name == name:
 518                 return jb
 519
 520         # the following is executed only if the job was not found
 521         msg = _('The job "%s" seems to be nonexistent') % name
 522         raise src.SatException(msg)
 523
 524     def str_of_length(self, text, length):
 525         '''Takes a string text of any length and returns
 526            the most close string of length "length".
 527
 528         :param text str: any string
 529         :param length int: a length for the returned string
 530         :return: the most close string of length "length"
 531         :rtype: str
 532         '''
 533         if len(text) > length:
 534             text_out = text[:length-3] + '...'
 535         else:
 536             diff = length - len(text)
 537             before = " " * (diff/2)
 538             after = " " * (diff/2 + diff%2)
 539             text_out = before + text + after
 540
 541         return text_out
 542
 543     def display_status(self, len_col):
 544         '''Takes a lenght and construct the display of the current status
 545            of the jobs in an array that has a column for each host.
 546            It displays the job that is currently running on the host
 547            of the column.
 548
 549         :param len_col int: the size of the column
 550         :return: Nothing
 551         :rtype: N\A
 552         '''
 553
 554         display_line = ""
 555         for host_port in self.lhosts:
 556             jb = self.is_occupied(host_port)
 557             if not jb: # nothing running on the host
 558                 empty = self.str_of_length("empty", len_col)
 559                 display_line += "|" + empty
 560             else:
 561                 display_line += "|" + src.printcolors.printcInfo(
 562                                         self.str_of_length(jb.name, len_col))
 563
 564         self.logger.write("\r" + display_line + "|")
 565         self.logger.flush()
 566
 567
 568     def run_jobs(self):
 569         '''The main method. Runs all the jobs on every host.
 570            For each host, at a given time, only one job can be running.
 571            The jobs that have the field after (that contain the job that has
 572            to be run before it) are run after the previous job.
 573            This method stops when all the jobs are finished.
 574
 575         :return: Nothing
 576         :rtype: N\A
 577         '''
 578
 579         # Print header
 580         self.logger.write(src.printcolors.printcInfo(
 581                                                 _('Executing the jobs :\n')))
 582         text_line = ""
 583         for host_port in self.lhosts:
 584             host = host_port[0]
 585             port = host_port[1]
 586             if port == 22: # default value
 587                 text_line += "|" + self.str_of_length(host, self.len_columns)
 588             else:
 589                 text_line += "|" + self.str_of_length(
 590                                 "("+host+", "+str(port)+")", self.len_columns)
 591
 592         tiret_line = " " + "-"*(len(text_line)-1) + "\n"
 593         self.logger.write(tiret_line)
 594         self.logger.write(text_line + "|\n")
 595         self.logger.write(tiret_line)
 596         self.logger.flush()
 597
 598         # The infinite loop that runs the jobs
 599         l_jobs_not_started = self.dic_job_machine.keys()
 600         while len(self._l_jobs_finished) != len(self.dic_job_machine.keys()):
 601             new_job_start = False
 602             for host_port in self.lhosts:
 603
 604                 if self.is_occupied(host_port):
 605                     continue
 606
 607                 for jb in l_jobs_not_started:
 608                     if (jb.machine.host, jb.machine.port) != host_port:
 609                         continue
 610                     if jb.after == None:
 611                         jb.run(self.logger)
 612                         l_jobs_not_started.remove(jb)
 613                         new_job_start = True
 614                         break
 615                     else:
 616                         jb_before = self.findJobThatHasName(jb.after)
 617                         if jb_before.has_finished():
 618                             jb.run(self.logger)
 619                             l_jobs_not_started.remove(jb)
 620                             new_job_start = True
 621                             break
 622
 623             new_job_finished = self.update_jobs_states_list()
 624
 625             if new_job_start or new_job_finished:
 626                 self.gui.update_xml_file(self.ljobs)
 627                 # Display the current status
 628                 self.display_status(self.len_columns)
 629
 630             # Make sure that the proc is not entirely busy
 631             time.sleep(0.001)
 632
 633         self.logger.write("\n")
 634         self.logger.write(tiret_line)
 635         self.logger.write("\n\n")
 636
 637         self.gui.update_xml_file(self.ljobs)
 638         self.gui.last_update()
 639
 640     def write_all_results(self):
 641         '''Display all the jobs outputs.
 642
 643         :return: Nothing
 644         :rtype: N\A
 645         '''
 646
 647         for jb in self.dic_job_machine.keys():
 648             self.logger.write(src.printcolors.printcLabel(
 649                         "#------- Results for job %s -------#\n" % jb.name))
 650             jb.write_results(self.logger)
 651             self.logger.write("\n\n")
 652
 653 class Gui(object):
 654     '''Class to manage the the xml data that can be displayed in a browser to
 655        see the jobs states
 656     '''
 657
 658     """
 659     <?xml version='1.0' encoding='utf-8'?>
 660     <?xml-stylesheet type='text/xsl' href='job_report.xsl'?>
 661     <JobsReport>
 662       <infos>
 663         <info name="generated" value="2016-06-02 07:06:45"/>
 664       </infos>
 665       <hosts>
 666           <host name=is221553 port=22 distribution=UB12.04/>
 667           <host name=is221560 port=22/>
 668           <host name=is221553 port=22 distribution=FD20/>
 669       </hosts>
 670       <applications>
 671           <application name=SALOME-7.8.0/>
 672           <application name=SALOME-master/>
 673           <application name=MED-STANDALONE-master/>
 674           <application name=CORPUS/>
 675       </applications>
 676
 677       <jobs>
 678           <job name="7.8.0 FD22">
 679                 <host>is228809</host>
 680                 <port>2200</port>
 681                 <application>SALOME-7.8.0</application>
 682                 <user>adminuser</user>
 683                 <timeout>240</timeout>
 684                 <commands>
 685                     export DISPLAY=is221560
 686                     scp -p salome@is221560.intra.cea.fr:/export/home/salome/SALOME-7.7.1p1-src.tgz /local/adminuser
 687                     tar xf /local/adminuser/SALOME-7.7.1p1-src.tgz -C /local/adminuser
 688                 </commands>
 689                 <state>Not launched</state>
 690           </job>
 691
 692           <job name="master MG05">
 693                 <host>is221560</host>
 694                 <port>22</port>
 695                 <application>SALOME-master</application>
 696                 <user>salome</user>
 697                 <timeout>240</timeout>
 698                 <commands>
 699                     export DISPLAY=is221560
 700                     scp -p salome@is221560.intra.cea.fr:/export/home/salome/SALOME-7.7.1p1-src.tgz /local/adminuser
 701                     sat prepare SALOME-master
 702                     sat compile SALOME-master
 703                     sat check SALOME-master
 704                     sat launcher SALOME-master
 705                     sat test SALOME-master
 706                 </commands>
 707                 <state>Running since 23 min</state>
 708                 <!-- <state>time out</state> -->
 709                 <!-- <state>OK</state> -->
 710                 <!-- <state>KO</state> -->
 711                 <begin>10/05/2016 20h32</begin>
 712                 <end>10/05/2016 22h59</end>
 713           </job>
 714
 715       </jobs>
 716     </JobsReport>
 717
 718     """
 719
 720     def __init__(self, xml_file_path, l_jobs, stylesheet):
 721         # The path of the xml file
 722         self.xml_file_path = xml_file_path
 723         # The stylesheet
 724         self.stylesheet = stylesheet
 725         # Open the file in a writing stream
 726         self.xml_file = src.xmlManager.XmlLogFile(xml_file_path, "JobsReport")
 727         # Create the lines and columns
 728         self.initialize_array(l_jobs)
 729         # Write the wml file
 730         self.update_xml_file(l_jobs)
 731
 732     def initialize_array(self, l_jobs):
 733         l_dist = []
 734         l_applications = []
 735         for job in l_jobs:
 736             distrib = job.distribution
 737             if distrib not in l_dist:
 738                 l_dist.append(distrib)
 739
 740             application = job.application
 741             if application not in l_applications:
 742                 l_applications.append(application)
 743
 744         self.l_dist = l_dist
 745         self.l_applications = l_applications
 746
 747         # Update the hosts node
 748         self.xmldists = self.xml_file.add_simple_node("distributions")
 749         for dist_name in self.l_dist:
 750             src.xmlManager.add_simple_node(self.xmldists, "dist", attrib={"name" : dist_name})
 751
 752         # Update the applications node
 753         self.xmlapplications = self.xml_file.add_simple_node("applications")
 754         for application in self.l_applications:
 755             src.xmlManager.add_simple_node(self.xmlapplications, "application", attrib={"name" : application})
 756
 757         # Initialize the jobs node
 758         self.xmljobs = self.xml_file.add_simple_node("jobs")
 759
 760         # Initialize the info node (when generated)
 761         self.xmlinfos = self.xml_file.add_simple_node("infos", attrib={"name" : "last update", "JobsCommandStatus" : "running"})
 762
 763     def update_xml_file(self, l_jobs):
 764
 765         # Update the job names and status node
 766         for job in l_jobs:
 767             # Find the node corresponding to the job and delete it
 768             # in order to recreate it
 769             for xmljob in self.xmljobs.findall('job'):
 770                 if xmljob.attrib['name'] == job.name:
 771                     self.xmljobs.remove(xmljob)
 772
 773             # recreate the job node
 774             xmlj = src.xmlManager.add_simple_node(self.xmljobs, "job", attrib={"name" : job.name})
 775             src.xmlManager.add_simple_node(xmlj, "host", job.machine.host)
 776             src.xmlManager.add_simple_node(xmlj, "port", str(job.machine.port))
 777             src.xmlManager.add_simple_node(xmlj, "user", job.machine.user)
 778             src.xmlManager.add_simple_node(xmlj, "application", job.application)
 779             src.xmlManager.add_simple_node(xmlj, "distribution", job.distribution)
 780             src.xmlManager.add_simple_node(xmlj, "timeout", str(job.timeout))
 781             src.xmlManager.add_simple_node(xmlj, "commands", job.commands)
 782             src.xmlManager.add_simple_node(xmlj, "state", job.get_status())
 783             src.xmlManager.add_simple_node(xmlj, "begin", str(job._T0))
 784             src.xmlManager.add_simple_node(xmlj, "end", str(job._Tf))
 785             src.xmlManager.add_simple_node(xmlj, "out", job.out)
 786             src.xmlManager.add_simple_node(xmlj, "err", job.err)
 787
 788         # Update the date
 789         src.xmlManager.append_node_attrib(self.xmlinfos,
 790                     attrib={"value" :
 791                     datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
 792
 793         # Write the file
 794         self.write_xml_file()
 795
 796     def last_update(self):
 797         src.xmlManager.append_node_attrib(self.xmlinfos,
 798                     attrib={"JobsCommandStatus" : "finished"})
 799         # Write the file
 800         self.write_xml_file()
 801
 802     def write_xml_file(self):
 803         self.xml_file.write_tree(self.stylesheet)
 804
 805 def print_info(logger, arch, JobsFilePath):
 806     '''Prints information header..
 807
 808     :param logger src.logger.Logger: The logger instance
 809     :param arch str: a string that gives the architecture of the machine on
 810                      which the command is launched
 811     :param JobsFilePath str: The path of the file
 812                              that contains the jobs configuration
 813     :return: Nothing
 814     :rtype: N\A
 815     '''
 816     info = [
 817         (_("Platform"), arch),
 818         (_("File containing the jobs configuration"), JobsFilePath)
 819     ]
 820
 821     smax = max(map(lambda l: len(l[0]), info))
 822     for i in info:
 823         sp = " " * (smax - len(i[0]))
 824         src.printcolors.print_value(logger, sp + i[0], i[1], 2)
 825     logger.write("\n", 2)
 826
 827 ##
 828 # Describes the command
 829 def description():
 830     return _("The jobs command launches maintenances that are described"
 831              " in the dedicated jobs configuration file.")
 832
 833 ##
 834 # Runs the command.
 835 def run(args, runner, logger):
 836     (options, args) = parser.parse_args(args)
 837
 838     jobs_cfg_files_dir = runner.cfg.SITE.jobs.config_path
 839
 840     # Make sure the path to the jobs config files directory exists
 841     if not os.path.exists(jobs_cfg_files_dir):
 842         logger.write(_("Creating directory %s\n") %
 843                      src.printcolors.printcLabel(jobs_cfg_files_dir), 1)
 844         os.mkdir(jobs_cfg_files_dir)
 845
 846     # list option : display all the available config files
 847     if options.list:
 848         lcfiles = []
 849         if not options.no_label:
 850             sys.stdout.write("------ %s\n" %
 851                              src.printcolors.printcHeader(jobs_cfg_files_dir))
 852
 853         for f in sorted(os.listdir(jobs_cfg_files_dir)):
 854             if not f.endswith('.pyconf'):
 855                 continue
 856             cfilename = f[:-7]
 857             lcfiles.append(cfilename)
 858             sys.stdout.write("%s\n" % cfilename)
 859         return 0
 860
 861     # Make sure the jobs_config option has been called
 862     if not options.jobs_cfg:
 863         message = _("The option --jobs_config is required\n")
 864         raise src.SatException( message )
 865
 866     # Make sure the invoked file exists
 867     file_jobs_cfg = os.path.join(jobs_cfg_files_dir, options.jobs_cfg)
 868     if not file_jobs_cfg.endswith('.pyconf'):
 869         file_jobs_cfg += '.pyconf'
 870
 871     if not os.path.exists(file_jobs_cfg):
 872         message = _("The file %s does not exist.\n") % file_jobs_cfg
 873         logger.write(src.printcolors.printcError(message), 1)
 874         message = _("The possible files are :\n")
 875         logger.write( src.printcolors.printcInfo(message), 1)
 876         for f in sorted(os.listdir(jobs_cfg_files_dir)):
 877             if not f.endswith('.pyconf'):
 878                 continue
 879             jobscfgname = f[:-7]
 880             sys.stdout.write("%s\n" % jobscfgname)
 881         raise src.SatException( _("No corresponding file") )
 882
 883     print_info(logger, runner.cfg.VARS.dist, file_jobs_cfg)
 884
 885     # Read the config that is in the file
 886     config_jobs = src.read_config_from_a_file(file_jobs_cfg)
 887     if options.only_jobs:
 888         l_jb = src.pyconf.Sequence()
 889         for jb in config_jobs.jobs:
 890             if jb.name in options.only_jobs:
 891                 l_jb.append(jb,
 892                 "Adding a job that was given in only_jobs option parameters")
 893         config_jobs.jobs = l_jb
 894
 895     # Initialization
 896     today_jobs = Jobs(runner, logger, config_jobs)
 897     # SSH connection to all machines
 898     today_jobs.ssh_connection_all_machines()
 899     if options.test_connection:
 900         return 0
 901
 902     gui = None
 903     if options.publish:
 904         gui = Gui("/export/home/serioja/LOGS/test.xml", today_jobs.ljobs, "job_report.xsl")
 905
 906     today_jobs.gui = gui
 907
 908     try:
 909         # Run all the jobs contained in config_jobs
 910         today_jobs.run_jobs()
 911     except KeyboardInterrupt:
 912         logger.write("\n\n%s\n\n" %
 913                 (src.printcolors.printcWarning(_("Forced interruption"))), 1)
 914     finally:
 915         # find the potential not finished jobs and kill them
 916         for jb in today_jobs.ljobs:
 917             if not jb.has_finished():
 918                 jb.kill_remote_process()
 919
 920         # Output the results
 921         today_jobs.write_all_results()