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

import org.apache.storm.Config;
import org.apache.storm.blobstore.ClientBlobStore;
import org.apache.storm.blobstore.InputStreamWithMeta;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.KeyNotFoundException;
import org.apache.storm.utils.ShellUtils.ExitCodeException;
import org.apache.storm.utils.ShellUtils.ShellCommandExecutor;
import org.apache.storm.utils.Utils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.PrintWriter;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
import static org.apache.storm.utils.Utils.OR;

/**
 * Class to download and manage files from the blobstore.  It uses an LRU cache
 * to determine which files to keep so they can be reused and which files to delete.
 */
public class Localizer {
  public static final Logger LOG = LoggerFactory.getLogger(Localizer.class);
  public static final String USERCACHE = "usercache";
  public static final String FILECACHE = "filecache";

  // sub directories to store either files or uncompressed archives respectively
  public static final String FILESDIR = "files";
  public static final String ARCHIVESDIR = "archives";

  private static final String TO_UNCOMPRESS = "_tmp_";
  
  
  
  private final Map<String, Object> _conf;
  private final int _threadPoolSize;
  // thread pool for initial download
  private final ExecutorService _execService;
  // thread pool for updates
  private final ExecutorService _updateExecService;
  private final int _blobDownloadRetries;

  // track resources - user to resourceSet
  private final ConcurrentMap<String, LocalizedResourceSet> _userRsrc = new
      ConcurrentHashMap<String, LocalizedResourceSet>();

  private final String _localBaseDir;

  // cleanup
  private long _cacheTargetSize;
  private long _cacheCleanupPeriod;
  private ScheduledExecutorService _cacheCleanupService;

  public Localizer(Map<String, Object> conf, String baseDir) {
    _conf = conf;
    _localBaseDir = baseDir;
    // default cache size 10GB, converted to Bytes
    _cacheTargetSize = Utils.getInt(_conf.get(Config.SUPERVISOR_LOCALIZER_CACHE_TARGET_SIZE_MB),
            10 * 1024).longValue() << 20;
    // default 10 minutes.
    _cacheCleanupPeriod = Utils.getInt(_conf.get(
            Config.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS), 10 * 60 * 1000).longValue();

    // if we needed we could make config for update thread pool size
    _threadPoolSize = Utils.getInt(_conf.get(Config.SUPERVISOR_BLOBSTORE_DOWNLOAD_THREAD_COUNT), 5);
    _blobDownloadRetries = Utils.getInt(_conf.get(
            Config.SUPERVISOR_BLOBSTORE_DOWNLOAD_MAX_RETRIES), 3);

    _execService = Executors.newFixedThreadPool(_threadPoolSize);
    _updateExecService = Executors.newFixedThreadPool(_threadPoolSize);
    reconstructLocalizedResources();
  }

  // For testing, it allows setting size in bytes
  protected void setTargetCacheSize(long size) {
    _cacheTargetSize = size;
  }

  // For testing, be careful as it doesn't clone
  ConcurrentMap<String, LocalizedResourceSet> getUserResources() {
    return _userRsrc;
  }

  public void startCleaner() {
    _cacheCleanupService = new ScheduledThreadPoolExecutor(1,
        new ThreadFactoryBuilder()
            .setNameFormat("Localizer Cache Cleanup")
            .build());

    _cacheCleanupService.scheduleWithFixedDelay(new Runnable() {
          @Override
          public void run() {
            handleCacheCleanup();
          }
        }, _cacheCleanupPeriod, _cacheCleanupPeriod, TimeUnit.MILLISECONDS);
  }

  public void shutdown() {
    if (_cacheCleanupService != null) {
      _cacheCleanupService.shutdown();
    }
    if (_execService != null) {
      _execService.shutdown();
    }
    if (_updateExecService != null) {
      _updateExecService.shutdown();
    }
  }

  // baseDir/supervisor/usercache/
  protected File getUserCacheDir() {
    return new File(_localBaseDir, USERCACHE);
  }

  // baseDir/supervisor/usercache/user1/
  protected File getLocalUserDir(String userName) {
    return new File(getUserCacheDir(), userName);
  }

  // baseDir/supervisor/usercache/user1/filecache
  public File getLocalUserFileCacheDir(String userName) {
    return new File(getLocalUserDir(userName), FILECACHE);
  }

