support truncate suro file hourly/daily

suro-localfile/src/main/java/com/netflix/suro/sink/localfile/Granularity.java

@@ -0,0 +1,47 @@
+package com.netflix.suro.sink.localfile;
+
+import org.joda.time.*;
+
+/**
+ * Created by [email protected] on 9/30/14.
+ */
+public enum Granularity {
+
+    HOUR{
+        @Override
+        public DateTime truncate(DateTime time) {
+            final MutableDateTime mutableDateTime = time.toMutableDateTime();
+            mutableDateTime.setMillisOfSecond(0);
+            mutableDateTime.setSecondOfMinute(0);
+            mutableDateTime.setMinuteOfHour(0);
+            return mutableDateTime.toDateTime();
+        }
+
+        @Override
+        public ReadablePeriod getUnits(int n) {
+            return Hours.hours(n);
+        }
+    } ,
+    DAY{
+        @Override
+        public DateTime truncate(DateTime time) {
+            final MutableDateTime mutableDateTime = time.toMutableDateTime();
+            mutableDateTime.setMillisOfDay(0);
+            return mutableDateTime.toDateTime();
+        }
+
+        @Override
+        public ReadablePeriod getUnits(int n) {
+            return Days.days(n);
+        }
+    };
+
+    public abstract DateTime truncate(DateTime time);
+    public abstract ReadablePeriod getUnits(int n);
+
+    public final DateTime next(DateTime time){
+        return truncate(time.plus(getUnits(1)));
+    }
+
+
+}

suro-localfile/src/main/java/com/netflix/suro/sink/localfile/LocalFileSink.java

@@ -86,6 +86,8 @@ public class LocalFileSink extends QueuedSink implements Sink {
 
     private boolean messageWrittenInRotation = false;
 
+    private Granularity granularity;
+
     @JsonCreator
     public LocalFileSink(
             @JsonProperty("outputDir") String outputDir,
@@ -97,6 +99,7 @@ public LocalFileSink(
             @JsonProperty("queue4Sink") MessageQueue4Sink queue4Sink,
             @JsonProperty("batchSize") int batchSize,
             @JsonProperty("batchTimeout") int batchTimeout,
+            @JsonProperty("granularity") String gran,
             @JacksonInject TrafficController trafficController,
             @JacksonInject SpaceChecker spaceChecker) {
         if (!outputDir.endsWith("/")) {
@@ -113,6 +116,10 @@ public LocalFileSink(
         this.trafficController = trafficController;
         this.spaceChecker = spaceChecker;
 
+        if(gran!=null){
+            granularity = Granularity.valueOf(gran.toUpperCase());
+        }
+
         Monitors.registerObject(outputDir.replace('/', '_'), this);
         initialize("localfile_" + outputDir.replace('/', '_'),
                 queue4Sink == null ? new MemoryQueue4Sink(10000) : queue4Sink, batchSize, batchTimeout);
@@ -189,7 +196,14 @@ private void rotate() throws IOException {
 
         filePath = newName;
 
-        nextRotation = new DateTime().plus(rotationPeriod).getMillis();
+        DateTime currentDateTime = new DateTime();
+        nextRotation = currentDateTime.plus(rotationPeriod).getMillis();
+        if(granularity!=null){
+            long expectedBreak = granularity.next(currentDateTime).getMillis();
+            if(nextRotation >expectedBreak){
+                nextRotation = expectedBreak;
+            }
+        }
 
         if (!spaceChecker.hasEnoughSpace()) {
             trafficController.stopTakingTraffic();