0

Here is my Test config (used netcat+ logger as console)

\#START OF CONFIG FILE

\#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 = 4444

\# Describe the sink
a1.sinks.k1.type = logger

\#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

\#====END OF CONFIG FILE

Now I issued the following command to use my specific config:

$bin/flume-ng agent --conf conf --conf-file conf/netcat_dump.conf  --name a1 -Dflume.root.logger=DEBUG,console

Use netcat command and enter the following text:

$netcat localhost 4444 

This is First Event being sent to flume through netcat

Now, If you look at the Flume Console, you see the truncated log line.

2013-11-25 15:33:20,862  ---- Event: { headers:{} body: 54 68 69 73 20 69 73 20 46 69 72 73 
74 20 45 76 **This is First Ev** }
2013-11-25 15:33:20,862  ---- Events processed = 1

Note: I'd tried with most of the channel parameters but didn't help.

1
  • If anyone looking at this thread, please kindly try with the first demo sample listed in the flume site [link] (flume.apache.org/FlumeUserGuide.html) and share you views. Try inputting a big string and check the flume console(I'm sure you'd face the same issue i.e truncating chars)
    – Avis
    Nov 26, 2013 at 9:54

2 Answers 2

1

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
1

The above answer written by Sarus works except change the a1.sinks.k1.type to the whole class path name which should include the package name. In addition, for Flume 1.6.0, copy your compiled jar to the lib folder under the installed flume path. You can also use System.out.pritnln instead of using log. Something like below

if(event!=null){
          System.out.println(EventHelper.dumpEvent(event,event.getBody().length));
          status = Status.READY; 
      }else{
          System.out.println("Event is null");
          status = Status.BACKOFF; 
      }

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.