scheduler.java

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

//
// This file is part of the OpenNMS(R) Application.
//
// OpenNMS(R) is Copyright (C) 2002-2003 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 Jan 31: Cleaned up some unused imports.
//
// 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/
//

package org.opennms.netmgt.capsd;

import java.lang.reflect.UndeclaredThrowableException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.Category;
import org.opennms.core.fiber.PausableFiber;
import org.opennms.core.queue.FifoQueue;
import org.opennms.core.queue.FifoQueueException;
import org.opennms.core.utils.ThreadCategory;
import org.opennms.netmgt.config.CapsdConfigFactory;
import org.opennms.netmgt.config.DatabaseConnectionFactory;

/**
 * This class implements a simple scheduler to ensure that Capsd rescans occurs
 * at the expected intervals.
 * 
 * @author <a href="mailto:[e-mail]">Mike Davidson </a>
 * @author <a href="http://www.opennms.org/">OpenNMS </a>
 * 
 */
final class Scheduler implements Runnable, PausableFiber {
    /**
     * The prefix for the fiber name.
     */
    private static final String FIBER_NAME = "Capsd Scheduler";

    /**
     * SQL used to retrieve list of nodes from the node table.
     */
    private static final String SQL_RETRIEVE_NODES = "SELECT nodeid FROM node WHERE nodetype != 'D'";

    /**
     * SQL used to retrieve the last poll time for all the managed interfaces
     * belonging to a particular node.
     */
    private static final String SQL_GET_LAST_POLL_TIME = "SELECT iplastcapsdpoll FROM ipinterface WHERE nodeid=? AND (ismanaged = 'M' OR ismanaged = 'N')";

    /**
     * Special identifier used in place of a valid node id in order to schedule
     * an SMB reparenting using the rescan scheduler.
     */
    private static final int SMB_REPARENTING_IDENTIFIER = -1;

    /**
     * The name of this fiber.
     */
    private String m_name;

    /**
     * The status for this fiber.
     */
    private int m_status;

    /**
     * The worker thread that executes this instance.
     */
    private Thread m_worker;

    /**
     * List of NodeInfo objects representing each of the nodes in the database
     * capability of being scheduled.
     */
    private List m_knownNodes;

    /**
     * The configured interval (in milliseconds) between rescans
     */
    private long m_interval;

    /**
     * The configured initial sleep (in milliseconds) prior to scheduling
     * rescans
     */
    private long m_initialSleep;

    /**
     * The rescan queue where new RescanProcessor objects are enqueued for
     * execution.
     */
    private FifoQueue m_rescanQ;

    /**
     * This class encapsulates the information about a node necessary to
     * schedule the node for rescans.
     */
    final class NodeInfo {
        int m_nodeId;

        Timestamp m_lastScanned;

        long m_interval;

        boolean m_scheduled;

        NodeInfo(int nodeId, Timestamp lastScanned, long interval) {
            m_nodeId = nodeId;
            m_lastScanned = lastScanned;
            m_interval = interval;
            m_scheduled = false;
        }

        NodeInfo(int nodeId, Date lastScanned, long interval) {
            m_nodeId = nodeId;
            m_lastScanned = new Timestamp(lastScanned.getTime());
            m_interval = interval;
            m_scheduled = false;
        }

        boolean isScheduled() {
            return m_scheduled;
        }

        int getNodeId() {
            return m_nodeId;
        }

        Timestamp getLastScanned() {
            return m_lastScanned;
        }

        long getRescanInterval() {
            return m_interval;
        }

        void setScheduled(boolean scheduled) {
            m_scheduled = scheduled;
        }

        void setLastScanned(Date lastScanned) {
            m_lastScanned = new Timestamp(lastScanned.getTime());
        }

        void setLastScanned(Timestamp lastScanned) {
            m_lastScanned = lastScanned;
        }

        boolean timeForRescan() {
            if (System.currentTimeMillis() >= (m_lastScanned.getTime() + m_interval))
                return true;
            else
                return false;
        }
    }

