/*
 * 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.hbase.quotas;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ScheduledChore;
import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionReplicaUtil;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.master.MetricsMaster;
import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
import org.apache.hbase.thirdparty.com.google.common.collect.Iterables;
import org.apache.hbase.thirdparty.com.google.common.collect.Multimap;

import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.SpaceQuota;

/**
 * Reads the currently received Region filesystem-space use reports and acts on those which
 * violate a defined quota.
 */
@InterfaceAudience.Private
public class QuotaObserverChore extends ScheduledChore {
  private static final Logger LOG = LoggerFactory.getLogger(QuotaObserverChore.class);
  static final String QUOTA_OBSERVER_CHORE_PERIOD_KEY =
      "hbase.master.quotas.observer.chore.period";
  static final int QUOTA_OBSERVER_CHORE_PERIOD_DEFAULT = 1000 * 60 * 1; // 1 minutes in millis

  static final String QUOTA_OBSERVER_CHORE_DELAY_KEY =
      "hbase.master.quotas.observer.chore.delay";
  static final long QUOTA_OBSERVER_CHORE_DELAY_DEFAULT = 1000L * 15L; // 15 seconds in millis

  static final String QUOTA_OBSERVER_CHORE_TIMEUNIT_KEY =
      "hbase.master.quotas.observer.chore.timeunit";
  static final String QUOTA_OBSERVER_CHORE_TIMEUNIT_DEFAULT = TimeUnit.MILLISECONDS.name();

  static final String QUOTA_OBSERVER_CHORE_REPORT_PERCENT_KEY =
      "hbase.master.quotas.observer.report.percent";
  static final double QUOTA_OBSERVER_CHORE_REPORT_PERCENT_DEFAULT= 0.95;

  static final String REGION_REPORT_RETENTION_DURATION_KEY =
      "hbase.master.quotas.region.report.retention.millis";
  static final long REGION_REPORT_RETENTION_DURATION_DEFAULT =
      1000 * 60 * 10; // 10 minutes

  private final Connection conn;
  private final Configuration conf;
  private final MasterQuotaManager quotaManager;
  private final MetricsMaster metrics;
  /*
   * Callback that changes in quota snapshots are passed to.
   */
  private final SpaceQuotaSnapshotNotifier snapshotNotifier;

  /*
   * Preserves the state of quota snapshots for tables and namespaces
   */
  private final Map<TableName,SpaceQuotaSnapshot> tableQuotaSnapshots;
  private final Map<TableName,SpaceQuotaSnapshot> readOnlyTableQuotaSnapshots;
  private final Map<String,SpaceQuotaSnapshot> namespaceQuotaSnapshots;
  private final Map<String,SpaceQuotaSnapshot> readOnlyNamespaceSnapshots;

  // The time, in millis, that region reports should be kept by the master
  private final long regionReportLifetimeMillis;

  /*
   * Encapsulates logic for tracking the state of a table/namespace WRT space quotas
   */
  private QuotaSnapshotStore<TableName> tableSnapshotStore;
  private QuotaSnapshotStore<String> namespaceSnapshotStore;

  public QuotaObserverChore(HMaster master, MetricsMaster metrics) {
    this(
        master.getConnection(), master.getConfiguration(),
        master.getSpaceQuotaSnapshotNotifier(), master.getMasterQuotaManager(),
        master, metrics);
  }

  QuotaObserverChore(
      Connection conn, Configuration conf, SpaceQuotaSnapshotNotifier snapshotNotifier,
      MasterQuotaManager quotaManager, Stoppable stopper, MetricsMaster metrics) {
    super(
        QuotaObserverChore.class.getSimpleName(), stopper, getPeriod(conf),
        getInitialDelay(conf), getTimeUnit(conf));
    this.conn = conn;
    this.conf = conf;
    this.metrics = metrics;
    this.quotaManager = quotaManager;
    this.snapshotNotifier = Objects.requireNonNull(snapshotNotifier);
    this.tableQuotaSnapshots = new ConcurrentHashMap<>();
    this.readOnlyTableQuotaSnapshots = Collections.unmodifiableMap(tableQuotaSnapshots);
    this.namespaceQuotaSnapshots = new ConcurrentHashMap<>();
    this.readOnlyNamespaceSnapshots = Collections.unmodifiableMap(namespaceQuotaSnapshots);
    this.regionReportLifetimeMillis = conf.getLong(
        REGION_REPORT_RETENTION_DURATION_KEY, REGION_REPORT_RETENTION_DURATION_DEFAULT);
  }

