Package Products :: Package DataCollector :: Module zenmodeler |
[hide private]
[frames] | no frames]

# Source Code for Module Products.DataCollector.zenmodeler

       1  ##########################################################################
       2  #
       3  # This program is part of Zenoss Core, an open source monitoring platform.
       4  # Copyright (C) 2007, Zenoss Inc.
       5  #
       6  # This program is free software; you can redistribute it and/or modify it
       7  # under the terms of the GNU General Public License version 2 as published by
       8  # the Free Software Foundation.
       9  #
      10  # For complete information please visit: http://www.zenoss.com/oss/
      11  #
      12  ###########################################################################
      13
      14  __doc__= """Discover (aka model) a device and its components.
      15  For instance, find out what Ethernet interfaces and hard disks a server
      16  has available.
      17  This information should change much less frequently than performance metrics.
      18  """
      19
      20  # IMPORTANT! The import of the pysamba.twisted.reactor module should come before
      21  # any other libraries that might possibly use twisted. This will ensure that
      22  # the proper WmiReactor is installed before anyone else grabs a reference to
      23  # the wrong reactor.
      24  import pysamba.twisted.reactor
      25
      26  import Globals
      27  from Products.ZenWin.WMIClient import WMIClient
      28  from Products.ZenWin.utils import addNTLMv2Option, setNTLMv2Auth
      29  from Products.ZenHub.PBDaemon import FakeRemote, PBDaemon
      30  from Products.ZenUtils.DaemonStats import DaemonStats
      31  from Products.ZenUtils.Driver import drive, driveLater
      32  from Products.ZenUtils.Utils import unused
      33  from Products.ZenEvents.ZenEventClasses import Heartbeat, Error
      34
      35  from PythonClient   import PythonClient
      36  from SshClient      import SshClient
      37  from TelnetClient   import TelnetClient, buildOptions as TCbuildOptions
      38  from SnmpClient     import SnmpClient
      39  from PortscanClient import PortscanClient
      40
      41  from Products.DataCollector import Classifier
      42
      43  from twisted.internet import reactor
      44  from twisted.internet.defer import succeed
      45
      46  import time
      47  import types
      48  import re
      49  import DateTime
      50
      51  import os
      52  import os.path
      53  import sys
      54  import traceback
      55  from random import randint
      56
      57  defaultPortScanTimeout = 5
      58  defaultParallel = 1
      59  defaultProtocol = "ssh"
      60  defaultPort = 22
      61
      62  # needed for Twisted's PB (Perspective Broker) to work
      63  from Products.DataCollector import DeviceProxy
      64  from Products.DataCollector import Plugins
      65  unused(DeviceProxy, Plugins)
      66
      67 -class ZenModeler(PBDaemon):
      68      """
      69      Daemon class to attach to zenhub and pass along
      70      device configuration information.
      71      """
      72
      73      name = 'zenmodeler'
      74      initialServices = PBDaemon.initialServices + ['ModelerService']
      75
      76      generateEvents = True
      77      configCycleInterval = 360
      78
      79      classCollectorPlugins = ()
      80
      81 -    def __init__(self, single=False ):
      82          """
      83          Initalizer
      84
      85          @param single: collect from a single device?
      86          @type single: boolean
      87          """
      88          PBDaemon.__init__(self)
      89          # FIXME: cleanup --force option #2660
      90          self.options.force = True
      91          self.start = None
      92          self.rrdStats = DaemonStats()
      93          self.single = single
      94          if self.options.device:
      95              self.single = True
      96          self.modelerCycleInterval = self.options.cycletime
      97          self.collage = float( self.options.collage ) / 1440.0
      98          self.pendingNewClients = False
      99          self.clients = []
     100          self.finished = []
     101          self.devicegen = None
     102
     103          # Delay start for between 10 and 60 minutes when run as a daemon.
     104          self.started = False
     105          self.startDelay = 0
     106          if self.options.daemon:
     107              if self.options.now:
     108                  self.log.debug("Run as a daemon, starting immediately.")
     109              else:
     110                  # self.startDelay = randint(10, 60) * 60
     111                  self.startDelay = randint(10, 60) * 1
     112                  self.log.info("Run as a daemon, waiting %s seconds to start." %
     113                                self.startDelay)
     114          else:
     115              self.log.debug("Run in foreground, starting immediately.")
     116
     117
     118 -    def reportError(self, error):
     119          """
     120          Log errors that have occurred
     121
     122          @param error: error message
     123          @type error: string
     124          """
     125          self.log.error("Error occured: %s", error)
     126
     127
     128 -    def connected(self):
     129          """
     130          Called after connected to the zenhub service
     131          """
     132          d = self.configure()
     133          d.addCallback(self.heartbeat)
     134          d.addErrback(self.reportError)
     135
     136
     137 -    def configure(self):
     138          """
     139          Get our configuration from zenhub
     140          """
     141          # add in the code to fetch cycle time, etc.
     142          def inner(driver):
     143              """
     144              Generator function to gather our configuration
     145
     146              @param driver: driver object
     147              @type driver: driver object
     148              """
     149              self.log.debug('fetching monitor properties')
     150              yield self.config().callRemote('propertyItems')
     151              items = dict(driver.next())
     152              if self.options.cycletime == 0:
     153                  self.modelerCycleInterval = items.get('modelerCycleInterval',
     154                                                        self.modelerCycleInterval)
     155              self.configCycleInterval = items.get('configCycleInterval',
     156                                                   self.configCycleInterval)
     157              reactor.callLater(self.configCycleInterval * 60, self.configure)
     158
     159              self.log.debug("Getting threshold classes...")
     160              yield self.config().callRemote('getThresholdClasses')
     161              self.remote_updateThresholdClasses(driver.next())
     162
     163              self.log.debug("Fetching default RRDCreateCommand...")
     164              yield self.config().callRemote('getDefaultRRDCreateCommand')
     165              createCommand = driver.next()
     166
     167              self.log.debug("Getting collector thresholds...")
     168              yield self.config().callRemote('getCollectorThresholds')
     169              self.rrdStats.config(self.options.monitor,
     170                                   self.name,
     171                                   driver.next(),
     172                                   createCommand)
     173
     174              self.log.debug("Getting collector plugins for each DeviceClass")
     175              yield self.config().callRemote('getClassCollectorPlugins')
     176              self.classCollectorPlugins = driver.next()
     177
     178          return drive(inner)
     179
     180
     181 -    def config(self):
     182          """
     183          Get the ModelerService
     184          """
     185          return self.services.get('ModelerService', FakeRemote())
     186
     187
     188 -    def selectPlugins(self, device, transport):
     189          """
     190          Build a list of active plugins for a device, based on:
     191
     192          * the --collect command-line option which is a regex
     193          * the --ignore command-line option which is a regex
     194          * transport which is a string describing the type of plugin
     195
     196          @param device: device to collect against
     197          @type device: string
     198          @param transport: python, ssh, snmp, telnet, cmd
     199          @type transport: string
     200          @return: results of the plugin
     201          @type: string
     202          @todo: determine if an event for the collector AND the device should be sent
     203          """
     204          plugins = []
     205          valid_loaders = []
     206          for loader in device.plugins:
     207              try:
     208                  plugin= loader.create()
     209                  self.log.debug( "Loaded plugin %s" % plugin.name() )
     210                  plugins.append( plugin )
     211                  valid_loaders.append( loader )
     212
     213              except (SystemExit, KeyboardInterrupt), ex:
     214                  self.log.info( "Interrupted by external signal (%s)" % str(ex) )
     215                  raise
     216
     217              except Plugins.PluginImportError, import_error:
     218                  import socket
     219                  component, _ = os.path.splitext( os.path.basename( sys.argv[0] ) )
     220                  collector_host= socket.gethostname()
     221                  # NB: an import errror affects all devices,
     222                  #     so report the issue against the collector
     223                  # TOOD: determine if an event for the collector AND the device should be sent
     224                  evt= { "eventClass":"/Status/Update", "component":component,
     225                        "agent":collector_host, "device":collector_host,
     226                        "severity":Error }
     227
     228                  info= "Problem loading plugin %s" % import_error.plugin
     229                  self.log.error( info )
     230                  evt[ 'summary' ]= info
     231
     232                  info= import_error.traceback
     233                  self.log.error( info )
     234                  evt[ 'message' ]= info
     235
     236                  info= ("Due to import errors, removing the %s plugin"
     237                         " from this collection cycle.") % import_error.plugin
     238                  self.log.error( info )
     239                  evt[ 'message' ] += "%s\n" % info
     240                  self.sendEvent( evt )
     241
     242          # Make sure that we don't generate messages for bad loaders again
     243          # NB: doesn't update the device's zProperties
     244          if len( device.plugins ) != len( valid_loaders ):
     245              device.plugins= valid_loaders
     246
     247          # Create functions to search for what plugins we will and
     248          # won't supply to the device
     249          collectTest = lambda x: False
     250          ignoreTest = lambda x: False
     251          if self.options.collectPlugins:
     252              collectTest = re.compile(self.options.collectPlugins).search
     253          elif self.options.ignorePlugins:
     254              ignoreTest = re.compile(self.options.ignorePlugins).search
     255
     256          result = []
     257          for plugin in plugins:
     258              if plugin.transport != transport:
     259                  continue
     260              name = plugin.name()
     261              if ignoreTest(name):
     262                  self.log.debug("Ignoring %s on %s because of --ignore flag",
     263                                 name, device.id)
     264              elif collectTest(name):
     265                  self.log.debug("Using %s on %s because of --collect flag",
     266                                 name, device.id)
     267                  result.append(plugin)
     268              elif not self.options.collectPlugins:
     269                  self.log.debug("Using %s on %s", name, device.id)
     270                  result.append(plugin)
     271          return result
     272
     273
     274
     275 -    def collectDevice(self, device):
     276          """
     277          Collect data from a single device.
     278
     279          @param device: device to collect against
     280          @type device: string
     281          """
     282          clientTimeout = getattr(device, 'zCollectorClientTimeout', 180)
     283          ip = device.manageIp
     284          timeout = clientTimeout + time.time()
     285          self.wmiCollect(device, ip, timeout)
     286          self.pythonCollect(device, ip, timeout)
     287          self.cmdCollect(device, ip, timeout)
     288          self.snmpCollect(device, ip, timeout)
     289          self.portscanCollect(device, ip, timeout)
     290
     291
     292
     293 -    def wmiCollect(self, device, ip, timeout):
     294          """
     295          Start the Windows Management Instrumentation (WMI) collector
     296
     297          @param device: device to collect against
     298          @type device: string
     299          @param ip: IP address of device to collect against
     300          @type ip: string
     301          @param timeout: timeout before failing the connection
     302          @type timeout: integer
     303          """
     304          if self.options.nowmi:
     305              return
     306
     307          client = None
     308          try:
     309              plugins = self.selectPlugins(device, 'wmi')
     310              if not plugins:
     311                  self.log.info("No WMI plugins found for %s" % device.id)
     312                  return
     313              if self.checkCollection(device):
     314                  self.log.info('WMI collector method for device %s' % device.id)
     315                  self.log.info("plugins: %s",
     316                          ", ".join(map(lambda p: p.name(), plugins)))
     317                  client = WMIClient(device, self, plugins)
     318              if not client or not plugins:
     319                  self.log.warn("WMI collector creation failed")
     320                  return
     321          except (SystemExit, KeyboardInterrupt):
     322              raise
     323          except Exception:
     324              self.log.exception("Error opening WMI collector")
     325          self.addClient(client, timeout, 'WMI', device.id)
     326
     327
     328
     329 -    def pythonCollect(self, device, ip, timeout):
     330          """
     331          Start local Python collection client.
     332
     333          @param device: device to collect against
     334          @type device: string
     335          @param ip: IP address of device to collect against
     336          @type ip: string
     337          @param timeout: timeout before failing the connection
     338          @type timeout: integer
     339          """
     340          client = None
     341          try:
     342              plugins = self.selectPlugins(device, "python")
     343              if not plugins:
     344                  self.log.info("No Python plugins found for %s" % device.id)
     345                  return
     346              if self.checkCollection(device):
     347                  self.log.info('Python collection device %s' % device.id)
     348                  self.log.info("plugins: %s",
     349                          ", ".join(map(lambda p: p.name(), plugins)))
     350                  client = PythonClient(device, self, plugins)
     351              if not client or not plugins:
     352                  self.log.warn("Python client creation failed")
     353                  return
     354          except (SystemExit, KeyboardInterrupt): raise
     355          except:
     356              self.log.exception("Error opening pythonclient")
     357          self.addClient(client, timeout, 'python', device.id)
     358
     359
     360 -    def cmdCollect(self, device, ip, timeout):
     361          """
     362          Start shell command collection client.
     363
     364          @param device: device to collect against
     365          @type device: string
     366          @param ip: IP address of device to collect against
     367          @type ip: string
     368          @param timeout: timeout before failing the connection
     369          @type timeout: integer
     370          """
     371          client = None
     372          clientType = 'snmp' # default to SNMP if we can't figure out a protocol
     373
     374          hostname = device.id
     375          try:
     376              plugins = self.selectPlugins(device,"command")
     377              if not plugins:
     378                  self.log.info("No command plugins found for %s" % hostname)
     379                  return
     380
     381              protocol = getattr(device, 'zCommandProtocol', defaultProtocol)
     382              commandPort = getattr(device, 'zCommandPort', defaultPort)
     383
     384              if protocol == "ssh":
     385                  client = SshClient(hostname, ip, commandPort,
     386                                     options=self.options,
     387                                     plugins=plugins, device=device,
     388                                     datacollector=self, isLoseConnection=True)
     389                  clientType = 'ssh'
     390                  self.log.info('Using SSH collection method for device %s'
     391                                % hostname)
     392
     393              elif protocol == 'telnet':
     394                  if commandPort == 22: commandPort = 23 #set default telnet
     395                  client = TelnetClient(hostname, ip, commandPort,
     396                                        options=self.options,
     397                                        plugins=plugins, device=device,
     398                                        datacollector=self)
     399                  clientType = 'telnet'
     400                  self.log.info('Using telnet collection method for device %s'
     401                                % hostname)
     402
     403              else:
     404                  info = ("Unknown protocol %s for device %s -- "
     405                          "defaulting to %s collection method" %
     406                          (protocol, hostname, clientType ))
     407                  self.log.warn( info )
     408                  import socket
     409                  component, _ = os.path.splitext( os.path.basename( sys.argv[0] ) )
     410                  collector_host= socket.gethostname()
     411                  evt= { "eventClass":"/Status/Update", "agent":collector_host,
     412                        "device":hostname, "severity":Error }
     413                  evt[ 'summary' ]= info
     414                  self.sendEvent( evt )
     415                  return
     416
     417              if not client:
     418                  self.log.warn("Shell command collector creation failed")
     419              else:
     420                  self.log.info("plugins: %s",
     421                      ", ".join(map(lambda p: p.name(), plugins)))
     422          except (SystemExit, KeyboardInterrupt): raise
     423          except:
     424              self.log.exception("Error opening command collector")
     425          self.addClient(client, timeout, clientType, device.id)
     426
     427
     428
     429 -    def snmpCollect(self, device, ip, timeout):
     430          """
     431          Start SNMP collection client.
     432
     433          @param device: device to collect against
     434          @type device: string
     435          @param ip: IP address of device to collect against
     436          @type ip: string
     437          @param timeout: timeout before failing the connection
     438          @type timeout: integer
     439          """
     440          client = None
     441          try:
     442              hostname = device.id
     443              if getattr( device, "zSnmpMonitorIgnore", True ):
     444                  self.log.info("SNMP monitoring off for %s" % hostname)
     445                  return
     446
     447              if not ip:
     448                  self.log.info("No manage IP for %s" % hostname)
     449                  return
     450
     451              plugins = []
     452              plugins = self.selectPlugins(device,"snmp")
     453              if not plugins:
     454                  self.log.info("No SNMP plugins found for %s" % hostname)
     455                  return
     456
     457              if self.checkCollection(device):
     458                  self.log.info('SNMP collection device %s' % hostname)
     459                  self.log.info("plugins: %s",
     460                                ", ".join(map(lambda p: p.name(), plugins)))
     461                  client = SnmpClient(device.id, ip, self.options,
     462                                      device, self, plugins)
     463              if not client or not plugins:
     464                  self.log.warn("SNMP collector creation failed")
     465                  return
     466          except (SystemExit, KeyboardInterrupt): raise
     467          except:
     468              self.log.exception("Error opening the SNMP collector")
     469          self.addClient(client, timeout, 'SNMP', device.id)
     470
     471
     472  ######## need to make async test for snmp work at some point -EAD #########
     473      # def checkSnmpConnection(self, device):
     474      #     """
     475      #     Check to see if our current community string is still valid
     476      #
     477      #     @param device: the device against which we will check
     478      #     @type device: a Device instance
     479      #     @return: result is None or a tuple containing
     480      #             (community, port, version, snmp name)
     481      #     @rtype: deferred: Twisted deferred
     482      #     """
     483      #     from pynetsnmp.twistedsnmp import AgentProxy
     484      #
     485      #     def inner(driver):
     486      #         self.log.debug("Checking SNMP community %s on %s",
     487      #                         device.zSnmpCommunity, device.id)
     488      #
     489      #         oid = ".1.3.6.1.2.1.1.5.0"
     490      #         proxy = AgentProxy(device.id,
     491      #                            device.zSnmpPort,
     492      #                            timeout=device.zSnmpTimeout,
     493      #                            community=device.zSnmpCommunity,
     494      #                            snmpVersion=device.zSnmpVer,
     495      #                            tries=2)
     496      #         proxy.open()
     497      #         yield proxy.get([oid])
     498      #         devname = driver.next().values()[0]
     499      #         if devname:
     500      #             yield succeed(True)
     501      #         yield succeed(False)
     502      #
     503      #     return drive(inner)
     504
     505
     506 -    def addClient(self, device, timeout, clientType, name):
     507          """
     508          If device is not None, schedule the device to be collected.
     509          Otherwise log an error.
     510
     511          @param device: device to collect against
     512          @type device: string
     513          @param timeout: timeout before failing the connection
     514          @type timeout: integer
     515          @param clientType: description of the plugin type
     516          @type clientType: string
     517          @param name: plugin name
     518          @type name: string
     519          """
     520          if device:
     521              device.timeout = timeout
     522              device.timedOut = False
     523              self.clients.append(device)
     524              device.run()
     525          else:
     526              self.log.warn('Unable to create a %s collector for %s',
     527                            clientType, name)
     528
     529
     530      # XXX double-check this, once the implementation is in place
     531 -    def portscanCollect(self, device, ip, timeout):
     532          """
     533          Start portscan collection client.
     534
     535          @param device: device to collect against
     536          @type device: string
     537          @param ip: IP address of device to collect against
     538          @type ip: string
     539          @param timeout: timeout before failing the connection
     540          @type timeout: integer
     541          """
     542          client = None
     543          try:
     544              hostname = device.id
     545              plugins = self.selectPlugins(device, "portscan")
     546              if not plugins:
     547                  self.log.info("No portscan plugins found for %s" % hostname)
     548                  return
     549              if self.checkCollection(device):
     550                  self.log.info('Portscan collector method for device %s'
     551                                % hostname)
     552                  self.log.info("plugins: %s",
     553                      ", ".join(map(lambda p: p.name(), plugins)))
     554                  client = PortscanClient(device.id, ip, self.options,
     555                                          device, self, plugins)
     556              if not client or not plugins:
     557                  self.log.warn("Portscan collector creation failed")
     558                  return
     559          except (SystemExit, KeyboardInterrupt): raise
     560          except:
     561              self.log.exception("Error opening portscan collector")
     562          self.addClient(client, timeout, 'portscan', device.id)
     563
     564
     565 -    def checkCollection(self, device):
     566          """
     567          See how old the data is that we've collected
     568
     569          @param device: device to collect against
     570          @type device: string
     571          @return: is the SNMP status number > 0 and is the last collection time + collage older than now?
     572          @type: boolean
     573          """
     574          age = device.getSnmpLastCollection() + self.collage
     575          if device.getSnmpStatusNumber() > 0 and age >= DateTime.DateTime():
     576              self.log.info("Skipped collection of %s" % device.id)
     577              return False
     578          return True
     579
     580
     581 -    def clientFinished(self, collectorClient):
     582          """
     583          Callback that processes the return values from a device.
     584          Python iterable.
     585
     586          @param collectorClient: collector instance
     587          @type collectorClient: collector class
     588          @return: Twisted deferred object
     589          @type: Twisted deferred object
     590          """
     591          device = collectorClient.device
     592          self.log.debug("Client for %s finished collecting", device.id)
     593          def processClient(driver):
     594              try:
     595                  pluginStats = {}
     596                  self.log.debug("Processing data for device %s", device.id)
     597                  devchanged = False
     598                  maps = []
     599                  for plugin, results in collectorClient.getResults():
     600                      if plugin is None: continue
     601                      self.log.debug("Processing plugin %s on device %s ...",
     602                                     plugin.name(), device.id)
     603                      if not results:
     604                          self.log.warn("The plugin %s returned no results.",
     605                                        plugin.name())
     606                          continue
     607
     608                      self.log.debug("Plugin %s results = %s", plugin.name(), results)
     609                      datamaps = []
     610                      try:
     611                          results = plugin.preprocess(results, self.log)
     612                          if results:
     613                              datamaps = plugin.process(device, results, self.log)
     614                          if datamaps:
     615                              pluginStats.setdefault(plugin.name(), plugin.weight)
     616
     617                      except (SystemExit, KeyboardInterrupt), ex:
     618                          self.log.info( "Plugin %s terminated due to external"
     619                                        " signal (%s)" % (plugin.name(), str(ex) )
     620                                        )
     621                          continue
     622
     623                      except Exception, ex:
     624                          # NB: don't discard the plugin, as it might be a
     625                          #     temporary issue
     626                          #     Also, report it against the device, rather than at
     627                          #     a collector as it might be just for this device.
     628                          import socket
     629                          component= os.path.splitext(
     630                              os.path.basename( sys.argv[0] )
     631                              )[0]
     632                          collector_host= socket.gethostname()
     633                          evt= { "eventClass":"/Status/Update",
     634                                "agent":collector_host, "device":device.id,
     635                                "severity":Error }
     636
     637                          info= "Problem while executing plugin %s" %plugin.name()
     638                          self.log.error( info )
     639                          evt[ 'summary' ]= info
     640
     641                          info= traceback.format_exc()
     642                          self.log.error( info )
     643                          evt[ 'message' ]= info
     644                          self.sendEvent( evt )
     645                          continue
     646
     647                      # allow multiple maps to be returned from one plugin
     648                      if type(datamaps) not in (types.ListType, types.TupleType):
     649                          datamaps = [datamaps,]
     650                      if datamaps:
     651                          maps += [m for m in datamaps if m]
     652                  if maps:
     653                      deviceClass = Classifier.classifyDevice(pluginStats,
     654                                                  self.classCollectorPlugins)
     655                      yield self.config().callRemote(
     656                                                  'applyDataMaps', device.id,
     657                                                  maps, deviceClass)
     658                      if driver.next():
     659                          devchanged = True
     660                  if devchanged:
     661                      self.log.info("Changes in configuration applied")
     662                  else:
     663                      self.log.info("No change in configuration detected")
     664                  yield self.config().callRemote('setSnmpLastCollection',
     665                                                 device.id)
     666                  driver.next()
     667              except Exception, ex:
     668                  self.log.exception(ex)
     669                  raise
     670
     671          def processClientFinished(result):
     672              """
     673              Called after the client collection finishes
     674
     675              @param result: object (unused)
     676              @type result: object
     677              """
     678              if not result:
     679                  self.log.debug("Client %s finished" % device.id)
     680              else:
     681                  self.log.error("Client %s finished with message: %s" %
     682                                 (device.id, result))
     683              try:
     684                  self.clients.remove(collectorClient)
     685                  self.finished.append(collectorClient)
     686              except ValueError:
     687                  self.log.debug("Client %s not found in in the list"
     688                                " of active clients",
     689                                device.id)
     690              d = drive(self.fillCollectionSlots)
     691              d.addErrback(self.fillError)
     692
     693          d = drive(processClient)
     694          d.addBoth(processClientFinished)
     695
     696
     697
     698 -    def fillError(self, reason):
     699          """
     700          Twisted errback routine to log an error when
     701          unable to collect some data
     702
     703          @param reason: error message
     704          @type reason: string
     705          """
     706          self.log.error("Unable to fill collection slots: %s" % reason)
     707
     708
     709 -    def cycleTime(self):
     710          """
     711          Return our cycle time (in minutes)
     712
     713          @return: cycle time
     714          @rtype: integer
     715          """
     716          return self.modelerCycleInterval * 60
     717
     718
     719 -    def heartbeat(self, ignored=None):
     720          """
     721          Twisted keep-alive mechanism to ensure that
     722          we're still connected to zenhub
     723
     724          @param ignored: object (unused)
     725          @type ignored: object
     726          """
     727          ARBITRARY_BEAT = 30
     728          reactor.callLater(ARBITRARY_BEAT, self.heartbeat)
     729          if self.options.cycle:
     730              evt = dict(eventClass=Heartbeat,
     731                         component='zenmodeler',
     732                         device=self.options.monitor,
     733                         timeout=3*ARBITRARY_BEAT)
     734              self.sendEvent(evt)
     735              self.niceDoggie(self.cycleTime())
     736
     737          # We start modeling from here to accomodate the startup delay.
     738          if not self.started:
     739              self.started = True
     740              reactor.callLater(self.startDelay, self.main)
     741
     742
     743 -    def checkStop(self, unused = None):
     744          """
     745          Check to see if there's anything to do.
     746          If there isn't, report our statistics and exit.
     747
     748          @param unused: unused (unused)
     749          @type unused: string
     750          """
     751          if self.clients: return
     752          if self.devicegen: return
     753
     754          if self.start:
     755              runTime = time.time() - self.start
     756              self.start = None
     757              self.log.info("Scan time: %0.2f seconds", runTime)
     758              devices = len(self.finished)
     759              timedOut = len([c for c in self.finished if c.timedOut])
     760              self.sendEvents(
     761                  self.rrdStats.gauge('cycleTime', self.cycleTime(), runTime) +
     762                  self.rrdStats.gauge('devices', self.cycleTime(), devices) +
     763                  self.rrdStats.gauge('timedOut', self.cycleTime(), timedOut)
     764                  )
     765              if not self.options.cycle:
     766                  self.stop()
     767              self.finished = []
     768
     769
     770 -    def fillCollectionSlots(self, driver):
     771          """
     772          An iterator which either returns a device to collect or
     773          calls checkStop()
     774
     775          @param driver: driver object
     776          @type driver: driver object
     777          """
     778          count = len(self.clients)
     779          while count < self.options.parallel and self.devicegen \
     780              and not self.pendingNewClients:
     781
     782              self.pendingNewClients = True
     783              try:
     784                  device = self.devicegen.next()
     785                  yield self.config().callRemote('getDeviceConfig', [device])
     786                  # just collect one device, and let the timer add more
     787                  devices = driver.next()
     788                  if devices:
     789                      self.collectDevice(devices[0])
     790              except StopIteration:
     791                  self.devicegen = None
     792
     793              self.pendingNewClients = False
     794              break
     795
     796          update = len(self.clients)
     797          if update != count and update != 1:
     798              self.log.info('Running %d clients', update)
     799          else:
     800              self.log.debug('Running %d clients', update)
     801          self.checkStop()
     802
     803
     804 -    def buildOptions(self):
     805          """
     806          Build our list of command-line options
     807          """
     808          PBDaemon.buildOptions(self)
     809          self.parser.add_option('--debug',
     810                  dest='debug', action="store_true", default=False,
     811                  help="Don't fork threads for processing")
     812          self.parser.add_option('--nowmi',
     813                  dest='nowmi', action="store_true", default=False,
     814                  help="Do not execute WMI plugins")
     815          self.parser.add_option('--parallel', dest='parallel',
     816                  type='int', default=defaultParallel,
     817                  help="Number of devices to collect from in parallel")
     818          self.parser.add_option('--cycletime',
     819                  dest='cycletime',default=720,type='int',
     820                  help="Run collection every x minutes")
     821          self.parser.add_option('--ignore',
     822                  dest='ignorePlugins',default="",
     823                  help="Modeler plugins to ignore. Takes a regular expression")
     824          self.parser.add_option('--collect',
     825                  dest='collectPlugins',default="",
     826                  help="Modeler plugins to use. Takes a regular expression")
     827          self.parser.add_option('-p', '--path', dest='path',
     828                  help="Start class path for collection ie /NetworkDevices")
     829          self.parser.add_option('-d', '--device', dest='device',
     830                  help="Fully qualified device name ie www.confmon.com")
     831          self.parser.add_option('-a', '--collage',
     832                  dest='collage', default=0, type='float',
     833                  help="Do not collect from devices whose collect date " +
     834                          "is within this many minutes")
     835          self.parser.add_option('--writetries',
     836                  dest='writetries',default=2,type='int',
     837                  help="Number of times to try to write if a "
     838                       "read conflict is found")
     839          # FIXME: cleanup --force option #2660
     840          self.parser.add_option("-F", "--force",
     841                      dest="force", action='store_true', default=True,
     842                      help="Force collection of config data (deprecated)")
     843          self.parser.add_option('--portscantimeout', dest='portscantimeout',
     844                  type='int', default=defaultPortScanTimeout,
     845                  help="Time to wait for connection failures when port scanning")
     846          self.parser.add_option('--now',
     847                  dest='now', action="store_true", default=False,
     848                  help="Start daemon now, do not sleep before starting")
     849          TCbuildOptions(self.parser, self.usage)
     850          addNTLMv2Option(self.parser)
     851
     852
     853
     854 -    def processOptions(self):
     855          """
     856          Check what the user gave us vs what we'll accept
     857          for command-line options
     858          """
     859          if not self.options.path and not self.options.device:
     860              self.options.path = "/Devices"
     861          if self.options.ignorePlugins and self.options.collectPlugins:
     862              raise SystemExit( "Only one of --ignore or --collect"
     863                               " can be used at a time")
     864          setNTLMv2Auth(self.options)
     865
     866 -    def _timeoutClients(self):
     867          """
     868          The guts of the timeoutClients method (minus the twisted reactor
     869          stuff). Breaking this part out as a separate method facilitates unit
     870          testing.
     871          """
     872          active = []
     873          for client in self.clients:
     874              if client.timeout < time.time():
     875                  self.log.warn("Client %s timeout", client.hostname)
     876                  self.finished.append(client)
     877                  client.timedOut = True
     878                  client.stop()
     879              else:
     880                  active.append(client)
     881          self.clients = active
     882
     883
     884
     885 -    def timeoutClients(self, unused=None):
     886          """
     887          Check to see which clients have timed out and which ones haven't.
     888          Stop processing anything that's timed out.
     889
     890          @param unused: unused (unused)
     891          @type unused: string
     892          """
     893          reactor.callLater(1, self.timeoutClients)
     894          self._timeoutClients()
     895          d = drive(self.fillCollectionSlots)
     896          d.addCallback(self.checkStop)
     897          d.addErrback(self.fillError)
     898
     899
     900
     901 -    def reactorLoop(self):
     902          """
     903          Twisted main loop
     904          """
     905          reactor.startRunning()
     906          while reactor.running:
     907              try:
     908                  while reactor.running:
     909                      reactor.runUntilCurrent()
     910                      timeout = reactor.timeout()
     911                      reactor.doIteration(timeout)
     912              except:
     913                  if reactor.running:
     914                      self.log.exception("Unexpected error in main loop.")
     915
     916
     917
     918 -    def getDeviceList(self):
     919          """
     920          Get the list of devices for which we are collecting:
     921          * if -d devicename was used, use the devicename
     922          * if a class path flag was supplied, gather the devices
     923            along that organizer
     924          * otherwise get all of the devices associated with our collector
     925
     926          @return: list of devices
     927          @rtype: list
     928          """
     929          if self.options.device:
     930              self.log.info("Collecting for device %s", self.options.device)
     931              return succeed([self.options.device])
     932
     933          self.log.info("Collecting for path %s", self.options.path)
     934          return self.config().callRemote('getDeviceListByOrganizer',
     935                                          self.options.path,
     936                                          self.options.monitor)
     937
     938
     939 -    def mainLoop(self, driver):
     940          """
     941          Main collection loop, a Python iterable
     942
     943          @param driver: driver object
     944          @type driver: driver object
     945          @return: Twisted deferred object
     946          @rtype: Twisted deferred object
     947          """
     948          if self.options.cycle:
     949              driveLater(self.cycleTime(), self.mainLoop)
     950
     951          if self.clients:
     952              self.log.error("Modeling cycle taking too long")
     953              return
     954
     955          self.start = time.time()
     956
     957          self.log.debug("Starting collector loop...")
     958          yield self.getDeviceList()
     959          self.devicegen = iter(driver.next())
     960          d = drive(self.fillCollectionSlots)
     961          d.addErrback(self.fillError)
     962          yield d
     963          driver.next()
     964          self.log.debug("Collection slots filled")
     965
     966
     967
     968 -    def main(self, unused=None):
     969          """
     970          Wrapper around the mainLoop
     971
     972          @param unused: unused (unused)
     973          @type unused: string
     974          @return: Twisted deferred object
     975          @rtype: Twisted deferred object
     976          """
     977          self.finished = []
     978          d = drive(self.mainLoop)
     979          d.addCallback(self.timeoutClients)
     980          return d
     981
     982
     983
     984 -    def remote_deleteDevice(self, device):
     985          """
     986          Stub function
     987
     988          @param device: device name (unused)
     989          @type device: string
     990          @todo: implement
     991          """
     992          # we fetch the device list before every scan
     993          self.log.debug("Asynch deleteDevice %s" % device)
     994
     995
     996  if __name__ == '__main__':
     997      dc = ZenModeler()
     998      dc.processOptions()
     999      reactor.run = dc.reactorLoop
    1000      dc.run()
    1001