geode-wan/src/main/java/org/apache/geode/cache/client/internal/locator/wan/LocatorDiscovery.java - geode - Git at Google

 /*
  * 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.geode.cache.client.internal.locator.wan;

 import java.io.IOException;
 import java.util.concurrent.ConcurrentHashMap;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.distributed.internal.DistributionConfig;
 import org.apache.geode.distributed.internal.WanLocatorDiscoverer;
 import org.apache.geode.distributed.internal.tcpserver.TcpClient;
 import org.apache.geode.internal.admin.remote.DistributionLocatorId;
 import org.apache.geode.internal.tcp.ConnectionException;
 import org.apache.geode.logging.internal.log4j.api.LogService;

 /**
  * This class represent a runnable task which exchange the locator information with local
  * locators(within the site) as well as remote locators (across the site)
  *
  * @since GemFire 7.0
  */
 public class LocatorDiscovery {

   private static final Logger logger = LogService.getLogger();

   private WanLocatorDiscoverer discoverer;

   private DistributionLocatorId locatorId;

   private LocatorMembershipListener locatorListener;

   RemoteLocatorJoinRequest request;

   TcpClient locatorClient;

   public static final int WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT =
       Integer.getInteger("WANLocator.CONNECTION_RETRY_ATTEMPT", 50000).intValue();

   public static final int WAN_LOCATOR_CONNECTION_INTERVAL =
       Integer.getInteger("WANLocator.CONNECTION_INTERVAL", 10000).intValue();

   public static final int WAN_LOCATOR_PING_INTERVAL =
       Integer.getInteger("WANLocator.PING_INTERVAL", 10000).intValue();

   public LocatorDiscovery(WanLocatorDiscoverer discoverer, DistributionLocatorId locator,
       RemoteLocatorJoinRequest request, LocatorMembershipListener locatorListener) {
     this.discoverer = discoverer;
     this.locatorId = locator;
     this.request = request;
     this.locatorListener = locatorListener;
     this.locatorClient = new TcpClient();
   }

   /**
    * When a batch fails, then this keeps the last time when a failure was logged . We don't want to
    * swamp the logs in retries due to same batch failures.
    */
   private final ConcurrentHashMap<DistributionLocatorId, long[]> failureLogInterval =
       new ConcurrentHashMap<DistributionLocatorId, long[]>();

   /**
    * The maximum size of {@link #failureLogInterval} beyond which it will start logging all failure
    * instances. Hopefully this should never happen in practice.
    */
   private static final int FAILURE_MAP_MAXSIZE = Integer
       .getInteger(DistributionConfig.GEMFIRE_PREFIX + "GatewaySender.FAILURE_MAP_MAXSIZE", 1000000);

   /**
    * The maximum interval for logging failures of the same event in millis.
    */
   private static final int FAILURE_LOG_MAX_INTERVAL = Integer.getInteger(
       DistributionConfig.GEMFIRE_PREFIX + "LocatorDiscovery.FAILURE_LOG_MAX_INTERVAL", 300000);

   public boolean skipFailureLogging(DistributionLocatorId locatorId) {
     boolean skipLogging = false;
     if (this.failureLogInterval.size() < FAILURE_MAP_MAXSIZE) {
       long[] logInterval = this.failureLogInterval.get(locatorId);
       if (logInterval == null) {
         logInterval = this.failureLogInterval.putIfAbsent(locatorId,
             new long[] {System.currentTimeMillis(), 1000});
       }
       if (logInterval != null) {
         long currentTime = System.currentTimeMillis();
         if ((currentTime - logInterval[0]) < logInterval[1]) {
           skipLogging = true;
         } else {
           logInterval[0] = currentTime;
           if (logInterval[1] <= (FAILURE_LOG_MAX_INTERVAL / 2)) {
             logInterval[1] *= 2;
           }
         }
       }
     }
     return skipLogging;
   }


   public class LocalLocatorDiscovery implements Runnable {
     @Override
     public void run() {
       exchangeLocalLocators();
     }
   }

   public class RemoteLocatorDiscovery implements Runnable {
     @Override
     public void run() {
       exchangeRemoteLocators();
     }
   }

   private WanLocatorDiscoverer getDiscoverer() {
     return this.discoverer;
   }

   private void exchangeLocalLocators() {
     int retryAttempt = 1;
     while (!getDiscoverer().isStopped()) {
       try {
         RemoteLocatorJoinResponse response =
             (RemoteLocatorJoinResponse) locatorClient.requestToServer(locatorId.getHost(), request,
                 WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
         if (response != null) {
           LocatorHelper.addExchangedLocators(response.getLocators(), this.locatorListener);
           logger.info("Locator discovery task exchanged locator information {} with {}: {}.",
               new Object[] {request.getLocator(), locatorId, response.getLocators()});
           break;
         }
       } catch (IOException ioe) {
         if (retryAttempt == WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT) {
           ConnectionException coe =
               new ConnectionException("Not able to connect to local locator after "
                   + WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT + " retry attempts", ioe);
           logger.fatal(String.format(
               "Locator discovery task could not exchange locator information %s with %s after %s retry attempts.",
               new Object[] {request.getLocator(), locatorId, retryAttempt}),
               coe);
           break;
         }
         if (skipFailureLogging(locatorId)) {
           logger.warn(
               "Locator discovery task could not exchange locator information {} with {} after {} retry attempts. Retrying in {} ms.",
               new Object[] {request.getLocator(), locatorId, retryAttempt,
                   WAN_LOCATOR_CONNECTION_INTERVAL});
         }
         try {
           Thread.sleep(WAN_LOCATOR_CONNECTION_INTERVAL);
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
         }
         retryAttempt++;
         continue;
       } catch (ClassNotFoundException classNotFoundException) {
         logger.fatal("Locator discovery task encountred unexpected exception",
             classNotFoundException);
         break;
       }
     }
   }

   public void exchangeRemoteLocators() {
     int retryAttempt = 1;
     DistributionLocatorId remoteLocator = this.locatorId;
     while (!getDiscoverer().isStopped()) {
       RemoteLocatorJoinResponse response;
       try {
         response =
             (RemoteLocatorJoinResponse) locatorClient.requestToServer(remoteLocator.getHost(),
                 request, WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
         if (response != null) {
           LocatorHelper.addExchangedLocators(response.getLocators(), this.locatorListener);
           logger.info("Locator discovery task exchanged locator information {} with {}: {}.",
               new Object[] {request.getLocator(), locatorId, response.getLocators()});
           RemoteLocatorPingRequest pingRequest = new RemoteLocatorPingRequest("");
           while (true) {
             Thread.sleep(WAN_LOCATOR_PING_INTERVAL);
             RemoteLocatorPingResponse pingResponse =
                 (RemoteLocatorPingResponse) locatorClient.requestToServer(remoteLocator.getHost(),
                     pingRequest, WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
             if (pingResponse != null) {
               continue;
             }
             break;
           }
         }
       } catch (IOException ioe) {
         if (retryAttempt == WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT) {
           logger.fatal(String.format(
               "Locator discovery task could not exchange locator information %s with %s after %s retry attempts.",
               new Object[] {request.getLocator(), remoteLocator, retryAttempt}),
               ioe);
           break;
         }
         if (skipFailureLogging(remoteLocator)) {
           logger.warn(
               "Locator discovery task could not exchange locator information {} with {} after {} retry attempts. Retrying in {} ms.",
               new Object[] {request.getLocator(), remoteLocator, retryAttempt,
                   WAN_LOCATOR_CONNECTION_INTERVAL});
         }
         try {
           Thread.sleep(WAN_LOCATOR_CONNECTION_INTERVAL);
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
         }
         retryAttempt++;
         continue;
       } catch (ClassNotFoundException classNotFoundException) {
         logger.fatal("Locator discovery task encountred unexpected exception",
             classNotFoundException);
         break;
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
       }
     }
   }

 }
	/*
	* 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.geode.cache.client.internal.locator.wan;

	import java.io.IOException;
	import java.util.concurrent.ConcurrentHashMap;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.distributed.internal.DistributionConfig;
	import org.apache.geode.distributed.internal.WanLocatorDiscoverer;
	import org.apache.geode.distributed.internal.tcpserver.TcpClient;
	import org.apache.geode.internal.admin.remote.DistributionLocatorId;
	import org.apache.geode.internal.tcp.ConnectionException;
	import org.apache.geode.logging.internal.log4j.api.LogService;

	/**
	* This class represent a runnable task which exchange the locator information with local
	* locators(within the site) as well as remote locators (across the site)
	*
	* @since GemFire 7.0
	*/
	public class LocatorDiscovery {

	private static final Logger logger = LogService.getLogger();

	private WanLocatorDiscoverer discoverer;

	private DistributionLocatorId locatorId;

	private LocatorMembershipListener locatorListener;

	RemoteLocatorJoinRequest request;

	TcpClient locatorClient;

	public static final int WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT =
	Integer.getInteger("WANLocator.CONNECTION_RETRY_ATTEMPT", 50000).intValue();

	public static final int WAN_LOCATOR_CONNECTION_INTERVAL =
	Integer.getInteger("WANLocator.CONNECTION_INTERVAL", 10000).intValue();

	public static final int WAN_LOCATOR_PING_INTERVAL =
	Integer.getInteger("WANLocator.PING_INTERVAL", 10000).intValue();

	public LocatorDiscovery(WanLocatorDiscoverer discoverer, DistributionLocatorId locator,
	RemoteLocatorJoinRequest request, LocatorMembershipListener locatorListener) {
	this.discoverer = discoverer;
	this.locatorId = locator;
	this.request = request;
	this.locatorListener = locatorListener;
	this.locatorClient = new TcpClient();
	}

	/**
	* When a batch fails, then this keeps the last time when a failure was logged . We don't want to
	* swamp the logs in retries due to same batch failures.
	*/
	private final ConcurrentHashMap<DistributionLocatorId, long[]> failureLogInterval =
	new ConcurrentHashMap<DistributionLocatorId, long[]>();

	/**
	* The maximum size of {@link #failureLogInterval} beyond which it will start logging all failure
	* instances. Hopefully this should never happen in practice.
	*/
	private static final int FAILURE_MAP_MAXSIZE = Integer
	.getInteger(DistributionConfig.GEMFIRE_PREFIX + "GatewaySender.FAILURE_MAP_MAXSIZE", 1000000);

	/**
	* The maximum interval for logging failures of the same event in millis.
	*/
	private static final int FAILURE_LOG_MAX_INTERVAL = Integer.getInteger(
	DistributionConfig.GEMFIRE_PREFIX + "LocatorDiscovery.FAILURE_LOG_MAX_INTERVAL", 300000);

	public boolean skipFailureLogging(DistributionLocatorId locatorId) {
	boolean skipLogging = false;
	if (this.failureLogInterval.size() < FAILURE_MAP_MAXSIZE) {
	long[] logInterval = this.failureLogInterval.get(locatorId);
	if (logInterval == null) {
	logInterval = this.failureLogInterval.putIfAbsent(locatorId,
	new long[] {System.currentTimeMillis(), 1000});
	}
	if (logInterval != null) {
	long currentTime = System.currentTimeMillis();
	if ((currentTime - logInterval[0]) < logInterval[1]) {
	skipLogging = true;
	} else {
	logInterval[0] = currentTime;
	if (logInterval[1] <= (FAILURE_LOG_MAX_INTERVAL / 2)) {
	logInterval[1] *= 2;
	}
	}
	}
	}
	return skipLogging;
	}


	public class LocalLocatorDiscovery implements Runnable {
	@Override
	public void run() {
	exchangeLocalLocators();
	}
	}

	public class RemoteLocatorDiscovery implements Runnable {
	@Override
	public void run() {
	exchangeRemoteLocators();
	}
	}

	private WanLocatorDiscoverer getDiscoverer() {
	return this.discoverer;
	}

	private void exchangeLocalLocators() {
	int retryAttempt = 1;
	while (!getDiscoverer().isStopped()) {
	try {
	RemoteLocatorJoinResponse response =
	(RemoteLocatorJoinResponse) locatorClient.requestToServer(locatorId.getHost(), request,
	WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
	if (response != null) {
	LocatorHelper.addExchangedLocators(response.getLocators(), this.locatorListener);
	logger.info("Locator discovery task exchanged locator information {} with {}: {}.",
	new Object[] {request.getLocator(), locatorId, response.getLocators()});
	break;
	}
	} catch (IOException ioe) {
	if (retryAttempt == WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT) {
	ConnectionException coe =
	new ConnectionException("Not able to connect to local locator after "
	+ WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT + " retry attempts", ioe);
	logger.fatal(String.format(
	"Locator discovery task could not exchange locator information %s with %s after %s retry attempts.",
	new Object[] {request.getLocator(), locatorId, retryAttempt}),
	coe);
	break;
	}
	if (skipFailureLogging(locatorId)) {
	logger.warn(
	"Locator discovery task could not exchange locator information {} with {} after {} retry attempts. Retrying in {} ms.",
	new Object[] {request.getLocator(), locatorId, retryAttempt,
	WAN_LOCATOR_CONNECTION_INTERVAL});
	}
	try {
	Thread.sleep(WAN_LOCATOR_CONNECTION_INTERVAL);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	}
	retryAttempt++;
	continue;
	} catch (ClassNotFoundException classNotFoundException) {
	logger.fatal("Locator discovery task encountred unexpected exception",
	classNotFoundException);
	break;
	}
	}
	}

	public void exchangeRemoteLocators() {
	int retryAttempt = 1;
	DistributionLocatorId remoteLocator = this.locatorId;
	while (!getDiscoverer().isStopped()) {
	RemoteLocatorJoinResponse response;
	try {
	response =
	(RemoteLocatorJoinResponse) locatorClient.requestToServer(remoteLocator.getHost(),
	request, WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
	if (response != null) {
	LocatorHelper.addExchangedLocators(response.getLocators(), this.locatorListener);
	logger.info("Locator discovery task exchanged locator information {} with {}: {}.",
	new Object[] {request.getLocator(), locatorId, response.getLocators()});
	RemoteLocatorPingRequest pingRequest = new RemoteLocatorPingRequest("");
	while (true) {
	Thread.sleep(WAN_LOCATOR_PING_INTERVAL);
	RemoteLocatorPingResponse pingResponse =
	(RemoteLocatorPingResponse) locatorClient.requestToServer(remoteLocator.getHost(),
	pingRequest, WanLocatorDiscoverer.WAN_LOCATOR_CONNECTION_TIMEOUT, true);
	if (pingResponse != null) {
	continue;
	}
	break;
	}
	}
	} catch (IOException ioe) {
	if (retryAttempt == WAN_LOCATOR_CONNECTION_RETRY_ATTEMPT) {
	logger.fatal(String.format(
	"Locator discovery task could not exchange locator information %s with %s after %s retry attempts.",
	new Object[] {request.getLocator(), remoteLocator, retryAttempt}),
	ioe);
	break;
	}
	if (skipFailureLogging(remoteLocator)) {
	logger.warn(
	"Locator discovery task could not exchange locator information {} with {} after {} retry attempts. Retrying in {} ms.",
	new Object[] {request.getLocator(), remoteLocator, retryAttempt,
	WAN_LOCATOR_CONNECTION_INTERVAL});
	}
	try {
	Thread.sleep(WAN_LOCATOR_CONNECTION_INTERVAL);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	}
	retryAttempt++;
	continue;
	} catch (ClassNotFoundException classNotFoundException) {
	logger.fatal("Locator discovery task encountred unexpected exception",
	classNotFoundException);
	break;
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	}
	}
	}

	}