  // baseDir/supervisor/usercache/user1/filecache/files
  protected File getCacheDirForFiles(File dir) {
    return new File(dir, FILESDIR);
  }

  // get the directory to put uncompressed archives in
  // baseDir/supervisor/usercache/user1/filecache/archives
  protected File getCacheDirForArchives(File dir) {
    return new File(dir, ARCHIVESDIR);
  }

  protected void addLocalizedResourceInDir(String dir, LocalizedResourceSet lrsrcSet,
      boolean uncompress) {
    File[] lrsrcs = readCurrentBlobs(dir);

    if (lrsrcs != null) {
      for (File rsrc : lrsrcs) {
        LOG.info("add localized in dir found: " + rsrc);
        /// strip off .suffix
        String path = rsrc.getPath();
        int p = path.lastIndexOf('.');
        if (p > 0) {
          path = path.substring(0, p);
        }
        LOG.debug("local file is: {} path is: {}", rsrc.getPath(), path);
        LocalizedResource lrsrc = new LocalizedResource(new File(path).getName(), path,
            uncompress);
        lrsrcSet.add(lrsrc.getKey(), lrsrc, uncompress);
      }
    }
  }

  protected File[] readDirContents(String location) {
    File dir = new File(location);
    File[] files = null;
    if (dir.exists()) {
      files = dir.listFiles();
    }
    return files;
  }

  // Looks for files in the directory with .current suffix
  protected File[] readCurrentBlobs(String location) {
    File dir = new File(location);
    File[] files = null;
    if (dir.exists()) {
      files = dir.listFiles(new FilenameFilter() {
        @Override
        public boolean accept(File dir, String name) {
          return name.toLowerCase().endsWith(Utils.DEFAULT_CURRENT_BLOB_SUFFIX);
        }
      });
    }
    return files;
  }

  // Check to see if there are any existing files already localized.
  protected void reconstructLocalizedResources() {
    try {
      LOG.info("Reconstruct localized resource: " + getUserCacheDir().getPath());
      File[] users = readDirContents(getUserCacheDir().getPath());

      if (users != null) {
        for (File userDir : users) {
          String user = userDir.getName();
          LOG.debug("looking in: {} for user: {}", userDir.getPath(), user);
          LocalizedResourceSet newSet = new LocalizedResourceSet(user);
          LocalizedResourceSet lrsrcSet = _userRsrc.putIfAbsent(user, newSet);
          if (lrsrcSet == null) {
            lrsrcSet = newSet;
          }
          addLocalizedResourceInDir(getCacheDirForFiles(getLocalUserFileCacheDir(user)).getPath(),
              lrsrcSet, false);
          addLocalizedResourceInDir(
              getCacheDirForArchives(getLocalUserFileCacheDir(user)).getPath(),
              lrsrcSet, true);
        }
      } else {
        LOG.warn("No left over resources found for any user during reconstructing of local resources at: {}", getUserCacheDir().getPath());
      }
    } catch (Exception e) {
      LOG.error("ERROR reconstructing localized resources", e);
    }
  }

  // ignores invalid user/topo/key
  public synchronized void removeBlobReference(String key, String user, String topo,
      boolean uncompress) throws AuthorizationException, KeyNotFoundException {
    LocalizedResourceSet lrsrcSet = _userRsrc.get(user);
    if (lrsrcSet != null) {
      LocalizedResource lrsrc = lrsrcSet.get(key, uncompress);
      if (lrsrc != null) {
        LOG.debug("removing blob reference to: {} for topo: {}", key, topo);
        lrsrc.removeReference(topo);
      } else {
        LOG.warn("trying to remove non-existent blob, key: " + key + " for user: " + user +
            " topo: " + topo);
      }
    } else {
      LOG.warn("trying to remove blob for non-existent resource set for user: " + user + " key: "
          + key + " topo: " + topo);
    }
  }

  public synchronized void addReferences(List<LocalResource> localresource, String user,
       String topo) {
    LocalizedResourceSet lrsrcSet = _userRsrc.get(user);
    if (lrsrcSet != null) {
      for (LocalResource blob : localresource) {
        LocalizedResource lrsrc = lrsrcSet.get(blob.getBlobName(), blob.shouldUncompress());
        if (lrsrc != null) {
          lrsrc.addReference(topo);
          LOG.debug("added reference for topo: {} key: {}", topo, blob);
        } else {
          LOG.warn("trying to add reference to non-existent blob, key: " + blob + " topo: " + topo);
        }
      }
    } else {
      LOG.warn("trying to add reference to non-existent local resource set, " +
          "user: " + user + " topo: " + topo);
    }
  }

