<commit_before>package com.twitter.mesos.scheduler.configuration;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;

import javax.annotation.Nullable;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

import org.apache.commons.lang.StringUtils;

import com.twitter.common.args.Arg;
import com.twitter.common.args.CmdLine;
import com.twitter.common.args.constraints.Positive;
import com.twitter.common.base.MorePreconditions;
import com.twitter.mesos.Tasks;
import com.twitter.mesos.gen.Constraint;
import com.twitter.mesos.gen.Identity;
import com.twitter.mesos.gen.JobConfiguration;
import com.twitter.mesos.gen.LimitConstraint;
import com.twitter.mesos.gen.TaskConstraint;
import com.twitter.mesos.gen.TwitterTaskInfo;
import com.twitter.mesos.gen.TwitterTaskInfo._Fields;
import com.twitter.mesos.gen.ValueConstraint;
import com.twitter.mesos.scheduler.CommandLineExpander;
import com.twitter.mesos.scheduler.ThermosJank;
import com.twitter.mesos.scheduler.configuration.ValueParser.ParseException;

/**
 * Manages translation from a string-mapped configuration to a concrete configuration type, and
 * defaults for optional values.
 *
 * TODO(<NAME>): Add input validation to all fields (strings not empty, positive ints, etc).
 */
public final class ConfigurationManager {

  public static final String DEDICATED_ATTRIBUTE = "dedicated";

  @VisibleForTesting public static final String HOST_CONSTRAINT = "host";
  @VisibleForTesting public static final String RACK_CONSTRAINT = "rack";
  @VisibleForTesting public static final String START_COMMAND_FIELD = "start_command";

  @VisibleForTesting
  @Positive
  @CmdLine(name = "max_tasks_per_job", help = "Maximum number of allowed tasks in a single job.")
  public static final Arg<Integer> MAX_TASKS_PER_JOB = Arg.create(500);

  @CmdLine(name = "constrain_legacy_executor",
      help = "If true, only allow the legacy executor to run on specially-tagged machines.")
  public static final Arg<Boolean> CONSTRAIN_LEGACY_EXECUTOR = Arg.create(true);

  @CmdLine(name = "require_contact_email",
      help = "If true, reject jobs that do not specify a contact email address.")
  public static final Arg<Boolean> REQUIRE_CONTACT_EMAIL = Arg.create(true);

  private static final Logger LOG = Logger.getLogger(ConfigurationManager.class.getName());

  private static final Pattern GOOD_IDENTIFIER_PATTERN = Pattern.compile("[\\w\\-\\.]+");

  private static final int MAX_IDENTIFIER_LENGTH = 255;

  private interface FieldSanitizer {
    void sanitize(TwitterTaskInfo task) throws TaskDescriptionException;
  }

  private static class DefaultField implements FieldSanitizer {
    private final _Fields field;
    private final Object defaultValue;

    DefaultField(_Fields field, Object defaultValue) {
      this.field = field;
      this.defaultValue = defaultValue;
    }

    @Override public void sanitize(TwitterTaskInfo task) {
      if (!task.isSet(field)) {
        task.setFieldValue(field, defaultValue);
      }
    }
  }

  private static class RequiredField implements FieldSanitizer {
    private final _Fields field;

    RequiredField(_Fields field) {
      this.field = field;
    }

    @Override public void sanitize(TwitterTaskInfo task) throws TaskDescriptionException {
      if (!task.isSet(field)) {
        throw new TaskDescriptionException("Field " + field.getFieldName() + " is required.");
      }
    }
  }