  @Override
  protected void chore() {
    try {
      if (LOG.isTraceEnabled()) {
        LOG.trace("Refreshing space quotas in RegionServer");
      }
      long start = System.nanoTime();
      _chore();
      if (metrics != null) {
        metrics.incrementQuotaObserverTime((System.nanoTime() - start) / 1_000_000);
      }
    } catch (IOException e) {
      LOG.warn("Failed to process quota reports and update quota state. Will retry.", e);
    }
  }

  void _chore() throws IOException {
    // Get the total set of tables that have quotas defined. Includes table quotas
    // and tables included by namespace quotas.
    TablesWithQuotas tablesWithQuotas = fetchAllTablesWithQuotasDefined();
    if (LOG.isTraceEnabled()) {
      LOG.trace("Found following tables with quotas: " + tablesWithQuotas);
    }

    if (metrics != null) {
      // Set the number of namespaces and tables with quotas defined
      metrics.setNumSpaceQuotas(tablesWithQuotas.getTableQuotaTables().size()
          + tablesWithQuotas.getNamespacesWithQuotas().size());
    }

    // The current "view" of region space use. Used henceforth.
    final Map<RegionInfo,Long> reportedRegionSpaceUse = quotaManager.snapshotRegionSizes();
    if (LOG.isTraceEnabled()) {
      LOG.trace(
          "Using " + reportedRegionSpaceUse.size() + " region space use reports: " +
          reportedRegionSpaceUse);
    }

    // Remove the "old" region reports
    pruneOldRegionReports();

    // Create the stores to track table and namespace snapshots
    initializeSnapshotStores(reportedRegionSpaceUse);
    // Report the number of (non-expired) region size reports
    if (metrics != null) {
      metrics.setNumRegionSizeReports(reportedRegionSpaceUse.size());
    }

    // Filter out tables for which we don't have adequate regionspace reports yet.
    // Important that we do this after we instantiate the stores above
    // This gives us a set of Tables which may or may not be violating their quota.
    // To be safe, we want to make sure that these are not in violation.
    Set<TableName> tablesInLimbo = tablesWithQuotas.filterInsufficientlyReportedTables(
        tableSnapshotStore);

    if (LOG.isTraceEnabled()) {
      LOG.trace("Filtered insufficiently reported tables, left with " +
          reportedRegionSpaceUse.size() + " regions reported");
    }

    for (TableName tableInLimbo : tablesInLimbo) {
      final SpaceQuotaSnapshot currentSnapshot = tableSnapshotStore.getCurrentState(tableInLimbo);
      SpaceQuotaStatus currentStatus = currentSnapshot.getQuotaStatus();
      if (currentStatus.isInViolation()) {
        if (LOG.isTraceEnabled()) {
          LOG.trace("Moving " + tableInLimbo + " out of violation because fewer region sizes were"
              + " reported than required.");
        }
        SpaceQuotaSnapshot targetSnapshot = new SpaceQuotaSnapshot(
            SpaceQuotaStatus.notInViolation(), currentSnapshot.getUsage(),
            currentSnapshot.getLimit());
        this.snapshotNotifier.transitionTable(tableInLimbo, targetSnapshot);
        // Update it in the Table QuotaStore so that memory is consistent with no violation.
        tableSnapshotStore.setCurrentState(tableInLimbo, targetSnapshot);
        // In case of Disable SVP, we need to enable the table as it moves out of violation
        if (SpaceViolationPolicy.DISABLE == currentStatus.getPolicy().orElse(null)) {
          QuotaUtil.enableTableIfNotEnabled(conn, tableInLimbo);
        }
      }
    }

    // Transition each table to/from quota violation based on the current and target state.
    // Only table quotas are enacted.
    final Set<TableName> tablesWithTableQuotas = tablesWithQuotas.getTableQuotaTables();
    processTablesWithQuotas(tablesWithTableQuotas);

    // For each Namespace quota, transition each table in the namespace in or out of violation
    // only if a table quota violation policy has not already been applied.
    final Set<String> namespacesWithQuotas = tablesWithQuotas.getNamespacesWithQuotas();
    final Multimap<String,TableName> tablesByNamespace = tablesWithQuotas.getTablesByNamespace();
    processNamespacesWithQuotas(namespacesWithQuotas, tablesByNamespace);
  }

