YARN-5382. RM does not audit log kill request for active applications. Contributed by Vrushali C

(cherry picked from commit 5c95bb315ba605b3bed77966a99a63854234e2c5)

Conflicts:

	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java
	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAuditLogger.java

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.server.resourcemanager;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.security.AccessControlException;
 import java.text.MessageFormat;
@@ -147,8 +148,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSyst
 import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystemUtil;
 import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppKillByClientEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppMoveEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
@@ -747,15 +747,18 @@ public class ClientRMService extends AbstractService implements
     }
 
     if (application.isAppFinalStateStored()) {
-      RMAuditLogger.logSuccess(callerUGI.getShortUserName(),
-          AuditConstants.KILL_APP_REQUEST, "ClientRMService", applicationId,
-          callerContext);
       return KillApplicationResponse.newInstance(true);
     }
 
-    this.rmContext.getDispatcher().getEventHandler().handle(
-        new RMAppEvent(applicationId, RMAppEventType.KILL,
-        "Application killed by user."));
+    String message = "Kill application " + applicationId + " received from "
+        + callerUGI;
+    InetAddress remoteAddress = Server.getRemoteIp();
+    if (null != remoteAddress) {
+      message += " at " + remoteAddress.getHostAddress();
+    }
+    this.rmContext.getDispatcher().getEventHandler()
+        .handle(new RMAppKillByClientEvent(applicationId, message, callerUGI,
+            remoteAddress));
 
     // For UnmanagedAMs, return true so they don't retry
     return KillApplicationResponse.newInstance(

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java

@@ -76,12 +76,12 @@ public class RMAuditLogger {
     public static final String LIST_RESERVATION_REQUEST = "List " +
             "Reservation Request";
   }
-  
+
   static String createSuccessLog(String user, String operation, String target,
       ApplicationId appId, ApplicationAttemptId attemptId,
       ContainerId containerId) {
     return createSuccessLog(user, operation, target, appId, attemptId,
-        containerId, null);
+        containerId, null, Server.getRemoteIp());
   }
 
   /**
@@ -89,10 +89,13 @@ public class RMAuditLogger {
    */
   static String createSuccessLog(String user, String operation, String target,
       ApplicationId appId, ApplicationAttemptId attemptId,
-      ContainerId containerId, CallerContext callerContext) {
+      ContainerId containerId, CallerContext callerContext,
+      InetAddress ip) {
     StringBuilder b = new StringBuilder();
     start(Keys.USER, user, b);
-    addRemoteIP(b);
+    if (ip != null) {
+      add(Keys.IP, ip.getHostAddress(), b);
+    }
     add(Keys.OPERATION, operation, b);
     add(Keys.TARGET, target ,b);
     add(Keys.RESULT, AuditConstants.SUCCESS, b);
@@ -178,10 +181,37 @@ public class RMAuditLogger {
       ApplicationId appId, CallerContext callerContext) {
     if (LOG.isInfoEnabled()) {
       LOG.info(createSuccessLog(user, operation, target, appId, null, null,
-          callerContext));
+          callerContext, Server.getRemoteIp()));
     }
   }
 
