capsd.java

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

        try {
            SnmpPeerFactory.reload();
        } catch (MarshalException ex) {
            log.error("Failed to load SNMP configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (ValidationException ex) {
            log.error("Failed to load SNMP configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (IOException ex) {
            log.error("Failed to load SNMP configuration", ex);
            throw new UndeclaredThrowableException(ex);
        }

        // Get connection to the database and use it to sync the
        // content of the database with the latest configuration
        // information.
        // 
        // First any new services are added to the services table
        // with a call to syncServices().
        //
        // Secondly the management state of interfaces and services
        // in the database is updated based on the latest configuration
        // information with a call to syncManagementState()
        //
        // Lastly the primary snmp interface state ('isSnmpPrimary')
        // of all interfaces which support SNMP is updated based on
        // the latest configuration information via a call to
        // syncSnmpPrimaryState()
        java.sql.Connection conn = null;
        try {
            conn = DatabaseConnectionFactory.getInstance().getConnection();
            if (log.isDebugEnabled()) {
                log.debug("init: Loading services into database...");
            }
            CapsdConfigFactory.getInstance().syncServices(conn);
            if (log.isDebugEnabled()) {
                log.debug("init: Syncing management state...");
            }
            CapsdConfigFactory.getInstance().syncManagementState(conn);
            if (log.isDebugEnabled()) {
                log.debug("init: Syncing primary SNMP interface state...");
            }
            CapsdConfigFactory.getInstance().syncSnmpPrimaryState(conn);
        } catch (SQLException sqlE) {
            log.fatal("SQL Exception while syncing database with latest configuration information.", sqlE);
            throw new UndeclaredThrowableException(sqlE);
        } catch (Throwable t) {
            log.fatal("Unknown error while syncing database with latest configuration information.", t);
            throw new UndeclaredThrowableException(t);
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (Exception e) {
            }
        }

        // Create the suspect event and rescan thread pools
        //
        m_suspectRunner = new RunnableConsumerThreadPool("Capsd Suspect Pool", 0.0f, 0.0f, CapsdConfigFactory.getInstance().getMaxSuspectThreadPoolSize());

        m_rescanRunner = new RunnableConsumerThreadPool("Capsd Rescan Pool", 0.0f, // Only
                                                                                    // stop
                                                                                    // thread
                                                                                    // if
                                                                                    // nothing
                                                                                    // in
                                                                                    // queue
                0.0f, // Always start thread if queue is not
                // empty and max threads has not been reached.
                CapsdConfigFactory.getInstance().getMaxRescanThreadPoolSize());

        // Create the rescan scheduler
        //
        if (log.isDebugEnabled()) {
            log.debug("init: Creating rescan scheduler");
        }
        try {
            // During instantiation, the scheduler will load the
            // list of known nodes from the database
            m_scheduler = new Scheduler(m_rescanRunner.getRunQueue());
        } catch (SQLException sqlE) {
            log.error("Failed to initialize the rescan scheduler.", sqlE);
            throw new UndeclaredThrowableException(sqlE);
        } catch (Throwable t) {
            log.error("Failed to initialize the rescan scheduler.", t);
            throw new UndeclaredThrowableException(t);
        }

        // Create an event receiver.
        //
        try {
            if (log.isDebugEnabled()) {
                log.debug("init: Creating event broadcast event receiver");
            }

            m_receiver = new BroadcastEventProcessor(m_suspectRunner.getRunQueue(), m_scheduler);
        } catch (Throwable t) {
            log.error("Failed to initialized the broadcast event receiver", t);
            throw new UndeclaredThrowableException(t);
        }
    }

    /**
     * Start the Capsd threads.
     */
    public void start() {
        ThreadCategory.setPrefix(LOG4J_CATEGORY);
        Category log = ThreadCategory.getInstance();

        m_status = STARTING;

        // Start the suspect event and rescan thread pools
        //
        if (log.isDebugEnabled()) {
            log.debug("start: Starting runnable thread pools...");
        }
        m_suspectRunner.start();
        m_rescanRunner.start();

        // Start the rescan scheduler
        //
        if (log.isDebugEnabled()) {
            log.debug("start: Starting rescan scheduler");
        }
        m_scheduler.start();

        m_status = RUNNING;
    }

    public void pause() {
        if (m_status != RUNNING)
            return;

        m_status = PAUSE_PENDING;

        Category log = ThreadCategory.getInstance();

        // TBD - Pause all threads

        m_status = PAUSED;

        if (log.isDebugEnabled()) {
            log.debug("pause: Finished pausing all threads");
        }
    }

    public void resume() {
        if (m_status != PAUSED) {
            return;
        }

        m_status = RESUME_PENDING;

        Category log = ThreadCategory.getInstance();

        // TBD - Resume all threads

        m_status = RUNNING;

        if (log.isDebugEnabled()) {
            log.debug("pause: Finished resuming all threads");
        }
    }

    /**
     * Returns a name/id for this process
     */
    public String getName() {
        return "OpenNMS.Capsd";
    }

    /**
     * Returns the current status
     */
    public synchronized int getStatus() {
        return m_status;
    }

    /**
     * Used to retrieve the local host name/address. The name/address of the
     * machine on which Capsd is running.
     */
    public static String getLocalHostAddress() {
        return m_address;
    }

    public static Capsd getInstance() {
        return m_singleton;
    }

    static Object getDbSyncLock() {
        return m_dbSyncLock;
    }

    /**
     * This method is used by other managed beans to forward an IP Address for
     * capability scanning. The If the interface converts properly then it is
     * scanned as a suspect interface for the discovery of all the services and
     * other interfaces that exists on the node.
     * 
     * @param ifAddr
     *            The address of the suspect interface.
     * 
     * @throws java.net.UnknownHostException
     *             Thrown if the address cannot be converted to aa proper
     *             internet address.
     */
    public void scanSuspectInterface(String ifAddr) throws UnknownHostException {
        String prefix = ThreadCategory.getPrefix();
        try {
            ThreadCategory.setPrefix(LOG4J_CATEGORY);
            InetAddress addr = InetAddress.getByName(ifAddr);
            SuspectEventProcessor proc = new SuspectEventProcessor(addr.getHostAddress());
            proc.run();
        } finally {
            ThreadCategory.setPrefix(prefix);
        }
    }

    /**
     * This method is used to force an existing node to be capability rescaned.
     * The main reason for its existance is as a hook for JMX managed beans to
     * invoke forced rescans allowing the main rescan logic to remain in the
     * capsd agent.
     * 
     * @param nodeId
     *            The node identifier from the database.
     */
    public void rescanInterfaceParent(Integer nodeId) {
        String prefix = ThreadCategory.getPrefix();
        try {
            ThreadCategory.setPrefix(LOG4J_CATEGORY);
            m_scheduler.forceRescan(nodeId.intValue());
        } finally {
            ThreadCategory.setPrefix(prefix);
        }
    }

} // end Capsd class