  /**
   * This function either returns the blob in the existing cache or if it doesn't exist in the
   * cache, it will download the blob and will block until the download is complete.
   */
  public LocalizedResource getBlob(LocalResource localResource, String user, String topo,
       File userFileDir) throws AuthorizationException, KeyNotFoundException, IOException {
    ArrayList<LocalResource> arr = new ArrayList<LocalResource>();
    arr.add(localResource);
    List<LocalizedResource> results = getBlobs(arr, user, topo, userFileDir);
    if (results.isEmpty() || results.size() != 1) {
      throw new IOException("Unknown error getting blob: " + localResource + ", for user: " + user +
          ", topo: " + topo);
    }
    return results.get(0);
  }

  protected boolean isLocalizedResourceDownloaded(LocalizedResource lrsrc) {
    File rsrcFileCurrent = new File(lrsrc.getCurrentSymlinkPath());
    File rsrcFileWithVersion = new File(lrsrc.getFilePathWithVersion());
    File versionFile = new File(lrsrc.getVersionFilePath());
    return (rsrcFileWithVersion.exists() && rsrcFileCurrent.exists() && versionFile.exists());
  }

  protected boolean isLocalizedResourceUpToDate(LocalizedResource lrsrc,
      ClientBlobStore blobstore) throws AuthorizationException, KeyNotFoundException {
    String localFile = lrsrc.getFilePath();
    long nimbusBlobVersion = Utils.nimbusVersionOfBlob(lrsrc.getKey(), blobstore);
    long currentBlobVersion = Utils.localVersionOfBlob(localFile);
    return (nimbusBlobVersion == currentBlobVersion);
  }

  protected ClientBlobStore getClientBlobStore() {
    return Utils.getClientBlobStoreForSupervisor(_conf);
  }

  /**
   * This function updates blobs on the supervisor. It uses a separate thread pool and runs
   * asynchronously of the download and delete.
   */
  public List<LocalizedResource> updateBlobs(List<LocalResource> localResources,
       String user) throws AuthorizationException, KeyNotFoundException, IOException {
    LocalizedResourceSet lrsrcSet = _userRsrc.get(user);
    ArrayList<LocalizedResource> results = new ArrayList<>();
    ArrayList<Callable<LocalizedResource>> updates = new ArrayList<>();

    if (lrsrcSet == null) {
      // resource set must have been removed
      return results;
    }
    ClientBlobStore blobstore = null;
    try {
      blobstore = getClientBlobStore();
      for (LocalResource localResource: localResources) {
        String key = localResource.getBlobName();
        LocalizedResource lrsrc = lrsrcSet.get(key, localResource.shouldUncompress());
        if (lrsrc == null) {
          LOG.warn("blob requested for update doesn't exist: {}", key);
          continue;
        } else if ((boolean)OR(_conf.get(Config.DISABLE_SYMLINKS), false)) {
          LOG.warn("symlinks are disabled so blobs cannot be downloaded.");
          continue;
        } else {
          // update it if either the version isn't the latest or if any local blob files are missing
          if (!isLocalizedResourceUpToDate(lrsrc, blobstore) ||
              !isLocalizedResourceDownloaded(lrsrc)) {
            LOG.debug("updating blob: {}", key);
            updates.add(new DownloadBlob(this, _conf, key, new File(lrsrc.getFilePath()), user,
                lrsrc.isUncompressed(), true));
          }
        }
      }
    } finally {
      if(blobstore != null) {
        blobstore.shutdown();
      }
    }
    try {
      List<Future<LocalizedResource>> futures = _updateExecService.invokeAll(updates);
      for (Future<LocalizedResource> futureRsrc : futures) {
        try {
          LocalizedResource lrsrc = futureRsrc.get();
          // put the resource just in case it was removed at same time by the cleaner
          LocalizedResourceSet newSet = new LocalizedResourceSet(user);
          LocalizedResourceSet newlrsrcSet = _userRsrc.putIfAbsent(user, newSet);
          if (newlrsrcSet == null) {
            newlrsrcSet = newSet;
          }
          newlrsrcSet.putIfAbsent(lrsrc.getKey(), lrsrc, lrsrc.isUncompressed());
          results.add(lrsrc);
        }
        catch (ExecutionException e) {
          LOG.error("Error updating blob: ", e);
          if (e.getCause() instanceof AuthorizationException) {
            throw (AuthorizationException)e.getCause();
          }
          if (e.getCause() instanceof KeyNotFoundException) {
            throw (KeyNotFoundException)e.getCause();
          }
        }
      }
    } catch (RejectedExecutionException re) {
      LOG.error("Error updating blobs : ", re);
    } catch (InterruptedException ie) {
      throw new IOException("Interrupted Exception", ie);
    }
    return results;
  }

