protected boolean emitMetricsJson()

in ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java [182:266]
76 lines of code
20 McCabe index (conditional complexity)

  protected boolean emitMetricsJson(String connectUrl, String jsonData) {
    int timeout = getTimeoutSeconds() * 1000;
    HttpURLConnection connection = null;
    try {
      if (connectUrl == null) {
        throw new IOException("Unknown URL. Unable to connect to metrics collector.");
      }
      connection = connectUrl.startsWith("https") ?
          getSSLConnection(connectUrl) : getConnection(connectUrl);

      if (LOG.isDebugEnabled()) {
        LOG.debug("emitMetricsJson to " + connectUrl + ", " + jsonData);
      }
      AppCookieManager appCookieManager = getAppCookieManager();
      String appCookie = appCookieManager.getCachedAppCookie(connectUrl);
      if (appCookie != null) {
        if (LOG.isInfoEnabled()) {
          LOG.info("Using cached app cookie for URL:" + connectUrl);
        }
        connection.setRequestProperty(COOKIE, appCookie);
      }

      int statusCode = emitMetricsJson(connection, timeout, jsonData);

      if (statusCode == HttpStatus.SC_UNAUTHORIZED ) {
        String wwwAuthHeader = connection.getHeaderField(WWW_AUTHENTICATE);
        if (LOG.isInfoEnabled()) {
          LOG.info("Received WWW-Authentication header:" + wwwAuthHeader + ", for URL:" + connectUrl);
        }
        if (wwwAuthHeader != null && wwwAuthHeader.trim().startsWith(NEGOTIATE)) {
          appCookie = appCookieManager.getAppCookie(connectUrl, true);
          if (appCookie != null) {
            cleanupInputStream(connection.getInputStream());
            connection = connectUrl.startsWith("https") ?
                getSSLConnection(connectUrl) : getConnection(connectUrl);
            connection.setRequestProperty(COOKIE, appCookie);
            statusCode = emitMetricsJson(connection, timeout, jsonData);
          }
        } else {
          // no supported authentication type found
          // we would let the original response propagate
          LOG.error("Unsupported WWW-Authentication header:" + wwwAuthHeader+ ", for URL:" + connectUrl);
        }
      }

      if (statusCode != 200) {
        LOG.info("Unable to POST metrics to collector, " + connectUrl + ", " +
            "statusCode = " + statusCode);
      } else {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Metrics posted to Collector " + connectUrl);
        }
      }
      cleanupInputStream(connection.getInputStream());
      // reset failedCollectorConnectionsCounter to "0"
      failedCollectorConnectionsCounter.set(0);
      return true;
    } catch (IOException ioe) {
      StringBuilder errorMessage =
          new StringBuilder("Unable to connect to collector, " + connectUrl + "\n"
                  + "This exceptions will be ignored for next " + NUMBER_OF_SKIPPED_COLLECTOR_EXCEPTIONS + " times\n");
      try {
        if ((connection != null)) {
          errorMessage.append(cleanupInputStream(connection.getErrorStream()));
        }
      } catch (IOException e) {
        //NOP
      }

      if (failedCollectorConnectionsCounter.getAndIncrement() == 0) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(errorMessage, ioe);
        } else {
          LOG.info(errorMessage);
        }
        throw new UnableToConnectException(ioe).setConnectUrl(connectUrl);
      } else {
        failedCollectorConnectionsCounter.compareAndSet(NUMBER_OF_SKIPPED_COLLECTOR_EXCEPTIONS, 0);
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format("Ignoring %s AMS connection exceptions", NUMBER_OF_SKIPPED_COLLECTOR_EXCEPTIONS));
        }
        return false;
      }
    }
  }