Merge branch 'upstream_master' into store_s3cred

# Conflicts:
#	weed/s3api/filer_util.go

19 files changed, 264 insertions, 20 deletions

diff --git a/other/java/client/pom.xml b/other/java/client/pom.xml
index efbf304c4..4bfc5ab8f 100644
--- a/other/java/client/pom.xml
+++ b/other/java/client/pom.xml
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.2</version>
+    <version>1.5.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -68,6 +68,11 @@
             <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>javax.annotation</groupId>
+            <artifactId>javax.annotation-api</artifactId>
+            <version>1.3.2</version>
+        </dependency>
     </dependencies>
 
     <distributionManagement>
diff --git a/other/java/client/pom.xml.deploy b/other/java/client/pom.xml.deploy
index 9efc21373..c3c960a28 100644
--- a/other/java/client/pom.xml.deploy
+++ b/other/java/client/pom.xml.deploy
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.2</version>
+    <version>1.5.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
diff --git a/other/java/client/pom_debug.xml b/other/java/client/pom_debug.xml
index 04ff52730..acdf621a5 100644
--- a/other/java/client/pom_debug.xml
+++ b/other/java/client/pom_debug.xml
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.2</version>
+    <version>1.5.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
diff --git a/other/java/client/src/main/java/seaweedfs/client/ChunkCache.java b/other/java/client/src/main/java/seaweedfs/client/ChunkCache.java
index 7afa2dca0..58870d742 100644
--- a/other/java/client/src/main/java/seaweedfs/client/ChunkCache.java
+++ b/other/java/client/src/main/java/seaweedfs/client/ChunkCache.java
@@ -15,7 +15,6 @@ public class ChunkCache {
         }
         this.cache = CacheBuilder.newBuilder()
                 .maximumSize(maxEntries)
-                .weakValues()
                 .expireAfterAccess(1, TimeUnit.HOURS)
                 .build();
     }
diff --git a/other/java/client/src/main/java/seaweedfs/client/FilerClient.java b/other/java/client/src/main/java/seaweedfs/client/FilerClient.java
index 035b2c852..7338d5bee 100644
--- a/other/java/client/src/main/java/seaweedfs/client/FilerClient.java
+++ b/other/java/client/src/main/java/seaweedfs/client/FilerClient.java
@@ -275,9 +275,9 @@ public class FilerClient {
         try {
             FilerProto.CreateEntryResponse createEntryResponse =
                     filerGrpcClient.getBlockingStub().createEntry(FilerProto.CreateEntryRequest.newBuilder()
-                    .setDirectory(parent)
-                    .setEntry(entry)
-                    .build());
+                            .setDirectory(parent)
+                            .setEntry(entry)
+                            .build());
             if (Strings.isNullOrEmpty(createEntryResponse.getError())) {
                 return true;
             }
@@ -333,4 +333,13 @@ public class FilerClient {
         return true;
     }
 
+    public Iterator<FilerProto.SubscribeMetadataResponse> watch(String prefix, String clientName, long sinceNs) {
+        return filerGrpcClient.getBlockingStub().subscribeMetadata(FilerProto.SubscribeMetadataRequest.newBuilder()
+                .setPathPrefix(prefix)
+                .setClientName(clientName)
+                .setSinceNs(sinceNs)
+                .build()
+        );
+    }
+
 }
diff --git a/other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java b/other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java
index ab2407dec..2b530d2dd 100644
--- a/other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java
+++ b/other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java
@@ -19,6 +19,7 @@ public class SeaweedRead {
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedRead.class);
 
     static ChunkCache chunkCache = new ChunkCache(4);