  /**
   * This function either returns the blobs in the existing cache or if they don't exist in the
   * cache, it downloads them in parallel (up to SUPERVISOR_BLOBSTORE_DOWNLOAD_THREAD_COUNT)
   * and will block until all of them have been downloaded
   */
  public synchronized List<LocalizedResource> getBlobs(List<LocalResource> localResources,
      String user, String topo, File userFileDir)
      throws AuthorizationException, KeyNotFoundException, IOException {
    if ((boolean)OR(_conf.get(Config.DISABLE_SYMLINKS), false)) {
      throw new KeyNotFoundException("symlinks are disabled so blobs cannot be downloaded.");
    }
    LocalizedResourceSet newSet = new LocalizedResourceSet(user);
    LocalizedResourceSet lrsrcSet = _userRsrc.putIfAbsent(user, newSet);
    if (lrsrcSet == null) {
      lrsrcSet = newSet;
    }
    ArrayList<LocalizedResource> results = new ArrayList<>();
    ArrayList<Callable<LocalizedResource>> downloads = new ArrayList<>();

    ClientBlobStore blobstore = null;
    try {
      blobstore = getClientBlobStore();
      for (LocalResource localResource: localResources) {
        String key = localResource.getBlobName();
        boolean uncompress = localResource.shouldUncompress();
        LocalizedResource lrsrc = lrsrcSet.get(key, localResource.shouldUncompress());
        boolean isUpdate = false;
        if ((lrsrc != null) && (lrsrc.isUncompressed() == localResource.shouldUncompress()) &&
            (isLocalizedResourceDownloaded(lrsrc))) {
          if (isLocalizedResourceUpToDate(lrsrc, blobstore)) {
            LOG.debug("blob already exists: {}", key);
            lrsrc.addReference(topo);
            results.add(lrsrc);
            continue;
          }
          LOG.debug("blob exists but isn't up to date: {}", key);
          isUpdate = true;
        }

        // go off to blobstore and get it
        // assume dir passed in exists and has correct permission
        LOG.debug("fetching blob: {}", key);
        File downloadDir = getCacheDirForFiles(userFileDir);
        File localFile = new File(downloadDir, key);
        if (uncompress) {
          // for compressed file, download to archives dir
          downloadDir = getCacheDirForArchives(userFileDir);
          localFile = new File(downloadDir, key);
        }
        downloadDir.mkdir();
        downloads.add(new DownloadBlob(this, _conf, key, localFile, user, uncompress,
            isUpdate));
      }
    } finally {
      if(blobstore !=null) {
        blobstore.shutdown();
      }
    }
    try {
      List<Future<LocalizedResource>> futures = _execService.invokeAll(downloads);
      for (Future<LocalizedResource> futureRsrc: futures) {
        LocalizedResource lrsrc = futureRsrc.get();
        lrsrc.addReference(topo);
        lrsrcSet.add(lrsrc.getKey(), lrsrc, lrsrc.isUncompressed());
        results.add(lrsrc);
      }
    } catch (ExecutionException e) {
      if (e.getCause() instanceof AuthorizationException)
        throw (AuthorizationException)e.getCause();
      else if (e.getCause() instanceof KeyNotFoundException) {
        throw (KeyNotFoundException)e.getCause();
      } else {
        throw new IOException("Error getting blobs", e);
      }
    } catch (RejectedExecutionException re) {
      throw new IOException("RejectedExecutionException: ", re);
    } catch (InterruptedException ie) {
      throw new IOException("Interrupted Exception", ie);
    }
    return results;
  }

