| /* |
| * 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.stratos.cep.extension; |
| |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.log4j.Logger; |
| import org.apache.stratos.messaging.broker.publish.EventPublisher; |
| import org.apache.stratos.messaging.broker.publish.EventPublisherPool; |
| import org.apache.stratos.messaging.domain.topology.*; |
| import org.apache.stratos.messaging.event.health.stat.MemberFaultEvent; |
| import org.apache.stratos.messaging.message.receiver.topology.TopologyManager; |
| import org.apache.stratos.messaging.util.MessagingUtil; |
| import org.wso2.siddhi.core.config.SiddhiContext; |
| import org.wso2.siddhi.core.event.StreamEvent; |
| import org.wso2.siddhi.core.event.in.InEvent; |
| import org.wso2.siddhi.core.event.in.InListEvent; |
| import org.wso2.siddhi.core.persistence.ThreadBarrier; |
| import org.wso2.siddhi.core.query.QueryPostProcessingElement; |
| import org.wso2.siddhi.core.query.processor.window.RunnableWindowProcessor; |
| import org.wso2.siddhi.core.query.processor.window.WindowProcessor; |
| import org.wso2.siddhi.core.util.collection.queue.scheduler.ISchedulerSiddhiQueue; |
| import org.wso2.siddhi.core.util.collection.queue.scheduler.SchedulerSiddhiQueue; |
| import org.wso2.siddhi.core.util.collection.queue.scheduler.SchedulerSiddhiQueueGrid; |
| import org.wso2.siddhi.query.api.definition.AbstractDefinition; |
| import org.wso2.siddhi.query.api.expression.Expression; |
| import org.wso2.siddhi.query.api.expression.Variable; |
| import org.wso2.siddhi.query.api.expression.constant.IntConstant; |
| import org.wso2.siddhi.query.api.expression.constant.LongConstant; |
| import org.wso2.siddhi.query.api.extension.annotation.SiddhiExtension; |
| |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| |
| /** |
| * CEP window processor to handle faulty member instances. This window processor is responsible for |
| * publishing MemberFault event if health stats are not received within a given time window. |
| */ |
| @SiddhiExtension(namespace = "stratos", function = "faultHandling") |
| public class FaultHandlingWindowProcessor extends WindowProcessor implements RunnableWindowProcessor { |
| private static final Logger log = Logger.getLogger(FaultHandlingWindowProcessor.class); |
| private static final String ACTIVATE_TIMEOUT_KEY = "cep.fault.handler.extension.activate.timeout"; |
| private static final int ACTIVATE_TIMEOUT = |
| Integer.getInteger(ACTIVATE_TIMEOUT_KEY, 60 * 1000 * 15); |
| private static final int TIME_OUT = 60 * 1000; |
| private ScheduledExecutorService faultHandleScheduler; |
| private ScheduledFuture<?> lastSchedule; |
| private ThreadBarrier threadBarrier; |
| private long timeToKeep; |
| private ISchedulerSiddhiQueue<StreamEvent> window; |
| private EventPublisher healthStatPublisher = EventPublisherPool |
| .getPublisher(MessagingUtil.Topics.HEALTH_STAT_TOPIC.getTopicName()); |
| private Map<String, Object> MemberFaultEventMap = new HashMap<String, Object>(); |
| private Map<String, Object> memberFaultEventMessageMap = new HashMap<String, Object>(); |
| |
| // Map of member id's to their last received health event time stamp |
| private ConcurrentHashMap<String, Long> memberTimeStampMap = new ConcurrentHashMap<String, Long>(); |
| private volatile boolean isActive; |
| private volatile boolean hasMemberTimeStampMapInitialized; |
| private long startTime = System.currentTimeMillis(); |
| |
| // Event receiver to receive topology events published by cloud-controller |
| private CEPTopologyEventReceiver cepTopologyEventReceiver = new CEPTopologyEventReceiver(this); |
| |
| // Stratos member id attribute index in stream execution plan |
| private int memberIdAttrIndex; |
| |
| @Override |
| protected void processEvent(InEvent event) { |
| addDataToMap(event); |
| } |
| |
| @Override |
| protected void processEvent(InListEvent listEvent) { |
| for (int i = 0, size = listEvent.getActiveEvents(); i < size; i++) { |
| addDataToMap((InEvent) listEvent.getEvent(i)); |
| } |
| } |
| |
| /** |
| * Add new entry to time stamp map from the received event. |
| * |
| * @param event Event received by Siddhi. |
| */ |
| private void addDataToMap(InEvent event) { |
| if (!isActive) { |
| log.info("Received first event. Marking fault handling window processor as active"); |
| isActive = true; |
| } |
| String id = (String) event.getData()[memberIdAttrIndex]; |
| //checking whether this member is the topology. |
| //sometimes there can be a delay between publishing member terminated events |
| //and actually terminating instances. Hence CEP might get events for already terminated members |
| //so we are checking the topology for the member existence |
| Member member = getMemberFromId(id); |
| if (null == member) { |
| log.warn(String.format("Member with [id] %s not found in the topology. Event rejected", id)); |
| return; |
| } |
| if (StringUtils.isNotEmpty(id)) { |
| memberTimeStampMap.put(id, event.getTimeStamp()); |
| } else { |
| log.warn("NULL member id found in the event received. Event rejected."); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug("Event received from [member-id] " + id + " [time-stamp] " + event.getTimeStamp()); |
| } |
| } |
| |
| @Override |
| public Iterator<StreamEvent> iterator() { |
| return window.iterator(); |
| } |
| |
| @Override |
| public Iterator<StreamEvent> iterator(String predicate) { |
| if (siddhiContext.isDistributedProcessingEnabled()) { |
| return ((SchedulerSiddhiQueueGrid<StreamEvent>) window).iterator(predicate); |
| } else { |
| return window.iterator(); |
| } |
| } |
| |
| /** |
| * Retrieve the current activated members from the topology and initialize the timestamp map. |
| * This will allow the system to recover from a restart |
| * |
| * @param topology Topology model object |
| */ |
| boolean loadTimeStampMapFromTopology(Topology topology) { |
| long currentTimeStamp = System.currentTimeMillis(); |
| if (topology == null || topology.getServices() == null) { |
| return false; |
| } |
| // TODO make this efficient by adding APIs to messaging component |
| for (Service service : topology.getServices()) { |
| if (service.getClusters() != null) { |
| for (Cluster cluster : service.getClusters()) { |
| if (cluster.getMembers() != null) { |
| for (Member member : cluster.getMembers()) { |
| // we are checking faulty status only in previously activated members |
| if (member != null && MemberStatus.Active.equals(member.getStatus())) { |
| // Initialize the member time stamp map from the topology at the beginning |
| memberTimeStampMap.putIfAbsent(member.getMemberId(), currentTimeStamp); |
| } |
| } |
| } |
| } |
| } |
| } |
| hasMemberTimeStampMapInitialized = true; |
| if (log.isInfoEnabled()) { |
| log.info("Member timestamps were successfully loaded from the topology: [timestamps] " + |
| Arrays.toString(memberTimeStampMap.entrySet().toArray())); |
| } |
| return true; |
| } |
| |
| private Member getMemberFromId(String memberId) { |
| if (StringUtils.isEmpty(memberId)) { |
| return null; |
| } |
| if (TopologyManager.isInitialized()) { |
| try { |
| TopologyManager.acquireReadLock(); |
| if (TopologyManager.getTopology().getServices() == null) { |
| return null; |
| } |
| // TODO make this efficient by adding APIs to messaging component |
| for (Service service : TopologyManager.getTopology().getServices()) { |
| if (service.getClusters() != null) { |
| for (Cluster cluster : service.getClusters()) { |
| if (cluster.getMembers() != null) { |
| for (Member member : cluster.getMembers()) { |
| if (memberId.equals(member.getMemberId())) { |
| return member; |
| } |
| } |
| } |
| } |
| } |
| } |
| } catch (Exception e) { |
| log.error("Error while reading topology" + e); |
| } finally { |
| TopologyManager.releaseReadLock(); |
| } |
| } |
| return null; |
| } |
| |
| private void publishMemberFault(Member member) { |
| if (member == null) { |
| log.warn("Failed to publish member fault event. Member object is null"); |
| return; |
| } |
| log.info("Publishing member fault event for [member-id] " + member.getMemberId()); |
| |
| MemberFaultEvent memberFaultEvent = new MemberFaultEvent(member.getClusterId(), member.getClusterInstanceId(), |
| member.getMemberId(), member.getPartitionId(), member.getNetworkPartitionId(), 0); |
| |
| memberFaultEventMessageMap.put("message", memberFaultEvent); |
| healthStatPublisher.publish(MemberFaultEventMap, true); |
| } |
| |
| @Override |
| public void run() { |
| try { |
| // wait until the first event OR given timeout to expire in order to activate this window processor |
| // this is to prevent false positives at the CEP startup |
| if (!isActive && System.currentTimeMillis() - startTime > ACTIVATE_TIMEOUT) { |
| log.info("Activation wait timeout has expired. Marking fault handling window processor as active"); |
| isActive = true; |
| } |
| // do not process events until memberTimeStampMap is initialized and window processor is activated |
| // memberTimeStampMap will be initialized only after receiving the complete topology event |
| if (!(isActive && hasMemberTimeStampMapInitialized)) { |
| return; |
| } |
| threadBarrier.pass(); |
| |
| for (Object o : memberTimeStampMap.entrySet()) { |
| Map.Entry pair = (Map.Entry) o; |
| long currentTime = System.currentTimeMillis(); |
| Long eventTimeStamp = (Long) pair.getValue(); |
| |
| if ((currentTime - eventTimeStamp) > TIME_OUT) { |
| String memberId = (String) pair.getKey(); |
| Member member = getMemberFromId(memberId); |
| if (member != null) { |
| log.info("Faulty member detected [member-id] " + pair.getKey() + " with [last time-stamp] " + |
| eventTimeStamp + " [time-out] " + TIME_OUT + " milliseconds"); |
| publishMemberFault(member); |
| } |
| memberTimeStampMap.remove(memberId); |
| } |
| } |
| if (log.isDebugEnabled()) { |
| log.debug("Fault handling processor iteration completed with [time-stamp map length] " + |
| memberTimeStampMap.size() + " [time-stamp map] " + memberTimeStampMap); |
| } |
| } catch (Throwable t) { |
| log.error(t.getMessage(), t); |
| } finally { |
| if (lastSchedule != null) { |
| lastSchedule.cancel(false); |
| } |
| lastSchedule = faultHandleScheduler.schedule(this, timeToKeep, TimeUnit.MILLISECONDS); |
| } |
| } |
| |
| @Override |
| protected Object[] currentState() { |
| return new Object[]{window.currentState()}; |
| } |
| |
| @Override |
| protected void restoreState(Object[] data) { |
| window.restoreState(data); |
| window.restoreState((Object[]) data[0]); |
| window.reSchedule(); |
| } |
| |
| @Override |
| protected void init(Expression[] parameters, QueryPostProcessingElement nextProcessor, |
| AbstractDefinition streamDefinition, String elementId, boolean async, SiddhiContext |
| siddhiContext) { |
| if (parameters[0] instanceof IntConstant) { |
| timeToKeep = ((IntConstant) parameters[0]).getValue(); |
| } else { |
| timeToKeep = ((LongConstant) parameters[0]).getValue(); |
| } |
| |
| String memberIdAttrName = ((Variable) parameters[1]).getAttributeName(); |
| memberIdAttrIndex = streamDefinition.getAttributePosition(memberIdAttrName); |
| |
| if (this.siddhiContext.isDistributedProcessingEnabled()) { |
| window = new SchedulerSiddhiQueueGrid<StreamEvent>(elementId, this, this.siddhiContext, this.async); |
| } else { |
| window = new SchedulerSiddhiQueue<StreamEvent>(this); |
| } |
| MemberFaultEventMap |
| .put("org.apache.stratos.messaging.event.health.stat.MemberFaultEvent", memberFaultEventMessageMap); |
| |
| //Ordinary scheduling |
| window.schedule(); |
| if (log.isInfoEnabled()) { |
| log.info(String.format("Fault handling window processor initialized with [timeToKeep] %s, " + |
| "[memberIdAttrName] %s, [memberIdAttrIndex] %s, [distributed-enabled] %s, " + |
| "[activate-timeout] %d", timeToKeep, memberIdAttrName, memberIdAttrIndex, |
| siddhiContext.isDistributedProcessingEnabled(), ACTIVATE_TIMEOUT)); |
| } |
| } |
| |
| @Override |
| public void schedule() { |
| if (lastSchedule != null) { |
| lastSchedule.cancel(false); |
| } |
| lastSchedule = faultHandleScheduler.schedule(this, timeToKeep, TimeUnit.MILLISECONDS); |
| } |
| |
| @Override |
| public void scheduleNow() { |
| if (lastSchedule != null) { |
| lastSchedule.cancel(false); |
| } |
| lastSchedule = faultHandleScheduler.schedule(this, 0, TimeUnit.MILLISECONDS); |
| } |
| |
| @Override |
| public void setScheduledExecutorService(ScheduledExecutorService scheduledExecutorService) { |
| this.faultHandleScheduler = scheduledExecutorService; |
| } |
| |
| @Override |
| public void setThreadBarrier(ThreadBarrier threadBarrier) { |
| this.threadBarrier = threadBarrier; |
| } |
| |
| @Override |
| public void destroy() { |
| // terminate topology listener thread |
| cepTopologyEventReceiver.destroy(); |
| window = null; |
| } |
| |
| ConcurrentHashMap<String, Long> getMemberTimeStampMap() { |
| return memberTimeStampMap; |
| } |
| } |
