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

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