eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ji...@apache.org
Subject incubator-eagle git commit: [MINOR] Enhance new publisher AlertEagleStorePlugin
Date Fri, 28 Oct 2016 02:10:59 GMT
Repository: incubator-eagle
Updated Branches:
  refs/heads/master f07777706 -> 2e5cb719c


[MINOR] Enhance new publisher AlertEagleStorePlugin

https://issues.apache.org/jira/browse/EAGLE-681

* implement alert apis on jdbc & mongodb
* add a rest api for searching a alert by alertId

Author: Zhao, Qingwen <qingwzhao@apache.org>

Closes #575 from qingwen220/EAGLE-681.


Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/2e5cb719
Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/2e5cb719
Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/2e5cb719

Branch: refs/heads/master
Commit: 2e5cb719c3a49da920d8d6807e10bcfdb6e55934
Parents: f077777
Author: Zhao, Qingwen <qingwzhao@apache.org>
Authored: Fri Oct 28 10:10:50 2016 +0800
Committer: zombieJ <smith3816@gmail.com>
Committed: Fri Oct 28 10:10:50 2016 +0800

----------------------------------------------------------------------
 .../alert/engine/model/AlertPublishEvent.java   |  5 +-
 .../alert/engine/model/AlertStreamEvent.java    |  4 +-
 .../AbstractMetadataChangeNotifyService.java    |  4 +-
 .../impl/ZKMetadataChangeNotifyService.java     |  4 +-
 .../publisher/AlertPublishSpecListener.java     |  2 +-
 .../publisher/impl/AlertEagleStorePlugin.java   |  2 +-
 .../alert/engine/runner/AlertPublisherBolt.java |  3 +-
 .../metadata/resource/MetadataResource.java     |  8 +-
 .../eagle/alert/metadata/IMetadataDao.java      |  2 +
 .../eagle/alert/metadata/MetadataUtils.java     |  4 +
 .../metadata/impl/InMemMetadataDaoImpl.java     | 10 +++
 .../metadata/impl/JdbcDatabaseHandler.java      | 94 ++++++--------------
 .../metadata/impl/JdbcMetadataDaoImpl.java      | 11 ++-
 .../metadata/impl/MongoMetadataDaoImpl.java     | 23 ++++-
 .../alert/resource/impl/MongoImplTest.java      | 11 +++
 15 files changed, 106 insertions(+), 81 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertPublishEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertPublishEvent.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertPublishEvent.java
index 9d73c2d..3453e55 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertPublishEvent.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertPublishEvent.java
@@ -96,7 +96,8 @@ public class AlertPublishEvent {
         AlertPublishEvent alertEvent = new AlertPublishEvent();
         alertEvent.setAlertId(UUID.randomUUID().toString());
         alertEvent.setPolicyId(event.getPolicyId());
-        if (event.getExtraData() != null) {
+        alertEvent.setAlertTimestamp(event.getCreatedTime());
+        if (event.getExtraData() != null && !event.getExtraData().isEmpty()) {
             alertEvent.setSiteId(event.getExtraData().get(SITE_ID_KEY).toString());
             alertEvent.setPolicyValue(event.getExtraData().get(POLICY_VALUE_KEY).toString());
             alertEvent.setAppIds((List<String>) event.getExtraData().get(APP_IDS_KEY));
@@ -104,4 +105,6 @@ public class AlertPublishEvent {
         alertEvent.setAlertData(event.getDataMap());
         return alertEvent;
     }
+
 }
+

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
index e502244..442c885 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
@@ -72,8 +72,8 @@ public class AlertStreamEvent extends StreamEvent {
             }
         }
         return String.format("AlertStreamEvent[stream=%S,timestamp=%s,data=[%s], policyId=%s,
createdBy=%s, metaVersion=%s]",
-            this.getStreamId(), DateTimeUtil.millisecondsToHumanDateWithMilliseconds(this.getTimestamp()),
-            StringUtils.join(dataStrings, ","), this.getPolicyId(), this.getCreatedBy(),
this.getMetaVersion());
+                this.getStreamId(), DateTimeUtil.millisecondsToHumanDateWithMilliseconds(this.getTimestamp()),
+                StringUtils.join(dataStrings, ","), this.getPolicyId(), this.getCreatedBy(),
this.getMetaVersion());
     }
 
     public String getCreatedBy() {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
index bbb6ca1..3264e61 100755
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/AbstractMetadataChangeNotifyService.java
@@ -103,8 +103,8 @@ public abstract class AbstractMetadataChangeNotifyService implements IMetadataCh
         alertPublishSpecListeners.forEach(s -> s.onAlertPublishSpecChange(alertPublishSpec,
sds));
     }
 
