Updated Branches: refs/heads/2334-loghost-squash [created] fde149125
ACCUMULO-2334 Remove ACCUMULO_LOG_HOST in favor of pull host and port log-forwarding from zookeeper Advertising both the host and port for log4j gives us a couple of benefits. We can do away with ACCUMULO_LOG_HOST, simplify the code to always do the same thing (pull from zookeeper), and gain robust failover if the monitor is moved to a different host or is restarted with a random port (does not require any other service restart to become aware). Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/fde14912 Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/fde14912 Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/fde14912 Branch: refs/heads/2334-loghost-squash Commit: fde149125478a294c0a04009ad6c96a7aea27675 Parents: 19e6b10 Author: Josh Elser <[email protected]> Authored: Thu Feb 6 21:18:49 2014 -0500 Committer: Josh Elser <[email protected]> Committed: Fri Feb 7 17:12:23 2014 -0500 ---------------------------------------------------------------------- .../1GB/native-standalone/accumulo-env.sh | 1 - conf/examples/1GB/standalone/accumulo-env.sh | 1 - .../2GB/native-standalone/accumulo-env.sh | 1 - conf/examples/2GB/standalone/accumulo-env.sh | 1 - .../3GB/native-standalone/accumulo-env.sh | 1 - conf/examples/3GB/standalone/accumulo-env.sh | 1 - .../512MB/native-standalone/accumulo-env.sh | 1 - conf/examples/512MB/standalone/accumulo-env.sh | 1 - .../org/apache/accumulo/core/Constants.java | 2 +- .../apache/accumulo/core/util/AddressUtil.java | 4 + .../org/apache/accumulo/server/Accumulo.java | 74 +------------ .../accumulo/server/monitor/LogService.java | 20 +++- .../apache/accumulo/server/monitor/Monitor.java | 17 ++- .../server/watcher/MonitorLog4jWatcher.java | 110 +++++++++++++++++++ 14 files changed, 148 insertions(+), 87 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/1GB/native-standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/1GB/native-standalone/accumulo-env.sh b/conf/examples/1GB/native-standalone/accumulo-env.sh index aa4a1d0..868a665 100755 --- a/conf/examples/1GB/native-standalone/accumulo-env.sh +++ b/conf/examples/1GB/native-standalone/accumulo-env.sh @@ -52,7 +52,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx64m -Xms64m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx128m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/1GB/standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/1GB/standalone/accumulo-env.sh b/conf/examples/1GB/standalone/accumulo-env.sh index 1707f3d..e25b028 100755 --- a/conf/examples/1GB/standalone/accumulo-env.sh +++ b/conf/examples/1GB/standalone/accumulo-env.sh @@ -52,7 +52,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx64m -Xms64m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx128m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/2GB/native-standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/2GB/native-standalone/accumulo-env.sh b/conf/examples/2GB/native-standalone/accumulo-env.sh index ef74ca7..848fb91 100755 --- a/conf/examples/2GB/native-standalone/accumulo-env.sh +++ b/conf/examples/2GB/native-standalone/accumulo-env.sh @@ -51,7 +51,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx128m -Xms128m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx256m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/2GB/standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/2GB/standalone/accumulo-env.sh b/conf/examples/2GB/standalone/accumulo-env.sh index 75014c2..74ddee5 100755 --- a/conf/examples/2GB/standalone/accumulo-env.sh +++ b/conf/examples/2GB/standalone/accumulo-env.sh @@ -52,7 +52,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx128m -Xms128m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx256m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/3GB/native-standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/3GB/native-standalone/accumulo-env.sh b/conf/examples/3GB/native-standalone/accumulo-env.sh index ae0da11..9fe07e7 100755 --- a/conf/examples/3GB/native-standalone/accumulo-env.sh +++ b/conf/examples/3GB/native-standalone/accumulo-env.sh @@ -51,7 +51,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx256m -Xms256m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx1g -Xms256m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/3GB/standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/3GB/standalone/accumulo-env.sh b/conf/examples/3GB/standalone/accumulo-env.sh index 7edd938..7a0992c 100755 --- a/conf/examples/3GB/standalone/accumulo-env.sh +++ b/conf/examples/3GB/standalone/accumulo-env.sh @@ -51,7 +51,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx256m -Xms256m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx1g -Xms256m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/512MB/native-standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/512MB/native-standalone/accumulo-env.sh b/conf/examples/512MB/native-standalone/accumulo-env.sh index 749a678..2e15473 100755 --- a/conf/examples/512MB/native-standalone/accumulo-env.sh +++ b/conf/examples/512MB/native-standalone/accumulo-env.sh @@ -51,7 +51,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx64m -Xms64m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx128m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/conf/examples/512MB/standalone/accumulo-env.sh ---------------------------------------------------------------------- diff --git a/conf/examples/512MB/standalone/accumulo-env.sh b/conf/examples/512MB/standalone/accumulo-env.sh index 9beb059..45d64ef 100755 --- a/conf/examples/512MB/standalone/accumulo-env.sh +++ b/conf/examples/512MB/standalone/accumulo-env.sh @@ -51,7 +51,6 @@ test -z "$ACCUMULO_MONITOR_OPTS" && export ACCUMULO_MONITOR_OPTS="${POLICY} -Xmx test -z "$ACCUMULO_GC_OPTS" && export ACCUMULO_GC_OPTS="-Xmx64m -Xms64m" test -z "$ACCUMULO_GENERAL_OPTS" && export ACCUMULO_GENERAL_OPTS="-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -Djava.net.preferIPv4Stack=true" test -z "$ACCUMULO_OTHER_OPTS" && export ACCUMULO_OTHER_OPTS="-Xmx128m -Xms64m" -export ACCUMULO_LOG_HOST=`(grep -v '^#' $ACCUMULO_HOME/conf/monitor | egrep -v '^[[:space:]]*$' ; echo localhost ) 2>/dev/null | head -1` # what do when the JVM runs out of heap memory export ACCUMULO_KILL_CMD='kill -9 %p' http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/core/src/main/java/org/apache/accumulo/core/Constants.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java b/core/src/main/java/org/apache/accumulo/core/Constants.java index 9bb3419..abf486a 100644 --- a/core/src/main/java/org/apache/accumulo/core/Constants.java +++ b/core/src/main/java/org/apache/accumulo/core/Constants.java @@ -64,7 +64,7 @@ public class Constants { public static final String ZGC_LOCK = ZGC + "/lock"; public static final String ZMONITOR = "/monitor"; - public static final String ZMONITOR_LOG4J_PORT = ZMONITOR + "/log4j_port"; + public static final String ZMONITOR_LOG4J_ADDR = ZMONITOR + "/log4j_addr"; public static final String ZCONFIG = "/config"; http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/core/src/main/java/org/apache/accumulo/core/util/AddressUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/util/AddressUtil.java b/core/src/main/java/org/apache/accumulo/core/util/AddressUtil.java index af9a1a6..e33a919 100644 --- a/core/src/main/java/org/apache/accumulo/core/util/AddressUtil.java +++ b/core/src/main/java/org/apache/accumulo/core/util/AddressUtil.java @@ -44,6 +44,10 @@ public class AddressUtil extends org.apache.accumulo.fate.util.AddressUtil { return new TSocket(addr.getHostName(), addr.getPort()); } + static public String getHostAddress(InetSocketAddress addr) { + return addr.getAddress().getHostAddress(); + } + static public String toString(InetSocketAddress addr) { return addr.getAddress().getHostAddress() + ":" + addr.getPort(); } http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/server/src/main/java/org/apache/accumulo/server/Accumulo.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/accumulo/server/Accumulo.java b/server/src/main/java/org/apache/accumulo/server/Accumulo.java index f4da33b..cc08c99 100644 --- a/server/src/main/java/org/apache/accumulo/server/Accumulo.java +++ b/server/src/main/java/org/apache/accumulo/server/Accumulo.java @@ -32,23 +32,18 @@ import org.apache.accumulo.core.trace.DistributedTrace; import org.apache.accumulo.core.util.AddressUtil; import org.apache.accumulo.core.util.UtilWaitThread; import org.apache.accumulo.core.util.Version; -import org.apache.accumulo.core.zookeeper.ZooUtil; import org.apache.accumulo.server.client.HdfsZooInstance; import org.apache.accumulo.server.conf.ServerConfiguration; import org.apache.accumulo.server.util.time.SimpleTimer; +import org.apache.accumulo.server.watcher.MonitorLog4jWatcher; import org.apache.accumulo.server.zookeeper.ZooReaderWriter; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hdfs.DistributedFileSystem; -import org.apache.log4j.LogManager; import org.apache.log4j.Logger; -import org.apache.log4j.helpers.FileWatchdog; import org.apache.log4j.helpers.LogLog; -import org.apache.log4j.xml.DOMConfigurator; import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.WatchedEvent; -import org.apache.zookeeper.Watcher; public class Accumulo { @@ -87,58 +82,7 @@ public class Accumulo { log.error("creating remote sink for trace spans", ex); } } - - private static class LogMonitor extends FileWatchdog implements Watcher { - String path; - - protected LogMonitor(String instance, String filename, int delay) { - super(filename); - setDelay(delay); - this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_PORT; - } - - private void setMonitorPort() { - try { - String port = new String(ZooReaderWriter.getInstance().getData(path, null), Constants.UTF8); - System.setProperty("org.apache.accumulo.core.host.log.port", port); - log.info("Changing monitor log4j port to "+port); - doOnChange(); - } catch (Exception e) { - log.error("Error reading zookeeper data for monitor log4j port", e); - } - } - - @Override - public void run() { - try { - if (ZooReaderWriter.getInstance().getZooKeeper().exists(path, this) != null) - setMonitorPort(); - log.info("Set watch for monitor log4j port"); - } catch (Exception e) { - log.error("Unable to set watch for monitor log4j port " + path); - } - super.run(); - } - - @Override - protected void doOnChange() { - LogManager.resetConfiguration(); - new DOMConfigurator().doConfigure(filename, LogManager.getLoggerRepository()); - } - - @Override - public void process(WatchedEvent event) { - setMonitorPort(); - if (event.getPath() != null) { - try { - ZooReaderWriter.getInstance().exists(event.getPath(), this); - } catch (Exception ex) { - log.error("Unable to reset watch for monitor log4j port", ex); - } - } - } - } - + public static void init(FileSystem fs, ServerConfiguration config, String application) throws UnknownHostException { System.setProperty("org.apache.accumulo.core.application", application); @@ -151,11 +95,6 @@ public class Accumulo { String localhost = InetAddress.getLocalHost().getHostName(); System.setProperty("org.apache.accumulo.core.ip.localhost.hostname", localhost); - if (System.getenv("ACCUMULO_LOG_HOST") != null) - System.setProperty("org.apache.accumulo.core.host.log", System.getenv("ACCUMULO_LOG_HOST")); - else - System.setProperty("org.apache.accumulo.core.host.log", localhost); - int logPort = config.getConfiguration().getPort(Property.MONITOR_LOG4J_PORT); System.setProperty("org.apache.accumulo.core.host.log.port", Integer.toString(logPort)); @@ -167,12 +106,9 @@ public class Accumulo { } // Turn off messages about not being able to reach the remote logger... we protect against that. LogLog.setQuietMode(true); - - // Configure logging - if (logPort==0) - new LogMonitor(config.getInstance().getInstanceID(), logConfig, 5000).start(); - else - DOMConfigurator.configureAndWatch(logConfig, 5000); + + // Configure logging using information advetised in zookeeper by the monitor + new MonitorLog4jWatcher(config.getInstance().getInstanceID(), logConfig, 5000).start(); log.info(application + " starting"); log.info("Instance " + config.getInstance().getInstanceID()); http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java b/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java index 10ef9e4..e6db13c 100644 --- a/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java +++ b/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java @@ -78,14 +78,26 @@ public class LogService extends org.apache.log4j.AppenderSkeleton { } } - static void startLogListener(AccumuloConfiguration conf, String instanceId) { + /** + * Place the host:port advertisement for the Monitor's Log4j listener in ZooKeeper + * @param conf configuration for the instance + * @param instanceId instanceId for the instance + * @param hostAddress Address that monitor process is bound to + */ + static void startLogListener(AccumuloConfiguration conf, String instanceId, String hostAddress) { try { SocketServer server = new SocketServer(conf.getPort(Property.MONITOR_LOG4J_PORT)); - ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(instanceId) + Constants.ZMONITOR_LOG4J_PORT, - Integer.toString(server.getLocalPort()).getBytes(Constants.UTF8), NodeExistsPolicy.OVERWRITE); + + // getLocalPort will return the actual ephemeral port used when '0' was provided. + String logForwardingAddr = hostAddress + ":" + server.getLocalPort(); + + log.debug("Setting monitor log4j log-forwarding address to: " + logForwardingAddr); + + ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(instanceId) + Constants.ZMONITOR_LOG4J_ADDR, + logForwardingAddr.getBytes(Constants.UTF8), NodeExistsPolicy.OVERWRITE); new Daemon(server).start(); } catch (Throwable t) { - log.info("Unable to listen to cluster-wide ports", t); + log.info("Unable to start/advertise Log4j listener for log-forwarding to monitor", t); } } http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java b/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java index dcb80fd..8cb0d84 100644 --- a/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java +++ b/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java @@ -482,20 +482,27 @@ public class Monitor { server.addServlet(ShellServlet.class, "/shell"); server.start(); - + InetSocketAddress monitorAddress = null; try { hostname = InetAddress.getLocalHost().getHostName(); log.debug("Using " + hostname + " to advertise monitor location in ZooKeeper"); - String monitorAddress = org.apache.accumulo.core.util.AddressUtil.toString(new InetSocketAddress(hostname, server.getPort())); - ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(instance) + Constants.ZMONITOR, monitorAddress.getBytes(Constants.UTF8), + monitorAddress = new InetSocketAddress(hostname, server.getPort()); + String monitorAddressAndPort = org.apache.accumulo.core.util.AddressUtil.toString(monitorAddress); + ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(instance) + Constants.ZMONITOR, monitorAddressAndPort.getBytes(Constants.UTF8), NodeExistsPolicy.OVERWRITE); log.info("Set monitor address in zookeeper to " + monitorAddress); } catch (Exception ex) { - log.error("Unable to set monitor address in zookeeper"); + log.error("Unable to set monitor HTTP address in zookeeper", ex); + } + + if (null != monitorAddress) { + LogService.startLogListener(Monitor.getSystemConfiguration(), instance.getInstanceID(), + org.apache.accumulo.core.util.AddressUtil.getHostAddress(monitorAddress)); + } else { + log.warn("Not starting log4j listener as we could not determine address to use"); } - LogService.startLogListener(Monitor.getSystemConfiguration(), instance.getInstanceID()); new Daemon(new LoggingRunnable(log, new ZooKeeperStatus()), "ZooKeeperStatus").start(); http://git-wip-us.apache.org/repos/asf/accumulo/blob/fde14912/server/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java b/server/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java new file mode 100644 index 0000000..b084c64 --- /dev/null +++ b/server/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.accumulo.server.watcher; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.zookeeper.ZooUtil; +import org.apache.accumulo.server.zookeeper.ZooReaderWriter; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.log4j.helpers.FileWatchdog; +import org.apache.log4j.xml.DOMConfigurator; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; + +import com.google.common.net.HostAndPort; + +/** + * Watcher that updates the monitor's log4j port from ZooKeeper in a system property + */ +public class MonitorLog4jWatcher extends FileWatchdog implements Watcher { + private static final Logger log = Logger.getLogger(MonitorLog4jWatcher.class); + + private static final String HOST_PROPERTY_NAME = "org.apache.accumulo.core.host.log"; + private static final String PORT_PROPERTY_NAME = "org.apache.accumulo.core.host.log.port"; + + protected String path; + + /** + * @param zkPath + * @param filename + * @param delay + * @param propertyName + */ + public MonitorLog4jWatcher(String instance, String filename, int delay) { + super(filename); + setDelay(delay); + this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR; + } + + @Override + public void run() { + try { + // Initially set the logger if the Monitor's log4j advertisement node exists + if (ZooReaderWriter.getInstance().getZooKeeper().exists(path, this) != null) + updateMonitorLog4jLocation(); + log.info("Set watch for Monitor Log4j watcher"); + } catch (Exception e) { + log.error("Unable to set watch for Monitor Log4j watcher on " + path); + } + + super.run(); + } + + @Override + protected void doOnChange() { + // Force a reset on the logger's configuration + LogManager.resetConfiguration(); + new DOMConfigurator().doConfigure(filename, LogManager.getLoggerRepository()); + } + + @Override + public void process(WatchedEvent event) { + // We got an update, process the data in the node + updateMonitorLog4jLocation(); + + if (event.getPath() != null) { + try { + ZooReaderWriter.getInstance().exists(event.getPath(), this); + } catch (Exception ex) { + log.error("Unable to reset watch for Monitor Log4j watcher", ex); + } + } + } + + /** + * Read the host and port information for the Monitor's log4j socket and update + * the system properties so that, on logger refresh, it sees the new information. + */ + protected void updateMonitorLog4jLocation() { + try { + String hostPortString = new String(ZooReaderWriter.getInstance().getData(path, null), Constants.UTF8); + HostAndPort hostAndPort = HostAndPort.fromString(hostPortString); + + System.setProperty(HOST_PROPERTY_NAME, hostAndPort.getHostText()); + System.setProperty(PORT_PROPERTY_NAME, Integer.toString(hostAndPort.getPort())); + + log.info("Changing monitor log4j address to " + hostAndPort.toString()); + + doOnChange(); + } catch (IllegalArgumentException e) { + log.error("Could not parse host and port information", e); + } catch (Exception e) { + log.error("Error reading zookeeper data for Monitor Log4j watcher", e); + } + } +}
