Issue #1350: Fix stats aggregation

Change-Id: I652db6f6f5bc0cab8fc9e85969bf57f1c2382da3

Former-commit-id: 71574f210d [formerly 94df67d354] [formerly 71574f210d [formerly 94df67d354] [formerly 2b124db62a [formerly b06d51266d600e0b04f7fe10bc36e312a2275428]]]
Former-commit-id: 2b124db62a
Former-commit-id: 99f19186ef [formerly 5a45254db0]
Former-commit-id: 9a51a3d3b2

edexOsgi/com.raytheon.uf.common.localization/src/com/raytheon/uf/common/localization/LocalizationFile.java

@@ -34,6 +34,7 @@ import com.raytheon.uf.common.localization.ILocalizationAdapter.ListResponse;
 import com.raytheon.uf.common.localization.LocalizationContext.LocalizationLevel;
 import com.raytheon.uf.common.localization.exception.LocalizationException;
 import com.raytheon.uf.common.localization.exception.LocalizationOpFailedException;
+import com.raytheon.uf.common.serialization.JAXBManager;
 import com.raytheon.uf.common.status.IUFStatusHandler;
 import com.raytheon.uf.common.status.UFStatus;
 import com.raytheon.uf.common.status.UFStatus.Priority;
@@ -165,7 +166,7 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
     private LocalizationLevel protectedLevel;
 
     /** File changed observers */
-    private Set<ILocalizationFileObserver> observers = new HashSet<ILocalizationFileObserver>();
+    private final Set<ILocalizationFileObserver> observers = new HashSet<ILocalizationFileObserver>();
 
     /** Flag to set if file has been requested */
     protected boolean fileRequested = false;
@@ -187,8 +188,8 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
      * @return
      */
     boolean isNull() {
-        return adapter == null && path == null && context == null
+        return (adapter == null) && (path == null) && (context == null)
-                && file == null;
+                && (file == null);
     }
 
     LocalizationFile(ILocalizationAdapter adapter, LocalizationContext context,
@@ -279,7 +280,7 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
             adapter.retrieve(this);
         }
 