  static class DownloadBlob implements Callable<LocalizedResource> {

    private Localizer _localizer;
    private Map _conf;
    private String _key;
    private File _localFile;
    private String _user;
    private boolean _uncompress;
    private boolean _isUpdate;

    public DownloadBlob(Localizer localizer, Map conf, String key, File localFile,
        String user, boolean uncompress, boolean update) {
      _localizer = localizer;
      _conf = conf;
      _key = key;
      _localFile = localFile;
      _user = user;
      _uncompress = uncompress;
      _isUpdate = update;
    }

    @Override
    public LocalizedResource call()
        throws AuthorizationException, KeyNotFoundException, IOException  {
      return _localizer.downloadBlob(_conf, _key, _localFile, _user, _uncompress,
        _isUpdate);
    }
  }

  private LocalizedResource downloadBlob(Map conf, String key, File localFile,
      String user, boolean uncompress, boolean isUpdate)
      throws AuthorizationException, KeyNotFoundException, IOException {
    ClientBlobStore blobstore = null;
    try {
      blobstore = getClientBlobStore();
      long nimbusBlobVersion = Utils.nimbusVersionOfBlob(key, blobstore);
      long oldVersion = Utils.localVersionOfBlob(localFile.toString());
      FileOutputStream out = null;
      PrintWriter writer = null;
      int numTries = 0;
      String localizedPath = localFile.toString();
      String localFileWithVersion = Utils.constructBlobWithVersionFileName(localFile.toString(),
              nimbusBlobVersion);
      String localVersionFile = Utils.constructVersionFileName(localFile.toString());
      String downloadFile = localFileWithVersion;
      if (uncompress) {
        // we need to download to temp file and then unpack into the one requested
        downloadFile = new File(localFile.getParent(), TO_UNCOMPRESS + localFile.getName()).toString();
      }
      while (numTries < _blobDownloadRetries) {
        out = new FileOutputStream(downloadFile);
        numTries++;
        try {
          if (!Utils.canUserReadBlob(blobstore.getBlobMeta(key), user)) {
            throw new AuthorizationException(user + " does not have READ access to " + key);
          }
          InputStreamWithMeta in = blobstore.getBlob(key);
          byte[] buffer = new byte[1024];
          int len;
          while ((len = in.read(buffer)) >= 0) {
            out.write(buffer, 0, len);
          }
          out.close();
          in.close();
          if (uncompress) {
            Utils.unpack(new File(downloadFile), new File(localFileWithVersion));
            LOG.debug("uncompressed " + downloadFile + " to: " + localFileWithVersion);
          }

          // Next write the version.
          LOG.info("Blob: " + key + " updated with new Nimbus-provided version: " +
              nimbusBlobVersion + " local version was: " + oldVersion);
          // The false parameter ensures overwriting the version file, not appending
          writer = new PrintWriter(
              new BufferedWriter(new FileWriter(localVersionFile, false)));
          writer.println(nimbusBlobVersion);
          writer.close();

          try {
            setBlobPermissions(conf, user, localFileWithVersion);
            setBlobPermissions(conf, user, localVersionFile);

            // Update the key.current symlink. First create tmp symlink and do
            // move of tmp to current so that the operation is atomic.
            String tmp_uuid_local = java.util.UUID.randomUUID().toString();
            LOG.debug("Creating a symlink @" + localFile + "." + tmp_uuid_local + " , " +
                "linking to: " + localFile + "." + nimbusBlobVersion);
            File uuid_symlink = new File(localFile + "." + tmp_uuid_local);

            Files.createSymbolicLink(uuid_symlink.toPath(),
                Paths.get(Utils.constructBlobWithVersionFileName(localFile.toString(),
                        nimbusBlobVersion)));
            File current_symlink = new File(Utils.constructBlobCurrentSymlinkName(
                    localFile.toString()));
            Files.move(uuid_symlink.toPath(), current_symlink.toPath(), ATOMIC_MOVE);
          } catch (IOException e) {
            // if we fail after writing the version file but before we move current link we need to
            // restore the old version to the file
            try {
              PrintWriter restoreWriter = new PrintWriter(
                  new BufferedWriter(new FileWriter(localVersionFile, false)));
              restoreWriter.println(oldVersion);
              restoreWriter.close();
            } catch (IOException ignore) {}
            throw e;
          }

          String oldBlobFile = localFile + "." + oldVersion;
          try {
            // Remove the old version. Note that if a number of processes have that file open,
            // the OS will keep the old blob file around until they all close the handle and only
            // then deletes it. No new process will open the old blob, since the users will open the
            // blob through the "blob.current" symlink, which always points to the latest version of
            // a blob. Remove the old version after the current symlink is updated as to not affect
            // anyone trying to read it.
            if ((oldVersion != -1) && (oldVersion != nimbusBlobVersion)) {
              LOG.info("Removing an old blob file:" + oldBlobFile);
              Files.delete(Paths.get(oldBlobFile));
            }
          } catch (IOException e) {
            // At this point we have downloaded everything and moved symlinks.  If the remove of
            // old fails just log an error
            LOG.error("Exception removing old blob version: " + oldBlobFile);
          }

          break;
        } catch (AuthorizationException ae) {
          // we consider this non-retriable exceptions
          if (out != null) {
            out.close();
          }
          new File(downloadFile).delete();
          throw ae;
        } catch (IOException | KeyNotFoundException e) {
          if (out != null) {
            out.close();
          }
          if (writer != null) {
            writer.close();
          }
          new File(downloadFile).delete();
          if (uncompress) {
            try {
              FileUtils.deleteDirectory(new File(localFileWithVersion));
            } catch (IOException ignore) {}
          }
          if (!isUpdate) {
            // don't want to remove existing version file if its an update
            new File(localVersionFile).delete();
          }

          if (numTries < _blobDownloadRetries) {
            LOG.error("Failed to download blob, retrying", e);
          } else {
            throw e;
          }
        }
      }
      return new LocalizedResource(key, localizedPath, uncompress);
    } finally {
      if(blobstore != null) {
        blobstore.shutdown();
      }
    }
  }

