| /* |
| * 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.autoscaler.event.receiver.health; |
| |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.stratos.autoscaler.context.AutoscalerContext; |
| import org.apache.stratos.autoscaler.monitor.cluster.ClusterMonitor; |
| import org.apache.stratos.messaging.domain.topology.Cluster; |
| import org.apache.stratos.messaging.domain.topology.Member; |
| import org.apache.stratos.messaging.domain.topology.Service; |
| import org.apache.stratos.messaging.event.health.stat.*; |
| import org.apache.stratos.messaging.listener.health.stat.*; |
| import org.apache.stratos.messaging.message.receiver.health.stat.HealthStatEventReceiver; |
| import org.apache.stratos.messaging.message.receiver.topology.TopologyManager; |
| |
| import java.util.concurrent.ExecutorService; |
| |
| /** |
| * A thread for processing topology messages and updating the topology data structure. |
| */ |
| public class AutoscalerHealthStatEventReceiver { |
| |
| private static final Log log = LogFactory.getLog(AutoscalerHealthStatEventReceiver.class); |
| private boolean terminated = false; |
| |
| private HealthStatEventReceiver healthStatEventReceiver; |
| private ExecutorService executorService; |
| |
| public AutoscalerHealthStatEventReceiver() { |
| this.healthStatEventReceiver = HealthStatEventReceiver.getInstance(); |
| addEventListeners(); |
| } |
| |
| // public void execute() { |
| // healthStatEventReceiver.setExecutorService(executorService); |
| // healthStatEventReceiver.execute(); |
| // |
| // if (log.isInfoEnabled()) { |
| // log.info("Autoscaler health stat event receiver thread started"); |
| // } |
| // } |
| |
| private void addEventListeners() { |
| // Listen to health stat events that affect clusters |
| healthStatEventReceiver.addEventListener(new AverageLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| AverageLoadAverageEvent averageLoadAverageEvent = (AverageLoadAverageEvent) event; |
| String clusterId = averageLoadAverageEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleAverageLoadAverageEvent(averageLoadAverageEvent); |
| } |
| |
| }); |
| |
| healthStatEventReceiver.addEventListener(new AverageMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| AverageMemoryConsumptionEvent averageMemoryConsumptionEvent = (AverageMemoryConsumptionEvent) event; |
| String clusterId = averageMemoryConsumptionEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleAverageMemoryConsumptionEvent(averageMemoryConsumptionEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new AverageRequestsInFlightEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| AverageRequestsInFlightEvent averageRequestsInFlightEvent = (AverageRequestsInFlightEvent) event; |
| String clusterId = averageRequestsInFlightEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleAverageRequestsInFlightEvent(averageRequestsInFlightEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new AverageRequestsServingCapabilityEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| AverageRequestsServingCapabilityEvent averageRequestsServingCapabilityEvent = (AverageRequestsServingCapabilityEvent) event; |
| String clusterId = averageRequestsServingCapabilityEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| if (monitor instanceof ClusterMonitor) { |
| ClusterMonitor vmClusterMonitor = (ClusterMonitor) monitor; |
| vmClusterMonitor.handleAverageRequestsServingCapabilityEvent(averageRequestsServingCapabilityEvent); |
| } |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new GradientOfLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| GradientOfLoadAverageEvent gradientOfLoadAverageEvent = (GradientOfLoadAverageEvent) event; |
| String clusterId = gradientOfLoadAverageEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleGradientOfLoadAverageEvent(gradientOfLoadAverageEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new GradientOfMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| GradientOfMemoryConsumptionEvent gradientOfMemoryConsumptionEvent = (GradientOfMemoryConsumptionEvent) event; |
| String clusterId = gradientOfMemoryConsumptionEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleGradientOfMemoryConsumptionEvent(gradientOfMemoryConsumptionEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new GradientOfRequestsInFlightEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| GradientOfRequestsInFlightEvent gradientOfRequestsInFlightEvent = (GradientOfRequestsInFlightEvent) event; |
| String clusterId = gradientOfRequestsInFlightEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleGradientOfRequestsInFlightEvent(gradientOfRequestsInFlightEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberAverageLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberAverageLoadAverageEvent memberAverageLoadAverageEvent = (MemberAverageLoadAverageEvent) event; |
| String memberId = memberAverageLoadAverageEvent.getMemberId(); |
| Member member = getMemberByMemberId(memberId); |
| if (null == member) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member not found in the Topology: [member] %s", memberId)); |
| } |
| return; |
| } |
| if (!member.isActive()) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member activated event has not received for the member %s. " |
| + "Therefore ignoring" + " the health stat", memberId)); |
| } |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| String clusterId = member.getClusterId(); |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberAverageLoadAverageEvent(memberAverageLoadAverageEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberAverageMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberAverageMemoryConsumptionEvent memberAverageMemoryConsumptionEvent = (MemberAverageMemoryConsumptionEvent) event; |
| String memberId = memberAverageMemoryConsumptionEvent.getMemberId(); |
| Member member = getMemberByMemberId(memberId); |
| if (null == member) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member not found in the Topology: [member] %s", memberId)); |
| } |
| return; |
| } |
| if (!member.isActive()) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member activated event has not received for the member %s. " |
| + "Therefore ignoring" + " the health stat", memberId)); |
| } |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| String clusterId = member.getClusterId(); |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberAverageMemoryConsumptionEvent(memberAverageMemoryConsumptionEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberFaultEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberFaultEvent memberFaultEvent = (MemberFaultEvent) event; |
| String clusterId = memberFaultEvent.getClusterId(); |
| String memberId = memberFaultEvent.getMemberId(); |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member fault event received: [member] %s ", memberId)); |
| } |
| if (memberId == null || memberId.isEmpty()) { |
| log.error("Member id not found in received message"); |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberFaultEvent(memberFaultEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberGradientOfLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberGradientOfLoadAverageEvent memberGradientOfLoadAverageEvent = (MemberGradientOfLoadAverageEvent) event; |
| String memberId = memberGradientOfLoadAverageEvent.getMemberId(); |
| Member member = getMemberByMemberId(memberId); |
| if (null == member) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member not found in the Topology: [member] %s", memberId)); |
| } |
| return; |
| } |
| if (!member.isActive()) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member activated event has not received for the member %s. " |
| + "Therefore ignoring" + " the health stat", memberId)); |
| } |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| String clusterId = member.getClusterId(); |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberGradientOfLoadAverageEvent(memberGradientOfLoadAverageEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberGradientOfMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberGradientOfMemoryConsumptionEvent memberGradientOfMemoryConsumptionEvent |
| = (MemberGradientOfMemoryConsumptionEvent) event; |
| String memberId = memberGradientOfMemoryConsumptionEvent.getMemberId(); |
| Member member = getMemberByMemberId(memberId); |
| if (null == member) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member not found in the Topology: [member] %s", memberId)); |
| } |
| return; |
| } |
| if (!member.isActive()) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member activated event has not received for the member %s. " |
| + "Therefore ignoring" + " the health stat", memberId)); |
| } |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| String clusterId = member.getClusterId(); |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberGradientOfMemoryConsumptionEvent(memberGradientOfMemoryConsumptionEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberSecondDerivativeOfLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| MemberSecondDerivativeOfLoadAverageEvent memberSecondDerivativeOfLoadAverageEvent = (MemberSecondDerivativeOfLoadAverageEvent) event; |
| String memberId = memberSecondDerivativeOfLoadAverageEvent.getMemberId(); |
| Member member = getMemberByMemberId(memberId); |
| if (null == member) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member not found in the Topology: [member] %s", memberId)); |
| } |
| return; |
| } |
| if (!member.isActive()) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("Member activated event has not received for the member %s. " |
| + "Therefore ignoring" + " the health stat", memberId)); |
| } |
| return; |
| } |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| String clusterId = member.getClusterId(); |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleMemberSecondDerivativeOfLoadAverageEvent(memberSecondDerivativeOfLoadAverageEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new MemberSecondDerivativeOfMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new SecondDerivativeOfLoadAverageEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| SecondDerivativeOfLoadAverageEvent secondDerivativeOfLoadAverageEvent = (SecondDerivativeOfLoadAverageEvent) event; |
| String clusterId = secondDerivativeOfLoadAverageEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleSecondDerivativeOfLoadAverageEvent(secondDerivativeOfLoadAverageEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new SecondDerivativeOfMemoryConsumptionEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| SecondDerivativeOfMemoryConsumptionEvent secondDerivativeOfMemoryConsumptionEvent = (SecondDerivativeOfMemoryConsumptionEvent) event; |
| String clusterId = secondDerivativeOfMemoryConsumptionEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleSecondDerivativeOfMemoryConsumptionEvent(secondDerivativeOfMemoryConsumptionEvent); |
| } |
| }); |
| |
| healthStatEventReceiver.addEventListener(new SecondDerivativeOfRequestsInFlightEventListener() { |
| @Override |
| protected void onEvent(org.apache.stratos.messaging.event.Event event) { |
| SecondDerivativeOfRequestsInFlightEvent secondDerivativeOfRequestsInFlightEvent |
| = (SecondDerivativeOfRequestsInFlightEvent) event; |
| String clusterId = secondDerivativeOfRequestsInFlightEvent.getClusterId(); |
| AutoscalerContext asCtx = AutoscalerContext.getInstance(); |
| ClusterMonitor monitor; |
| monitor = asCtx.getClusterMonitor(clusterId); |
| if (null == monitor) { |
| if (log.isDebugEnabled()) { |
| log.debug(String.format("A cluster monitor is not found in autoscaler context " |
| + "[cluster] %s", clusterId)); |
| } |
| return; |
| } |
| monitor.handleSecondDerivativeOfRequestsInFlightEvent(secondDerivativeOfRequestsInFlightEvent); |
| } |
| }); |
| } |
| |
| private Member getMemberByMemberId(String memberId) { |
| try { |
| TopologyManager.acquireReadLock(); |
| for (Service service : TopologyManager.getTopology().getServices()) { |
| for (Cluster cluster : service.getClusters()) { |
| if (cluster.memberExists(memberId)) { |
| return cluster.getMember(memberId); |
| } |
| } |
| } |
| return null; |
| } finally { |
| TopologyManager.releaseReadLock(); |
| } |
| } |
| |
| public void terminate() { |
| this.terminated = true; |
| } |
| |
| public ExecutorService getExecutorService() { |
| return executorService; |
| } |
| |
| public void setExecutorService(ExecutorService executorService) { |
| this.executorService = executorService; |
| } |
| } |
