broadcasteventprocessor.java

来自「opennms得相关源码 请大家看看」· Java 代码 · 共 1,678 行 · 第 1/5 页

//
// This file is part of the OpenNMS(R) Application.
//
// OpenNMS(R) is Copyright (C) 2002-2005 The OpenNMS Group, Inc.  All rights reserved.
// OpenNMS(R) is a derivative work, containing both original code, included code and modified
// code that was published under the GNU General Public License. Copyrights for modified 
// and included code are below.
//
// OpenNMS(R) is a registered trademark of The OpenNMS Group, Inc.
//
// Modifications:
//
// 2003 Nov 11: Merged changes from Rackspace project
// 2003 Jan 31: Cleaned up some unused imports.
// 2004 Sep 08: Completely reorganize to clean up the delete code.
//
// Original code base Copyright (C) 1999-2001 Oculan Corp.  All rights reserved.
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.                                                            
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//       
// For more information contact: 
//      OpenNMS Licensing       <[e-mail]>
//      http://www.opennms.org/
//      http://www.opennms.com/
//
// Tab Size = 8
//

package org.opennms.netmgt.capsd;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Category;
import org.opennms.core.queue.FifoQueue;
import org.opennms.core.utils.ThreadCategory;
import org.opennms.netmgt.EventConstants;
import org.opennms.netmgt.config.CapsdConfigFactory;
import org.opennms.netmgt.config.DatabaseConnectionFactory;
import org.opennms.netmgt.config.OpennmsServerConfigFactory;
import org.opennms.netmgt.eventd.EventIpcManagerFactory;
import org.opennms.netmgt.eventd.EventListener;
import org.opennms.netmgt.utils.XmlrpcUtil;
import org.opennms.netmgt.xml.event.Event;

/**
 * @author <a href="mailto:[e-mail]">Matt Brozowski </a>
 * @author <a href="http://www.opennms.org/">OpenNMS </a>
 */
final class BroadcastEventProcessor implements EventListener {

    /**
     * SQL statement used to add an interface/server mapping into the database;
     */
    private static String SQL_ADD_INTERFACE_TO_SERVER = "INSERT INTO serverMap VALUES (?, ?)";

    /**
     * SQL statement used to add an interface/service mapping into the database.
     */
    private static String SQL_ADD_SERVICE_TO_MAPPING = "INSERT INTO serviceMap VALUES (?, ?)";

    /**
     * SQL statement used to delete all services mapping to a specified
     * interface from the database.
     */
    private static String SQL_DELETE_ALL_SERVICES_INTERFACE_MAPPING = "DELETE FROM serviceMap WHERE ipaddr = ?";

    /**
     * SQL statement used to delete an interface/server mapping from the
     * database.
     */
    private static String SQL_DELETE_INTERFACE_ON_SERVER = "DELETE FROM serverMap WHERE ipaddr = ? AND servername = ?";

    /**
     * SQL statement used to delete an interface/service mapping from the
     * database.
     */
    private static String SQL_DELETE_SERVICE_INTERFACE_MAPPING = "DELETE FROM serviceMap WHERE ipaddr = ? AND servicemapname = ?";

    /**
     * SQL statement used to verify if an ipinterface with the specified ip
     * address exists in the database and retrieve the nodeid if exists.
     */
    private static String SQL_QUERY_IPADDRESS_EXIST = "SELECT nodeid FROM ipinterface WHERE ipaddr = ? AND isManaged !='D'";

    /**
     * SQL statement used to query the 'node' and 'ipinterface' tables to verify
     * if a specified ipaddr and node label have already exist in the database.
     */
    private static String SQL_QUERY_IPINTERFACE_EXIST = "SELECT nodelabel, ipaddr FROM node, ipinterface WHERE node.nodeid = ipinterface.nodeid AND node.nodelabel = ? AND ipinterface.ipaddr = ? AND isManaged !='D' AND nodeType !='D'";

    /**
     * SQL statement used to query if a node with the specified nodelabel exist
     * in the database, and the nodeid from the database if exists.
     */
    private static String SQL_QUERY_NODE_EXIST = "SELECT nodeid, dpname FROM node WHERE nodelabel = ? AND nodeType !='D'";

    /**
     * SQL statement used to verify if an ifservice with the specified ip
     * address and service name exists in the database.
     */
    private static String SQL_QUERY_SERVICE_EXIST = "SELECT nodeid FROM ifservices, service WHERE ifservices.serviceid = service.serviceid AND ipaddr = ? AND servicename = ? AND status !='D'";

    /**
     * SQL statement used to query if an interface/service mapping already
     * exists in the database.
     */
    private static String SQL_QUERY_SERVICE_MAPPING_EXIST = "SELECT * FROM serviceMap WHERE ipaddr = ? AND servicemapname = ?";

    /**
     * SQL query to retrieve nodeid of a particulary interface address
     */
    private static String SQL_RETRIEVE_NODEID = "select nodeid from ipinterface where ipaddr=? and isManaged!='D'";

    /**
     * SQL statement used to retrieve the serviced id from the database with a
     * specified service name.
     */
    private static String SQL_RETRIEVE_SERVICE_ID = "SELECT serviceid FROM service WHERE servicename = ?";

    /**
     * Determines if deletePropagation is enabled in the Outage Manager.
     * 
     * @return true if deletePropagation is enable, false otherwise
     */
    public static boolean isPropagationEnabled() {
        return CapsdConfigFactory.getInstance().getDeletePropagationEnabled();
    }

