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

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

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.util.FSUtils;

/**
 * Thread that walks over the filesystem, and computes the mappings
 * Region -> BestHost and Region -> {@code Map<HostName, fractional-locality-of-region>}
 *
 */
@InterfaceAudience.Private
class FSRegionScanner implements Runnable {
  static private final Logger LOG = LoggerFactory.getLogger(FSRegionScanner.class);

  private Path regionPath;

  /**
   * The file system used
   */
  private FileSystem fs;

  /**
   * Maps each region to the RS with highest locality for that region.
   */
  private final Map<String,String> regionToBestLocalityRSMapping;

  /**
   * Maps region encoded names to maps of hostnames to fractional locality of
   * that region on that host.
   */
  private Map<String, Map<String, Float>> regionDegreeLocalityMapping;

  FSRegionScanner(FileSystem fs, Path regionPath,
                  Map<String, String> regionToBestLocalityRSMapping,
                  Map<String, Map<String, Float>> regionDegreeLocalityMapping) {
    this.fs = fs;
    this.regionPath = regionPath;
    this.regionToBestLocalityRSMapping = regionToBestLocalityRSMapping;
    this.regionDegreeLocalityMapping = regionDegreeLocalityMapping;
  }

  @Override
  public void run() {
    try {
      // empty the map for each region
      Map<String, AtomicInteger> blockCountMap = new HashMap<>();

      //get table name
      String tableName = regionPath.getParent().getName();
      int totalBlkCount = 0;

      // ignore null
      FileStatus[] cfList = fs.listStatus(regionPath, new FSUtils.FamilyDirFilter(fs));
      if (null == cfList) {
        return;
      }

      // for each cf, get all the blocks information
      for (FileStatus cfStatus : cfList) {
        if (!cfStatus.isDirectory()) {
          // skip because this is not a CF directory
          continue;
        }

        FileStatus[] storeFileLists = fs.listStatus(cfStatus.getPath());
        if (null == storeFileLists) {
          continue;
        }

        for (FileStatus storeFile : storeFileLists) {
          BlockLocation[] blkLocations =
            fs.getFileBlockLocations(storeFile, 0, storeFile.getLen());
          if (null == blkLocations) {
            continue;
          }

          totalBlkCount += blkLocations.length;
          for(BlockLocation blk: blkLocations) {
            for (String host: blk.getHosts()) {
              AtomicInteger count = blockCountMap.get(host);
              if (count == null) {
                count = new AtomicInteger(0);
                blockCountMap.put(host, count);
              }
              count.incrementAndGet();
            }
          }
        }
      }

      if (regionToBestLocalityRSMapping != null) {
        int largestBlkCount = 0;
        String hostToRun = null;
        for (Map.Entry<String, AtomicInteger> entry : blockCountMap.entrySet()) {
          String host = entry.getKey();

          int tmp = entry.getValue().get();
          if (tmp > largestBlkCount) {
            largestBlkCount = tmp;
            hostToRun = host;
          }
        }

        // empty regions could make this null
        if (null == hostToRun) {
          return;
        }

        if (hostToRun.endsWith(".")) {
          hostToRun = hostToRun.substring(0, hostToRun.length()-1);
        }
        String name = tableName + ":" + regionPath.getName();
        synchronized (regionToBestLocalityRSMapping) {
          regionToBestLocalityRSMapping.put(name,  hostToRun);
        }
      }

      if (regionDegreeLocalityMapping != null && totalBlkCount > 0) {
        Map<String, Float> hostLocalityMap = new HashMap<>();
        for (Map.Entry<String, AtomicInteger> entry : blockCountMap.entrySet()) {
          String host = entry.getKey();
          if (host.endsWith(".")) {
            host = host.substring(0, host.length() - 1);
          }
          // Locality is fraction of blocks local to this host.
          float locality = ((float)entry.getValue().get()) / totalBlkCount;
          hostLocalityMap.put(host, locality);
        }
        // Put the locality map into the result map, keyed by the encoded name
        // of the region.
        regionDegreeLocalityMapping.put(regionPath.getName(), hostLocalityMap);
      }
    } catch (IOException e) {
      LOG.warn("Problem scanning file system", e);
    } catch (RuntimeException e) {
      LOG.warn("Problem scanning file system", e);
    }
  }
}