  public static final String EXECUTOR_CONSTRAINT = "executor";
  public static final String LEGACY_EXECUTOR_VALUE = "legacy";
  @VisibleForTesting
  public static final Constraint LEGACY_EXECUTOR = new Constraint(EXECUTOR_CONSTRAINT,
      TaskConstraint.value(new ValueConstraint(false, ImmutableSet.of(LEGACY_EXECUTOR_VALUE))));
  private static final Iterable<FieldSanitizer> SANITIZERS = ImmutableList.<FieldSanitizer>builder()
      .add(new RequiredField(_Fields.NUM_CPUS))
      .add(new RequiredField(_Fields.RAM_MB))
      .add(new RequiredField(_Fields.DISK_MB))
      .add(new DefaultField(_Fields.IS_DAEMON, false))
      .add(new DefaultField(_Fields.PRIORITY, 0))
      .add(new DefaultField(_Fields.PRODUCTION, false))
      .add(new DefaultField(_Fields.HEALTH_CHECK_INTERVAL_SECS, 30))
      .add(new DefaultField(_Fields.MAX_TASK_FAILURES, 1))
      .add(new DefaultField(_Fields.TASK_LINKS, Sets.<String>newHashSet()))
      .add(new DefaultField(_Fields.REQUESTED_PORTS, Sets.<String>newHashSet()))
      .add(new DefaultField(_Fields.CONSTRAINTS, Sets.<Constraint>newHashSet()))
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) {
          if (!Iterables.any(task.getConstraints(), hasName(HOST_CONSTRAINT))) {
            task.addToConstraints(hostLimitConstraint(1));
          }
        }
      })
      // TODO(<NAME>): Remove this once the old executor is retired.
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) throws TaskDescriptionException {
          // Apply an executor:legacy constraint to all non-thermos tasks.
          if (!Tasks.isThermos(task)) {
            Iterables.removeIf(task.getConstraints(), hasName(LEGACY_EXECUTOR.getName()));
            if (CONSTRAIN_LEGACY_EXECUTOR.get()) {
              task.addToConstraints(LEGACY_EXECUTOR);
            }
          }
        }
      })
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) {
          if (!isDedicated(task)
              && task.isProduction()
              && task.isIsDaemon()
              && !Iterables.any(task.getConstraints(), hasName(RACK_CONSTRAINT))) {

            task.addToConstraints(rackLimitConstraint(1));
          }
        }
      })
      .build();

  private static final List<Field<?>> FIELDS = ImmutableList.<Field<?>>builder()
      .add(new TypedField<String>(String.class, false, "hdfs_path", null) {
        @Override boolean isSet(TwitterTaskInfo task) { return task.isSetHdfsPath(); }

        @Override void apply(TwitterTaskInfo task, String value) throws TaskDescriptionException {
          task.setHdfsPath(value);
        }
      })
      .add(new TypedField<String>(String.class, true, START_COMMAND_FIELD, "") {
        @Override boolean isSet(TwitterTaskInfo task) { return task.isSetStartCommand(); }

        @Override void apply(TwitterTaskInfo task, String value) throws TaskDescriptionException {
          task.setStartCommand(value);<commit_msg>Always constrain legacy executors to legacy hosts.<commit_after>package com.twitter.mesos.scheduler.configuration;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;

import javax.annotation.Nullable;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

import org.apache.commons.lang.StringUtils;

import com.twitter.common.args.Arg;
import com.twitter.common.args.CmdLine;
import com.twitter.common.args.constraints.Positive;
import com.twitter.common.base.MorePreconditions;
import com.twitter.mesos.Tasks;
import com.twitter.mesos.gen.Constraint;
import com.twitter.mesos.gen.Identity;
import com.twitter.mesos.gen.JobConfiguration;
import com.twitter.mesos.gen.LimitConstraint;
import com.twitter.mesos.gen.TaskConstraint;
import com.twitter.mesos.gen.TwitterTaskInfo;
import com.twitter.mesos.gen.TwitterTaskInfo._Fields;
import com.twitter.mesos.gen.ValueConstraint;
import com.twitter.mesos.scheduler.CommandLineExpander;
import com.twitter.mesos.scheduler.ThermosJank;
import com.twitter.mesos.scheduler.configuration.ValueParser.ParseException;

/**
 * Manages translation from a string-mapped configuration to a concrete configuration type, and
 * defaults for optional values.
 *
 * TODO(<NAME>): Add input validation to all fields (strings not empty, positive ints, etc).
 */
public final class ConfigurationManager {

  public static final String DEDICATED_ATTRIBUTE = "dedicated";

  @VisibleForTesting public static final String HOST_CONSTRAINT = "host";
  @VisibleForTesting public static final String RACK_CONSTRAINT = "rack";
  @VisibleForTesting public static final String START_COMMAND_FIELD = "start_command";

  @VisibleForTesting
  @Positive
  @CmdLine(name = "max_tasks_per_job", help = "Maximum number of allowed tasks in a single job.")
  public static final Arg<Integer> MAX_TASKS_PER_JOB = Arg.create(500);

  @CmdLine(name = "require_contact_email",
      help = "If true, reject jobs that do not specify a contact email address.")
  public static final Arg<Boolean> REQUIRE_CONTACT_EMAIL = Arg.create(true);

  private static final Logger LOG = Logger.getLogger(ConfigurationManager.class.getName());

  private static final Pattern GOOD_IDENTIFIER_PATTERN = Pattern.compile("[\\w\\-\\.]+");

  private static final int MAX_IDENTIFIER_LENGTH = 255;

  private interface FieldSanitizer {
    void sanitize(TwitterTaskInfo task) throws TaskDescriptionException;
  }

  private static class DefaultField implements FieldSanitizer {
    private final _Fields field;
    private final Object defaultValue;

    DefaultField(_Fields field, Object defaultValue) {
      this.field = field;
      this.defaultValue = defaultValue;
    }

    @Override public void sanitize(TwitterTaskInfo task) {
      if (!task.isSet(field)) {
        task.setFieldValue(field, defaultValue);
      }
    }
  }

  private static class RequiredField implements FieldSanitizer {
    private final _Fields field;

    RequiredField(_Fields field) {
      this.field = field;
    }

    @Override public void sanitize(TwitterTaskInfo task) throws TaskDescriptionException {
      if (!task.isSet(field)) {
        throw new TaskDescriptionException("Field " + field.getFieldName() + " is required.");
      }
    }
  }

  public static final String EXECUTOR_CONSTRAINT = "executor";
  public static final String LEGACY_EXECUTOR_VALUE = "legacy";
  @VisibleForTesting
  public static final Constraint LEGACY_EXECUTOR = new Constraint(EXECUTOR_CONSTRAINT,
      TaskConstraint.value(new ValueConstraint(false, ImmutableSet.of(LEGACY_EXECUTOR_VALUE))));
  private static final Iterable<FieldSanitizer> SANITIZERS = ImmutableList.<FieldSanitizer>builder()
      .add(new RequiredField(_Fields.NUM_CPUS))
      .add(new RequiredField(_Fields.RAM_MB))
      .add(new RequiredField(_Fields.DISK_MB))
      .add(new DefaultField(_Fields.IS_DAEMON, false))
      .add(new DefaultField(_Fields.PRIORITY, 0))
      .add(new DefaultField(_Fields.PRODUCTION, false))
      .add(new DefaultField(_Fields.HEALTH_CHECK_INTERVAL_SECS, 30))
      .add(new DefaultField(_Fields.MAX_TASK_FAILURES, 1))
      .add(new DefaultField(_Fields.TASK_LINKS, Sets.<String>newHashSet()))
      .add(new DefaultField(_Fields.REQUESTED_PORTS, Sets.<String>newHashSet()))
      .add(new DefaultField(_Fields.CONSTRAINTS, Sets.<Constraint>newHashSet()))
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) {
          if (!Iterables.any(task.getConstraints(), hasName(HOST_CONSTRAINT))) {
            task.addToConstraints(hostLimitConstraint(1));
          }
        }
      })
      // TODO(<NAME>): Remove this once the old executor is retired.
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) throws TaskDescriptionException {
          // Apply an executor:legacy constraint to all non-thermos tasks.
          if (!Tasks.isThermos(task)) {
            Iterables.removeIf(task.getConstraints(), hasName(LEGACY_EXECUTOR.getName()));
            task.addToConstraints(LEGACY_EXECUTOR);
          }
        }
      })
      .add(new FieldSanitizer() {
        @Override public void sanitize(TwitterTaskInfo task) {
          if (!isDedicated(task)
              && task.isProduction()
              && task.isIsDaemon()
              && !Iterables.any(task.getConstraints(), hasName(RACK_CONSTRAINT))) {

            task.addToConstraints(rackLimitConstraint(1));
          }
        }
      })
      .build();

  private static final List<Field<?>> FIELDS = ImmutableList.<Field<?>>builder()
      .add(new TypedField<String>(String.class, false, "hdfs_path", null) {
        @Override boolean isSet(TwitterTaskInfo task) { return task.isSetHdfsPath(); }

        @Override void apply(TwitterTaskInfo task, String value) throws TaskDescriptionException {
          task.setHdfsPath(value);
        }
      })
      .add(new TypedField<String>(String.class, true, START_COMMAND_FIELD, "") {
        @Override boolean isSet(TwitterTaskInfo task) { return task.isSetStartCommand(); }

        @Override void apply(TwitterTaskInfo task, String value) throws TaskDescriptionException {
          task.setStartCommand(value);
{'fraction_non_alphanumeric': 0.1025527519551424, 'fraction_numerical': 0.0017706949977866313, 'mean_word_length': 4.007388252678242, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1375940', 'n_tokens_mistral': 4017, 'n_tokens_neox': 3872, 'n_words': 860}