From 85880ce841fd1612c69fee2bf57fce70dfd7b4b8 Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Wed, 12 Jan 2011 16:18:20 -0800
Subject: [PATCH] FLUME-471: Add metrics to batching/unbatching/gzip/gunzip decorators

---
 .../flume/handlers/batch/BatchingDecorator.java    |   16 +++-
 .../flume/handlers/batch/GunzipDecorator.java      |   46 +++++++++++-
 .../flume/handlers/batch/GzipDecorator.java        |   26 ++++++-
 .../flume/handlers/batch/UnbatchingDecorator.java  |   22 ++++++
 .../flume/handlers/hdfs/WriteableEvent.java        |    6 +-
 .../flume/handlers/batch/TestBatching.java         |   80 ++++++++++++++++----
 6 files changed, 173 insertions(+), 23 deletions(-)

diff --git a/src/java/com/cloudera/flume/handlers/batch/BatchingDecorator.java b/src/java/com/cloudera/flume/handlers/batch/BatchingDecorator.java
index f363b80..517242c 100644
--- a/src/java/com/cloudera/flume/handlers/batch/BatchingDecorator.java
+++ b/src/java/com/cloudera/flume/handlers/batch/BatchingDecorator.java
@@ -102,12 +102,22 @@ public class BatchingDecorator<S extends EventSink> extends
     return e.get(BATCH_SIZE) != null && e.get(BATCH_DATA) != null;
   }
 
+  /**
+   * This is a method setup seems trivial but is used in testing.
+   */
+  protected void endBatchTimeout() throws IOException, InterruptedException {
+    endBatch();
+    timeoutCount.incrementAndGet();
+
+  }
+
   protected synchronized void endBatch() throws IOException,
       InterruptedException {
     if (events.size() > 0) {
       Event be = batchevent(events);
       super.append(be);
       events.clear();
+    } else {
       emptyCount.incrementAndGet();
     }
     lastBatchTime = Clock.unixTime();
@@ -135,7 +145,8 @@ public class BatchingDecorator<S extends EventSink> extends
           LOG.warn("Batch timeout thread did not exit in a timely fashion");
         }
       } catch (InterruptedException e) {
-        LOG.warn("Interrupted while waiting for batch timeout thread to finish");
+        LOG
+            .warn("Interrupted while waiting for batch timeout thread to finish");
       }
     }
 