  public void setBlobPermissions(Map conf, String user, String path)
      throws IOException {

    if (!Utils.getBoolean(conf.get(Config.SUPERVISOR_RUN_WORKER_AS_USER), false)) {
      return;
    }
    String wlCommand = Utils.getString(conf.get(Config.SUPERVISOR_WORKER_LAUNCHER), "");
    if (wlCommand.isEmpty()) {
      String stormHome = System.getProperty("storm.home");
      wlCommand = stormHome + "/bin/worker-launcher";
    }
    List<String> command = new ArrayList<String>(Arrays.asList(wlCommand, user, "blob", path));

    String[] commandArray = command.toArray(new String[command.size()]);
    ShellCommandExecutor shExec = new ShellCommandExecutor(commandArray);
    LOG.info("Setting blob permissions, command: {}", Arrays.toString(commandArray));

    try {
      shExec.execute();
      LOG.debug("output: {}", shExec.getOutput());
    } catch (ExitCodeException e) {
      int exitCode = shExec.getExitCode();
      LOG.warn("Exit code from worker-launcher is : " + exitCode, e);
      LOG.debug("output: {}", shExec.getOutput());
      throw new IOException("Setting blob permissions failed" +
          " (exitCode=" + exitCode + ") with output: " + shExec.getOutput(), e);
    }
  }


  public synchronized void handleCacheCleanup() {
    LocalizedResourceRetentionSet toClean = new LocalizedResourceRetentionSet(_cacheTargetSize);
    // need one large set of all and then clean via LRU
    for (LocalizedResourceSet t : _userRsrc.values()) {
      toClean.addResources(t);
      LOG.debug("Resources to be cleaned after adding {} : {}", t.getUser(), toClean);
    }
    toClean.cleanup();
    LOG.debug("Resource cleanup: {}", toClean);
    for (LocalizedResourceSet t : _userRsrc.values()) {
      if (t.getSize() == 0) {
        String user = t.getUser();

        LOG.debug("removing empty set: {}", user);
        File userFileCacheDir = getLocalUserFileCacheDir(user);
        getCacheDirForFiles(userFileCacheDir).delete();
        getCacheDirForArchives(userFileCacheDir).delete();
        getLocalUserFileCacheDir(user).delete();
        boolean dirsRemoved = getLocalUserDir(user).delete();
        // to catch race with update thread
        if (dirsRemoved) {
          _userRsrc.remove(user);
        }
      }
    }
  }
}
