Added Framework Authentication.

Modified `MesosScheduler.java` and `configuration.md`.  Now `mapred.mesos.framework.principal`, `mapred.mesos.framework.secretfile`, `mapred.mesos.framework.user`, and `mapred.mesos.framework.name` are configureable options. Addresses issue #53

Added Support for Framework Authentication

Added Support for Framework Authentication

Author: DarinJ

configuration.md

@@ -141,6 +141,36 @@ default values.
       role configured in "mapred.mesos.role".
     </description>
   </property>
+  <property>
+    <name>mapred.mesos.framework.name</name>
+    <value>hadoop</value>
+    <description>
+      This is the Mesos framework name.  Defaults to Hadoop plus port information.
+    </description>
+  </property>
+  <property>
+    <name>mapred.mesos.framework.principal</name>
+    <value>hadoop</value>
+    <description>
+      This is the Mesos framework principal. It is used for framework authentication.
+      Consult the Mesos documentation for details.
+    </description>
+  </property>
+  <property>
+    <name>mapred.mesos.framework.secretfile</name>
+    <value>/location/secretfile</value>
+    <description>
+      Location of the file holding the Mesos framework secret. It is used for framework authentication.
+      Consult the Mesos documentation for details.  Caution: avoid newline characters, some editor place these before end of file.
+    </description>
+  </property>
+  <property>
+    <name>mapred.mesos.framework.user</name>
+    <value>hadoop</value>
+    <description>
+      This is the user the Mesos framework runs as. If left unset, it defaults to the user running the scheduler.
+    </description>
+  </property>
 
   <!-- If you're using a custom Mesos Containerizer -->
   <property>

src/main/java/org/apache/hadoop/mapred/MesosScheduler.java

@@ -1,6 +1,7 @@
 package org.apache.hadoop.mapred;
 
 import com.codahale.metrics.Meter;
+import com.google.protobuf.ByteString;
 import org.apache.commons.httpclient.HttpHost;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -14,6 +15,7 @@
 import org.apache.mesos.hadoop.Metrics;
 
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -364,10 +366,35 @@ public synchronized void start() throws IOException {
     String master = conf.get("mapred.mesos.master", "local");
 
     try {
-      FrameworkInfo frameworkInfo = FrameworkInfo.newBuilder().setUser("") // Let Mesos fill in the user.
-          .setCheckpoint(conf.getBoolean("mapred.mesos.checkpoint", false)).setRole(conf.get("mapred.mesos.role", "*")).setName("Hadoop: (RPC port: " + jobTracker.port + "," + " WebUI port: " + jobTracker.infoPort + ")").build();
-
-      driver = new MesosSchedulerDriver(this, frameworkInfo, master);
+      FrameworkInfo frameworkInfo;
+      FrameworkInfo.Builder frameworkInfoBuilder = FrameworkInfo.newBuilder()
+              .setUser(conf.get("mapred.mesos.framework.user", "")) // Let Mesos fill in the user.
+              .setCheckpoint(conf.getBoolean("mapred.mesos.checkpoint", false))
+              .setRole(conf.get("mapred.mesos.role", "*"))
+              .setName(conf.get("mapred.mesos.framework.name", "Hadoop: (RPC port: " + jobTracker.port + ","
+                      + " WebUI port: " + jobTracker.infoPort + ")"));
+
+      Credential credential = null;
+
+      String frameworkPrincipal = conf.get("mapred.mesos.framework.principal");
+      if (frameworkPrincipal != null) {
+        frameworkInfoBuilder.setPrincipal(frameworkPrincipal);
+        String secretFile = conf.get("mapred.mesos.framework.secretfile");
+        if (secretFile != null) {
+          credential = Credential.newBuilder()
+                  .setSecret(ByteString.readFrom(new FileInputStream(secretFile)))
+                  .setPrincipal(frameworkPrincipal)
+                  .build();
+        }
+      }
+      if (credential == null) {
+        LOG.info("Creating Schedule Driver");
+        driver = new MesosSchedulerDriver(this, frameworkInfoBuilder.build(), master);
+      } else {
+        LOG.info("Creatingg Schedule Driver, attempting to authenticate with Principal: " + credential.getPrincipal()
+                + ", secret:" + credential.getSecret());
+        driver = new MesosSchedulerDriver(this, frameworkInfoBuilder.build(), master, credential);
+      }
       driver.start();
     } catch (Exception e) {
       // If the MesosScheduler can't be loaded, the JobTracker won't be useful