AMBARI-14384 Ambari Metrics doesn't use SPNEGO to authenticate (dsen)

ambari-metrics/ambari-metrics-common/pom.xml

@@ -72,6 +72,16 @@
       <artifactId>hadoop-annotations</artifactId>
       <version>2.6.0</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-auth</artifactId>
+      <version>2.7.3</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>2.7.3</version>
+    </dependency>
     <dependency>
       <groupId>org.codehaus.jackson</groupId>
       <artifactId>jackson-mapper-asl</artifactId>

ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java

@@ -19,11 +19,14 @@ package org.apache.hadoop.metrics2.sink.timeline;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authentication.client.AuthenticationException;
 import org.codehaus.jackson.map.AnnotationIntrospector;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.annotate.JsonSerialize;
 import org.codehaus.jackson.xc.JaxbAnnotationIntrospector;
 
+import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
 import javax.net.ssl.HttpsURLConnection;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLSocketFactory;
@@ -140,9 +143,11 @@ public abstract class AbstractTimelineMetricsSink {
         if (LOG.isDebugEnabled()) {
           LOG.debug(String.format("Ignoring %s AMS connection exceptions", NUMBER_OF_SKIPPED_COLLECTOR_EXCEPTIONS));
         }
-        return false;
       }
+    } catch (AuthenticationException e) {
+      LOG.error("AuthenticationException while posting metrics to metrics collector", e);
     }
+    return false;
   }
 
   protected boolean emitMetrics(TimelineMetrics metrics) {
@@ -186,18 +191,43 @@ public abstract class AbstractTimelineMetricsSink {
     }
     return sb.toString();
   }
+  /**
+   * Uses UGI to find and renew (if needed) kerberos ticket and opens authenticated connection
+   *
+   * @param spec - requested URL
+   * @param userGroupInformation
+   * @return HttpURLConnection - opened authenticated connection
+   * @throws IOException
+   * @throws AuthenticationException
+   */
+  protected HttpURLConnection getAuthenticatedConnection(String spec, UserGroupInformation userGroupInformation) throws IOException, AuthenticationException {
+    //renew ticked if needed
+    if (userGroupInformation != null) {
+      userGroupInformation.checkTGTAndReloginFromKeytab();
+    }
+    //open authenticated connection
+    AuthenticatedURL.Token token = new AuthenticatedURL.Token();
+    URL url = new URL(spec);
+    return new AuthenticatedURL().openConnection(url, token);
+  }
 
   // Get a connection
-  protected HttpURLConnection getConnection(String spec) throws IOException {
-    return (HttpURLConnection) new URL(spec).openConnection();
+  protected HttpURLConnection getConnection(String spec) throws IOException, AuthenticationException {
+    UserGroupInformation userGroupInformation = UserGroupInformation.getLoginUser();
+
+    if (UserGroupInformation.isSecurityEnabled() && userGroupInformation != null &&
+            userGroupInformation.getAuthenticationMethod() == UserGroupInformation.AuthenticationMethod.KERBEROS) {
+      return getAuthenticatedConnection(spec, userGroupInformation);
+    } else {
+      return (HttpURLConnection) new URL(spec).openConnection();
+    }
   }
 
   // Get an ssl connection
   protected HttpsURLConnection getSSLConnection(String spec)
-    throws IOException, IllegalStateException {
+          throws IOException, IllegalStateException, AuthenticationException {
 
-    HttpsURLConnection connection = (HttpsURLConnection) (new URL(spec)
-      .openConnection());
+    HttpsURLConnection connection = (HttpsURLConnection) getConnection(spec);
 
     connection.setSSLSocketFactory(sslSocketFactory);
 

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/configuration/hadoop-metrics2.properties.xml

@@ -82,25 +82,22 @@ resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue
 *.sink.timeline.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
 *.sink.timeline.period={{metrics_collection_period}}
 *.sink.timeline.sendInterval={{metrics_report_interval}}000
-*.sink.timeline.slave.host.name={{hostname}}
-*.sink.timeline.zookeeper.quorum={{zookeeper_quorum}}
-*.sink.timeline.protocol={{metric_collector_protocol}}
-*.sink.timeline.port={{metric_collector_port}}
+*.sink.timeline.slave.host.name = {{hostname}}
 
 # HTTPS properties
 *.sink.timeline.truststore.path = {{metric_truststore_path}}
 *.sink.timeline.truststore.type = {{metric_truststore_type}}
 *.sink.timeline.truststore.password = {{metric_truststore_password}}
 
-datanode.sink.timeline.collector={{metric_collector_hosts}}
-namenode.sink.timeline.collector={{metric_collector_hosts}}
-resourcemanager.sink.timeline.collector={{metric_collector_hosts}}
-nodemanager.sink.timeline.collector={{metric_collector_hosts}}
-jobhistoryserver.sink.timeline.collector={{metric_collector_hosts}}
-journalnode.sink.timeline.collector={{metric_collector_hosts}}
-maptask.sink.timeline.collector={{metric_collector_hosts}}
-reducetask.sink.timeline.collector={{metric_collector_hosts}}
-applicationhistoryserver.sink.timeline.collector={{metric_collector_hosts}}
+datanode.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+namenode.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+resourcemanager.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+nodemanager.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+jobhistoryserver.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+journalnode.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+maptask.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+reducetask.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+applicationhistoryserver.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
 
 resourcemanager.sink.timeline.tagsForPrefix.yarn=Queue