-        if (isDirectory == false && !file.exists()) {
+        if ((isDirectory == false) && !file.exists()) {
             try {
                 file.getParentFile().mkdirs();
             } catch (Throwable t) {
@@ -352,8 +353,8 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
             // Read in the bytes
             int offset = 0;
             int numRead = 0;
-            while (offset < rval.length
+            while ((offset < rval.length)
-                    && (numRead = is.read(rval, offset, rval.length - offset)) >= 0) {
+                    && ((numRead = is.read(rval, offset, rval.length - offset)) >= 0)) {
                 offset += numRead;
             }
 
@@ -548,7 +549,7 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
      * @return true if the file exists
      */
     public boolean exists() {
-        return isNull() == false && adapter.exists(this);
+        return (isNull() == false) && adapter.exists(this);
     }
 
     /**
@@ -596,6 +597,44 @@ public final class LocalizationFile implements Comparable<LocalizationFile> {
         }
     }
 
+    /**
+     * Returns the object version of this jaxb serialized file. Returns null if
+     * the file does not exist or is empty.
+     * 
+     * @param <T>
+     * @param resultClass
+     * @param manager
+     * @return
+     * @throws LocalizationException
+     */
+    public <T> T jaxbUnmarshal(Class<T> resultClass, JAXBManager manager)
+            throws LocalizationException {
+        File f = getFile();
+        if (f.exists() && (f.length() > 0)) {
+            InputStream is = null;
+            try {
+                is = openInputStream();
+                T object = resultClass.cast(manager
+                        .jaxbUnmarshalFromInputStream(is));
+                return object;
+            } catch (Exception e) {
+                throw new LocalizationException("Could not unmarshal file "
+                        + file.getName(), e);
+            } finally {
+                if (is != null) {
+                    try {
+                        is.close();
+                    } catch (IOException e) {
+                        statusHandler.handle(Priority.WARN,
+                                "Failed to close input stream for file", e);
+                    }
+                }
+            }
+        }
+
+        return null;
+    }
+
     @Override
     public String toString() {
         return context + IPathManager.SEPARATOR + path;

edexOsgi/com.raytheon.uf.common.stats/src/com/raytheon/uf/common/stats/xml/StatisticsEvent.java

@@ -19,6 +19,7 @@
  **/
 package com.raytheon.uf.common.stats.xml;
 
+import java.lang.reflect.Method;
 import java.util.List;
 
 import javax.xml.bind.annotation.XmlAccessType;
@@ -28,6 +29,7 @@ import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlElements;
 import javax.xml.bind.annotation.XmlRootElement;
 
+import com.raytheon.uf.common.event.Event;
 import com.raytheon.uf.common.serialization.annotations.DynamicSerialize;
 import com.raytheon.uf.common.serialization.annotations.DynamicSerializeElement;
 
@@ -72,6 +74,12 @@ public class StatisticsEvent {
     @DynamicSerializeElement
     private List<StatisticsAggregate> aggregateList;
 
+    private Class<? extends Event> typeClass = null;
+
+    private List<Method> groupByMethods = null;
+
+    private List<Method> aggregateMethods = null;
+
     /**
      * @return the aggregateList
      */
@@ -146,4 +154,29 @@ public class StatisticsEvent {
     public void setType(String type) {
         this.type = type;
     }
+
+    public Class<? extends Event> getTypeClass() {
+        return typeClass;
+    }
+
+    public void setTypeClass(Class<? extends Event> typeClass) {
+        this.typeClass = typeClass;
+    }
+
+    public List<Method> getGroupByMethods() {
+        return groupByMethods;
+    }
+
+    public void setGroupByMethods(List<Method> groupByMethods) {
+        this.groupByMethods = groupByMethods;
+    }
+
+    public List<Method> getAggregateMethods() {
+        return aggregateMethods;
+    }
+
+    public void setAggregateMethods(List<Method> aggregateMethods) {
+        this.aggregateMethods = aggregateMethods;
+    }
+
 }

edexOsgi/com.raytheon.uf.common.util/src/com/raytheon/uf/common/util/ReflectionUtil.java

@@ -83,7 +83,7 @@ public final class ReflectionUtil {
         }
     }
 
-    public static Object getter(Object object, String name)
+    public static Method getGetterMethod(Class<?> clazz, String name)
             throws ReflectionException {
         // Assume camel cased names - capitalize first letter...
         String method = Character.toUpperCase(name.charAt(0))
@@ -92,13 +92,23 @@ public final class ReflectionUtil {
             Method m;
             try {
                 // Try common 'get' first...
-                m = object.getClass().getMethod("get" + method);
+                m = clazz.getMethod("get" + method);
             } catch (NoSuchMethodException e) {
                 // Try 'is' as a prefix
-                m = object.getClass().getMethod("is" + method);
+                m = clazz.getMethod("is" + method);
             }
 
-            return m.invoke(object, (Object[]) null);
+            return m;
+        } catch (Exception e) {
+            throw new ReflectionException(e);
+        }
+    }
+
+    public static Object getter(Object object, String name)
+            throws ReflectionException {
+        try {
+            return getGetterMethod(object.getClass(), name).invoke(object,
+                    (Object[]) null);
         } catch (Exception e) {
             throw new ReflectionException(e);
         }

edexOsgi/com.raytheon.uf.edex.purgesrv/res/spring/purge-spring.xml

@@ -53,7 +53,7 @@
 		
 		<!-- schedule the timer to purge outgoing directory -->
 		<route id="purgeOutgoingScheduled">
-			<from uri="purgeOutgoingCron" />
+			<from ref="purgeOutgoingCron" />
 			<bean ref="purgeOutgoing" method="purge" />
 		</route>
       
@@ -67,7 +67,7 @@
 		
 	    <!-- Purge on Scheduled timer -->
 		<route id="purgeScheduled">
-			<from uri="purgeCron" />
+			<from ref="purgeCron" />
 			<doTry>
 				<bean ref="purgeManager" method="executePurge" />
 				<doCatch>
@@ -80,7 +80,7 @@
 		
 		<!-- schedule the timer to purge log directories -->
 		<route id="purgeLogScheduled">
-			<from uri="purgeLogsCron" />
+			<from ref="purgeLogsCron" />
 			<bean ref="purgeLogs" method="purge" />
 		</route>
 

edexOsgi/com.raytheon.uf.edex.stats/META-INF/MANIFEST.MF

@@ -16,6 +16,7 @@ Require-Bundle: com.raytheon.uf.common.serialization;bundle-version="1.12.1174",
  com.raytheon.uf.common.serialization.comm;bundle-version="1.12.1174",
  com.raytheon.uf.common.stats;bundle-version="1.0.0",
  com.raytheon.uf.common.dataplugin;bundle-version="1.12.1174",
+ com.raytheon.uf.common.util,
  org.springframework;bundle-version="2.5.6",
  com.raytheon.uf.edex.core;bundle-version="1.12.1174",
  com.raytheon.uf.common.status;bundle-version="1.12.1174",

edexOsgi/com.raytheon.uf.edex.stats/res/spring/edex-process-stats.xml

@@ -12,7 +12,6 @@
 
     <bean id="aggregateManager" class="com.raytheon.uf.edex.stats.AggregateManager">
         <constructor-arg value="${stats.period}"/>
-        <constructor-arg value="${stats.scanInterval}"/>
     </bean>
 
    	<bean id="edexStatsRegistered" factory-bean="clusteredCamelContextMgr"
@@ -24,9 +23,10 @@
             xmlns="http://camel.apache.org/schema/spring"
             errorHandlerRef="errorHandler"
             autoStartup="false">
+            <endpoint id="statsScanTimer" uri="timer://scanStats?period=${stats.scanInterval}m"/>
 
             <route id="statsTableScan">
-              <from uri="timer://scanStats?period=15m" />
+              <from ref="statsScanTimer" />
                 <doTry>
                    <bean ref="statsPurge" method="purgeAggregates"/>
                    <bean ref="aggregateManager" method="scan"/>

edexOsgi/com.raytheon.uf.edex.stats/resources/edexProcessStats.properties

@@ -1,4 +1,4 @@
 # scan interval of stats table in minutes
-stats.scanInterval=15
+stats.scanInterval=2
 # bucket interval or period of when to aggregate in minutes
-stats.period=2
+stats.period=5

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/AggregateManager.java

@@ -20,31 +20,30 @@
 package com.raytheon.uf.edex.stats;
 
 import java.lang.reflect.Method;
-import java.util.ArrayList;
 import java.util.Calendar;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.TimeZone;
 
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Multimap;
 import com.raytheon.uf.common.event.Event;
-import com.raytheon.uf.common.serialization.SerializationException;
 import com.raytheon.uf.common.serialization.SerializationUtil;
 import com.raytheon.uf.common.stats.AggregateRecord;
 import com.raytheon.uf.common.stats.StatsRecord;
 import com.raytheon.uf.common.stats.xml.StatisticsAggregate;
-import com.raytheon.uf.common.stats.xml.StatisticsConfig;
 import com.raytheon.uf.common.stats.xml.StatisticsEvent;
 import com.raytheon.uf.common.stats.xml.StatisticsGroup;
 import com.raytheon.uf.common.status.IUFStatusHandler;
 import com.raytheon.uf.common.status.UFStatus;
 import com.raytheon.uf.common.time.TimeRange;
-import com.raytheon.uf.edex.database.dao.CoreDao;
+import com.raytheon.uf.common.time.util.TimeUtil;
-import com.raytheon.uf.edex.database.dao.DaoConfig;
+import com.raytheon.uf.common.util.CollectionUtil;
+import com.raytheon.uf.edex.stats.dao.AggregateRecordDao;
 import com.raytheon.uf.edex.stats.dao.StatsDao;
-import com.raytheon.uf.edex.stats.handler.StatsHandler;
 import com.raytheon.uf.edex.stats.util.ConfigLoader;
 
 /**
@@ -60,119 +59,67 @@ import com.raytheon.uf.edex.stats.util.ConfigLoader;
  * ------------ ---------- ----------- --------------------------
  * Aug 21, 2012            jsanchez    Stored the aggregate buckets in the db.
  * Nov 07, 2012   1317     mpduff      Updated Configuration Files.
- * 
+ * Nov 28, 2012   1350     rjpeter     Simplied aggregation and added aggregation with current db aggregate records.
  * </pre>
  * 
  * @author jsanchez
  * 
  */
 public class AggregateManager {
-    private class TimeRangeKey extends TimeRange {
-        private static final long serialVersionUID = 4603487307433273159L;
-
-        public TimeRangeKey(Calendar cal1, Calendar cal2) {
-            super(cal1, cal2);
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (o != null && o instanceof TimeRange) {
-                TimeRange other = (TimeRange) o;
-
-                return getStart().equals(other.getStart())
-                        && getEnd().equals(other.getEnd());
-            }
-
-            return false;
-        }
-
-        @Override
-        public int hashCode() {
-            return 1;
-        }
-
-        @Override
-        public String toString() {
-            return super.toString();
-        }
-    }
-
     private static final IUFStatusHandler statusHandler = UFStatus
             .getHandler(AggregateManager.class);
 
     /** In minutes */
     private int bucketInterval;
 
-    /** In minutes */
-    private int scanInterval;
-
     /** default value */
     private static final int defaultBucketInterval = 5;
 
     /** default value */
     private static final int defaultScanInterval = 15;
 
-    /** loads localized copies of the statsConfig */
+    public AggregateManager(String bucketInterval) {
-    private final ConfigLoader configLoader;
+        validateIntervals(bucketInterval);
-
-    private final CoreDao aggregateRecordDao = new CoreDao(DaoConfig.forClass(
-            "metadata", AggregateRecord.class));
-
-    public AggregateManager(String bucketInterval, String scanInterval)
-            throws Exception {
-        configLoader = new ConfigLoader();
-        validateIntervals(bucketInterval, scanInterval);
-        configLoader.load();
-        StatsHandler.setValidEventTypes(configLoader.getConfigurations());
     }
 
     /**
-     * Performs the aggregation based on the statsConfig file.
+     * Aggregates the grouped events and then aggregates them with any matching
+     * range in DB and stores the updated aggregate.
      * 
-     * @param key
+     * @param dao
-     * @param data
+     * @param statsEvent
+     * @param timeRange
+     * @param groupedEvents
      */
-    private void aggregate(StatisticsEvent statsEvent, TimeRange timeRange,
+    private void aggregate(AggregateRecordDao dao, StatisticsEvent statsEvent,
-            List<Event> data) {
+            TimeRange timeRange, Multimap<String, Event> groupedEvents) {
-        Calendar start = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
+        Calendar start = TimeUtil.newCalendar(TimeZone.getTimeZone("GMT"));
         start.setTime(timeRange.getStart());
 
-        Calendar end = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
+        Calendar end = TimeUtil.newCalendar(TimeZone.getTimeZone("GMT"));
         end.setTime(timeRange.getEnd());
 
-        // collect grouping names from stats config
-        List<String> groupByColumns = new ArrayList<String>();
-        for (StatisticsGroup groupBy : statsEvent.getGroupList()) {
-            String column = groupBy.getName();
-            groupByColumns.add(column);
-        }
-
-        // breaks data into groups
-        Map<String, List<Event>> map = divideIntoGroups(data, groupByColumns);
-
         // perform aggregate functions on the grouped data
-        for (String groupKey : map.keySet()) {
+        for (String groupKey : groupedEvents.keySet()) {
+            Collection<Event> groupData = groupedEvents.get(groupKey);
+            Iterator<Method> aggrMethodIter = statsEvent.getAggregateMethods()
+                    .iterator();
+            Iterator<StatisticsAggregate> statAggrIter = statsEvent
+                    .getAggregateList().iterator();
+            int count = groupData.size();
+
+            while (aggrMethodIter.hasNext() && statAggrIter.hasNext()) {
+                String field = statAggrIter.next().getField();
+                Method m = aggrMethodIter.next();
 
-            List<Event> groupData = map.get(groupKey);
-            for (StatisticsAggregate aggregate : statsEvent.getAggregateList()) {
-                String field = aggregate.getField();
                 try {
-                    double[] values = new double[groupData.size()];
+                    double max = -Double.MAX_VALUE;
-                    String methodName = getterMethodName(field);
-                    for (int i = 0; i < groupData.size(); i++) {
-                        Object obj = groupData.get(i);
-                        Class<?> clazz = obj.getClass();
-                        Method m = clazz.getMethod(methodName, new Class<?>[0]);
-                        Number number = (Number) m.invoke(obj, new Object[0]);
-                        values[i] = number.doubleValue();
-                    }
-
-                    double count = values.length;
-                    double max = 0;
                     double min = Double.MAX_VALUE;
                     double sum = 0;
 
-                    for (double value : values) {
+                    for (Event event : groupData) {
+                        Number number = (Number) m.invoke(event, new Object[0]);
+                        double value = number.doubleValue();
                         sum += value;
                         if (value > max) {
                             max = value;
@@ -189,7 +136,7 @@ public class AggregateManager {
                     record.setMin(min);
                     record.setMax(max);
                     record.setCount(count);
-                    aggregateRecordDao.persist(record);
+                    dao.mergeRecord(record);
                 } catch (Exception e) {
                     statusHandler.error("Unable to aggregate '" + field + "'",
                             e);
@@ -206,73 +153,13 @@ public class AggregateManager {
      * @param date
      * @return
      */
-    private TimeRangeKey createTimeRangeKey(Calendar date) {
+    private TimeRange createTimeRange(Calendar date) {
         Calendar start = getBucketStartTime(date);
         Calendar end = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
         end.setTimeInMillis(start.getTimeInMillis());
         end.add(Calendar.MINUTE, bucketInterval);
 
-        TimeRangeKey timeRangeKey = new TimeRangeKey(start, end);
+        return new TimeRange(start, end);
-
-        return timeRangeKey;
-    }
-
-    /**
-     * Breaks the list of data into groups based on groupByColumns. The key is a
-     * concatenation of the column values (i.e. datatype.username). This method
-     * can group data to n-number of levels.
-     * 
-     * @param data
-     * @param groupByColumns
-     * @return
-     */
-    private Map<String, List<Event>> divideIntoGroups(List<Event> data,
-            List<String> groupByColumns) {
-        Map<String, List<Event>> map = new HashMap<String, List<Event>>();
-        map.put("", data);
-        for (String column : groupByColumns) {
-
-            List<Map<String, List<Event>>> listOfMaps = new ArrayList<Map<String, List<Event>>>();
-            for (String parent : map.keySet()) {
-                List<Event> list = map.get(parent);
-                listOfMaps.add(group(list, column, parent));
-            }
-
-            map.clear();
-
-            // replace map with grouped data
-            for (Map<String, List<Event>> m : listOfMaps) {
-                for (String k : m.keySet()) {
-                    map.put(k, m.get(k));
-                }
-            }
-        }
-
-        return map;
-    }
-
-    /**
-     * Extracts the events from the stats records.
-     * 
-     * @param records
-     * @return
-     */
-    private List<Event> extractEvents(List<StatsRecord> records) {
-        List<Event> eventsList = new ArrayList<Event>(records.size());
-
-        for (StatsRecord record : records) {
-            try {
-                Event event = SerializationUtil.transformFromThrift(
-                        Event.class, record.getEvent());
-                eventsList.add(event);
-            } catch (SerializationException e) {
-                statusHandler
-                        .error("Error trying to transform event. Aggregation may be inaccurate. ",
-                                e);
-            }
-        }
-
-        return eventsList;
     }
 
     /**
@@ -302,155 +189,133 @@ public class AggregateManager {
         return start;
     }
 
-    /**
-     * Returns the name of the getter method for the parameter
-     * 
-     * @param parameter
-     * @return
-     */
-    private String getterMethodName(String parameter) {
-        return "get" + parameter.substring(0, 1).toUpperCase()
-                + parameter.substring(1);
-    }
-
-    /**
-     * Helper method to group data to one level.
-     * 
-     * @param data
-     * @param column
-     * @param parent
-     * @return
-     */
-    private Map<String, List<Event>> group(List<Event> data, String column,
-            String parent) {
-        Map<String, List<Event>> map = new HashMap<String, List<Event>>();
-        String methodName = getterMethodName(column);
-        for (Event rec : data) {
-            try {
-                Class<?> clazz = rec.getClass();
-                Method m = clazz.getMethod(methodName, new Class<?>[0]);
-                String value = column + ":"
-                        + String.valueOf(m.invoke(rec, new Object[0]));
-                if (parent.length() > 0) {
-                    value = parent + "-" + value;
-                }
-                List<Event> list = map.get(value);
-                if (list == null) {
-                    list = new ArrayList<Event>();
-                }
-                list.add(rec);
-                map.put(value, list);
-            } catch (Exception e) {
-                statusHandler.error("Error creating groups", e);
-            }
-        }
-
-        return map;
-    }
-
-    /**
-     * Retrieve StatRecords from the metadata.event.stats table. This method
-     * does not retrieve records of the current bucket.
-     */
-    private void retrieveStatRecords(StatsDao statsRecordDao,
-            Map<String, Map<TimeRangeKey, List<StatsRecord>>> aggregateBuckets)
-            throws Exception {
-        Calendar current = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
-        for (StatisticsConfig statsConfig : configLoader.getConfigurations()) {
-            for (StatisticsEvent event : statsConfig.getEvents()) {
-                String eventType = event.getType();
-                // Does not retrieve stat records of current bucket.
-                // this method should always return a valid array.
-                StatsRecord[] records = statsRecordDao.retrieveRecords(
-                        getBucketStartTime(current), eventType);
-                sort(eventType, records, aggregateBuckets);
-            }
-        }
-    }
-
     /**
      * Scans the stats table to be stored in buckets and aggregate if necessary.
      */
     public void scan() throws Exception {
+        long t0 = System.currentTimeMillis();
+        ConfigLoader configLoader = ConfigLoader.getInstance();
         StatsDao statsRecordDao = new StatsDao();
-        Map<String, Map<TimeRangeKey, List<StatsRecord>>> aggregateBuckets = new HashMap<String, Map<TimeRangeKey, List<StatsRecord>>>();
+        AggregateRecordDao aggrRecordDao = new AggregateRecordDao();
 
-        // retrieves records and sorts in buckets
+        Map<String, StatisticsEvent> statsMap = configLoader.getTypeView();
-        retrieveStatRecords(statsRecordDao, aggregateBuckets);
 
-        // loops through map to aggregate buckets
+        // latest time to pull
-        for (StatisticsConfig statsConfig : configLoader.getConfigurations()) {
+        Calendar timeToProcess = Calendar.getInstance(TimeZone
-            for (StatisticsEvent event : statsConfig.getEvents()) {
+                .getTimeZone("GMT"));
-                String eventType = event.getType();
+        int count = 0;
+
+        // process the events by type
+        for (Map.Entry<String, StatisticsEvent> entry : statsMap.entrySet()) {
+            String type = entry.getKey();
+            StatisticsEvent event = entry.getValue();
+            List<StatsRecord> records = null;
+
+            do {
+                // retrieve stats in blocks of 1000
+                records = statsRecordDao.retrieveRecords(timeToProcess, type,
+                        2000);
+
+                if (!CollectionUtil.isNullOrEmpty(records)) {
+                    // sort events into time buckets
+                    Map<TimeRange, Multimap<String, Event>> timeMap = sort(
+                            event, records);
+
+                    for (Map.Entry<TimeRange, Multimap<String, Event>> timeMapEntry : timeMap
+                            .entrySet()) {
+                        aggregate(aggrRecordDao, event, timeMapEntry.getKey(),
+                                timeMapEntry.getValue());
+                    }
 
-                Map<TimeRangeKey, List<StatsRecord>> map = aggregateBuckets
-                        .get(eventType);
-                // map should never be null, since it will be set in the 'sort'
-                // method.
-                for (Iterator<Map.Entry<TimeRangeKey, List<StatsRecord>>> iter = map
-                        .entrySet().iterator(); iter.hasNext();) {
-                    Entry<TimeRangeKey, List<StatsRecord>> element = iter
-                            .next();
-                    TimeRangeKey tr = element.getKey();
-                    List<StatsRecord> records = element.getValue();
-                    if (!records.isEmpty()) {
-                        List<Event> data = extractEvents(records);
-                        aggregate(event, tr, data);
                     try {
                         statsRecordDao.deleteAll(records);
                     } catch (Exception e) {
-                            statusHandler.error("Error deleting stat records",
+                        statusHandler.error("Error deleting stat records", e);
+                    }
+
+                    count += records.size();
+                }
+            } while (!CollectionUtil.isNullOrEmpty(records));
+        }
+
+        long t1 = System.currentTimeMillis();
+        statusHandler.info("Aggregated " + count + " stat events in "
+                + (t1 - t0) + " ms");
+    }
+
+    /**
+     * Sorts the records into time buckets and groups by the underlying Event.
+     * 
+     * @param records
+     * @return
+     */
+    private Map<TimeRange, Multimap<String, Event>> sort(
+            StatisticsEvent statEvent, List<StatsRecord> records) {
+        Map<TimeRange, Multimap<String, Event>> rval = new HashMap<TimeRange, Multimap<String, Event>>();
+        TimeRange timeRange = null;
+        Multimap<String, Event> eventsByGroup = null;
+        final Object[] EMPTY_OBJ_ARR = new Object[0];
+        StringBuilder group = new StringBuilder();
+
+        for (StatsRecord record : records) {
+            if ((timeRange == null)
+                    || !timeRange.contains(record.getDate().getTime())) {
+                // Create bucket based on stats record date
+                timeRange = createTimeRange(record.getDate());
+                eventsByGroup = rval.get(timeRange);
+                if (eventsByGroup == null) {
+                    eventsByGroup = ArrayListMultimap.create();
+                    rval.put(timeRange, eventsByGroup);
+                }
+            }
+
+            try {
+                // get underlying event
+                Event event = SerializationUtil.transformFromThrift(
+                        Event.class, record.getEvent());
+
+                // determine group
+                boolean addDelim = false;
+                Iterator<Method> gMethodIter = statEvent.getGroupByMethods()
+                        .iterator();
+                Iterator<StatisticsGroup> gFieldNameIter = statEvent
+                        .getGroupList().iterator();
+                group.setLength(0);
+
+                while (gMethodIter.hasNext() && gFieldNameIter.hasNext()) {
+                    Method m = gMethodIter.next();
+                    String field = gFieldNameIter.next().getName();
+                    String gVal = String
+                            .valueOf(m.invoke(event, EMPTY_OBJ_ARR));
+
+                    if (addDelim) {
+                        group.append('-');
+                    } else {
+                        addDelim = true;
+                    }
+
+                    group.append(field).append(':').append(gVal);
+                }
+
+                eventsByGroup.put(group.toString(), event);
+            } catch (Exception e) {
+                statusHandler
+                        .error("Error processing event. Aggregation may be inaccurate. ",
                                 e);
             }
         }
-                    iter.remove();
+
-                }
+        return rval;
-            }
-        }
     }
 
     /**
-     * Stores the results into proper aggregate buckets. This method assumes
+     * Tests if the bucket interval is a valid value. If value is invalid then
-     * that the records are in date order.
+     * value will be set to default value.
-     * 
-     * @param events
-     */
-    private void sort(String eventType, StatsRecord[] records,
-            Map<String, Map<TimeRangeKey, List<StatsRecord>>> aggregateBuckets)
-            throws Exception {
-        Map<TimeRangeKey, List<StatsRecord>> map = aggregateBuckets
-                .get(eventType);
-        if (map == null) {
-            map = new HashMap<TimeRangeKey, List<StatsRecord>>();
-            aggregateBuckets.put(eventType, map);
-        }
-
-        TimeRangeKey timeRange = null;
-        for (StatsRecord record : records) {
-            if (timeRange == null
-                    || !timeRange.contains(record.getDate().getTime())) {
-                // Create bucket based on stats record date
-                timeRange = createTimeRangeKey(record.getDate());
-            }
-
-            List<StatsRecord> bucketList = map.get(timeRange);
-            if (bucketList == null) {
-                bucketList = new ArrayList<StatsRecord>();
-                map.put(timeRange, bucketList);
-            }
-            bucketList.add(record);
-        }
-    }
-
-    /**
-     * Tests if the bucket interval and the scan interval are valid values. If
-     * values are invalid then values will be set to default values.
      * 
      * @param bucketInt
-     * @param scanInt
      * @return
      */
-    private void validateIntervals(String bucketInt, String scanInt) {
+    private void validateIntervals(String bucketInt) {
         try {
             bucketInterval = Integer.parseInt(bucketInt);
         } catch (NumberFormatException e) {
@@ -460,26 +325,6 @@ public class AggregateManager {
                     + defaultBucketInterval + "'");
         }
 
-        try {
-            scanInterval = Integer.parseInt(scanInt);
-        } catch (NumberFormatException e) {
-            scanInterval = defaultScanInterval;
-            statusHandler.info("'" + scanInt
-                    + "' is not a valid scan interval value. Setting to '"
-                    + defaultScanInterval + "'");
-        }
-
-        if (scanInterval < bucketInterval) {
-            scanInterval = defaultBucketInterval;
-            bucketInterval = defaultBucketInterval;
-            statusHandler
-                    .info("The bucket interval can not be greater than the scan interval. Setting scan interval to '"
-                            + defaultBucketInterval
-                            + "' and bucket interval to '"
-                            + bucketInterval
-                            + "'");
-        }
-
         int incrementsWithinHour = bucketInterval;
         // checks if period is larger than 60 minutes
         if (bucketInterval > 60) {
@@ -492,5 +337,4 @@ public class AggregateManager {
                             + bucketInterval + "'");
         }
     }
-
 }

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/dao/AggregateRecordDao.java

@@ -0,0 +1,64 @@
+package com.raytheon.uf.edex.stats.dao;
+
+import java.util.List;
+
+import com.raytheon.uf.common.dataquery.db.QueryParam.QueryOperand;
+import com.raytheon.uf.common.stats.AggregateRecord;
+import com.raytheon.uf.common.util.CollectionUtil;
+import com.raytheon.uf.edex.database.DataAccessLayerException;
+import com.raytheon.uf.edex.database.dao.CoreDao;
+import com.raytheon.uf.edex.database.dao.DaoConfig;
+import com.raytheon.uf.edex.database.query.DatabaseQuery;
+
+public class AggregateRecordDao extends CoreDao {
+    /**
+     * Creates a new data access object
+     */
+    public AggregateRecordDao() {
+        super(DaoConfig.forClass("metadata", AggregateRecord.class));
+    }
+
+    /**
+     * Retrieves stat records that has a date before the limit.
+     * 
+     * @param limit
+     * @param eventType
+     * @param maxResults
+     *            if greater than 0 will limit database results to maxResults
+     * @return an array of stat records. If an error occurs, then an array of
+     *         size 0 will be returned.
+     * @throws DataAccessLayerException
+     */
+    @SuppressWarnings("unchecked")
+    public void mergeRecord(AggregateRecord newRecord)
+            throws DataAccessLayerException {
+        DatabaseQuery query = new DatabaseQuery(AggregateRecord.class);
+        query.addQueryParam("eventType", newRecord.getEventType(),
+                QueryOperand.EQUALS);
+        query.addQueryParam("field", newRecord.getField(), QueryOperand.EQUALS);
+        query.addQueryParam("grouping", newRecord.getGrouping(),
+                QueryOperand.EQUALS);
+        query.addQueryParam("startDate", newRecord.getStartDate(),
+                QueryOperand.EQUALS);
+        query.addQueryParam("endDate", newRecord.getEndDate(),
+                QueryOperand.EQUALS);
+
+        List<AggregateRecord> results = (List<AggregateRecord>) queryByCriteria(query);
+        if (!CollectionUtil.isNullOrEmpty(results)) {
+            // shouldn't be able to get multiple results, just merge with first
+            // and update
+            AggregateRecord prevRecord = results.get(0);
+            prevRecord.setCount(prevRecord.getCount() + newRecord.getCount());
+            prevRecord.setSum(prevRecord.getSum() + newRecord.getSum());
+            if (newRecord.getMin() < prevRecord.getMin()) {
+                prevRecord.setMin(newRecord.getMin());
+            }
+            if (newRecord.getMax() > prevRecord.getMax()) {
+                prevRecord.setMax(newRecord.getMax());
+            }
+            update(prevRecord);
+        } else {
+            persist(newRecord);
+        }
+    }
+}

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/dao/StatsDao.java

@@ -5,17 +5,12 @@ import java.util.List;
 
 import com.raytheon.uf.common.dataquery.db.QueryParam.QueryOperand;
 import com.raytheon.uf.common.stats.StatsRecord;
-import com.raytheon.uf.common.status.IUFStatusHandler;
-import com.raytheon.uf.common.status.UFStatus;
 import com.raytheon.uf.edex.database.DataAccessLayerException;
 import com.raytheon.uf.edex.database.dao.CoreDao;
 import com.raytheon.uf.edex.database.dao.DaoConfig;
 import com.raytheon.uf.edex.database.query.DatabaseQuery;
 
 public class StatsDao extends CoreDao {
-    private static final transient IUFStatusHandler statusHandler = UFStatus
-            .getHandler(StatsDao.class);
-
     /**
      * Creates a new data access object
      */
@@ -28,30 +23,24 @@ public class StatsDao extends CoreDao {
      * 
      * @param limit
      * @param eventType
+     * @param maxResults
+     *            if greater than 0 will limit database results to maxResults
      * @return an array of stat records. If an error occurs, then an array of
      *         size 0 will be returned.
      * @throws DataAccessLayerException
      */
-    public StatsRecord[] retrieveRecords(Calendar limit, String eventType) {
+    @SuppressWarnings("unchecked")
+    public List<StatsRecord> retrieveRecords(Calendar limit, String eventType,
+            int maxResults) throws DataAccessLayerException {
         DatabaseQuery query = new DatabaseQuery(StatsRecord.class);
         query.addQueryParam("eventType", eventType, QueryOperand.EQUALS);
         query.addQueryParam("date", limit, QueryOperand.LESSTHAN);
-        query.setMaxResults(1000);
+        query.addOrder("date", true);
 
-        // TODO Need to make StatsDao to keep track to determine next 1000
+        if (maxResults > 0) {
-        // results.
+            query.setMaxResults(maxResults);
-        StatsRecord[] records = null;
-        try {
-            List<?> objects = queryByCriteria(query);
-            records = new StatsRecord[objects.size()];
-            for (int i = 0; i < records.length; i++) {
-                records[i] = (StatsRecord) objects.get(i);
-            }
-        } catch (DataAccessLayerException e) {
-            records = new StatsRecord[0];
-            statusHandler.error("Error querying the stats table", e);
         }
 
-        return records;
+        return (List<StatsRecord>) queryByCriteria(query);
     }
 }

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/handler/GraphDataHandler.java

@@ -104,7 +104,7 @@ public class GraphDataHandler implements IRequestHandler<GraphDataRequest> {
      * @throws Exception
      */
     private GraphDataResponse getMetaData() throws Exception {
-        ConfigLoader loader = new ConfigLoader();
+        ConfigLoader loader = ConfigLoader.getInstance();
         loader.load();
         List<StatisticsConfig> configList = loader.getConfigurations();
         GraphDataResponse response = new GraphDataResponse();
@@ -136,8 +136,7 @@ public class GraphDataHandler implements IRequestHandler<GraphDataRequest> {
         }
 
         if (request.getField() != null) {
-            query.addQueryParam(FIELD, request.getField(),
+            query.addQueryParam(FIELD, request.getField(), QueryOperand.EQUALS);
-                    QueryOperand.EQUALS);
         }
 
         List<?> results = dao.queryByCriteria(query);
@@ -192,7 +191,7 @@ public class GraphDataHandler implements IRequestHandler<GraphDataRequest> {
      */
     private String getDisplayUnit(String category, String type, String dataType)
             throws Exception {
-        ConfigLoader loader = new ConfigLoader();
+        ConfigLoader loader = ConfigLoader.getInstance();
         loader.load();
         List<StatisticsConfig> configList = loader.getConfigurations();
 

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/handler/StatsHandler.java

@@ -97,7 +97,7 @@ public class StatsHandler {
      * @throws Exception
      */
     protected void loadEventValidTypes() throws Exception {
-        ConfigLoader configLoader = new ConfigLoader();
+        ConfigLoader configLoader = ConfigLoader.getInstance();
         configLoader.load();
         HashSet<String> myValidEventTypes = new HashSet<String>();
 

edexOsgi/com.raytheon.uf.edex.stats/src/com/raytheon/uf/edex/stats/util/ConfigLoader.java

@@ -20,11 +20,17 @@
 package com.raytheon.uf.edex.stats.util;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
-import javax.xml.bind.JAXBException;
+import com.raytheon.uf.common.event.Event;
-
 import com.raytheon.uf.common.localization.IPathManager;
 import com.raytheon.uf.common.localization.LocalizationContext;
 import com.raytheon.uf.common.localization.LocalizationContext.LocalizationType;
@@ -34,8 +40,11 @@ import com.raytheon.uf.common.serialization.JAXBManager;
 import com.raytheon.uf.common.stats.xml.StatisticsAggregate;
 import com.raytheon.uf.common.stats.xml.StatisticsConfig;
 import com.raytheon.uf.common.stats.xml.StatisticsEvent;
+import com.raytheon.uf.common.stats.xml.StatisticsGroup;
 import com.raytheon.uf.common.status.IUFStatusHandler;
 import com.raytheon.uf.common.status.UFStatus;
+import com.raytheon.uf.common.util.ReflectionException;
+import com.raytheon.uf.common.util.ReflectionUtil;
 
 /**
  * Loads StatisticsConfig files from localization.
@@ -47,7 +56,7 @@ import com.raytheon.uf.common.status.UFStatus;
  * ------------ ---------- ----------- --------------------------
  * Aug 21, 2012            jsanchez    Updated error handling and validated config files.
  * Nov 07, 2012   1317     mpduff      Update config files.
- *
+ * Nov 29, 2012   1350     rjpeter     Updated to static, fixed localization, increased validation.
  * </pre>
  * 
  * @author jsanchez
@@ -58,20 +67,31 @@ public class ConfigLoader {
     private static final IUFStatusHandler statusHandler = UFStatus
             .getHandler(ConfigLoader.class);
 
-    private static final JAXBManager jaxbManager;
+    private final JAXBManager jaxbManager;
-    static {
+
-        try {
+    private static final ConfigLoader instance = new ConfigLoader();
-            jaxbManager = new JAXBManager(StatisticsConfig.class);
-        } catch (JAXBException e) {
-            throw new ExceptionInInitializerError(e);
-        }
-    }
 
     private final IPathManager pm = PathManagerFactory.getPathManager();
 
-    private final List<StatisticsConfig> configurations = new ArrayList<StatisticsConfig>();
+    private List<StatisticsConfig> configurations = Collections.emptyList();
 
-    private final String STATS_DIR = "stats";
+    private Map<String, StatisticsEvent> classToEventMap = Collections
+            .emptyMap();
+
+    private static final String STATS_DIR = "stats";
+
+    public static ConfigLoader getInstance() {
+        return instance;
+    }
+
+    private ConfigLoader() {
+        try {
+            jaxbManager = new JAXBManager(StatisticsConfig.class);
+            load();
+        } catch (Exception e) {
+            throw new ExceptionInInitializerError(e);
+        }
+    }
 
     /**
      * Returns a list of all StatisticsConfig files.
@@ -82,65 +102,184 @@ public class ConfigLoader {
         return configurations;
     }
 
+    /**
+     * Returns a map of event type to statistics event.
+     * 
+     * @return
+     */
+    public Map<String, StatisticsEvent> getTypeView() {
+        return classToEventMap;
+    }
+
     /**
      * Loads the StatisticsConfig files in the STATS_DIR directory.
      */
     public void load() throws Exception {
         LocalizationContext[] searchContext = pm
                 .getLocalSearchHierarchy(LocalizationType.EDEX_STATIC);
-        LocalizationFile[] localizationFiles = null;
+        Map<String, LocalizationFile> statConfs = new HashMap<String, LocalizationFile>();
+
+        // grab all stats from contexts, allowing overwrite by name
         for (LocalizationContext ctx : searchContext) {
-            localizationFiles = pm.listFiles(ctx, STATS_DIR, null, false, true);
+            LocalizationFile[] localizationFiles = pm.listFiles(ctx, STATS_DIR,
-            if (localizationFiles != null && localizationFiles.length > 0) {
+                    null, false, true);
-                break;
+            for (LocalizationFile lf : localizationFiles) {
+                String name = lf.getName();
+                if (!statConfs.containsKey(name)) {
+                    statConfs.put(name, lf);
+                }
             }
         }
 
-        if (localizationFiles != null && localizationFiles.length > 0) {
+        if (!statConfs.isEmpty()) {
-            configurations.clear();
+            List<StatisticsConfig> myConfigurations = new ArrayList<StatisticsConfig>(
-            for (LocalizationFile localizationFile : localizationFiles) {
+                    statConfs.size());
-                if (localizationFile.getFile() != null
+            Map<String, StatisticsEvent> myEvents = new HashMap<String, StatisticsEvent>();
-                        && localizationFile.getFile().exists()) {
+
-                    StatisticsConfig config = (StatisticsConfig) jaxbManager
+            for (LocalizationFile lf : statConfs.values()) {
-                            .jaxbUnmarshalFromXmlFile(localizationFile
+                StatisticsConfig config = lf.jaxbUnmarshal(
-                                    .getFile());
+                        StatisticsConfig.class, jaxbManager);
-                    config = validateAggregates(config);
+                if (config != null) {
-                    configurations.add(config);
+                    validate(myEvents, config);
+                    if (!config.getEvents().isEmpty()) {
+                        myConfigurations.add(config);
                     }
                 }
             }
+
+            configurations = myConfigurations;
+            classToEventMap = myEvents;
+        }
     }
 
     /**
-     * Removes the aggregate if its not a numerical parameter.
+     * Validates the StatisticsConfig remove invalid
+     * events/aggregates/duplicates.
      * 
      * @param config
      */
-    private StatisticsConfig validateAggregates(StatisticsConfig config)
+    private void validate(Map<String, StatisticsEvent> eventMap,
-            throws ClassNotFoundException {
+            StatisticsConfig config) {
-        List<StatisticsAggregate> aggregates = new ArrayList<StatisticsAggregate>();
+        for (Iterator<StatisticsEvent> iter = config.getEvents().iterator(); iter
-        for (StatisticsEvent event : config.getEvents()) {
+                .hasNext();) {
-            Class<?> clazz = Class.forName(event.getType());
+            StatisticsEvent event = iter.next();
-            aggregates = new ArrayList<StatisticsAggregate>();
+            String eventType = event.getType();
-            for (StatisticsAggregate aggregate : event.getAggregateList()) {
+            if (!eventMap.containsKey(eventType)) {
-                String aggregateField = aggregate.getField();
                 try {
-                    Field field = clazz.getDeclaredField(aggregateField);
+                    Class<?> clazz = Class.forName(eventType);
-                    if (!field.getType().isPrimitive()) {
+                    // verify the type is an Event
+                    clazz.asSubclass(Event.class);
+
+                    // validate groupBy fields can be found
+                    List<StatisticsGroup> groups = event.getGroupList();
+                    if (groups != null) {
+                        List<Method> groupByMethods = new ArrayList<Method>(
+                                groups.size());
+                        Set<String> currentFields = new HashSet<String>();
+
+                        for (Iterator<StatisticsGroup> groupIter = groups
+                                .iterator(); groupIter.hasNext();) {
+                            StatisticsGroup group = groupIter.next();
+                            String fieldName = group.getName();
+                            if (!currentFields.contains(fieldName)) {
+                                try {
+                                    Method m = ReflectionUtil.getGetterMethod(
+                                            clazz, fieldName);
+                                    groupByMethods.add(m);
+                                    currentFields.add(fieldName);
+                                } catch (ReflectionException e) {
+                                    groupIter.remove();
                                     statusHandler
-                                .info("'"
+                                            .warn("'"
+                                                    + fieldName
+                                                    + "' does not have getter method. Group being removed");
+                                }
+                            } else {
+                                statusHandler
+                                        .warn("'"
+                                                + fieldName
+                                                + "' already defined. Duplicate group being removed.");
+                            }
+                        }
+
+                        event.setGroupByMethods(groupByMethods);
+                    }
+
+                    // validate aggregate methods can be found and are primitive
+                    List<StatisticsAggregate> curAggregates = event
+                            .getAggregateList();
+                    List<Method> aggregateMethods = new ArrayList<Method>(
+                            curAggregates.size());
+                    Set<String> currentFields = new HashSet<String>();
+
+                    for (Iterator<StatisticsAggregate> aggrIter = curAggregates
+                            .iterator(); aggrIter.hasNext();) {
+                        StatisticsAggregate aggregate = aggrIter.next();
+                        String aggregateField = aggregate.getField();
+
+                        try {
+                            Field field = clazz
+                                    .getDeclaredField(aggregateField);
+                            if (field.getType().isPrimitive()) {
+                                if (!currentFields.contains(aggregateField)) {
+                                    try {
+                                        Method m = ReflectionUtil
+                                                .getGetterMethod(clazz,
+                                                        aggregateField);
+                                        aggregateMethods.add(m);
+                                        currentFields.add(aggregateField);
+                                    } catch (ReflectionException e) {
+                                        aggrIter.remove();
+                                        statusHandler
+                                                .warn("'"
+                                                        + aggregateField
+                                                        + "' does not have getter method. Aggregate being removed");
+                                    }
+                                } else {
+                                    aggrIter.remove();
+                                    statusHandler
+                                            .warn("'"
+                                                    + aggregateField
+                                                    + "' already defined. Duplicate aggregate being removed. ");
+                                }
+                            } else {
+                                aggrIter.remove();
+                                statusHandler
+                                        .warn("'"
                                                 + aggregateField
                                                 + "' not a primitive type. Aggregate being removed. ");
                             }
-                    aggregates.add(aggregate);
                         } catch (NoSuchFieldException e) {
-                    statusHandler.info("'" + aggregateField
+                            aggrIter.remove();
+                            statusHandler
+                                    .warn("'"
+                                            + aggregateField
                                             + "' not a valid field. Aggregate being removed. ");
                         }
                     }
-            event.setAggregateList(aggregates);
-        }
 
-        return config;
+                    if (!curAggregates.isEmpty()) {
+                        event.setAggregateMethods(aggregateMethods);
+                        eventMap.put(eventType, event);
+                    } else {
+                        iter.remove();
+                    }
+                } catch (ClassNotFoundException e) {
+                    iter.remove();
+                    statusHandler.warn("'" + eventType
+                            + "' not a valid type. Type being removed. ");
+                } catch (ClassCastException e) {
+                    iter.remove();
+                    statusHandler.warn("'" + eventType
+                            + "' not an Event type. Type being removed. ");
+                }
+            } else {
+                iter.remove();
+                statusHandler
+                        .warn("'"
+                                + eventType
+                                + "' is already defined.  StatisticsEvent being skipped.");
+            }
+        }
     }
 }