You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/12/18 23:03:01 UTC
[1/2] storm git commit: STORM-2854 Expose IEventLogger to make event
logging pluggable
Repository: storm
Updated Branches:
refs/heads/master c893c8dd8 -> 3af8dda88
STORM-2854 Expose IEventLogger to make event logging pluggable
* expose option to register IEventLogger similar to metrics consumer
* change the interface of IEventLogger slightly
* allow argument
* the change is technically not backward compatible but in real we can treat it's OK
* cause we don't provide a chance to implement custom IEventLogger and plug to topology
* Fix EventInfo to contain origin type of values instead of String-converted values
* open possibility to extend FileBasedEventLogger and provide different format of log message
* document the change
* address review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6bf3094c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6bf3094c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6bf3094c
Branch: refs/heads/master
Commit: 6bf3094c60f0af665f0129bb9f68c0ec7836b05f
Parents: b246d37
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Thu Dec 14 15:53:05 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Dec 19 08:02:38 2017 +0900
----------------------------------------------------------------------
conf/storm.yaml.example | 9 ++-
docs/Eventlogging.md | 33 ++++++++++-
.../src/jvm/org/apache/storm/Config.java | 39 ++++++++++++
.../org/apache/storm/daemon/StormCommon.java | 3 +
.../apache/storm/metric/EventLoggerBolt.java | 62 ++++++++++++++++----
.../storm/metric/FileBasedEventLogger.java | 25 ++++----
.../org/apache/storm/metric/IEventLogger.java | 41 ++++++++++---
.../storm/validation/ConfigValidation.java | 20 +++++++
8 files changed, 199 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/conf/storm.yaml.example
----------------------------------------------------------------------
diff --git a/conf/storm.yaml.example b/conf/storm.yaml.example
index f396fde..6d75963 100644
--- a/conf/storm.yaml.example
+++ b/conf/storm.yaml.example
@@ -73,4 +73,11 @@
# argument:
# - endpoint: "metrics-collector.mycompany.org"
#
-# storm.cluster.metrics.consumer.publish.interval.secs: 60
\ No newline at end of file
+# storm.cluster.metrics.consumer.publish.interval.secs: 60
+
+# Event Logger
+# topology.event.logger.register:
+# - class: "org.apache.storm.metric.FileBasedEventLogger"
+# - class: "org.mycompany.MyEventLogger"
+# arguments:
+# endpoint: "event-logger.mycompany.org"
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/docs/Eventlogging.md
----------------------------------------------------------------------
diff --git a/docs/Eventlogging.md b/docs/Eventlogging.md
index 0895232..6fbf1ed 100644
--- a/docs/Eventlogging.md
+++ b/docs/Eventlogging.md
@@ -68,7 +68,8 @@ Eventlogging works by sending the events (tuples) from each component to an inte
## Extending eventlogging
-Storm provides an `IEventLogger` interface which is used by the event logger bolt to log the events. The default implementation for this is a FileBasedEventLogger which logs the events to an events.log file ( `logs/workers-artifacts/<topology-id>/<worker-port>/events.log`). Alternate implementations of the `IEventLogger` interface can be added to extend the event logging functionality (say build a search index or log the events in a database etc)
+
+Storm provides an `IEventLogger` interface which is used by the event logger bolt to log the events.
```java
/**
@@ -79,7 +80,7 @@ public interface IEventLogger {
/**
* Invoked during eventlogger bolt prepare.
*/
- void prepare(Map stormConf, TopologyContext context);
+ void prepare(Map stormConf, Map<String, Object> arguments, TopologyContext context);
/**
* Invoked when the {@link EventLoggerBolt} receives a tuple from the spouts or bolts that has event logging enabled.
@@ -94,3 +95,31 @@ public interface IEventLogger {
void close();
}
```
+
+The default implementation for this is a FileBasedEventLogger which logs the events to an events.log file ( `logs/workers-artifacts/<topology-id>/<worker-port>/events.log`).
+Alternate implementations of the `IEventLogger` interface can be added to extend the event logging functionality (say build a search index or log the events in a database etc)
+
+If you just want to use FileBasedEventLogger but with changing the log format, you can simply implement your own by extending FileBasedEventLogger and override `buildLogMessage(EventInfo)` to provide log line explicitly.
+
+To register event logger to your topology, add to your topology's configuration like:
+
+```java
+conf.registerEventLogger(org.apache.storm.metric.FileBasedEventLogger.class);
+```
+
+You can refer [Config#registerEventLogger](javadocs/org/apache/storm/Config.html#registerEventLogger-java.lang.Class-) and overloaded methods from javadoc.
+
+Otherwise edit the storm.yaml config file:
+
+```yaml
+topology.event.logger.register:
+ - class: "org.apache.storm.metric.FileBasedEventLogger"
+ - class: "org.mycompany.MyEventLogger"
+ arguments:
+ endpoint: "event-logger.mycompany.org"
+```
+
+When you implement your own event logger, `arguments` is passed to Map<String, Object> when [IEventLogger#prepare](javadocs/org/apache/storm/metric/IEventLogger.html#prepare-java.util.Map-java.lang.Map-org.apache.storm.task.TopologyContext-) is called.
+
+Please keep in mind that EventLoggerBolt is just a kind of Bolt, so whole throughput of the topology will go down when registered event loggers cannot keep up handling incoming events, so you may want to take care of the Bolt like normal Bolt.
+One of idea to avoid this is making your implementation of IEventLogger as `non-blocking` fashion.
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index cc47e14..9a41cc6 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -17,6 +17,7 @@
*/
package org.apache.storm;
+import org.apache.storm.metric.IEventLogger;
import org.apache.storm.serialization.IKryoDecorator;
import org.apache.storm.serialization.IKryoFactory;
import org.apache.storm.validation.ConfigValidation;
@@ -379,6 +380,18 @@ public class Config extends HashMap<String, Object> {
public static final String TOPOLOGY_ACKER_EXECUTORS = "topology.acker.executors";
/**
+ * A list of classes implementing IEventLogger (See storm.yaml.example for exact config format).
+ * Each listed class will be routed all the events sampled from emitting tuples.
+ * If there's no class provided to the option, default event logger will be initialized and used
+ * unless you disable event logger executor.
+ *
+ * Note that EventLoggerBolt takes care of all the implementations of IEventLogger, hence registering
+ * many implementations (especially they're implemented as 'blocking' manner) would slow down overall topology.
+ */
+ @isListEntryCustom(entryValidatorClasses={EventLoggerRegistryValidator.class})
+ public static final String TOPOLOGY_EVENT_LOGGER_REGISTER = "topology.event.logger.register";
+
+ /**
* How many executors to spawn for event logger.
*
* <p>By not setting this variable or setting it as null, Storm will set the number of eventlogger executors
@@ -1788,6 +1801,32 @@ public class Config extends HashMap<String, Object> {
registerSerialization(this, klass, serializerClass);
}
+ public void registerEventLogger(Class<? extends IEventLogger> klass, Map<String, Object> argument) {
+ registerEventLogger(this, klass, argument);
+ }
+
+ public void registerEventLogger(Class<? extends IEventLogger> klass) {
+ registerEventLogger(this, klass, null);
+ }
+
+ public static void registerEventLogger(Map<String, Object> conf, Class<? extends IEventLogger> klass, Map<String, Object> argument) {
+ Map<String, Object> m = new HashMap<>();
+ m.put("class", klass.getCanonicalName());
+ m.put("arguments", argument);
+
+ List<Map<String, Object>> l = (List<Map<String, Object>>)conf.get(TOPOLOGY_EVENT_LOGGER_REGISTER);
+ if (l == null) {
+ l = new ArrayList<>();
+ }
+ l.add(m);
+
+ conf.put(TOPOLOGY_EVENT_LOGGER_REGISTER, l);
+ }
+
+ public static void registerEventLogger(Map<String, Object> conf, Class<? extends IEventLogger> klass) {
+ registerEventLogger(conf, klass, null);
+ }
+
public static void registerMetricsConsumer(Map<String, Object> conf, Class klass, Object argument, long parallelismHint) {
HashMap m = new HashMap();
m.put("class", klass.getCanonicalName());
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java b/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
index 617e7ae..f6053b9 100644
--- a/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
+++ b/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
@@ -94,6 +94,9 @@ public class StormCommon {
public static final String TOPOLOGY_METRICS_CONSUMER_EXPAND_MAP_TYPE = "expandMapType";
public static final String TOPOLOGY_METRICS_CONSUMER_METRIC_NAME_SEPARATOR = "metricNameSeparator";
+ public static final String TOPOLOGY_EVENT_LOGGER_CLASS = "class";
+ public static final String TOPOLOGY_EVENT_LOGGER_ARGUMENTS = "arguments";
+
@Deprecated
public static String getStormId(final IStormClusterState stormClusterState, final String topologyName) {
return stormClusterState.getTopoId(topologyName).get();
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java b/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
index 901375d..4a1549f 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
@@ -15,8 +15,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.metric;
+import static org.apache.storm.daemon.StormCommon.TOPOLOGY_EVENT_LOGGER_ARGUMENTS;
+import static org.apache.storm.daemon.StormCommon.TOPOLOGY_EVENT_LOGGER_CLASS;
+import static org.apache.storm.metric.IEventLogger.EventInfo;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.storm.Config;
import org.apache.storm.task.IBolt;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
@@ -24,9 +34,6 @@ import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Map;
-import static org.apache.storm.metric.IEventLogger.EventInfo;
-
public class EventLoggerBolt implements IBolt {
private static final Logger LOG = LoggerFactory.getLogger(EventLoggerBolt.class);
@@ -39,13 +46,19 @@ public class EventLoggerBolt implements IBolt {
public static final String FIELD_COMPONENT_ID = "component-id";
public static final String FIELD_MESSAGE_ID = "message-id";
- private IEventLogger eventLogger;
+ private List<IEventLogger> eventLoggers;
@Override
public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
LOG.info("EventLoggerBolt prepare called");
- eventLogger = new FileBasedEventLogger();
- eventLogger.prepare(topoConf, context);
+
+ eventLoggers = new ArrayList<>();
+ List<Map<String, Object>> registerInfo = (List<Map<String, Object>>) topoConf.get(Config.TOPOLOGY_EVENT_LOGGER_REGISTER);
+ if (registerInfo != null && !registerInfo.isEmpty()) {
+ initializeEventLoggers(topoConf, context, registerInfo);
+ } else {
+ initializeDefaultEventLogger(topoConf, context);
+ }
}
@Override
@@ -53,15 +66,42 @@ public class EventLoggerBolt implements IBolt {
LOG.debug("** EventLoggerBolt got tuple from sourceComponent {}, with values {}", input.getSourceComponent(), input.getValues());
Object msgId = input.getValueByField(FIELD_MESSAGE_ID);
- EventInfo eventInfo = new EventInfo(input.getValueByField(FIELD_TS).toString(), input.getSourceComponent(),
- String.valueOf(input.getSourceTask()), msgId == null ? "" : msgId.toString(),
- input.getValueByField(FIELD_VALUES).toString());
+ EventInfo eventInfo = new EventInfo(input.getLongByField(FIELD_TS), input.getSourceComponent(),
+ input.getSourceTask(), msgId, (List<Object>) input.getValueByField(FIELD_VALUES));
- eventLogger.log(eventInfo);
+ for (IEventLogger eventLogger : eventLoggers) {
+ eventLogger.log(eventInfo);
+ }
}
@Override
public void cleanup() {
- eventLogger.close();
+ for (IEventLogger eventLogger : eventLoggers) {
+ eventLogger.close();
+ }
+ }
+
+ private void initializeEventLoggers(Map<String, Object> topoConf, TopologyContext context, List<Map<String, Object>> registerInfo) {
+ for (Map<String, Object> info : registerInfo) {
+ String className = (String) info.get(TOPOLOGY_EVENT_LOGGER_CLASS);
+ Map<String, Object> arguments = (Map<String, Object>) info.get(TOPOLOGY_EVENT_LOGGER_ARGUMENTS);
+
+ IEventLogger eventLogger;
+ try {
+ eventLogger = (IEventLogger) Class.forName(className).newInstance();
+ } catch (Exception e) {
+ throw new RuntimeException("Could not instantiate a class listed in config under section "
+ + Config.TOPOLOGY_EVENT_LOGGER_REGISTER + " with fully qualified name " + className, e);
+ }
+
+ eventLogger.prepare(topoConf, arguments, context);
+ eventLoggers.add(eventLogger);
+ }
+ }
+
+ private void initializeDefaultEventLogger(Map<String, Object> topoConf, TopologyContext context) {
+ FileBasedEventLogger eventLogger = new FileBasedEventLogger();
+ eventLogger.prepare(topoConf, null, context);
+ eventLoggers.add(eventLogger);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
index 8f5632a..3d99685 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
@@ -15,12 +15,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.storm.metric;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.utils.ConfigUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+package org.apache.storm.metric;
import java.io.BufferedWriter;
import java.io.File;
@@ -35,6 +31,11 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.utils.ConfigUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
public class FileBasedEventLogger implements IEventLogger {
private static final Logger LOG = LoggerFactory.getLogger(FileBasedEventLogger.class);
@@ -63,7 +64,7 @@ public class FileBasedEventLogger implements IEventLogger {
@Override
public void run() {
try {
- if(dirty) {
+ if (dirty) {
eventLogWriter.flush();
dirty = false;
}
@@ -79,7 +80,7 @@ public class FileBasedEventLogger implements IEventLogger {
@Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context) {
+ public void prepare(Map<String, Object> conf, Map<String, Object> arguments, TopologyContext context) {
String stormId = context.getStormId();
int port = context.getThisWorkerPort();
@@ -87,12 +88,12 @@ public class FileBasedEventLogger implements IEventLogger {
* Include the topology name & worker port in the file name so that
* multiple event loggers can log independently.
*/
- String workersArtifactRoot = ConfigUtils.workerArtifactsRoot(topoConf, stormId, port);
+ String workersArtifactRoot = ConfigUtils.workerArtifactsRoot(conf, stormId, port);
Path path = Paths.get(workersArtifactRoot, "events.log");
File dir = path.toFile().getParentFile();
if (!dir.exists()) {
- dir.mkdirs();
+ dir.mkdirs();
}
initLogWriter(path);
setUpFlushTask();
@@ -102,7 +103,7 @@ public class FileBasedEventLogger implements IEventLogger {
public void log(EventInfo event) {
try {
//TODO: file rotation
- eventLogWriter.write(event.toString());
+ eventLogWriter.write(buildLogMessage(event));
eventLogWriter.newLine();
dirty = true;
} catch (IOException ex) {
@@ -111,6 +112,10 @@ public class FileBasedEventLogger implements IEventLogger {
}
}
+ protected String buildLogMessage(EventInfo event) {
+ return event.toString();
+ }
+
@Override
public void close() {
try {
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
index 70c5a8d..e310d16 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
@@ -20,6 +20,7 @@ package org.apache.storm.metric;
import org.apache.storm.task.TopologyContext;
import java.util.Date;
+import java.util.List;
import java.util.Map;
/**
@@ -31,13 +32,14 @@ public interface IEventLogger {
/**
* A wrapper for the fields that we would log.
*/
- public static class EventInfo {
- String ts;
- String component;
- String task;
- String messageId;
- String values;
- EventInfo(String ts, String component, String task, String messageId, String values) {
+ class EventInfo {
+ private long ts;
+ private String component;
+ private int task;
+ private Object messageId;
+ private List<Object> values;
+
+ public EventInfo(long ts, String component, int task, Object messageId, List<Object> values) {
this.ts = ts;
this.component = component;
this.task = task;
@@ -45,6 +47,26 @@ public interface IEventLogger {
this.values = values;
}
+ public long getTs() {
+ return ts;
+ }
+
+ public String getComponent() {
+ return component;
+ }
+
+ public int getTask() {
+ return task;
+ }
+
+ public Object getMessageId() {
+ return messageId;
+ }
+
+ public List<Object> getValues() {
+ return values;
+ }
+
/**
* Returns a default formatted string with fields separated by ","
*
@@ -52,11 +74,12 @@ public interface IEventLogger {
*/
@Override
public String toString() {
- return new Date(Long.parseLong(ts)).toString() + "," + component + "," + task + "," + messageId + "," + values;
+ return new Date(ts).toString() + "," + component + "," + String.valueOf(task) + ","
+ + (messageId == null ? "" : messageId.toString()) + "," + values.toString();
}
}
- void prepare(Map<String, Object> topoConf, TopologyContext context);
+ void prepare(Map<String, Object> conf, Map<String, Object> arguments, TopologyContext context);
/**
* This method would be invoked when the {@link EventLoggerBolt} receives a tuple from the spouts or bolts that has
http://git-wip-us.apache.org/repos/asf/storm/blob/6bf3094c/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
index c78af22..b89b603 100644
--- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
+++ b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
@@ -501,6 +501,26 @@ public class ConfigValidation {
}
}
+ public static class EventLoggerRegistryValidator extends Validator {
+
+ @Override
+ public void validateField(String name, Object o) {
+ if(o == null) {
+ return;
+ }
+ SimpleTypeValidator.validateField(name, Map.class, o);
+ if(!((Map<?, ?>) o).containsKey("class") ) {
+ throw new IllegalArgumentException( "Field " + name + " must have map entry with key: class");
+ }
+
+ SimpleTypeValidator.validateField(name, String.class, ((Map<?, ?>) o).get("class"));
+
+ if(((Map<?, ?>) o).containsKey("arguments") ) {
+ SimpleTypeValidator.validateField(name, Map.class, ((Map<?, ?>) o).get("arguments"));
+ }
+ }
+ }
+
public static class MapOfStringToMapOfStringToObjectValidator extends Validator {
@Override
public void validateField(String name, Object o) {
[2/2] storm git commit: Merge branch 'STORM-2854'
Posted by ka...@apache.org.
Merge branch 'STORM-2854'
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/3af8dda8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/3af8dda8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/3af8dda8
Branch: refs/heads/master
Commit: 3af8dda88aaff2e62be0d6aafd8bb1cbaaa07f8d
Parents: c893c8d 6bf3094
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Dec 19 08:02:55 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Dec 19 08:02:55 2017 +0900
----------------------------------------------------------------------
conf/storm.yaml.example | 9 ++-
docs/Eventlogging.md | 33 ++++++++++-
.../src/jvm/org/apache/storm/Config.java | 39 ++++++++++++
.../org/apache/storm/daemon/StormCommon.java | 3 +
.../apache/storm/metric/EventLoggerBolt.java | 62 ++++++++++++++++----
.../storm/metric/FileBasedEventLogger.java | 25 ++++----
.../org/apache/storm/metric/IEventLogger.java | 41 ++++++++++---
.../storm/validation/ConfigValidation.java | 20 +++++++
8 files changed, 199 insertions(+), 33 deletions(-)
----------------------------------------------------------------------