Package Products :: Package ZenEvents :: Module zensyslog |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenEvents.zensyslog

      1  #! /usr/bin/env python
      2  ##############################################################################
      3  #
      4  # Copyright (C) Zenoss, Inc. 2008, 2011, all rights reserved.
      5  #
      6  # This content is made available according to terms specified in
      7  # License.zenoss under the directory where your Zenoss product is installed.
      8  #
      9  ##############################################################################
     10
     11
     12  __doc__ = """zensyslog
     13
     14  Turn syslog messages into events.
     15
     16  """
     17
     18  import time
     19  import socket
     20  import os
     21  import logging
     22
     23  from twisted.internet.protocol import DatagramProtocol
     24  from twisted.internet import reactor, defer, udp
     25  from twisted.python import failure
     26
     27  import Globals
     28  import zope.interface
     29  import zope.component
     30
     31
     32  from Products.ZenCollector.daemon import CollectorDaemon
     33  from Products.ZenCollector.interfaces import ICollector, ICollectorPreferences,\
     34                                               IEventService, \
     35                                               IScheduledTask
     36  from Products.ZenCollector.tasks import SimpleTaskFactory,\
     37                                          SimpleTaskSplitter,\
     38                                          BaseTask, TaskStates
     39  from Products.ZenUtils.observable import ObservableMixin
     40
     41  from Products.ZenEvents.SyslogProcessing import SyslogProcessor
     42
     43  from Products.ZenUtils.Utils import zenPath
     44  from Products.ZenUtils.IpUtil import asyncNameLookup
     45
     46  from Products.ZenEvents.EventServer import Stats
     47  from Products.ZenUtils.Utils import unused
     48  from Products.ZenCollector.services.config import DeviceProxy
     49  unused(Globals, DeviceProxy)
     50
     51  COLLECTOR_NAME = 'zensyslog'
     52  log = logging.getLogger("zen.%s" % COLLECTOR_NAME)
     53
     54
     55 -class SyslogPreferences(object):
     56      zope.interface.implements(ICollectorPreferences)
     57
     58 -    def __init__(self):
     59          """
     60          Constructs a new PingCollectionPreferences instance and
     61          provides default values for needed attributes.
     62          """
     63          self.collectorName = COLLECTOR_NAME
     64          self.defaultRRDCreateCommand = None
     65          self.configCycleInterval = 20 # minutes
     66          self.cycleInterval = 5 * 60 # seconds
     67
     68          # The configurationService attribute is the fully qualified class-name
     69          # of our configuration service that runs within ZenHub
     70          self.configurationService = 'Products.ZenHub.services.SyslogConfig'
     71
     72          # Will be filled in based on buildOptions
     73          self.options = None
     74
     75          self.configCycleInterval = 20*60
     76
     77 -    def postStartupTasks(self):
     78          task = SyslogTask(COLLECTOR_NAME, configId=COLLECTOR_NAME)
     79          yield task
     80
     81 -    def buildOptions(self, parser):
     82          """
     83          Command-line options to be supported
     84          """
     85          SYSLOG_PORT = 514
     86          try:
     87              SYSLOG_PORT = socket.getservbyname('syslog', 'udp')
     88          except socket.error:
     89              pass
     90
     91          parser.add_option('--parsehost', dest='parsehost',
     92                             action='store_true', default=False,
     93                             help='Try to parse the hostname part of a syslog HEADER'
     94                             )
     95          parser.add_option('--stats', dest='stats',
     96                             action='store_true', default=False,
     97                             help='Print statistics to log every 2 secs')
     98          parser.add_option('--logorig', dest='logorig',
     99                             action='store_true', default=False,
    100                             help='Log the original message')
    101          parser.add_option('--logformat', dest='logformat',
    102                             default='human',
    103                             help='Human-readable (/var/log/messages) or raw (wire)'
    104                             )
    105          parser.add_option('--minpriority', dest='minpriority',
    106                             default=6, type='int',
    107                             help='Minimum priority message that zensyslog will accept'
    108                              )
    109          parser.add_option('--syslogport', dest='syslogport',
    110                             default=SYSLOG_PORT, type='int',
    111                             help='Port number to use for syslog events'
    112                              )
    113          parser.add_option('--listenip', dest='listenip',
    114                             default='0.0.0.0',
    115                             help='IP address to listen on. Default is %default'
    116                              )
    117          parser.add_option('--useFileDescriptor',
    118                              dest='useFileDescriptor', type='int',
    119                             help='Read from an existing connection rather opening a new port.'
    120                             , default=None)
    121          parser.add_option('--noreverseLookup', dest='noreverseLookup',
    122                             action='store_true', default=False,
    123                             help="Don't convert the remote device's IP address to a hostname."
    124                             )
    125
    126 -    def postStartup(self):
    127          daemon = zope.component.getUtility(ICollector)
    128          daemon.defaultPriority = 1
    129
    130
    131 -class SyslogTask(BaseTask, DatagramProtocol):
    132      """
    133      Listen for syslog messages and turn them into events
    134      Connects to the TrapService service in zenhub.
    135      """
    136      zope.interface.implements(IScheduledTask)
    137
    138      SYSLOG_DATE_FORMAT = '%b %d %H:%M:%S'
    139      SAMPLE_DATE = 'Apr 10 15:19:22'
    140
    141 -    def __init__(self, taskName, configId,
    142                   scheduleIntervalSeconds=3600, taskConfig=None):
    143          BaseTask.__init__(self, taskName, configId,
    144                   scheduleIntervalSeconds, taskConfig)
    145          self.log = log
    146
    147          # Needed for interface
    148          self.name = taskName
    149          self.configId = configId
    150          self.state = TaskStates.STATE_IDLE
    151          self.interval = scheduleIntervalSeconds
    152          self._preferences = taskConfig
    153          self._daemon = zope.component.getUtility(ICollector)
    154          self._eventService = zope.component.queryUtility(IEventService)
    155          self._preferences = self._daemon
    156
    157          self.options = self._daemon.options
    158
    159          self.stats = Stats()
    160
    161          if not self.options.useFileDescriptor\
    162               and self.options.syslogport < 1024:
    163              self._daemon.openPrivilegedPort('--listen', '--proto=udp',
    164                                      '--port=%s:%d'
    165                                       % (self.options.listenip,
    166                                      self.options.syslogport))
    167          self._daemon.changeUser()
    168          self.minpriority = self.options.minpriority
    169          self.processor = None
    170
    171          if self.options.logorig:
    172              self.olog = logging.getLogger('origsyslog')
    173              self.olog.setLevel(20)
    174              self.olog.propagate = False
    175              lname = zenPath('log/origsyslog.log')
    176              hdlr = logging.FileHandler(lname)
    177              hdlr.setFormatter(logging.Formatter('%(message)s'))
    178              self.olog.addHandler(hdlr)
    179
    180          if self.options.useFileDescriptor is not None:
    181              self.useUdpFileDescriptor(int(self.options.useFileDescriptor))
    182          else:
    183              reactor.listenUDP(self.options.syslogport, self,
    184                                interface=self.options.listenip)
    185
    186          #   yield self.model().callRemote('getDefaultPriority')
    187          self.processor = SyslogProcessor(self._eventService.sendEvent,
    188                      self.options.minpriority, self.options.parsehost,
    189                      self.options.monitor, self._daemon.defaultPriority)
    190
    191 -    def doTask(self):
    192          """
    193          This is a wait-around task since we really are called
    194          asynchronously.
    195          """
    196          return defer.succeed("Waiting for syslog messages...")
    197
    198 -    def useUdpFileDescriptor(self, fd):
    199          s = socket.fromfd(fd, socket.AF_INET, socket.SOCK_DGRAM)
    200          os.close(fd)
    201          port = s.getsockname()[1]
    202          transport = udp.Port(port, self)
    203          s.setblocking(0)
    204          transport.socket = s
    205          transport.fileno = s.fileno
    206          transport.connected = 1
    207          transport._realPortNumber = port
    208          self.transport = transport
    209          # hack around startListening not being called
    210          self.numPorts = 1
    211          transport.startReading()
    212
    213 -    def expand(self, msg, client_address):
    214          """
    215          Expands a syslog message into a string format suitable for writing
    216          to the filesystem such that it appears the same as it would
    217          had the message been logged by the syslog daemon.
    218
    219          @param msg: syslog message
    220          @type msg: string
    221          @param client_address: IP info of the remote device (ipaddr, port)
    222          @type client_address: tuple of (string, number)
    223          @return: message
    224          @rtype: string
    225          """
    226          # pri := facility * severity
    227          stop = msg.find('>')
    228
    229          # check for a datestamp.  default to right now if date not present
    230          start = stop + 1
    231          stop = start + len(SyslogTask.SAMPLE_DATE)
    232          dateField = msg[start:stop]
    233          try:
    234              date = time.strptime(dateField,
    235                                   SyslogTask.SYSLOG_DATE_FORMAT)
    236              year = time.localtime()[0]
    237              date = (year, ) + date[1:]
    238              start = stop + 1
    239          except ValueError:
    240
    241          # date not present, so use today's date
    242              date = time.localtime()
    243
    244          # check for a hostname.  default to localhost if not present
    245          stop = msg.find(' ', start)
    246          if msg[stop - 1] == ':':
    247              hostname = client_address[0]
    248          else:
    249              hostname = msg[start:stop]
    250              start = stop + 1
    251
    252          # the message content
    253          body = msg[start:]
    254
    255          # assemble the message
    256          prettyTime = time.strftime(SyslogTask.SYSLOG_DATE_FORMAT, date)
    257          message = '%s %s %s' % (prettyTime, hostname, body)
    258          return message
    259
    260 -    def datagramReceived(self, msg, client_address):
    261          """
    262          Consume the network packet
    263
    264          @param msg: syslog message
    265          @type msg: string
    266          @param client_address: IP info of the remote device (ipaddr, port)
    267          @type client_address: tuple of (string, number)
    268          """
    269          (ipaddr, port) = client_address
    270          if self.options.logorig:
    271              if self.options.logformat == 'human':
    272                  message = self.expand(msg, client_address)
    273              else:
    274                  message = msg
    275              self.olog.info(message)
    276
    277          if self.options.noreverseLookup:
    278              d = defer.succeed(ipaddr)
    279          else:
    280              d = asyncNameLookup(ipaddr)
    281          d.addBoth(self.gotHostname, (msg, ipaddr, time.time()))
    282
    283 -    def gotHostname(self, response, data):
    284          """
    285          Send the resolved address, if possible, and the event via the thread
    286
    287          @param response: Twisted response
    288          @type response: Twisted response
    289          @param data: (msg, ipaddr, rtime)
    290          @type data: tuple of (string, string, datetime object)
    291          """
    292          (msg, ipaddr, rtime) = data
    293          if isinstance(response, failure.Failure):
    294              host = ipaddr
    295          else:
    296              host = response
    297          if self.processor:
    298              self.processor.process(msg, ipaddr, host, rtime)
    299
    300 -    def displayStatistics(self):
    301          totalTime, totalEvents, maxTime = self.stats.report()
    302          display = "%d events processed in %.2f seconds" % (
    303                        totalEvents,
    304                        totalTime)
    305          if totalEvents > 0:
    306              display += """
    307  %.5f average seconds per event
    308  Maximum processing time for one event was %.5f""" % (
    309                         (totalTime / totalEvents), maxTime)
    310          return display
    311
    312 -    def cleanup(self):
    313          status = self.displayStatistics()
    314          self.log.info(status)
    315
    316
    317 -class SyslogConfigTask(ObservableMixin):
    318      """
    319      Receive a configuration object containing the default priority
    320      """
    321      zope.interface.implements(IScheduledTask)
    322
    323 -    def __init__(self, taskName, configId,
    324                   scheduleIntervalSeconds=3600, taskConfig=None):
    325          super(SyslogConfigTask, self).__init__()
    326
    327          # Needed for ZCA interface contract
    328          self.name = taskName
    329          self.configId = configId
    330          self.state = TaskStates.STATE_IDLE
    331          self.interval = scheduleIntervalSeconds
    332          self._preferences = taskConfig
    333          self._daemon = zope.component.getUtility(ICollector)
    334
    335          self._daemon.defaultPriority = self._preferences.defaultPriority
    336
    337 -    def doTask(self):
    338          return defer.succeed("Already updated default syslog priority...")
    339
    340 -    def cleanup(self):
    341          pass
    342
    343
    344 -class SyslogDaemon(CollectorDaemon):
    345
    346      _frameworkFactoryName = "nosip"
    347
    348
    349  if __name__=='__main__':
    350      myPreferences = SyslogPreferences()
    351      myTaskFactory = SimpleTaskFactory(SyslogConfigTask)
    352      myTaskSplitter = SimpleTaskSplitter(myTaskFactory)
    353      daemon = SyslogDaemon(myPreferences, myTaskSplitter)
    354      daemon.run()
    355