001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.quotas;
019
020import java.io.IOException;
021import java.util.Collections;
022import java.util.HashMap;
023import java.util.HashSet;
024import java.util.Iterator;
025import java.util.Map;
026import java.util.Map.Entry;
027import java.util.Optional;
028import java.util.Set;
029import java.util.concurrent.ConcurrentHashMap;
030import org.apache.commons.lang3.builder.HashCodeBuilder;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileSystem;
033import org.apache.hadoop.hbase.DoNotRetryIOException;
034import org.apache.hadoop.hbase.NamespaceDescriptor;
035import org.apache.hadoop.hbase.RegionStateListener;
036import org.apache.hadoop.hbase.TableName;
037import org.apache.hadoop.hbase.client.Connection;
038import org.apache.hadoop.hbase.client.RegionInfo;
039import org.apache.hadoop.hbase.master.MasterServices;
040import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
041import org.apache.hadoop.hbase.master.procedure.SwitchRpcThrottleProcedure;
042import org.apache.hadoop.hbase.namespace.NamespaceAuditor;
043import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
044import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
045import org.apache.yetus.audience.InterfaceAudience;
046import org.apache.yetus.audience.InterfaceStability;
047import org.slf4j.Logger;
048import org.slf4j.LoggerFactory;
049
050import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
051import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
052import org.apache.hbase.thirdparty.com.google.protobuf.TextFormat;
053
054import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
055import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsRpcThrottleEnabledRequest;
056import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsRpcThrottleEnabledResponse;
057import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetQuotaRequest;
058import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetQuotaResponse;
059import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchExceedThrottleQuotaRequest;
060import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchExceedThrottleQuotaResponse;
061import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchRpcThrottleRequest;
062import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchRpcThrottleResponse;
063import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.FileArchiveNotificationRequest;
064import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.FileArchiveNotificationRequest.FileWithSize;
065
066/**
067 * Master Quota Manager. It is responsible for initialize the quota table on the first-run and
068 * provide the admin operations to interact with the quota table. TODO: FUTURE: The master will be
069 * responsible to notify each RS of quota changes and it will do the "quota aggregation" when the
070 * QuotaScope is CLUSTER.
071 */
072@InterfaceAudience.Private
073@InterfaceStability.Evolving
074public class MasterQuotaManager implements RegionStateListener {
075  private static final Logger LOG = LoggerFactory.getLogger(MasterQuotaManager.class);
076  private static final Map<RegionInfo, Long> EMPTY_MAP =
077    Collections.unmodifiableMap(new HashMap<>());
078
079  private final MasterServices masterServices;
080  private NamedLock<String> namespaceLocks;
081  private NamedLock<TableName> tableLocks;
082  private NamedLock<String> userLocks;
083  private NamedLock<String> regionServerLocks;
084  private boolean initialized = false;
085  private NamespaceAuditor namespaceQuotaManager;
086  private ConcurrentHashMap<RegionInfo, SizeSnapshotWithTimestamp> regionSizes;
087
088  public MasterQuotaManager(MasterServices masterServices) {
089    this.masterServices = masterServices;
090  }
091
092  public void start() throws IOException {
093    // If the user doesn't want the quota support skip all the initializations.
094    if (!QuotaUtil.isQuotaEnabled(masterServices.getConfiguration())) {
095      LOG.info("Quota support disabled");
096      return;
097    }
098
099    // Create the quota table if missing
100    if (!masterServices.getTableDescriptors().exists(QuotaUtil.QUOTA_TABLE_NAME)) {
101      LOG.info("Quota table not found. Creating...");
102      createQuotaTable();
103    }
104
105    LOG.info("Initializing quota support");
106    namespaceLocks = new NamedLock<>();
107    tableLocks = new NamedLock<>();
108    userLocks = new NamedLock<>();
109    regionServerLocks = new NamedLock<>();
110    regionSizes = new ConcurrentHashMap<>();
111
112    namespaceQuotaManager = new NamespaceAuditor(masterServices);
113    namespaceQuotaManager.start();
114    initialized = true;
115  }
116
117  public void stop() {
118  }
119
120  public boolean isQuotaInitialized() {
121    return initialized && namespaceQuotaManager.isInitialized();
122  }
123
124  /*
125   * ========================================================================== Admin operations to
126   * manage the quota table
127   */
128  public SetQuotaResponse setQuota(final SetQuotaRequest req)
129    throws IOException, InterruptedException {
130    checkQuotaSupport();
131
132    if (req.hasUserName()) {
133      userLocks.lock(req.getUserName());
134      try {
135        if (req.hasTableName()) {
136          setUserQuota(req.getUserName(), ProtobufUtil.toTableName(req.getTableName()), req);
137        } else if (req.hasNamespace()) {
138          setUserQuota(req.getUserName(), req.getNamespace(), req);
139        } else {
140          setUserQuota(req.getUserName(), req);
141        }
142      } finally {
143        userLocks.unlock(req.getUserName());
144      }
145    } else if (req.hasTableName()) {
146      TableName table = ProtobufUtil.toTableName(req.getTableName());
147      tableLocks.lock(table);
148      try {
149        setTableQuota(table, req);
150      } finally {
151        tableLocks.unlock(table);
152      }
153    } else if (req.hasNamespace()) {
154      namespaceLocks.lock(req.getNamespace());
155      try {
156        setNamespaceQuota(req.getNamespace(), req);
157      } finally {
158        namespaceLocks.unlock(req.getNamespace());
159      }
160    } else if (req.hasRegionServer()) {
161      regionServerLocks.lock(req.getRegionServer());
162      try {
163        setRegionServerQuota(req.getRegionServer(), req);
164      } finally {
165        regionServerLocks.unlock(req.getRegionServer());
166      }
167    } else {
168      throw new DoNotRetryIOException(new UnsupportedOperationException(
169        "a user, a table, a namespace or region server must be specified"));
170    }
171    return SetQuotaResponse.newBuilder().build();
172  }
173
174  public void setUserQuota(final String userName, final SetQuotaRequest req)
175    throws IOException, InterruptedException {
176    setQuota(req, new SetQuotaOperations() {
177      @Override
178      public GlobalQuotaSettingsImpl fetch() throws IOException {
179        return new GlobalQuotaSettingsImpl(req.getUserName(), null, null, null,
180          QuotaUtil.getUserQuota(masterServices.getConnection(), userName));
181      }
182
183      @Override
184      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
185        QuotaUtil.addUserQuota(masterServices.getConnection(), userName, quotaPojo.toQuotas());
186      }
187
188      @Override
189      public void delete() throws IOException {
190        QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName);
191      }
192
193      @Override
194      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
195        masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, quotaPojo);
196      }
197
198      @Override
199      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
200        masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, quotaPojo);
201      }
202    });
203  }
204
205  public void setUserQuota(final String userName, final TableName table, final SetQuotaRequest req)
206    throws IOException, InterruptedException {
207    setQuota(req, new SetQuotaOperations() {
208      @Override
209      public GlobalQuotaSettingsImpl fetch() throws IOException {
210        return new GlobalQuotaSettingsImpl(userName, table, null, null,
211          QuotaUtil.getUserQuota(masterServices.getConnection(), userName, table));
212      }
213
214      @Override
215      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
216        QuotaUtil.addUserQuota(masterServices.getConnection(), userName, table,
217          quotaPojo.toQuotas());
218      }
219
220      @Override
221      public void delete() throws IOException {
222        QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName, table);
223      }
224
225      @Override
226      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
227        masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, table, quotaPojo);
228      }
229
230      @Override
231      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
232        masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, table, quotaPojo);
233      }
234    });
235  }
236
237  public void setUserQuota(final String userName, final String namespace, final SetQuotaRequest req)
238    throws IOException, InterruptedException {
239    setQuota(req, new SetQuotaOperations() {
240      @Override
241      public GlobalQuotaSettingsImpl fetch() throws IOException {
242        return new GlobalQuotaSettingsImpl(userName, null, namespace, null,
243          QuotaUtil.getUserQuota(masterServices.getConnection(), userName, namespace));
244      }
245
246      @Override
247      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
248        QuotaUtil.addUserQuota(masterServices.getConnection(), userName, namespace,
249          quotaPojo.toQuotas());
250      }
251
252      @Override
253      public void delete() throws IOException {
254        QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName, namespace);
255      }
256
257      @Override
258      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
259        masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, namespace, quotaPojo);
260      }
261
262      @Override
263      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
264        masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, namespace, quotaPojo);
265      }
266    });
267  }
268
269  public void setTableQuota(final TableName table, final SetQuotaRequest req)
270    throws IOException, InterruptedException {
271    setQuota(req, new SetQuotaOperations() {
272      @Override
273      public GlobalQuotaSettingsImpl fetch() throws IOException {
274        return new GlobalQuotaSettingsImpl(null, table, null, null,
275          QuotaUtil.getTableQuota(masterServices.getConnection(), table));
276      }
277
278      @Override
279      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
280        QuotaUtil.addTableQuota(masterServices.getConnection(), table, quotaPojo.toQuotas());
281      }
282
283      @Override
284      public void delete() throws IOException {
285        SpaceQuotaSnapshot currSnapshotOfTable =
286          QuotaTableUtil.getCurrentSnapshotFromQuotaTable(masterServices.getConnection(), table);
287        QuotaUtil.deleteTableQuota(masterServices.getConnection(), table);
288        if (currSnapshotOfTable != null) {
289          SpaceQuotaStatus quotaStatus = currSnapshotOfTable.getQuotaStatus();
290          if (
291            SpaceViolationPolicy.DISABLE == quotaStatus.getPolicy().orElse(null)
292              && quotaStatus.isInViolation()
293          ) {
294            QuotaUtil.enableTableIfNotEnabled(masterServices.getConnection(), table);
295          }
296        }
297      }
298
299      @Override
300      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
301        masterServices.getMasterCoprocessorHost().preSetTableQuota(table, quotaPojo);
302      }
303
304      @Override
305      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
306        masterServices.getMasterCoprocessorHost().postSetTableQuota(table, quotaPojo);
307      }
308    });
309  }
310
311  public void setNamespaceQuota(final String namespace, final SetQuotaRequest req)
312    throws IOException, InterruptedException {
313    setQuota(req, new SetQuotaOperations() {
314      @Override
315      public GlobalQuotaSettingsImpl fetch() throws IOException {
316        return new GlobalQuotaSettingsImpl(null, null, namespace, null,
317          QuotaUtil.getNamespaceQuota(masterServices.getConnection(), namespace));
318      }
319
320      @Override
321      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
322        QuotaUtil.addNamespaceQuota(masterServices.getConnection(), namespace,
323          quotaPojo.toQuotas());
324      }
325
326      @Override
327      public void delete() throws IOException {
328        QuotaUtil.deleteNamespaceQuota(masterServices.getConnection(), namespace);
329      }
330
331      @Override
332      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
333        masterServices.getMasterCoprocessorHost().preSetNamespaceQuota(namespace, quotaPojo);
334      }
335
336      @Override
337      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
338        masterServices.getMasterCoprocessorHost().postSetNamespaceQuota(namespace, quotaPojo);
339      }
340    });
341  }
342
343  public void setRegionServerQuota(final String regionServer, final SetQuotaRequest req)
344    throws IOException, InterruptedException {
345    setQuota(req, new SetQuotaOperations() {
346      @Override
347      public GlobalQuotaSettingsImpl fetch() throws IOException {
348        return new GlobalQuotaSettingsImpl(null, null, null, regionServer,
349          QuotaUtil.getRegionServerQuota(masterServices.getConnection(), regionServer));
350      }
351
352      @Override
353      public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
354        QuotaUtil.addRegionServerQuota(masterServices.getConnection(), regionServer,
355          quotaPojo.toQuotas());
356      }
357
358      @Override
359      public void delete() throws IOException {
360        QuotaUtil.deleteRegionServerQuota(masterServices.getConnection(), regionServer);
361      }
362
363      @Override
364      public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
365        masterServices.getMasterCoprocessorHost().preSetRegionServerQuota(regionServer, quotaPojo);
366      }
367
368      @Override
369      public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException {
370        masterServices.getMasterCoprocessorHost().postSetRegionServerQuota(regionServer, quotaPojo);
371      }
372    });
373  }
374
375  public void setNamespaceQuota(NamespaceDescriptor desc) throws IOException {
376    if (initialized) {
377      this.namespaceQuotaManager.addNamespace(desc);
378    }
379  }
380
381  public void removeNamespaceQuota(String namespace) throws IOException {
382    if (initialized) {
383      this.namespaceQuotaManager.deleteNamespace(namespace);
384    }
385  }
386
387  public SwitchRpcThrottleResponse switchRpcThrottle(SwitchRpcThrottleRequest request)
388    throws IOException {
389    boolean rpcThrottle = request.getRpcThrottleEnabled();
390    if (initialized) {
391      masterServices.getMasterCoprocessorHost().preSwitchRpcThrottle(rpcThrottle);
392      boolean oldRpcThrottle = masterServices.getRpcThrottleStateStore().get();
393      if (rpcThrottle != oldRpcThrottle) {
394        LOG.info("{} switch rpc throttle from {} to {}", masterServices.getClientIdAuditPrefix(),
395          oldRpcThrottle, rpcThrottle);
396        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
397        SwitchRpcThrottleProcedure procedure =
398          new SwitchRpcThrottleProcedure(rpcThrottle, masterServices.getServerName(), latch);
399        masterServices.getMasterProcedureExecutor().submitProcedure(procedure);
400        latch.await();
401      } else {
402        LOG.warn("Skip switch rpc throttle to {} because it's the same with old value",
403          rpcThrottle);
404      }
405      SwitchRpcThrottleResponse response = SwitchRpcThrottleResponse.newBuilder()
406        .setPreviousRpcThrottleEnabled(oldRpcThrottle).build();
407      masterServices.getMasterCoprocessorHost().postSwitchRpcThrottle(oldRpcThrottle, rpcThrottle);
408      return response;
409    } else {
410      LOG.warn("Skip switch rpc throttle to {} because rpc quota is disabled", rpcThrottle);
411      return SwitchRpcThrottleResponse.newBuilder().setPreviousRpcThrottleEnabled(false).build();
412    }
413  }
414
415  public IsRpcThrottleEnabledResponse isRpcThrottleEnabled(IsRpcThrottleEnabledRequest request)
416    throws IOException {
417    if (initialized) {
418      masterServices.getMasterCoprocessorHost().preIsRpcThrottleEnabled();
419      boolean enabled = isRpcThrottleEnabled();
420      IsRpcThrottleEnabledResponse response =
421        IsRpcThrottleEnabledResponse.newBuilder().setRpcThrottleEnabled(enabled).build();
422      masterServices.getMasterCoprocessorHost().postIsRpcThrottleEnabled(enabled);
423      return response;
424    } else {
425      LOG.warn("Skip get rpc throttle because rpc quota is disabled");
426      return IsRpcThrottleEnabledResponse.newBuilder().setRpcThrottleEnabled(false).build();
427    }
428  }
429
430  public boolean isRpcThrottleEnabled() throws IOException {
431    return initialized && masterServices.getRpcThrottleStateStore().get();
432  }
433
434  public SwitchExceedThrottleQuotaResponse
435    switchExceedThrottleQuota(SwitchExceedThrottleQuotaRequest request) throws IOException {
436    boolean enabled = request.getExceedThrottleQuotaEnabled();
437    if (initialized) {
438      masterServices.getMasterCoprocessorHost().preSwitchExceedThrottleQuota(enabled);
439      boolean previousEnabled =
440        QuotaUtil.isExceedThrottleQuotaEnabled(masterServices.getConnection());
441      if (previousEnabled == enabled) {
442        LOG.warn("Skip switch exceed throttle quota to {} because it's the same with old value",
443          enabled);
444      } else {
445        QuotaUtil.switchExceedThrottleQuota(masterServices.getConnection(), enabled);
446        LOG.info("{} switch exceed throttle quota from {} to {}",
447          masterServices.getClientIdAuditPrefix(), previousEnabled, enabled);
448      }
449      SwitchExceedThrottleQuotaResponse response = SwitchExceedThrottleQuotaResponse.newBuilder()
450        .setPreviousExceedThrottleQuotaEnabled(previousEnabled).build();
451      masterServices.getMasterCoprocessorHost().postSwitchExceedThrottleQuota(previousEnabled,
452        enabled);
453      return response;
454    } else {
455      LOG.warn("Skip switch exceed throttle quota to {} because quota is disabled", enabled);
456      return SwitchExceedThrottleQuotaResponse.newBuilder()
457        .setPreviousExceedThrottleQuotaEnabled(false).build();
458    }
459  }
460
461  public boolean isExceedThrottleQuotaEnabled() throws IOException {
462    return initialized
463      ? QuotaUtil.isExceedThrottleQuotaEnabled(masterServices.getConnection())
464      : false;
465  }
466
467  private void setQuota(final SetQuotaRequest req, final SetQuotaOperations quotaOps)
468    throws IOException, InterruptedException {
469    if (req.hasRemoveAll() && req.getRemoveAll() == true) {
470      quotaOps.preApply(null);
471      quotaOps.delete();
472      quotaOps.postApply(null);
473      return;
474    }
475
476    // Apply quota changes
477    GlobalQuotaSettingsImpl currentQuota = quotaOps.fetch();
478    if (LOG.isTraceEnabled()) {
479      LOG.trace(
480        "Current quota for request(" + TextFormat.shortDebugString(req) + "): " + currentQuota);
481    }
482    // Call the appropriate "pre" CP hook with the current quota value (may be null)
483    quotaOps.preApply(currentQuota);
484    // Translate the protobuf request back into a POJO
485    QuotaSettings newQuota = QuotaSettings.buildFromProto(req);
486    if (LOG.isTraceEnabled()) {
487      LOG.trace("Deserialized quota from request: " + newQuota);
488    }
489
490    // Merge the current quota settings with the new quota settings the user provided.
491    //
492    // NB: while SetQuotaRequest technically allows for multi types of quotas to be set in one
493    // message, the Java API (in Admin/AsyncAdmin) does not. Assume there is only one type.
494    GlobalQuotaSettingsImpl mergedQuota = currentQuota.merge(newQuota);
495    if (LOG.isTraceEnabled()) {
496      LOG.trace("Computed merged quota from current quota and user request: " + mergedQuota);
497    }
498
499    // Submit new changes
500    if (mergedQuota == null) {
501      quotaOps.delete();
502    } else {
503      quotaOps.update(mergedQuota);
504    }
505    // Advertise the final result via the "post" CP hook
506    quotaOps.postApply(mergedQuota);
507  }
508
509  public void checkNamespaceTableAndRegionQuota(TableName tName, int regions) throws IOException {
510    if (initialized) {
511      namespaceQuotaManager.checkQuotaToCreateTable(tName, regions);
512    }
513  }
514
515  public void checkAndUpdateNamespaceRegionQuota(TableName tName, int regions) throws IOException {
516    if (initialized) {
517      namespaceQuotaManager.checkQuotaToUpdateRegion(tName, regions);
518    }
519  }
520
521  /** Returns cached region count, or -1 if quota manager is disabled or table status not found */
522  public int getRegionCountOfTable(TableName tName) throws IOException {
523    if (initialized) {
524      return namespaceQuotaManager.getRegionCountOfTable(tName);
525    }
526    return -1;
527  }
528
529  @Override
530  public void onRegionMerged(RegionInfo mergedRegion) throws IOException {
531    if (initialized) {
532      namespaceQuotaManager.updateQuotaForRegionMerge(mergedRegion);
533    }
534  }
535
536  @Override
537  public void onRegionSplit(RegionInfo hri) throws IOException {
538    if (initialized) {
539      namespaceQuotaManager.checkQuotaToSplitRegion(hri);
540    }
541  }
542
543  /**
544   * Remove table from namespace quota.
545   * @param tName - The table name to update quota usage.
546   * @throws IOException Signals that an I/O exception has occurred.
547   */
548  public void removeTableFromNamespaceQuota(TableName tName) throws IOException {
549    if (initialized) {
550      namespaceQuotaManager.removeFromNamespaceUsage(tName);
551    }
552  }
553
554  public NamespaceAuditor getNamespaceQuotaManager() {
555    return this.namespaceQuotaManager;
556  }
557
558  /**
559   * Encapsulates CRUD quota operations for some subject.
560   */
561  private static interface SetQuotaOperations {
562    /**
563     * Fetches the current quota settings for the subject.
564     */
565    GlobalQuotaSettingsImpl fetch() throws IOException;
566
567    /**
568     * Deletes the quota for the subject.
569     */
570    void delete() throws IOException;
571
572    /**
573     * Persist the given quota for the subject.
574     */
575    void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException;
576
577    /**
578     * Performs some action before {@link #update(GlobalQuotaSettingsImpl)} with the current quota
579     * for the subject.
580     */
581    void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException;
582
583    /**
584     * Performs some action after {@link #update(GlobalQuotaSettingsImpl)} with the resulting quota
585     * from the request action for the subject.
586     */
587    void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException;
588  }
589
590  /*
591   * ========================================================================== Helpers
592   */
593
594  private void checkQuotaSupport() throws IOException {
595    if (!QuotaUtil.isQuotaEnabled(masterServices.getConfiguration())) {
596      throw new DoNotRetryIOException(new UnsupportedOperationException("quota support disabled"));
597    }
598    if (!initialized) {
599      long maxWaitTime = masterServices.getConfiguration()
600        .getLong("hbase.master.wait.for.quota.manager.init", 30000); // default is 30 seconds.
601      long startTime = EnvironmentEdgeManager.currentTime();
602      do {
603        try {
604          Thread.sleep(100);
605        } catch (InterruptedException e) {
606          LOG.warn("Interrupted while waiting for Quota Manager to be initialized.");
607          break;
608        }
609      } while (!initialized && (EnvironmentEdgeManager.currentTime() - startTime) < maxWaitTime);
610      if (!initialized) {
611        throw new IOException("Quota manager is uninitialized, please retry later.");
612      }
613    }
614  }
615
616  private void createQuotaTable() throws IOException {
617    masterServices.createSystemTable(QuotaUtil.QUOTA_TABLE_DESC);
618  }
619
620  private static class NamedLock<T> {
621    private final HashSet<T> locks = new HashSet<>();
622
623    public void lock(final T name) throws InterruptedException {
624      synchronized (locks) {
625        while (locks.contains(name)) {
626          locks.wait();
627        }
628        locks.add(name);
629      }
630    }
631
632    public void unlock(final T name) {
633      synchronized (locks) {
634        locks.remove(name);
635        locks.notifyAll();
636      }
637    }
638  }
639
640  @Override
641  public void onRegionSplitReverted(RegionInfo hri) throws IOException {
642    if (initialized) {
643      this.namespaceQuotaManager.removeRegionFromNamespaceUsage(hri);
644    }
645  }
646
647  /**
648   * Holds the size of a region at the given time, millis since the epoch.
649   */
650  private static class SizeSnapshotWithTimestamp {
651    private final long size;
652    private final long time;
653
654    public SizeSnapshotWithTimestamp(long size, long time) {
655      this.size = size;
656      this.time = time;
657    }
658
659    public long getSize() {
660      return size;
661    }
662
663    public long getTime() {
664      return time;
665    }
666
667    @Override
668    public boolean equals(Object o) {
669      if (o instanceof SizeSnapshotWithTimestamp) {
670        SizeSnapshotWithTimestamp other = (SizeSnapshotWithTimestamp) o;
671        return size == other.size && time == other.time;
672      }
673      return false;
674    }
675
676    @Override
677    public int hashCode() {
678      HashCodeBuilder hcb = new HashCodeBuilder();
679      return hcb.append(size).append(time).toHashCode();
680    }
681
682    @Override
683    public String toString() {
684      StringBuilder sb = new StringBuilder(32);
685      sb.append("SizeSnapshotWithTimestamp={size=").append(size).append("B, ");
686      sb.append("time=").append(time).append("}");
687      return sb.toString();
688    }
689  }
690
691  void initializeRegionSizes() {
692    assert regionSizes == null;
693    this.regionSizes = new ConcurrentHashMap<>();
694  }
695
696  public void addRegionSize(RegionInfo hri, long size, long time) {
697    if (regionSizes == null) {
698      return;
699    }
700    regionSizes.put(hri, new SizeSnapshotWithTimestamp(size, time));
701  }
702
703  public Map<RegionInfo, Long> snapshotRegionSizes() {
704    if (regionSizes == null) {
705      return EMPTY_MAP;
706    }
707
708    Map<RegionInfo, Long> copy = new HashMap<>();
709    for (Entry<RegionInfo, SizeSnapshotWithTimestamp> entry : regionSizes.entrySet()) {
710      copy.put(entry.getKey(), entry.getValue().getSize());
711    }
712    return copy;
713  }
714
715  int pruneEntriesOlderThan(long timeToPruneBefore, QuotaObserverChore quotaObserverChore) {
716    if (regionSizes == null) {
717      return 0;
718    }
719    int numEntriesRemoved = 0;
720    Iterator<Entry<RegionInfo, SizeSnapshotWithTimestamp>> iterator =
721      regionSizes.entrySet().iterator();
722    while (iterator.hasNext()) {
723      RegionInfo regionInfo = iterator.next().getKey();
724      long currentEntryTime = regionSizes.get(regionInfo).getTime();
725      // do not prune the entries if table is in violation and
726      // violation policy is disable to avoid cycle of enable/disable.
727      // Please refer HBASE-22012 for more details.
728      // prune entries older than time.
729      if (
730        currentEntryTime < timeToPruneBefore
731          && !isInViolationAndPolicyDisable(regionInfo.getTable(), quotaObserverChore)
732      ) {
733        iterator.remove();
734        numEntriesRemoved++;
735      }
736    }
737    return numEntriesRemoved;
738  }
739
740  /**
741   * Method to check if a table is in violation and policy set on table is DISABLE.
742   * @param tableName          tableName to check.
743   * @param quotaObserverChore QuotaObserverChore instance
744   * @return returns true if table is in violation and policy is disable else false.
745   */
746  private boolean isInViolationAndPolicyDisable(TableName tableName,
747    QuotaObserverChore quotaObserverChore) {
748    boolean isInViolationAtTable = false;
749    boolean isInViolationAtNamespace = false;
750    SpaceViolationPolicy tablePolicy = null;
751    SpaceViolationPolicy namespacePolicy = null;
752    // Get Current Snapshot for the given table
753    SpaceQuotaSnapshot tableQuotaSnapshot = quotaObserverChore.getTableQuotaSnapshot(tableName);
754    SpaceQuotaSnapshot namespaceQuotaSnapshot =
755      quotaObserverChore.getNamespaceQuotaSnapshot(tableName.getNamespaceAsString());
756    if (tableQuotaSnapshot != null) {
757      // check if table in violation
758      isInViolationAtTable = tableQuotaSnapshot.getQuotaStatus().isInViolation();
759      Optional<SpaceViolationPolicy> policy = tableQuotaSnapshot.getQuotaStatus().getPolicy();
760      if (policy.isPresent()) {
761        tablePolicy = policy.get();
762      }
763    }
764    if (namespaceQuotaSnapshot != null) {
765      // check namespace in violation
766      isInViolationAtNamespace = namespaceQuotaSnapshot.getQuotaStatus().isInViolation();
767      Optional<SpaceViolationPolicy> policy = namespaceQuotaSnapshot.getQuotaStatus().getPolicy();
768      if (policy.isPresent()) {
769        namespacePolicy = policy.get();
770      }
771    }
772    return (tablePolicy == SpaceViolationPolicy.DISABLE && isInViolationAtTable)
773      || (namespacePolicy == SpaceViolationPolicy.DISABLE && isInViolationAtNamespace);
774  }
775
776  public void processFileArchivals(FileArchiveNotificationRequest request, Connection conn,
777    Configuration conf, FileSystem fs) throws IOException {
778    final HashMultimap<TableName, Entry<String, Long>> archivedFilesByTable = HashMultimap.create();
779    // Group the archived files by table
780    for (FileWithSize fileWithSize : request.getArchivedFilesList()) {
781      TableName tn = ProtobufUtil.toTableName(fileWithSize.getTableName());
782      archivedFilesByTable.put(tn,
783        Maps.immutableEntry(fileWithSize.getName(), fileWithSize.getSize()));
784    }
785    if (LOG.isTraceEnabled()) {
786      LOG.trace("Grouped archived files by table: " + archivedFilesByTable);
787    }
788    // Report each set of files to the appropriate object
789    for (TableName tn : archivedFilesByTable.keySet()) {
790      final Set<Entry<String, Long>> filesWithSize = archivedFilesByTable.get(tn);
791      final FileArchiverNotifier notifier =
792        FileArchiverNotifierFactoryImpl.getInstance().get(conn, conf, fs, tn);
793      notifier.addArchivedFiles(filesWithSize);
794    }
795  }
796
797  /**
798   * Removes each region size entry where the RegionInfo references the provided TableName.
799   * @param tableName tableName.
800   */
801  public void removeRegionSizesForTable(TableName tableName) {
802    regionSizes.keySet().removeIf(regionInfo -> regionInfo.getTable().equals(tableName));
803  }
804}