@@ -170,8 +181,7 @@ public class BatchingDecorator<S extends EventSink> extends
           continue;
         }
         try {
-          endBatch();
-          timeoutCount.incrementAndGet();
+          endBatchTimeout();
         } catch (IOException e) {
           LOG.error("IOException when ending batch!", e);
           timeoutThreadDone = true;
diff --git a/src/java/com/cloudera/flume/handlers/batch/GunzipDecorator.java b/src/java/com/cloudera/flume/handlers/batch/GunzipDecorator.java
index 3df379b..e48283f 100644
--- a/src/java/com/cloudera/flume/handlers/batch/GunzipDecorator.java
+++ b/src/java/com/cloudera/flume/handlers/batch/GunzipDecorator.java
@@ -20,6 +20,8 @@ package com.cloudera.flume.handlers.batch;
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.zip.GZIPInputStream;
 
 import com.cloudera.flume.conf.Context;
@@ -28,12 +30,24 @@ import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.core.EventSinkDecorator;
 import com.cloudera.flume.handlers.hdfs.WriteableEvent;
+import com.cloudera.flume.reporter.ReportEvent;
 import com.google.common.base.Preconditions;
 
 /**
  * This gunzip's any event that is gzip'ed, otherwise events just pass through.
  */
 public class GunzipDecorator<S extends EventSink> extends EventSinkDecorator<S> {
+
+  public static final String R_PASSTHROUGH = "passthroughCount";
+  public static final String R_GZIPCOUNT = "gzippedCount";
+  public static final String R_GZIPSIZE = "gzippedSize";
+  public static final String R_GUNZIPSIZE = "gunzippedSize";
+
+  AtomicLong passthrough = new AtomicLong(0);
+  AtomicLong gzipCnt = new AtomicLong(0);
+  AtomicLong gzipSize = new AtomicLong(0);
+  AtomicLong gunzipSize = new AtomicLong(0);
+
   public GunzipDecorator(S s) {
     super(s);
   }
@@ -44,23 +58,53 @@ public class GunzipDecorator<S extends EventSink> extends EventSinkDecorator<S>
     return e.get(GZDOC) != null;
   }
 
+  /**
+   * This is a terrible hack to expose the number bytes that the compressed
+   * input stream got decompressed to.
+   */
+  static class GZIPInputStreamExposeSize extends GZIPInputStream {
+    GZIPInputStreamExposeSize(InputStream in) throws IOException {
+      super(in);
+    }
+
+    long getDecompressSize() {
+      // inf is protected by GZIPInputStream
+      return inf.getBytesWritten();
+    }
+  }
+
   @Override
   public void append(Event e) throws IOException, InterruptedException {
 
     byte[] bs = e.get(GZDOC);
     if (bs == null) {
       super.append(e);
+      passthrough.incrementAndGet();
       return;
     }
 
+    gzipCnt.incrementAndGet();
+    gzipSize.addAndGet(bs.length);
     ByteArrayInputStream bais = new ByteArrayInputStream(bs);
-    GZIPInputStream gzis = new GZIPInputStream(bais);
+    GZIPInputStreamExposeSize gzis = new GZIPInputStreamExposeSize(bais);
     DataInputStream dis = new DataInputStream(gzis);
 
     WriteableEvent out = new WriteableEvent();
     out.readFields(dis);
+    long gunSz = gzis.getDecompressSize(); // must be done before closed.
     dis.close();
     super.append(out);
+    gunzipSize.addAndGet(gunSz);
+  }
+
+  @Override
+  public ReportEvent getMetrics() {
+    ReportEvent rpt = super.getMetrics();
+    rpt.setLongMetric(R_PASSTHROUGH, passthrough.get());
+    rpt.setLongMetric(R_GZIPCOUNT, gzipCnt.get());
+    rpt.setLongMetric(R_GZIPSIZE, gzipSize.get());
+    rpt.setLongMetric(R_GUNZIPSIZE, gunzipSize.get());
+    return rpt;
   }
 
   public static SinkDecoBuilder builder() {
diff --git a/src/java/com/cloudera/flume/handlers/batch/GzipDecorator.java b/src/java/com/cloudera/flume/handlers/batch/GzipDecorator.java
index bbd0451..c9ef4b1 100644
--- a/src/java/com/cloudera/flume/handlers/batch/GzipDecorator.java
+++ b/src/java/com/cloudera/flume/handlers/batch/GzipDecorator.java
@@ -19,6 +19,7 @@ package com.cloudera.flume.handlers.batch;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.zip.GZIPOutputStream;
 
 import com.cloudera.flume.conf.Context;
@@ -28,12 +29,22 @@ import com.cloudera.flume.core.EventImpl;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.core.EventSinkDecorator;
 import com.cloudera.flume.handlers.hdfs.WriteableEvent;
+import com.cloudera.flume.reporter.ReportEvent;
 import com.google.common.base.Preconditions;
 
 /**
  * This gzips each event as it passes through the decorator.
  */
 public class GzipDecorator<S extends EventSink> extends EventSinkDecorator<S> {
+
+  public static final String R_EVENTCOUNT = "eventsCount";
+  public static final String R_EVENTSIZE = "eventsSize";
+  public static final String R_GZIPSIZE = "gzippedSize";
+
+  AtomicLong eventCount = new AtomicLong(0);
+  AtomicLong eventSize = new AtomicLong(0);
+  AtomicLong gzipSize = new AtomicLong(0);
+
   public GzipDecorator(S s) {
     super(s);
   }
@@ -42,6 +53,7 @@ public class GzipDecorator<S extends EventSink> extends EventSinkDecorator<S> {
   public void append(Event e) throws IOException, InterruptedException {
     WriteableEvent we = new WriteableEvent(e);
     byte[] bs = we.toBytes();
+    eventSize.addAndGet(bs.length);
 
     ByteArrayOutputStream baos = new ByteArrayOutputStream();
     GZIPOutputStream gzos = new GZIPOutputStream(baos);
@@ -49,8 +61,20 @@ public class GzipDecorator<S extends EventSink> extends EventSinkDecorator<S> {
     gzos.close();
 
     Event gze = new EventImpl(new byte[0]);
-    gze.set(GunzipDecorator.GZDOC, baos.toByteArray());
+    byte[] compressed = baos.toByteArray();
+    gze.set(GunzipDecorator.GZDOC, compressed);
     super.append(gze);
+    gzipSize.addAndGet(compressed.length);
+    eventCount.incrementAndGet();
+  }
+
+  @Override
+  public ReportEvent getMetrics() {
+    ReportEvent rpt = super.getMetrics();
+    rpt.setLongMetric(R_EVENTCOUNT, eventCount.get());
+    rpt.setLongMetric(R_EVENTSIZE, eventSize.get());
+    rpt.setLongMetric(R_GZIPSIZE, gzipSize.get());
+    return rpt;
   }
 
   public static SinkDecoBuilder builder() {
diff --git a/src/java/com/cloudera/flume/handlers/batch/UnbatchingDecorator.java b/src/java/com/cloudera/flume/handlers/batch/UnbatchingDecorator.java
index 9f0ba62..b4b2605 100644
--- a/src/java/com/cloudera/flume/handlers/batch/UnbatchingDecorator.java
+++ b/src/java/com/cloudera/flume/handlers/batch/UnbatchingDecorator.java
@@ -22,6 +22,7 @@ import java.io.DataInput;
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicLong;
 
 import com.cloudera.flume.conf.Context;
 import com.cloudera.flume.conf.SinkFactory.SinkDecoBuilder;
@@ -29,6 +30,7 @@ import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventSink;
 import com.cloudera.flume.core.EventSinkDecorator;
 import com.cloudera.flume.handlers.hdfs.WriteableEvent;
+import com.cloudera.flume.reporter.ReportEvent;
 import com.google.common.base.Preconditions;
 
 /**
@@ -38,6 +40,14 @@ import com.google.common.base.Preconditions;
 public class UnbatchingDecorator<S extends EventSink> extends
     EventSinkDecorator<S> {
 
+  public static final String R_PASSTHROUGH = "passthroughCount";
+  public static final String R_BATCHED_IN = "batchCount";
+  public static final String R_BATCHED_OUT = "unbatchedCount";
+
+  AtomicLong passthrough = new AtomicLong(0);
+  AtomicLong batchCnt = new AtomicLong(0);
+  AtomicLong unbatchedCnt = new AtomicLong(0);
+
   public UnbatchingDecorator(S s) {
     super(s);
   }
@@ -50,19 +60,31 @@ public class UnbatchingDecorator<S extends EventSink> extends
   public void append(Event e) throws IOException, InterruptedException {
     if (!BatchingDecorator.isBatch(e)) {
       super.append(e);
+      passthrough.incrementAndGet();
       return;
     }
 
     int sz = ByteBuffer.wrap(e.get(BatchingDecorator.BATCH_SIZE)).getInt();
     byte[] data = e.get(BatchingDecorator.BATCH_DATA);
     DataInput in = new DataInputStream(new ByteArrayInputStream(data));
+    batchCnt.incrementAndGet();
     for (int i = 0; i < sz; i++) {
       WriteableEvent we = new WriteableEvent();
       we.readFields(in);
       super.append(we);
+      unbatchedCnt.incrementAndGet();
     }
   }
 
+  @Override
+  public ReportEvent getMetrics() {
+    ReportEvent rpt = super.getMetrics();
+    rpt.setLongMetric(R_PASSTHROUGH, passthrough.get());
+    rpt.setLongMetric(R_BATCHED_IN, batchCnt.get());
+    rpt.setLongMetric(R_BATCHED_OUT, unbatchedCnt.get());
+    return rpt;
+  }
+
   public static SinkDecoBuilder builder() {
     return new SinkDecoBuilder() {
       @Override
diff --git a/src/java/com/cloudera/flume/handlers/hdfs/WriteableEvent.java b/src/java/com/cloudera/flume/handlers/hdfs/WriteableEvent.java
index b484ae3..08ed666 100644
--- a/src/java/com/cloudera/flume/handlers/hdfs/WriteableEvent.java
+++ b/src/java/com/cloudera/flume/handlers/hdfs/WriteableEvent.java
@@ -115,15 +115,15 @@ public class WriteableEvent extends EventBaseImpl implements Writable {
     // TODO (jon) Compare to java.nio implementation
     byte[] body = new byte[len];
     in.readFully(body);
-
+    
     long time = in.readLong();
-
+    
     int prioidx = in.readInt();
     assert (Priority.values().length > prioidx);
     Priority prio = Priority.values()[prioidx];
 
     long nanos = in.readLong();
-
+    
     String host = in.readUTF();
 
     Map<String, byte[]> fields = unserializeMap(in);
diff --git a/src/javatest/com/cloudera/flume/handlers/batch/TestBatching.java b/src/javatest/com/cloudera/flume/handlers/batch/TestBatching.java
index 4d8ea2f..27f9995 100644
--- a/src/javatest/com/cloudera/flume/handlers/batch/TestBatching.java
+++ b/src/javatest/com/cloudera/flume/handlers/batch/TestBatching.java
@@ -32,12 +32,14 @@ import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventImpl;
 import com.cloudera.flume.core.EventSink;
+import com.cloudera.flume.core.EventSource;
+import com.cloudera.flume.core.EventUtil;
 import com.cloudera.flume.core.FanOutSink;
+import com.cloudera.flume.handlers.debug.ConsoleEventSink;
 import com.cloudera.flume.handlers.debug.MemorySinkSource;
 import com.cloudera.flume.handlers.hdfs.WriteableEvent;
 import com.cloudera.flume.reporter.ReportEvent;
 import com.cloudera.flume.reporter.aggregator.CounterSink;
-import com.cloudera.util.Clock;
 
 /**
  * This tests batching/unbatching and gzip/gunzip compression
@@ -128,7 +130,7 @@ public class TestBatching {
       InterruptedException {
 
     MemorySinkSource mem = new MemorySinkSource();
-    BatchingDecorator<EventSink> b = new BatchingDecorator<EventSink>(mem, 100,
+    BatchingDecorator<EventSink> b = new BatchingDecorator<EventSink>(mem, 10,
         0);
     b.open();
     for (int i = 0; i < 100; i++) {
@@ -156,8 +158,8 @@ public class TestBatching {
     LOG.info(String.format("before: %d  gzip: %d  gunzip: %d", origsz, gzipsz,
         ungzsz));
 
-    Assert.assertTrue(origsz > gzipsz); // got some benefit for compressing?
-    Assert.assertEquals(origsz, ungzsz); // uncompress is same size as
+    assertTrue(origsz > gzipsz); // got some benefit for compressing?
+    assertEquals(origsz, ungzsz); // uncompress is same size as
     // precompressed?
 
   }
@@ -173,24 +175,72 @@ public class TestBatching {
   @Test
   public void testEmptyBatches() throws FlumeSpecException, IOException,
       InterruptedException {
-    EventSink snk = FlumeBuilder.buildSink(new Context(),
-        "{ batch(2,100) => console }");
+    BatchingDecorator<EventSink> snk = new BatchingDecorator<EventSink>(
+        new ConsoleEventSink(), 2, 100000);
     snk.open();
 
     for (int i = 0; i < 10; i++) {
-      Clock.sleep(1000);
-      snk.append(new EventImpl(("test " + i).getBytes()));
+      if (i % 4 == 0) {
+        snk.append(new EventImpl(("test " + i).getBytes()));
+
+      }
+      snk.endBatchTimeout();
     }
     snk.close();
 
     ReportEvent rpt = snk.getMetrics();
     LOG.info(rpt.toString());
-    assertEquals(Long.valueOf(0), rpt.getLongMetric(BatchingDecorator.R_FILLED));
-    assertEquals(Long.valueOf(10), rpt.getLongMetric(BatchingDecorator.R_EMPTY));
-    // this is timing based and there is a little play with these numbers.
-    assertTrue(rpt.getLongMetric(BatchingDecorator.R_TRIGGERS) > 97);
-    assertTrue(rpt.getLongMetric(BatchingDecorator.R_TRIGGERS) < 102);
-    assertTrue(rpt.getLongMetric(BatchingDecorator.R_TIMEOUTS) > 97);
-    assertTrue(rpt.getLongMetric(BatchingDecorator.R_TIMEOUTS) < 102);
+    assertEquals(0, (long) rpt.getLongMetric(BatchingDecorator.R_FILLED));
+    assertEquals(8, (long) rpt.getLongMetric(BatchingDecorator.R_EMPTY));
+    // Extra trigger happens on close
+    assertEquals(11, (long) rpt.getLongMetric(BatchingDecorator.R_TRIGGERS));
+  }
+
+  @Test
+  public void testBatchingMetrics() throws IOException, InterruptedException {
+    EventSource src = MemorySinkSource.cannedData("this is a data", 200);
+    MemorySinkSource mem = new MemorySinkSource();
+    UnbatchingDecorator<EventSink> unbatch = new UnbatchingDecorator<EventSink>(
+        mem);
+    GunzipDecorator<EventSink> gunzip = new GunzipDecorator<EventSink>(unbatch);
+
+    int evts = 5;
+    int latency = 0; // 0 never triggers on time
+    GzipDecorator<EventSink> gzip = new GzipDecorator<EventSink>(gunzip);
+    BatchingDecorator<EventSink> batch = new BatchingDecorator<EventSink>(gzip,
+        evts, latency);
+
+    src.open();
+    batch.open();
+    EventUtil.dumpAll(src, batch);
+    src.close();
+    batch.close();
+
+    // check metrics.
+    ReportEvent ubRpt = unbatch.getMetrics();
+    assertEquals(40, (long) ubRpt
+        .getLongMetric(UnbatchingDecorator.R_BATCHED_IN));
+    assertEquals(200, (long) ubRpt
+        .getLongMetric(UnbatchingDecorator.R_BATCHED_OUT));
+    assertEquals(0, (long) ubRpt
+        .getLongMetric(UnbatchingDecorator.R_PASSTHROUGH));
+
+    ReportEvent bRpt = batch.getMetrics();
+    assertEquals(1, (long) bRpt.getLongMetric(BatchingDecorator.R_EMPTY));
+    assertEquals(40, (long) bRpt.getLongMetric(BatchingDecorator.R_FILLED));
+    assertEquals(0, (long) bRpt.getLongMetric(BatchingDecorator.R_TIMEOUTS));
+    assertEquals(41, (long) bRpt.getLongMetric(BatchingDecorator.R_TRIGGERS));
+
+    ReportEvent gzr = gzip.getMetrics();
+    assertTrue(gzr.getLongMetric(GzipDecorator.R_GZIPSIZE) < gzr
+        .getLongMetric(GzipDecorator.R_EVENTSIZE));
+    assertEquals(40, (long) gzr.getLongMetric(GzipDecorator.R_EVENTCOUNT));
+
+    ReportEvent ugzr = gunzip.getMetrics();
+    assertTrue(ugzr.getLongMetric(GunzipDecorator.R_GZIPSIZE) < ugzr
+        .getLongMetric(GunzipDecorator.R_GUNZIPSIZE));
+    assertEquals(40, (long) ugzr.getLongMetric(GunzipDecorator.R_GZIPCOUNT));
+    assertEquals(0, (long) ugzr.getLongMetric(GunzipDecorator.R_PASSTHROUGH));
   }
+
 }
-- 
1.7.0.4