    /**
     * FIXME: finish the doc here
     * 
     * @return Returns the xmlrpc.
     */
    public static boolean isXmlRpcEnabled() {
        return CapsdConfigFactory.getInstance().getXmlrpc().equals("true");
    }

    /**
     * local openNMS server name
     */
    private String m_localServer = null;

    /**
     * Set of event ueis that we should notify when we receive and when a
     * success or failure occurs.
     */
    private Set m_notifySet = new HashSet();

    /**
     * The Capsd rescan scheduler
     */
    private Scheduler m_scheduler;

    /**
     * The location where suspectInterface events are enqueued for processing.
     */
    private FifoQueue m_suspectQ;

    /**
     * Constructor
     * 
     * @param suspectQ
     *            The queue where new SuspectEventProcessor objects are enqueued
     *            for running..
     * @param scheduler
     *            Rescan scheduler.
     */
    BroadcastEventProcessor(FifoQueue suspectQ, Scheduler scheduler) {
        // Suspect queue
        //
        m_suspectQ = suspectQ;

        // Scheduler
        //
        m_scheduler = scheduler;

        // the local servername
        m_localServer = OpennmsServerConfigFactory.getInstance().getServerName();

        // Subscribe to eventd
        //
        createMessageSelectorAndSubscribe();
    }

    /**
     * Unsubscribe from eventd
     */
    public void close() {
        EventIpcManagerFactory.getInstance().getManager().removeEventListener(this);
    }

    /**
     * Counts the number of interfaces on the node other than a given interface
     * 
     * @param dbConn
     *            the database connection
     * @param nodeid
     *            the node to check interfaces for
     * @param ipAddr
     *            the interface not to include in the count
     * @return the numer of interfaces other than the given one
     * @throws SQLException
     *             if an error occurs talking to the database
     */
    private int countOtherInterfacesOnNode(Connection dbConn, long nodeId, String ipAddr) throws SQLException {
        Category log = ThreadCategory.getInstance(getClass());

        final String DB_COUNT_OTHER_INTERFACES_ON_NODE = "SELECT count(*) FROM ipinterface WHERE nodeID=? and ipAddr != ? and isManaged != 'D'";

        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            stmt = dbConn.prepareStatement(DB_COUNT_OTHER_INTERFACES_ON_NODE);
            stmt.setLong(1, nodeId);
            stmt.setString(2, ipAddr);
            rs = stmt.executeQuery();
            int count = 0;
            while (rs.next()) {
                count = rs.getInt(1);
            }

            if (log.isDebugEnabled())
                log.debug("countServicesForInterface: count services for interface " + nodeId + "/" + ipAddr + ": found " + count);

            return count;
        } finally {
            try {
                if (rs != null) rs.close();
            } finally {
                if (stmt != null) stmt.close();
            }
        }        
    }

    /**
     * Counts the number of other non deleted services associated with the
     * interface defined by nodeid/ipAddr
     * 
     * @param dbConn
     *            the database connection
     * @param nodeId
     *            the node to chck
     * @param ipAddr
     *            the interface to check
     * @param service
     *            the name of the service not to include
     * @return the number of non deleted services, other than serviceId
     */
    private int countOtherServicesOnInterface(Connection dbConn, long nodeId, String ipAddr, String service) throws SQLException {

        Category log = ThreadCategory.getInstance(getClass());

        final String DB_COUNT_OTHER_SERVICES_ON_IFACE = "SELECT count(*) FROM ifservices, service " + "WHERE ifservices.serviceId = service.serviceId AND ifservices.status != 'D' " + "AND ifservices.nodeID=? AND ifservices.ipAddr=? AND service.servicename != ?";

        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            stmt = dbConn.prepareStatement(DB_COUNT_OTHER_SERVICES_ON_IFACE);
            stmt.setLong(1, nodeId);
            stmt.setString(2, ipAddr);
            stmt.setString(3, service);
            rs = stmt.executeQuery();
            int count = 0;
            while (rs.next()) {
                count = rs.getInt(1);
            }

            if (log.isDebugEnabled())
                log.debug("countServicesForInterface: count services for interface " + nodeId + "/" + ipAddr + ": found " + count);

            return count;
        } finally {
            try {
                if (rs != null) rs.close();
            } finally {
                if (stmt != null) stmt.close();
            }
        }        
    }

    /**
     * Counts the number of non deleted services on a node on interfaces other
     * than a given interface
     * 
     * @param dbConn
     *            the database connection
     * @param nodeId
     *            the nodeid to check
     * @param ipAddr
     *            the address of the interface not to include
     * @return the number of non deleted services on other interfaces
     */
    private int countServicesOnOtherInterfaces(Connection dbConn, long nodeId, String ipAddr) throws SQLException {
        Category log = ThreadCategory.getInstance(getClass());

        final String DB_COUNT_SERVICES_ON_OTHER_INTERFACES = "SELECT count(*) FROM ifservices WHERE nodeID=? and ipAddr != ? and status != 'D'";

        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            stmt = dbConn.prepareStatement(DB_COUNT_SERVICES_ON_OTHER_INTERFACES);
            stmt.setLong(1, nodeId);
            stmt.setString(2, ipAddr);
            rs = stmt.executeQuery();

            int count = 0;
            while (rs.next()) {
                count = rs.getInt(1);
            }

            if (log.isDebugEnabled())