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 2016/06/02 07:07:58 UTC
[19/46] incubator-eagle git commit: [EAGLE-325] Initialize next-gen
alert engine code on branch-0.5
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/InMemScheduleConext.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/InMemScheduleConext.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/InMemScheduleConext.java
deleted file mode 100644
index 84a4061..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/InMemScheduleConext.java
+++ /dev/null
@@ -1,148 +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.coordinator.provider;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.internal.MonitoredStream;
-import org.apache.eagle.alert.coordination.model.internal.PolicyAssignment;
-import org.apache.eagle.alert.coordination.model.internal.StreamGroup;
-import org.apache.eagle.alert.coordination.model.internal.Topology;
-import org.apache.eagle.alert.coordinator.IScheduleContext;
-import org.apache.eagle.alert.coordinator.model.TopologyUsage;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.Publishment;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-
-/**
- * @since Mar 28, 2016
- *
- */
-public class InMemScheduleConext implements IScheduleContext {
-
- private Map<String, Topology> topologies = new HashMap<String, Topology>();
- private Map<String, TopologyUsage> usages = new HashMap<String, TopologyUsage>();
- private Map<String, PolicyDefinition> policies = new HashMap<String, PolicyDefinition>();
- private Map<String, Kafka2TupleMetadata> datasources = new HashMap<String, Kafka2TupleMetadata>();
- private Map<String, PolicyAssignment> policyAssignments = new HashMap<String, PolicyAssignment>();
- private Map<String, StreamDefinition> schemas = new HashMap<String, StreamDefinition>();
- private Map<StreamGroup, MonitoredStream> monitoredStreams = new HashMap<StreamGroup, MonitoredStream>();
- private Map<String, Publishment> publishments = new HashMap<String, Publishment>();
-
- public InMemScheduleConext() {
- }
-
- public InMemScheduleConext(IScheduleContext context) {
- this.topologies = new HashMap<String, Topology>(context.getTopologies());
- this.usages = new HashMap<String, TopologyUsage>(context.getTopologyUsages());
- this.policies = new HashMap<String, PolicyDefinition>(context.getPolicies());
- this.datasources = new HashMap<String, Kafka2TupleMetadata>(context.getDataSourceMetadata());
- this.policyAssignments = new HashMap<String, PolicyAssignment>(context.getPolicyAssignments());
- this.schemas = new HashMap<String, StreamDefinition>(context.getStreamSchemas());
- this.monitoredStreams = new HashMap<StreamGroup, MonitoredStream>(context.getMonitoredStreams());
- this.publishments = new HashMap<String, Publishment>(context.getPublishments());
- }
-
- public InMemScheduleConext(Map<String, Topology> topologies2, Map<String, PolicyAssignment> assignments,
- Map<String, Kafka2TupleMetadata> kafkaSources, Map<String, PolicyDefinition> policies2,
- Map<String, Publishment> publishments2, Map<String, StreamDefinition> streamDefinitions,
- Map<StreamGroup, MonitoredStream> monitoredStreamMap, Map<String, TopologyUsage> usages2) {
- this.topologies = topologies2;
- this.policyAssignments = assignments;
- this.datasources = kafkaSources;
- this.policies = policies2;
- this.publishments = publishments2;
- this.schemas = streamDefinitions;
- this.monitoredStreams = monitoredStreamMap;
- this.usages = usages2;
- }
-
- public Map<String, Topology> getTopologies() {
- return topologies;
- }
-
- public void addTopology(Topology topo) {
- topologies.put(topo.getName(), topo);
- }
-
- public Map<String, TopologyUsage> getTopologyUsages() {
- return usages;
- }
-
- public void addTopologyUsages(TopologyUsage usage) {
- usages.put(usage.getTopoName(), usage);
- }
-
- public Map<String, PolicyDefinition> getPolicies() {
- return policies;
- }
-
- public void addPoilcy(PolicyDefinition pd) {
- this.policies.put(pd.getName(), pd);
- }
-
- public Map<String, Kafka2TupleMetadata> getDatasources() {
- return datasources;
- }
-
- public void setDatasources(Map<String, Kafka2TupleMetadata> datasources) {
- this.datasources = datasources;
- }
-
- public void addDataSource(Kafka2TupleMetadata dataSource) {
- this.datasources.put(dataSource.getName(), dataSource);
- }
-
- @Override
- public Map<String, Kafka2TupleMetadata> getDataSourceMetadata() {
- return datasources;
- }
-
- public void setPolicyOrderedTopologies(Map<String, PolicyAssignment> policyAssignments) {
- this.policyAssignments = policyAssignments;
- }
-
- public Map<String, PolicyAssignment> getPolicyAssignments() {
- return this.policyAssignments;
- }
-
- @Override
- public Map<String, StreamDefinition> getStreamSchemas() {
- return schemas;
- }
-
- public void addSchema(StreamDefinition schema) {
- this.schemas.put(schema.getStreamId(), schema);
- }
-
- public void setStreamSchemas(Map<String, StreamDefinition> schemas) {
- this.schemas = schemas;
- }
-
- @Override
- public Map<StreamGroup, MonitoredStream> getMonitoredStreams() {
- return monitoredStreams;
- }
-
- @Override
- public Map<String, Publishment> getPublishments() {
- return publishments;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/ScheduleContextBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/ScheduleContextBuilder.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/ScheduleContextBuilder.java
deleted file mode 100644
index d4d6c0c..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/provider/ScheduleContextBuilder.java
+++ /dev/null
@@ -1,400 +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.coordinator.provider;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordination.model.WorkSlot;
-import org.apache.eagle.alert.coordination.model.internal.MonitoredStream;
-import org.apache.eagle.alert.coordination.model.internal.PolicyAssignment;
-import org.apache.eagle.alert.coordination.model.internal.StreamGroup;
-import org.apache.eagle.alert.coordination.model.internal.StreamWorkSlotQueue;
-import org.apache.eagle.alert.coordination.model.internal.Topology;
-import org.apache.eagle.alert.coordinator.IScheduleContext;
-import org.apache.eagle.alert.coordinator.model.AlertBoltUsage;
-import org.apache.eagle.alert.coordinator.model.GroupBoltUsage;
-import org.apache.eagle.alert.coordinator.model.TopologyUsage;
-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.service.IMetadataServiceClient;
-import org.apache.eagle.alert.service.MetadataServiceClientImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.typesafe.config.Config;
-
-/**
- * FIXME: this class focus on correctness, not the efficiency now. There might
- * be problem when metadata size grows too big.
- *
- * @since May 3, 2016
- *
- */
-public class ScheduleContextBuilder {
-
- private static final Logger LOG = LoggerFactory.getLogger(ScheduleContextBuilder.class);
- private static final String UNIQUE_BOLT_ID = "%s-%s";// toponame-boltname
-
- private IMetadataServiceClient client;
-
- private Map<String, Topology> topologies;
- private Map<String, PolicyAssignment> assignments;
- private Map<String, Kafka2TupleMetadata> kafkaSources;
- private Map<String, PolicyDefinition> policies;
- private Map<String, Publishment> publishments;
- private Map<String, StreamDefinition> streamDefinitions;
- private Map<StreamGroup, MonitoredStream> monitoredStreamMap;
- private Map<String, TopologyUsage> usages;
-
- public ScheduleContextBuilder(Config config) {
- client = new MetadataServiceClientImpl(config);
- }
-
- public ScheduleContextBuilder(IMetadataServiceClient client) {
- this.client = client;
- }
-
- /**
- * Built a shcedule context for metadata client service.
- *
- * @return
- */
- public IScheduleContext buildContext() {
- topologies = listToMap(client.listTopologies());
- kafkaSources = listToMap(client.listDataSources());
- policies = listToMap(client.listPolicies());
- publishments = listToMap(client.listPublishment());
- streamDefinitions = listToMap(client.listStreams());
-
- // TODO: See ScheduleState comments on how to improve the storage
- ScheduleState state = client.getVersionedSpec();
- assignments = listToMap(state == null ? new ArrayList<PolicyAssignment>() : cleanupDeprecatedAssignments(state.getAssignments()));
-
- monitoredStreamMap = listToMap(state == null ? new ArrayList<MonitoredStream>() : cleanupDeprecatedStreamsAndAssignment(state.getMonitoredStreams()));
-
- // build based on existing data
- usages = buildTopologyUsage();
-
- // copy to shedule context now
- return new InMemScheduleConext(topologies, assignments, kafkaSources, policies, publishments,
- streamDefinitions, monitoredStreamMap, usages);
- }
-
- /**
- * 1.
- * <pre>
- * Check for deprecated policy stream group with its assigned monitored stream groups.
- * If this is unmatched, we think the policy' stream group has been changed, remove the policy assignments
- * If finally, no assignment refer to a given monitored stream, this monitored stream could be removed.
- * Log when every time a remove happens.
- * </pre>
- * 2.
- * <pre>
- * if monitored stream's queue's is on non-existing topology, remove it.
- * </pre>
- * @param monitoredStreams
- * @return
- */
- private List<MonitoredStream> cleanupDeprecatedStreamsAndAssignment(List<MonitoredStream> monitoredStreams) {
- List<MonitoredStream> result = new ArrayList<MonitoredStream>(monitoredStreams);
-
- // clear deprecated streams
- clearMonitoredStreams(monitoredStreams);
-
- // build queueId-> streamGroup
- Map<String, StreamGroup> queue2StreamGroup = new HashMap<String, StreamGroup>();
- for (MonitoredStream ms : result) {
- for (StreamWorkSlotQueue q : ms.getQueues()) {
- queue2StreamGroup.put(q.getQueueId(), ms.getStreamGroup());
- }
- }
-
- // decide the assignment delete set
- Set<StreamGroup> usedGroups = new HashSet<StreamGroup>();
- Set<String> toRemove = new HashSet<String>();
- // check if queue is still referenced by policy assignments
- for (PolicyAssignment assignment : assignments.values()) {
- PolicyDefinition def = policies.get(assignment.getPolicyName());
- StreamGroup group = queue2StreamGroup.get(assignment.getQueueId());
- if (group == null || !Objects.equals(group.getStreamPartitions(), def.getPartitionSpec())) {
- LOG.warn(" policy assgiment {} 's policy group {} is different to the monitored stream's partition group {}, "
- + "this indicates a policy stream partition spec change, the assignment would be removed! ",
- assignment, def.getPartitionSpec(), group == null ? "'not found'" :group.getStreamPartitions());
- toRemove.add(assignment.getPolicyName());
- } else {
- usedGroups.add(group);
- }
- }
-
- // remove useless
- assignments.keySet().removeAll(toRemove);
- // remove non-referenced monitored streams
- result.removeIf((t) -> {
- boolean used = usedGroups.contains(t.getStreamGroup());
- if (!used) {
- LOG.warn("monitor stream with stream group {} is not referenced, "
- + "this monitored stream and its worker queu will be removed", t.getStreamGroup());
- }
- return !used;
- });
-
- return result;
- }
-
- private void clearMonitoredStreams(List<MonitoredStream> monitoredStreams) {
- Iterator<MonitoredStream> it = monitoredStreams.iterator();
- while (it.hasNext()) {
- MonitoredStream ms = it.next();
- Iterator<StreamWorkSlotQueue> queueIt = ms.getQueues().iterator();
- // clean queue that underly topology is changed(removed/down)
- while (queueIt.hasNext()) {
- StreamWorkSlotQueue queue = queueIt.next();
- boolean deprecated = false;
- for (WorkSlot ws : queue.getWorkingSlots()) {
- // check if topology available or bolt available
- if (!topologies.containsKey(ws.topologyName)
- || !topologies.get(ws.topologyName).getAlertBoltIds().contains(ws.boltId)) {
- deprecated = true;
- break;
- }
- }
- if (deprecated) {
- queueIt.remove();
- }
- }
-
- if (ms.getQueues().isEmpty()) {
- it.remove();
- }
- }
- }
-
- private List<PolicyAssignment> cleanupDeprecatedAssignments(List<PolicyAssignment> list) {
- List<PolicyAssignment> result = new ArrayList<PolicyAssignment>(list);
- Iterator<PolicyAssignment> paIt = result.iterator();
- while (paIt.hasNext()) {
- PolicyAssignment assignment = paIt.next();
- if (!policies.containsKey(assignment.getPolicyName())) {
- LOG.info("Policy assignment {} 's policy not found, this assignment will be removed!", assignment);
- paIt.remove();
- }
- }
- return result;
- }
-
- private <T, K> Map<K, T> listToMap(List<T> collections) {
- Map<K, T> maps = new HashMap<K, T>(collections.size());
- for (T t : collections) {
- maps.put(getKey(t), t);
- }
- return maps;
- }
-
- /*
- * One drawback, once we add class, this code need to be changed!
- */
- @SuppressWarnings("unchecked")
- private <T, K> K getKey(T t) {
- if (t instanceof Topology) {
- return (K) ((Topology) t).getName();
- } else if (t instanceof PolicyAssignment) {
- return (K) ((PolicyAssignment) t).getPolicyName();
- } else if (t instanceof Kafka2TupleMetadata) {
- return (K) ((Kafka2TupleMetadata) t).getName();
- } else if (t instanceof PolicyDefinition) {
- return (K) ((PolicyDefinition) t).getName();
- } else if (t instanceof Publishment) {
- return (K) ((Publishment) t).getName();
- } else if (t instanceof StreamDefinition) {
- return (K) ((StreamDefinition) t).getStreamId();
- } else if (t instanceof MonitoredStream) {
- return (K) ((MonitoredStream) t).getStreamGroup();
- }
- throw new RuntimeException("unexpected key class " + t.getClass());
- }
-
- private Map<String, TopologyUsage> buildTopologyUsage() {
- Map<String, TopologyUsage> usages = new HashMap<String, TopologyUsage>();
-
- // pre-build data structure for help
- Map<String, Set<MonitoredStream>> topo2MonitorStream = new HashMap<String, Set<MonitoredStream>>();
- Map<String, Set<String>> topo2Policies = new HashMap<String, Set<String>>();
- // simply assume no bolt with the same id
- Map<String, Set<String>> bolt2Policies = new HashMap<String, Set<String>>();
- // simply assume no bolt with the same id
- Map<String, Set<StreamGroup>> bolt2Partition = new HashMap<String, Set<StreamGroup>>();
- // simply assume no bolt with the same id
- Map<String, Set<String>> bolt2QueueIds = new HashMap<String, Set<String>>();
- Map<String, StreamWorkSlotQueue> queueMap = new HashMap<String, StreamWorkSlotQueue>();
-
- preBuildQueue2TopoMap(topo2MonitorStream, topo2Policies, bolt2Policies, bolt2Partition, bolt2QueueIds, queueMap);
-
- for (Topology t : topologies.values()) {
- TopologyUsage u = new TopologyUsage(t.getName());
- // add group/bolt usages
- for (String grpBolt : t.getGroupNodeIds()) {
- GroupBoltUsage grpUsage = new GroupBoltUsage(grpBolt);
- u.getGroupUsages().put(grpBolt, grpUsage);
- }
- for (String alertBolt : t.getAlertBoltIds()) {
- String uniqueBoltId = String.format(UNIQUE_BOLT_ID, t.getName(), alertBolt);
-
- AlertBoltUsage alertUsage = new AlertBoltUsage(alertBolt);
- u.getAlertUsages().put(alertBolt, alertUsage);
- // complete usage
- addBoltUsageInfo(bolt2Policies, bolt2Partition, bolt2QueueIds, uniqueBoltId, alertUsage, queueMap);
- }
-
- // policy -- policy assignment
- if (topo2Policies.containsKey(u.getTopoName())) {
- u.getPolicies().addAll(topo2Policies.get(u.getTopoName()));
- }
-
- // data source
- buildTopologyDataSource(u);
-
- // topology usage monitored stream -- from monitored steams' queue slot item info
- if (topo2MonitorStream.containsKey(u.getTopoName())) {
- u.getMonitoredStream().addAll(topo2MonitorStream.get(u.getTopoName()));
- }
-
- usages.put(u.getTopoName(), u);
- }
-
- return usages;
- }
-
- private void addBoltUsageInfo(Map<String, Set<String>> bolt2Policies,
- Map<String, Set<StreamGroup>> bolt2Partition, Map<String, Set<String>> bolt2QueueIds, String uniqueAlertBolt,
- AlertBoltUsage alertUsage, Map<String, StreamWorkSlotQueue> queueMap) {
- //
- if (bolt2Policies.containsKey(uniqueAlertBolt)) {
- alertUsage.getPolicies().addAll(bolt2Policies.get(uniqueAlertBolt));
- }
- //
- if (bolt2Partition.containsKey(uniqueAlertBolt)) {
- alertUsage.getPartitions().addAll(bolt2Partition.get(uniqueAlertBolt));
- }
- //
- if (bolt2QueueIds.containsKey(uniqueAlertBolt)) {
- for (String qId : bolt2QueueIds.get(uniqueAlertBolt)) {
- if (queueMap.containsKey(qId)) {
- alertUsage.getReferQueues().add(queueMap.get(qId));
- } else {
- LOG.error(" queue id {} not found in queue map !", qId);
- }
- }
- }
- }
-
- private void buildTopologyDataSource(TopologyUsage u) {
- for (String policyName : u.getPolicies()) {
- PolicyDefinition def = policies.get(policyName);
- if (def != null) {
- u.getDataSources().addAll(findDatasource(def));
- } else {
- LOG.error(" policy not find {}, but reference in topology usage {} !", policyName, u.getTopoName());
- }
- }
- }
-
- private List<String> findDatasource(PolicyDefinition def) {
- List<String> result = new ArrayList<String>();
- List<String> inputStreams = def.getInputStreams();
- for (String is : inputStreams) {
- StreamDefinition ss = this.streamDefinitions.get(is);
- if (ss == null) {
- LOG.error("policy {} referenced stream definition {} not found in definiton !", def.getName(), is);
- } else {
- result.add(ss.getDataSource());
- }
- }
- return result;
- }
-
- private void preBuildQueue2TopoMap(
- Map<String, Set<MonitoredStream>> topo2MonitorStream,
- Map<String, Set<String>> topo2Policies,
- Map<String, Set<String>> bolt2Policies,
- Map<String, Set<StreamGroup>> bolt2Partition,
- Map<String, Set<String>> bolt2QueueIds,
- Map<String, StreamWorkSlotQueue> queueMap) {
- // pre-build structure
- // why don't reuse the queue.getPolicies
- Map<String, Set<String>> queue2Policies= new HashMap<String, Set<String>>();
- for (PolicyAssignment pa : assignments.values()) {
- if (!queue2Policies.containsKey(pa.getQueueId())) {
- queue2Policies.put(pa.getQueueId(), new HashSet<String>());
- }
- queue2Policies.get(pa.getQueueId()).add(pa.getPolicyName());
- }
-
- for (MonitoredStream stream : monitoredStreamMap.values()) {
- for (StreamWorkSlotQueue q : stream.getQueues()) {
- queueMap.put(q.getQueueId(), q);
- Set<String> policiesOnQ = queue2Policies.containsKey(q.getQueueId()) ? queue2Policies.get(q.getQueueId()) : new HashSet<String>();
-
- for (WorkSlot slot : q.getWorkingSlots()) {
- // topo2monitoredstream
- if (!topo2MonitorStream.containsKey(slot.getTopologyName())) {
- topo2MonitorStream.put(slot.getTopologyName(), new HashSet<MonitoredStream>());
- }
- topo2MonitorStream.get(slot.getTopologyName()).add(stream);
-
- // topo2policy
- if (!topo2Policies.containsKey(slot.getTopologyName())) {
- topo2Policies.put(slot.getTopologyName(), new HashSet<String>());
- }
- topo2Policies.get(slot.getTopologyName()).addAll(policiesOnQ);
-
- // bolt2Policy
- if (!bolt2Policies.containsKey(getUniqueBoltId(slot))) {
- bolt2Policies.put(getUniqueBoltId(slot), new HashSet<String>());
- }
- bolt2Policies.get(getUniqueBoltId(slot)).addAll(policiesOnQ);
-
- // bolt2Queue
- if (!bolt2QueueIds.containsKey(getUniqueBoltId(slot))) {
- bolt2QueueIds.put(getUniqueBoltId(slot), new HashSet<String>());
- }
- bolt2QueueIds.get(getUniqueBoltId(slot)).add(q.getQueueId());
-
- // bolt2Partition
- if (!bolt2Partition.containsKey(getUniqueBoltId(slot))) {
- bolt2Partition.put(getUniqueBoltId(slot), new HashSet<StreamGroup>());
- }
- bolt2Partition.get(getUniqueBoltId(slot)).add(stream.getStreamGroup());
- }
- }
- }
- }
-
- private String getUniqueBoltId(WorkSlot slot) {
- return String.format(UNIQUE_BOLT_ID, slot.getTopologyName(), slot.getBoltId());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/resource/CoordinatorResource.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/resource/CoordinatorResource.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/resource/CoordinatorResource.java
deleted file mode 100644
index 6b8495e..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/resource/CoordinatorResource.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.coordinator.resource;
-
-import javax.ws.rs.GET;
-import javax.ws.rs.POST;
-import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordinator.Coordinator;
-import org.apache.eagle.alert.coordinator.ScheduleOption;
-import org.apache.eagle.alert.utils.JsonUtils;
-
-/**
- * This is to provide API access even we don't have ZK as intermediate access.
- * FIXME : more elogant status code
- *
- * @since Mar 24, 2016 <br/>
- */
-@Path("/coordinator")
-@Produces({ "application/json" })
-public class CoordinatorResource {
-
- // sprint config here?
- private Coordinator alertCoordinator = new Coordinator();
-
- @GET
- @Path("/assignments")
- public String getAssignments() throws Exception {
- ScheduleState state = alertCoordinator.getState();
- return JsonUtils.writeValueAsString(state);
- }
-
- @POST
- @Path("/build")
- public String build() throws Exception {
- ScheduleOption option = new ScheduleOption();
- ScheduleState state = alertCoordinator.schedule(option);
- return JsonUtils.writeValueAsString(state);
- }
-
- /**
- * Manually update the topology usages, for administration
- *
- * @return
- */
- @POST
- @Path("/refreshUsages")
- public String refreshUsages() {
- // TODO
- return "";
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/DynamicPolicyLoader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/DynamicPolicyLoader.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/DynamicPolicyLoader.java
deleted file mode 100644
index 15333da..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/DynamicPolicyLoader.java
+++ /dev/null
@@ -1,97 +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.coordinator.trigger;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.service.IMetadataServiceClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Poll policy change and notify listeners
- */
-public class DynamicPolicyLoader implements Runnable{
- private static Logger LOG = LoggerFactory.getLogger(DynamicPolicyLoader.class);
-
- private IMetadataServiceClient client;
- // initial cachedPolicies should be empty
- private Map<String, PolicyDefinition> cachedPolicies = new HashMap<>();
- private List<PolicyChangeListener> listeners = new ArrayList<>();
-
- public DynamicPolicyLoader(IMetadataServiceClient client){
- this.client = client;
- }
-
- public synchronized void addPolicyChangeListener(PolicyChangeListener listener){
- listeners.add(listener);
- }
-
- /**
- * When it is run at the first time, due to cachedPolicies being empty, all existing policies are expected
- * to be addedPolicies
- */
- @SuppressWarnings("unchecked")
- @Override
- public void run() {
- // we should catch every exception to avoid zombile thread
- try {
- List<PolicyDefinition> current = client.listPolicies();
- Map<String, PolicyDefinition> currPolicies = new HashMap<>();
- current.forEach(pe -> currPolicies.put(pe.getName(), pe));
-
- Collection<String> addedPolicies = CollectionUtils.subtract(currPolicies.keySet(), cachedPolicies.keySet());
- Collection<String> removedPolicies = CollectionUtils.subtract(cachedPolicies.keySet(), currPolicies.keySet());
- Collection<String> potentiallyModifiedPolicies = CollectionUtils.intersection(currPolicies.keySet(), cachedPolicies.keySet());
-
- List<String> reallyModifiedPolicies = new ArrayList<>();
- for (String updatedPolicy : potentiallyModifiedPolicies) {
- if (!currPolicies.get(updatedPolicy).equals(cachedPolicies.get(updatedPolicy))) {
- reallyModifiedPolicies.add(updatedPolicy);
- }
- }
-
- boolean policyChanged = false;
- if (addedPolicies.size() != 0 ||
- removedPolicies.size() != 0 ||
- reallyModifiedPolicies.size() != 0) {
- policyChanged = true;
- }
-
- if (!policyChanged) {
- LOG.info("policy is not changed since last run");
- return;
- }
- synchronized (this) {
- for (PolicyChangeListener listener : listeners) {
- listener.onPolicyChange(current, addedPolicies, removedPolicies, reallyModifiedPolicies);
- }
- }
-
- // reset cached policies
- cachedPolicies = currPolicies;
- } catch (Throwable t) {
- LOG.error("error loading policy, but continue to run", t);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/PolicyChangeListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/PolicyChangeListener.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/PolicyChangeListener.java
deleted file mode 100644
index cd5265d..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/java/org/apache/eagle/alert/coordinator/trigger/PolicyChangeListener.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
- *
- * 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
- * limit
- */
-package org.apache.eagle.alert.coordinator.trigger;
-
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-
-public interface PolicyChangeListener {
- void onPolicyChange(List<PolicyDefinition> allPolicies, Collection<String> addedPolicies, Collection<String> removedPolicies, Collection<String> modifiedPolicies);
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/application.conf
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/application.conf b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/application.conf
deleted file mode 100644
index 23ee161..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/application.conf
+++ /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
-#
-# 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.
-{
- "coordinator" : {
- "policiesPerBolt" : 5,
- "boltParallelism" : 5,
- "policyDefaultParallelism" : 5,
- "boltLoadUpbound": 0.8,
- "topologyLoadUpbound" : 0.8,
- "numOfAlertBoltsPerTopology" : 5,
- "zkConfig" : {
- "zkQuorum" : "localhost:2181",
- "zkRoot" : "/alert",
- "zkSessionTimeoutMs" : 10000,
- "connectionTimeoutMs" : 10000,
- "zkRetryTimes" : 3,
- "zkRetryInterval" : 3000
- },
- "metadataService" : {
- "host" : "localhost",
- "port" : 8080,
- "context" : "/api"
- },
- "metadataDynamicCheck" : {
- "initDelayMillis" : 1000,
- "delayMillis" : 30000
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/log4j.properties b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/log4j.properties
deleted file mode 100644
index d4bc126..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/resources/log4j.properties
+++ /dev/null
@@ -1,21 +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.
-
-log4j.rootLogger=INFO, stdout
-
-# standard output
-log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %p [%t] %c{2}[%L]: %m%n
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/WEB-INF/web.xml
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/WEB-INF/web.xml b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/WEB-INF/web.xml
deleted file mode 100644
index 1aa925e..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/WEB-INF/web.xml
+++ /dev/null
@@ -1,87 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-<web-app xmlns="http://java.sun.com/xml/ns/javaee"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
- http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
- version="3.0">
- <welcome-file-list>
- <welcome-file>index.html</welcome-file>
- </welcome-file-list>
- <servlet>
- <servlet-name>Jersey Web Application</servlet-name>
- <servlet-class>com.sun.jersey.spi.container.servlet.ServletContainer</servlet-class>
- <init-param>
- <param-name>com.sun.jersey.config.property.packages</param-name>
- <param-value>io.swagger.jaxrs.json,io.swagger.jaxrs.listing,org.apache.eagle.alert.coordinator.resource,org.codehaus.jackson.jaxrs</param-value>
- </init-param>
- <init-param>
- <param-name>com.sun.jersey.spi.container.ContainerRequestFilters</param-name>
- <param-value>com.sun.jersey.api.container.filter.GZIPContentEncodingFilter;com.sun.jersey.api.container.filter.PostReplaceFilter</param-value>
- </init-param>
- <init-param>
- <param-name>com.sun.jersey.spi.container.ContainerResponseFilters</param-name>
- <param-value>com.sun.jersey.api.container.filter.GZIPContentEncodingFilter</param-value>
- </init-param>
- <load-on-startup>1</load-on-startup>
- </servlet>
- <!-- Servlet for swagger initialization only, no URL mapping. -->
- <servlet>
- <servlet-name>swaggerConfig</servlet-name>
- <servlet-class>io.swagger.jaxrs.config.DefaultJaxrsConfig</servlet-class>
- <init-param>
- <param-name>api.version</param-name>
- <param-value>1.0.0</param-value>
- </init-param>
- <init-param>
- <param-name>swagger.api.basepath</param-name>
- <param-value>/api</param-value>
- </init-param>
- <load-on-startup>2</load-on-startup>
- </servlet>
-
- <servlet-mapping>
- <servlet-name>Jersey Web Application</servlet-name>
- <url-pattern>/api/*</url-pattern>
- </servlet-mapping>
- <filter>
- <filter-name>CorsFilter</filter-name>
- <!-- this should be replaced by tomcat ones, see also metadata resource -->
- <filter-class>org.apache.eagle.alert.resource.SimpleCORSFiler</filter-class>
- <init-param>
- <param-name>cors.allowed.origins</param-name>
- <param-value>*</param-value>
- </init-param>
- <init-param>
- <param-name>cors.allowed.headers</param-name>
- <param-value>Authorization,Origin, No-Cache, X-Requested-With, If-Modified-Since, Pragma, Last-Modified, Cache-Control, Expires, Content-Type, X-E4M-With, Accept</param-value>
- </init-param>
- <init-param>
- <param-name>cors.allowed.methods</param-name>
- <param-value>GET,POST,HEAD,OPTIONS,PUT,DELETE</param-value>
- </init-param>
- <init-param>
- <param-name>cors.support.credentials</param-name>
- <param-value>true</param-value>
- </init-param>
- </filter>
- <filter-mapping>
- <filter-name>CorsFilter</filter-name>
- <url-pattern>/*</url-pattern>
- </filter-mapping>
-</web-app>
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/index.html
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/index.html b/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/index.html
deleted file mode 100644
index 1c4ea76..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/main/webapp/index.html
+++ /dev/null
@@ -1,18 +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.
- -->
-UMP Coordinator service!
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/CoordinatorTest.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/CoordinatorTest.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/CoordinatorTest.java
deleted file mode 100644
index 78b72b3..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/CoordinatorTest.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
- *
- * 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.alert.coordinator;
-
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.eagle.alert.config.ConfigBusConsumer;
-import org.apache.eagle.alert.config.ConfigBusProducer;
-import org.apache.eagle.alert.config.ConfigChangeCallback;
-import org.apache.eagle.alert.config.ConfigValue;
-import org.apache.eagle.alert.config.ZKConfig;
-import org.apache.eagle.alert.config.ZKConfigBuilder;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordinator.Coordinator;
-import org.apache.eagle.alert.coordinator.ScheduleOption;
-import org.apache.eagle.alert.service.IMetadataServiceClient;
-import org.apache.eagle.alert.service.MetadataServiceClientImpl;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-
-/**
- * @since May 5, 2016
- *
- */
-public class CoordinatorTest {
-
- @SuppressWarnings({ "resource", "unused" })
- @Ignore
- @Test
- public void test() throws Exception {
- before();
- Config config = ConfigFactory.load().getConfig("coordinator");
- ZKConfig zkConfig = ZKConfigBuilder.getZKConfig(config);
- ConfigBusProducer producer = new ConfigBusProducer(zkConfig);
- IMetadataServiceClient client = new MetadataServiceClientImpl(config);
-
- Coordinator coordinator = new Coordinator(config, producer, client);
- ScheduleOption option = new ScheduleOption();
- ScheduleState state = coordinator.schedule(option);
- String v = state.getVersion();
-
- AtomicBoolean validated = new AtomicBoolean(false);
- ConfigBusConsumer consumer = new ConfigBusConsumer(zkConfig, "topo1/spout", new ConfigChangeCallback() {
- @Override
- public void onNewConfig(ConfigValue value) {
- String vId = value.getValue().toString();
- Assert.assertEquals(v, vId);
- validated.set(true);
- }
- });
-
- Thread.sleep(1000);
- Assert.assertTrue(validated.get());
- }
-
- @SuppressWarnings({ "resource", "unused" })
- @Test
- public void test_01() throws Exception {
- before();
- Config config = ConfigFactory.load().getConfig("coordinator");
- ZKConfig zkConfig = ZKConfigBuilder.getZKConfig(config);
- ConfigBusProducer producer = new ConfigBusProducer(zkConfig);
- IMetadataServiceClient client = ScheduleContextBuilderTest.getSampleMetadataService();
-
- Coordinator coordinator = new Coordinator(config, producer, client);
- ScheduleOption option = new ScheduleOption();
- ScheduleState state = coordinator.schedule(option);
- String v = state.getVersion();
-
- // TODO : assert version
-
- CountDownLatch latch = new CountDownLatch(1);
- AtomicBoolean validated = new AtomicBoolean(false);
- ConfigBusConsumer consumer = new ConfigBusConsumer(zkConfig, "topo1/spout", new ConfigChangeCallback() {
- @Override
- public void onNewConfig(ConfigValue value) {
- String vId = value.getValue().toString();
- Assert.assertEquals(v, vId);
- validated.set(true);
- latch.countDown();
- }
- });
-
- latch.await(3, TimeUnit.SECONDS);
- Assert.assertTrue(validated.get());
- }
-
- @Ignore
- @Test
- public void test_main() throws Exception {
- before();
-
- Coordinator.main(null);
- }
-
- @Before
- public void before() {
- System.setProperty("config.resource", "/test-application.conf");
- ConfigFactory.invalidateCaches();
- ConfigFactory.load().getConfig("coordinator");
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/DynamicPolicyLoaderTest.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/DynamicPolicyLoaderTest.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/DynamicPolicyLoaderTest.java
deleted file mode 100644
index 155a9e5..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/DynamicPolicyLoaderTest.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
- * *
- * * 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.alert.coordinator;
-
-/**
- * Since 4/28/16.
- */
-public class DynamicPolicyLoaderTest {
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/MetadataServiceClientImplTest.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/MetadataServiceClientImplTest.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/MetadataServiceClientImplTest.java
deleted file mode 100644
index e2ea031..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/MetadataServiceClientImplTest.java
+++ /dev/null
@@ -1,57 +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.alert.coordinator;
-
-import org.apache.eagle.alert.config.ConfigBusProducer;
-import org.apache.eagle.alert.config.ZKConfig;
-import org.apache.eagle.alert.config.ZKConfigBuilder;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordinator.Coordinator;
-import org.apache.eagle.alert.service.MetadataServiceClientImpl;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-
-/**
- * @since May 9, 2016
- *
- */
-public class MetadataServiceClientImplTest {
-
- @Ignore
- @Test
- public void addScheduleState() throws Exception {
- ConfigFactory.invalidateCaches();
- System.setProperty("config.resource", "/test-application.conf");
- Config config = ConfigFactory.load("test-application.conf").getConfig("coordinator");
- MetadataServiceClientImpl client = new MetadataServiceClientImpl(config);
-
- ScheduleState ss = new ScheduleState();
- ss.setVersion("spec_version_1463764252582");
-
- client.addScheduleState(ss);
-
- client.close();
-
- ss.setVersion("spec_version_1464764252582");
- ZKConfig zkConfig = ZKConfigBuilder.getZKConfig(config);
- ConfigBusProducer producer = new ConfigBusProducer(zkConfig);
- Coordinator.postSchedule(client, ss, producer);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/ScheduleContextBuilderTest.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/ScheduleContextBuilderTest.java b/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/ScheduleContextBuilderTest.java
deleted file mode 100644
index f2e67de..0000000
--- a/eagle-core/eagle-alert/alert/alert-coordinator/src/test/java/org/apache/alert/coordinator/ScheduleContextBuilderTest.java
+++ /dev/null
@@ -1,281 +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.alert.coordinator;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.alert.coordinator.mock.InMemMetadataServiceClient;
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.ScheduleState;
-import org.apache.eagle.alert.coordination.model.Tuple2StreamMetadata;
-import org.apache.eagle.alert.coordination.model.WorkSlot;
-import org.apache.eagle.alert.coordination.model.internal.MonitoredStream;
-import org.apache.eagle.alert.coordination.model.internal.PolicyAssignment;
-import org.apache.eagle.alert.coordination.model.internal.StreamGroup;
-import org.apache.eagle.alert.coordination.model.internal.StreamWorkSlotQueue;
-import org.apache.eagle.alert.coordination.model.internal.Topology;
-import org.apache.eagle.alert.coordinator.IScheduleContext;
-import org.apache.eagle.alert.coordinator.model.AlertBoltUsage;
-import org.apache.eagle.alert.coordinator.model.TopologyUsage;
-import org.apache.eagle.alert.coordinator.provider.ScheduleContextBuilder;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.Publishment;
-import org.apache.eagle.alert.engine.coordinator.StreamColumn;
-import org.apache.eagle.alert.engine.coordinator.StreamColumn.Type;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamPartition;
-import org.apache.eagle.alert.engine.coordinator.StreamSortSpec;
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * @since May 5, 2016
- *
- */
-public class ScheduleContextBuilderTest {
-
- @Test
- public void test() {
- InMemMetadataServiceClient client = getSampleMetadataService();
-
- ScheduleContextBuilder builder = new ScheduleContextBuilder(client);
-
- IScheduleContext context = builder.buildContext();
-
- // assert topology usage
- Map<String, TopologyUsage> usages = context.getTopologyUsages();
- Assert.assertEquals(1, usages.get(TOPO1).getMonitoredStream().size());
- Assert.assertTrue(usages.get(TOPO1).getPolicies().contains(TEST_POLICY_1));
-
- String alertBolt0 = TOPO1 + "-alert-" + "0";
- String alertBolt1 = TOPO1 + "-alert-" + "1";
- String alertBolt2 = TOPO1 + "-alert-" + "2";
- for (AlertBoltUsage u : usages.get(TOPO1).getAlertUsages().values()) {
- if (u.getBoltId().equals(alertBolt0) || u.getBoltId().equals(alertBolt1)
- || u.getBoltId().equals(alertBolt2)) {
- Assert.assertEquals(1, u.getPolicies().size());
- Assert.assertTrue(u.getPolicies().contains(TEST_POLICY_1));
- Assert.assertEquals(1, u.getPartitions().size());
- Assert.assertEquals(1, u.getReferQueues().size());
- }
- }
- }
-
- @Test
- public void test_remove_policy() {
- InMemMetadataServiceClient client = getSampleMetadataService();
- ScheduleContextBuilder builder = new ScheduleContextBuilder(client);
-
- PolicyAssignment assignment1 = client.getVersionedSpec().getAssignments().get(0);
-
- IScheduleContext context = builder.buildContext();
- Assert.assertTrue(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
- StreamWorkSlotQueue queue = SchedulerTest.getQueue(context, assignment1.getQueueId()).getRight();
-
- client.listPolicies().remove(0);
- context = builder.buildContext();
- Assert.assertFalse(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
-
- WorkSlot slot = queue.getWorkingSlots().get(0);
- Set<String> topoPolicies = context.getTopologyUsages().get(slot.topologyName).getPolicies();
- Assert.assertFalse(topoPolicies.contains(TEST_DATASOURCE_1));
- Assert.assertEquals(0, topoPolicies.size());
- }
-
- @Test
- public void test_changed_policy() {
- InMemMetadataServiceClient client = getSampleMetadataService();
- ScheduleContextBuilder builder = new ScheduleContextBuilder(client);
- PolicyAssignment assignment1 = client.getVersionedSpec().getAssignments().get(0);
-
- IScheduleContext context = builder.buildContext();
- Assert.assertTrue(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
-
- StreamWorkSlotQueue queue = SchedulerTest.getQueue(context, assignment1.getQueueId()).getRight();
-
- PolicyDefinition pd1 = client.listPolicies().get(0);
- // add a new group by column : need to replace the partiton spec, to
- // avoid reference same object in
- // on jvm (no serialization and deserialization)
- StreamPartition par = new StreamPartition(pd1.getPartitionSpec().get(0));
- par.getColumns().add("s1");
- pd1.getPartitionSpec().clear();
- pd1.getPartitionSpec().add(par);
-
- context = builder.buildContext();
-
- // assert the policy assignment is removed
- Assert.assertFalse(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
- // assert the monitored stream is removed as no policy on it now.
- Assert.assertEquals(0, context.getMonitoredStreams().size());
- // assert the topology usage doesn't contain policy
- WorkSlot slot = queue.getWorkingSlots().get(0);
- TopologyUsage topologyUsage = context.getTopologyUsages().get(slot.topologyName);
- Set<String> topoPolicies = topologyUsage.getPolicies();
- Assert.assertFalse(topoPolicies.contains(TEST_DATASOURCE_1));
- Assert.assertEquals(0, topoPolicies.size());
- // assert the topology usage doesn't contain the monitored stream
- Assert.assertEquals(0, topologyUsage.getMonitoredStream().size());
- // assert the alert bolt usage doesn't have the queue reference
- Assert.assertEquals(0, topologyUsage.getAlertBoltUsage(slot.getBoltId()).getReferQueues().size());
- }
-
- @Test
- public void test_renamed_topologies() {
- InMemMetadataServiceClient client = getSampleMetadataService();
- ScheduleContextBuilder builder = new ScheduleContextBuilder(client);
-
- IScheduleContext context = builder.buildContext();
- Assert.assertTrue(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
-
- Topology t = client.listTopologies().get(0);
- t.setName("newName");
-
- context = builder.buildContext();
- Assert.assertFalse(context.getPolicyAssignments().containsKey(TEST_POLICY_1));
- }
-
- private static final String TOPO1 = "topo1";
- private static final String V1 = "v1";
- private static final String COL1 = "col1";
- private static final String OUT_STREAM1 = "out-stream1";
- private static final String TEST_POLICY_1 = "test-policy-1";
- private static final String TEST_STREAM_DEF_1 = "testStreamDef";
- private static final String TEST_DATASOURCE_1 = "test-datasource-1";
- private static StreamPartition par;
- private static String queueId;
- private static StreamGroup streamGroup;
-
- public static InMemMetadataServiceClient getSampleMetadataService() {
- InMemMetadataServiceClient client = new InMemMetadataServiceClient();
- client.listTopologies().add(createSampleTopology());
- client.listDataSources().add(createKafka2TupleMetadata());
- // client.listSpoutMetadata().add(createS)
- client.listPolicies().add(createPolicy());
- client.listPublishment().add(createPublishment());
- client.listStreams().add(createStreamDefinition());
- client.addScheduleState(createScheduleState());
- return client;
- }
-
- private static ScheduleState createScheduleState() {
- ScheduleState ss = new ScheduleState();
- ss.setVersion(V1);
-
- ss.getMonitoredStreams().add(createMonitoredStream());
- ss.getAssignments().add(createAssignment());
-
- return ss;
- }
-
- private static MonitoredStream createMonitoredStream() {
- MonitoredStream ms = new MonitoredStream(streamGroup);
- ms.setVersion(V1);
-
- List<WorkSlot> slots = new ArrayList<WorkSlot>();
- WorkSlot slot0 = new WorkSlot(TOPO1, TOPO1 + "-alert-" + 0);
- WorkSlot slot1 = new WorkSlot(TOPO1, TOPO1 + "-alert-" + 1);
- WorkSlot slot2 = new WorkSlot(TOPO1, TOPO1 + "-alert-" + 2);
- slots.add(slot0);
- slots.add(slot1);
- slots.add(slot2);
-
- StreamWorkSlotQueue q = new StreamWorkSlotQueue(streamGroup, false, new HashMap<>(), slots);
- ms.addQueues(q);
- queueId = q.getQueueId();
- return ms;
- }
-
- private static PolicyAssignment createAssignment() {
- PolicyAssignment pa = new PolicyAssignment(TEST_POLICY_1, queueId);
- return pa;
- }
-
- private static PolicyDefinition createPolicy() {
- PolicyDefinition def = new PolicyDefinition();
- def.setName(TEST_POLICY_1);
- def.setInputStreams(Arrays.asList(TEST_STREAM_DEF_1));
- def.setOutputStreams(Arrays.asList(OUT_STREAM1));
- def.setParallelismHint(5);
-
- streamGroup = new StreamGroup();
- par = new StreamPartition();
- par.setStreamId(TEST_STREAM_DEF_1);
- par.getColumns().add(COL1);
- StreamSortSpec sortSpec = new StreamSortSpec();
-// sortSpec.setColumn("col1");
- sortSpec.setWindowMargin(3);
- sortSpec.setWindowPeriod("PT1M");
-
- par.setSortSpec(sortSpec);
- streamGroup.addStreamPartition(par);
-
- List<StreamPartition> lists = new ArrayList<StreamPartition>();
- lists.add(par);
- def.setPartitionSpec(lists);
- return def;
- }
-
- private static StreamDefinition createStreamDefinition() {
- StreamDefinition def = new StreamDefinition();
- def.setStreamId(TEST_STREAM_DEF_1);
- def.setDataSource(TEST_DATASOURCE_1);
-
- StreamColumn col = new StreamColumn();
- col.setName(COL1);
- col.setRequired(true);
- col.setType(Type.STRING);
- def.getColumns().add(col);
-
- return def;
- }
-
- private static Publishment createPublishment() {
- Publishment pub = new Publishment();
- pub.setType("KAFKA");
- pub.setName("test-stream-output");
- pub.setPolicyIds(Arrays.asList(TEST_POLICY_1));
- return pub;
- }
-
- private static Kafka2TupleMetadata createKafka2TupleMetadata() {
- Kafka2TupleMetadata ktm = new Kafka2TupleMetadata();
- ktm.setName(TEST_DATASOURCE_1);
- ktm.setSchemeCls("SchemeClass");
- ktm.setTopic("tupleTopic");
- ktm.setType("KAFKA");
- ktm.setCodec(new Tuple2StreamMetadata());
- return ktm;
- }
-
- private static Topology createSampleTopology() {
- Topology t = new Topology(TOPO1, 3, 10);
- for (int i = 0; i < t.getNumOfGroupBolt(); i++) {
- t.getGroupNodeIds().add(t.getName() + "-grp-" + i);
- }
- for (int i = 0; i < t.getNumOfAlertBolt(); i++) {
- t.getAlertBoltIds().add(t.getName() + "-alert-" + i);
- }
- return t;
- }
-
-}