You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2017/03/09 01:53:20 UTC
[1/4] accumulo git commit: ACCUMULO-4409 Create
AccumuloMonitorAppender
Repository: accumulo
Updated Branches:
refs/heads/master 861db793d -> cbdf02e33
ACCUMULO-4409 Create AccumuloMonitorAppender
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9bc9ec32
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9bc9ec32
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9bc9ec32
Branch: refs/heads/master
Commit: 9bc9ec32c74ba0a313ac8fa0c3f7df5f6a3c6deb
Parents: a1cf8e2
Author: Christopher Tubbs <ct...@apache.org>
Authored: Thu Feb 23 22:55:24 2017 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Mon Mar 6 16:41:43 2017 -0500
----------------------------------------------------------------------
.../monitor/util/AccumuloMonitorAppender.java | 125 +++++++++++++++++++
1 file changed, 125 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9bc9ec32/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
new file mode 100644
index 0000000..5802920
--- /dev/null
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
@@ -0,0 +1,125 @@
+/*
+ * 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.monitor.util;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.zookeeper.ZooCache;
+import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
+import org.apache.accumulo.server.client.HdfsZooInstance;
+import org.apache.log4j.AsyncAppender;
+import org.apache.log4j.net.SocketAppender;
+
+import com.google.common.net.HostAndPort;
+
+public class AccumuloMonitorAppender extends AsyncAppender {
+
+ private final ScheduledExecutorService executorService;
+ private final AtomicBoolean trackerScheduled;
+
+ /**
+ * A Log4j Appender which follows the registered location of the active Accumulo monitor service, and forwards log messages to it
+ */
+ public AccumuloMonitorAppender() {
+ // create the background thread to watch for updates to monitor location
+ trackerScheduled = new AtomicBoolean(false);
+ executorService = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread t = new Thread(runnable, "MonitorLog4jLocationTracker");
+ t.setDaemon(true);
+ return t;
+ });
+ }
+
+ @Override
+ public void activateOptions() {
+ if (trackerScheduled.compareAndSet(false, true)) {
+ executorService.scheduleAtFixedRate(new MonitorTracker(HdfsZooInstance.getInstance()), 5, 10, TimeUnit.SECONDS);
+ }
+ super.activateOptions();
+ }
+
+ @Override
+ public void close() {
+ if (!executorService.isShutdown()) {
+ executorService.shutdownNow();
+ }
+ super.close();
+ }
+
+ private class MonitorTracker implements Runnable {
+
+ private final String path;
+ private final ZooCache zooCache;
+
+ private byte[] lastLocation;
+ private SocketAppender lastSocketAppender;
+
+ public MonitorTracker(Instance instance) {
+ this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
+ this.zooCache = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
+
+ this.lastLocation = null;
+ this.lastSocketAppender = null;
+ }
+
+ @Override
+ public void run() {
+ byte[] loc = zooCache.get(path);
+ if (!Arrays.equals(loc, lastLocation)) {
+ // something changed
+ switchAppender(loc);
+ }
+ }
+
+ private void switchAppender(byte[] loc) {
+ // remove and close the last one, if it was non-null
+ if (lastSocketAppender != null) {
+ AccumuloMonitorAppender.this.removeAppender(lastSocketAppender);
+ lastSocketAppender.close();
+ }
+
+ // create a new one, if it is non-null
+ if (loc != null) {
+
+ int defaultPort = Integer.parseUnsignedInt(Property.MONITOR_LOG4J_PORT.getDefaultValue());
+ HostAndPort remote = HostAndPort.fromString(new String(loc, UTF_8));
+
+ SocketAppender socketAppender = new SocketAppender();
+ socketAppender.setRemoteHost(remote.getHostText());
+ socketAppender.setPort(remote.getPortOrDefault(defaultPort));
+
+ lastLocation = loc;
+ lastSocketAppender = socketAppender;
+
+ socketAppender.activateOptions();
+ AccumuloMonitorAppender.this.addAppender(socketAppender);
+ }
+ }
+
+ }
+
+}
[2/4] accumulo git commit: ACCUMULO-4409 More logging bootstrap
cleanup
Posted by ct...@apache.org.
ACCUMULO-4409 More logging bootstrap cleanup
Clean up unneeded system properties, environment variables, logging
configuration, and related classes.
Fix AccumuloMonitorAppender to reload when monitor restarts in the same
location.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/ff7525ef
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/ff7525ef
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/ff7525ef
Branch: refs/heads/master
Commit: ff7525efea12c090a8c8fa9fa3d75f780cadee42
Parents: 9bc9ec3
Author: Christopher Tubbs <ct...@apache.org>
Authored: Tue Feb 28 18:43:21 2017 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Mon Mar 6 17:15:26 2017 -0500
----------------------------------------------------------------------
assemble/bin/accumulo | 10 +-
assemble/conf/log4j-monitor.properties | 14 +-
assemble/conf/log4j-service.properties | 38 ++---
assemble/conf/log4j.properties | 12 +-
assemble/conf/templates/accumulo-env.sh | 51 ++++---
.../org/apache/accumulo/server/Accumulo.java | 17 +--
.../server/watcher/Log4jConfiguration.java | 59 -------
.../server/watcher/MonitorLog4jWatcher.java | 152 -------------------
.../server/watcher/MonitorLog4jWatcherTest.java | 50 ------
.../monitor/util/AccumuloMonitorAppender.java | 70 ++++++---
.../start/util/AsyncSocketAppender.java | 111 --------------
.../start/util/AsyncSocketAppenderTest.java | 78 ----------
12 files changed, 122 insertions(+), 540 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/assemble/bin/accumulo
----------------------------------------------------------------------
diff --git a/assemble/bin/accumulo b/assemble/bin/accumulo
index b10d931..04b9836 100755
--- a/assemble/bin/accumulo
+++ b/assemble/bin/accumulo
@@ -111,14 +111,12 @@ function main() {
if [ -n "$ACCUMULO_JAVA_PREFIX" ]; then
JAVA=($ACCUMULO_JAVA_PREFIX $JAVA)
fi
-
- CLASSPATH="${lib}/accumulo-start.jar:${conf}:${lib}/slf4j-api.jar:${lib}/slf4j-log4j12.jar:${lib}/log4j.jar:${CLASSPATH}"
- exec "${JAVA[@]}" "-Dapp=$cmd" \
+ CLASSPATH="${conf}:${lib}/*:${CLASSPATH}"
+ export CLASSPATH
+
+ exec "${JAVA[@]}" \
"${JAVA_OPTS[@]}" \
- -classpath "${CLASSPATH}" \
- -Dhadoop.home.dir="${HADOOP_PREFIX}" \
- -Dzookeeper.home.dir="${ZOOKEEPER_HOME}" \
org.apache.accumulo.start.Main \
"$@"
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/assemble/conf/log4j-monitor.properties
----------------------------------------------------------------------
diff --git a/assemble/conf/log4j-monitor.properties b/assemble/conf/log4j-monitor.properties
index 8004677..acddc49 100644
--- a/assemble/conf/log4j-monitor.properties
+++ b/assemble/conf/log4j-monitor.properties
@@ -16,23 +16,21 @@
## Log4j 1.2 file that configures logging for Accumulo Monitor
## The system properties referenced below are configured by accumulo-env.sh
-## Write out INFO and higher to log file
+## Define a log file appender
log4j.appender.file=org.apache.log4j.RollingFileAppender
-log4j.appender.file.File=${accumulo.log.dir}/${accumulo.service.id}.log
+log4j.appender.file.File=${accumulo.log.dir}/${accumulo.application}.log
log4j.appender.file.MaxFileSize=100MB
log4j.appender.file.MaxBackupIndex=10
log4j.appender.file.Threshold=INFO
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{ISO8601} [%-8c{2}] %-5p: %m%n
-## Keep the last few log messages for display to the user in Monitor GUI
+## Define an appender for the Accumulo Monitor to log to its own web GUI
log4j.appender.gui=org.apache.accumulo.server.monitor.LogService
-log4j.appender.gui.Keep=50
log4j.appender.gui.Threshold=WARN
-## Log accumulo messages to file and gui appenders
-log4j.logger.org.apache.accumulo=INFO, file, gui
-log4j.additivity.org.apache.accumulo=false
+## Append monitor logs to its own web GUI
+log4j.logger.org.apache.accumulo=INHERITED, gui
-# Log non-accumulo messages to file
+## Append most logs to file
log4j.rootLogger=INFO, file
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/assemble/conf/log4j-service.properties
----------------------------------------------------------------------
diff --git a/assemble/conf/log4j-service.properties b/assemble/conf/log4j-service.properties
index e7212e0..bc01971 100644
--- a/assemble/conf/log4j-service.properties
+++ b/assemble/conf/log4j-service.properties
@@ -13,43 +13,47 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-## Log4j 1.2 file that configures logging for all Accumulo services (Master, TabletServer, GC & Tracer) except Monitor
+## Log4j 1.2 file that configures logging for all Accumulo services (Master, TabletServer, GC, and Tracer) except Monitor
## The system properties referenced below are configured by accumulo-env.sh
-## Write out INFO and higher to log file
+## Define a log file appender
log4j.appender.file=org.apache.log4j.RollingFileAppender
-log4j.appender.file.File=${accumulo.log.dir}/${accumulo.service.id}.log
+log4j.appender.file.File=${accumulo.log.dir}/${accumulo.application}.log
log4j.appender.file.MaxFileSize=100MB
log4j.appender.file.MaxBackupIndex=10
log4j.appender.file.Threshold=INFO
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{ISO8601} [%-8c{2}] %-5p: %m%n
-## Forwards Accumulo messages at WARN or higher to Accumulo Monitor
-log4j.appender.monitor=org.apache.accumulo.start.util.AsyncSocketAppender
-log4j.appender.monitor.RemoteHost=${accumulo.monitor.host}
-log4j.appender.monitor.Port=${accumulo.monitor.log.port}
-log4j.appender.monitor.Application=${accumulo.service.id}
+## Define an appender to send important logs to the the primary Accumulo Monitor
+## The primary monitor is the one currently holding a shared lock in ZooKeeper,
+## and is typically the one that started first.
+log4j.appender.monitor=org.apache.accumulo.monitor.util.AccumuloMonitorAppender
log4j.appender.monitor.Threshold=WARN
-## Configures Audit logs which are OFF by default.
+## Uncomment to define a log file appender for audit logs
#log4j.appender.audit=org.apache.log4j.DailyRollingFileAppender
-#log4j.appender.audit.File=${accumulo.log.dir}/${accumulo.audit.log}
+#log4j.appender.audit.File=${accumulo.log.dir}/${accumulo.application}.audit
#log4j.appender.audit.DatePattern='.'yyyy-MM-dd
#log4j.appender.audit.layout=org.apache.log4j.PatternLayout
-#log4j.appender.audit.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS/Z} [%c{2}] %-5p: %m%n
-#log4j.logger.org.apache.accumulo.audit=INFO, audit
+#log4j.appender.audit.layout.ConversionPattern=%d{ISO8601} [%c{2}] %-5p: %m%n
#log4j.additivity.org.apache.accumulo.audit=false
-## Uncomment above and comment out line below to turn Audit logging ON
+
+## Change this log level from OFF to one of the following to enable audit logging:
+## INFO
+## enables audit logging (inherit appenders from root logger)
+## INFO, audit
+## enables audit logging using the audit log appender
+## (requires audit log file appender above to be uncommented)
log4j.logger.org.apache.accumulo.audit=OFF
-# Log Accumulo messages to file & monitor
-log4j.logger.org.apache.accumulo=INFO, file, monitor
-log4j.additivity.org.apache.accumulo=false
+## Append logs to the primary Accumulo Monitor
+log4j.logger.org.apache.accumulo=INHERITED, monitor
+## Constrain some particularly spammy loggers
log4j.logger.org.apache.accumulo.core.file.rfile.bcfile=INFO
log4j.logger.org.mortbay.log=WARN
log4j.logger.org.apache.zookeeper=ERROR
-# Log non-accumulo messages to file
+## Append most logs to file
log4j.rootLogger=INFO, file
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/assemble/conf/log4j.properties
----------------------------------------------------------------------
diff --git a/assemble/conf/log4j.properties b/assemble/conf/log4j.properties
index 5e602f2..1258fa6 100644
--- a/assemble/conf/log4j.properties
+++ b/assemble/conf/log4j.properties
@@ -13,22 +13,24 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-# Log4j 1.2 file that configures logging for Accumulo commands
+## Log4j 1.2 file that configures logging for clients and utility commands
-# Setup console appender
+## Define a console appender
log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Target=System.out
+log4j.appender.console.Threshold=ALL
log4j.appender.console.layout.ConversionPattern=%d{ISO8601} [%-8c{2}] %-5p: %m%n
log4j.appender.console.layout=org.apache.log4j.PatternLayout
-# hide AUDIT messages in the shell, alternatively you could send them to a different logger
+## Hide audit logs generated by the shell
log4j.logger.org.apache.accumulo.shell.Shell.audit=WARN
+## Constrain some particularly spammy loggers
log4j.logger.org.apache.accumulo.core.file.rfile.bcfile.Compression=WARN
-log4j.logger.org.apache.accumulo.test.TestRandomDeletes=WARN
log4j.logger.org.apache.commons.vfs2.impl.DefaultFileSystemManager=WARN
log4j.logger.org.apache.hadoop.io.compress=WARN
log4j.logger.org.apache.zookeeper=ERROR
log4j.logger.org.mortbay.log=WARN
-# By default, log everything at INFO or higher to the console
+## Append most logs to console
log4j.rootLogger=INFO, console
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/assemble/conf/templates/accumulo-env.sh
----------------------------------------------------------------------
diff --git a/assemble/conf/templates/accumulo-env.sh b/assemble/conf/templates/accumulo-env.sh
index a58d816..5e3864b 100644
--- a/assemble/conf/templates/accumulo-env.sh
+++ b/assemble/conf/templates/accumulo-env.sh
@@ -41,7 +41,7 @@ export ZOOKEEPER_HOME="${ZOOKEEPER_HOME:-/path/to/zookeeper}"
##################################################################
## JVM options set for all processes. Extra options can be passed in by setting ACCUMULO_JAVA_OPTS to an array of options.
-JAVA_OPTS=("${ACCUMULO_JAVA_OPTS[@]}"
+JAVA_OPTS=("${ACCUMULO_JAVA_OPTS[@]}"
'-XX:+UseConcMarkSweepGC'
'-XX:CMSInitiatingOccupancyFraction=75'
'-XX:+CMSClassUnloadingEnabled'
@@ -55,37 +55,54 @@ ${bin}/accumulo-util build-native &> /dev/null
## JVM options set for individual applications
case "$cmd" in
-master) JAVA_OPTS=("${JAVA_OPTS[@]}" ${masterHigh_masterLow}) ;;
-monitor) JAVA_OPTS=("${JAVA_OPTS[@]}" ${monitorHigh_monitorLow}) ;;
-gc) JAVA_OPTS=("${JAVA_OPTS[@]}" ${gcHigh_gcLow}) ;;
-tserver) JAVA_OPTS=("${JAVA_OPTS[@]}" ${tServerHigh_tServerLow}) ;;
-shell) JAVA_OPTS=("${JAVA_OPTS[@]}" ${shellHigh_shellLow}) ;;
-*) JAVA_OPTS=("${JAVA_OPTS[@]}" ${otherHigh_otherLow}) ;;
+ master) JAVA_OPTS=("${JAVA_OPTS[@]}" ${masterHigh_masterLow}) ;;
+ monitor) JAVA_OPTS=("${JAVA_OPTS[@]}" ${monitorHigh_monitorLow}) ;;
+ gc) JAVA_OPTS=("${JAVA_OPTS[@]}" ${gcHigh_gcLow}) ;;
+ tserver) JAVA_OPTS=("${JAVA_OPTS[@]}" ${tServerHigh_tServerLow}) ;;
+ shell) JAVA_OPTS=("${JAVA_OPTS[@]}" ${shellHigh_shellLow}) ;;
+ *) JAVA_OPTS=("${JAVA_OPTS[@]}" ${otherHigh_otherLow}) ;;
esac
-## JVM options set for logging. Review logj4 properties files to see how they are used.
-JAVA_OPTS=("${JAVA_OPTS[@]}"
+## JVM options set for logging. Review logj4 properties files to see how they are used.
+JAVA_OPTS=("${JAVA_OPTS[@]}"
"-Daccumulo.log.dir=${ACCUMULO_LOG_DIR}"
- "-Daccumulo.service.id=${cmd}${ACCUMULO_SERVICE_INSTANCE}_$(hostname)"
- "-Daccumulo.audit.log=$(hostname).audit")
+ "-Daccumulo.application=${ACCUMULO_CMD}${ACCUMULO_SERVICE_INSTANCE}_$(hostname)")
case "$cmd" in
-monitor) JAVA_OPTS=("${JAVA_OPTS[@]}" "-Dlog4j.configuration=file:${conf}/log4j-monitor.properties") ;;
-gc|master|tserver|tracer) JAVA_OPTS=("${JAVA_OPTS[@]}" "-Dlog4j.configuration=file:${conf}/log4j-service.properties") ;;
-*) JAVA_OPTS=("${JAVA_OPTS[@]}" "-Dlog4j.configuration=file:${conf}/log4j.properties") ;;
+ monitor)
+ JAVA_OPTS=("${JAVA_OPTS[@]}" "-Dlog4j.configuration=log4j-monitor.properties")
+ ;;
+ gc|master|tserver|tracer)
+ JAVA_OPTS=("${JAVA_OPTS[@]}" "-Dlog4j.configuration=log4j-service.properties")
+ ;;
+ *)
+ # let log4j use its default behavior (log4j.xml, log4j.properties)
+ true
+ ;;
esac
export JAVA_OPTS
+## External class path items for Java system class loader (dependencies not included with Accumulo)
+CLASSPATH="$(find "$ZOOKEEPER_HOME"/{,lib} "$HADOOP_PREFIX"/share/hadoop/{common,common/lib,hdfs,mapreduce,yarn} -maxdepth 1 -name '*.jar' \
+ -and -not -name '*slf4j*' \
+ -and -not -name '*fatjar*' \
+ -and -not -name '*-javadoc*' \
+ -and -not -name '*-sources*.jar' \
+ -and -not -name '*-test*.jar' \
+ -print0 | tr '\0' ':')$CLASSPATH"
+CLASSPATH="${HADOOP_CONF_DIR}:${CLASSPATH}"
+export CLASSPATH
+
############################
# Variables set to a default
############################
export MALLOC_ARENA_MAX=${MALLOC_ARENA_MAX:-1}
## Add Hadoop native libraries to shared library paths given operating system
-case "$(uname)" in
-Darwin) export DYLD_LIBRARY_PATH="${HADOOP_PREFIX}/lib/native:${DYLD_LIBRARY_PATH}" ;;
-*) export LD_LIBRARY_PATH="${HADOOP_PREFIX}/lib/native:${LD_LIBRARY_PATH}" ;;
+case "$(uname)" in
+ Darwin) export DYLD_LIBRARY_PATH="${HADOOP_PREFIX}/lib/native:${DYLD_LIBRARY_PATH}" ;;
+ *) export LD_LIBRARY_PATH="${HADOOP_PREFIX}/lib/native:${LD_LIBRARY_PATH}" ;;
esac
###############################################
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/server/base/src/main/java/org/apache/accumulo/server/Accumulo.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/Accumulo.java b/server/base/src/main/java/org/apache/accumulo/server/Accumulo.java
index e423497..e849ae3 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/Accumulo.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/Accumulo.java
@@ -16,6 +16,7 @@
*/
package org.apache.accumulo.server;
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.nio.charset.StandardCharsets.UTF_8;
import java.io.File;
@@ -30,7 +31,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.util.AddressUtil;
@@ -44,7 +44,6 @@ import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
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;
@@ -52,8 +51,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
-import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
-
public class Accumulo {
private static final Logger log = Logger.getLogger(Accumulo.class);
@@ -107,18 +104,6 @@ public class Accumulo {
public static void init(VolumeManager fs, ServerConfigurationFactory serverConfig, String application) throws IOException {
final AccumuloConfiguration conf = serverConfig.getConfiguration();
- final Instance instance = serverConfig.getInstance();
-
- String logConfigFile = System.getProperty("log4j.configuration");
- if (logConfigFile != null) {
- if (logConfigFile.startsWith("file:")) {
- logConfigFile = logConfigFile.split(":")[1];
- }
- // Set up polling log4j updates and log-forwarding using information advertised in zookeeper by the monitor
- MonitorLog4jWatcher logConfigWatcher = new MonitorLog4jWatcher(instance.getInstanceID(), logConfigFile);
- logConfigWatcher.setDelay(5000L);
- logConfigWatcher.start();
- }
log.info(application + " starting");
log.info("Instance " + serverConfig.getInstance().getInstanceID());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/server/base/src/main/java/org/apache/accumulo/server/watcher/Log4jConfiguration.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/watcher/Log4jConfiguration.java b/server/base/src/main/java/org/apache/accumulo/server/watcher/Log4jConfiguration.java
deleted file mode 100644
index 68e4897..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/watcher/Log4jConfiguration.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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 static java.util.Objects.requireNonNull;
-
-import java.io.File;
-
-import org.apache.log4j.LogManager;
-import org.apache.log4j.PropertyConfigurator;
-import org.apache.log4j.xml.DOMConfigurator;
-
-/**
- * Encapsulate calls to PropertyConfigurator or DOMConfigurator to set up logging
- */
-public class Log4jConfiguration {
-
- private final boolean usingProperties;
- private final String filename;
- private final File log4jFile;
-
- public Log4jConfiguration(String filename) {
- requireNonNull(filename, "log4j configuration filename must not be null");
- usingProperties = filename.endsWith(".properties");
- this.filename = filename;
- log4jFile = new File(filename);
- }
-
- public boolean isUsingProperties() {
- return usingProperties;
- }
-
- public void resetLogger() {
- // Force a reset on the logger's configuration, but only if the configured log4j file actually exists
- // If we reset the configuration blindly, the ITs will not get any logging as they don't set it up on their own
- if (log4jFile.exists() && log4jFile.isFile() && log4jFile.canRead()) {
- LogManager.resetConfiguration();
- if (usingProperties) {
- new PropertyConfigurator().doConfigure(filename, LogManager.getLoggerRepository());
- } else {
- new DOMConfigurator().doConfigure(filename, LogManager.getLoggerRepository());
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java b/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
deleted file mode 100644
index a0783a4..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
+++ /dev/null
@@ -1,152 +0,0 @@
-/*
- * 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 static java.nio.charset.StandardCharsets.UTF_8;
-
-import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.zookeeper.ZooUtil;
-import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
-import org.apache.log4j.Appender;
-import org.apache.log4j.LogManager;
-import org.apache.log4j.Logger;
-import org.apache.log4j.helpers.FileWatchdog;
-import org.apache.zookeeper.KeeperException.NoNodeException;
-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 = "accumulo.monitor.host";
- private static final String PORT_PROPERTY_NAME = "accumulo.monitor.log.port";
-
- private final Object lock;
- private final Log4jConfiguration logConfig;
- private boolean loggingDisabled = false;
- protected String path;
-
- public MonitorLog4jWatcher(String instance, String filename) {
- super(filename);
- this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
- this.lock = new Object();
- this.logConfig = new Log4jConfiguration(filename);
- doOnChange();
- }
-
- boolean isUsingProperties() {
- return logConfig.isUsingProperties();
- }
-
- String getPath() {
- return path;
- }
-
- @Override
- public void run() {
- try {
- // Initially set the logger if the Monitor's log4j advertisement node exists
- if (ZooReaderWriter.getInstance().exists(path, this))
- 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
- public void doOnChange() {
- // this method gets called in the parent class' constructor
- // I'm not sure of a better way to get around this. The final modifier helps though.
- if (null == lock) {
- return;
- }
-
- synchronized (lock) {
- // We might triggered by file-reloading or from ZK update.
- // Either way will result in log-forwarding being restarted
- loggingDisabled = false;
- log.info("Enabled log-forwarding");
- logConfig.resetLogger();
- }
- }
-
- @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), UTF_8);
- 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 (NoNodeException e) {
- // Not sure on the synchronization guarantees for Loggers and Appenders
- // on configuration reload
- synchronized (lock) {
- // Don't need to try to re-disable'ing it.
- if (loggingDisabled) {
- return;
- }
-
- Logger logger = LogManager.getLogger("org.apache.accumulo");
- if (null != logger) {
- // TODO ACCUMULO-2343 Create a specific appender for log-forwarding to the monitor
- // that can replace the AsyncAppender+SocketAppender.
- Appender appender = logger.getAppender("ASYNC");
- if (null != appender) {
- log.info("Closing log-forwarding appender");
- appender.close();
- log.info("Removing log-forwarding appender");
- logger.removeAppender(appender);
- loggingDisabled = true;
- }
- }
- }
- } 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);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/server/base/src/test/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcherTest.java
----------------------------------------------------------------------
diff --git a/server/base/src/test/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcherTest.java b/server/base/src/test/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcherTest.java
deleted file mode 100644
index 53c3c6e..0000000
--- a/server/base/src/test/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcherTest.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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 static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.zookeeper.ZooUtil;
-import org.junit.Before;
-import org.junit.Test;
-
-public class MonitorLog4jWatcherTest {
- private static final String INSTANCE = "instance";
- private static final String FILENAME = "something_logger.xml";
-
- private MonitorLog4jWatcher w;
-
- @Before
- public void setUp() throws Exception {
- w = new MonitorLog4jWatcher(INSTANCE, FILENAME);
- }
-
- @Test
- public void testGetters() {
- assertFalse(w.isUsingProperties());
- assertEquals(ZooUtil.getRoot(INSTANCE) + Constants.ZMONITOR_LOG4J_ADDR, w.getPath());
- }
-
- @Test
- public void testPropertyDetection() {
- w = new MonitorLog4jWatcher(INSTANCE, FILENAME + ".properties");
- assertTrue(w.isUsingProperties());
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
index 5802920..8a855d0 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
@@ -18,7 +18,6 @@ package org.apache.accumulo.monitor.util;
import static java.nio.charset.StandardCharsets.UTF_8;
-import java.util.Arrays;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -33,6 +32,7 @@ import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.log4j.AsyncAppender;
import org.apache.log4j.net.SocketAppender;
+import org.apache.zookeeper.data.Stat;
import com.google.common.net.HostAndPort;
@@ -48,7 +48,7 @@ public class AccumuloMonitorAppender extends AsyncAppender {
// create the background thread to watch for updates to monitor location
trackerScheduled = new AtomicBoolean(false);
executorService = Executors.newSingleThreadScheduledExecutor(runnable -> {
- Thread t = new Thread(runnable, "MonitorLog4jLocationTracker");
+ Thread t = new Thread(runnable, AccumuloMonitorAppender.class.getSimpleName() + " Location Tracker");
t.setDaemon(true);
return t;
});
@@ -56,8 +56,10 @@ public class AccumuloMonitorAppender extends AsyncAppender {
@Override
public void activateOptions() {
+ // only schedule it once (in case options get activated more than once); not sure if this is possible
if (trackerScheduled.compareAndSet(false, true)) {
- executorService.scheduleAtFixedRate(new MonitorTracker(HdfsZooInstance.getInstance()), 5, 10, TimeUnit.SECONDS);
+ // wait 5 seconds, then run every 5 seconds
+ executorService.scheduleAtFixedRate(new MonitorTracker(), 5, 5, TimeUnit.SECONDS);
}
super.activateOptions();
}
@@ -72,52 +74,78 @@ public class AccumuloMonitorAppender extends AsyncAppender {
private class MonitorTracker implements Runnable {
- private final String path;
- private final ZooCache zooCache;
+ private String path;
+ private ZooCache zooCache;
- private byte[] lastLocation;
+ private long lastModifiedTransactionId;
private SocketAppender lastSocketAppender;
- public MonitorTracker(Instance instance) {
- this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
- this.zooCache = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
+ public MonitorTracker() {
- this.lastLocation = null;
+ // path and zooCache are lazily set the first time this tracker is run
+ // this allows the tracker to be constructed and scheduled during log4j initialization without
+ // triggering any actual logs from the Accumulo or ZooKeeper code
+ this.path = null;
+ this.zooCache = null;
+
+ this.lastModifiedTransactionId = 0;
this.lastSocketAppender = null;
}
@Override
public void run() {
- byte[] loc = zooCache.get(path);
- if (!Arrays.equals(loc, lastLocation)) {
- // something changed
- switchAppender(loc);
+ try {
+ // lazily set up path and zooCache (see comment in constructor)
+ if (this.zooCache == null) {
+ Instance instance = HdfsZooInstance.getInstance();
+ this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
+ this.zooCache = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
+ }
+
+ // get the current location from the cache and update if necessary
+ Stat stat = new Stat();
+ byte[] loc = zooCache.get(path, stat);
+ long modifiedTransactionId = stat.getMzxid();
+
+ // modifiedTransactionId will be 0 if no current location
+ // lastModifiedTransactionId will be 0 if we've never seen a location
+ if (modifiedTransactionId != lastModifiedTransactionId) {
+ // replace old socket on every change, even if new location is the same as old location
+ // if modifiedTransactionId changed, then the monitor restarted and the old socket is dead now
+ switchAppender(loc, modifiedTransactionId);
+ }
+ } catch (Exception e) {
+ // dump any non-fatal problems to the console, but let it run again
+ e.printStackTrace();
}
}
- private void switchAppender(byte[] loc) {
+ private void switchAppender(byte[] newLocation, long newModifiedTransactionId) {
// remove and close the last one, if it was non-null
if (lastSocketAppender != null) {
AccumuloMonitorAppender.this.removeAppender(lastSocketAppender);
lastSocketAppender.close();
}
- // create a new one, if it is non-null
- if (loc != null) {
+ // create a new SocketAppender, if new location is non-null
+ if (newLocation != null) {
int defaultPort = Integer.parseUnsignedInt(Property.MONITOR_LOG4J_PORT.getDefaultValue());
- HostAndPort remote = HostAndPort.fromString(new String(loc, UTF_8));
+ HostAndPort remote = HostAndPort.fromString(new String(newLocation, UTF_8));
SocketAppender socketAppender = new SocketAppender();
+ socketAppender.setApplication(System.getProperty("accumulo.application", "unknown"));
socketAppender.setRemoteHost(remote.getHostText());
socketAppender.setPort(remote.getPortOrDefault(defaultPort));
- lastLocation = loc;
- lastSocketAppender = socketAppender;
-
socketAppender.activateOptions();
AccumuloMonitorAppender.this.addAppender(socketAppender);
+
+ lastSocketAppender = socketAppender;
}
+
+ // update lastModifiedTransactionId, even if the new one is 0 (no new location)
+ lastModifiedTransactionId = newModifiedTransactionId;
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/start/src/main/java/org/apache/accumulo/start/util/AsyncSocketAppender.java
----------------------------------------------------------------------
diff --git a/start/src/main/java/org/apache/accumulo/start/util/AsyncSocketAppender.java b/start/src/main/java/org/apache/accumulo/start/util/AsyncSocketAppender.java
deleted file mode 100644
index db64ecb..0000000
--- a/start/src/main/java/org/apache/accumulo/start/util/AsyncSocketAppender.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.start.util;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.log4j.AsyncAppender;
-import org.apache.log4j.net.SocketAppender;
-import org.apache.log4j.spi.LoggingEvent;
-
-/**
- * An asynchronous appender that maintains its own internal socket appender. Unlike <code>AsyncAppender</code>, this appender can be configured with a Log4J
- * properties file, although in that case no additional appenders can be added.
- */
-public class AsyncSocketAppender extends AsyncAppender {
-
- private final SocketAppender socketAppender;
- private final AtomicBoolean activated = new AtomicBoolean(false);
-
- /**
- * Creates a new appender.
- */
- public AsyncSocketAppender() {
- socketAppender = new SocketAppender();
- }
-
- /**
- * Creates a new appender using the given socket appender internally. Use this constructor for testing only.
- */
- AsyncSocketAppender(SocketAppender socketAppender) {
- this.socketAppender = socketAppender;
- }
-
- @Override
- public void append(final LoggingEvent event) {
- // Lazy attachment, to avoid calling non-final method in constructor
- if (!isAttached(socketAppender)) {
- addAppender(socketAppender);
- }
-
- // Lazy activation / connection too, to allow setting host and port
- if (activated.compareAndSet(false, true)) {
- socketAppender.activateOptions();
- }
-
- super.append(event);
- }
-
- // SocketAppender delegate methods
-
- public String getApplication() {
- return socketAppender.getApplication();
- }
-
- // super.getLocationInfo() will always agree with socketAppender
- public int getPort() {
- return socketAppender.getPort();
- }
-
- public int getReconnectionDelay() {
- return socketAppender.getReconnectionDelay();
- }
-
- public String getRemoteHost() {
- return socketAppender.getRemoteHost();
- }
-
- public boolean isAdvertiseViaMulticastDNS() {
- return socketAppender.isAdvertiseViaMulticastDNS();
- }
-
- public void setAdvertiseViaMulticastDNS(boolean advertiseViaMulticastDNS) {
- socketAppender.setAdvertiseViaMulticastDNS(advertiseViaMulticastDNS);
- }
-
- public void setApplication(String lapp) {
- socketAppender.setApplication(lapp);
- }
-
- @Override
- public void setLocationInfo(boolean locationInfo) {
- super.setLocationInfo(locationInfo);
- socketAppender.setLocationInfo(locationInfo);
- }
-
- public void setPort(int port) {
- socketAppender.setPort(port);
- }
-
- public void setReconnectionDelay(int delay) {
- socketAppender.setReconnectionDelay(delay);
- }
-
- public void setRemoteHost(String host) {
- socketAppender.setRemoteHost(host);
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ff7525ef/start/src/test/java/org/apache/accumulo/start/util/AsyncSocketAppenderTest.java
----------------------------------------------------------------------
diff --git a/start/src/test/java/org/apache/accumulo/start/util/AsyncSocketAppenderTest.java b/start/src/test/java/org/apache/accumulo/start/util/AsyncSocketAppenderTest.java
deleted file mode 100644
index 6c73782..0000000
--- a/start/src/test/java/org/apache/accumulo/start/util/AsyncSocketAppenderTest.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.start.util;
-
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
-import org.apache.log4j.net.SocketAppender;
-import org.apache.log4j.spi.LoggingEvent;
-import org.easymock.EasyMock;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-public class AsyncSocketAppenderTest {
- private SocketAppender sa;
- private AsyncSocketAppender asa;
-
- @Before
- public void setUp() throws Exception {
- sa = EasyMock.createMock(SocketAppender.class);
- }
-
- @Test
- public void testDelegates() {
- asa = new AsyncSocketAppender();
- asa.setApplication("myapp");
- asa.setLocationInfo(true);
- asa.setPort(1234);
- asa.setReconnectionDelay(56);
- asa.setRemoteHost("remotehost");
- Assert.assertEquals("myapp", asa.getApplication());
- Assert.assertEquals(true, asa.getLocationInfo()); // not really delegating
- Assert.assertEquals(1234, asa.getPort());
- Assert.assertEquals(56, asa.getReconnectionDelay());
- Assert.assertEquals("remotehost", asa.getRemoteHost());
- }
-
- @Test
- public void testSetLocationInfo() {
- sa.setLocationInfo(true);
- EasyMock.replay(sa);
- asa = new AsyncSocketAppender(sa);
- asa.setLocationInfo(true);
- EasyMock.verify(sa);
- }
-
- @Test
- public void testAppend() {
- asa = new AsyncSocketAppender(sa);
- Assert.assertFalse(asa.isAttached(sa));
- LoggingEvent event1 = new LoggingEvent("java.lang.String", Logger.getRootLogger(), Level.INFO, "event1", null);
- LoggingEvent event2 = new LoggingEvent("java.lang.Integer", Logger.getRootLogger(), Level.WARN, "event2", null);
- sa.activateOptions();
- sa.doAppend(event1);
- sa.doAppend(event2);
- sa.close();
- EasyMock.replay(sa);
- asa.doAppend(event1);
- asa.doAppend(event2);
- asa.close(); // forces events to be appended to socket appender
- Assert.assertTrue(asa.isAttached(sa));
- EasyMock.verify(sa);
- }
-}
[4/4] accumulo git commit: Merge branch
'ACCUMULO-4409-custom-appender'
Posted by ct...@apache.org.
Merge branch 'ACCUMULO-4409-custom-appender'
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/cbdf02e3
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/cbdf02e3
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/cbdf02e3
Branch: refs/heads/master
Commit: cbdf02e330a5de7ad5d224a295befa33d99cfc18
Parents: 861db79 5afbacc
Author: Christopher Tubbs <ct...@apache.org>
Authored: Wed Mar 8 20:52:22 2017 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Wed Mar 8 20:52:22 2017 -0500
----------------------------------------------------------------------
assemble/bin/accumulo | 10 +-
assemble/conf/log4j-monitor.properties | 14 +-
assemble/conf/log4j-service.properties | 38 ++--
assemble/conf/log4j.properties | 12 +-
assemble/conf/templates/accumulo-env.sh | 51 +++--
.../org/apache/accumulo/server/Accumulo.java | 17 +-
.../server/watcher/Log4jConfiguration.java | 59 -----
.../server/watcher/MonitorLog4jWatcher.java | 152 -------------
.../server/watcher/MonitorLog4jWatcherTest.java | 50 -----
.../monitor/util/AccumuloMonitorAppender.java | 222 +++++++++++++++++++
.../util/AccumuloMonitorAppenderTest.java | 184 +++++++++++++++
.../start/util/AsyncSocketAppender.java | 111 ----------
.../start/util/AsyncSocketAppenderTest.java | 78 -------
13 files changed, 479 insertions(+), 519 deletions(-)
----------------------------------------------------------------------
[3/4] accumulo git commit: ACCUMULO-4409 Add tests for monitor
appender
Posted by ct...@apache.org.
ACCUMULO-4409 Add tests for monitor appender
Changes to make AccumuloMonitorAppender more testable and the tests to
accompany it.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/5afbacc0
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/5afbacc0
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/5afbacc0
Branch: refs/heads/master
Commit: 5afbacc0af9a6d7f9c610969eafcfed2734364a2
Parents: ff7525e
Author: Christopher Tubbs <ct...@apache.org>
Authored: Wed Mar 8 20:50:44 2017 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Wed Mar 8 20:50:44 2017 -0500
----------------------------------------------------------------------
.../monitor/util/AccumuloMonitorAppender.java | 193 +++++++++++++------
.../util/AccumuloMonitorAppenderTest.java | 184 ++++++++++++++++++
2 files changed, 315 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/5afbacc0/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
index 8a855d0..a965f0c 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppender.java
@@ -18,10 +18,13 @@ package org.apache.accumulo.monitor.util;
import static java.nio.charset.StandardCharsets.UTF_8;
+import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.function.Supplier;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Instance;
@@ -30,16 +33,19 @@ import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.ZooCache;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
import org.apache.accumulo.server.client.HdfsZooInstance;
+import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.AsyncAppender;
import org.apache.log4j.net.SocketAppender;
import org.apache.zookeeper.data.Stat;
import com.google.common.net.HostAndPort;
-public class AccumuloMonitorAppender extends AsyncAppender {
+public class AccumuloMonitorAppender extends AsyncAppender implements AutoCloseable {
- private final ScheduledExecutorService executorService;
- private final AtomicBoolean trackerScheduled;
+ final ScheduledExecutorService executorService;
+ final AtomicBoolean trackerScheduled;
+ private int frequency = 0;
+ private MonitorTracker tracker = null;
/**
* A Log4j Appender which follows the registered location of the active Accumulo monitor service, and forwards log messages to it
@@ -54,12 +60,33 @@ public class AccumuloMonitorAppender extends AsyncAppender {
});
}
+ public void setFrequency(int millis) {
+ if (millis > 0) {
+ frequency = millis;
+ }
+ }
+
+ public int getFrequency() {
+ return frequency;
+ }
+
+ // this is just for testing
+ void setTracker(MonitorTracker monitorTracker) {
+ tracker = monitorTracker;
+ }
+
@Override
public void activateOptions() {
// only schedule it once (in case options get activated more than once); not sure if this is possible
if (trackerScheduled.compareAndSet(false, true)) {
- // wait 5 seconds, then run every 5 seconds
- executorService.scheduleAtFixedRate(new MonitorTracker(), 5, 5, TimeUnit.SECONDS);
+ if (frequency <= 0) {
+ // use default rate of 5 seconds between each check
+ frequency = 5000;
+ }
+ if (tracker == null) {
+ tracker = new MonitorTracker(this, new ZooCacheLocationSupplier(), new SocketAppenderFactory());
+ }
+ executorService.scheduleWithFixedDelay(tracker, frequency, frequency, TimeUnit.MILLISECONDS);
}
super.activateOptions();
}
@@ -72,80 +99,122 @@ public class AccumuloMonitorAppender extends AsyncAppender {
super.close();
}
- private class MonitorTracker implements Runnable {
+ static class MonitorLocation {
+ private final String location;
+ private final long modId;
- private String path;
- private ZooCache zooCache;
-
- private long lastModifiedTransactionId;
- private SocketAppender lastSocketAppender;
+ public MonitorLocation(long modId, byte[] location) {
+ this.modId = modId;
+ this.location = location == null ? null : new String(location, UTF_8);
+ }
- public MonitorTracker() {
+ public boolean hasLocation() {
+ return location != null;
+ }
- // path and zooCache are lazily set the first time this tracker is run
- // this allows the tracker to be constructed and scheduled during log4j initialization without
- // triggering any actual logs from the Accumulo or ZooKeeper code
- this.path = null;
- this.zooCache = null;
+ public String getLocation() {
+ return location;
+ }
- this.lastModifiedTransactionId = 0;
- this.lastSocketAppender = null;
+ @Override
+ public boolean equals(Object obj) {
+ if (obj != null && obj instanceof MonitorLocation) {
+ MonitorLocation other = (MonitorLocation) obj;
+ return modId == other.modId && Objects.equals(location, other.location);
+ }
+ return false;
}
@Override
- public void run() {
- try {
- // lazily set up path and zooCache (see comment in constructor)
- if (this.zooCache == null) {
- Instance instance = HdfsZooInstance.getInstance();
- this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
- this.zooCache = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
- }
+ public int hashCode() {
+ return Long.hashCode(modId);
+ }
+ }
- // get the current location from the cache and update if necessary
- Stat stat = new Stat();
- byte[] loc = zooCache.get(path, stat);
- long modifiedTransactionId = stat.getMzxid();
-
- // modifiedTransactionId will be 0 if no current location
- // lastModifiedTransactionId will be 0 if we've never seen a location
- if (modifiedTransactionId != lastModifiedTransactionId) {
- // replace old socket on every change, even if new location is the same as old location
- // if modifiedTransactionId changed, then the monitor restarted and the old socket is dead now
- switchAppender(loc, modifiedTransactionId);
- }
- } catch (Exception e) {
- // dump any non-fatal problems to the console, but let it run again
- e.printStackTrace();
+ private static class ZooCacheLocationSupplier implements Supplier<MonitorLocation> {
+
+ // path and zooCache are lazily set the first time this tracker is run
+ // this allows the tracker to be constructed and scheduled during log4j initialization without
+ // triggering any actual logs from the Accumulo or ZooKeeper code
+ private String path = null;
+ private ZooCache zooCache = null;
+
+ @Override
+ public MonitorLocation get() {
+ // lazily set up path and zooCache (see comment in constructor)
+ if (this.zooCache == null) {
+ Instance instance = HdfsZooInstance.getInstance();
+ this.path = ZooUtil.getRoot(instance) + Constants.ZMONITOR_LOG4J_ADDR;
+ this.zooCache = new ZooCacheFactory().getZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut());
}
+
+ // get the current location from the cache and update if necessary
+ Stat stat = new Stat();
+ byte[] loc = zooCache.get(path, stat);
+ // mzxid is 0 if location does not exist and the non-zero transaction id of the last modification otherwise
+ return new MonitorLocation(stat.getMzxid(), loc);
}
+ }
- private void switchAppender(byte[] newLocation, long newModifiedTransactionId) {
- // remove and close the last one, if it was non-null
- if (lastSocketAppender != null) {
- AccumuloMonitorAppender.this.removeAppender(lastSocketAppender);
- lastSocketAppender.close();
- }
+ private static class SocketAppenderFactory implements Function<MonitorLocation,AppenderSkeleton> {
+ @Override
+ public AppenderSkeleton apply(MonitorLocation loc) {
+ int defaultPort = Integer.parseUnsignedInt(Property.MONITOR_LOG4J_PORT.getDefaultValue());
+ HostAndPort remote = HostAndPort.fromString(loc.getLocation());
+
+ SocketAppender socketAppender = new SocketAppender();
+ socketAppender.setApplication(System.getProperty("accumulo.application", "unknown"));
+ socketAppender.setRemoteHost(remote.getHostText());
+ socketAppender.setPort(remote.getPortOrDefault(defaultPort));
- // create a new SocketAppender, if new location is non-null
- if (newLocation != null) {
+ return socketAppender;
+ }
+ }
- int defaultPort = Integer.parseUnsignedInt(Property.MONITOR_LOG4J_PORT.getDefaultValue());
- HostAndPort remote = HostAndPort.fromString(new String(newLocation, UTF_8));
+ static class MonitorTracker implements Runnable {
- SocketAppender socketAppender = new SocketAppender();
- socketAppender.setApplication(System.getProperty("accumulo.application", "unknown"));
- socketAppender.setRemoteHost(remote.getHostText());
- socketAppender.setPort(remote.getPortOrDefault(defaultPort));
+ private final AccumuloMonitorAppender parentAsyncAppender;
+ private final Supplier<MonitorLocation> currentLocationSupplier;
+ private final Function<MonitorLocation,AppenderSkeleton> appenderFactory;
- socketAppender.activateOptions();
- AccumuloMonitorAppender.this.addAppender(socketAppender);
+ private MonitorLocation lastLocation;
+ private AppenderSkeleton lastSocketAppender;
- lastSocketAppender = socketAppender;
- }
+ public MonitorTracker(AccumuloMonitorAppender appender, Supplier<MonitorLocation> currentLocationSupplier,
+ Function<MonitorLocation,AppenderSkeleton> appenderFactory) {
+ this.parentAsyncAppender = Objects.requireNonNull(appender);
+ this.appenderFactory = Objects.requireNonNull(appenderFactory);
+ this.currentLocationSupplier = Objects.requireNonNull(currentLocationSupplier);
- // update lastModifiedTransactionId, even if the new one is 0 (no new location)
- lastModifiedTransactionId = newModifiedTransactionId;
+ this.lastLocation = new MonitorLocation(0, null);
+ this.lastSocketAppender = null;
+ }
+
+ @Override
+ public void run() {
+ try {
+ MonitorLocation currentLocation = currentLocationSupplier.get();
+ // detect change
+ if (!currentLocation.equals(lastLocation)) {
+ // clean up old appender
+ if (lastSocketAppender != null) {
+ parentAsyncAppender.removeAppender(lastSocketAppender);
+ lastSocketAppender.close();
+ lastSocketAppender = null;
+ }
+ // create a new one
+ if (currentLocation.hasLocation()) {
+ lastSocketAppender = appenderFactory.apply(currentLocation);
+ lastSocketAppender.activateOptions();
+ parentAsyncAppender.addAppender(lastSocketAppender);
+ }
+ // update the last location only if switching was successful
+ lastLocation = currentLocation;
+ }
+ } catch (Exception e) {
+ // dump any non-fatal problems to the console, but let it run again
+ e.printStackTrace();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/5afbacc0/server/monitor/src/test/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppenderTest.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/test/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppenderTest.java b/server/monitor/src/test/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppenderTest.java
new file mode 100644
index 0000000..cd4eb03
--- /dev/null
+++ b/server/monitor/src/test/java/org/apache/accumulo/monitor/util/AccumuloMonitorAppenderTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.monitor.util;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.fail;
+
+import java.util.Enumeration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.accumulo.monitor.util.AccumuloMonitorAppender.MonitorLocation;
+import org.apache.accumulo.monitor.util.AccumuloMonitorAppender.MonitorTracker;
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.spi.LoggingEvent;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.Timeout;
+
+public class AccumuloMonitorAppenderTest {
+
+ @Rule
+ public Timeout timeout = new Timeout(10, TimeUnit.SECONDS);
+
+ @Rule
+ public ExpectedException exception = ExpectedException.none();
+
+ @Test
+ public void testActivateOptions() {
+ try (AccumuloMonitorAppender appender = new AccumuloMonitorAppender()) {
+ appender.executorService.shutdown();
+ // simulate tracker having already been scheduled
+ appender.trackerScheduled.compareAndSet(false, true);
+ appender.activateOptions();
+ // activateOptions should not trigger a RejectedExecutionException, because we tricked it into thinking it was already called, and therefore it did not
+ // schedule the tracker after shutting down
+ }
+
+ exception.expect(RejectedExecutionException.class);
+ try (AccumuloMonitorAppender appender = new AccumuloMonitorAppender()) {
+ appender.executorService.shutdown();
+ appender.activateOptions();
+ fail("Calling activateOptions should have triggered a RejectedExecutionException");
+ // this ensures that the activateOptions correctly attempts to schedule a worker
+ }
+ }
+
+ @Test
+ public void testExecutorService() throws InterruptedException, ExecutionException {
+ ScheduledExecutorService executorService = null;
+ AtomicLong counter = new AtomicLong(2);
+ try (AccumuloMonitorAppender appender = new AccumuloMonitorAppender()) {
+ executorService = appender.executorService;
+
+ // make sure executor service is started and running
+ Assert.assertEquals(false, executorService.isShutdown());
+ Assert.assertEquals(false, executorService.isTerminated());
+
+ // make sure executor service executes tasks
+ ScheduledFuture<Long> future = executorService.schedule(() -> counter.getAndIncrement(), 1, TimeUnit.MILLISECONDS);
+ Assert.assertEquals(Long.valueOf(2), future.get());
+ Assert.assertEquals(3, counter.get());
+
+ // schedule a task that won't finish
+ executorService.schedule(() -> counter.getAndIncrement(), 1, TimeUnit.DAYS);
+
+ // make sure executor service is still running
+ Assert.assertEquals(false, executorService.isShutdown());
+ Assert.assertEquals(false, executorService.isTerminated());
+ }
+ // verify that closing the appender shuts down the executor service threads
+ Assert.assertEquals(true, executorService.isShutdown());
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ Assert.assertEquals(true, executorService.isTerminated());
+
+ // verify executor service did not wait for scheduled task to run
+ Assert.assertEquals(3, counter.get());
+ }
+
+ @Test
+ public void testMonitorTracker() throws InterruptedException {
+ AtomicLong currentLoc = new AtomicLong(0);
+ Supplier<MonitorLocation> locSupplier = () -> {
+ long loc = currentLoc.get();
+ // for simplicity, create the location name from a number (0 represents no location)
+ byte[] location = loc == 0 ? null : ("loc" + loc).getBytes(UTF_8);
+ return new MonitorLocation(loc, location);
+ };
+ Function<MonitorLocation,AppenderSkeleton> appenderFactory = newLocation -> new AppenderSkeleton() {
+
+ {
+ this.name = "Appender for " + newLocation.getLocation();
+ }
+
+ @Override
+ public boolean requiresLayout() {
+ return false;
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ protected void append(LoggingEvent event) {}
+
+ };
+
+ try (AccumuloMonitorAppender parent = new AccumuloMonitorAppender()) {
+ parent.setFrequency(1); // make it check frequently (every 1 millisecond)
+ parent.setTracker(new MonitorTracker(parent, locSupplier, appenderFactory));
+ parent.activateOptions();
+
+ // initially there are no appenders
+ Assert.assertTrue(parent.getAllAppenders() == null);
+ updateLocAndVerify(currentLoc, parent, 0);
+ updateLocAndVerify(currentLoc, parent, 10);
+
+ // verify it's the same after a few times
+ // this verifies the logic in the tracker's run method which compares current location with last to see if a change occurred
+ AppenderSkeleton lastAppender = (AppenderSkeleton) parent.getAllAppenders().nextElement();
+ for (int x = 0; x < 10; x++) {
+ Thread.sleep(10);
+ AppenderSkeleton currentAppender = (AppenderSkeleton) parent.getAllAppenders().nextElement();
+ Assert.assertSame(lastAppender, currentAppender);
+ }
+
+ updateLocAndVerify(currentLoc, parent, 3);
+ updateLocAndVerify(currentLoc, parent, 0);
+ updateLocAndVerify(currentLoc, parent, 0);
+ updateLocAndVerify(currentLoc, parent, 12);
+ updateLocAndVerify(currentLoc, parent, 0);
+ updateLocAndVerify(currentLoc, parent, 335);
+
+ updateLocAndVerify(currentLoc, parent, 0);
+ // verify we removed all the appenders
+ Assert.assertFalse(parent.getAllAppenders().hasMoreElements());
+ }
+ }
+
+ private static void updateLocAndVerify(AtomicLong currentLoc, AccumuloMonitorAppender parent, int newLoc) {
+ // set the new location
+ currentLoc.set(newLoc);
+ // wait for the appender to notice the change
+ while (!verifyAppender(parent, newLoc == 0 ? null : ("loc" + newLoc))) {}
+ }
+
+ private static boolean verifyAppender(AccumuloMonitorAppender parent, String newLocName) {
+ Enumeration<?> childAppenders = parent.getAllAppenders();
+ if (newLocName == null) {
+ return childAppenders == null || !childAppenders.hasMoreElements();
+ }
+ if (childAppenders == null || !childAppenders.hasMoreElements()) {
+ return false;
+ }
+ AppenderSkeleton child = (AppenderSkeleton) childAppenders.nextElement();
+ if (childAppenders.hasMoreElements()) {
+ Assert.fail("Appender should never have more than one child");
+ }
+ return ("Appender for " + newLocName).equals(child.getName());
+ }
+
+}