package io.lumify.core.bootstrap.lib;

import io.lumify.core.config.Configuration;
import io.lumify.core.exception.LumifyException;
import io.lumify.core.util.LumifyLogger;
import io.lumify.core.util.LumifyLoggerFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.security.NoSuchAlgorithmException;

public class HdfsLibCacheLoader extends LibLoader {
    private static final LumifyLogger LOGGER = LumifyLoggerFactory.getLogger(HdfsLibCacheLoader.class);
    private File hdfsLibCacheTempDirectory;

    @Override
    public void loadLibs(Configuration configuration) {
        LOGGER.info("Loading libs using %s", HdfsLibCacheLoader.class.getName());

        String hdfsLibCacheDirectory = configuration.get(Configuration.HDFS_LIB_CACHE_SOURCE_DIRECTORY, null);
        if (hdfsLibCacheDirectory == null) {
            LOGGER.warn("Skipping HDFS libcache. Configuration parameter %s not found", Configuration.HDFS_LIB_CACHE_SOURCE_DIRECTORY);
            return;
        }

        setHdfsLibCacheTempDirectory(configuration);

        String hdfsLibCacheUser = getHdfsLibCacheUser(configuration);
        FileSystem hdfsFileSystem = getFileSystem(configuration, hdfsLibCacheUser);
        File libCacheDirectory = ensureLocalLibCacheDirectory();

        try {
            syncLibCache(hdfsFileSystem, new Path(hdfsLibCacheDirectory), libCacheDirectory);
        } catch (Exception ex) {
            throw new LumifyException(String.format("Could not sync HDFS libcache. %s -> %s", hdfsLibCacheDirectory, libCacheDirectory.getAbsolutePath()), ex);
        }
    }

    private void setHdfsLibCacheTempDirectory(Configuration configuration) {
        String hdfsLibCacheTempDirectoryString = configuration.get(Configuration.HDFS_LIB_CACHE_TEMP_DIRECTORY, null);
        if (hdfsLibCacheTempDirectoryString == null) {
            File baseDir = new File(System.getProperty("java.io.tmpdir"));
            hdfsLibCacheTempDirectory = new File(baseDir, "lumify-hdfslibcache");
        } else {
            hdfsLibCacheTempDirectory = new File(hdfsLibCacheTempDirectoryString);
        }
    }

    private String getHdfsLibCacheUser(Configuration configuration) {
        String hdfsLibCacheUser = configuration.get(Configuration.HDFS_LIB_CACHE_HDFS_USER);
        if (hdfsLibCacheUser == null) {
            hdfsLibCacheUser = "hadoop";
            LOGGER.warn("Configuration parameter %s was not set; defaulting to HDFS user '%s'.", Configuration.HDFS_LIB_CACHE_HDFS_USER, hdfsLibCacheUser);
        } else {
            LOGGER.info("Connecting to HDFS as user '%s'", hdfsLibCacheUser);
        }
        return hdfsLibCacheUser;
    }

    private File ensureLocalLibCacheDirectory() {
        LOGGER.debug("Using local lib cache directory: %s", hdfsLibCacheTempDirectory.getAbsolutePath());
        hdfsLibCacheTempDirectory.mkdirs();
        return hdfsLibCacheTempDirectory;
    }

    private FileSystem getFileSystem(Configuration configuration, String user) {
        try {
            String hdfsRootDir = configuration.get(Configuration.HADOOP_URL);
            return FileSystem.get(new URI(hdfsRootDir), configuration.toHadoopConfiguration(), user);
        } catch (Exception ex) {
            throw new LumifyException("Could not open HDFS file system.", ex);
        }
    }

    private static void syncLibCache(FileSystem fs, Path source, File destDir) throws IOException, NoSuchAlgorithmException {
        if (!fs.exists(source)) {
            throw new LumifyException(String.format("Could not sync HDFS directory %s. Directory does not exist.", source));
        }

        addFilesFromHdfs(fs, source, destDir);
    }

