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

# Source Code for Module Products.ZenEvents.zensyslog

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