<commit_before>/* 
 * Licensed 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.ngrinder.agent.service;

import static net.grinder.message.console.AgentControllerState.INACTIVE;
import static net.grinder.message.console.AgentControllerState.WRONG_REGION;
import static org.ngrinder.agent.model.ClustedAgentRequest.RequestType.SHARE_AGENT_SYSTEM_DATA_MODEL;
import static org.ngrinder.agent.model.ClustedAgentRequest.RequestType.STOP_AGENT;
import static org.ngrinder.agent.repository.AgentManagerSpecification.startWithRegion;
import static org.ngrinder.common.util.CollectionUtils.newHashMap;
import static org.ngrinder.common.util.TypeConvertUtil.convert;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.annotation.PostConstruct;

import net.grinder.common.processidentity.AgentIdentity;
import net.grinder.engine.controller.AgentControllerIdentityImplementation;
import net.grinder.util.thread.InterruptibleRunnable;
import net.sf.ehcache.Ehcache;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.mutable.MutableInt;
import org.ngrinder.agent.model.AgentInfo;
import org.ngrinder.agent.model.ClustedAgentRequest;
import org.ngrinder.infra.logger.CoreLogger;
import org.ngrinder.infra.schedule.ScheduledTask;
import org.ngrinder.model.User;
import org.ngrinder.monitor.controller.model.SystemDataModel;
import org.ngrinder.perftest.service.AgentManager;
import org.ngrinder.region.service.RegionService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
import org.springframework.cache.Cache.ValueWrapper;
import org.springframework.cache.CacheManager;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;

import com.google.common.collect.Maps;

/**
 * Cluster enabled version of {@link AgentManagerService}.
 * 
 * @author <NAME>
 * @since 3.1
 */
public class ClusteredAgentManagerService extends AgentManagerService {
	private static final Logger LOGGER = LoggerFactory.getLogger(ClusteredAgentManagerService.class);

	@Autowired
	private CacheManager cacheManager;

	private Cache agentRequestCache;

	private Cache agentMonitorCache;

	private Cache agentMonioringTargetsCache;

	@Autowired
	private ScheduledTask scheduledTask;

	@Autowired
	private RegionService regionService;

	/**
	 * Initialize.
	 */
	@PostConstruct
	public void init() {
		agentMonioringTargetsCache = getCacheManager().getCache("agent_monitoring_targets");
		if (getConfig().isCluster()) {
			agentRequestCache = getCacheManager().getCache("agent_request");
			agentMonitorCache = getCacheManager().getCache("agent_monitoring");
			scheduledTask.addScheduledTaskEvery3Sec(new InterruptibleRunnable() {
				@Override
				public void interruptibleRun() {
					List<String> keysWithExpiryCheck = convert(((Ehcache) agentRequestCache.getNativeCache())
									.getKeysWithExpiryCheck());
					String region = getConfig().getRegion() + "|";
					for (String each : keysWithExpiryCheck) {
						try {
							if (each.startsWith(region) && agentRequestCache.get(each) != null) {
								ClustedAgentRequest agentRequest = convert(agentRequestCache.get(each).get());
								AgentControllerIdentityImplementation agentIdentity = getLocalAgentIdentityByIpAndName(
												agentRequest.getAgentIp(), agentRequest.getAgentName());
								if (agentIdentity != null) {
									agentRequest.getRequestType().process(ClusteredAgentManagerService.this,
													agentIdentity);
								}
							}

						} catch (Exception e) {
							CoreLogger.LOGGER.error(e.getMessage(), e);
						}
						agentRequestCache.evict(each);
					}
				}
			});
		}
	}