    private static void addFilesFromHdfs(FileSystem fs, Path source, File destDir) throws IOException, NoSuchAlgorithmException {
        LOGGER.debug("Adding files from HDFS %s -> %s", source.toString(), destDir.getAbsolutePath());
        RemoteIterator<LocatedFileStatus> hdfsFiles = fs.listFiles(source, true);
        while (hdfsFiles.hasNext()) {
            LocatedFileStatus hdfsFile = hdfsFiles.next();
            if (hdfsFile.isDirectory()) {
                continue;
            }

            File locallyCachedFile = getLocalCacheFileName(hdfsFile, destDir);
            if (locallyCachedFile.exists()) {
                LOGGER.info("HDFS file %s already cached at %s. Skipping sync.", hdfsFile.getPath().toString(), locallyCachedFile.getPath());
            } else {
                fs.copyToLocalFile(hdfsFile.getPath(), new Path(locallyCachedFile.getAbsolutePath()));
                locallyCachedFile.setLastModified(hdfsFile.getModificationTime());
                LOGGER.info("Caching HDFS file %s -> %s", hdfsFile.getPath().toString(), locallyCachedFile.getPath());
            }

            addLibFile(locallyCachedFile);
        }
    }

    private static File getLocalCacheFileName(LocatedFileStatus hdfsFile, File destdir) {
        String filename = hdfsFile.getPath().getName();
        String baseFilename = filename.substring(0, filename.lastIndexOf('.'));
        String extension = filename.substring(filename.lastIndexOf('.'));
        String cacheFilename = baseFilename + "-" + hdfsFile.getModificationTime() + extension;
        return new File(destdir, cacheFilename);
    }

}
@@ -657,52 +657,8 @@
 ss);
-%0A    private File hdfsLibCacheTempDirectory;
 %0A%0A  
@@ -1139,19 +1139,49 @@
         
-set
+File libCacheDirectory = getLocal
 HdfsLibC
@@ -1176,36 +1176,32 @@
 ocalHdfsLibCache
-Temp
 Directory(config
@@ -1210,17 +1210,16 @@
 ation);%0A
-%0A
         
@@ -1367,73 +1367,8 @@
 er);
-%0A        File libCacheDirectory = ensureLocalLibCacheDirectory();
 %0A%0A  
@@ -1699,16 +1699,21 @@
 ate 
-void set
+File getLocal
 Hdfs
@@ -1712,36 +1712,32 @@
 ocalHdfsLibCache
-Temp
 Directory(Config
@@ -1872,32 +1872,71 @@
 RECTORY, null);%0A
+        File libCacheDirectory = null;%0A
         if (hdfs
@@ -2061,32 +2061,24 @@
         
-hdfsL
+l
 ibCache
-Temp
 Director
@@ -2123,16 +2123,202 @@
 ache%22);%0A
+            LOGGER.info(%22Configuration parameter %25s was not set; defaulting local libcache dir to %25s%22, Configuration.HDFS_LIB_CACHE_TEMP_DIRECTORY, libCacheDirectory.getAbsolutePath());%0A
         
@@ -2338,32 +2338,24 @@
         
-hdfsL
+l
 ibCache
-Temp
 Director
@@ -2401,33 +2401,263 @@
 tring);%0A        
-%7D
+    LOGGER.info(%22Using local lib cache directory: %25s%22, libCacheDirectory.getAbsolutePath());%0A        %7D%0A%0A        if (!libCacheDirectory.exists()) %7B%0A            libCacheDirectory.mkdirs();%0A        %7D%0A%0A        return libCacheDirectory;
 %0A    %7D%0A%0A    priv
@@ -3197,257 +3197,8 @@
  %7D%0A%0A
-    private File ensureLocalLibCacheDirectory() %7B%0A        LOGGER.debug(%22Using local lib cache directory: %25s%22, hdfsLibCacheTempDirectory.getAbsolutePath());%0A        hdfsLibCacheTempDirectory.mkdirs();%0A        return hdfsLibCacheTempDirectory;%0A    %7D%0A%0A
     

Check to see if cache directory exists before creating it