Map<Short, Map<Short, ContainerStorageStats>> getContainerStorageStats(long referenceTimeInMs) throws StoreException {
    if (!enabled.get()) {
      throw new StoreException(String.format("BlobStoreStats is not enabled or closing for store %s", storeId),
          StoreErrorCodes.StoreShuttingDown);
    }
    Map<Short, Map<Short, Long>> validSizeMap = null;
    Map<Short, Map<Short, Long>> physicalUsageMap = null;
    Map<Short, Map<Short, Long>> numberStoreKeyMap = null;
    ScanResults currentScanResults = scanResults.get();
    if (currentScanResults != null && isWithinRange(currentScanResults.containerForecastStartTimeMs,
        currentScanResults.containerForecastEndTimeMs, referenceTimeInMs)) {
      validSizeMap = currentScanResults.getValidSizePerContainer(referenceTimeInMs);
      physicalUsageMap = currentScanResults.getContainerPhysicalStorageUsage();
      numberStoreKeyMap = currentScanResults.getContainerNumberOfStoreKeys();
    } else {
      if (isScanning && isWithinRange(indexScanner.newScanResults.containerForecastStartTimeMs,
          indexScanner.newScanResults.containerForecastEndTimeMs, referenceTimeInMs)) {
        scanLock.lock();
        try {
          if (isScanning) {
            if (waitCondition.await(waitTimeoutInSecs, TimeUnit.SECONDS)) {
              currentScanResults = scanResults.get();
              if (isWithinRange(currentScanResults.containerForecastStartTimeMs,
                  currentScanResults.containerForecastEndTimeMs, referenceTimeInMs)) {
                validSizeMap = currentScanResults.getValidSizePerContainer(referenceTimeInMs);
                physicalUsageMap = currentScanResults.getContainerPhysicalStorageUsage();
                numberStoreKeyMap = currentScanResults.getContainerNumberOfStoreKeys();
              }
            } else {
              metrics.blobStoreStatsIndexScannerErrorCount.inc();
              logger.error("Timed out while waiting for BlobStoreStats index scan to complete for store {}", storeId);
            }
          } else {
            currentScanResults = scanResults.get();
            if (isWithinRange(currentScanResults.containerForecastStartTimeMs,
                currentScanResults.containerForecastEndTimeMs, referenceTimeInMs)) {
              validSizeMap = currentScanResults.getValidSizePerContainer(referenceTimeInMs);
              physicalUsageMap = currentScanResults.getContainerPhysicalStorageUsage();
              numberStoreKeyMap = currentScanResults.getContainerNumberOfStoreKeys();
            }
          }
        } catch (InterruptedException e) {
          metrics.blobStoreStatsIndexScannerErrorCount.inc();
          throw new IllegalStateException(
              String.format("Illegal state, wait for scan to complete is interrupted for store %s", storeId), e);
        } finally {
          scanLock.unlock();
        }
      }
      if (validSizeMap == null) {
        // retValue could be null in three scenarios:
        // 1. referenceTimeInMs is outside of current forecast coverage and there is no ongoing scan.
        // 2. timed out while waiting for an ongoing scan.
        // 3. rare edge case where currentScanResults updated twice since the start of the wait.
        return collectContainerStorageStats(referenceTimeInMs);
      }
    }
    Map<Short, Map<Short, ContainerStorageStats>> retValue = new HashMap<>();
    for (short accountId : validSizeMap.keySet()) {
      for (short containerId : validSizeMap.get(accountId).keySet()) {
        retValue.computeIfAbsent(accountId, k -> new HashMap<>())
            .put(containerId, new ContainerStorageStats(containerId, validSizeMap.get(accountId).get(containerId),
                physicalUsageMap.get(accountId).get(containerId), numberStoreKeyMap.get(accountId).get(containerId)));
      }
    }
    return retValue;
  }
Gets the storage stats for all serviceIds and their containerIds as of now (the time when the API is called).
Storage stats is comprised of 3 values: 1. valid data size (logicalStorageUsage) 2. physical data size 3. number of blobs.
The following data are considered as valid data for this API:
1. PUT with no expiry and no corresponding DELETE
2. PUT expiring at t_exp but t_exp_ref < t_exp
3. PUT with corresponding DELETE at time t_delete but t_del_ref < t_delete
For this API, t_ref is specified by the given reference time.
For physical data size, all the records will be taken into consideration, including DELETED PUT, even DELETE record itself.
For number of blobs, it includes all different blob ids.
@param referenceTimeInMs the reference time in ms until which deletes and expiration are relevant
@return the storage stats of each container in the form of a nested {@link Map} of accountId to another map of containerId
to {@link ContainerStorageStats}.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java