+  /**
+   * Create a readable and parseable audit log string for a successful event.
+   *
+   * @param user
+   *          User who made the service request to the ResourceManager.
+   * @param operation
+   *          Operation requested by the user.
+   * @param target
+   *          The target on which the operation is being performed.
+   * @param appId
+   *          Application Id in which operation was performed.
+   * @param ip
+   *          The ip address of the caller.
+   *
+   *          <br>
+   *          <br>
+   *          Note that the {@link RMAuditLogger} uses tabs ('\t') as a key-val
+   *          delimiter and hence the value fields should not contains tabs
+   *          ('\t').
+   */
+  public static void logSuccess(String user, String operation, String target,
+      ApplicationId appId, InetAddress ip) {
+    if (LOG.isInfoEnabled()) {
+      LOG.info(createSuccessLog(user, operation, target, appId, null, null,
+          null, ip));
+    }
+  }
 
   /**
    * Create a readable and parseable audit log string for a successful event.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.resourcemanager.rmapp;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
@@ -74,6 +75,8 @@ import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
 import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
 import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEventType;
+import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger;
+import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.RMServerUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.blacklist.BlacklistManager;
@@ -819,7 +822,7 @@ public class RMAppImpl implements RMApp, Recoverable {
 
       if (oldState != getState()) {
         LOG.info(appID + " State change from " + oldState + " to "
-            + getState());
+            + getState() + " on event=" + event.getType());
       }
     } finally {
       this.writeLock.unlock();
@@ -1241,6 +1244,25 @@ public class RMAppImpl implements RMApp, Recoverable {
     };
   }
 
+  /**
+   * Log the audit event for kill by client.
+   *
+   * @param event
+   *          The {@link RMAppEvent} to be logged
+   */
+  static void auditLogKillEvent(RMAppEvent event) {
+    if (event instanceof RMAppKillByClientEvent) {
+      RMAppKillByClientEvent killEvent = (RMAppKillByClientEvent) event;
+      UserGroupInformation callerUGI = killEvent.getCallerUGI();
+      String userName = null;
+      if (callerUGI != null) {
+        userName = callerUGI.getShortUserName();
+      }
+      InetAddress remoteIP = killEvent.getIp();
+      RMAuditLogger.logSuccess(userName, AuditConstants.KILL_APP_REQUEST,
+          "RMAppImpl", event.getApplicationId(), remoteIP);
+    }
+  }
 
   private static class AppKilledTransition extends FinalTransition {
     public AppKilledTransition() {
@@ -1251,6 +1273,7 @@ public class RMAppImpl implements RMApp, Recoverable {
     public void transition(RMAppImpl app, RMAppEvent event) {
       app.diagnostics.append(event.getDiagnosticMsg());
       super.transition(app, event);
+      RMAppImpl.auditLogKillEvent(event);
     };
   }
 
@@ -1264,6 +1287,7 @@ public class RMAppImpl implements RMApp, Recoverable {
       app.handler.handle(
           new RMAppAttemptEvent(app.currentAttempt.getAppAttemptId(),
               RMAppAttemptEventType.KILL, event.getDiagnosticMsg()));
+      RMAppImpl.auditLogKillEvent(event);
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppKillByClientEvent.java

@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.rmapp;
+
+import java.net.InetAddress;
+
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+
+/**
+ * An event class that is used to help with logging information
+ * when an application KILL event is needed.
+ *
+ */
+public class RMAppKillByClientEvent extends RMAppEvent {
+
+  private final UserGroupInformation callerUGI;
+  private final InetAddress ip;
+
+  /**
+   * constructor to create an event used for logging during user driven kill
+   * invocations.
+   *
+   * @param appId application id
+   * @param diagnostics message about the kill event
+   * @param callerUGI caller's user and group information
+   * @param remoteIP ip address of the caller
+   */
+  public RMAppKillByClientEvent(ApplicationId appId, String diagnostics,
+      UserGroupInformation callerUGI, InetAddress remoteIP) {
+    super(appId, RMAppEventType.KILL, diagnostics);
+    this.callerUGI = callerUGI;
+    this.ip = remoteIP;
+  }
+
+  /**
+   * returns the {@link UserGroupInformation} information.
+   * @return UserGroupInformation
+   */
+  public final UserGroupInformation getCallerUGI() {
+    return callerUGI;
+  }
+
+  /**
+   * returns the ip address stored in this event.
+   * @return remoteIP
+   */
+  public final InetAddress getIp() {
+    return ip;
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAuditLogger.java

@@ -23,6 +23,7 @@ import static org.mockito.Mockito.when;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 
 import com.google.protobuf.BlockingService;
 import com.google.protobuf.RpcController;
@@ -105,18 +106,37 @@ public class TestRMAuditLogger {
     testSuccessLogFormatHelper(checkIP, appId, attemptId, containerId, null);
   }
 
+  private void testSuccessLogFormatHelper(boolean checkIP, ApplicationId appId,
+      ApplicationAttemptId attemptId, ContainerId containerId,
+      CallerContext callerContext) {
+    testSuccessLogFormatHelper(checkIP, appId, attemptId, containerId,
+        callerContext, Server.getRemoteIp());
+  }
+
   /**
    * Test the AuditLog format for successful events.
    */
   private void testSuccessLogFormatHelper(boolean checkIP, ApplicationId appId,
       ApplicationAttemptId attemptId, ContainerId containerId,
-      CallerContext callerContext) {
-    String sLog = RMAuditLogger.createSuccessLog(USER, OPERATION, TARGET,
-        appId, attemptId, containerId, callerContext);
+      CallerContext callerContext, InetAddress remoteIp) {
+
+    String sLog;
+    if (checkIP) {
+      sLog = RMAuditLogger.createSuccessLog(USER, OPERATION, TARGET, appId,
+          attemptId, containerId, callerContext, remoteIp);
+    } else {
+      sLog = RMAuditLogger.createSuccessLog(USER, OPERATION, TARGET, appId,
+          attemptId, containerId, callerContext, null);
+    }
     StringBuilder expLog = new StringBuilder();
     expLog.append("USER=test\t");
     if (checkIP) {
-      InetAddress ip = Server.getRemoteIp();
+      InetAddress ip;
+      if(remoteIp != null) {
+        ip = remoteIp;
+      } else {
+        ip = Server.getRemoteIp();
+      }
       expLog.append(Keys.IP.name() + "=" + ip.getHostAddress() + "\t");
     }
     expLog.append("OPERATION=oper\tTARGET=tgt\tRESULT=SUCCESS");
@@ -141,6 +161,13 @@ public class TestRMAuditLogger {
     assertEquals(expLog.toString(), sLog);
   }
 
+  private void testSuccessLogFormatHelperWithIP(boolean checkIP,
+      ApplicationId appId, ApplicationAttemptId attemptId,
+      ContainerId containerId, InetAddress ip) {
+    testSuccessLogFormatHelper(checkIP, appId, attemptId, containerId, null,
+        ip);
+  }
+
   /**
    * Test the AuditLog format for successful events passing nulls.
    */
@@ -157,6 +184,33 @@ public class TestRMAuditLogger {
     assertEquals(expLog.toString(), sLog);
   }
 
+  /**
+   * Tests the SuccessLog with two IP addresses.
+   *
+   * @param checkIP
+   * @param appId
+   * @param attemptId
+   * @param containerId
+   */
+  private void testSuccessLogFormatHelperWithIP(boolean checkIP,
+      ApplicationId appId, ApplicationAttemptId attemptId,
+      ContainerId containerId) {
+    testSuccessLogFormatHelperWithIP(checkIP, appId, attemptId, containerId,
+        InetAddress.getLoopbackAddress());
+    byte[] ipAddr = new byte[] {100, 10, 10, 1};
+
+    InetAddress addr = null;
+    try {
+      addr = InetAddress.getByAddress(ipAddr);
+    } catch (UnknownHostException uhe) {
+      // should not happen as long as IP address format
+      // stays the same
+      Assert.fail("Check ip address being constructed");
+    }
+    testSuccessLogFormatHelperWithIP(checkIP, appId, attemptId, containerId,
+        addr);
+  }
+
   /**
    * Test the AuditLog format for successful events with the various
    * parameters.
@@ -180,6 +234,7 @@ public class TestRMAuditLogger {
     testSuccessLogFormatHelper(checkIP, APPID, ATTEMPTID, CONTAINERID,
         new CallerContext.Builder(CALLER_CONTEXT).setSignature(CALLER_SIGNATURE)
             .build());
+    testSuccessLogFormatHelperWithIP(checkIP, APPID, ATTEMPTID, CONTAINERID);
     testSuccessLogNulls(checkIP);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java

@@ -37,6 +37,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.ipc.Server;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -254,7 +255,7 @@ public class TestRMAppTransitions {
 
     ApplicationMasterService masterService =
         new ApplicationMasterService(rmContext, scheduler);
-    
+
     if(submissionContext == null) {
       submissionContext = new ApplicationSubmissionContextPBImpl();
     }
@@ -302,7 +303,7 @@ public class TestRMAppTransitions {
   private static void assertStartTimeSet(RMApp application) {
     Assert.assertTrue("application start time is not greater than 0",
         application.getStartTime() > 0);
-    Assert.assertTrue("application start time is before currentTime", 
+    Assert.assertTrue("application start time is before currentTime",
         application.getStartTime() <= System.currentTimeMillis());
   }
 
@@ -321,7 +322,7 @@ public class TestRMAppTransitions {
     assertStartTimeSet(application);
     Assert.assertTrue("application finish time is not greater than 0",
         (application.getFinishTime() > 0));
-    Assert.assertTrue("application finish time is not >= than start time",
+    Assert.assertTrue("application finish time is not >= start time",
         (application.getFinishTime() >= application.getStartTime()));
   }
 
@@ -546,11 +547,14 @@ public class TestRMAppTransitions {
   public void testAppNewKill() throws IOException {
     LOG.info("--- START: testAppNewKill ---");
 
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppNewKill", new String[] {"foo_group"});
+
     RMApp application = createNewTestApp(null);
     // NEW => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
     application.handle(event);
     rmDispatcher.await();
     sendAppUpdateSavedEvent(application);
@@ -601,9 +605,13 @@ public class TestRMAppTransitions {
 
     RMApp application = testCreateAppNewSaving(null);
     // NEW_SAVING => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppNewSavingKill", new String[] {"foo_group"});
+
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
+
     application.handle(event);
     rmDispatcher.await();
     sendAppUpdateSavedEvent(application);
@@ -650,10 +658,15 @@ public class TestRMAppTransitions {
   public void testAppSubmittedKill() throws IOException, InterruptedException {
     LOG.info("--- START: testAppSubmittedKill---");
     RMApp application = testCreateAppSubmittedNoRecovery(null);
+
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppSubmittedKill", new String[] {"foo_group"});
+
     // SUBMITTED => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
+
     application.handle(event);
     rmDispatcher.await();
     sendAppUpdateSavedEvent(application);
@@ -703,9 +716,13 @@ public class TestRMAppTransitions {
     LOG.info("--- START: testAppAcceptedKill ---");
     RMApp application = testCreateAppAccepted(null);
     // ACCEPTED => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppAcceptedKill", new String[] {"foo_group"});
+
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
+
     application.handle(event);
     rmDispatcher.await();
 
@@ -751,9 +768,14 @@ public class TestRMAppTransitions {
 
     RMApp application = testCreateAppRunning(null);
     // RUNNING => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppRunningKill", new String[] {"foo_group"});
+
+    // SUBMITTED => KILLED event RMAppEventType.KILL
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
+
     application.handle(event);
     rmDispatcher.await();
 
@@ -917,9 +939,14 @@ public class TestRMAppTransitions {
     RMApp application = testCreateAppRunning(null);
 
     // RUNNING => KILLED event RMAppEventType.KILL
-    RMAppEvent event =
-        new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL,
-        "Application killed by user.");
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+        "fooTestAppKilledKill", new String[] {"foo_group"});
+
+    // SUBMITTED => KILLED event RMAppEventType.KILL
+    RMAppEvent event = new RMAppKillByClientEvent(
+        application.getApplicationId(), "Application killed by user.", fooUser,
+        Server.getRemoteIp());
+
     application.handle(event);
     rmDispatcher.await();
     sendAttemptUpdateSavedEvent(application);