Merge pull request #15 from chrislusf/master

sync

4 files changed, 48 insertions, 11 deletions

diff --git a/weed/util/compression.go b/weed/util/compression.go
index 2881a7bfd..cf3ac7c57 100644
--- a/weed/util/compression.go
+++ b/weed/util/compression.go
@@ -12,15 +12,44 @@ import (
 	"github.com/klauspost/compress/zstd"
 )
 
+var (
+	UnsupportedCompression = fmt.Errorf("unsupported compression")
+)
+
+func MaybeGzipData(input []byte) []byte {
+	if IsGzippedContent(input) {
+		return input
+	}
+	gzipped, err := GzipData(input)
+	if err != nil {
+		return input
+	}
+	if len(gzipped)*10 > len(input)*9 {
+		return input
+	}
+	return gzipped
+}
+
+func MaybeDecompressData(input []byte) []byte {
+	uncompressed, err := DecompressData(input)
+	if err != nil {
+		if err != UnsupportedCompression {
+			glog.Errorf("decompressed data: %v", err)
+		}
+		return input
+	}
+	return uncompressed
+}
+
 func GzipData(input []byte) ([]byte, error) {
 	buf := new(bytes.Buffer)
 	w, _ := gzip.NewWriterLevel(buf, flate.BestSpeed)
 	if _, err := w.Write(input); err != nil {
-		glog.V(2).Infoln("error compressing data:", err)
+		glog.V(2).Infof("error gzip data: %v", err)
 		return nil, err
 	}
 	if err := w.Close(); err != nil {
-		glog.V(2).Infoln("error closing compressed data:", err)
+		glog.V(2).Infof("error closing gzipped data: %v", err)
 		return nil, err
 	}
 	return buf.Bytes(), nil
@@ -39,7 +68,7 @@ func DecompressData(input []byte) ([]byte, error) {
 	if IsZstdContent(input) {
 		return unzstdData(input)
 	}
-	return input, fmt.Errorf("unsupported compression")
+	return input, UnsupportedCompression
 }
 
 func ungzipData(input []byte) ([]byte, error) {
@@ -48,7 +77,7 @@ func ungzipData(input []byte) ([]byte, error) {
 	defer r.Close()
 	output, err := ioutil.ReadAll(r)
 	if err != nil {
-		glog.V(2).Infoln("error uncompressing data:", err)
+		glog.V(2).Infof("error ungzip data: %v", err)
 	}
 	return output, err
 }
diff --git a/weed/util/constants.go b/weed/util/constants.go
index d48b9e32d..4782b5688 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 92)
+	VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 99)
 	COMMIT  = ""
 )
 
diff --git a/weed/util/http_util.go b/weed/util/http_util.go
index 7cc64ea85..eef24b930 100644
--- a/weed/util/http_util.go
+++ b/weed/util/http_util.go
@@ -174,7 +174,7 @@ func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) e
 	return readFn(r.Body)
 }
 
-func DownloadFile(fileUrl string) (filename string, header http.Header, rc io.ReadCloser, e error) {
+func DownloadFile(fileUrl string) (filename string, header http.Header, resp *http.Response, e error) {
 	response, err := client.Get(fileUrl)
 	if err != nil {
 		return "", nil, nil, err
@@ -188,7 +188,7 @@ func DownloadFile(fileUrl string) (filename string, header http.Header, rc io.Re
 			filename = strings.Trim(filename, "\"")
 		}
 	}
-	rc = response.Body
+	resp = response
 	return
 }
 
diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go
index f0486ac46..57f4b0115 100644
--- a/weed/util/log_buffer/log_read.go
+++ b/weed/util/log_buffer/log_read.go
@@ -2,6 +2,7 @@ package log_buffer
 
 import (
 	"bytes"
+	"fmt"
 	"time"
 
 	"github.com/golang/protobuf/proto"
@@ -11,13 +12,17 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
+var (
+	ResumeError = fmt.Errorf("resume")
+)
+
 func (logBuffer *LogBuffer) LoopProcessLogData(
 	startTreadTime time.Time,
 	waitForDataFn func() bool,
-	eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (err error) {
+	eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) {
 	// loop through all messages
 	var bytesBuf *bytes.Buffer
-	lastReadTime := startTreadTime
+	lastReadTime = startTreadTime
 	defer func() {
 		if bytesBuf != nil {
 			logBuffer.ReleaseMemory(bytesBuf)
@@ -48,10 +53,13 @@ func (logBuffer *LogBuffer) LoopProcessLogData(
 		for pos := 0; pos+4 < len(buf); {
 
 			size := util.BytesToUint32(buf[pos : pos+4])
+			if pos+4+int(size) > len(buf) {
+				err = ResumeError
+				glog.Errorf("LoopProcessLogData: read buffer %v read %d [%d,%d) from [0,%d)", lastReadTime, batchSize, pos, pos+int(size)+4, len(buf))
+				return
+			}
 			entryData := buf[pos+4 : pos+4+int(size)]
 
-			// fmt.Printf("read buffer read %d [%d,%d) from [0,%d)\n", batchSize, pos, pos+int(size)+4, len(buf))
-
 			logEntry := &filer_pb.LogEntry{}
 			if err = proto.Unmarshal(entryData, logEntry); err != nil {
 				glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err)