	/**
	 * Run a scheduled task to check the agent status.
	 * 
	 * @since 3.1
	 */
	public void checkAgentStatus() {
		List<AgentInfo> changeAgents = new ArrayList<AgentInfo>();
		List<AgentInfo> deleteAgents = new ArrayList<AgentInfo>();
		String curRegion = getConfig().getRegion();
		Set<String> regions = getRegions();

		Set<AgentIdentity> allAttachedAgents = getAgentManager().getAllAttachedAgents();
		Map<String, AgentControllerIdentityImplementation> attachedAgentMap = newHashMap(allAttachedAgents);
		for (AgentIdentity agentIdentity : allAttachedAgents) {
			AgentControllerIdentityImplementation existingAgent = convert(agentIdentity);
			attachedAgentMap.put(createAgentKey(existingAgent), existingAgent);
		}

		List<AgentInfo> agentsInDB = getAgentRepository().findAll();
		Map<String, AgentInfo> agentsInDBMap = Maps.newHashMap();
		// step1. check all agents in DB, whether they are attached to controller.
		for (AgentInfo eachAgentInDB : agentsInDB) {
			String keyOfAgentInDB = createAgentKey(eachAgentInDB);
			agentsInDBMap.put(keyOfAgentInDB, eachAgentInDB);
			AgentControllerIdentityImplementation agentIdentity = attachedAgentMap.remove(keyOfAgentInDB);
			String regionOfEachAgentInDB = extractRegionFromAgentRegion(eachAgentInDB.getRegion());

			if (agentIdentity != null) {// if the agent attached to current controller
				if (StringUtils.equals(regionOfEachAgentInDB, curRegion)) {
					if (!hasSamePortAndStatus(eachAgentInDB, agentIdentity)) {
						fillUp(eachAgentInDB, agentIdentity);
						changeAgents.add(eachAgentInDB);
					}
				} else if (eachAgentInDB.getStatus() != WRONG_REGION) {
					// the region config is wrong
					eachAgentInDB.setStatus(WRONG_REGION);
					changeAgents.add(eachAgentInDB);

				}
			} else { 
				if (eachAgentInDB.getStatus() != INACTIVE) {
					eachAgentInDB.setStatus(INACTIVE);
					changeAgents.add(eachAgentInDB);
				}
				// the agent in DB is not attached to current controller
				if (StringUtils.equals(regionOfEachAgentInDB, curRegion)) {
					// the agent WAS attached to this controller before, but it is down.
					
				}
				else if (!regions.contains(regionOfEachAgentInDB)) {
					// this agent in DB 's region is not in any region
					deleteAgents.add(eachAgentInDB);
				}
			}<commit_msg>[NOBTS] Rollback changes.<commit_after>/* 
 * Licensed 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.ngrinder.agent.service;

import static net.grinder.message.console.AgentControllerState.INACTIVE;
import static net.grinder.message.console.AgentControllerState.WRONG_REGION;
import static org.ngrinder.agent.model.ClustedAgentRequest.RequestType.SHARE_AGENT_SYSTEM_DATA_MODEL;
import static org.ngrinder.agent.model.ClustedAgentRequest.RequestType.STOP_AGENT;
import static org.ngrinder.agent.repository.AgentManagerSpecification.startWithRegion;
import static org.ngrinder.common.util.CollectionUtils.newHashMap;
import static org.ngrinder.common.util.TypeConvertUtil.convert;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.annotation.PostConstruct;

import net.grinder.common.processidentity.AgentIdentity;
import net.grinder.engine.controller.AgentControllerIdentityImplementation;
import net.grinder.util.thread.InterruptibleRunnable;
import net.sf.ehcache.Ehcache;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.mutable.MutableInt;
import org.ngrinder.agent.model.AgentInfo;
import org.ngrinder.agent.model.ClustedAgentRequest;
import org.ngrinder.infra.logger.CoreLogger;
import org.ngrinder.infra.schedule.ScheduledTask;
import org.ngrinder.model.User;
import org.ngrinder.monitor.controller.model.SystemDataModel;
import org.ngrinder.perftest.service.AgentManager;
import org.ngrinder.region.service.RegionService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
import org.springframework.cache.Cache.ValueWrapper;
import org.springframework.cache.CacheManager;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;

import com.google.common.collect.Maps;

/**
 * Cluster enabled version of {@link AgentManagerService}.
 * 
 * @author <NAME>
 * @since 3.1
 */
public class ClusteredAgentManagerService extends AgentManagerService {
	private static final Logger LOGGER = LoggerFactory.getLogger(ClusteredAgentManagerService.class);

	@Autowired
	private CacheManager cacheManager;

	private Cache agentRequestCache;

	private Cache agentMonitorCache;

	private Cache agentMonioringTargetsCache;

	@Autowired
	private ScheduledTask scheduledTask;

	@Autowired
	private RegionService regionService;

	/**
	 * Initialize.
	 */
	@PostConstruct
	public void init() {
		agentMonioringTargetsCache = getCacheManager().getCache("agent_monitoring_targets");
		if (getConfig().isCluster()) {
			agentRequestCache = getCacheManager().getCache("agent_request");
			agentMonitorCache = getCacheManager().getCache("agent_monitoring");
			scheduledTask.addScheduledTaskEvery3Sec(new InterruptibleRunnable() {
				@Override
				public void interruptibleRun() {
					List<String> keysWithExpiryCheck = convert(((Ehcache) agentRequestCache.getNativeCache())
									.getKeysWithExpiryCheck());
					String region = getConfig().getRegion() + "|";
					for (String each : keysWithExpiryCheck) {
						try {
							if (each.startsWith(region) && agentRequestCache.get(each) != null) {
								ClustedAgentRequest agentRequest = convert(agentRequestCache.get(each).get());
								AgentControllerIdentityImplementation agentIdentity = getLocalAgentIdentityByIpAndName(
												agentRequest.getAgentIp(), agentRequest.getAgentName());
								if (agentIdentity != null) {
									agentRequest.getRequestType().process(ClusteredAgentManagerService.this,
													agentIdentity);
								}
							}

						} catch (Exception e) {
							CoreLogger.LOGGER.error(e.getMessage(), e);
						}
						agentRequestCache.evict(each);
					}
				}
			});
		}
	}

	/**
	 * Run a scheduled task to check the agent status.
	 * 
	 * @since 3.1
	 */
	public void checkAgentStatus() {
		List<AgentInfo> changeAgents = new ArrayList<AgentInfo>();
		List<AgentInfo> deleteAgents = new ArrayList<AgentInfo>();
		String curRegion = getConfig().getRegion();
		Set<String> regions = getRegions();

		Set<AgentIdentity> allAttachedAgents = getAgentManager().getAllAttachedAgents();
		Map<String, AgentControllerIdentityImplementation> attachedAgentMap = newHashMap(allAttachedAgents);
		for (AgentIdentity agentIdentity : allAttachedAgents) {
			AgentControllerIdentityImplementation existingAgent = convert(agentIdentity);
			attachedAgentMap.put(createAgentKey(existingAgent), existingAgent);
		}

		List<AgentInfo> agentsInDB = getAgentRepository().findAll();
		Map<String, AgentInfo> agentsInDBMap = Maps.newHashMap();
		// step1. check all agents in DB, whether they are attached to controller.
		for (AgentInfo eachAgentInDB : agentsInDB) {
			String keyOfAgentInDB = createAgentKey(eachAgentInDB);
			agentsInDBMap.put(keyOfAgentInDB, eachAgentInDB);
			AgentControllerIdentityImplementation agentIdentity = attachedAgentMap.remove(keyOfAgentInDB);
			String regionOfEachAgentInDB = extractRegionFromAgentRegion(eachAgentInDB.getRegion());

			if (agentIdentity != null) {// if the agent attached to current controller
				if (StringUtils.equals(regionOfEachAgentInDB, curRegion)) {
					if (!hasSamePortAndStatus(eachAgentInDB, agentIdentity)) {
						fillUp(eachAgentInDB, agentIdentity);
						changeAgents.add(eachAgentInDB);
					}
				} else if (eachAgentInDB.getStatus() != WRONG_REGION) {
					// the region config is wrong
					eachAgentInDB.setStatus(WRONG_REGION);
					changeAgents.add(eachAgentInDB);

				}
			} else { // the agent in DB is not attached to current controller
				if (StringUtils.equals(regionOfEachAgentInDB, curRegion)) {
					// the agent WAS attached to this controller before, but it is down.
					if (eachAgentInDB.getStatus() != INACTIVE) {
						eachAgentInDB.setStatus(INACTIVE);
						changeAgents.add(eachAgentInDB);
					}
				} else if (!regions.contains(regionOfEachAgentInDB)) {
					// this agent in DB 's region is not in any region
					deleteAgents.add(eachAgentInDB);
				}
			}
{'fraction_non_alphanumeric': 0.09600666818216261, 'fraction_numerical': 0.0018185951352580132, 'mean_word_length': 4.837240159221583, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1798565', 'n_tokens_mistral': 4283, 'n_tokens_neox': 3745, 'n_words': 820}