  void initializeSnapshotStores(Map<RegionInfo,Long> regionSizes) {
    Map<RegionInfo,Long> immutableRegionSpaceUse = Collections.unmodifiableMap(regionSizes);
    if (tableSnapshotStore == null) {
      tableSnapshotStore = new TableQuotaSnapshotStore(conn, this, immutableRegionSpaceUse);
    } else {
      tableSnapshotStore.setRegionUsage(immutableRegionSpaceUse);
    }
    if (namespaceSnapshotStore == null) {
      namespaceSnapshotStore = new NamespaceQuotaSnapshotStore(
          conn, this, immutableRegionSpaceUse);
    } else {
      namespaceSnapshotStore.setRegionUsage(immutableRegionSpaceUse);
    }
  }

  /**
   * Processes each {@code TableName} which has a quota defined and moves it in or out of
   * violation based on the space use.
   *
   * @param tablesWithTableQuotas The HBase tables which have quotas defined
   */
  void processTablesWithQuotas(final Set<TableName> tablesWithTableQuotas) throws IOException {
    long numTablesInViolation = 0L;
    for (TableName table : tablesWithTableQuotas) {
      final SpaceQuota spaceQuota = tableSnapshotStore.getSpaceQuota(table);
      if (spaceQuota == null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Unexpectedly did not find a space quota for " + table
              + ", maybe it was recently deleted.");
        }
        continue;
      }
      final SpaceQuotaSnapshot currentSnapshot = tableSnapshotStore.getCurrentState(table);
      final SpaceQuotaSnapshot targetSnapshot = tableSnapshotStore.getTargetState(table, spaceQuota);
      if (LOG.isTraceEnabled()) {
        LOG.trace("Processing " + table + " with current=" + currentSnapshot + ", target="
            + targetSnapshot);
      }
      updateTableQuota(table, currentSnapshot, targetSnapshot);