    /**
     * Constructs a new instance of the scheduler.
     * 
     */
    Scheduler(FifoQueue rescanQ) throws SQLException {
        m_rescanQ = rescanQ;

        m_name = FIBER_NAME;
        m_status = START_PENDING;
        m_worker = null;

        m_knownNodes = Collections.synchronizedList(new LinkedList());
        Category log = ThreadCategory.getInstance(getClass());

        // Get rescan interval from configuration factory
        //
        m_interval = CapsdConfigFactory.getInstance().getRescanFrequency();
        if (log.isDebugEnabled())
            log.debug("Scheduler: rescan interval(millis): " + m_interval);

        // Get initial rescan sleep time from configuration factory
        //
        m_initialSleep = CapsdConfigFactory.getInstance().getInitialSleepTime();
        if (log.isDebugEnabled())
            log.debug("Scheduler: initial rescan sleep time(millis): " + m_initialSleep);

        // Schedule SMB Reparenting using special nodeId (-1)
        //
        // Schedule this node in such a way that it will be
        // scheduled immediately and SMB reparenting will take place
        Date lastSmbReparenting = new Date();
        lastSmbReparenting.setTime(System.currentTimeMillis() - m_interval);

        if (log.isDebugEnabled())
            log.debug("Scheduler: scheduling SMB reparenting...");
        NodeInfo smbInfo = new NodeInfo(SMB_REPARENTING_IDENTIFIER, lastSmbReparenting, m_interval);
        m_knownNodes.add(smbInfo);

        // Load actual known nodes from the database
        //
        loadKnownNodes();
        if (log.isDebugEnabled())
            log.debug("Scheduler: done loading known nodes, node count: " + m_knownNodes.size());
    }

    /**
     * Builds a list of NodeInfo objects representing each of the nodes in the
     * database capable of being scheduled for rescan.
     * 
     * @throws SQLException
     *             if there is a problem accessing the database.
     */
    private void loadKnownNodes() throws SQLException {
        Category log = ThreadCategory.getInstance(getClass());

        Connection db = null;

        PreparedStatement nodeStmt = null;
        PreparedStatement ifStmt = null;
        try {
            db = DatabaseConnectionFactory.getInstance().getConnection();
            // Prepare SQL statements in advance
            //
            nodeStmt = db.prepareStatement(SQL_RETRIEVE_NODES);
            ifStmt = db.prepareStatement(SQL_GET_LAST_POLL_TIME);

            // Retrieve non-deleted nodes from the node table in the database
            //
            ResultSet rs = nodeStmt.executeQuery();

            while (rs.next()) {
                // Retrieve an interface from the ipInterface table in
                // the database for its last polled/scanned time

                int nodeId = rs.getInt(1);
                ifStmt.setInt(1, nodeId); // set nodeid
                if (log.isDebugEnabled())
                    log.debug("loadKnownNodes: retrieved nodeid " + nodeId + ", now getting last poll time.");

                ResultSet rset = ifStmt.executeQuery();
                if (rset.next()) {
                    Timestamp lastPolled = rset.getTimestamp(1);
                    if (lastPolled != null && rset.wasNull() == false) {
                        if (log.isDebugEnabled())
                            log.debug("loadKnownNodes: adding node " + nodeId + " with last poll time " + lastPolled);
                        NodeInfo nodeInfo = new NodeInfo(nodeId, lastPolled, m_interval);
                        m_knownNodes.add(nodeInfo);
                    }
                } else {
                    if (log.isDebugEnabled())
                        log.debug("Node w/ nodeid " + nodeId + " has no managed interfaces from which to retrieve a last poll time...it will not be scheduled.");
                }
            }
        } finally {
            try {
                if (nodeStmt != null)
                    nodeStmt.close();
            } catch (Exception e) {
            }

            try {
                if (ifStmt != null)
                    ifStmt.close();
            } catch (Exception e) {
            }

            try {
                if (db != null)
                    db.close();
            } catch (Exception e) {
            }
        }

    }

    /**
     * Creates a NodeInfo object representing the specified node and adds it to
     * the known node list for scheduling.
     * 
     * @param nodeId
     *            Id of node to be scheduled
     * 
     * @throws SQLException
     *             if there is any problem accessing the database
     */
    void scheduleNode(int nodeId) throws SQLException {
        Category log = ThreadCategory.getInstance(getClass());

        // Retrieve last poll time for the node from the ipInterface
        // table.
        Connection db = null;
        try {