catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DmaapHealth.java

   1 package org.openecomp.sdc.be.components.distribution.engine;
   2
   3 import org.apache.commons.validator.routines.UrlValidator;
   4 import org.apache.http.client.utils.URIUtils;
   5 import org.openecomp.sdc.be.config.BeEcompErrorManager;
   6 import org.openecomp.sdc.be.config.ConfigurationManager;
   7 import org.openecomp.sdc.be.config.DmaapConsumerConfiguration;
   8 import org.openecomp.sdc.common.api.HealthCheckInfo;
   9 import org.slf4j.Logger;
  10 import org.slf4j.LoggerFactory;
  11 import org.springframework.stereotype.Component;
  12
  13 import javax.annotation.PostConstruct;
  14 import javax.annotation.PreDestroy;
  15 import java.io.IOException;
  16 import java.net.InetAddress;
  17 import java.net.URI;
  18 import java.net.URISyntaxException;
  19 import java.util.concurrent.Executors;
  20 import java.util.concurrent.ScheduledExecutorService;
  21 import java.util.concurrent.ScheduledFuture;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.atomic.AtomicBoolean;
  24
  25 import static org.apache.commons.lang3.StringUtils.*;
  26 import static org.openecomp.sdc.common.api.Constants.HC_COMPONENT_DMAAP_ENGINE;
  27
  28 @Component("dmaapHealth")
  29 public class DmaapHealth {
  30
  31
  32     protected static final String DMAAP_HEALTH_LOG_CONTEXT = "dmaap.healthcheck";
  33     private static final String DMAAP_HEALTH_CHECK_STR = "dmaapHealthCheck";
  34     private static final Logger log = LoggerFactory.getLogger(DmaapHealth.class);
  35     private static final Logger logHealth = LoggerFactory.getLogger(DMAAP_HEALTH_LOG_CONTEXT);
  36     private HealthCheckInfo healthCheckInfo = DmaapHealth.HealthCheckInfoResult.UNAVAILABLE.getHealthCheckInfo();
  37     private long healthCheckReadTimeout = 20;
  38     private long reconnectInterval = 5;
  39     private HealthCheckScheduledTask healthCheckScheduledTask = null ;
  40     private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
  41     private ScheduledFuture<?> scheduledFuture = null;
  42     private DmaapConsumerConfiguration configuration = null ;
  43
  44     private volatile AtomicBoolean lastHealthState = new AtomicBoolean(false);
  45     private volatile AtomicBoolean reportedHealthState = null;
  46
  47     public enum HealthCheckInfoResult {
  48         OK(new HealthCheckInfo(HC_COMPONENT_DMAAP_ENGINE, HealthCheckInfo.HealthCheckStatus.UP, null, DmaapStatusDescription.OK.getDescription())),
  49         UNAVAILABLE(new HealthCheckInfo(HC_COMPONENT_DMAAP_ENGINE, HealthCheckInfo.HealthCheckStatus.DOWN, null, DmaapStatusDescription.UNAVAILABLE.getDescription())),
  50         DOWN(new HealthCheckInfo(HC_COMPONENT_DMAAP_ENGINE, HealthCheckInfo.HealthCheckStatus.DOWN, null, DmaapStatusDescription.DOWN.getDescription()));
  51
  52         private HealthCheckInfo healthCheckInfo;
  53         HealthCheckInfoResult(HealthCheckInfo healthCheckInfo) {
  54             this.healthCheckInfo = healthCheckInfo;
  55         }
  56         public HealthCheckInfo getHealthCheckInfo() {
  57             return healthCheckInfo;
  58         }
  59     }
  60
  61     public enum DmaapStatusDescription {
  62         OK("OK"), UNAVAILABLE("Dmaap is not available"),DOWN("DOWN"), NOT_CONFIGURED("Dmaap configuration is missing/wrong ");
  63
  64         private String desc;
  65         DmaapStatusDescription(String desc) {
  66             this.desc = desc;
  67         }
  68         public String getDescription() {
  69             return desc;
  70         }
  71
  72     }
  73
  74     @PostConstruct
  75     public DmaapHealth init() {
  76         log.trace("Enter init method of Dmaap health");
  77         synchronized (DmaapHealth.class){
  78             this.configuration = ConfigurationManager.getConfigurationManager().getConfiguration().getDmaapConsumerConfiguration();
  79
  80             Integer pollingInterval = configuration.getPollingInterval();
  81             if (pollingInterval != null && pollingInterval!=0) {
  82                 reconnectInterval = pollingInterval;
  83             }
  84             Integer healthCheckReadTimeoutConfig = configuration.getTimeoutMs();
  85             if (healthCheckReadTimeoutConfig != null) {
  86                 this.healthCheckReadTimeout = healthCheckReadTimeoutConfig;
  87             }
  88             this.healthCheckScheduledTask = new HealthCheckScheduledTask( configuration ); //what is the representation? csv? delimiter? json or other
  89             startHealthCheckTask(true);
  90         }
  91         log.trace("Exit init method of DistributionEngineClusterHealth");
  92         return this;
  93     }
  94
  95     @PreDestroy
  96     protected void destroy() {
  97         if (scheduledFuture != null) {
  98             scheduledFuture.cancel(true);
  99             scheduledFuture = null;
 100         }
 101         if (scheduler != null) {
 102             scheduler.shutdown();
 103         }
 104     }
 105
 106     /**
 107      * Start health check task.
 108      *
 109      * @param startTask
 110      */
 111     public void startHealthCheckTask( boolean startTask ) {
 112         synchronized (DmaapHealth.class){
 113             if (startTask && this.scheduledFuture == null) {
 114                 this.scheduledFuture = this.scheduler.scheduleAtFixedRate(this.healthCheckScheduledTask , 0, reconnectInterval, TimeUnit.SECONDS);
 115             }
 116         }
 117     }
 118
 119     public void report(Boolean isUp){
 120         if (reportedHealthState == null)
 121             reportedHealthState = new AtomicBoolean(isUp);
 122         reportedHealthState.set(isUp);
 123     }
 124
 125     public void logAlarm(boolean lastHealthState) {
 126         try{
 127             if ( lastHealthState ) {
 128                 BeEcompErrorManager.getInstance().logDmaapHealthCheckRecovery( DMAAP_HEALTH_CHECK_STR );
 129             } else {
 130                 BeEcompErrorManager.getInstance().logDmaapHealthCheckError( DMAAP_HEALTH_CHECK_STR );
 131             }
 132         }catch( Exception e ){
 133             log.debug("cannot logAlarm -> {}" ,e );
 134         }
 135     }
 136
 137     public DmaapConsumerConfiguration getConfiguration() {
 138         return configuration;
 139     }
 140
 141     public HealthCheckInfo getHealthCheckInfo() {
 142         return healthCheckInfo;
 143     }
 144
 145     /**
 146      * Health Check Task Scheduler - infinite check.
 147      */
 148     public class HealthCheckScheduledTask implements Runnable {
 149         private final DmaapConsumerConfiguration config;
 150         private static final int timeout = 8192;
 151
 152         public HealthCheckScheduledTask(final DmaapConsumerConfiguration config){
 153             this.config = config;
 154         }
 155         @Override
 156         public void run() {
 157             logHealth.trace("Executing Dmaap Health Check Task - Start");
 158             boolean prevIsReachable = false;
 159             boolean reachable = false;
 160             //first try simple ping
 161             try{
 162                 if ( reportedHealthState != null ){
 163                     reachable = reportedHealthState.get();
 164                 }
 165                 else{
 166                     reachable =  false;
 167                 }
 168                 prevIsReachable = lastHealthState.getAndSet( reachable );
 169                 healthCheckInfo = reachable ? HealthCheckInfoResult.OK.healthCheckInfo : HealthCheckInfoResult.DOWN.healthCheckInfo;
 170             }
 171             catch( Exception e ){
 172                 log.debug("{} | cannot check connectivity -> {}",DMAAP_HEALTH_CHECK_STR, e );
 173                 prevIsReachable = lastHealthState.getAndSet(false);
 174                 healthCheckInfo = HealthCheckInfoResult.UNAVAILABLE.healthCheckInfo;
 175             }
 176             if (prevIsReachable != lastHealthState.get())
 177                 logAlarm( lastHealthState.get() );
 178         }
 179
 180
 181         /**
 182          * @deprecated (health is reported outside from EnvironmentEngine consumer fetch)
 183          */
 184         @Deprecated
 185         public boolean isICMPReachable( ) throws IOException{
 186             try{
 187                 String hostname = getUrlHost(config.getHosts());
 188                 return InetAddress.getByName( hostname ).isReachable(timeout);
 189             }catch( URISyntaxException e ){
 190                 log.debug("{} | malformed host configuration -> ",DMAAP_HEALTH_CHECK_STR , e);
 191             }
 192             return false;
 193         }
 194     }
 195
 196     public static String getUrlHost(String qualifiedHost) throws URISyntaxException{
 197         //region - parse complex format ex. <http://URL:PORT>
 198         try{
 199             UrlValidator validator = new UrlValidator();
 200             if (validator.isValid(qualifiedHost)){
 201                 return URIUtils.extractHost(new URI(qualifiedHost)).getHostName();
 202             }else{
 203                 log.debug("{} | invalid url format, continuing ", DMAAP_HEALTH_CHECK_STR );
 204             }
 205         }catch(URISyntaxException e){
 206             log.debug("{} | invalid url format, continuing {} ", DMAAP_HEALTH_CHECK_STR , e);
 207         }
 208         //endregion
 209
 210         //region - try shortcut format <URL> or <URL:PORT>
 211         if ( countMatches( qualifiedHost , ":") <= 1){
 212             String[] address = qualifiedHost.split(":");
 213             if ( address.length>0 && isNotBlank(address[0]) ){
 214                 return address[0];
 215             }
 216         }
 217         //endregion
 218         throw new URISyntaxException( qualifiedHost , "invalid hostname, expecting a single <host:port> ,  (valid ex. www.google.com:80 | www.google.com | http:\\\\www.google.com:8181)");
 219     }
 220 }