      if (targetSnapshot.getQuotaStatus().isInViolation()) {
        numTablesInViolation++;
      }
    }
    // Report the number of tables in violation
    if (metrics != null) {
      metrics.setNumTableInSpaceQuotaViolation(numTablesInViolation);
    }
  }

  /**
   * Processes each namespace which has a quota defined and moves all of the tables contained
   * in that namespace into or out of violation of the quota. Tables which are already in
   * violation of a quota at the table level which <em>also</em> have a reside in a namespace
   * with a violated quota will not have the namespace quota enacted. The table quota takes
   * priority over the namespace quota.
   *
   * @param namespacesWithQuotas The set of namespaces that have quotas defined
   * @param tablesByNamespace A mapping of namespaces and the tables contained in those namespaces
   */
  void processNamespacesWithQuotas(
      final Set<String> namespacesWithQuotas,
      final Multimap<String,TableName> tablesByNamespace) throws IOException {
    long numNamespacesInViolation = 0L;
    for (String namespace : namespacesWithQuotas) {
      // Get the quota definition for the namespace
      final SpaceQuota spaceQuota = namespaceSnapshotStore.getSpaceQuota(namespace);
      if (spaceQuota == null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Could not get Namespace space quota for " + namespace
              + ", maybe it was recently deleted.");
        }
        continue;
      }
      final SpaceQuotaSnapshot currentSnapshot = namespaceSnapshotStore.getCurrentState(namespace);
      final SpaceQuotaSnapshot targetSnapshot = namespaceSnapshotStore.getTargetState(
          namespace, spaceQuota);
      if (LOG.isTraceEnabled()) {
        LOG.trace("Processing " + namespace + " with current=" + currentSnapshot + ", target="
            + targetSnapshot);
      }
      updateNamespaceQuota(namespace, currentSnapshot, targetSnapshot, tablesByNamespace);

      if (targetSnapshot.getQuotaStatus().isInViolation()) {
        numNamespacesInViolation++;
      }
    }

    // Report the number of namespaces in violation
    if (metrics != null) {
      metrics.setNumNamespacesInSpaceQuotaViolation(numNamespacesInViolation);
    }
  }

  /**
   * Updates the hbase:quota table with the new quota policy for this <code>table</code>
   * if necessary.
   *
   * @param table The table being checked
   * @param currentSnapshot The state of the quota on this table from the previous invocation.
   * @param targetSnapshot The state the quota should be in for this table.
   */
  void updateTableQuota(
      TableName table, SpaceQuotaSnapshot currentSnapshot, SpaceQuotaSnapshot targetSnapshot)
          throws IOException {
    final SpaceQuotaStatus currentStatus = currentSnapshot.getQuotaStatus();
    final SpaceQuotaStatus targetStatus = targetSnapshot.getQuotaStatus();

    // If we're changing something, log it.
    if (!currentSnapshot.equals(targetSnapshot)) {
      this.snapshotNotifier.transitionTable(table, targetSnapshot);
      // Update it in memory
      tableSnapshotStore.setCurrentState(table, targetSnapshot);

      // If the target is none, we're moving out of violation. Update the hbase:quota table
      SpaceViolationPolicy currPolicy = currentStatus.getPolicy().orElse(null);
      SpaceViolationPolicy targetPolicy = targetStatus.getPolicy().orElse(null);
      if (!targetStatus.isInViolation()) {
        // In case of Disable SVP, we need to enable the table as it moves out of violation
        if (isDisableSpaceViolationPolicy(currPolicy, targetPolicy)) {
          QuotaUtil.enableTableIfNotEnabled(conn, table);
        }
        if (LOG.isDebugEnabled()) {
          LOG.debug(table + " moved into observance of table space quota.");
        }
      } else {
        // We're either moving into violation or changing violation policies
        if (currPolicy != targetPolicy && SpaceViolationPolicy.DISABLE == currPolicy) {
          // In case of policy switch, we need to enable the table if current policy is Disable SVP
          QuotaUtil.enableTableIfNotEnabled(conn, table);
        } else if (SpaceViolationPolicy.DISABLE == targetPolicy) {
          // In case of Disable SVP, we need to disable the table as it moves into violation
          QuotaUtil.disableTableIfNotDisabled(conn, table);
        }
        if (LOG.isDebugEnabled()) {
          LOG.debug(
            table + " moved into violation of table space quota with policy of " + targetPolicy);
        }
      }
    } else if (LOG.isTraceEnabled()) {
      // Policies are the same, so we have nothing to do except log this. Don't need to re-update
      // the quota table
      if (!currentStatus.isInViolation()) {
        LOG.trace(table + " remains in observance of quota.");
      } else {
        LOG.trace(table + " remains in violation of quota.");
      }
    }
  }

  /**
   * Method to check whether we are dealing with DISABLE {@link SpaceViolationPolicy}. In such a
   * case, currPolicy or/and targetPolicy will be having DISABLE policy.
   * @param currPolicy currently set space violation policy
   * @param targetPolicy new space violation policy
   * @return true if is DISABLE space violation policy; otherwise false
   */
  private boolean isDisableSpaceViolationPolicy(final SpaceViolationPolicy currPolicy,
      final SpaceViolationPolicy targetPolicy) {
    return SpaceViolationPolicy.DISABLE == currPolicy
        || SpaceViolationPolicy.DISABLE == targetPolicy;
  }

  /**
   * Updates the hbase:quota table with the target quota policy for this <code>namespace</code>
   * if necessary.
   *
   * @param namespace The namespace being checked
   * @param currentSnapshot The state of the quota on this namespace from the previous invocation
   * @param targetSnapshot The state the quota should be in for this namespace
   * @param tablesByNamespace A mapping of tables in namespaces.
   */
  void updateNamespaceQuota(
      String namespace, SpaceQuotaSnapshot currentSnapshot, SpaceQuotaSnapshot targetSnapshot,
      final Multimap<String,TableName> tablesByNamespace) throws IOException {
    final SpaceQuotaStatus targetStatus = targetSnapshot.getQuotaStatus();

    // When the policies differ, we need to move into or out of violation
    if (!currentSnapshot.equals(targetSnapshot)) {
      // We want to have a policy of "NONE", moving out of violation
      if (!targetStatus.isInViolation()) {
        for (TableName tableInNS : tablesByNamespace.get(namespace)) {
          // If there is a quota on this table in violation
          if (tableSnapshotStore.getCurrentState(tableInNS).getQuotaStatus().isInViolation()) {
            // Table-level quota violation policy is being applied here.
            if (LOG.isTraceEnabled()) {
              LOG.trace("Not activating Namespace violation policy because a Table violation"
                  + " policy is already in effect for " + tableInNS);
            }
          } else {
            LOG.info(tableInNS + " moving into observance of namespace space quota");
            this.snapshotNotifier.transitionTable(tableInNS, targetSnapshot);
          }
        }
      // We want to move into violation at the NS level
      } else {
        // Moving tables in the namespace into violation or to a different violation policy
        for (TableName tableInNS : tablesByNamespace.get(namespace)) {
          final SpaceQuotaSnapshot tableQuotaSnapshot =
                tableSnapshotStore.getCurrentState(tableInNS);
          final boolean hasTableQuota =
              !Objects.equals(QuotaSnapshotStore.NO_QUOTA, tableQuotaSnapshot);
          if (hasTableQuota && tableQuotaSnapshot.getQuotaStatus().isInViolation()) {
            // Table-level quota violation policy is being applied here.
            if (LOG.isTraceEnabled()) {
              LOG.trace("Not activating Namespace violation policy because a Table violation"
                  + " policy is already in effect for " + tableInNS);
            }
          } else {
            // No table quota present or a table quota present that is not in violation
            LOG.info(tableInNS + " moving into violation of namespace space quota with policy "
                + targetStatus.getPolicy());
            this.snapshotNotifier.transitionTable(tableInNS, targetSnapshot);
          }
        }
      }
      // Update the new state in memory for this namespace
      namespaceSnapshotStore.setCurrentState(namespace, targetSnapshot);
    } else {
      // Policies are the same
      if (!targetStatus.isInViolation()) {
        // Both are NONE, so we remain in observance
        if (LOG.isTraceEnabled()) {
          LOG.trace(namespace + " remains in observance of quota.");
        }
      } else {
        // Namespace quota is still in violation, need to enact if the table quota is not
        // taking priority.
        for (TableName tableInNS : tablesByNamespace.get(namespace)) {
          // Does a table policy exist
          if (tableSnapshotStore.getCurrentState(tableInNS).getQuotaStatus().isInViolation()) {
            // Table-level quota violation policy is being applied here.
            if (LOG.isTraceEnabled()) {
              LOG.trace("Not activating Namespace violation policy because Table violation"
                  + " policy is already in effect for " + tableInNS);
            }
          } else {
            // No table policy, so enact namespace policy
            LOG.info(tableInNS + " moving into violation of namespace space quota");
            this.snapshotNotifier.transitionTable(tableInNS, targetSnapshot);
          }
        }
      }
    }
  }

  /**
   * Removes region reports over a certain age.
   */
  void pruneOldRegionReports() {
    final long now = EnvironmentEdgeManager.currentTime();
    final long pruneTime = now - regionReportLifetimeMillis;
    final int numRemoved = quotaManager.pruneEntriesOlderThan(pruneTime,this);
    if (LOG.isTraceEnabled()) {
      LOG.trace("Removed " + numRemoved + " old region size reports that were older than "
          + pruneTime + ".");
    }
  }

  /**
   * Computes the set of all tables that have quotas defined. This includes tables with quotas
   * explicitly set on them, in addition to tables that exist namespaces which have a quota
   * defined.
   */
  TablesWithQuotas fetchAllTablesWithQuotasDefined() throws IOException {
    final Scan scan = QuotaTableUtil.makeScan(null);
    final TablesWithQuotas tablesWithQuotas = new TablesWithQuotas(conn, conf);
    try (final QuotaRetriever scanner = new QuotaRetriever()) {
      scanner.init(conn, scan);
      for (QuotaSettings quotaSettings : scanner) {
        // Only one of namespace and tablename should be 'null'
        final String namespace = quotaSettings.getNamespace();
        final TableName tableName = quotaSettings.getTableName();
        if (QuotaType.SPACE != quotaSettings.getQuotaType()) {
          continue;
        }

        if (namespace != null) {
          assert tableName == null;
          // Collect all of the tables in the namespace
          TableName[] tablesInNS = conn.getAdmin().listTableNamesByNamespace(namespace);
          for (TableName tableUnderNs : tablesInNS) {
            if (LOG.isTraceEnabled()) {
              LOG.trace("Adding " + tableUnderNs + " under " +  namespace
                  + " as having a namespace quota");
            }
            tablesWithQuotas.addNamespaceQuotaTable(tableUnderNs);
          }
        } else {
          assert tableName != null;
          if (LOG.isTraceEnabled()) {
            LOG.trace("Adding " + tableName + " as having table quota.");
          }
          // namespace is already null, must be a non-null tableName
          tablesWithQuotas.addTableQuotaTable(tableName);
        }
      }
      return tablesWithQuotas;
    }
  }

  QuotaSnapshotStore<TableName> getTableSnapshotStore() {
    return tableSnapshotStore;
  }

  QuotaSnapshotStore<String> getNamespaceSnapshotStore() {
    return namespaceSnapshotStore;
  }

  /**
   * Returns an unmodifiable view over the current {@link SpaceQuotaSnapshot} objects
   * for each HBase table with a quota defined.
   */
  public Map<TableName,SpaceQuotaSnapshot> getTableQuotaSnapshots() {
    return readOnlyTableQuotaSnapshots;
  }

  /**
   * Returns an unmodifiable view over the current {@link SpaceQuotaSnapshot} objects
   * for each HBase namespace with a quota defined.
   */
  public Map<String,SpaceQuotaSnapshot> getNamespaceQuotaSnapshots() {
    return readOnlyNamespaceSnapshots;
  }

  /**
   * Fetches the {@link SpaceQuotaSnapshot} for the given table.
   */
  SpaceQuotaSnapshot getTableQuotaSnapshot(TableName table) {
    SpaceQuotaSnapshot state = this.tableQuotaSnapshots.get(table);
    if (state == null) {
      // No tracked state implies observance.
      return QuotaSnapshotStore.NO_QUOTA;
    }
    return state;
  }

  /**
   * Stores the quota state for the given table.
   */
  void setTableQuotaSnapshot(TableName table, SpaceQuotaSnapshot snapshot) {
    this.tableQuotaSnapshots.put(table, snapshot);
  }

  /**
   * Fetches the {@link SpaceQuotaSnapshot} for the given namespace from this chore.
   */
  SpaceQuotaSnapshot getNamespaceQuotaSnapshot(String namespace) {
    SpaceQuotaSnapshot state = this.namespaceQuotaSnapshots.get(namespace);
    if (state == null) {
      // No tracked state implies observance.
      return QuotaSnapshotStore.NO_QUOTA;
    }
    return state;
  }

  /**
   * Stores the given {@code snapshot} for the given {@code namespace} in this chore.
   */
  void setNamespaceQuotaSnapshot(String namespace, SpaceQuotaSnapshot snapshot) {
    this.namespaceQuotaSnapshots.put(namespace, snapshot);
  }

  /**
   * Extracts the period for the chore from the configuration.
   *
   * @param conf The configuration object.
   * @return The configured chore period or the default value in the given timeunit.
   * @see #getTimeUnit(Configuration)
   */
  static int getPeriod(Configuration conf) {
    return conf.getInt(QUOTA_OBSERVER_CHORE_PERIOD_KEY,
        QUOTA_OBSERVER_CHORE_PERIOD_DEFAULT);
  }

  /**
   * Extracts the initial delay for the chore from the configuration.
   *
   * @param conf The configuration object.
   * @return The configured chore initial delay or the default value in the given timeunit.
   * @see #getTimeUnit(Configuration)
   */
  static long getInitialDelay(Configuration conf) {
    return conf.getLong(QUOTA_OBSERVER_CHORE_DELAY_KEY,
        QUOTA_OBSERVER_CHORE_DELAY_DEFAULT);
  }

  /**
   * Extracts the time unit for the chore period and initial delay from the configuration. The
   * configuration value for {@link #QUOTA_OBSERVER_CHORE_TIMEUNIT_KEY} must correspond to
   * a {@link TimeUnit} value.
   *
   * @param conf The configuration object.
   * @return The configured time unit for the chore period and initial delay or the default value.
   */
  static TimeUnit getTimeUnit(Configuration conf) {
    return TimeUnit.valueOf(conf.get(QUOTA_OBSERVER_CHORE_TIMEUNIT_KEY,
        QUOTA_OBSERVER_CHORE_TIMEUNIT_DEFAULT));
  }

  /**
   * Extracts the percent of Regions for a table to have been reported to enable quota violation
   * state change.
   *
   * @param conf The configuration object.
   * @return The percent of regions reported to use.
   */
  static Double getRegionReportPercent(Configuration conf) {
    return conf.getDouble(QUOTA_OBSERVER_CHORE_REPORT_PERCENT_KEY,
        QUOTA_OBSERVER_CHORE_REPORT_PERCENT_DEFAULT);
  }

  /**
   * A container which encapsulates the tables that have either a table quota or are contained in a
   * namespace which have a namespace quota.
   */
  static class TablesWithQuotas {
    private final Set<TableName> tablesWithTableQuotas = new HashSet<>();
    private final Set<TableName> tablesWithNamespaceQuotas = new HashSet<>();
    private final Connection conn;
    private final Configuration conf;

    public TablesWithQuotas(Connection conn, Configuration conf) {
      this.conn = Objects.requireNonNull(conn);
      this.conf = Objects.requireNonNull(conf);
    }

    Configuration getConfiguration() {
      return conf;
    }

    /**
     * Adds a table with a table quota.
     */
    public void addTableQuotaTable(TableName tn) {
      tablesWithTableQuotas.add(tn);
    }

    /**
     * Adds a table with a namespace quota.
     */
    public void addNamespaceQuotaTable(TableName tn) {
      tablesWithNamespaceQuotas.add(tn);
    }

    /**
     * Returns true if the given table has a table quota.
     */
    public boolean hasTableQuota(TableName tn) {
      return tablesWithTableQuotas.contains(tn);
    }

    /**
     * Returns true if the table exists in a namespace with a namespace quota.
     */
    public boolean hasNamespaceQuota(TableName tn) {
      return tablesWithNamespaceQuotas.contains(tn);
    }

    /**
     * Returns an unmodifiable view of all tables with table quotas.
     */
    public Set<TableName> getTableQuotaTables() {
      return Collections.unmodifiableSet(tablesWithTableQuotas);
    }

    /**
     * Returns an unmodifiable view of all tables in namespaces that have
     * namespace quotas.
     */
    public Set<TableName> getNamespaceQuotaTables() {
      return Collections.unmodifiableSet(tablesWithNamespaceQuotas);
    }

    public Set<String> getNamespacesWithQuotas() {
      Set<String> namespaces = new HashSet<>();
      for (TableName tn : tablesWithNamespaceQuotas) {
        namespaces.add(tn.getNamespaceAsString());
      }
      return namespaces;
    }

    /**
     * Returns a view of all tables that reside in a namespace with a namespace
     * quota, grouped by the namespace itself.
     */
    public Multimap<String,TableName> getTablesByNamespace() {
      Multimap<String,TableName> tablesByNS = HashMultimap.create();
      for (TableName tn : tablesWithNamespaceQuotas) {
        tablesByNS.put(tn.getNamespaceAsString(), tn);
      }
      return tablesByNS;
    }

    /**
     * Filters out all tables for which the Master currently doesn't have enough region space
     * reports received from RegionServers yet.
     */
    public Set<TableName> filterInsufficientlyReportedTables(
        QuotaSnapshotStore<TableName> tableStore) throws IOException {
      final double percentRegionsReportedThreshold = getRegionReportPercent(getConfiguration());
      Set<TableName> tablesToRemove = new HashSet<>();
      for (TableName table : Iterables.concat(tablesWithTableQuotas, tablesWithNamespaceQuotas)) {
        // Don't recompute a table we've already computed
        if (tablesToRemove.contains(table)) {
          continue;
        }
        final int numRegionsInTable = getNumRegions(table);
        // If the table doesn't exist (no regions), bail out.
        if (numRegionsInTable == 0) {
          if (LOG.isTraceEnabled()) {
            LOG.trace("Filtering " + table + " because no regions were reported");
          }
          tablesToRemove.add(table);
          continue;
        }
        final int reportedRegionsInQuota = getNumReportedRegions(table, tableStore);
        final double ratioReported = ((double) reportedRegionsInQuota) / numRegionsInTable;
        if (ratioReported < percentRegionsReportedThreshold) {
          if (LOG.isTraceEnabled()) {
            LOG.trace("Filtering " + table + " because " + reportedRegionsInQuota  + " of " +
                numRegionsInTable + " regions were reported.");
          }
          tablesToRemove.add(table);
        } else if (LOG.isTraceEnabled()) {
          LOG.trace("Retaining " + table + " because " + reportedRegionsInQuota  + " of " +
              numRegionsInTable + " regions were reported.");
        }
      }
      for (TableName tableToRemove : tablesToRemove) {
        tablesWithTableQuotas.remove(tableToRemove);
        tablesWithNamespaceQuotas.remove(tableToRemove);
      }
      return tablesToRemove;
    }

    /**
     * Computes the total number of regions in a table.
     */
    int getNumRegions(TableName table) throws IOException {
      List<RegionInfo> regions = this.conn.getAdmin().getRegions(table);
      if (regions == null) {
        return 0;
      }
      // Filter the region replicas if any and return the original number of regions for a table.
      RegionReplicaUtil.removeNonDefaultRegions(regions);
      return regions.size();
    }

    /**
     * Computes the number of regions reported for a table.
     */
    int getNumReportedRegions(TableName table, QuotaSnapshotStore<TableName> tableStore)
        throws IOException {
      return Iterables.size(tableStore.filterBySubject(table));
    }

    @Override
    public String toString() {
      final StringBuilder sb = new StringBuilder(32);
      sb.append(getClass().getSimpleName())
          .append(": tablesWithTableQuotas=")
          .append(this.tablesWithTableQuotas)
          .append(", tablesWithNamespaceQuotas=")
          .append(this.tablesWithNamespaceQuotas);
      return sb.toString();
    }
  }
}