http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java deleted file mode 100644 index 1df24b9..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher; - -import java.util.List; - -import org.apache.eagle.alert.engine.model.AlertStreamEvent; - -/** - * Dedup Eagle entities. - * TODO: need support de-dup field values - */ -public interface AlertDeduplicator { - - List<AlertStreamEvent> dedup(AlertStreamEvent event); - - void setDedupIntervalMin(String intervalMin); - -}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java deleted file mode 100644 index fdb01a7..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.engine.coordinator.Publishment; - -import java.util.List; - -public interface AlertPublishListener { - void onPublishChange(List<Publishment> added, - List<Publishment> removed, - List<Publishment> afterModified, - List<Publishment> beforeModified); -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java deleted file mode 100644 index 5f97743..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher; - -import java.io.Closeable; -import java.util.List; -import java.util.Map; - -import org.apache.eagle.alert.engine.coordinator.Publishment; -import org.apache.eagle.alert.engine.model.AlertStreamEvent; -import org.apache.eagle.alert.engine.publisher.impl.PublishStatus; - -import com.typesafe.config.Config; - -/** - * Created on 2/10/16. - * Notification Plug-in interface which provide abstraction layer to notify to different system - */ -public interface AlertPublishPlugin extends Closeable { - /** - * Init alert publish plugin. - * - * @param config - * @param publishment - * @param configProperties - storm config that would be useful for some implementation - * @throws Exception - */ - @SuppressWarnings("rawtypes") - void init(Config config, Publishment publishment, Map configProperties) throws Exception; - - void update(String dedupIntervalMin, Map<String, Object> pluginProperties); - - void close(); - - void onAlert(AlertStreamEvent event) throws Exception; - - List<AlertStreamEvent> dedup(AlertStreamEvent event); - - PublishStatus getStatus(); - -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPluginProvider.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPluginProvider.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPluginProvider.java deleted file mode 100644 index 77eea40..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPluginProvider.java +++ /dev/null @@ -1,24 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.engine.coordinator.PublishmentType; - -public interface AlertPublishPluginProvider { - PublishmentType getPluginType(); -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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 deleted file mode 100644 index b5401cb..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishSpecListener.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.coordination.model.PublishSpec; -import org.apache.eagle.alert.engine.coordinator.PolicyDefinition; -import org.apache.eagle.alert.engine.coordinator.StreamDefinition; - -import java.util.Map; - -public interface AlertPublishSpecListener { - void onAlertPublishSpecChange(PublishSpec spec, Map<String, StreamDefinition> sds); - - void onAlertPolicyChange(Map<String, PolicyDefinition> pds, Map<String, StreamDefinition> sds); -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java deleted file mode 100644 index 9717e2b..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublisher.java +++ /dev/null @@ -1,37 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import java.io.Serializable; -import java.util.Map; - -import org.apache.eagle.alert.engine.coordinator.PublishPartition; -import org.apache.eagle.alert.engine.model.AlertStreamEvent; - -import com.typesafe.config.Config; - -public interface AlertPublisher extends AlertPublishListener, Serializable { - @SuppressWarnings("rawtypes") - void init(Config config, Map stormConfig); - - String getName(); - - void nextEvent(PublishPartition partition, AlertStreamEvent event); - - void close(); - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java deleted file mode 100644 index 991bba0..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertSink.java +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import java.io.Serializable; - -public interface AlertSink extends Serializable { - void open() throws Exception; - - void close() throws Exception; -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertStreamFilter.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertStreamFilter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertStreamFilter.java deleted file mode 100644 index 71c2a8e..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertStreamFilter.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.engine.model.AlertStreamEvent; - -public interface AlertStreamFilter { - /** - * Filter Stream Event, if skipped, return null. - */ - AlertStreamEvent filter(AlertStreamEvent event); -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PipeStreamFilter.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PipeStreamFilter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PipeStreamFilter.java deleted file mode 100644 index a6cc3e5..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PipeStreamFilter.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.engine.model.AlertStreamEvent; - -import java.util.ArrayList; -import java.util.List; - -public class PipeStreamFilter implements AlertStreamFilter { - - private final List<AlertStreamFilter> filters; - - public PipeStreamFilter(AlertStreamFilter... filters) { - this.filters = new ArrayList<>(); - for (AlertStreamFilter filter : filters) { - this.filters.add(filter); - } - } - - @Override - public AlertStreamEvent filter(AlertStreamEvent event) { - AlertStreamEvent current = event; - for (AlertStreamFilter filter : this.filters) { - if (current == null) { - return null; - } - current = filter.filter(current); - } - return current; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java deleted file mode 100644 index e716fbe..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishConstants.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package org.apache.eagle.alert.engine.publisher; - -public class PublishConstants { - public static final String NOTIFICATION_TYPE = "type"; - public static final String EMAIL_NOTIFICATION = "email"; - public static final String KAFKA_STORE = "kafka"; - public static final String EAGLE_STORE = "eagleStore"; - - // email specific constants - public static final String SUBJECT = "subject"; - public static final String SENDER = "sender"; - public static final String RECIPIENTS = "recipients"; - public static final String TEMPLATE = "template"; - - // kafka specific constants - public static final String TOPIC = "topic"; - public static final String BROKER_LIST = "kafka_broker"; - public static final String WRITE_MODE = "kafka_write_mode"; - - // local rotated file constants - public static final String FILE_NAME = "fileName"; - public static final String ROTATE_EVERY_KB = "rotate_every_kb"; - public static final String NUMBER_OF_FILES = "number_of_files"; - - // slack specific constants - public static final String TOKEN = "token"; - public static final String CHANNELS = "channels"; - public static final String SEVERITYS = "severitys"; - public static final String URL_TEMPLATE = "urltemplate"; - - public static final String ALERT_EMAIL_TIME_PROPERTY = "timestamp"; - public static final String ALERT_EMAIL_COUNT_PROPERTY = "count"; - public static final String ALERT_EMAIL_ALERTLIST_PROPERTY = "alertList"; - public static final String ALERT_EMAIL_ORIGIN_PROPERTY = "alertEmailOrigin"; - - public static final String ALERT_EMAIL_SUBJECT = "alertSubject"; - public static final String ALERT_EMAIL_BODY = "alertBody"; - public static final String ALERT_EMAIL_STREAM_ID = "streamId"; - public static final String ALERT_EMAIL_TIME = "alertTime"; - public static final String ALERT_EMAIL_POLICY_ID = "policyId"; - public static final String ALERT_EMAIL_ALERT_ID = "alertId"; - public static final String ALERT_EMAIL_ALERT_CATEGORY = "alertCategory"; - public static final String ALERT_EMAIL_ALERT_SEVERITY = "alertSeverity"; - public static final String ALERT_EMAIL_ALERT_DATA = "alertData"; - public static final String ALERT_EMAIL_ALERT_DATA_DESC = "alertDataDesc"; - public static final String ALERT_EMAIL_CREATOR = "creator"; - public static final String ALERT_EMAIL_ALERT_DETAIL_URL = "alertDetailUrl"; - public static final String ALERT_EMAIL_POLICY_DETAIL_URL = "policyDetailUrl"; - public static final String ALERT_EMAIL_HOME_URL = "homeUrl"; - public static final String ALERT_EMAIL_VERSION = "version"; -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishementTypeLoader.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishementTypeLoader.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishementTypeLoader.java deleted file mode 100644 index 820d70e..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/PublishementTypeLoader.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * <p/> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p/> - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.eagle.alert.engine.publisher; - -import org.apache.eagle.alert.engine.coordinator.PublishmentType; -import org.apache.eagle.common.utils.ReflectionsHelper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.LinkedList; -import java.util.List; - -public class PublishementTypeLoader { - private static final Logger LOGGER = LoggerFactory.getLogger(PublishementTypeLoader.class); - - private final List<PublishmentType> publishmentTypeSet; - - private PublishementTypeLoader() { - this.publishmentTypeSet = new LinkedList<>(); - LOGGER.info("Loading alert publish plugins ..."); - for (Class<? extends AlertPublishPluginProvider> clazz: ReflectionsHelper.getInstance().getSubTypesOf(AlertPublishPluginProvider.class)) { - LOGGER.debug("Loading alert publish plugin: {}", clazz); - try { - PublishmentType type = clazz.newInstance().getPluginType(); - this.publishmentTypeSet.add(type); - LOGGER.info("Loaded alert publish plugin {}:{}", type.getName(), type.getType()); - } catch (InstantiationException | IllegalAccessException e) { - LOGGER.error("Failed to get instantiate alert publish plugin provider: {}", clazz, e); - } - } - LOGGER.info("Loaded {} alert publish plugins", this.publishmentTypeSet.size()); - } - - private static final PublishementTypeLoader INSTANCE = new PublishementTypeLoader(); - - public static List<PublishmentType> loadPublishmentTypes() { - return INSTANCE.getPublishmentTypes(); - } - - public List<PublishmentType> getPublishmentTypes() { - return publishmentTypeSet; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupCache.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupCache.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupCache.java deleted file mode 100644 index abb83d6..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupCache.java +++ /dev/null @@ -1,223 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher.dedup; - -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedDeque; - -import org.apache.commons.lang.time.DateUtils; -import org.apache.commons.lang3.StringUtils; -import org.apache.eagle.alert.engine.coordinator.StreamDefinition; -import org.apache.eagle.alert.engine.model.AlertStreamEvent; -import org.apache.eagle.alert.engine.publisher.impl.EventUniq; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.base.Objects; -import com.typesafe.config.Config; - -public class DedupCache { - - private static final Logger LOG = LoggerFactory.getLogger(DedupCache.class); - - private static final long CACHE_MAX_EXPIRE_TIME_IN_DAYS = 30; - - public static final String DEDUP_COUNT = "dedupCount"; - public static final String DOC_ID = "docId"; - public static final String DEDUP_FIRST_OCCURRENCE = "dedupFirstOccurrenceTime"; - - private long lastUpdated = -1; - private Map<EventUniq, ConcurrentLinkedDeque<DedupValue>> events = new ConcurrentHashMap<EventUniq, ConcurrentLinkedDeque<DedupValue>>(); - - @SuppressWarnings("unused") - private Config config; - - private String publishName; - - public DedupCache(Config config, String publishName) { - this.config = config; - this.publishName = publishName; - } - - public Map<EventUniq, ConcurrentLinkedDeque<DedupValue>> getEvents() { - if (lastUpdated < 0 - || System.currentTimeMillis() - lastUpdated > CACHE_MAX_EXPIRE_TIME_IN_DAYS * DateUtils.MILLIS_PER_DAY - || events.size() <= 0) { - lastUpdated = System.currentTimeMillis(); - } - return events; - } - - public boolean contains(EventUniq eventEniq) { - return this.getEvents().containsKey(eventEniq); - } - - public void removeEvent(EventUniq eventEniq) { - if (this.contains(eventEniq)) { - this.events.remove(eventEniq); - } - } - - public List<AlertStreamEvent> dedup(AlertStreamEvent event, EventUniq eventEniq, - String dedupStateField, String stateFieldValue, - String stateCloseValue) { - DedupValue[] dedupValues = this.addOrUpdate(eventEniq, event, stateFieldValue, stateCloseValue); - if (dedupValues != null) { - // any of dedupValues won't be null - if (dedupValues.length == 2) { - // emit last event which includes count of dedup events & new state event - return Arrays.asList( - this.mergeEventWithDedupValue(event, dedupValues[0], dedupStateField), - this.mergeEventWithDedupValue(event, dedupValues[1], dedupStateField)); - } else if (dedupValues.length == 1) { - //populate firstOccurrenceTime & count - return Arrays.asList(this.mergeEventWithDedupValue(event, dedupValues[0], dedupStateField)); - } - } - // duplicated, will be ignored - return null; - } - - public synchronized DedupValue[] addOrUpdate(EventUniq eventEniq, AlertStreamEvent event, String stateFieldValue, String stateCloseValue) { - Map<EventUniq, ConcurrentLinkedDeque<DedupValue>> events = this.getEvents(); - if (!events.containsKey(eventEniq) - || (events.containsKey(eventEniq) - && events.get(eventEniq).size() > 0 - && !StringUtils.equalsIgnoreCase(stateFieldValue, - events.get(eventEniq).getLast().getStateFieldValue()))) { - DedupValue[] dedupValues = this.add(eventEniq, event, stateFieldValue, stateCloseValue); - return dedupValues; - } else { - // update count - this.updateCount(eventEniq); - return null; - } - } - - private DedupValue[] add(EventUniq eventEniq, AlertStreamEvent event, String stateFieldValue, String stateCloseValue) { - DedupValue dedupValue = null; - if (!events.containsKey(eventEniq)) { - dedupValue = createDedupValue(eventEniq, event, stateFieldValue); - ConcurrentLinkedDeque<DedupValue> dedupValues = new ConcurrentLinkedDeque<>(); - dedupValues.add(dedupValue); - // skip the event which put failed due to concurrency - events.put(eventEniq, dedupValues); - LOG.info("{} Add new dedup key {}, and value {}", this.publishName, eventEniq, dedupValues); - } else if (!StringUtils.equalsIgnoreCase(stateFieldValue, - events.get(eventEniq).getLast().getStateFieldValue())) { - // existing a de-dup value, try update or reset - DedupValue lastDedupValue = events.get(eventEniq).getLast(); - dedupValue = updateDedupValue(lastDedupValue, eventEniq, event, stateFieldValue, stateCloseValue); - LOG.info("{} Update dedup key {}, and value {}", this.publishName, eventEniq, dedupValue); - } - if (dedupValue == null) { - return null; - } - return new DedupValue[] {dedupValue}; - } - - private DedupValue updateDedupValue(DedupValue lastDedupValue, EventUniq eventEniq, AlertStreamEvent event, String stateFieldValue, String stateCloseValue) { - if (lastDedupValue.getFirstOccurrence() >= eventEniq.timestamp) { - // if dedup value happens later then event, dedup state changes. - return null; - } - - if (lastDedupValue.getStateFieldValue().equals(stateCloseValue) - && eventEniq.timestamp < lastDedupValue.getCloseTime()) { - DedupValue dv = createDedupValue(eventEniq, event, stateFieldValue); - lastDedupValue.resetTo(dv); - } else { - // update lastDedupValue, set closeTime when close - lastDedupValue.setStateFieldValue(stateFieldValue); - if (stateFieldValue.equals(stateCloseValue)) { - lastDedupValue.setCloseTime(eventEniq.timestamp); // when close an event, set closeTime for further check - } - } - return lastDedupValue; - } - - private DedupValue createDedupValue(EventUniq eventEniq, AlertStreamEvent event, String stateFieldValue) { - DedupValue dedupValue; - dedupValue = new DedupValue(); - dedupValue.setFirstOccurrence(eventEniq.timestamp); - int idx = event.getSchema().getColumnIndex(DOC_ID); - if (idx >= 0) { - dedupValue.setDocId(event.getData()[idx].toString()); - } else { - dedupValue.setDocId(""); - } - dedupValue.setCount(1); - dedupValue.setCloseTime(0); - dedupValue.setStateFieldValue(stateFieldValue); - return dedupValue; - } - - private DedupValue updateCount(EventUniq eventEniq) { - ConcurrentLinkedDeque<DedupValue> dedupValues = events.get(eventEniq); - if (dedupValues == null || dedupValues.size() <= 0) { - LOG.warn("{} No dedup values found for {}, cannot update count", this.publishName, eventEniq); - return null; - } else { - DedupValue dedupValue = dedupValues.getLast(); - dedupValue.setCount(dedupValue.getCount() + 1); - String updateMsg = String.format( - "%s Update count for dedup key %s, value %s and count %s", this.publishName, eventEniq, - dedupValue.getStateFieldValue(), dedupValue.getCount()); - if (LOG.isDebugEnabled()) { - LOG.debug(updateMsg); - } - return dedupValue; - } - } - - private AlertStreamEvent mergeEventWithDedupValue(AlertStreamEvent originalEvent, - DedupValue dedupValue, String dedupStateField) { - AlertStreamEvent event = new AlertStreamEvent(); - Object[] newdata = new Object[originalEvent.getData().length]; - for (int i = 0; i < originalEvent.getData().length; i++) { - newdata[i] = originalEvent.getData()[i]; - } - event.setData(newdata); - event.setStreamId(originalEvent.getStreamId()); - event.setSchema(originalEvent.getSchema()); - event.setPolicyId(originalEvent.getPolicyId()); - event.setCreatedTime(originalEvent.getCreatedTime()); - event.setCreatedBy(originalEvent.getCreatedBy()); - event.setTimestamp(originalEvent.getTimestamp()); - StreamDefinition streamDefinition = event.getSchema(); - for (int i = 0; i < event.getData().length; i++) { - String colName = streamDefinition.getColumns().get(i).getName(); - if (Objects.equal(colName, dedupStateField)) { - event.getData()[i] = dedupValue.getStateFieldValue(); - } - if (Objects.equal(colName, DEDUP_COUNT)) { - event.getData()[i] = dedupValue.getCount(); - } - if (Objects.equal(colName, DEDUP_FIRST_OCCURRENCE)) { - event.getData()[i] = dedupValue.getFirstOccurrence(); - } - if (Objects.equal(colName, DOC_ID)) { - event.getData()[i] = dedupValue.getDocId(); - } - } - return event; - } - -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupEntity.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupEntity.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupEntity.java deleted file mode 100644 index 86bc9b3..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupEntity.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher.dedup; - -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.ConcurrentLinkedDeque; - -import org.apache.eagle.alert.engine.publisher.impl.EventUniq; - -public class DedupEntity { - - private String publishName; - private EventUniq eventEniq; - private List<DedupValue> dedupValues = new ArrayList<DedupValue>(); - - public DedupEntity(String publishName, EventUniq eventEniq, ConcurrentLinkedDeque<DedupValue> dedupValues) { - this.publishName = publishName; - this.eventEniq = eventEniq; - this.dedupValues.addAll(dedupValues); - } - - public DedupEntity(String publishName, EventUniq eventEniq, List<DedupValue> dedupValues) { - this.publishName = publishName; - this.eventEniq = eventEniq; - this.dedupValues = dedupValues; - } - - public String getPublishName() { - return publishName; - } - - public void setPublishName(String publishName) { - this.publishName = publishName; - } - - public EventUniq getEventEniq() { - return eventEniq; - } - - public void setEventEniq(EventUniq eventEniq) { - this.eventEniq = eventEniq; - } - - public List<DedupValue> getDedupValues() { - return dedupValues; - } - - public void setDedupValues(List<DedupValue> dedupValues) { - this.dedupValues = dedupValues; - } - - public ConcurrentLinkedDeque<DedupValue> getDedupValuesInConcurrentLinkedDeque() { - ConcurrentLinkedDeque<DedupValue> result = new ConcurrentLinkedDeque<DedupValue>(); - result.addAll(this.getDedupValues()); - return result; - } - -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupValue.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupValue.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupValue.java deleted file mode 100644 index c1b3573..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/DedupValue.java +++ /dev/null @@ -1,96 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher.dedup; - -import com.google.common.base.Objects; - -public class DedupValue { - - private long firstOccurrence; - private String stateFieldValue; - private long count; - private long closeTime; - private String docId; - - public DedupValue() { - } - - public void resetTo(DedupValue dv) { - this.docId = dv.docId; - this.firstOccurrence = dv.firstOccurrence; - this.count = dv.count; - this.closeTime = dv.closeTime; - this.stateFieldValue = dv.stateFieldValue; - } - - public long getCount() { - return count; - } - - public void setCount(long count) { - this.count = count; - } - - public String getDocId() { - return docId; - } - - public void setDocId(String docId) { - this.docId = docId; - } - - public long getFirstOccurrence() { - return firstOccurrence; - } - - public void setFirstOccurrence(long firstOccurence) { - this.firstOccurrence = firstOccurence; - } - - public void setCloseTime(long closeTime) { - this.closeTime = closeTime; - } - - public long getCloseTime() { - return closeTime; - } - - public String getStateFieldValue() { - return stateFieldValue; - } - - public void setStateFieldValue(String stateFieldValue) { - this.stateFieldValue = stateFieldValue; - } - - @Override - public boolean equals(Object dedupValue) { - return Objects.equal(this.getStateFieldValue(), ((DedupValue) dedupValue).getStateFieldValue()); - } - - @Override - public int hashCode() { - return this.stateFieldValue == null ? "".hashCode() : this.stateFieldValue.hashCode(); - } - - @Override - public String toString() { - return String.format("DedupValue[state: %s, count: %s, first occurrence %s]", - stateFieldValue, count, firstOccurrence); - } - -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/ExtendedDeduplicator.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/ExtendedDeduplicator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/ExtendedDeduplicator.java deleted file mode 100644 index ce86ab4..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/dedup/ExtendedDeduplicator.java +++ /dev/null @@ -1,81 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher.dedup; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.eagle.alert.engine.publisher.AlertDeduplicator; - -import com.typesafe.config.Config; - -public abstract class ExtendedDeduplicator implements AlertDeduplicator { - - private Map<String, String> properties; - private DedupCache dedupCache; - private List<String> customDedupFields; - private String dedupStateField; - private Config config; - private String publishName; - - public ExtendedDeduplicator(Config config, - Map<String, String> properties, - List<String> customDedupFields, - String dedupStateField, - DedupCache dedupCache, - String publishName) { - this.properties = properties; - if (this.properties == null) { - this.properties = new HashMap<String, String>(); - } - this.dedupCache = dedupCache; - this.customDedupFields = customDedupFields; - this.dedupStateField = dedupStateField; - this.config = config; - this.publishName = publishName; - } - - public Map<String, String> getProperties() { - return this.properties; - } - - public DedupCache getDedupCache() { - return this.dedupCache; - } - - public List<String> getCustomDedupFields() { - return customDedupFields; - } - - public String getDedupStateField() { - return dedupStateField; - } - - public Config getConfig() { - return config; - } - - public String getPublishName() { - return publishName; - } - - public void setPublishName(String publishName) { - this.publishName = publishName; - } - -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java deleted file mode 100644 index 9010c93..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailConstants.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package org.apache.eagle.alert.engine.publisher.email; - -public class AlertEmailConstants { - - public static final String CONN_PLAINTEXT = "plaintext"; - public static final String CONN_TLS = "tls"; - public static final String CONN_SSL = "ssl"; - - public static final String CONF_MAIL_HOST = "mail.smtp.host"; - public static final String CONF_MAIL_PORT = "mail.smtp.port"; - public static final String CONF_MAIL_AUTH = "mail.smtp.auth"; - public static final String CONF_AUTH_USER = "mail.username"; - public static final String CONF_AUTH_PASSWORD = "mail.password"; - public static final String CONF_MAIL_CONN = "mail.connection"; - public static final String CONF_MAIL_DEBUG = "mail.debug"; -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java deleted file mode 100644 index 77cb1a7..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailContext.java +++ /dev/null @@ -1,81 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher.email; - -import java.util.Map; - -/** - * alert email bean - * one email consists of a list of email component. - */ -public class AlertEmailContext { - private Map<String, String> alertContext; - private String sender; - private String subject; - private String recipients; - private String velocityTplFile; - private String cc; - - public Map<String, String> getAlertContext() { - return alertContext; - } - - public void setAlertContext(Map<String, String> alertContext) { - this.alertContext = alertContext; - } - - public String getVelocityTplFile() { - return velocityTplFile; - } - - public void setVelocityTplFile(String velocityTplFile) { - this.velocityTplFile = velocityTplFile; - } - - public String getRecipients() { - return recipients; - } - - public void setRecipients(String recipients) { - this.recipients = recipients; - } - - public String getSender() { - return sender; - } - - public void setSender(String sender) { - this.sender = sender; - } - - public String getSubject() { - return subject; - } - - public void setSubject(String subject) { - this.subject = subject; - } - - public String getCc() { - return cc; - } - - public void setCc(String cc) { - this.cc = cc; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java deleted file mode 100644 index 1bcac17..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGenerator.java +++ /dev/null @@ -1,224 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -/** - * - */ -package org.apache.eagle.alert.engine.publisher.email; - -import org.apache.commons.httpclient.URIException; -import org.apache.commons.httpclient.util.URIUtil; -import org.apache.eagle.alert.engine.model.AlertStreamEvent; -import org.apache.eagle.alert.engine.publisher.PublishConstants; -import org.apache.eagle.common.DateTimeUtil; -import org.apache.eagle.common.Version; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.*; - -public class AlertEmailGenerator { - private String tplFile; - private String sender; - private String recipients; - private String subject; - - private String serverHost = "localhost"; - private int serverPort = 80; - - private Properties properties; - - private ThreadPoolExecutor executorPool; - - private static final Logger LOG = LoggerFactory.getLogger(AlertEmailGenerator.class); - - private static final long MAX_TIMEOUT_MS = 60000; - - public boolean sendAlertEmail(AlertStreamEvent entity) { - return sendAlertEmail(entity, recipients, null); - } - - public boolean sendAlertEmail(AlertStreamEvent entity, String recipients) { - return sendAlertEmail(entity, recipients, null); - } - - public boolean sendAlertEmail(AlertStreamEvent event, String recipients, String cc) { - AlertEmailContext email = new AlertEmailContext(); - Map<String, String> alertContext = buildAlertContext(event); - email.setAlertContext(alertContext); - email.setVelocityTplFile(tplFile); - if (event.getCategory() != null) { - email.setSubject(String.format("[Eagle Alert][%s][%s] %s", - event.getSeverity(), event.getCategory(), event.getSubject() != null ? event.getSubject() : subject)); - } else { - email.setSubject(String.format("[Eagle Alert][%s] %s", - event.getSeverity(), event.getSubject() != null ? event.getSubject() : subject)); - } - email.setSender(sender); - email.setRecipients(recipients); - email.setCc(cc); - - /** asynchronized email sending. */ - AlertEmailSender thread = new AlertEmailSender(email, properties); - - if (this.executorPool == null) { - throw new IllegalStateException("Invoking thread executor pool but it's is not set yet"); - } - - LOG.info("Sending email in asynchronous to: " + recipients + ", cc: " + cc); - Future<?> future = this.executorPool.submit(thread); - Boolean status; - try { - future.get(MAX_TIMEOUT_MS, TimeUnit.MILLISECONDS); - status = true; - //LOG.info(String.format("Successfully send email to %s", recipients)); - } catch (InterruptedException | ExecutionException e) { - status = false; - LOG.error(String.format("Failed to send email to %s, due to:%s", recipients, e), e); - } catch (TimeoutException e) { - status = false; - LOG.error(String.format("Failed to send email to %s due to timeout exception, max timeout: %s ms ", recipients, MAX_TIMEOUT_MS), e); - } - return status; - } - - private String getAlertBody(AlertStreamEvent event) { - if (event.getBody() == null) { - return String.format("Alert policy \"%s\" was triggered: %s", event.getPolicyId(), generateAlertDataDesc(event)); - } else { - return event.getBody(); - } - } - - private String generateAlertDataDesc(AlertStreamEvent event) { - if (event.getDataMap() == null) { - return "N/A"; - } - StringBuilder sb = new StringBuilder(); - for (Map.Entry<String, Object> entry : event.getDataMap().entrySet()) { - sb.append(entry.getKey()).append("=").append(entry.getValue()).append(" "); - } - return sb.toString(); - } - - private Map<String, String> buildAlertContext(AlertStreamEvent event) { - Map<String, String> alertContext = new HashMap<>(); - - if (event.getContext() != null) { - for (Map.Entry<String, Object> entry : event.getContext().entrySet()) { - if (entry.getValue() == null) { - alertContext.put(entry.getKey(), "N/A"); - } else { - alertContext.put(entry.getKey(), entry.getValue().toString()); - } - } - } - - alertContext.put(PublishConstants.ALERT_EMAIL_SUBJECT, event.getSubject()); - alertContext.put(PublishConstants.ALERT_EMAIL_BODY, getAlertBody(event)); - alertContext.put(PublishConstants.ALERT_EMAIL_POLICY_ID, event.getPolicyId()); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_ID, event.getAlertId()); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_DATA, event.getDataMap().toString()); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_DATA_DESC, generateAlertDataDesc(event)); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_CATEGORY, event.getCategory()); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_SEVERITY, event.getSeverity().toString()); - alertContext.put(PublishConstants.ALERT_EMAIL_TIME, DateTimeUtil.millisecondsToHumanDateWithSeconds(event.getCreatedTime())); - alertContext.put(PublishConstants.ALERT_EMAIL_STREAM_ID, event.getStreamId()); - alertContext.put(PublishConstants.ALERT_EMAIL_CREATOR, event.getCreatedBy()); - alertContext.put(PublishConstants.ALERT_EMAIL_VERSION, Version.version); - - String rootUrl = this.getServerPort() == 80 ? String.format("http://%s";, this.getServerHost()) - : String.format("http://%s:%s";, this.getServerHost(), this.getServerPort()); - try { - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_DETAIL_URL, - String.format("%s/#/site/%s/alert/detail/%s", rootUrl, event.getSiteId(), URIUtil.encodeQuery(event.getAlertId(), "UTF-8"))); - alertContext.put(PublishConstants.ALERT_EMAIL_POLICY_DETAIL_URL, - String.format("%s/#/site/%s/policy/detail/%s", rootUrl, event.getSiteId(), URIUtil.encodeQuery(event.getPolicyId(), "UTF-8"))); - } catch (URIException e) { - LOG.warn(e.getMessage(), e); - alertContext.put(PublishConstants.ALERT_EMAIL_ALERT_DETAIL_URL, - String.format("%s/#/site/%s/alert/detail/%s?timestamp=%s", event.getSiteId(), rootUrl, event.getAlertId(), event.getTimestamp())); - alertContext.put(PublishConstants.ALERT_EMAIL_POLICY_DETAIL_URL, - String.format("%s/#/site/%s/policy/detail/%s", event.getSiteId(), rootUrl, event.getPolicyId())); - } - alertContext.put(PublishConstants.ALERT_EMAIL_HOME_URL, rootUrl); - return alertContext; - } - - public String getTplFile() { - return tplFile; - } - - public void setTplFile(String tplFile) { - this.tplFile = tplFile; - } - - public String getSender() { - return sender; - } - - public void setSender(String sender) { - this.sender = sender; - } - - public String getRecipients() { - return recipients; - } - - public void setRecipients(String recipients) { - this.recipients = recipients; - } - - public String getSubject() { - return subject; - } - - public void setSubject(String subject) { - this.subject = subject; - } - - public Properties getProperties() { - return properties; - } - - public void setProperties(Properties properties) { - this.properties = properties; - } - - public void setExecutorPool(ThreadPoolExecutor executorPool) { - this.executorPool = executorPool; - } - - public String getServerHost() { - return serverHost; - } - - public void setServerHost(String serverHost) { - this.serverHost = serverHost; - } - - public int getServerPort() { - return serverPort; - } - - public void setServerPort(int serverPort) { - this.serverPort = serverPort; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGeneratorBuilder.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGeneratorBuilder.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGeneratorBuilder.java deleted file mode 100644 index f6debab..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailGeneratorBuilder.java +++ /dev/null @@ -1,77 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher.email; - -import java.util.Properties; -import java.util.concurrent.ThreadPoolExecutor; - -public class AlertEmailGeneratorBuilder { - private AlertEmailGenerator generator; - - private AlertEmailGeneratorBuilder() { - generator = new AlertEmailGenerator(); - } - - public static AlertEmailGeneratorBuilder newBuilder() { - return new AlertEmailGeneratorBuilder(); - } - - public AlertEmailGeneratorBuilder withSubject(String subject) { - generator.setSubject(subject); - return this; - } - - public AlertEmailGeneratorBuilder withSender(String sender) { - generator.setSender(sender); - return this; - } - - public AlertEmailGeneratorBuilder withRecipients(String recipients) { - generator.setRecipients(recipients); - return this; - } - - public AlertEmailGeneratorBuilder withTplFile(String tplFile) { - generator.setTplFile(tplFile); - return this; - } - - public AlertEmailGeneratorBuilder withMailProps(Properties mailProps) { - generator.setProperties(mailProps); - return this; - } - - public AlertEmailGeneratorBuilder withExecutorPool(ThreadPoolExecutor threadPoolExecutor) { - generator.setExecutorPool(threadPoolExecutor); - return this; - } - - public AlertEmailGeneratorBuilder withServerHost(String serverHost) { - generator.setServerHost(serverHost); - return this; - } - - public AlertEmailGeneratorBuilder withServerPort(int serverPort) { - generator.setServerPort(serverPort); - return this; - } - - public AlertEmailGenerator build() { - return this.generator; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailSender.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailSender.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailSender.java deleted file mode 100644 index f573215..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/AlertEmailSender.java +++ /dev/null @@ -1,127 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher.email; - -import org.apache.eagle.alert.engine.publisher.PublishConstants; -import org.apache.eagle.common.DateTimeUtil; -import org.apache.velocity.VelocityContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.lang.management.ManagementFactory; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Properties; - -public class AlertEmailSender implements Runnable { - - protected final List<Map<String, String>> alertContexts = new ArrayList<Map<String, String>>(); - protected final String configFileName; - protected final String subject; - protected final String sender; - protected final String recipients; - protected final String cc; - protected final String origin; - protected boolean sentSuccessfully = false; - - private static final Logger LOG = LoggerFactory.getLogger(AlertEmailSender.class); - private static final int MAX_RETRY_COUNT = 3; - - private Properties mailProps; - private String threadName; - - /** - * Derived class may have some additional context properties to add. - * - * @param context velocity context - * @param env environment - */ - protected void additionalContext(VelocityContext context, String env) { - // By default there's no additional context added - } - - public AlertEmailSender(AlertEmailContext alertEmail) { - this.recipients = alertEmail.getRecipients(); - this.configFileName = alertEmail.getVelocityTplFile(); - this.subject = alertEmail.getSubject(); - this.sender = alertEmail.getSender(); - this.cc = alertEmail.getCc(); - - this.alertContexts.add(alertEmail.getAlertContext()); - String tmp = ManagementFactory.getRuntimeMXBean().getName(); - this.origin = tmp.split("@")[1] + "(pid:" + tmp.split("@")[0] + ")"; - threadName = Thread.currentThread().getName(); - LOG.info("Initialized " + threadName + ": origin is : " + this.origin + ", recipient of the email: " + this.recipients + ", velocity TPL file: " + this.configFileName); - } - - public AlertEmailSender(AlertEmailContext alertEmail, Properties mailProps) { - this(alertEmail); - this.mailProps = mailProps; - } - - @Override - public void run() { - int count = 0; - boolean success = false; - while (count++ < MAX_RETRY_COUNT && !success) { - LOG.info("Sending email, tried: " + count + ", max: " + MAX_RETRY_COUNT); - try { - final EagleMailClient client; - if (mailProps != null) { - client = new EagleMailClient(mailProps); - } else { - client = new EagleMailClient(); - } - - final VelocityContext context = new VelocityContext(); - generateCommonContext(context); - LOG.info("After calling generateCommonContext..."); - - if (recipients == null || recipients.equals("")) { - LOG.error("Recipients is null, skip sending emails "); - return; - } - String title = subject; - - success = client.send(sender, recipients, cc, title, configFileName, context, null); - LOG.info("Success of sending email: " + success); - if (!success && count < MAX_RETRY_COUNT) { - LOG.info("Sleep for a while before retrying"); - Thread.sleep(10 * 1000); - } - } catch (Exception e) { - LOG.warn("Sending mail exception", e); - } - } - if (success) { - sentSuccessfully = true; - LOG.info(String.format("Successfully send email, thread: %s", threadName)); - } else { - LOG.warn(String.format("Fail sending email after tries %s times, thread: %s", MAX_RETRY_COUNT, threadName)); - } - } - - private void generateCommonContext(VelocityContext context) { - context.put(PublishConstants.ALERT_EMAIL_TIME_PROPERTY, DateTimeUtil.millisecondsToHumanDateWithSeconds(System.currentTimeMillis())); - context.put(PublishConstants.ALERT_EMAIL_COUNT_PROPERTY, alertContexts.size()); - context.put(PublishConstants.ALERT_EMAIL_ALERTLIST_PROPERTY, alertContexts); - context.put(PublishConstants.ALERT_EMAIL_ORIGIN_PROPERTY, origin); - } - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/EagleMailClient.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/EagleMailClient.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/EagleMailClient.java deleted file mode 100755 index 7147894..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/email/EagleMailClient.java +++ /dev/null @@ -1,230 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ -package org.apache.eagle.alert.engine.publisher.email; - -import org.apache.velocity.Template; -import org.apache.velocity.VelocityContext; -import org.apache.velocity.app.VelocityEngine; -import org.apache.velocity.exception.ResourceNotFoundException; -import org.apache.velocity.runtime.RuntimeConstants; -import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.StringWriter; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import javax.activation.DataHandler; -import javax.activation.DataSource; -import javax.activation.FileDataSource; -import javax.mail.*; -import javax.mail.internet.*; - -public class EagleMailClient { - private static final Logger LOG = LoggerFactory.getLogger(EagleMailClient.class); - private static final String BASE_PATH = "templates/"; - - private VelocityEngine velocityEngine; - private Session session; - - public EagleMailClient() { - this(new Properties()); - } - - public EagleMailClient(final Properties config) { - try { - velocityEngine = new VelocityEngine(); - velocityEngine.setProperty(RuntimeConstants.RESOURCE_LOADER, "classpath"); - velocityEngine.setProperty("classpath.resource.loader.class", ClasspathResourceLoader.class.getName()); - velocityEngine.init(); - - config.put("mail.transport.protocol", "smtp"); - if (Boolean.parseBoolean(config.getProperty(AlertEmailConstants.CONF_MAIL_AUTH))) { - session = Session.getInstance(config, new Authenticator() { - protected PasswordAuthentication getPasswordAuthentication() { - return new PasswordAuthentication( - config.getProperty(AlertEmailConstants.CONF_AUTH_USER), - config.getProperty(AlertEmailConstants.CONF_AUTH_PASSWORD) - ); - } - }); - } else { - session = Session.getInstance(config, new Authenticator() { - }); - } - - final String debugMode = config.getProperty(AlertEmailConstants.CONF_MAIL_DEBUG, "false"); - final boolean debug = Boolean.parseBoolean(debugMode); - LOG.info("Set email debug mode: " + debugMode); - session.setDebug(debug); - } catch (Exception e) { - LOG.error("Failed to connect to smtp server", e); - } - } - - private boolean sendInternal(String from, String to, String cc, String title, String content) { - Message msg = new MimeMessage(session); - try { - msg.setFrom(new InternetAddress(from)); - msg.setSubject(title); - if (to != null) { - msg.setRecipients(Message.RecipientType.TO, InternetAddress.parse(to)); - } - if (cc != null) { - msg.setRecipients(Message.RecipientType.CC, InternetAddress.parse(cc)); - } - //msg.setRecipients(Message.RecipientType.BCC, InternetAddress.parse(DEFAULT_BCC_ADDRESS)); - msg.setContent(content, "text/html;charset=utf-8"); - LOG.info(String.format("Going to send mail: from[%s], to[%s], cc[%s], title[%s]", from, to, cc, title)); - Transport.send(msg); - return true; - } catch (AddressException e) { - LOG.info("Failed to send mail, got an AddressException: " + e.getMessage(), e); - return false; - } catch (MessagingException e) { - LOG.info("Failed to send mail, got an AddressException: " + e.getMessage(), e); - return false; - } - } - - private boolean sendInternal(String from, String to, String cc, String title, String content, List<MimeBodyPart> attachments) { - MimeMessage mail = new MimeMessage(session); - try { - mail.setFrom(new InternetAddress(from)); - mail.setSubject(title); - if (to != null) { - mail.setRecipients(Message.RecipientType.TO, InternetAddress.parse(to)); - } - if (cc != null) { - mail.setRecipients(Message.RecipientType.CC, InternetAddress.parse(cc)); - } - - //mail.setRecipients(Message.RecipientType.BCC, InternetAddress.parse(DEFAULT_BCC_ADDRESS)); - - MimeBodyPart mimeBodyPart = new MimeBodyPart(); - mimeBodyPart.setContent(content, "text/html;charset=utf-8"); - - Multipart multipart = new MimeMultipart(); - multipart.addBodyPart(mimeBodyPart); - - for (MimeBodyPart attachment : attachments) { - multipart.addBodyPart(attachment); - } - - mail.setContent(multipart); - // mail.setContent(content, "text/html;charset=utf-8"); - LOG.info(String.format("Going to send mail: from[%s], to[%s], cc[%s], title[%s]", from, to, cc, title)); - Transport.send(mail); - return true; - } catch (AddressException e) { - LOG.info("Failed to send mail, got an AddressException: " + e.getMessage(), e); - return false; - } catch (MessagingException e) { - LOG.info("Failed to send mail, got an AddressException: " + e.getMessage(), e); - return false; - } - } - - public boolean send(String from, String to, String cc, String title, - String content) { - return this.sendInternal(from, to, cc, title, content); - } - - public boolean send(String from, String to, String cc, String title, - String templatePath, VelocityContext context) { - Template t = null; - try { - t = velocityEngine.getTemplate(BASE_PATH + templatePath); - } catch (ResourceNotFoundException ex) { - // ignored - } - if (t == null) { - try { - t = velocityEngine.getTemplate(templatePath); - } catch (ResourceNotFoundException e) { - t = velocityEngine.getTemplate("/" + templatePath); - } - } - final StringWriter writer = new StringWriter(); - t.merge(context, writer); - if (LOG.isDebugEnabled()) { - LOG.debug(writer.toString()); - } - - return this.send(from, to, cc, title, writer.toString()); - } - - public boolean send(String from, String to, String cc, String title, - String templatePath, VelocityContext context, Map<String, File> attachments) { - if (attachments == null || attachments.isEmpty()) { - return send(from, to, cc, title, templatePath, context); - } - Template t = null; - - List<MimeBodyPart> mimeBodyParts = new ArrayList<MimeBodyPart>(); - - for (Map.Entry<String, File> entry : attachments.entrySet()) { - final String attachment = entry.getKey(); - final File attachmentFile = entry.getValue(); - final MimeBodyPart mimeBodyPart = new MimeBodyPart(); - if (attachmentFile != null && attachmentFile.exists()) { - DataSource source = new FileDataSource(attachmentFile); - try { - mimeBodyPart.setDataHandler(new DataHandler(source)); - mimeBodyPart.setFileName(attachment); - mimeBodyPart.setDisposition(MimeBodyPart.ATTACHMENT); - mimeBodyPart.setContentID(attachment); - mimeBodyParts.add(mimeBodyPart); - } catch (MessagingException e) { - LOG.error("Generate mail failed, got exception while attaching files: " + e.getMessage(), e); - } - } else { - LOG.error("Attachment: " + attachment + " is null or not exists"); - } - } - - try { - t = velocityEngine.getTemplate(BASE_PATH + templatePath); - } catch (ResourceNotFoundException ex) { - // ignored - } - - if (t == null) { - try { - t = velocityEngine.getTemplate(templatePath); - } catch (ResourceNotFoundException e) { - try { - t = velocityEngine.getTemplate("/" + templatePath); - } catch (Exception ex) { - LOG.error("Template not found:" + "/" + templatePath, ex); - } - } - } - - final StringWriter writer = new StringWriter(); - t.merge(context, writer); - if (LOG.isDebugEnabled()) { - LOG.debug(writer.toString()); - } - - return this.sendInternal(from, to, cc, title, writer.toString(), mimeBodyParts); - } -} http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AbstractPublishPlugin.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AbstractPublishPlugin.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AbstractPublishPlugin.java deleted file mode 100644 index b155bb8..0000000 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/impl/AbstractPublishPlugin.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.eagle.alert.engine.publisher.impl; - -import com.google.common.base.Joiner; -import com.typesafe.config.Config; -import org.apache.commons.lang3.StringUtils; -import org.apache.eagle.alert.engine.codec.IEventSerializer; -import org.apache.eagle.alert.engine.coordinator.OverrideDeduplicatorSpec; -import org.apache.eagle.alert.engine.coordinator.Publishment; -import org.apache.eagle.alert.engine.model.AlertStreamEvent; -import org.apache.eagle.alert.engine.publisher.AlertDeduplicator; -import org.apache.eagle.alert.engine.publisher.AlertPublishPlugin; -import org.apache.eagle.alert.engine.publisher.dedup.DedupCache; -import org.apache.eagle.alert.engine.publisher.dedup.ExtendedDeduplicator; -import org.slf4j.Logger; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -/** - * @since Jun 3, 2016. - */ -public abstract class AbstractPublishPlugin implements AlertPublishPlugin { - - protected AlertDeduplicator deduplicator; - protected PublishStatus status; - protected IEventSerializer serializer; - protected String pubName; - - @SuppressWarnings("rawtypes") - @Override - public void init(Config config, Publishment publishment, Map conf) throws Exception { - DedupCache dedupCache = new DedupCache(config, publishment.getName()); - OverrideDeduplicatorSpec spec = publishment.getOverrideDeduplicator(); - if (spec != null && StringUtils.isNotBlank(spec.getClassName())) { - try { - this.deduplicator = (ExtendedDeduplicator) Class.forName( - spec.getClassName()).getConstructor( - Config.class, - Map.class, - List.class, - String.class, - DedupCache.class, - String.class).newInstance( - config, - spec.getProperties(), - publishment.getDedupFields(), - publishment.getDedupStateField(), - dedupCache, - publishment.getName()); - getLogger().info("{} initiliazed extended deduplicator {} with properties {} successfully", - publishment.getName(), spec.getClassName(), Joiner.on(",").withKeyValueSeparator(">").join( - spec.getProperties() == null ? new HashMap<String, String>() : spec.getProperties())); - } catch (Throwable t) { - getLogger().error(String.format("initialize extended deduplicator %s failed", spec.getClassName()), t); - } - } else { - this.deduplicator = new DefaultDeduplicator(publishment.getDedupIntervalMin(), - publishment.getDedupFields(), publishment.getDedupStateField(), publishment.getDedupStateCloseValue(), dedupCache); - this.pubName = publishment.getName(); - } - String serializerClz = publishment.getSerializer(); - try { - Object obj = Class.forName(serializerClz).getConstructor(Map.class).newInstance(conf); - if (!(obj instanceof IEventSerializer)) { - throw new Exception(String.format("serializer %s of publishment %s is not subclass to %s!", - publishment.getSerializer(), - publishment.getName(), - IEventSerializer.class.getName())); - } - serializer = (IEventSerializer) obj; - } catch (Exception e) { - getLogger().error(String.format("initialized failed, use default StringEventSerializer, failure message : {}", e.getMessage()), e); - serializer = new StringEventSerializer(conf); - } - } - - @Override - public void update(String dedupIntervalMin, Map<String, Object> pluginProperties) { - deduplicator.setDedupIntervalMin(dedupIntervalMin); - } - - @Override - public List<AlertStreamEvent> dedup(AlertStreamEvent event) { - return deduplicator.dedup(event); - } - - @Override - public PublishStatus getStatus() { - return status; - } - - protected abstract Logger getLogger(); - -}
