You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eagle.apache.org by ha...@apache.org on 2017/04/03 11:54:45 UTC
[37/84] [partial] eagle git commit: Clean repo for eagle site
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java
deleted file mode 100644
index 255f1e6..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java
+++ /dev/null
@@ -1,124 +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.metric;
-
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-import org.apache.eagle.alert.metric.sink.MetricSink;
-import org.apache.eagle.alert.metric.sink.MetricSinkRepository;
-import org.apache.eagle.alert.metric.source.MetricSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-public class MetricSystem implements IMetricSystem {
- private final Config config;
- private Map<MetricSink, Config> sinks = new HashMap<>();
- private MetricRegistry registry = new MetricRegistry();
- private volatile boolean running;
- private volatile boolean initialized;
- private static final Logger LOG = LoggerFactory.getLogger(MetricSystem.class);
- private final Map<String, Object> metricTags = new HashMap<>();
- private int scheduleDurationSeconds = 10;
-
- public MetricSystem(Config config) {
- this.config = config;
- if (this.config.hasPath(MetricConfigs.DURATION_SECONDS_CONF)) {
- this.scheduleDurationSeconds = this.config.getInt(MetricConfigs.DURATION_SECONDS_CONF);
- LOG.info("Override {}: {}",MetricConfigs.DURATION_SECONDS_CONF, this.scheduleDurationSeconds);
- }
- }
-
- public static MetricSystem load(Config config) {
- MetricSystem instance = new MetricSystem(config);
- instance.loadFromConfig();
- return instance;
- }
-
- /**
- * Add additional tags.
- */
- @Override
- public void tags(Map<String, Object> metricTags) {
- this.metricTags.putAll(metricTags);
- }
-
- @Override
- public void start() {
- if (initialized) {
- throw new IllegalStateException("Attempting to initialize a MetricsSystem that is already initialized");
- }
- sinks.forEach((sink, conf) -> {
- sink.prepare(conf.withValue(MetricConfigs.TAGS_FIELD_NAME, ConfigFactory.parseMap(metricTags).root()), registry);
- });
- initialized = true;
- }
-
- @Override
- public void schedule() {
- if (running) {
- throw new IllegalStateException("Attempting to start a MetricsSystem that is already running");
- }
- sinks.keySet().forEach((sink) -> sink.start(this.scheduleDurationSeconds, TimeUnit.SECONDS));
- running = true;
- }
-
- public void loadFromConfig() {
- loadSinksFromConfig();
- }
-
- private void loadSinksFromConfig() {
- Config sinkCls = config.hasPath(MetricConfigs.METRIC_SINK_CONF) ? config.getConfig(MetricConfigs.METRIC_SINK_CONF) : null;
- if (sinkCls == null) {
- // do nothing
- } else {
- for (String sinkType : sinkCls.root().unwrapped().keySet()) {
- register(MetricSinkRepository.createSink(sinkType), config.getConfig(MetricConfigs.METRIC_SINK_CONF + "." + sinkType));
- }
- }
- }
-
- @Override
- public void stop() {
- sinks.keySet().forEach(MetricSink::stop);
- }
-
- @Override
- public void report() {
- sinks.keySet().forEach(MetricSink::report);
- }
-
- @Override
- public void register(MetricSink sink, Config config) {
- sinks.put(sink, config);
- LOG.info("Registered {}", sink);
- }
-
- @Override
- public void register(MetricSource source) {
- registry().registerAll(source.registry());
- }
-
- @Override
- public MetricRegistry registry() {
- return registry;
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
deleted file mode 100644
index 229bbd9..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
+++ /dev/null
@@ -1,112 +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.metric.entity;
-
-import org.apache.eagle.common.DateTimeUtil;
-
-import com.codahale.metrics.Counter;
-import com.codahale.metrics.Gauge;
-import com.codahale.metrics.Histogram;
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.Snapshot;
-import com.codahale.metrics.Timer;
-
-import java.util.Map;
-import java.util.TreeMap;
-
-public class MetricEvent extends TreeMap<String, Object> {
-
- private static final long serialVersionUID = 6862373651636342744L;
-
- public static Builder of(String name) {
- return new Builder(name);
- }
-
- /**
- * TODO: Refactor according to ConsoleReporter.
- */
- public static class Builder {
- private final String name;
- private MetricEvent instance;
-
- public Builder(String name) {
- this.instance = new MetricEvent();
- this.name = name;
- }
-
- public MetricEvent build() {
- this.instance.put("name", name);
- if (!this.instance.containsKey("timestamp")) {
- this.instance.put("timestamp", DateTimeUtil.getCurrentTimestamp());
- }
- return this.instance;
- }
-
- public Builder from(Counter value) {
- // this.instance.put("type","counter");
- this.instance.put("count", value.getCount());
- return this;
- }
-
- @SuppressWarnings( {"rawtypes", "unchecked"})
- public Builder from(Gauge gauge) {
- Object value = gauge.getValue();
- if (value instanceof Map) {
- Map<? extends String, ?> map = (Map<? extends String, ?>) value;
- this.instance.putAll(map);
- } else {
- this.instance.put("value", value);
- }
- return this;
- }
-
- public Builder from(Histogram value) {
- this.instance.put("count", value.getCount());
- Snapshot snapshot = value.getSnapshot();
- this.instance.put("min", snapshot.getMin());
- this.instance.put("max", snapshot.getMax());
- this.instance.put("mean", snapshot.getMean());
- this.instance.put("stddev", snapshot.getStdDev());
- this.instance.put("median", snapshot.getMedian());
- this.instance.put("75thPercentile", snapshot.get75thPercentile());
- this.instance.put("95thPercentile", snapshot.get95thPercentile());
- this.instance.put("98thPercentile", snapshot.get98thPercentile());
- this.instance.put("99thPercentile", snapshot.get99thPercentile());
- this.instance.put("999thPercentile", snapshot.get999thPercentile());
- return this;
- }
-
- public Builder from(Meter value) {
- this.instance.put("value", value.getCount());
- this.instance.put("15MinRate", value.getFifteenMinuteRate());
- this.instance.put("5MinRate", value.getFiveMinuteRate());
- this.instance.put("mean", value.getMeanRate());
- this.instance.put("1MinRate", value.getOneMinuteRate());
- return this;
- }
-
- public Builder from(Timer value) {
- // this.instance.put("type","timer");
- this.instance.put("value", value.getCount());
- this.instance.put("15MinRate", value.getFifteenMinuteRate());
- this.instance.put("5MinRate", value.getFiveMinuteRate());
- this.instance.put("mean", value.getMeanRate());
- this.instance.put("1MinRate", value.getOneMinuteRate());
- return this;
- }
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
deleted file mode 100644
index bf0b365..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
+++ /dev/null
@@ -1,208 +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.metric.reporter;
-
-import org.apache.eagle.alert.metric.entity.MetricEvent;
-import org.apache.eagle.alert.utils.ByteUtils;
-import com.codahale.metrics.*;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Preconditions;
-import com.typesafe.config.Config;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.Map;
-import java.util.Properties;
-import java.util.SortedMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-public class KafkaReporter extends ScheduledReporter {
- private static final Logger LOG = LoggerFactory.getLogger(KafkaReporter.class);
- private final String topic;
- private final Properties properties;
- private final Producer<byte[], String> producer;
- private final Map<String, Object> additionalFields;
-
- protected KafkaReporter(MetricRegistry registry, MetricFilter filter, TimeUnit rateUnit, TimeUnit durationUnit, String topic, Properties config, Map<String, Object> additionalFields) {
- super(registry, "kafka-reporter", filter, rateUnit, durationUnit);
- this.topic = topic;
- this.properties = new Properties();
- Preconditions.checkNotNull(topic, "topic should not be null");
- // properties.put("bootstrap.servers", brokerList);
- // properties.put("metadata.broker.list", brokerList);
- properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- properties.put("request.required.acks", "1");
- properties.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- if (config != null) {
- LOG.info(config.toString());
- properties.putAll(config);
- }
- this.additionalFields = additionalFields;
- this.producer = new KafkaProducer<>(properties);
- LOG.info("Initialized kafka-reporter");
- }
-
- @SuppressWarnings("rawtypes")
- @Override
- public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters,
- SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) {
- for (SortedMap.Entry<String, Gauge> entry : gauges.entrySet()) {
- onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
- }
- for (SortedMap.Entry<String, Counter> entry : counters.entrySet()) {
- onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
- }
- for (SortedMap.Entry<String, Histogram> entry : histograms.entrySet()) {
- onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
- }
- for (SortedMap.Entry<String, Meter> entry : meters.entrySet()) {
- onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
- }
- for (SortedMap.Entry<String, Timer> entry : timers.entrySet()) {
- onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
- }
- }
-
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
- private void onMetricEvent(MetricEvent event) {
- try {
- if (additionalFields != null) {
- event.putAll(additionalFields);
- }
- // TODO: Support configurable partition key
- byte[] key = ByteUtils.intToBytes(event.hashCode());
- ProducerRecord<byte[], String> record = new ProducerRecord<>(topic, key, OBJECT_MAPPER.writeValueAsString(event));
- // TODO: Support configuration timeout
- this.producer.send(record).get(5, TimeUnit.SECONDS);
- } catch (JsonProcessingException e) {
- LOG.error("Failed to serialize {} as json", event, e);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
- LOG.error("Failed to produce message to topic {}", topic, e);
- }
- }
-
- @Override
- public void stop() {
- this.producer.close();
- super.stop();
- }
-
- @Override
- public void close() {
- this.producer.close();
- super.close();
- }
-
- public static Builder forRegistry(MetricRegistry registry) {
- return new Builder(registry);
- }
-
- public static class Builder {
- private final MetricRegistry registry;
- private TimeUnit rateUnit;
- private TimeUnit durationUnit;
- private MetricFilter filter;
- private String topic;
- private Properties properties;
- private Map<String, Object> additionalFields;
-
- private Builder(MetricRegistry registry) {
- this.registry = registry;
- this.rateUnit = TimeUnit.SECONDS;
- this.durationUnit = TimeUnit.MILLISECONDS;
- this.filter = MetricFilter.ALL;
- }
-
- /**
- * Convert rates to the given time unit.
- *
- * @param rateUnit a unit of time
- * @return {@code this}
- */
- public Builder convertRatesTo(TimeUnit rateUnit) {
- this.rateUnit = rateUnit;
- return this;
- }
-
- /**
- * Convert durations to the given time unit.
- *
- * @param durationUnit a unit of time
- * @return {@code this}
- */
- public Builder convertDurationsTo(TimeUnit durationUnit) {
- this.durationUnit = durationUnit;
- return this;
- }
-
- /**
- * Only report metrics which match the given filter.
- *
- * @param filter a {@link MetricFilter}
- * @return {@code this}
- */
- public Builder filter(MetricFilter filter) {
- this.filter = filter;
- return this;
- }
-
- public Builder topic(String topic) {
- this.topic = topic;
- return this;
- }
-
- /**
- * Builds a {@link ConsoleReporter} with the given properties.
- *
- * @return a {@link ConsoleReporter}
- */
- public KafkaReporter build() {
- if (topic == null && properties != null) {
- topic = properties.getProperty("topic");
- }
- return new KafkaReporter(registry, filter, rateUnit, durationUnit, topic, properties, additionalFields);
- }
-
- @SuppressWarnings("serial")
- public Builder config(Config config) {
- this.config(new Properties() {
- {
- putAll(config.root().unwrapped());
- }
- });
- return this;
- }
-
- public Builder config(Properties properties) {
- this.properties = properties;
- return this;
- }
-
- public Builder addFields(Map<String, Object> tags) {
- this.additionalFields = tags;
- return this;
- }
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
deleted file mode 100644
index 6ded685..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
+++ /dev/null
@@ -1,48 +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.metric.sink;
-
-import com.codahale.metrics.ConsoleReporter;
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import java.util.concurrent.TimeUnit;
-
-public class ConsoleSink implements MetricSink {
- private ConsoleReporter reporter;
-
- @Override
- public void prepare(Config config, MetricRegistry registry) {
- reporter = ConsoleReporter.forRegistry(registry).build();
- }
-
- @Override
- public void start(long period, TimeUnit unit) {
- reporter.start(period, unit);
- }
-
- @Override
- public void stop() {
- reporter.stop();
- reporter.close();
- }
-
- @Override
- public void report() {
- reporter.report();
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/ElasticSearchSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ElasticSearchSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ElasticSearchSink.java
deleted file mode 100644
index 3015be2..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ElasticSearchSink.java
+++ /dev/null
@@ -1,93 +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.metric.sink;
-
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import org.apache.eagle.alert.metric.MetricConfigs;
-import org.elasticsearch.metrics.ElasticsearchReporter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-public class ElasticSearchSink implements MetricSink {
-
- private ElasticsearchReporter reporter = null;
- private static final Logger LOG = LoggerFactory.getLogger(ElasticSearchSink.class);
-
- private static final String INDEX_DATEFORMAT_CONF = "indexDateFormat";
- private static final String TIMESTAMP_FIELD_CONF = "timestampField";
- private static final String HOSTS_CONF = "hosts";
- private static final String INDEX_CONF = "index";
-
- private static final String DEFAULT_INDEX_DATE_FORMAT = "yyyy-MM-dd";
- private static final String DEFAULT_TIMESTAMP_FIELD = "@timestamp";
-
- @Override
- public void prepare(Config config, MetricRegistry registry) {
- LOG.info("Preparing elasticsearch-sink");
- try {
- ElasticsearchReporter.Builder builder = ElasticsearchReporter.forRegistry(registry);
-
- if (config.hasPath(HOSTS_CONF)) {
- List<String> hosts = config.getStringList(HOSTS_CONF);
- builder.hosts(hosts.toArray(new String[hosts.size()]));
- }
-
- if (config.hasPath(INDEX_CONF)) {
- builder.index(config.getString(INDEX_CONF));
- }
-
- builder.indexDateFormat(config.hasPath(INDEX_DATEFORMAT_CONF)
- ? config.getString(INDEX_DATEFORMAT_CONF) : DEFAULT_INDEX_DATE_FORMAT);
-
- builder.timestampFieldname(config.hasPath(TIMESTAMP_FIELD_CONF)
- ? config.getString(TIMESTAMP_FIELD_CONF) : DEFAULT_TIMESTAMP_FIELD);
-
- if (config.hasPath(MetricConfigs.TAGS_FIELD_NAME)) {
- builder.additionalFields(config.getConfig(MetricConfigs.TAGS_FIELD_NAME).root().unwrapped());
- }
-
- reporter = builder.build();
- } catch (IOException e) {
- LOG.error(e.getMessage(), e);
- throw new IllegalStateException(e.getMessage(), e);
- }
- LOG.info("Initialized elasticsearch-sink");
- }
-
- @Override
- public void start(long period, TimeUnit unit) {
- LOG.info("Starting elasticsearch-sink");
- reporter.start(period, unit);
- }
-
- @Override
- public void stop() {
- LOG.info("Stopping elasticsearch-sink");
- reporter.stop();
- reporter.close();
- }
-
- @Override
- public void report() {
- reporter.report();
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/JmxSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/JmxSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/JmxSink.java
deleted file mode 100644
index 2d3ed93..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/JmxSink.java
+++ /dev/null
@@ -1,47 +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.metric.sink;
-
-import com.codahale.metrics.JmxReporter;
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import java.util.concurrent.TimeUnit;
-
-public class JmxSink implements MetricSink {
- private JmxReporter reporter;
-
- @Override
- public void prepare(Config config, MetricRegistry registry) {
- reporter = JmxReporter.forRegistry(registry).build();
- }
-
- @Override
- public void start(long period, TimeUnit unit) {
- reporter.start();
- }
-
- @Override
- public void stop() {
- reporter.stop();
- reporter.close();
- }
-
- @Override
- public void report() {
- // do nothing
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/KafkaSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/KafkaSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/KafkaSink.java
deleted file mode 100644
index f8bc8f3..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/KafkaSink.java
+++ /dev/null
@@ -1,66 +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.metric.sink;
-
-import org.apache.eagle.alert.metric.MetricConfigs;
-import org.apache.eagle.alert.metric.reporter.KafkaReporter;
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.concurrent.TimeUnit;
-
-public class KafkaSink implements MetricSink {
- private KafkaReporter reporter;
- private static final Logger LOG = LoggerFactory.getLogger(KafkaSink.class);
-
- @Override
- public void prepare(Config config, MetricRegistry registry) {
- LOG.debug("Preparing kafka-sink");
- KafkaReporter.Builder builder = KafkaReporter.forRegistry(registry)
- .topic(config.getString("topic"))
- .config(config);
-
- if (config.hasPath(MetricConfigs.TAGS_FIELD_NAME)) {
- builder.addFields(config.getConfig(MetricConfigs.TAGS_FIELD_NAME).root().unwrapped());
- }
-
- reporter = builder.build();
- LOG.info("Prepared kafka-sink");
- }
-
- @Override
- public void start(long period, TimeUnit unit) {
- LOG.info("Starting");
- reporter.start(period, unit);
- }
-
- @Override
- public void stop() {
- LOG.info("Stopping");
- reporter.stop();
-
- LOG.info("Closing");
- reporter.close();
- }
-
- @Override
- public void report() {
- reporter.report();
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSink.java
deleted file mode 100644
index 2030d8e..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSink.java
+++ /dev/null
@@ -1,31 +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.metric.sink;
-
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import java.util.concurrent.TimeUnit;
-
-public interface MetricSink {
- void prepare(Config config, MetricRegistry registry);
-
- void start(long period, TimeUnit unit);
-
- void stop();
-
- void report();
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSinkRepository.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSinkRepository.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSinkRepository.java
deleted file mode 100644
index 70d7331..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/MetricSinkRepository.java
+++ /dev/null
@@ -1,47 +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.metric.sink;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class MetricSinkRepository {
- private static final Map<String, Class<? extends MetricSink>> sinkTypeClassMapping = new HashMap<>();
-
- public static void register(String sinkType, Class<? extends MetricSink> sinkClass) {
- sinkTypeClassMapping.put(sinkType, sinkClass);
- }
-
- public static MetricSink createSink(String sinkType) {
- if (!sinkTypeClassMapping.containsKey(sinkType)) {
- throw new IllegalArgumentException("Unknown sink type: " + sinkType);
- }
- try {
- return sinkTypeClassMapping.get(sinkType).newInstance();
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException(e);
- }
- }
-
- static {
- register("kafka", KafkaSink.class);
- register("jmx", JmxSink.class);
- register("elasticsearch", ElasticSearchSink.class);
- register("stdout", ConsoleSink.class);
- register("logger", Slf4jSink.class);
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/sink/Slf4jSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/Slf4jSink.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/Slf4jSink.java
deleted file mode 100644
index c25c835..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/Slf4jSink.java
+++ /dev/null
@@ -1,76 +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.metric.sink;
-
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.Slf4jReporter;
-import com.typesafe.config.Config;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-public class Slf4jSink implements MetricSink {
- private Slf4jReporter reporter;
-
- @SuppressWarnings("serial")
- private static final Map<String, Slf4jReporter.LoggingLevel> LEVEL_MAPPING = new HashMap<String, Slf4jReporter.LoggingLevel>() {
- {
- put("INFO", Slf4jReporter.LoggingLevel.INFO);
- put("DEBUG", Slf4jReporter.LoggingLevel.DEBUG);
- put("ERROR", Slf4jReporter.LoggingLevel.ERROR);
- put("TRACE", Slf4jReporter.LoggingLevel.TRACE);
- put("WARN", Slf4jReporter.LoggingLevel.WARN);
- }
- };
-
- private static Slf4jReporter.LoggingLevel getLoggingLevel(String level) {
- if (LEVEL_MAPPING.containsKey(level.toUpperCase())) {
- return LEVEL_MAPPING.get(level.toUpperCase());
- } else {
- throw new IllegalArgumentException("Illegal logging level: " + level);
- }
- }
-
- @Override
- public void prepare(Config config, MetricRegistry registry) {
- reporter = Slf4jReporter.forRegistry(registry)
- .outputTo(LoggerFactory.getLogger("org.apache.eagle.alert.metric"))
- .withLoggingLevel(config.hasPath("level") ? getLoggingLevel(config.getString("level")) : Slf4jReporter.LoggingLevel.INFO)
- .convertRatesTo(TimeUnit.SECONDS)
- .convertDurationsTo(TimeUnit.MILLISECONDS)
- .build();
- }
-
- @Override
- public void start(long period, TimeUnit unit) {
- reporter.start(period, unit);
- }
-
- @Override
- public void stop() {
- reporter.stop();
- reporter.close();
- }
-
- @Override
- public void report() {
- reporter.report();
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/JVMMetricSource.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/JVMMetricSource.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/JVMMetricSource.java
deleted file mode 100644
index 8261a25..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/JVMMetricSource.java
+++ /dev/null
@@ -1,41 +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.metric.source;
-
-import com.codahale.metrics.JvmAttributeGaugeSet;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
-
-public class JVMMetricSource implements MetricSource {
-
- private MetricRegistry registry = new MetricRegistry();
-
- public JVMMetricSource() {
- registry.registerAll(new JvmAttributeGaugeSet());
- registry.registerAll(new MemoryUsageGaugeSet());
- }
-
- @Override
- public String name() {
- return "jvm";
- }
-
- @Override
- public MetricRegistry registry() {
- return registry;
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSource.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSource.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSource.java
deleted file mode 100644
index 180fa97..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSource.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.metric.source;
-
-import com.codahale.metrics.MetricRegistry;
-
-public interface MetricSource {
- String name();
-
- MetricRegistry registry();
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSourceWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSourceWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSourceWrapper.java
deleted file mode 100644
index d83576c..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/metric/source/MetricSourceWrapper.java
+++ /dev/null
@@ -1,39 +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.metric.source;
-
-import com.codahale.metrics.MetricRegistry;
-
-public class MetricSourceWrapper implements MetricSource {
- private final MetricRegistry registry;
- private final String name;
-
- public MetricSourceWrapper(String name, MetricRegistry registry) {
- this.name = name;
- this.registry = registry;
- }
-
- @Override
- public String name() {
- return name;
- }
-
- @Override
- public MetricRegistry registry() {
- return registry;
- }
-}
\ 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-common/src/main/java/org/apache/eagle/alert/resource/SimpleCORSFiler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/resource/SimpleCORSFiler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/resource/SimpleCORSFiler.java
deleted file mode 100644
index f0e0f87..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/resource/SimpleCORSFiler.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
- *
- * 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.resource;
-
-import java.io.IOException;
-
-import javax.servlet.Filter;
-import javax.servlet.FilterChain;
-import javax.servlet.FilterConfig;
-import javax.servlet.ServletException;
-import javax.servlet.ServletRequest;
-import javax.servlet.ServletResponse;
-import javax.servlet.http.HttpServletResponse;
-
-/**
- * A simple allow all CORS filter that works with swagger UI. Tomcat CORS filter
- * doesn't support Origin: null case which is the swagger UI request.
- *
- * @since Apr 15, 2016
- */
-public class SimpleCORSFiler implements Filter {
-
- @Override
- public void init(FilterConfig filterConfig) throws ServletException {
-
- }
-
- @Override
- public void doFilter(ServletRequest request, ServletResponse res, FilterChain chain) throws IOException,
- ServletException {
- HttpServletResponse response = (HttpServletResponse) res;
- response.setHeader("Access-Control-Allow-Origin", "*");
- response.setHeader("Access-Control-Allow-Methods", "HEAD, POST, GET, PUT, OPTIONS, DELETE");
- response.setHeader("Access-Control-Max-Age", "3600");
- response.setHeader("Access-Control-Allow-Headers",
- "Origin, Accept, X-Requested-With, Content-Type, Access-Control-Request-Method, Access-Control-Request-Headers, Authorization");
- chain.doFilter(request, response);
- }
-
- @Override
- public void destroy() {
-
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/IMetadataServiceClient.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/IMetadataServiceClient.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/IMetadataServiceClient.java
deleted file mode 100644
index efa6d0e..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/IMetadataServiceClient.java
+++ /dev/null
@@ -1,100 +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.service;
-
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.coordination.model.internal.Topology;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.Publishment;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamingCluster;
-import org.apache.eagle.alert.engine.model.AlertPublishEvent;
-
-import java.io.Closeable;
-import java.io.Serializable;
-import java.util.List;
-
-/**
- * service stub to get metadata from remote metadata service.
- */
-public interface IMetadataServiceClient extends Closeable, Serializable {
-
- // user metadta
- void addStreamingCluster(StreamingCluster cluster);
-
- void addStreamingClusters(List<StreamingCluster> clusters);
-
- List<StreamingCluster> listClusters();
-
- List<Topology> listTopologies();
-
- void addTopology(Topology t);
-
- void addTopologies(List<Topology> topologies);
-
- void addPolicy(PolicyDefinition policy);
-
- void addPolicies(List<PolicyDefinition> policies);
-
- List<PolicyDefinition> listPolicies();
-
- void addStreamDefinition(StreamDefinition streamDef);
-
- void addStreamDefinitions(List<StreamDefinition> streamDefs);
-
- List<StreamDefinition> listStreams();
-
- void addDataSource(Kafka2TupleMetadata k2t);
-
- void addDataSources(List<Kafka2TupleMetadata> k2ts);
-
- List<Kafka2TupleMetadata> listDataSources();
-
- void addPublishment(Publishment pub);
-
- void addPublishments(List<Publishment> pubs);
-
- List<Publishment> listPublishment();
-
- // monitor metadata
- List<SpoutSpec> listSpoutMetadata();
-
- ScheduleState getVersionedSpec();
-
- ScheduleState getVersionedSpec(String version);
-
- void addScheduleState(ScheduleState state);
-
- void clear();
-
- void clearScheduleState(int maxCapacity);
-
- // for topology mgmt
-
- // for alert event
- List<AlertPublishEvent> listAlertPublishEvent();
-
- void addAlertPublishEvent(AlertPublishEvent event);
-
- void addAlertPublishEvents(List<AlertPublishEvent> events);
-
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/MetadataServiceClientImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/MetadataServiceClientImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/MetadataServiceClientImpl.java
deleted file mode 100644
index 4565d96..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/service/MetadataServiceClientImpl.java
+++ /dev/null
@@ -1,305 +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.service;
-
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.coordination.model.internal.Topology;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.Publishment;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamingCluster;
-
-import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider;
-import com.sun.jersey.api.client.Client;
-import com.sun.jersey.api.client.ClientResponse;
-import com.sun.jersey.api.client.GenericType;
-import com.sun.jersey.api.client.WebResource;
-import com.sun.jersey.api.client.config.ClientConfig;
-import com.sun.jersey.api.client.config.DefaultClientConfig;
-import com.sun.jersey.client.urlconnection.URLConnectionClientHandler;
-import com.typesafe.config.Config;
-import org.apache.eagle.alert.engine.model.AlertPublishEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import javax.ws.rs.core.MediaType;
-
-public class MetadataServiceClientImpl implements IMetadataServiceClient {
- private static final long serialVersionUID = 3003976065082684128L;
-
- private static final Logger LOG = LoggerFactory.getLogger(MetadataServiceClientImpl.class);
-
- private static final String METADATA_SCHEDULESTATES_PATH = "/metadata/schedulestates";
- private static final String METADATA_PUBLISHMENTS_PATH = "/metadata/publishments";
- private static final String METADATA_DATASOURCES_PATH = "/metadata/datasources";
- private static final String METADATA_STREAMS_PATH = "/metadata/streams";
- private static final String METADATA_POLICIES_PATH = "/metadata/policies";
- private static final String METADATA_CLUSTERS_PATH = "/metadata/clusters";
- private static final String METADATA_TOPOLOGY_PATH = "/metadata/topologies";
- private static final String METADATA_ALERTS_PATH = "/metadata/alerts";
-
- private static final String METADATA_PUBLISHMENTS_BATCH_PATH = "/metadata/publishments/batch";
- private static final String METADATA_DATASOURCES_BATCH_PATH = "/metadata/datasources/batch";
- private static final String METADATA_STREAMS_BATCH_PATH = "/metadata/streams/batch";
- private static final String METADATA_POLICIES_BATCH_PATH = "/metadata/policies/batch";
- private static final String METADATA_CLUSTERS_BATCH_PATH = "/metadata/clusters/batch";
- private static final String METADATA_TOPOLOGY_BATCH_PATH = "/metadata/topologies/batch";
- private static final String METADATA_ALERTS_BATCH_PATH = "/metadata/alerts/batch";
-
- private static final String METADATA_CLEAR_PATH = "/metadata/clear";
- private static final String METADATA_CLEAR_SCHEDULESTATES_PATH = "/metadata/clear/schedulestates";
-
- private static final String EAGLE_CORRELATION_CONTEXT = "metadataService.context";
- public static final String EAGLE_CORRELATION_SERVICE_PORT = "metadataService.port";
- public static final String EAGLE_CORRELATION_SERVICE_HOST = "metadataService.host";
-
- protected static final String CONTENT_TYPE = "Content-Type";
-
- private String host;
- private int port;
- private String context;
- private transient Client client;
- private String basePath;
-
- public MetadataServiceClientImpl(Config config) {
- this(config.getString(EAGLE_CORRELATION_SERVICE_HOST), config.getInt(EAGLE_CORRELATION_SERVICE_PORT), config
- .getString(EAGLE_CORRELATION_CONTEXT));
- basePath = buildBasePath();
- }
-
- public MetadataServiceClientImpl(String host, int port, String context) {
- this.host = host;
- this.port = port;
- this.context = context;
- this.basePath = buildBasePath();
- ClientConfig cc = new DefaultClientConfig();
- cc.getProperties().put(DefaultClientConfig.PROPERTY_CONNECT_TIMEOUT, 60 * 1000);
- cc.getProperties().put(DefaultClientConfig.PROPERTY_READ_TIMEOUT, 60 * 1000);
- cc.getClasses().add(JacksonJsonProvider.class);
- cc.getProperties().put(URLConnectionClientHandler.PROPERTY_HTTP_URL_CONNECTION_SET_METHOD_WORKAROUND, true);
- this.client = Client.create(cc);
- client.addFilter(new com.sun.jersey.api.client.filter.GZIPContentEncodingFilter());
- }
-
- private String buildBasePath() {
- StringBuilder sb = new StringBuilder();
- sb.append("http://");
- sb.append(host);
- sb.append(":");
- sb.append(port);
- sb.append(context);
- return sb.toString();
- }
-
- @Override
- public void close() throws IOException {
- client.destroy();
- }
-
- @Override
- public List<SpoutSpec> listSpoutMetadata() {
- ScheduleState state = getVersionedSpec();
- return new ArrayList<>(state.getSpoutSpecs().values());
- }
-
- @Override
- public List<StreamingCluster> listClusters() {
- return list(METADATA_CLUSTERS_PATH, new GenericType<List<StreamingCluster>>() {
- });
- }
-
- @Override
- public List<PolicyDefinition> listPolicies() {
- return list(METADATA_POLICIES_PATH, new GenericType<List<PolicyDefinition>>() {
- });
- }
-
- @Override
- public List<StreamDefinition> listStreams() {
- return list(METADATA_STREAMS_PATH, new GenericType<List<StreamDefinition>>() {
- });
- }
-
- @Override
- public List<Kafka2TupleMetadata> listDataSources() {
- return list(METADATA_DATASOURCES_PATH, new GenericType<List<Kafka2TupleMetadata>>() {
- });
- }
-
- private <T> List<T> list(String path, GenericType<List<T>> type) {
- WebResource r = client.resource(basePath + path);
- LOG.info("Requesting {}", basePath + path);
- List<T> ret = r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).get(type);
- return ret;
- }
-
- @Override
- public List<Publishment> listPublishment() {
- return list(METADATA_PUBLISHMENTS_PATH, new GenericType<List<Publishment>>() {
- });
- }
-
- @Override
- public ScheduleState getVersionedSpec(String version) {
- return listOne(METADATA_SCHEDULESTATES_PATH + "/" + version, ScheduleState.class);
- }
-
- @Override
- public ScheduleState getVersionedSpec() {
- return listOne(METADATA_SCHEDULESTATES_PATH, ScheduleState.class);
- }
-
- private <T> T listOne(String path, Class<T> tClz) {
- LOG.info("Requesting {}", basePath + path);
- WebResource r = client.resource(basePath + path);
-
- ClientResponse resp = r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON)
- .get(ClientResponse.class);
- if (resp.getStatus() < 300) {
- try {
- return resp.getEntity(tClz);
- } catch (Exception e) {
- LOG.warn("List one entity failed, ignored and continue, path:{}", path, e);
- }
- } else {
- LOG.warn("Fail querying metadata service {}, http status: {}", basePath + path, resp.getStatus());
- }
- return null;
- }
-
- @Override
- public void addScheduleState(ScheduleState state) {
- WebResource r = client.resource(basePath + METADATA_SCHEDULESTATES_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(state);
- }
-
- @Override
- public List<Topology> listTopologies() {
- return list(METADATA_TOPOLOGY_PATH, new GenericType<List<Topology>>() {
- });
- }
-
- @Override
- public void addStreamingCluster(StreamingCluster cluster) {
- WebResource r = client.resource(basePath + METADATA_CLUSTERS_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(cluster);
- }
-
- @Override
- public void addStreamingClusters(List<StreamingCluster> clusters) {
- WebResource r = client.resource(basePath + METADATA_CLUSTERS_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(clusters);
- }
-
- @Override
- public void addTopology(Topology t) {
- WebResource r = client.resource(basePath + METADATA_TOPOLOGY_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(t);
- }
-
- @Override
- public void addTopologies(List<Topology> topologies) {
- WebResource r = client.resource(basePath + METADATA_TOPOLOGY_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(topologies);
- }
-
- @Override
- public void addPolicy(PolicyDefinition policy) {
- WebResource r = client.resource(basePath + METADATA_POLICIES_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(policy);
- }
-
- @Override
- public void addPolicies(List<PolicyDefinition> policies) {
- WebResource r = client.resource(basePath + METADATA_POLICIES_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(policies);
- }
-
- @Override
- public void addStreamDefinition(StreamDefinition streamDef) {
- WebResource r = client.resource(basePath + METADATA_STREAMS_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(streamDef);
- }
-
- @Override
- public void addStreamDefinitions(List<StreamDefinition> streamDefs) {
- WebResource r = client.resource(basePath + METADATA_STREAMS_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(streamDefs);
- }
-
- @Override
- public void addDataSource(Kafka2TupleMetadata k2t) {
- WebResource r = client.resource(basePath + METADATA_DATASOURCES_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(k2t);
- }
-
- @Override
- public void addDataSources(List<Kafka2TupleMetadata> k2ts) {
- WebResource r = client.resource(basePath + METADATA_DATASOURCES_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(k2ts);
- }
-
- @Override
- public void addPublishment(Publishment pub) {
- WebResource r = client.resource(basePath + METADATA_PUBLISHMENTS_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(pub);
- }
-
- @Override
- public void addPublishments(List<Publishment> pubs) {
- WebResource r = client.resource(basePath + METADATA_PUBLISHMENTS_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(pubs);
- }
-
- @Override
- public void clear() {
- WebResource r = client.resource(basePath + METADATA_CLEAR_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post();
- }
-
- @Override
- public void clearScheduleState(int maxCapacity) {
- WebResource r = client.resource(basePath + METADATA_CLEAR_SCHEDULESTATES_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(maxCapacity);
- }
-
- @Override
- public List<AlertPublishEvent> listAlertPublishEvent() {
- return list(METADATA_ALERTS_PATH, new GenericType<List<AlertPublishEvent>>(){});
- }
-
- @Override
- public void addAlertPublishEvent(AlertPublishEvent event) {
- WebResource r = client.resource(basePath + METADATA_ALERTS_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(event);
- }
-
- @Override
- public void addAlertPublishEvents(List<AlertPublishEvent> events) {
- WebResource r = client.resource(basePath + METADATA_ALERTS_BATCH_PATH);
- r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(events);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
deleted file mode 100644
index e047662..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringEmptyFunctionExtension.java
+++ /dev/null
@@ -1,86 +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.siddhiext;
-
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.core.executor.function.FunctionExecutor;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-public class StringEmptyFunctionExtension extends FunctionExecutor {
- /**
- * The initialization method for StringEmptyFunctionExtension, this method will be called before the other methods.
- *
- * @param attributeExpressionExecutors the executors of each function parameter
- * @param executionPlanContext the context of the execution plan
- */
- @Override
- protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
- if (attributeExpressionExecutors.length != 1) {
- throw new ExecutionPlanValidationException("Invalid no of arguments passed to string:empty() function, "
- + "required 1, but found " + attributeExpressionExecutors.length);
- }
-
- Attribute.Type attributeType = attributeExpressionExecutors[0].getReturnType();
- if (attributeType != Attribute.Type.STRING) {
- throw new ExecutionPlanValidationException("Invalid parameter type found for the argument of math:string() function, "
- + "required " + Attribute.Type.STRING
- + ", but found " + attributeType.toString());
- }
- }
-
- /**
- * The main execution method which will be called upon event arrival.
- * when there are more than one function parameter
- *
- * @param data the runtime values of function parameters
- * @return the function result
- */
- @Override
- protected Object execute(Object[] data) {
- return null;
- }
-
- @Override
- protected Object execute(Object data) {
- return !(data == null || ((String) data).isEmpty());
- }
-
- @Override
- public void start() {
- }
-
- @Override
- public void stop() {
- }
-
- @Override
- public Attribute.Type getReturnType() {
- return Attribute.Type.BOOL;
- }
-
- @Override
- public Object[] currentState() {
- return null;
- }
-
- @Override
- public void restoreState(Object[] state) {
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringListSizeFunctionExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringListSizeFunctionExtension.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringListSizeFunctionExtension.java
deleted file mode 100644
index 5d3d3ae..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringListSizeFunctionExtension.java
+++ /dev/null
@@ -1,100 +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.siddhiext;
-
-import org.apache.commons.collections.ListUtils;
-import org.apache.eagle.alert.utils.JsonUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.core.executor.function.FunctionExecutor;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-import java.util.List;
-
-public class StringListSizeFunctionExtension extends FunctionExecutor {
- private static final Logger LOG = LoggerFactory.getLogger(StringListSizeFunctionExtension.class);
-
- /**
- * The initialization method for StringListSizeFunctionExtension, this method will be called before the other methods.
- *
- * @param attributeExpressionExecutors the executors of each function parameter
- * @param executionPlanContext the context of the execution plan
- */
- @Override
- protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
- if (attributeExpressionExecutors.length != 1) {
- throw new ExecutionPlanValidationException("Invalid no of arguments passed to string:listSize() function, "
- + "required 1, but found " + attributeExpressionExecutors.length);
- }
-
- Attribute.Type attributeType = attributeExpressionExecutors[0].getReturnType();
- if (attributeType != Attribute.Type.STRING) {
- throw new ExecutionPlanValidationException("Invalid parameter type found for the argument of string:listSize() function, "
- + "required " + Attribute.Type.STRING
- + ", but found " + attributeType.toString());
- }
- }
-
- /**
- * The main execution method which will be called upon event arrival.
- * when there are more than one function parameter
- * This method calculates subtraction of two List Of Strings
- * Each String is a jobs string needs to be loaded
- * @param data the runtime values of function parameters
- * @return the function result
- */
- @Override
- protected Object execute(Object[] data) {
- return null;
- }
-
- @Override
- protected Object execute(Object data) {
- try {
- return JsonUtils.jsonStringToList((String)data).size();
- } catch (Exception e) {
- LOG.warn("exception found {}", e);
- return 0;
- }
- }
-
- @Override
- public void start() {
- }
-
- @Override
- public void stop() {
- }
-
- @Override
- public Attribute.Type getReturnType() {
- return Attribute.Type.INT;
- }
-
- @Override
- public Object[] currentState() {
- return null;
- }
-
- @Override
- public void restoreState(Object[] state) {
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringSubtractFunctionExtension.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringSubtractFunctionExtension.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringSubtractFunctionExtension.java
deleted file mode 100644
index 9d26adf..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/siddhiext/StringSubtractFunctionExtension.java
+++ /dev/null
@@ -1,106 +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.siddhiext;
-
-import org.apache.commons.collections.ListUtils;
-import org.apache.eagle.alert.utils.JsonUtils;
-import org.codehaus.jettison.json.JSONArray;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.wso2.siddhi.core.config.ExecutionPlanContext;
-import org.wso2.siddhi.core.executor.ExpressionExecutor;
-import org.wso2.siddhi.core.executor.function.FunctionExecutor;
-import org.wso2.siddhi.query.api.definition.Attribute;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-public class StringSubtractFunctionExtension extends FunctionExecutor {
- private static final Logger LOG = LoggerFactory.getLogger(StringSubtractFunctionExtension.class);
-
- /**
- * The initialization method for StringSubtractFunctionExtension, this method will be called before the other methods.
- *
- * @param attributeExpressionExecutors the executors of each function parameter
- * @param executionPlanContext the context of the execution plan
- */
- @Override
- protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
- if (attributeExpressionExecutors.length != 2) {
- throw new ExecutionPlanValidationException("Invalid no of arguments passed to string:subtract() function, "
- + "required 2, but found " + attributeExpressionExecutors.length);
- }
-
- Attribute.Type attributeType = attributeExpressionExecutors[0].getReturnType();
- if (attributeType != Attribute.Type.STRING) {
- throw new ExecutionPlanValidationException("Invalid parameter type found for the argument of string:subtract() function, "
- + "required " + Attribute.Type.STRING
- + ", but found " + attributeType.toString());
- }
- }
-
- /**
- * The main execution method which will be called upon event arrival.
- * when there are more than one function parameter
- * This method calculates subtraction of two List Of Strings
- * Each String is a jobs string needs to be loaded
- * @param data the runtime values of function parameters
- * @return the function result
- */
- @Override
- protected Object execute(Object[] data) {
- try {
- List<String> ths = JsonUtils.jsonStringToList((String) data[0]);
- List<String> rhs = JsonUtils.jsonStringToList((String) data[1]);
-
- return org.apache.commons.lang.StringUtils.join(ListUtils.subtract(ths, rhs), "\n");
- } catch (Exception e) {
- LOG.warn("exception found {}", e);
- return null;
- }
- }
-
- @Override
- protected Object execute(Object data) {
- return null;
- }
-
- @Override
- public void start() {
- }
-
- @Override
- public void stop() {
- }
-
- @Override
- public Attribute.Type getReturnType() {
- return Attribute.Type.STRING;
- }
-
- @Override
- public Object[] currentState() {
- return null;
- }
-
- @Override
- public void restoreState(Object[] state) {
- }
-}
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/AlertConstants.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/AlertConstants.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/AlertConstants.java
deleted file mode 100644
index 2740836..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/AlertConstants.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
- * <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.utils;
-
-public class AlertConstants {
- public static final String FIELD_0 = "f0";
- public static final String FIELD_1 = "f1";
- public static final String FIELD_2 = "f2";
- public static final String FIELD_3 = "f3";
-
- public static final String DEFAULT_SPOUT_NAME = "alertEngineSpout";
- public static final String DEFAULT_ROUTERBOLT_NAME = "streamRouterBolt";
-
- public static final String ALERT_SERVICE_ENDPOINT_NAME = "AlertService";
-
- public static final String COORDINATOR = "coordinator";
-
- public static final String KAFKA_BROKER_ZK_BASE_PATH = "spout.kafkaBrokerZkBasePath";
- public static final String KAFKA_BROKER_ZK_QUORUM = "spout.kafkaBrokerZkQuorum";
-}
\ 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-common/src/main/java/org/apache/eagle/alert/utils/ByteUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/ByteUtils.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/ByteUtils.java
deleted file mode 100644
index 53fc4ac..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/utils/ByteUtils.java
+++ /dev/null
@@ -1,188 +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.utils;
-
-import java.io.UnsupportedEncodingException;
-
-public class ByteUtils {
-
- public static double bytesToDouble(byte[] bytes, int offset) {
- return Double.longBitsToDouble(bytesToLong(bytes, offset));
- }
-
- public static double bytesToDouble(byte[] bytes) {
- return Double.longBitsToDouble(bytesToLong(bytes));
- }
-
- public static void doubleToBytes(double v, byte[] bytes) {
- doubleToBytes(v, bytes, 0);
- }
-
- public static void doubleToBytes(double v, byte[] bytes, int offset) {
- longToBytes(Double.doubleToLongBits(v), bytes, offset);
- }
-
- public static byte[] doubleToBytes(double v) {
- return longToBytes(Double.doubleToLongBits(v));
- }
-
- public static long bytesToLong(byte[] bytes) {
- return bytesToLong(bytes, 0);
- }
-
- public static long bytesToLong(byte[] bytes, int offset) {
- long value = 0;
- for (int i = 0; i < 8; i++) {
- value <<= 8;
- value |= (bytes[i + offset] & 0xFF);
- }
- return value;
- }
-
- public static void longToBytes(long v, byte[] bytes) {
- longToBytes(v, bytes, 0);
- }
-
- public static void longToBytes(long v, byte[] bytes, int offset) {
- long tmp = v;
- for (int i = 0; i < 8; i++) {
- bytes[offset + 7 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- }
-
- public static byte[] longToBytes(long v) {
- long tmp = v;
- byte[] b = new byte[8];
- for (int i = 0; i < 8; i++) {
- b[7 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- return b;
- }
-
- public static int bytesToInt(byte[] bytes) {
- return bytesToInt(bytes, 0);
- }
-
- public static int bytesToInt(byte[] bytes, int offset) {
- int value = 0;
- for (int i = 0; i < 4; i++) {
- value <<= 8;
- value |= (bytes[i + offset] & 0xFF);
- }
- return value;
- }
-
- public static void intToBytes(int v, byte[] bytes) {
- intToBytes(v, bytes, 0);
- }
-
- public static void intToBytes(int v, byte[] bytes, int offset) {
- int tmp = v;
- for (int i = 0; i < 4; i++) {
- bytes[offset + 3 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- }
-
- public static byte[] intToBytes(int v) {
- int tmp = v;
- byte[] b = new byte[4];
- for (int i = 0; i < 4; i++) {
- b[3 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- return b;
- }
-
- //////
-
- public static short bytesToShort(byte[] bytes) {
- return bytesToShort(bytes, 0);
- }
-
- public static short bytesToShort(byte[] bytes, int offset) {
- short value = 0;
- for (int i = 0; i < 2; i++) {
- value <<= 8;
- value |= (bytes[i + offset] & 0xFF);
- }
- return value;
- }
-
- public static void shortToBytes(short v, byte[] bytes) {
- shortToBytes(v, bytes, 0);
- }
-
- public static void shortToBytes(short v, byte[] bytes, int offset) {
- int tmp = v;
- for (int i = 0; i < 2; i++) {
- bytes[offset + 1 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- }
-
- public static byte[] shortToBytes(short v) {
- int tmp = v;
- byte[] b = new byte[2];
- for (int i = 0; i < 2; i++) {
- b[1 - i] = (byte) (tmp & 0xFF);
- tmp >>= 8;
- }
- return b;
- }
-
- public static byte[] concat(byte[]... arrays) {
- int length = 0;
- for (byte[] array : arrays) {
- length += array.length;
- }
- byte[] result = new byte[length];
- int pos = 0;
- for (byte[] array : arrays) {
- System.arraycopy(array, 0, result, pos, array.length);
- pos += array.length;
- }
- return result;
- }
-
- public static byte[] stringToBytes(String str) {
- try {
- return str.getBytes("UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new IllegalStateException(e);
- }
- }
-
- // public static void main(String[] args){
- // int a = "ThreadName".hashCode();
- // byte[] b = intToBytes(a);
- // byte[] c = intToBytes(1676687583);
- // String s = new String(b);
- // System.out.println(s);
-
- // byte[] d = intToBytes(8652353);
- // System.out.println(bytesToInt(d));
-
- // byte[] e = longToBytes(12131513513l);
- // System.out.println(bytesToLong(e));
- // if(12131513513l == bytesToLong(e)){
- // System.out.println("yes");
- // }
- // }
-}
\ No newline at end of file