src/nepi/resources/all/collector.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program 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
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 from nepi.execution.attribute import Attribute, Flags, Types
  21 from nepi.execution.trace import Trace, TraceAttr
  22 from nepi.execution.resource import ResourceManager, clsinit_copy, \
  23         ResourceState, ResourceAction
  24 from nepi.util.sshfuncs import ProcStatus
  25
  26 import os
  27 import tempfile
  28
  29 @clsinit_copy
  30 class Collector(ResourceManager):
  31     """ The collector is reponsible of collecting traces
  32     of a same type associated to RMs into a local directory.
  33
  34     .. class:: Class Args :
  35
  36         :param ec: The Experiment controller
  37         :type ec: ExperimentController
  38         :param guid: guid of the RM
  39         :type guid: int
  40
  41     """
  42     _rtype = "Collector"
  43     _help = "A Collector can be attached to a trace name on another " \
  44         "ResourceManager and will retrieve and store the trace content " \
  45         "in a local file at the end of the experiment"
  46     _backend_type = "all"
  47
  48     @classmethod
  49     def _register_attributes(cls):
  50         trace_name = Attribute("traceName",
  51                 "Name of the trace to be collected",
  52                 flags = Flags.Design)
  53
  54         store_dir = Attribute("storeDir",
  55                 "Path to local directory to store trace results",
  56                 default = tempfile.gettempdir(),
  57                 flags = Flags.Design)
  58
  59         user_run_id = Attribute("useRunId",
  60                 "If set to True stores traces into a sub directory named after "
  61                 "the RUN ID assigned by the EC",
  62                 type = Types.Bool,
  63                 default = False,
  64                 flags = Flags.Design)
  65
  66         sub_dir = Attribute("subDir",
  67                 "Sub directory to collect traces into",
  68                 flags = Flags.Design)
  69
  70         rename = Attribute("rename",
  71                 "Name to give to the collected trace file",
  72                 flags = Flags.Design)
  73
  74         cls._register_attribute(trace_name)
  75         cls._register_attribute(store_dir)
  76         cls._register_attribute(sub_dir)
  77         cls._register_attribute(rename)
  78         cls._register_attribute(useRunId)
  79
  80     def __init__(self, ec, guid):
  81         super(Collector, self).__init__(ec, guid)
  82         self._store_path =  None
  83
  84     @property
  85     def store_path(self):
  86         return self._store_path
  87
  88     def do_provision(self):
  89         trace_name = self.get("traceName")
  90         if not trace_name:
  91             self.fail()
  92
  93             msg = "No traceName was specified"
  94             self.error(msg)
  95             raise RuntimeError, msg
  96
  97         self._store_path = self.get("storeDir")
  98
  99         if self.get("useRunId"):
 100             self._store_path = os.path.join(self._store_path, self.ec.run_id)
 101
 102         subdir = self.get("subDir")
 103         if subdir:
 104             self._store_path = os.path.join(self._store_path, subdir)
 105
 106         msg = "Creating local directory at %s to store %s traces " % (
 107             store_dir, trace_name)
 108         self.info(msg)
 109
 110         try:
 111             os.makedirs(self.store_path)
 112         except OSError:
 113             pass
 114
 115         super(Collector, self).do_provision()
 116
 117     def do_deploy(self):
 118         self.do_discover()
 119         self.do_provision()
 120
 121         super(Collector, self).do_deploy()
 122
 123     def do_release(self):
 124         trace_name = self.get("traceName")
 125         rename = self.get("rename") or trace_name
 126
 127         msg = "Collecting '%s' traces to local directory %s" % (
 128             trace_name, self.store_path)
 129         self.info(msg)
 130
 131         rms = self.get_connected()
 132         for rm in rms:
 133             result = self.ec.trace(rm.guid, trace_name)
 134             fpath = os.path.join(self.store_path, "%d.%s" % (rm.guid,
 135                 rename))
 136             try:
 137                 f = open(fpath, "w")
 138                 f.write(result)
 139                 f.close()
 140             except:
 141                 import traceback
 142                 err = traceback.format_exc()
 143                 msg = "Couldn't retrieve trace %s for %d at %s " % (trace_name,
 144                         rm.guid, fpath)
 145                 self.error(msg, out = "", err = err)
 146                 continue
 147
 148         super(Collector, self).do_release()
 149
 150     def valid_connection(self, guid):
 151         # TODO: Validate!
 152         return True
 153