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                      datamaps = []
     609                      try:
     610                          results = plugin.preprocess(results, self.log)
     611                          if results:
     612                              datamaps = plugin.process(device, results, self.log)
     613                          if datamaps:
     614                              pluginStats.setdefault(plugin.name(), plugin.weight)
     615
     616                      except (SystemExit, KeyboardInterrupt), ex:
     617                          self.log.info( "Plugin %s terminated due to external"
     618                                        " signal (%s)" % (plugin.name(), str(ex) )
     619                                        )
     620                          continue
     621
     622                      except Exception, ex:
     623                          # NB: don't discard the plugin, as it might be a
     624                          #     temporary issue
     625                          #     Also, report it against the device, rather than at
     626                          #     a collector as it might be just for this device.
     627                          import socket
     628                          component= os.path.splitext(
     629                              os.path.basename( sys.argv[0] )
     630                              )[0]
     631                          collector_host= socket.gethostname()
     632                          evt= { "eventClass":"/Status/Update",
     633                                "agent":collector_host, "device":device.id,
     634                                "severity":Error }
     635
     636                          info= "Problem while executing plugin %s" %plugin.name()
     637                          self.log.error( info )
     638                          evt[ 'summary' ]= info
     639
     640                          info= traceback.format_exc()
     641                          self.log.error( info )
     642                          evt[ 'message' ]= info
     643                          self.sendEvent( evt )
     644                          continue
     645
     646                      # allow multiple maps to be returned from one plugin
     647                      if type(datamaps) not in (types.ListType, types.TupleType):
     648                          datamaps = [datamaps,]
     649                      if datamaps:
     650                          maps += [m for m in datamaps if m]
     651                  if maps:
     652                      deviceClass = Classifier.classifyDevice(pluginStats,
     653                                                  self.classCollectorPlugins)
     654                      yield self.config().callRemote(
     655                                                  'applyDataMaps', device.id,
     656                                                  maps, deviceClass)
     657                      if driver.next():
     658                          devchanged = True
     659                  if devchanged:
     660                      self.log.info("Changes in configuration applied")
     661                  else:
     662                      self.log.info("No change in configuration detected")
     663                  yield self.config().callRemote('setSnmpLastCollection',
     664                                                 device.id)
     665                  driver.next()
     666              except Exception, ex:
     667                  self.log.exception(ex)
     668                  raise
     669
     670          def processClientFinished(result):
     671              """
     672              Called after the client collection finishes
     673
     674              @param result: object (unused)
     675              @type result: object
     676              """
     677              if not result:
     678                  self.log.debug("Client %s finished" % device.id)
     679              else:
     680                  self.log.error("Client %s finished with message: %s" %
     681                                 (device.id, result))
     682              try:
     683                  self.clients.remove(collectorClient)
     684                  self.finished.append(collectorClient)
     685              except ValueError:
     686                  self.log.debug("Client %s not found in in the list"
     687                                " of active clients",
     688                                device.id)
     689              d = drive(self.fillCollectionSlots)
     690              d.addErrback(self.fillError)
     691
     692          d = drive(processClient)
     693          d.addBoth(processClientFinished)
     694
     695
     696
     697 -    def fillError(self, reason):
     698          """
     699          Twisted errback routine to log an error when
     700          unable to collect some data
     701
     702          @param reason: error message
     703          @type reason: string
     704          """
     705          self.log.error("Unable to fill collection slots: %s" % reason)
     706
     707
     708 -    def cycleTime(self):
     709          """
     710          Return our cycle time (in minutes)
     711
     712          @return: cycle time
     713          @rtype: integer
     714          """
     715          return self.modelerCycleInterval * 60
     716
     717
     718 -    def heartbeat(self, ignored=None):
     719          """
     720          Twisted keep-alive mechanism to ensure that
     721          we're still connected to zenhub
     722
     723          @param ignored: object (unused)
     724          @type ignored: object
     725          """
     726          ARBITRARY_BEAT = 30
     727          reactor.callLater(ARBITRARY_BEAT, self.heartbeat)
     728          if self.options.cycle:
     729              evt = dict(eventClass=Heartbeat,
     730                         component='zenmodeler',
     731                         device=self.options.monitor,
     732                         timeout=3*ARBITRARY_BEAT)
     733              self.sendEvent(evt)
     734              self.niceDoggie(self.cycleTime())
     735
     736          # We start modeling from here to accomodate the startup delay.
     737          if not self.started:
     738              self.started = True
     739              reactor.callLater(self.startDelay, self.main)
     740
     741
     742 -    def checkStop(self, unused = None):
     743          """
     744          Check to see if there's anything to do.
     745          If there isn't, report our statistics and exit.
     746
     747          @param unused: unused (unused)
     748          @type unused: string
     749          """
     750          if self.clients: return
     751          if self.devicegen: return
     752
     753          if self.start:
     754              runTime = time.time() - self.start
     755              self.start = None
     756              self.log.info("Scan time: %0.2f seconds", runTime)
     757              devices = len(self.finished)
     758              timedOut = len([c for c in self.finished if c.timedOut])
     759              self.sendEvents(
     760                  self.rrdStats.gauge('cycleTime', self.cycleTime(), runTime) +
     761                  self.rrdStats.gauge('devices', self.cycleTime(), devices) +
     762                  self.rrdStats.gauge('timedOut', self.cycleTime(), timedOut)
     763                  )
     764              if not self.options.cycle:
     765                  self.stop()
     766              self.finished = []
     767
     768
     769 -    def fillCollectionSlots(self, driver):
     770          """
     771          An iterator which either returns a device to collect or
     772          calls checkStop()
     773
     774          @param driver: driver object
     775          @type driver: driver object
     776          """
     777          count = len(self.clients)
     778          while count < self.options.parallel and self.devicegen \
     779              and not self.pendingNewClients:
     780
     781              self.pendingNewClients = True
     782              try:
     783                  device = self.devicegen.next()
     784                  yield self.config().callRemote('getDeviceConfig', [device])
     785                  # just collect one device, and let the timer add more
     786                  devices = driver.next()
     787                  if devices:
     788                      self.collectDevice(devices[0])
     789              except StopIteration:
     790                  self.devicegen = None
     791
     792              self.pendingNewClients = False
     793              break
     794
     795          update = len(self.clients)
     796          if update != count and update != 1:
     797              self.log.info('Running %d clients', update)
     798          else:
     799              self.log.debug('Running %d clients', update)
     800          self.checkStop()
     801
     802
     803 -    def buildOptions(self):
     804          """
     805          Build our list of command-line options
     806          """
     807          PBDaemon.buildOptions(self)
     808          self.parser.add_option('--debug',
     809                  dest='debug', action="store_true", default=False,
     810                  help="Don't fork threads for processing")
     811          self.parser.add_option('--nowmi',
     812                  dest='nowmi', action="store_true", default=False,
     813                  help="Do not execute WMI plugins")
     814          self.parser.add_option('--parallel', dest='parallel',
     815                  type='int', default=defaultParallel,
     816                  help="Number of devices to collect from in parallel")
     817          self.parser.add_option('--cycletime',
     818                  dest='cycletime',default=720,type='int',
     819                  help="Run collection every x minutes")
     820          self.parser.add_option('--ignore',
     821                  dest='ignorePlugins',default="",
     822                  help="Modeler plugins to ignore. Takes a regular expression")
     823          self.parser.add_option('--collect',
     824                  dest='collectPlugins',default="",
     825                  help="Modeler plugins to use. Takes a regular expression")
     826          self.parser.add_option('-p', '--path', dest='path',
     827                  help="Start class path for collection ie /NetworkDevices")
     828          self.parser.add_option('-d', '--device', dest='device',
     829                  help="Fully qualified device name ie www.confmon.com")
     830          self.parser.add_option('-a', '--collage',
     831                  dest='collage', default=0, type='float',
     832                  help="Do not collect from devices whose collect date " +
     833                          "is within this many minutes")
     834          self.parser.add_option('--writetries',
     835                  dest='writetries',default=2,type='int',
     836                  help="Number of times to try to write if a "
     837                       "read conflict is found")
     838          # FIXME: cleanup --force option #2660
     839          self.parser.add_option("-F", "--force",
     840                      dest="force", action='store_true', default=True,
     841                      help="Force collection of config data (deprecated)")
     842          self.parser.add_option('--portscantimeout', dest='portscantimeout',
     843                  type='int', default=defaultPortScanTimeout,
     844                  help="Time to wait for connection failures when port scanning")
     845          self.parser.add_option('--now',
     846                  dest='now', action="store_true", default=False,
     847                  help="Start daemon now, do not sleep before starting")
     848          TCbuildOptions(self.parser, self.usage)
     849          addNTLMv2Option(self.parser)
     850
     851
     852
     853 -    def processOptions(self):
     854          """
     855          Check what the user gave us vs what we'll accept
     856          for command-line options
     857          """
     858          if not self.options.path and not self.options.device:
     859              self.options.path = "/Devices"
     860          if self.options.ignorePlugins and self.options.collectPlugins:
     861              raise SystemExit( "Only one of --ignore or --collect"
     862                               " can be used at a time")
     863          setNTLMv2Auth(self.options)
     864
     865 -    def _timeoutClients(self):
     866          """
     867          The guts of the timeoutClients method (minus the twisted reactor
     868          stuff). Breaking this part out as a separate method facilitates unit
     869          testing.
     870          """
     871          active = []
     872          for client in self.clients:
     873              if client.timeout < time.time():
     874                  self.log.warn("Client %s timeout", client.hostname)
     875                  self.finished.append(client)
     876                  client.timedOut = True
     877                  client.stop()
     878              else:
     879                  active.append(client)
     880          self.clients = active
     881
     882
     883
     884 -    def timeoutClients(self, unused=None):
     885          """
     886          Check to see which clients have timed out and which ones haven't.
     887          Stop processing anything that's timed out.
     888
     889          @param unused: unused (unused)
     890          @type unused: string
     891          """
     892          reactor.callLater(1, self.timeoutClients)
     893          self._timeoutClients()
     894          d = drive(self.fillCollectionSlots)
     895          d.addCallback(self.checkStop)
     896          d.addErrback(self.fillError)
     897
     898
     899
     900 -    def reactorLoop(self):
     901          """
     902          Twisted main loop
     903          """
     904          reactor.startRunning()
     905          while reactor.running:
     906              try:
     907                  while reactor.running:
     908                      reactor.runUntilCurrent()
     909                      timeout = reactor.timeout()
     910                      reactor.doIteration(timeout)
     911              except:
     912                  if reactor.running:
     913                      self.log.exception("Unexpected error in main loop.")
     914
     915
     916
     917 -    def getDeviceList(self):
     918          """
     919          Get the list of devices for which we are collecting:
     920          * if -d devicename was used, use the devicename
     921          * if a class path flag was supplied, gather the devices
     922            along that organizer
     923          * otherwise get all of the devices associated with our collector
     924
     925          @return: list of devices
     926          @rtype: list
     927          """
     928          if self.options.device:
     929              self.log.info("Collecting for device %s", self.options.device)
     930              return succeed([self.options.device])
     931
     932          self.log.info("Collecting for path %s", self.options.path)
     933          return self.config().callRemote('getDeviceListByOrganizer',
     934                                          self.options.path,
     935                                          self.options.monitor)
     936
     937
     938 -    def mainLoop(self, driver):
     939          """
     940          Main collection loop, a Python iterable
     941
     942          @param driver: driver object
     943          @type driver: driver object
     944          @return: Twisted deferred object
     945          @rtype: Twisted deferred object
     946          """
     947          if self.options.cycle:
     948              driveLater(self.cycleTime(), self.mainLoop)
     949
     950          if self.clients:
     951              self.log.error("Modeling cycle taking too long")
     952              return
     953
     954          self.start = time.time()
     955
     956          self.log.debug("Starting collector loop...")
     957          yield self.getDeviceList()
     958          self.devicegen = iter(driver.next())
     959          d = drive(self.fillCollectionSlots)
     960          d.addErrback(self.fillError)
     961          yield d
     962          driver.next()
     963          self.log.debug("Collection slots filled")
     964
     965
     966
     967 -    def main(self, unused=None):
     968          """
     969          Wrapper around the mainLoop
     970
     971          @param unused: unused (unused)
     972          @type unused: string
     973          @return: Twisted deferred object
     974          @rtype: Twisted deferred object
     975          """
     976          self.finished = []
     977          d = drive(self.mainLoop)
     978          d.addCallback(self.timeoutClients)
     979          return d
     980
     981
     982
     983 -    def remote_deleteDevice(self, device):
     984          """
     985          Stub function
     986
     987          @param device: device name (unused)
     988          @type device: string
     989          @todo: implement
     990          """
     991          # we fetch the device list before every scan
     992          self.log.debug("Asynch deleteDevice %s" % device)
     993
     994
     995  if __name__ == '__main__':
     996      dc = ZenModeler()
     997      dc.processOptions()
     998      reactor.run = dc.reactorLoop
     999      dc.run()
    1000