-    protected void notifyAlertPublishBolt(Map<String, PolicyDefinition> pds) {
-        alertPublishSpecListeners.forEach(s -> s.onAlertPolicyChange(pds));
+    protected void notifyAlertPublishBolt(Map<String, PolicyDefinition> pds, Map<String,
StreamDefinition> sds) {
+        alertPublishSpecListeners.forEach(s -> s.onAlertPolicyChange(pds, sds));
     }
 
     public void close() throws IOException {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
index 7bf4984..8dc6fdd 100755
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/coordinator/impl/ZKMetadataChangeNotifyService.java
@@ -123,7 +123,7 @@ public class ZKMetadataChangeNotifyService extends AbstractMetadataChangeNotifyS
                     prePopulate(alertSpec, state.getPolicySnapshots());
                     notifyAlertBolt(alertSpec, sds);
                     if (state.getPublishSpecs().get(topologyId) != null) {
-                        notifyAlertPublishBolt(listToMap(state.getPolicySnapshots()));
+                        notifyAlertPublishBolt(listToMap(state.getPolicySnapshots()), sds);
                     }
                 }
                 break;
@@ -134,7 +134,7 @@ public class ZKMetadataChangeNotifyService extends AbstractMetadataChangeNotifyS
                 } else {
                     notifyAlertPublishBolt(pubSpec, sds);
                     if (state.getAlertSpecs().get(topologyId) != null) {
-                        notifyAlertPublishBolt(listToMap(state.getPolicySnapshots()));
+                        notifyAlertPublishBolt(listToMap(state.getPolicySnapshots()), sds);
                     }
                 }
                 break;

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
index d8429ca..b5401cb 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java
@@ -25,5 +25,5 @@ import java.util.Map;
 public interface AlertPublishSpecListener {
     void onAlertPublishSpecChange(PublishSpec spec, Map<String, StreamDefinition> sds);
     
-    void onAlertPolicyChange(Map<String, PolicyDefinition> pds);
+    void onAlertPolicyChange(Map<String, PolicyDefinition> pds, Map<String, StreamDefinition>
sds);
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AlertEagleStorePlugin.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AlertEagleStorePlugin.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AlertEagleStorePlugin.java
index b9b5f2e..f517078 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AlertEagleStorePlugin.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AlertEagleStorePlugin.java
@@ -78,4 +78,4 @@ public class AlertEagleStorePlugin extends AbstractPublishPlugin {
         return null;
     }
 
-}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/runner/AlertPublisherBolt.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/runner/AlertPublisherBolt.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/runner/AlertPublisherBolt.java
index f0ad0ae..95f2a8f 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/runner/AlertPublisherBolt.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/runner/AlertPublisherBolt.java
@@ -118,8 +118,9 @@ public class AlertPublisherBolt extends AbstractStreamBolt implements
AlertPubli
     }
 
     @Override
