From 8da4ea4f1d3c573e232e43beda0bbcf2b9df80dd Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Mon, 17 Jan 2011 23:49:18 -0800
Subject: [PATCH] FLUME-433: Agents end up in ERROR mode when all collectors are killed in BE or DFO mode

---
 .../cloudera/flume/agent/AgentFailChainSink.java   |    6 +-
 .../flume/agent/diskfailover/DiskFailoverDeco.java |   18 +-
 .../cloudera/flume/core/BackOffFailOverSink.java   |   11 +-
 .../com/cloudera/flume/core/CompositeSink.java     |    4 +
 .../flume/handlers/debug/StubbornAppendSink.java   |    5 +-
 .../cloudera/util/CappedExponentialBackoff.java    |    5 +-
 .../flume/agent/TestAgentFailChainSink.java        |  270 ++++++++++++++++++--
 7 files changed, 285 insertions(+), 34 deletions(-)

diff --git a/src/java/com/cloudera/flume/agent/AgentFailChainSink.java b/src/java/com/cloudera/flume/agent/AgentFailChainSink.java
index b75e8b5..0e18d77 100644
--- a/src/java/com/cloudera/flume/agent/AgentFailChainSink.java
+++ b/src/java/com/cloudera/flume/agent/AgentFailChainSink.java
@@ -75,6 +75,7 @@ public class AgentFailChainSink extends EventSink.Base {
       break;
     }
     case BE: {
+      thriftlist.add("null");
       String chains = AgentFailChainSink.genBestEffortChain(thriftlist
           .toArray(new String[0]));
       LOG.info("Setting failover chain to  " + chains);
@@ -156,8 +157,9 @@ public class AgentFailChainSink extends EventSink.Base {
    */
   public static String genDfoChain(String... chain) {
     String primaries = genBestEffortChain(chain);
-    String body = "< " + primaries + " ? {diskFailover => { insistentOpen =>  "
-        + primaries + " } } >";
+    String body = "< " + primaries
+        + " ? diskFailover insistentAppend stubbornAppend insistentOpen "
+        + primaries + " >";
 
     LOG.info("Setting dfo failover chain to  " + body);
     return body;
diff --git a/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java b/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
index 6e5fd48..413c2c8 100644
--- a/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
+++ b/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
@@ -63,7 +63,8 @@ public class DiskFailoverDeco extends EventSinkDecorator<EventSink> {
 
   CountDownLatch drainCompleted = null; // block close until subthread is
   // completed
-  CountDownLatch drainStarted = null; // blocks open until subthread is started
+  CountDownLatch drainStarted = null; // blocks open until subthread is
+  // started
   volatile IOException lastExn = null;
 
   final long checkmillis;
@@ -251,10 +252,12 @@ public class DiskFailoverDeco extends EventSinkDecorator<EventSink> {
           checkmillis = Long.parseLong(argv[1]);
         }
 
-        // TODO (jon) this will cause problems with multiple nodes in same JVM
+        // TODO (jon) this will cause problems with multiple nodes in
+        // same JVM
         FlumeNode node = FlumeNode.getInstance();
 
-        // this makes the dfo present to the when reporting on the FlumeNode
+        // this makes the dfo present to the when reporting on the
+        // FlumeNode
         String dfonode = context.getValue(LogicalNodeContext.C_LOGICAL);
         Preconditions.checkArgument(dfonode != null,
             "Context does not have a logical node name");
@@ -282,7 +285,6 @@ public class DiskFailoverDeco extends EventSinkDecorator<EventSink> {
     Map<String, Reportable> map = new HashMap<String, Reportable>();
     map.put(sink.getName(), sink);
     map.put(dfoMan.getName(), dfoMan);
-    map.put("drainSink." + sink.getName(), sink);
     if (drainSource != null) {
       // careful, drainSource can be null if deco not opened yet
       map.put("drainSource." + drainSource.getName(), drainSource);
@@ -290,4 +292,12 @@ public class DiskFailoverDeco extends EventSinkDecorator<EventSink> {
 
     return map;
   }
+
+  public DiskFailoverManager getFailoverManager() {
+    return dfoMan;
+  }
+
+  public RollSink getDFOWriter() {
+    return input;
+  }
 }
diff --git a/src/java/com/cloudera/flume/core/BackOffFailOverSink.java b/src/java/com/cloudera/flume/core/BackOffFailOverSink.java
index 093d1da..bd814a5 100644
--- a/src/java/com/cloudera/flume/core/BackOffFailOverSink.java
+++ b/src/java/com/cloudera/flume/core/BackOffFailOverSink.java
@@ -140,11 +140,12 @@ public class BackOffFailOverSink extends EventSink.Base {
       // isreset
       return;
     } catch (IOException ioe3) {
+      LOG.info(ioe3.getMessage());
       fails.incrementAndGet();
       primaryOk = false;
       backoffPolicy.backoff();
       backup.append(e);
-      backups.incrementAndGet();
+      backups.incrementAndGet();      
       super.append(e);
     }
   }
@@ -219,6 +220,14 @@ public class BackOffFailOverSink extends EventSink.Base {
     }
   }
 
+  public EventSink getPrimary() {
+	  return primary;
+  }
+  
+  public EventSink getBackup() {
+	  return backup;
+  }
+  
   @Override
   public String getName() {
     return "BackoffFailover";
diff --git a/src/java/com/cloudera/flume/core/CompositeSink.java b/src/java/com/cloudera/flume/core/CompositeSink.java
index 584d278..704f08b 100644
--- a/src/java/com/cloudera/flume/core/CompositeSink.java
+++ b/src/java/com/cloudera/flume/core/CompositeSink.java
@@ -83,5 +83,9 @@ public class CompositeSink extends EventSink.Base {
     super.getReports(namePrefix, reports);
     snk.getReports(namePrefix + getName() + ".", reports);
   }
+  
+  public EventSink getSink() {
+	  return snk;
+  }
 
 }
diff --git a/src/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java b/src/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
index b0f4115..adfe60e 100644
--- a/src/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
+++ b/src/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
@@ -59,11 +59,11 @@ public class StubbornAppendSink<S extends EventSink> extends
   @Override
   public void open() throws IOException, InterruptedException {
     Preconditions.checkNotNull(sink);
-    LOG.info("Opening Stubborn Append Sink " + this);
+    LOG.debug("Opening Stubborn Append Sink {}", this);
     isOpen.set(true); // slightly different than default semantics -- open
     // always keeps it in open state.
     sink.open();
-    LOG.info("Opening Stubborn Append Sink " + this);
+    LOG.debug("Opened Stubborn Append Sink {}", this);
   }
 
   @Override
@@ -73,6 +73,7 @@ public class StubbornAppendSink<S extends EventSink> extends
       appendSuccesses.incrementAndGet();
       return; // success case
     } catch (Exception ex) {
+      LOG.info(ex.getMessage());
       appendFails.incrementAndGet();
       super.close(); // close
 
diff --git a/src/java/com/cloudera/util/CappedExponentialBackoff.java b/src/java/com/cloudera/util/CappedExponentialBackoff.java
index 5066853..ad49ce2 100644
--- a/src/java/com/cloudera/util/CappedExponentialBackoff.java
+++ b/src/java/com/cloudera/util/CappedExponentialBackoff.java
@@ -67,7 +67,8 @@ public class CappedExponentialBackoff implements BackoffPolicy {
    * Has time progressed enough to do a retry attempt?
    */
   public boolean isRetryOk() {
-    return retryTime <= Clock.unixTime();
+    long now = Clock.unixTime();
+    return retryTime <= now;
   }
 
   /**
@@ -83,7 +84,7 @@ public class CappedExponentialBackoff implements BackoffPolicy {
   public void reset() {
     sleepIncrement = initialSleep;
     long cur = Clock.unixTime();
-    retryTime = cur;
+    retryTime = cur + initialSleep;
   }
 
   @Override
diff --git a/src/javatest/com/cloudera/flume/agent/TestAgentFailChainSink.java b/src/javatest/com/cloudera/flume/agent/TestAgentFailChainSink.java
index 1c6acae..2ea4df4 100644
--- a/src/javatest/com/cloudera/flume/agent/TestAgentFailChainSink.java
+++ b/src/javatest/com/cloudera/flume/agent/TestAgentFailChainSink.java
@@ -19,21 +19,34 @@ package com.cloudera.flume.agent;
 
 import static org.junit.Assert.assertEquals;
 
+import java.io.File;
+import java.io.IOException;
+
 import org.codehaus.jettison.json.JSONException;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.cloudera.flume.agent.diskfailover.DiskFailoverDeco;
+import com.cloudera.flume.agent.diskfailover.DiskFailoverManager;
 import com.cloudera.flume.conf.Context;
 import com.cloudera.flume.conf.FlumeArgException;
 import com.cloudera.flume.conf.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.LogicalNodeContext;
+import com.cloudera.flume.conf.ReportTestingContext;
+import com.cloudera.flume.core.BackOffFailOverSink;
 import com.cloudera.flume.core.CompositeSink;
+import com.cloudera.flume.core.Event;
+import com.cloudera.flume.core.EventImpl;
 import com.cloudera.flume.core.EventSink;
+import com.cloudera.flume.handlers.rolling.RollSink;
+import com.cloudera.flume.handlers.thrift.ThriftEventSource;
 import com.cloudera.flume.reporter.ReportEvent;
 import com.cloudera.flume.reporter.ReportManager;
 import com.cloudera.flume.reporter.ReportUtil;
+import com.cloudera.util.Clock;
+import com.cloudera.util.FileUtil;
 import com.cloudera.util.NetUtils;
 
 public class TestAgentFailChainSink {
@@ -58,14 +71,14 @@ public class TestAgentFailChainSink {
 
     // just check a sample of the values:
     assertEquals(0, (long) rpt.getLongMetric("naiveWal.loggedQ"));
-    assertEquals("foo1", rpt
-        .getStringMetric("drainSink.StubbornAppend.InsistentOpen."
+    assertEquals("foo1",
+        rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen."
             + "FailoverChainSink.primary.foo1.name"));
-    assertEquals("foo2", rpt
-        .getStringMetric("drainSink.StubbornAppend.InsistentOpen."
+    assertEquals("foo2",
+        rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen."
             + "FailoverChainSink.backup.BackoffFailover.primary.foo2.name"));
-    assertEquals("foo3", rpt
-        .getStringMetric("drainSink.StubbornAppend.InsistentOpen."
+    assertEquals("foo3",
+        rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen."
             + "FailoverChainSink.backup.BackoffFailover.backup.foo3.name"));
   }
 
@@ -102,13 +115,14 @@ public class TestAgentFailChainSink {
     LOG.info(ReportUtil.toJSONObject(rpt).toString());
 
     // just check a sample of the values:
-    assertEquals("foo1", rpt
-        .getStringMetric("primary.LazyOpenDecorator.StubbornAppend.foo1.name"));
-    assertEquals("foo2", rpt
-        .getStringMetric("backup.BackoffFailover.primary.LazyOpenDecorator."
+    assertEquals(
+        "foo1",
+        rpt.getStringMetric("primary.LazyOpenDecorator.StubbornAppend.foo1.name"));
+    assertEquals("foo2",
+        rpt.getStringMetric("backup.BackoffFailover.primary.LazyOpenDecorator."
             + "StubbornAppend.foo2.name"));
-    assertEquals("foo3", rpt
-        .getStringMetric("backup.BackoffFailover.backup.LazyOpenDecorator."
+    assertEquals("foo3",
+        rpt.getStringMetric("backup.BackoffFailover.backup.LazyOpenDecorator."
             + "StubbornAppend.foo3.name"));
   }
 
@@ -122,6 +136,90 @@ public class TestAgentFailChainSink {
   }
 
   /**
+   * have two destinations, send some messsage to primary, kill primary, send
+   * some to secondary, kill secondary, send some messages to null, restore
+   * secondary send some messagse to secondary.
+   */
+  @Test
+  public void testConfirmBEChain() throws FlumeSpecException, IOException,
+      InterruptedException {
+    // create sources
+    String c1 = "rpcSource(1234)";
+    ThriftEventSource c1Src = (ThriftEventSource) FlumeBuilder.buildSource(
+        LogicalNodeContext.testingContext(), c1);
+    c1Src.open();
+
+    String c2 = "rpcSource(1235)";
+    ThriftEventSource c2Src = (ThriftEventSource) FlumeBuilder.buildSource(
+        LogicalNodeContext.testingContext(), c2);
+    c2Src.open();
+
+    // create agentBEChain sink
+    String spec = "agentBEChain(\"localhost:1234\", \"localhost:1235\")";
+    EventSink snk = new CompositeSink(new Context(), spec);
+    snk.open();
+
+    Event e1 = new EventImpl("test 1".getBytes());
+    Event e2 = new EventImpl("test 2".getBytes());
+    Event e3 = new EventImpl("test 3".getBytes());
+    Event e4 = new EventImpl("test 4".getBytes());
+
+    // Everything is on and we send some messages
+    snk.append(e1);
+    Clock.sleep(100);
+    LOG.info(c1Src.getMetrics().toString());
+    assertEquals(1,
+        (long) c1Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    c1Src.next();
+    c1Src.close();
+
+    // Killed the first of the chain, should go to backup
+    // the number of events lost here is not consistent after close. this
+    // seems time based, and the first two seem to be lost
+    snk.append(e1);
+    Clock.sleep(20);
+    snk.append(e2);
+    Clock.sleep(20);
+    snk.append(e3);
+    Clock.sleep(20);
+    snk.append(e4);
+    Clock.sleep(20);
+
+    LOG.info(c2Src.getMetrics().toString());
+    assertEquals(2,
+        (long) c2Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    // 2 lost in network buffer, but two received in backup.  yay.
+    c2Src.next();
+    c2Src.next();
+    c2Src.close();
+
+    // all thrift sinks are closed now, we should loss messages
+    snk.append(e1); // lost
+    Clock.sleep(20);
+    snk.append(e2); // lost
+    Clock.sleep(20);
+    snk.append(e3); // lost
+    Clock.sleep(20);
+    snk.append(e4); // lost
+    Clock.sleep(20);
+
+    // re-open desination 1.
+    c1Src.open();
+    snk.append(e1);
+    Clock.sleep(20);
+    snk.append(e2);
+    Clock.sleep(20);
+    c1Src.next();
+    c1Src.close();
+    LOG.info(c1Src.getMetrics().toString());
+    // 2 events from prevoius + 1 from new open
+    // first one fails on reopen but next succeeds
+    assertEquals(2 + 1,
+        (long) c2Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    snk.close();
+  }
+
+  /**
    * These should fail if there are any exceptions thrown.
    */
   @Test
@@ -139,22 +237,22 @@ public class TestAgentFailChainSink {
     // just check a sample of the values:
     assertEquals(
         "foo1",
-        rpt
-            .getStringMetric("backup.DiskFailover.drainSink.LazyOpenDecorator.InsistentOpen."
-                + "FailoverChainSink.primary.LazyOpenDecorator.StubbornAppend.foo1.name"));
+        rpt.getStringMetric("backup.DiskFailover.LazyOpenDecorator."
+            + "InsistentAppend.StubbornAppend.InsistentOpen."
+            + "FailoverChainSink.primary.LazyOpenDecorator.StubbornAppend.foo1.name"));
 
     assertEquals(
         "foo2",
-        rpt
-            .getStringMetric("backup.DiskFailover.drainSink.LazyOpenDecorator.InsistentOpen."
-                + "FailoverChainSink.backup.BackoffFailover.primary.LazyOpenDecorator."
-                + "StubbornAppend.foo2.name"));
+        rpt.getStringMetric("backup.DiskFailover.LazyOpenDecorator."
+            + "InsistentAppend.StubbornAppend.InsistentOpen."
+            + "FailoverChainSink.backup.BackoffFailover.primary.LazyOpenDecorator."
+            + "StubbornAppend.foo2.name"));
     assertEquals(
         "foo3",
-        rpt
-            .getStringMetric("backup.DiskFailover.LazyOpenDecorator.InsistentOpen."
-                + "FailoverChainSink.backup.BackoffFailover.backup.LazyOpenDecorator."
-                + "StubbornAppend.foo3.name"));
+        rpt.getStringMetric("backup.DiskFailover.LazyOpenDecorator."
+            + "InsistentAppend.StubbornAppend.InsistentOpen."
+            + "FailoverChainSink.backup.BackoffFailover.backup.LazyOpenDecorator."
+            + "StubbornAppend.foo3.name"));
   }
 
   @Test
@@ -176,4 +274,130 @@ public class TestAgentFailChainSink {
     new CompositeSink(new Context(), spec);
   }
 
+  /**
+   * have two destinations, send some messsage to primary, kill primary, send
+   * some to secondary, kill secondary, send some messages to dfo log, restore
+   * secondary send some messages to secondary.
+   * 
+   * Recover some of secondary.
+   */
+  @Test
+  public void testConfirmDFOChain() throws FlumeSpecException, IOException,
+      InterruptedException {
+    // create sources
+    String c1 = "rpcSource(1234)";
+    ThriftEventSource c1Src = (ThriftEventSource) FlumeBuilder.buildSource(
+        LogicalNodeContext.testingContext()
+
+        , c1);
+    c1Src.open();
+
+    String c2 = "rpcSource(1235)";
+    ThriftEventSource c2Src = (ThriftEventSource) FlumeBuilder.buildSource(
+        LogicalNodeContext.testingContext(), c2);
+    c2Src.open();
+
+    // create agentDFOChain sink
+    File tmpDir = FileUtil.mktempdir();
+    String spec = "agentDFOChain(\"localhost:1234\", \"localhost:1235\")";
+    CompositeSink snk = new CompositeSink(new LogicalNodeContext(
+        tmpDir.getName(), tmpDir.getName()), spec);
+    snk.open();
+
+    Event e1 = new EventImpl("test 1".getBytes());
+    Event e2 = new EventImpl("test 2".getBytes());
+    Event e3 = new EventImpl("test 3".getBytes());
+    Event e4 = new EventImpl("test 4".getBytes());
+
+    // Everything is on and we send some messages.
+    snk.append(e1);
+    Clock.sleep(100);
+    LOG.info(c1Src.getMetrics().toString());
+    assertEquals(1,
+        (long) c1Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    // it got through, yay.
+    c1Src.next();
+    c1Src.close();
+
+    // Killed the first of the chain, should go to backup
+    // the number of events lost here is not consistent after close. this
+    // seems time based, and the first two seem to be lost
+    snk.append(e1);
+    Clock.sleep(20);
+    snk.append(e2);
+    Clock.sleep(20);
+    snk.append(e3);
+    Clock.sleep(20);
+    snk.append(e4);
+    Clock.sleep(20);
+       
+    LOG.info(c2Src.getMetrics().toString());
+    assertEquals(2,
+        (long) c2Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    // 2 lost in network buffer, but two received in backup. yay.
+    c2Src.next();
+    c2Src.next();
+    c2Src.close();
+
+    // all thrift sinks are closed now, we should end up in dfo
+    snk.append(e1); // lost in thrift sink buffer
+    Clock.sleep(20);
+    snk.append(e2); // lost in thrift sink buffer
+    Clock.sleep(20);
+    snk.append(e3); // written
+    Clock.sleep(20);
+    snk.append(e4); // written
+    Clock.sleep(20);
+    LOG.info(snk.getMetrics().toText());
+    assertEquals(
+        2,
+        (long) ReportUtil.getFlattenedReport(snk).getLongMetric(
+            "backup.DiskFailover.NaiveDiskFailover.writingEvts"));
+
+    // re-open destination 1
+    c1Src.open();
+    snk.append(e1);
+    Clock.sleep(20);
+    snk.append(e2);
+    Clock.sleep(20);
+    c1Src.next();
+
+    // get handle to roller in dfo log roller to provide data
+    AgentFailChainSink afcs = (AgentFailChainSink) snk.getSink();
+    BackOffFailOverSink bofos = (BackOffFailOverSink) ((CompositeSink) afcs.snk)
+        .getSink();
+    DiskFailoverDeco dfo = (DiskFailoverDeco) bofos.getBackup();
+    DiskFailoverManager dfm = dfo.getFailoverManager();
+    RollSink dfoWriter = dfo.getDFOWriter();
+    dfoWriter.rotate(); // allow dfo retry thread to go.
+
+    // give data some time to show up.
+    Clock.sleep(1000);
+    c1Src.next();
+    c1Src.next();
+    c1Src.next();
+    c1Src.next();
+    c1Src.close();
+    ReportEvent rpt = ReportUtil.getFlattenedReport(snk);
+    LOG.info(rpt.toString());
+
+    String written = "backup.DiskFailover.NaiveDiskFailover.writingEvts";
+    assertEquals(4, (long) rpt.getLongMetric(written));
+    // yay. all four events written to dfo log
+
+    String primary = "backup.DiskFailover."
+        + "LazyOpenDecorator.InsistentAppend.StubbornAppend."
+        + "InsistentOpen.FailoverChainSink.sentPrimary";
+    assertEquals(4, (long) rpt.getLongMetric(primary));
+    // yay all four go through to the path we wanted. (the primary after the
+    // disk failover)
+
+    // data from DFO log was sent.
+    assertEquals(2 + 4,
+        (long) c1Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+    // first one fails on reopen but next succeeds
+    assertEquals(2 + 1,
+        (long) c2Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
+
+  }
 }
-- 
1.7.1

