Your output is working as expected as the default logger sink will truncate the body content to 16 bytes. I don't believe you can override this behavior without creating your own custom LoggerSink as the current LoggerSink does not have any configuration parameters. I modified the existing LoggerSink below and named it AdvancedLoggerSink (a bit of a misnomer as it's not all that advanced).
The advanced logger sink adds a configuration parameter called maxBytes
which you can use to set how much of your log message gets output. The default is still 16 bytes but you can now overwrite this with whatever you want. If you set it to 0 then it will print the entire log message.
To get this working you need to download the flume binaries and then create a JAR file with the AdvancedLoggerSink class in it. You'll need to include the following flume jars which are located in the lib directory of the flume binary download when you compile and create your jar file:
- flume-ng-configuration-1.4.0.jar
- flume-ng-core-1.4.0.jar
- flume-ng-sdk-1.4.0.jar
- slf4j-api-1.6.1.jar
Assuming you create a jar file called advancedLoggerSink.jar
you would then place that into your flume plugin directory inside a directory called lib. The plugins directory defaults to $FLUME_HOME/plugins.d
but you can create it anywhere. Your directory structure should looks like this:
plugins.d/advanced-logger-sink/lib/advancedLoggerSink.jar
(Make sure you put the jar inside a directory called 'lib'. See the flume user guide for more info on the plugin directory layout http://flume.apache.org/FlumeUserGuide.html)
To run the flume agent use the following command:
flume-ng agent --plugins-path /path/to/your/plugins.d --conf /conf/directory --conf-file /conf/logger.flume --name a1 -Dflume.root.logger=INFO,console
Notice how I specified the plugins-path (the path where the plugins.d directory exists). Flume will automatically load the advancedLoggerSink inside the plugins.d directory.
Here's the AdvancedLoggerSink class:
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventHelper;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class AdvancedLoggerSink extends AbstractSink implements Configurable {
private static final int defaultMaxBytes = 16;
private int maxBytesProp;
private static final Logger logger = LoggerFactory
.getLogger(AdvancedLoggerSink.class);
@Override
public void configure(Context context) {
// maxBytes of 0 means to log the entire event
int maxBytesProp = context.getInteger("maxBytes", defaultMaxBytes);
if (maxBytesProp < 0) {
maxBytesProp = defaultMaxBytes;
}
this.maxBytesProp = maxBytesProp;
}
@Override
public Status process() throws EventDeliveryException {
Status result = Status.READY;
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
Event event = null;
try {
transaction.begin();
event = channel.take();
if (event != null) {
if (logger.isInfoEnabled()) {
logger.info("Event: " + EventHelper.dumpEvent(
event,
this.maxBytesProp == 0 ? event.getBody().length : this.maxBytesProp
));
}
} else {
// No event found, request back-off semantics from the sink
// runner
result = Status.BACKOFF;
}
transaction.commit();
} catch (Exception ex) {
transaction.rollback();
throw new EventDeliveryException("Failed to log event: " + event,
ex);
} finally {
transaction.close();
}
return result;
}
}
Your config file should then look like:
# example.conf: A single-node Flume configuration
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = AdvancedLoggerSink
# maxBytes is the maximum number of bytes to output for the body of the event
# the default is 16 bytes. If you set maxBytes to 0 then the entire record will
# be output.
a1.sinks.k1.maxBytes = 0
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1