-    public void onAlertPolicyChange(Map<String, PolicyDefinition> pds) {
+    public void onAlertPolicyChange(Map<String, PolicyDefinition> pds, Map<String,
StreamDefinition> sds) {
         this.policyDefinitionMap = pds;
+        this.streamDefinitionMap = sds;
     }
 
     private void wrapAlertPublishEvent(AlertStreamEvent event) {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata-service/src/main/java/org/apache/eagle/service/metadata/resource/MetadataResource.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata-service/src/main/java/org/apache/eagle/service/metadata/resource/MetadataResource.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata-service/src/main/java/org/apache/eagle/service/metadata/resource/MetadataResource.java
index f5b34b8..05455f0 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata-service/src/main/java/org/apache/eagle/service/metadata/resource/MetadataResource.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata-service/src/main/java/org/apache/eagle/service/metadata/resource/MetadataResource.java
@@ -17,7 +17,6 @@
 package org.apache.eagle.service.metadata.resource;
 
 import com.google.common.base.Preconditions;
-import javafx.scene.control.Alert;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
 import org.apache.eagle.alert.coordination.model.ScheduleState;
@@ -33,7 +32,6 @@ import org.apache.eagle.alert.metadata.impl.MetadataDaoFactory;
 import org.apache.eagle.alert.metadata.resource.Models;
 import org.apache.eagle.alert.metadata.resource.OpResult;
 import com.google.inject.Inject;
-import org.apache.storm.zookeeper.Op;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -501,6 +499,12 @@ public class MetadataResource {
         return dao.listAlertPublishEvent(size);
     }
 
+    @Path("/alerts/{alertId}")
+    @GET
+    public AlertPublishEvent getAlertPublishEvent(@PathParam("alertId") String alertId) {
+        return dao.getAlertPublishEvent(alertId);
+    }
+
     @Path("/topologies/{topologyName}")
     @DELETE
     public OpResult removeTopology(@PathParam("topologyName") String topologyName) {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/IMetadataDao.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/IMetadataDao.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/IMetadataDao.java
index d25c548..33a3c39 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/IMetadataDao.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/IMetadataDao.java
@@ -80,6 +80,8 @@ public interface IMetadataDao extends Closeable {
 
     List<AlertPublishEvent> listAlertPublishEvent(int size);
 
+    AlertPublishEvent getAlertPublishEvent(String alertId);
+
     OpResult addAlertPublishEvent(AlertPublishEvent event);
 
     ScheduleState getScheduleState(String versionId);

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/MetadataUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/MetadataUtils.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/MetadataUtils.java
index 7015a53..841a0a0 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/MetadataUtils.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/MetadataUtils.java
@@ -23,6 +23,7 @@ import org.apache.eagle.alert.coordination.model.internal.PolicyAssignment;
 import org.apache.eagle.alert.engine.coordinator.PublishmentType;
 import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
 import com.typesafe.config.Config;
+import org.apache.eagle.alert.engine.model.AlertPublishEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,6 +50,9 @@ public class MetadataUtils {
         if (t instanceof ScheduleState) {
             return ((ScheduleState) t).getVersion();
         }
+        if (t instanceof AlertPublishEvent) {
+            return ((AlertPublishEvent) t).getAlertId();
+        }
 
         try {
             Method m = t.getClass().getMethod("getName");

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/InMemMetadataDaoImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/InMemMetadataDaoImpl.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/InMemMetadataDaoImpl.java
index 0de13f1..ad7d353 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/InMemMetadataDaoImpl.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/InMemMetadataDaoImpl.java
@@ -207,6 +207,16 @@ public class InMemMetadataDaoImpl implements IMetadataDao {
     }
 
     @Override
+    public AlertPublishEvent getAlertPublishEvent(String alertId) {
+        Optional<AlertPublishEvent> op = alerts.stream().filter(alert -> alert.getAlertId().equals(alertId)).findAny();
+        if (op.isPresent()) {
+            return op.get();
+        } else {
+            return null;
+        }
+    }
+
+    @Override
     public OpResult addAlertPublishEvent(AlertPublishEvent event) {
         alerts.add(event);
         OpResult result = new OpResult();

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcDatabaseHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcDatabaseHandler.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcDatabaseHandler.java
index e484e09..d9580f1 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcDatabaseHandler.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcDatabaseHandler.java
@@ -42,8 +42,11 @@ public class JdbcDatabaseHandler {
     private static final String DELETE_STATEMENT = "DELETE FROM %s WHERE id=?";
     private static final String UPDATE_STATEMENT = "UPDATE %s set value=? WHERE id=?";
     private static final String QUERY_ALL_STATEMENT = "SELECT value FROM %s";
-    private static final String QUERY_CONDITION_STATEMENT = "SELECT value FROM %s WHERE id=?";
+    private static final String QUERY_CONDITION_STATEMENT = "SELECT value FROM %s WHERE id=%s";
     private static final String QUERY_ORDERBY_STATEMENT = "SELECT value FROM %s ORDER BY
id %s";
+    private static final String QUERY_ALL_STATEMENT_WITH_SIZE = "SELECT value FROM %s limit
%s";
+
+    public enum SortType { DESC, ASC }
 
     private Map<String, String> tblNameMap = new HashMap<>();
 
@@ -151,58 +154,20 @@ public class JdbcDatabaseHandler {
 
     public <T> List<T> list(Class<T> clz) {
         String tb = getTableName(clz.getSimpleName());
-        List<T> result = new LinkedList<>();
-        try {
-            Statement statement = connection.createStatement();
-            ResultSet rs = statement.executeQuery(String.format(QUERY_ALL_STATEMENT, tb));
-            while (rs.next()) {
-                //String key = rs.getString(1);
-                String json = rs.getString(1);
-                try {
-                    T obj = mapper.readValue(json, clz);
-                    result.add(obj);
-                } catch (IOException e) {
-                    LOG.error("deserialize config item failed!", e);
-                }
-            }
-            rs.close();
-            statement.close();
-        } catch (SQLException e) {
-            e.printStackTrace();
-        }
-        return result;
+        String query = String.format(QUERY_ALL_STATEMENT, tb);
+        return executeSelectStatement(clz, query);
+    }
+
+    public <T> List<T> listSubset(Class<T> clz, int size) {
+        String tb = getTableName(clz.getSimpleName());
+        String query = String.format(QUERY_ALL_STATEMENT_WITH_SIZE, tb, size);
+        return executeSelectStatement(clz, query);
     }
 
     public <T> T listWithFilter(String key, Class<T> clz) {
         String tb = getTableName(clz.getSimpleName());
-        List<T> result = new LinkedList<>();
-        PreparedStatement statement = null;
-        try {
-            statement = connection.prepareStatement(String.format(QUERY_CONDITION_STATEMENT,
tb));
-            statement.setString(1, key);
-            ResultSet rs = statement.executeQuery();
-            while (rs.next()) {
-                //String key = rs.getString(1);
-                String json = rs.getString(1);
-                try {
-                    T obj = mapper.readValue(json, clz);
-                    result.add(obj);
-                } catch (IOException e) {
-                    LOG.error("deserialize config item failed!", e);
-                }
-            }
-            rs.close();
-        } catch (SQLException e) {
-            e.printStackTrace();
-        } finally {
-            if (statement != null) {
-                try {
-                    statement.close();
-                } catch (SQLException e) {
-                    LOG.warn("Close statement failed");
-                }
-            }
-        }
+        String query = String.format(QUERY_CONDITION_STATEMENT, tb, key);
+        List<T> result = executeSelectStatement(clz, query);
         if (result.isEmpty()) {
             return null;
         } else {
@@ -212,11 +177,21 @@ public class JdbcDatabaseHandler {
 
     public <T> T listTop(Class<T> clz, String sortType) {
         String tb = getTableName(clz.getSimpleName());
+        String query = String.format(QUERY_ORDERBY_STATEMENT, tb, sortType);
+        List<T> result = executeSelectStatement(clz, query);
+        if (result.isEmpty()) {
+            return null;
+        } else {
+            return result.get(0);
+        }
+    }
+
+    public <T> List<T> executeSelectStatement(Class<T> clz, String query)
{
+        String tb = getTableName(clz.getSimpleName());
         List<T> result = new LinkedList<>();
-        PreparedStatement statement = null;
         try {
-            statement = connection.prepareStatement(String.format(QUERY_ORDERBY_STATEMENT,
tb, sortType));
-            ResultSet rs = statement.executeQuery();
+            Statement statement = connection.createStatement();
+            ResultSet rs = statement.executeQuery(query);
             while (rs.next()) {
                 //String key = rs.getString(1);
                 String json = rs.getString(1);
@@ -228,22 +203,11 @@ public class JdbcDatabaseHandler {
                 }
             }
             rs.close();
+            statement.close();
         } catch (SQLException e) {
             e.printStackTrace();
-        } finally {
-            if (statement != null) {
-                try {
-                    statement.close();
-                } catch (SQLException e) {
-                    LOG.warn("Close statement failed");
-                }
-            }
-        }
-        if (result.isEmpty()) {
-            return null;
-        } else {
-            return result.get(0);
         }
+        return result;
     }
 
     public <T> OpResult remove(String clzName, String key) {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcMetadataDaoImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcMetadataDaoImpl.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcMetadataDaoImpl.java
index 0b595ac..adf26d8 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcMetadataDaoImpl.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/JdbcMetadataDaoImpl.java
@@ -74,7 +74,12 @@ public class JdbcMetadataDaoImpl implements IMetadataDao {
 
     @Override
     public List<AlertPublishEvent> listAlertPublishEvent(int size) {
-        return null;
+        return handler.listSubset(AlertPublishEvent.class, size);
+    }
+
+    @Override
+    public AlertPublishEvent getAlertPublishEvent(String alertId) {
+        return handler.listWithFilter(alertId, AlertPublishEvent.class);
     }
 
     @Override
@@ -85,7 +90,7 @@ public class JdbcMetadataDaoImpl implements IMetadataDao {
 
     @Override
     public ScheduleState getScheduleState() {
-        return handler.listTop(ScheduleState.class, "DESC");
+        return handler.listTop(ScheduleState.class, JdbcDatabaseHandler.SortType.DESC.toString());
     }
 
     @Override
@@ -110,7 +115,7 @@ public class JdbcMetadataDaoImpl implements IMetadataDao {
 
     @Override
     public OpResult addAlertPublishEvent(AlertPublishEvent event) {
-        return null;
+        return handler.addOrReplace(AlertPublishEvent.class.getSimpleName(), event);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/MongoMetadataDaoImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/MongoMetadataDaoImpl.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/MongoMetadataDaoImpl.java
index ab42e7d..f8186ac 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/MongoMetadataDaoImpl.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/main/java/org/apache/eagle/alert/metadata/impl/MongoMetadataDaoImpl.java
@@ -125,6 +125,15 @@ public class MongoMetadataDaoImpl implements IMetadataDao {
             publishmentType.createIndex(doc1, io1);
         }
 
+        alerts = db.getCollection("alerts");
+        {
+            IndexOptions io1 = new IndexOptions().background(true).unique(true).name("alertIndex");
+            BsonDocument doc1 = new BsonDocument();
+            doc1.append("alertId", new BsonInt32(1));
+            alerts.createIndex(doc1, io1);
+        }
+
+
         // below is for schedule_specs and its splitted collections
         BsonDocument doc1 = new BsonDocument();
         IndexOptions io1 = new IndexOptions().background(true).name("versionIndex");
@@ -179,6 +188,8 @@ public class MongoMetadataDaoImpl implements IMetadataDao {
             filter.append("streamId", new BsonString(MetadataUtils.getKey(t)));
         } else if (t instanceof PublishmentType) {
             filter.append("type", new BsonString(MetadataUtils.getKey(t)));
+        } else if (t instanceof AlertPublishEvent) {
+            filter.append("alertId", new BsonString(MetadataUtils.getKey(t)));
         } else {
             filter.append("name", new BsonString(MetadataUtils.getKey(t)));
         }
@@ -307,12 +318,22 @@ public class MongoMetadataDaoImpl implements IMetadataDao {
 
     @Override
     public List<AlertPublishEvent> listAlertPublishEvent(int size) {
+        return list(alerts, AlertPublishEvent.class);
+    }
+
+    @Override
+    public AlertPublishEvent getAlertPublishEvent(String alertId) {
+        List<AlertPublishEvent> results = list(alerts, AlertPublishEvent.class);
+        Optional<AlertPublishEvent> op = results.stream().filter(alert -> alert.getAlertId().equals(alertId)).findAny();
+        if (op.isPresent()) {
+            return op.get();
+        }
         return null;
     }
 
     @Override
     public OpResult addAlertPublishEvent(AlertPublishEvent event) {
-        return null;
+        return addOrReplace(alerts, event);
     }
 
     private <T> OpResult addOne(MongoCollection<Document> collection, T t) {

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/2e5cb719/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/test/java/org/apache/eagle/service/alert/resource/impl/MongoImplTest.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/test/java/org/apache/eagle/service/alert/resource/impl/MongoImplTest.java
b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/test/java/org/apache/eagle/service/alert/resource/impl/MongoImplTest.java
index 7f08133..ff83b80 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/test/java/org/apache/eagle/service/alert/resource/impl/MongoImplTest.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-metadata-parent/alert-metadata/src/test/java/org/apache/eagle/service/alert/resource/impl/MongoImplTest.java
@@ -32,6 +32,7 @@ import org.apache.eagle.alert.coordination.model.internal.PolicyAssignment;
 import org.apache.eagle.alert.coordination.model.internal.StreamGroup;
 import org.apache.eagle.alert.coordination.model.internal.Topology;
 import org.apache.eagle.alert.engine.coordinator.*;
+import org.apache.eagle.alert.engine.model.AlertPublishEvent;
 import org.apache.eagle.alert.metadata.IMetadataDao;
 import org.apache.eagle.alert.metadata.impl.MongoMetadataDaoImpl;
 import org.apache.eagle.alert.metadata.resource.OpResult;
@@ -206,6 +207,16 @@ public class MongoImplTest {
             assigns = dao.listStreams();
             Assert.assertEquals(0, assigns.size());
         }
+        // alert
+        {
+            AlertPublishEvent alert = new AlertPublishEvent();
+            alert.setAlertTimestamp(System.currentTimeMillis());
+            alert.setAlertId(UUID.randomUUID().toString());
+            OpResult result = dao.addAlertPublishEvent(alert);
+            Assert.assertEquals(200, result.code);
+            List<AlertPublishEvent> alerts = dao.listAlertPublishEvent(2);
+            Assert.assertEquals(1, alerts.size());
+        }
     }
 
     private void test_addstate() {


Mime
View raw message