dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/DMaaPMRUtils.java

   1 /*\r
   2  * ===============================LICENSE_START======================================\r
   3  *  dcae-analytics\r
   4  * ================================================================================\r
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * ================================================================================\r
   7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   8  *  you may not use this file except in compliance with the License.\r
   9  *   You may obtain a copy of the License at\r
  10  *\r
  11  *          http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  *  Unless required by applicable law or agreed to in writing, software\r
  14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  *  See the License for the specific language governing permissions and\r
  17  *  limitations under the License.\r
  18  *  ============================LICENSE_END===========================================\r
  19  */\r
  20 \r
  21 package org.openecomp.dcae.apod.analytics.cdap.common.utils;\r
  22 \r
  23 import co.cask.cdap.api.metrics.Metrics;\r
  24 import com.google.common.base.Optional;\r
  25 import com.google.common.base.Stopwatch;\r
  26 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;\r
  27 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
  28 import org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils;\r
  29 import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;\r
  30 import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;\r
  31 import org.slf4j.Logger;\r
  32 import org.slf4j.LoggerFactory;\r
  33 \r
  34 import java.util.List;\r
  35 \r
  36 /**\r
  37  * Utility common methods for DMaaP MR functionality\r
  38  *\r
  39  * @author Rajiv Singla . Creation Date: 2/6/2017.\r
  40  */\r
  41 public abstract class DMaaPMRUtils {\r
  42 \r
  43     private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRUtils.class);\r
  44 \r
  45     private DMaaPMRUtils() {\r
  46         // private constructor\r
  47     }\r
  48 \r
  49 \r
  50     /**\r
  51      * Returns messages fetched from DMaaP MR Subscriber.\r
  52      *\r
  53      * @param subscriber DMaaP MR Subscriber instance\r
  54      * @param metrics CDAP metrics\r
  55      *\r
  56      * @return messages fetched from DMaaP MR topic\r
  57      */\r
  58     public static Optional<List<String>> getSubscriberMessages(final DMaaPMRSubscriber subscriber,\r
  59                                                                final Metrics metrics) {\r
  60 \r
  61         final Optional<DMaaPMRSubscriberResponse> subscriberResponseOptional =\r
  62                 getSubscriberResponse(subscriber, metrics);\r
  63 \r
  64         // If response is not present, unable to proceed\r
  65         if (!subscriberResponseOptional.isPresent()) {\r
  66             return Optional.absent();\r
  67         }\r
  68 \r
  69         final DMaaPMRSubscriberResponse subscriberResponse = subscriberResponseOptional.get();\r
  70 \r
  71         // If response code return by the subscriber call is not successful, unable to do proceed\r
  72         if (!HTTPUtils.isSuccessfulResponseCode(subscriberResponse.getResponseCode())) {\r
  73             LOG.error("Subscriber was unable to fetch messages properly.Subscriber Response Code: {} " +\r
  74                     "Unable to proceed further....", subscriberResponse.getResponseCode());\r
  75             metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_UNSUCCESSFUL_RESPONSES_METRIC, 1);\r
  76             return Optional.absent();\r
  77         }\r
  78 \r
  79         LOG.debug("Subscriber HTTP Response Status Code match successful:  {}", subscriberResponse,\r
  80                 HTTPUtils.HTTP_SUCCESS_STATUS_CODE);\r
  81 \r
  82         final List<String> actualMessages = subscriberResponse.getFetchedMessages();\r
  83 \r
  84         // If there are no message returned during from Subscriber, nothing to write to CDAP Stream\r
  85         if (actualMessages.isEmpty()) {\r
  86             LOG.debug("Subscriber Response has no messages. Nothing to write....");\r
  87             metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_RESPONSES_WITH_NO_MESSAGES_METRIC, 1);\r
  88             return Optional.absent();\r
  89         }\r
  90 \r
  91         LOG.debug("DMaaP MR Subscriber found new messages in DMaaP Topic. Message count: {}", actualMessages.size());\r
  92         metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_TOTAL_MESSAGES_PROCESSED_METRIC, actualMessages.size());\r
  93 \r
  94         return Optional.of(actualMessages);\r
  95 \r
  96     }\r
  97 \r
  98 \r
  99     /**\r
 100      * Get Subscriber response and records time taken to fetch messages. Returns Optional.None if Subscriber response\r
 101      * is null or response status code is not present\r
 102      *\r
 103      * @param subscriber - DMaaP Subscriber\r
 104      * @param metrics - CDAP Metrics collector\r
 105      *\r
 106      * @return - Optional of Subscriber Response\r
 107      */\r
 108     public static Optional<DMaaPMRSubscriberResponse> getSubscriberResponse(final DMaaPMRSubscriber subscriber,\r
 109                                                                             final Metrics metrics) {\r
 110 \r
 111         // Record all response count from subscriber\r
 112         metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_ALL_RESPONSES_COUNT_METRIC, 1);\r
 113 \r
 114         // Check how long it took for subscriber to respond\r
 115         final Stopwatch stopwatch = new Stopwatch();\r
 116         stopwatch.start();\r
 117 \r
 118         // Fetch messages from DMaaP MR Topic\r
 119         DMaaPMRSubscriberResponse subscriberResponse = null;\r
 120         try {\r
 121             subscriberResponse = subscriber.fetchMessages();\r
 122         } catch (DCAEAnalyticsRuntimeException e) {\r
 123             LOG.error("Error while fetching messages for DMaaP MR Topic: {}", e);\r
 124         }\r
 125 \r
 126         stopwatch.stop();\r
 127         final long subscriberResponseTimeMS = stopwatch.elapsedMillis();\r
 128 \r
 129         // If response is null is null or response code is null, unable to proceed nothing to do\r
 130         if (subscriberResponse == null || subscriberResponse.getResponseCode() == null) {\r
 131             LOG.error("Subscriber Response is null or subscriber Response code is null. Unable to proceed further...");\r
 132             return Optional.absent();\r
 133         }\r
 134 \r
 135         LOG.debug("Subscriber Response:{}, Subscriber HTTP Response Status Code {}, Subscriber Response Time(ms): {}",\r
 136                 subscriberResponse, subscriberResponse.getResponseCode(), subscriberResponseTimeMS);\r
 137 \r
 138         // Record subscriber response time\r
 139         metrics.gauge(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_RESPONSE_TIME_MS_METRIC, subscriberResponseTimeMS);\r
 140 \r
 141         return Optional.of(subscriberResponse);\r
 142     }\r
 143 \r
 144 }\r