From 65c48c4ab745dd183c2719eb96fba68b9e24a725 Mon Sep 17 00:00:00 2001 From: Hao Chen Date: Tue, 28 Feb 2017 23:57:27 +0800 Subject: [PATCH 1/2] Support stream group --- .../engine/coordinator/StreamDefinition.java | 21 ++++++++++++++++--- ....metric.HadoopMetricMonitorAppProdiver.xml | 3 +++ ...urity.auditlog.HdfsAuditLogAppProvider.xml | 1 + ...uditlog.HiveQueryMonitoringAppProvider.xml | 1 + 4 files changed, 23 insertions(+), 3 deletions(-) diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java index fd5d5a623c..0cb7bd5c6f 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java @@ -52,6 +52,8 @@ public class StreamDefinition implements Serializable { // Stream data source ID private String dataSource; + private String group = "Default"; + // private String streamSource; @@ -61,19 +63,22 @@ public class StreamDefinition implements Serializable { private List columns = new ArrayList<>(); public String toString() { - return String.format("StreamDefinition[streamId=%s, dataSource=%s, description=%s, validate=%s, timeseries=%s, columns=%s", + return String.format("StreamDefinition[group=%s,streamId=%s, dataSource=%s, description=%s, validate=%s, timeseries=%s, columns=%s", + group, streamId, dataSource, description, validate, timeseries, - columns); + columns + ); } @Override public int hashCode() { return new HashCodeBuilder() .append(this.streamId) + .append(this.group) .append(this.description) .append(this.validate) .append(this.timeseries) @@ -94,11 +99,12 @@ public boolean equals(Object obj) { } StreamDefinition streamDefinition = (StreamDefinition) obj; return Objects.equals(this.streamId, streamDefinition.streamId) + && Objects.equals(this.group, streamDefinition.group) && Objects.equals(this.description, streamDefinition.description) && Objects.equals(this.validate, streamDefinition.validate) && Objects.equals(this.timeseries, streamDefinition.timeseries) && Objects.equals(this.dataSource, streamDefinition.dataSource) - && Objects.equals(this.streamSource, streamDefinition.streamSource) + && Objects.equals(this.streamSource, streamDefinition.streamSource) && Objects.equals(this.siteId, streamDefinition.siteId) && CollectionUtils.isEqualCollection(this.columns, streamDefinition.columns); } @@ -187,8 +193,17 @@ public StreamDefinition copy() { copied.setDescription(this.getDescription()); copied.setSiteId(this.getSiteId()); copied.setStreamId(this.getStreamId()); + copied.setGroup(this.getGroup()); copied.setTimeseries(this.isTimeseries()); copied.setValidate(this.isValidate()); return copied; } + + public String getGroup() { + return group; + } + + public void setGroup(String group) { + this.group = group; + } } \ No newline at end of file diff --git a/eagle-metric/eagle-hadoop-metric/src/main/resources/META-INF/providers/org.apache.eagle.metric.HadoopMetricMonitorAppProdiver.xml b/eagle-metric/eagle-hadoop-metric/src/main/resources/META-INF/providers/org.apache.eagle.metric.HadoopMetricMonitorAppProdiver.xml index 51a9257bae..eff1a70d27 100644 --- a/eagle-metric/eagle-hadoop-metric/src/main/resources/META-INF/providers/org.apache.eagle.metric.HadoopMetricMonitorAppProdiver.xml +++ b/eagle-metric/eagle-hadoop-metric/src/main/resources/META-INF/providers/org.apache.eagle.metric.HadoopMetricMonitorAppProdiver.xml @@ -110,6 +110,7 @@ HADOOP_JMX_METRIC_STREAM Hadoop JMX Metric Stream including name node, resource manager, etc. + Hadoop Metric host @@ -141,6 +142,7 @@ SYSTEM_METRIC_STREAM System Metrics Stream including CPU, Network, Disk, etc. + System Metric host @@ -176,6 +178,7 @@ HADOOP_JMX_RESOURCE_STREAM Hadoop JMX Resource Stream including name node, resource manager, etc. + Hadoop Metric host diff --git a/eagle-security/eagle-security-hdfs-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HdfsAuditLogAppProvider.xml b/eagle-security/eagle-security-hdfs-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HdfsAuditLogAppProvider.xml index 19972579a5..426a78f248 100644 --- a/eagle-security/eagle-security-hdfs-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HdfsAuditLogAppProvider.xml +++ b/eagle-security/eagle-security-hdfs-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HdfsAuditLogAppProvider.xml @@ -193,6 +193,7 @@ hdfs_audit_log_enriched_stream + Hadoop Log Hdfs Audit Log Enriched Stream true true diff --git a/eagle-security/eagle-security-hive/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HiveQueryMonitoringAppProvider.xml b/eagle-security/eagle-security-hive/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HiveQueryMonitoringAppProvider.xml index 89fbfbd2bd..12375195d4 100644 --- a/eagle-security/eagle-security-hive/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HiveQueryMonitoringAppProvider.xml +++ b/eagle-security/eagle-security-hive/src/main/resources/META-INF/providers/org.apache.eagle.security.auditlog.HiveQueryMonitoringAppProvider.xml @@ -135,6 +135,7 @@ hdfs_audit_log_stream + Hadoop Log Hdfs Audit Log Stream true true From c63b8c37552f068a338635b5ea2551ec167def46 Mon Sep 17 00:00:00 2001 From: Hao Chen Date: Wed, 1 Mar 2017 00:36:31 +0800 Subject: [PATCH 2/2] Fix StreamDefinitionTest --- .../eagle/alert/engine/coordinator/StreamDefinition.java | 2 +- .../alert/engine/coordinator/StreamDefinitionTest.java | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java index 0cb7bd5c6f..73d3991e66 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java @@ -63,7 +63,7 @@ public class StreamDefinition implements Serializable { private List columns = new ArrayList<>(); public String toString() { - return String.format("StreamDefinition[group=%s,streamId=%s, dataSource=%s, description=%s, validate=%s, timeseries=%s, columns=%s", + return String.format("StreamDefinition[group=%s, streamId=%s, dataSource=%s, description=%s, validate=%s, timeseries=%s, columns=%s", group, streamId, dataSource, diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/test/java/org/apache/eagle/alert/engine/coordinator/StreamDefinitionTest.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/test/java/org/apache/eagle/alert/engine/coordinator/StreamDefinitionTest.java index e33ef075d1..c0d9213bde 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/test/java/org/apache/eagle/alert/engine/coordinator/StreamDefinitionTest.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/test/java/org/apache/eagle/alert/engine/coordinator/StreamDefinitionTest.java @@ -35,15 +35,15 @@ public void testStreamDefinition() { streamColumns.add(new StreamColumn.Builder().name("value").type(StreamColumn.Type.DOUBLE).build()); StreamDefinition streamDefinition = new StreamDefinition(); - Assert.assertEquals("StreamDefinition[streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[]", streamDefinition.toString()); + Assert.assertEquals("StreamDefinition[group=Default, streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[]", streamDefinition.toString()); streamDefinition.setColumns(streamColumns); Assert.assertEquals(3, streamDefinition.getColumnIndex("data")); Assert.assertEquals(-1, streamDefinition.getColumnIndex("DATA")); Assert.assertEquals(-1, streamDefinition.getColumnIndex("isYhd")); - Assert.assertEquals("StreamDefinition[streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[StreamColumn=name[name], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[host], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[flag], type=[bool], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[data], type=[long], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[value], type=[double], defaultValue=[null], required=[false], nodataExpression=[null]]", streamDefinition.toString()); + Assert.assertEquals("StreamDefinition[group=Default, streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[StreamColumn=name[name], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[host], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[flag], type=[bool], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[data], type=[long], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[value], type=[double], defaultValue=[null], required=[false], nodataExpression=[null]]", streamDefinition.toString()); StreamDefinition streamDefinition1 = streamDefinition.copy(); - Assert.assertEquals("StreamDefinition[streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[StreamColumn=name[name], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[host], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[flag], type=[bool], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[data], type=[long], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[value], type=[double], defaultValue=[null], required=[false], nodataExpression=[null]]", streamDefinition1.toString()); + Assert.assertEquals("StreamDefinition[group=Default, streamId=null, dataSource=null, description=null, validate=false, timeseries=false, columns=[StreamColumn=name[name], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[host], type=[string], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[flag], type=[bool], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[data], type=[long], defaultValue=[null], required=[false], nodataExpression=[null], StreamColumn=name[value], type=[double], defaultValue=[null], required=[false], nodataExpression=[null]]", streamDefinition1.toString()); Assert.assertTrue(streamDefinition1.equals(streamDefinition)); Assert.assertFalse(streamDefinition1 == streamDefinition);