Merge branch 'master' into head_check_all_chunks

weed/filer/meta_aggregator.go

@@ -71,6 +71,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, peer string
 
 	// when filer store is not shared by multiple filers
 	if peerSignature != f.Signature {
+		lastTsNs = 0
 		if prevTsNs, err := ma.readOffset(f, peer, peerSignature); err == nil {
 			lastTsNs = prevTsNs
 		}

weed/filer/sqlite/sqlite_store.go

@@ -0,0 +1,74 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
+	"github.com/chrislusf/seaweedfs/weed/filer/mysql"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	_ "modernc.org/sqlite"
+)
+
+func init() {
+	filer.Stores = append(filer.Stores, &SqliteStore{})
+}
+
+type SqliteStore struct {
+	abstract_sql.AbstractSqlStore
+}
+
+func (store *SqliteStore) GetName() string {
+	return "sqlite"
+}
+
+func (store *SqliteStore) Initialize(configuration util.Configuration, prefix string) (err error) {
+	dbFile := configuration.GetString(prefix + "dbFile")
+	createTable := `CREATE TABLE IF NOT EXISTS "%s" (
+		dirhash BIGINT,
+		name VARCHAR(1000),
+		directory TEXT,
+		meta BLOB,
+		PRIMARY KEY (dirhash, name)
+	) WITHOUT ROWID;`
+	upsertQuery := `INSERT INTO "%s"(dirhash,name,directory,meta)VALUES(?,?,?,?)
+	ON CONFLICT(dirhash,name) DO UPDATE SET
+		directory=excluded.directory,
+		meta=excluded.meta;
+	`
+	return store.initialize(
+		dbFile,
+		createTable,
+		upsertQuery,
+	)
+}
+
+func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (err error) {
+
+	store.SupportBucketTable = true
+	store.SqlGenerator = &mysql.SqlGenMysql{
+		CreateTableSqlTemplate: createTable,
+		DropTableSqlTemplate:   "drop table `%s`",
+		UpsertQueryTemplate:    upsertQuery,
+	}
+
+	var dbErr error
+	store.DB, dbErr = sql.Open("sqlite", dbFile)
+	if dbErr != nil {
+		store.DB.Close()
+		store.DB = nil
+		return fmt.Errorf("can not connect to %s error:%v", dbFile, err)
+	}
+
+	if err = store.DB.Ping(); err != nil {
+		return fmt.Errorf("connect to %s error:%v", dbFile, err)
+	}
+
+	if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil {
+		return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err)
+	}
+
+	return nil
+}

weed/filer/stream.go

@@ -3,13 +3,16 @@ package filer
 import (
 	"bytes"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"io"
 	"math"
 	"strings"
+	"time"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/stats"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/wdclient"
 )
 
 func StreamContent(masterClient wdclient.HasLookupFileIdFunction, w io.Writer, chunks []*filer_pb.FileChunk, offset int64, size int64) error {
@@ -35,15 +38,20 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, w io.Writer, c
 	for _, chunkView := range chunkViews {
 
 		urlStrings := fileId2Url[chunkView.FileId]
+		start := time.Now()
 		data, err := retriedFetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
+		stats.FilerRequestHistogram.WithLabelValues("chunkDownload").Observe(time.Since(start).Seconds())
 		if err != nil {
+			stats.FilerRequestCounter.WithLabelValues("chunkDownloadError").Inc()
 			return fmt.Errorf("read chunk: %v", err)
 		}
 
 		_, err = w.Write(data)
 		if err != nil {
+			stats.FilerRequestCounter.WithLabelValues("chunkDownloadedError").Inc()
 			return fmt.Errorf("write chunk: %v", err)
 		}
+		stats.FilerRequestCounter.WithLabelValues("chunkDownload").Inc()
 	}
 
 	return nil