AgentIndexRunner.java revision 1463
825N/A/*
825N/A * CDDL HEADER START
825N/A *
825N/A * The contents of this file are subject to the terms of the
859N/A * Common Development and Distribution License (the "License").
825N/A * You may not use this file except in compliance with the License.
825N/A *
825N/A * See LICENSE.txt included in this distribution for the specific
860N/A * language governing permissions and limitations under the License.
860N/A *
825N/A * When distributing Covered Code, include this CDDL HEADER in each
825N/A * file and include the License file at LICENSE.txt.
825N/A * If applicable, add the following below this CDDL HEADER, with the
825N/A * fields enclosed by brackets "[]" replaced with your own identifying
825N/A * information: Portions Copyright [yyyy] [name of copyright owner]
825N/A *
825N/A * CDDL HEADER END
825N/A */
825N/A
825N/A/*
825N/A * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
825N/A */
825N/Apackage org.opensolaris.opengrok.management;
825N/A
825N/Aimport java.io.File;
825N/Aimport java.util.Arrays;
857N/Aimport java.util.HashSet;
825N/Aimport java.util.Iterator;
825N/Aimport java.util.List;
825N/Aimport java.util.Set;
825N/Aimport java.util.logging.Level;
825N/Aimport java.util.logging.Logger;
825N/Aimport javax.management.ListenerNotFoundException;
825N/Aimport javax.management.MBeanNotificationInfo;
857N/Aimport javax.management.MBeanRegistration;
825N/Aimport javax.management.MBeanServer;
825N/Aimport javax.management.Notification;
825N/Aimport javax.management.NotificationEmitter;
825N/Aimport javax.management.NotificationFilter;
825N/Aimport javax.management.NotificationListener;
825N/Aimport javax.management.ObjectName;
825N/Aimport org.opensolaris.opengrok.configuration.RuntimeEnvironment;
825N/Aimport org.opensolaris.opengrok.history.HistoryGuru;
825N/Aimport org.opensolaris.opengrok.index.IndexChangedListener;
825N/Aimport org.opensolaris.opengrok.index.Indexer;
825N/A
825N/A/**
825N/A * AgentIndexRunner.
825N/A * @author Jan S Berg
825N/A */
825N/Apublic final class AgentIndexRunner implements AgentIndexRunnerMBean, NotificationListener,
825N/A MBeanRegistration, Runnable, IndexChangedListener, NotificationEmitter {
825N/A
825N/A private static final Logger log = Logger.getLogger(AgentIndexRunner.class.getName());
825N/A private transient static AgentIndexRunner indexerInstance = null;
825N/A private static final String NOTIFICATIONACTIONTYPE = "ogaaction";
825N/A private static final String NOTIFICATIONEXCEPTIONTYPE = "ogaexception";
825N/A private static final String NOTIFICATIONINFOSTRINGTYPE = "ogainfostring";
825N/A private static final String NOTIFICATIONINFOLONGTYPE = "ogainfolong";
825N/A private boolean enabled;
825N/A private transient Thread indexThread = null;
825N/A private RuntimeEnvironment env = null;
825N/A private long lastIndexStart = 0;
825N/A private long lastIndexFinish = 0;
825N/A private long lastIndexUsedTime = 0;
825N/A private Exception lastException = null;
825N/A private final Set<NotificationHolder> notifListeners =
825N/A new HashSet<NotificationHolder>();
825N/A private static long sequenceNo = 0;
825N/A private final StringBuilder notifications = new StringBuilder();
825N/A private static final int MAXMESSAGELENGTH = 50000;
825N/A
825N/A /**
825N/A * The only constructor is private, so other classes will only get an
825N/A * instance through the static factory method getInstance().
825N/A */
825N/A private AgentIndexRunner(boolean enabledParam) {
825N/A enabled = enabledParam;
825N/A }
825N/A
825N/A /**
825N/A * Static factory method to get an instance of AgentIndexRunner.
825N/A * @param enabledParam If {@code true}, the initial instance should be
825N/A * running or not.
825N/A * @return a singleton
825N/A */
825N/A @SuppressWarnings("PMD.AvoidSynchronizedAtMethodLevel")
825N/A public static synchronized AgentIndexRunner getInstance(boolean enabledParam) {
825N/A if (indexerInstance == null) {
825N/A indexerInstance = new AgentIndexRunner(enabledParam);
825N/A }
825N/A return indexerInstance;
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public ObjectName preRegister(MBeanServer serverParam, ObjectName name) {
825N/A return name;
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void postRegister(Boolean registrationDone) {
825N/A // not used
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void preDeregister() {
825N/A // not used
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void postDeregister() {
825N/A // not used
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void run() {
825N/A try {
825N/A //Indexer ind = new Indexer();
825N/A log.info("Running...");
825N/A lastIndexStart = System.currentTimeMillis();
825N/A lastException = null;
825N/A doNotify(NOTIFICATIONINFOLONGTYPE, "StartIndexing", Long.valueOf(lastIndexStart));
825N/A String configfile = Management.getInstance().getConfigurationFile();
825N/A if (configfile == null) {
825N/A doNotify(NOTIFICATIONEXCEPTIONTYPE, "Missing Configuration file", "");
825N/A }
825N/A File cfgFile = new File(configfile);
825N/A if (cfgFile.exists()) {
825N/A env = RuntimeEnvironment.getInstance();
825N/A log.log(Level.INFO, "Running indexer with configuration ''{0}''", configfile);
825N/A env.readConfiguration(cfgFile);
825N/A
825N/A int noThreads = Management.getInstance().getNumberOfThreads().intValue();
825N/A boolean update = Management.getInstance().getUpdateIndexDatabase().booleanValue();
825N/A String[] sublist = Management.getInstance().getSubFiles();
825N/A log.info("Update source repositories");
825N/A HistoryGuru.getInstance().updateRepositories();
825N/A List<String> subFiles = Arrays.asList(sublist);
825N/A log.log(Level.INFO, "Starting index, update {0} noThreads {1} subfiles {2}", new Object[]{String.valueOf(update), String.valueOf(noThreads), String.valueOf(subFiles.size())});
825N/A Indexer.doIndexerExecution(update, noThreads, subFiles, this);
825N/A log.info("Finished indexing");
825N/A lastIndexFinish = System.currentTimeMillis();
825N/A sendNotifications();
825N/A doNotify(NOTIFICATIONINFOLONGTYPE, "FinishedIndexing", Long.valueOf(lastIndexFinish));
825N/A lastIndexUsedTime = lastIndexFinish - lastIndexStart;
825N/A String publishhost = Management.getInstance().getPublishServerURL();
825N/A if ((publishhost == null) || (publishhost.equals(""))) {
825N/A log.warning("No publishhost given, not sending updates");
825N/A } else {
825N/A Indexer.sendToConfigHost(publishhost);
825N/A doNotify(NOTIFICATIONINFOSTRINGTYPE, "Published index", publishhost);
825N/A }
825N/A
825N/A
825N/A } else {
825N/A log.log(Level.WARNING, "Cannot Run indexing without proper configuration file ''{0}''", configfile);
825N/A doNotify(NOTIFICATIONEXCEPTIONTYPE, "Configuration file not valid", configfile);
825N/A }
825N/A } catch (Exception e) {
825N/A log.warning("Exception running indexing: " + e.getMessage());
825N/A log.log(Level.FINE, "run", e);
825N/A lastException = e;
825N/A }
825N/A }
825N/A
825N/A /**
825N/A * Disables indexer
825N/A */
825N/A @Override
825N/A public void disable() {
825N/A enabled = false;
825N/A }
825N/A
825N/A /**
825N/A * Enables the indexer
825N/A */
825N/A @Override
825N/A public void enable() {
825N/A enabled = true;
825N/A }
825N/A
825N/A /**
825N/A * Handle timer notifications to the purgatory.
825N/A * Will start the purger if it is enabled and return immediately.
825N/A */
825N/A @Override
825N/A public void handleNotification(Notification n, Object hb) {
825N/A if (n.getType().equals("timer.notification")) {
825N/A log.finer("Received timer notification");
825N/A if (enabled) {
825N/A index(false);
825N/A } else {
825N/A log.info("Indexing is disabled, doing nothing");
825N/A }
825N/A } else {
825N/A log.log(Level.WARNING, "Received unknown notification type ''{0}''", n.getType());
825N/A }
825N/A }
825N/A
825N/A /**
825N/A * The index method starts a thread that will
825N/A * start indexing part of the opengrok agent.
825N/A * @param waitForFinished if false the command returns immediately, if true
825N/A * it will return when the indexing is done.
825N/A */
825N/A @Override
825N/A public void index(boolean waitForFinished) {
825N/A log.info("Starting indexing.");
825N/A /*
825N/A * Synchronize here to make sure that you never get more than one
825N/A * indexing thread trying to start at the same time.
825N/A */
825N/A synchronized (this) {
825N/A if (indexThread != null) {
825N/A if (indexThread.isAlive()) {
825N/A log.warning("Previous indexer is still alive, will not start another.");
825N/A return;
825N/A }
825N/A log.fine("Previous indexer is no longer alive, starting a new one.");
825N/A }
825N/A indexThread = new Thread(this);
825N/A try {
825N/A indexThread.start();
825N/A if (!waitForFinished) {
825N/A return;
825N/A }
825N/A log.fine("Waiting for indexer to finish ...");
825N/A indexThread.join();
825N/A log.fine("indexer finished.");
825N/A } catch (Exception e) {
825N/A log.warning("Caught Exception while waiting for indexing to finish: "
825N/A + e.getMessage());
825N/A log.log(Level.FINE, "index", e);
825N/A }
825N/A return;
825N/A }
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void fileAdd(String path, String analyzer) {
825N/A log.log(Level.FINE, "Add ''{0}'' analyzer {1}", new Object[]{path, analyzer});
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void fileRemove(String path) {
825N/A log.log(Level.FINE, "File remove ''{0}''", path);
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void fileUpdate(String path) {
825N/A log.log(Level.FINE, "File updated ''{0}''", path);
825N/A addFileAction("U:", path);
825N/A }
825N/A
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void fileAdded(String path, String analyzer) {
825N/A log.log(Level.FINE, "Added ''{0}'' analyzer {1}", new Object[]{path, analyzer});
825N/A addFileAction("A:", path);
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public void fileRemoved(String path) {
825N/A log.log(Level.FINE, "File removed ''{0}''", path);
825N/A addFileAction("R:", path);
825N/A }
825N/A
825N/A private void addFileAction(String type, String path) {
825N/A notifications.append('\n');
825N/A notifications.append(type);
825N/A notifications.append(path);
825N/A if (notifications.length() > MAXMESSAGELENGTH) {
825N/A sendNotifications();
825N/A }
825N/A }
825N/A
825N/A private void sendNotifications() {
825N/A if (notifications.length() > 0) {
825N/A doNotify(NOTIFICATIONACTIONTYPE, "FilesInfo", notifications.toString());
825N/A notifications.delete(0, notifications.length());
825N/A }
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public long lastIndexTimeFinished() {
825N/A return lastIndexFinish;
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public long lastIndexTimeStarted() {
825N/A return lastIndexStart;
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public long lastIndexTimeUsed() {
825N/A return lastIndexUsedTime;
825N/A }
825N/A
825N/A /**
825N/A * {@inheritDoc}
825N/A */
825N/A @Override
825N/A public Exception getExceptions() {
return lastException;
}
/**
* {@inheritDoc}
*/
@Override
public void addNotificationListener(NotificationListener notiflistener,
NotificationFilter notfilt, Object obj) throws IllegalArgumentException
{
log.log(Level.CONFIG, "Adds a notify listener, with obj {0}", obj.toString());
if (notiflistener == null) {
throw new IllegalArgumentException("Must have legal NotificationListener");
}
synchronized (notifListeners) {
notifListeners.add(new NotificationHolder(notiflistener, notfilt, obj));
}
}
/**
* {@inheritDoc}
*/
@Override
public void removeNotificationListener(NotificationListener notiflistener)
throws ListenerNotFoundException
{
log.info("removes a notify listener, no obj");
boolean removed = false;
synchronized (notifListeners) {
Iterator<NotificationHolder> it = notifListeners.iterator();
while (it.hasNext()) {
NotificationHolder mnf = it.next();
if (mnf.getNL().equals(notiflistener)) {
it.remove();
removed = true;
}
}
}
if (!removed) {
throw new ListenerNotFoundException("Didn't remove the given NotificationListener");
}
}
/**
* {@inheritDoc}
*/
@Override
public void removeNotificationListener(NotificationListener notiflistener,
NotificationFilter filt, Object obj) throws ListenerNotFoundException
{
log.log(Level.CONFIG, "removes a notify listener obj {0}", obj);
boolean removed = false;
synchronized (notifListeners) {
Iterator<NotificationHolder> it = notifListeners.iterator();
while (it.hasNext()) {
NotificationHolder mnf = it.next();
if (mnf.getNL().equals(notiflistener)
&& ((mnf.getFilter() == null) || mnf.getFilter().equals(filt))
&& ((mnf.getFilter() == null) || mnf.getObj().equals(obj))) {
it.remove();
removed = true;
}
}
}
if (!removed) {
throw new ListenerNotFoundException("Didn't remove the given NotificationListener");
}
}
/**
* Method that the subclass can override, but doesn't have to
* @return MBeanNotificationInfo array of notification (and types) this
* class can emitt.
*/
@Override
public MBeanNotificationInfo[] getNotificationInfo() {
MBeanNotificationInfo[] info = new MBeanNotificationInfo[1];
String[] supptypes = { NOTIFICATIONACTIONTYPE, NOTIFICATIONINFOLONGTYPE,
NOTIFICATIONINFOSTRINGTYPE };
String name = "AgentIndexRunner";
String descr = "OpenGrok Indexer Notifications";
MBeanNotificationInfo minfo = new MBeanNotificationInfo(supptypes, name,
descr);
info[0] = minfo;
return info;
}
@SuppressWarnings("boxing")
private void doNotify(String type, String msg, Object userdata) {
try {
log.log(Level.CONFIG, "start notifying {0} listeners",
notifListeners.size());
long ts = System.currentTimeMillis();
sequenceNo++;
Notification notif = new Notification(type, this, sequenceNo, ts, msg);
notif.setUserData(userdata);
synchronized (notifListeners) {
for (NotificationHolder nl : notifListeners) {
log.log(Level.FINE, "having one with obj {0}", nl.getObj());
try {
if ((nl.getFilter() == null) ||
nl.getFilter().isNotificationEnabled(notif)) {
nl.getNL().handleNotification(notif, nl.getObj());
}
} catch (Exception exnot) {
log.log(Level.WARNING, "Ex " + exnot, exnot);
}
}
}
} catch (Exception ex) {
log.warning("Exception during notification sending: " + ex.getMessage());
log.log(Level.FINE, "doNotify", ex);
}
}
}