+    static VolumeIdCache volumeIdCache = new VolumeIdCache(4 * 1024);
 
     // returns bytesRead
     public static long read(FilerGrpcClient filerGrpcClient, List<VisibleInterval> visibleIntervals,
@@ -27,16 +28,28 @@ public class SeaweedRead {
 
         List<ChunkView> chunkViews = viewFromVisibles(visibleIntervals, position, bufferLength);
 
+        Map<String, FilerProto.Locations> knownLocations = new HashMap<>();
+
         FilerProto.LookupVolumeRequest.Builder lookupRequest = FilerProto.LookupVolumeRequest.newBuilder();
         for (ChunkView chunkView : chunkViews) {
             String vid = parseVolumeId(chunkView.fileId);
-            lookupRequest.addVolumeIds(vid);
+            FilerProto.Locations locations = volumeIdCache.getLocations(vid);
+            if (locations == null) {
+                lookupRequest.addVolumeIds(vid);
+            } else {
+                knownLocations.put(vid, locations);
+            }
         }
 
-        FilerProto.LookupVolumeResponse lookupResponse = filerGrpcClient
-                .getBlockingStub().lookupVolume(lookupRequest.build());
-
-        Map<String, FilerProto.Locations> vid2Locations = lookupResponse.getLocationsMapMap();
+        if (lookupRequest.getVolumeIdsCount() > 0) {
+            FilerProto.LookupVolumeResponse lookupResponse = filerGrpcClient
+                    .getBlockingStub().lookupVolume(lookupRequest.build());
+            Map<String, FilerProto.Locations> vid2Locations = lookupResponse.getLocationsMapMap();
+            for (Map.Entry<String, FilerProto.Locations> entry : vid2Locations.entrySet()) {
+                volumeIdCache.setLocations(entry.getKey(), entry.getValue());
+                knownLocations.put(entry.getKey(), entry.getValue());
+            }
+        }
 
         //TODO parallel this
         long readCount = 0;
@@ -50,7 +63,7 @@ public class SeaweedRead {
                 startOffset += gap;
             }
 
-            FilerProto.Locations locations = vid2Locations.get(parseVolumeId(chunkView.fileId));
+            FilerProto.Locations locations = knownLocations.get(parseVolumeId(chunkView.fileId));
             if (locations == null || locations.getLocationsCount() == 0) {
                 LOG.error("failed to locate {}", chunkView.fileId);
                 // log here!
diff --git a/other/java/client/src/main/java/seaweedfs/client/VolumeIdCache.java b/other/java/client/src/main/java/seaweedfs/client/VolumeIdCache.java
new file mode 100644
index 000000000..fd2649cc2
--- /dev/null
+++ b/other/java/client/src/main/java/seaweedfs/client/VolumeIdCache.java
@@ -0,0 +1,36 @@
+package seaweedfs.client;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
+import java.util.concurrent.TimeUnit;
+
+public class VolumeIdCache {
+
+    private Cache<String, FilerProto.Locations> cache = null;
+
+    public VolumeIdCache(int maxEntries) {
+        if (maxEntries == 0) {
+            return;
+        }
+        this.cache = CacheBuilder.newBuilder()
+                .maximumSize(maxEntries)
+                .expireAfterAccess(5, TimeUnit.MINUTES)
+                .build();
+    }
+
+    public FilerProto.Locations getLocations(String volumeId) {
+        if (this.cache == null) {
+            return null;
+        }
+        return this.cache.getIfPresent(volumeId);
+    }
+
+    public void setLocations(String volumeId, FilerProto.Locations locations) {
+        if (this.cache == null) {
+            return;
+        }
+        this.cache.put(volumeId, locations);
+    }
+
+}
diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto
index 11c29e6ec..f61b62fa3 100644
--- a/other/java/client/src/main/proto/filer.proto
+++ b/other/java/client/src/main/proto/filer.proto
@@ -348,3 +348,23 @@ message KvPutRequest {
 message KvPutResponse {
     string error = 1;
 }
+
+// path-based configurations
+message FilerConf {
+    int32 version = 1;
+    message PathConf {
+        string location_prefix = 1;
+        string collection = 2;
+        string replication = 3;
+        string ttl = 4;
+        enum DiskType {
+            NONE = 0;
+            HDD = 1;
+            SSD = 2;
+        }
+        DiskType disk_type = 5;
+        bool fsync = 6;
+        uint32 volume_growth_count = 7;
+    }
+    repeated PathConf locations = 2;
+}
diff --git a/other/java/examples/pom.xml b/other/java/examples/pom.xml
new file mode 100644
index 000000000..f7c48d0ab
--- /dev/null
+++ b/other/java/examples/pom.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.example</groupId>
+    <artifactId>unzip</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <dependencies>
+        <dependency>
+            <groupId>com.github.chrislusf</groupId>
+            <artifactId>seaweedfs-client</artifactId>
+            <version>1.5.6</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.github.chrislusf</groupId>
+            <artifactId>seaweedfs-hadoop2-client</artifactId>
+            <version>1.5.6</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>2.9.2</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+
+</project>
\ No newline at end of file
diff --git a/other/java/examples/src/main/java/com/seaweedfs/examples/UnzipFile.java b/other/java/examples/src/main/java/com/seaweedfs/examples/UnzipFile.java
new file mode 100644
index 000000000..0529a5c73
--- /dev/null
+++ b/other/java/examples/src/main/java/com/seaweedfs/examples/UnzipFile.java
@@ -0,0 +1,54 @@
+package com.seaweedfs.examples;
+
+import seaweed.hdfs.SeaweedInputStream;
+import seaweedfs.client.FilerClient;
+import seaweedfs.client.FilerGrpcClient;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+
+public class UnzipFile {
+
+    public static void main(String[] args) throws IOException {
+
+        FilerGrpcClient filerGrpcClient = new FilerGrpcClient("localhost", 18888);
+        FilerClient filerClient = new FilerClient(filerGrpcClient);
+
+        long startTime = System.currentTimeMillis();
+        parseZip("/Users/chris/tmp/test.zip");
+
+        long startTime2 = System.currentTimeMillis();
+
+        long localProcessTime = startTime2 - startTime;
+
+        SeaweedInputStream seaweedInputStream = new SeaweedInputStream(
+                filerGrpcClient,
+                new org.apache.hadoop.fs.FileSystem.Statistics(""),
+                "/",
+                filerClient.lookupEntry("/", "test.zip")
+        );
+        parseZip(seaweedInputStream);
+
+        long swProcessTime = System.currentTimeMillis() - startTime2;
+
+        System.out.println("Local time: " + localProcessTime);
+        System.out.println("SeaweedFS time: " + swProcessTime);
+
+    }
+
+    public static void parseZip(String filename) throws IOException {
+        FileInputStream fileInputStream = new FileInputStream(filename);
+        parseZip(fileInputStream);
+    }
+
+    public static void parseZip(InputStream is) throws IOException {
+        ZipInputStream zin = new ZipInputStream(is);
+        ZipEntry ze;
+        while ((ze = zin.getNextEntry()) != null) {
+            System.out.println(ze.getName());
+        }
+    }
+}
diff --git a/other/java/examples/src/main/java/com/seaweedfs/examples/WatchFiles.java b/other/java/examples/src/main/java/com/seaweedfs/examples/WatchFiles.java
new file mode 100644
index 000000000..e489cb3b1
--- /dev/null
+++ b/other/java/examples/src/main/java/com/seaweedfs/examples/WatchFiles.java
@@ -0,0 +1,46 @@
+package com.seaweedfs.examples;
+
+import seaweedfs.client.FilerClient;
+import seaweedfs.client.FilerProto;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.Iterator;
+
+public class WatchFiles {
+
+    public static void main(String[] args) throws IOException {
+
+        FilerClient filerClient = new FilerClient("localhost", 18888);
+
+        long sinceNs = (System.currentTimeMillis() - 3600 * 1000) * 1000000L;
+
+        Iterator<FilerProto.SubscribeMetadataResponse> watch = filerClient.watch(
+                "/buckets",
+                "exampleClientName",
+                sinceNs
+        );
+
+        System.out.println("Connected to filer, subscribing from " + new Date());
+
+        while (watch.hasNext()) {
+            FilerProto.SubscribeMetadataResponse event = watch.next();
+            FilerProto.EventNotification notification = event.getEventNotification();
+            if (!event.getDirectory().equals(notification.getNewParentPath())) {
+                // move an entry to a new directory, possibly with a new name
+                if (notification.hasOldEntry() && notification.hasNewEntry()) {
+                    System.out.println("moved " + event.getDirectory() + "/" + notification.getOldEntry().getName() + " to " + notification.getNewParentPath() + "/" + notification.getNewEntry().getName());
+                } else {
+                    System.out.println("this should not happen.");
+                }
+            } else if (notification.hasNewEntry() && !notification.hasOldEntry()) {
+                System.out.println("created entry " + event.getDirectory() + "/" + notification.getNewEntry().getName());
+            } else if (!notification.hasNewEntry() && notification.hasOldEntry()) {
+                System.out.println("deleted entry " + event.getDirectory() + "/" + notification.getOldEntry().getName());
+            } else if (notification.hasNewEntry() && notification.hasOldEntry()) {
+                System.out.println("updated entry " + event.getDirectory() + "/" + notification.getNewEntry().getName());
+            }
+        }
+
+    }
+}
diff --git a/other/java/hdfs2/dependency-reduced-pom.xml b/other/java/hdfs2/dependency-reduced-pom.xml
index 229fa673c..f7873a435 100644
--- a/other/java/hdfs2/dependency-reduced-pom.xml
+++ b/other/java/hdfs2/dependency-reduced-pom.xml
@@ -301,7 +301,7 @@
     </snapshotRepository>

   </distributionManagement>

   <properties>

-    <seaweedfs.client.version>1.5.2</seaweedfs.client.version>

+    <seaweedfs.client.version>1.5.6</seaweedfs.client.version>

     <hadoop.version>2.9.2</hadoop.version>

   </properties>

 </project>

diff --git a/other/java/hdfs2/pom.xml b/other/java/hdfs2/pom.xml
index 1b73b2811..bda0eba56 100644
--- a/other/java/hdfs2/pom.xml
+++ b/other/java/hdfs2/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.5.2</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 
diff --git a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
index ca67c3874..fb1f4c53b 100644
--- a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
+++ b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
@@ -25,6 +25,7 @@ public class SeaweedFileSystem extends FileSystem {
     public static final String FS_SEAWEED_FILER_PORT = "fs.seaweed.filer.port";
     public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_BUFFER_SIZE = "fs.seaweed.buffer.size";
+    public static final String FS_SEAWEED_REPLICATION = "fs.seaweed.replication";
     public static final int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class);
@@ -66,6 +67,12 @@ public class SeaweedFileSystem extends FileSystem {
     }
 
     @Override
+    public void close() throws IOException {
+        super.close();
+        this.seaweedFileSystemStore.close();
+    }
+
+    @Override
     public FSDataInputStream open(Path path, int bufferSize) throws IOException {
 
         LOG.debug("open path: {} bufferSize:{}", path, bufferSize);
@@ -91,7 +98,7 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            String replicaPlacement = String.format("%03d", replication - 1);
+            String replicaPlacement = this.getConf().get(FS_SEAWEED_REPLICATION, String.format("%03d", replication - 1));
             int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
             OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, seaweedBufferSize, replicaPlacement);
             return new FSDataOutputStream(outputStream, statistics);
diff --git a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
index 23556a578..cec309b3e 100644
--- a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
+++ b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
@@ -36,6 +36,14 @@ public class SeaweedFileSystemStore {
         this.conf = conf;
     }
 
+    public void close() {
+        try {
+            this.filerGrpcClient.shutdown();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
     public static String getParentDirectory(Path path) {
         return path.isRoot() ? "/" : path.getParent().toUri().getPath();
     }
diff --git a/other/java/hdfs3/dependency-reduced-pom.xml b/other/java/hdfs3/dependency-reduced-pom.xml
index c4847a9b9..20b52e20f 100644
--- a/other/java/hdfs3/dependency-reduced-pom.xml
+++ b/other/java/hdfs3/dependency-reduced-pom.xml
@@ -309,7 +309,7 @@
     </snapshotRepository>

   </distributionManagement>

   <properties>

-    <seaweedfs.client.version>1.5.2</seaweedfs.client.version>

+    <seaweedfs.client.version>1.5.6</seaweedfs.client.version>

     <hadoop.version>3.1.1</hadoop.version>

   </properties>

 </project>

diff --git a/other/java/hdfs3/pom.xml b/other/java/hdfs3/pom.xml
index 9e668bba8..85d8db859 100644
--- a/other/java/hdfs3/pom.xml
+++ b/other/java/hdfs3/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.5.2</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 
diff --git a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
index ca67c3874..fb1f4c53b 100644
--- a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
+++ b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
@@ -25,6 +25,7 @@ public class SeaweedFileSystem extends FileSystem {
     public static final String FS_SEAWEED_FILER_PORT = "fs.seaweed.filer.port";
     public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_BUFFER_SIZE = "fs.seaweed.buffer.size";
+    public static final String FS_SEAWEED_REPLICATION = "fs.seaweed.replication";
     public static final int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class);
@@ -66,6 +67,12 @@ public class SeaweedFileSystem extends FileSystem {
     }
 
     @Override
+    public void close() throws IOException {
+        super.close();
+        this.seaweedFileSystemStore.close();
+    }
+
+    @Override
     public FSDataInputStream open(Path path, int bufferSize) throws IOException {
 
         LOG.debug("open path: {} bufferSize:{}", path, bufferSize);
@@ -91,7 +98,7 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            String replicaPlacement = String.format("%03d", replication - 1);
+            String replicaPlacement = this.getConf().get(FS_SEAWEED_REPLICATION, String.format("%03d", replication - 1));
             int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
             OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, seaweedBufferSize, replicaPlacement);
             return new FSDataOutputStream(outputStream, statistics);
diff --git a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
index 23556a578..1ad809f9a 100644
--- a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
+++ b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java
@@ -36,6 +36,14 @@ public class SeaweedFileSystemStore {
         this.conf = conf;
     }
 
+    public void close() {
+        try {
+            this.filerGrpcClient.shutdown();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
     public static String getParentDirectory(Path path) {
         return path.isRoot() ? "/" : path.getParent().toUri().getPath();
     }
@@ -279,4 +287,4 @@ public class SeaweedFileSystemStore {
 
     }
 
-}
+}
\ No newline at end of file