/**
 * 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.scheduler.fair;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.security.Groups;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.w3c.dom.Element;
import org.w3c.dom.NamedNodeMap;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

import com.google.common.annotations.VisibleForTesting;

@Private
@Unstable
public abstract class QueuePlacementRule {
  protected boolean create;
  public static final Log LOG =
      LogFactory.getLog(QueuePlacementRule.class.getName());

  /**
   * Initializes the rule with any arguments.
   * 
   * @param args
   *    Additional attributes of the rule's xml element other than create.
   */
  public QueuePlacementRule initialize(boolean create, Map<String, String> args) {
    this.create = create;
    return this;
  }
  
  /**
   * 
   * @param requestedQueue
   *    The queue explicitly requested.
   * @param user
   *    The user submitting the app.
   * @param groups
   *    The groups of the user submitting the app.
   * @param configuredQueues
   *    The queues specified in the scheduler configuration.
   * @return
   *    The queue to place the app into. An empty string indicates that we should
   *    continue to the next rule, and null indicates that the app should be rejected.
   */
  public String assignAppToQueue(String requestedQueue, String user,
      Groups groups, Map<FSQueueType, Set<String>> configuredQueues)
      throws IOException {
   String queue = getQueueForApp(requestedQueue, user, groups,
        configuredQueues);
    if (create || configuredQueues.get(FSQueueType.LEAF).contains(queue)
        || configuredQueues.get(FSQueueType.PARENT).contains(queue)) {
      return queue;
    } else {
      return "";
    }
  }
  
  public void initializeFromXml(Element el)
      throws AllocationConfigurationException {
    boolean create = true;
    NamedNodeMap attributes = el.getAttributes();
    Map<String, String> args = new HashMap<String, String>();
    for (int i = 0; i < attributes.getLength(); i++) {
      Node node = attributes.item(i);
      String key = node.getNodeName();
      String value = node.getNodeValue();
      if (key.equals("create")) {
        create = Boolean.parseBoolean(value);
      } else {
        args.put(key, value);
      }
    }
    initialize(create, args);
  }
  
  /**
   * Returns true if this rule never tells the policy to continue.
   */
  public abstract boolean isTerminal();
  
  /**
   * Applies this rule to an app with the given requested queue and user/group
   * information.
   * 
   * @param requestedQueue
   *    The queue specified in the ApplicationSubmissionContext
   * @param user
   *    The user submitting the app.
   * @param groups
   *    The groups of the user submitting the app.
   * @return
   *    The name of the queue to assign the app to, or null to empty string
   *    continue to the next rule.
   */
  protected abstract String getQueueForApp(String requestedQueue, String user,
      Groups groups, Map<FSQueueType, Set<String>> configuredQueues)
      throws IOException;

  /**
   * Places apps in queues by username of the submitter
   */
  public static class User extends QueuePlacementRule {
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues) {
      return "root." + cleanName(user);
    }
    
    @Override
    public boolean isTerminal() {
      return create;
    }
  }
  
  /**
   * Places apps in queues by primary group of the submitter
   */
  public static class PrimaryGroup extends QueuePlacementRule {
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues)
        throws IOException {
      final List<String> groupList = groups.getGroups(user);
      if (groupList.isEmpty()) {
        throw new IOException("No groups returned for user " + user);
      }
      return "root." + cleanName(groupList.get(0));
    }
    
    @Override
    public boolean isTerminal() {
      return create;
    }
  }
  
  /**
   * Places apps in queues by secondary group of the submitter
   * 
   * Match will be made on first secondary group that exist in
   * queues
   */
  public static class SecondaryGroupExistingQueue extends QueuePlacementRule {
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues)
        throws IOException {
      List<String> groupNames = groups.getGroups(user);
      for (int i = 1; i < groupNames.size(); i++) {
        String group = cleanName(groupNames.get(i));
        if (configuredQueues.get(FSQueueType.LEAF).contains("root." + group)
            || configuredQueues.get(FSQueueType.PARENT).contains(
                "root." + group)) {
          return "root." + group;
        }
      }
      
      return "";
    }
        
    @Override
    public boolean isTerminal() {
      return false;
    }
  }

  /**
   * Places apps in queues with name of the submitter under the queue
   * returned by the nested rule.
   */
  public static class NestedUserQueue extends QueuePlacementRule {
    @VisibleForTesting
    QueuePlacementRule nestedRule;

    /**
     * Parse xml and instantiate the nested rule 
     */
    @Override
    public void initializeFromXml(Element el)
        throws AllocationConfigurationException {
      NodeList elements = el.getChildNodes();

      for (int i = 0; i < elements.getLength(); i++) {
        Node node = elements.item(i);
        if (node instanceof Element) {
          Element element = (Element) node;
          if ("rule".equals(element.getTagName())) {
            QueuePlacementRule rule = QueuePlacementPolicy
                .createAndInitializeRule(node);
            if (rule == null) {
              throw new AllocationConfigurationException(
                  "Unable to create nested rule in nestedUserQueue rule");
            }
            this.nestedRule = rule;
            break;
          } else {
            continue;
          }
        }
      }

      if (this.nestedRule == null) {
        throw new AllocationConfigurationException(
            "No nested rule specified in <nestedUserQueue> rule");
      }
      super.initializeFromXml(el);
    }
    
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues)
        throws IOException {
      // Apply the nested rule
      String queueName = nestedRule.assignAppToQueue(requestedQueue, user,
          groups, configuredQueues);
      
      if (queueName != null && queueName.length() != 0) {
        if (!queueName.startsWith("root.")) {
          queueName = "root." + queueName;
        }
        
        // Verify if the queue returned by the nested rule is an configured leaf queue,
        // if yes then skip to next rule in the queue placement policy
        if (configuredQueues.get(FSQueueType.LEAF).contains(queueName)) {
          return "";
        }
        return queueName + "." + cleanName(user);
      }
      return queueName;
    }

    @Override
    public boolean isTerminal() {
      return false;
    }
  }
  
  /**
   * Places apps in queues by requested queue of the submitter
   */
  public static class Specified extends QueuePlacementRule {
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues) {
      if (requestedQueue.equals(YarnConfiguration.DEFAULT_QUEUE_NAME)) {
        return "";
      } else {
        if (!requestedQueue.startsWith("root.")) {
          requestedQueue = "root." + requestedQueue;
        }
        return requestedQueue;
      }
    }
    
    @Override
    public boolean isTerminal() {
      return false;
    }
  }
  
  /**
   * Places apps in the specified default queue. If no default queue is
   * specified the app is placed in root.default queue.
   */
  public static class Default extends QueuePlacementRule {
    @VisibleForTesting
    String defaultQueueName;

    @Override
    public QueuePlacementRule initialize(boolean create,
        Map<String, String> args) {
      if (defaultQueueName == null) {
        defaultQueueName = "root." + YarnConfiguration.DEFAULT_QUEUE_NAME;
      }
      return super.initialize(create, args);
    }
    
    @Override
    public void initializeFromXml(Element el)
        throws AllocationConfigurationException {
      defaultQueueName = el.getAttribute("queue");
      if (defaultQueueName != null && !defaultQueueName.isEmpty()) {
        if (!defaultQueueName.startsWith("root.")) {
          defaultQueueName = "root." + defaultQueueName;
        }
      } else {
        defaultQueueName = "root." + YarnConfiguration.DEFAULT_QUEUE_NAME;
      }
      super.initializeFromXml(el);
    }

    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues) {
      return defaultQueueName;
    }

    @Override
    public boolean isTerminal() {
      return true;
    }
  }

  /**
   * Rejects all apps
   */
  public static class Reject extends QueuePlacementRule {
    @Override
    public String assignAppToQueue(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues) {
      return null;
    }
    
    @Override
    protected String getQueueForApp(String requestedQueue, String user,
        Groups groups, Map<FSQueueType, Set<String>> configuredQueues) {
      throw new UnsupportedOperationException();
    }
    
    @Override
    public boolean isTerminal() {
      return true;
    }
  }

  /**
   * Replace the periods in the username or groupname with "_dot_" and
   * remove trailing and leading whitespace.
   */
  protected String cleanName(String name) {
    name = name.trim();
    if (name.contains(".")) {
      String converted = name.replaceAll("\\.", "_dot_");
      LOG.warn("Name " + name + " is converted to " + converted
          + " when it is used as a queue name.");
      return converted;
    } else {
      return name;
    }
  }
}