Add support for distributed filer metadata store.

go/weed/filer.go

@@ -22,6 +22,8 @@ type FilerOptions struct {
 	defaultReplicaPlacement *string
 	dir                     *string
 	redirectOnRead          *bool
+	cassandra_server        *string
+	cassandra_keyspace      *string
 }
 
 func init() {
@@ -32,13 +34,15 @@ func init() {
 	f.dir = cmdFiler.Flag.String("dir", os.TempDir(), "directory to store meta data")
 	f.defaultReplicaPlacement = cmdFiler.Flag.String("defaultReplicaPlacement", "000", "default replication type if not specified")
 	f.redirectOnRead = cmdFiler.Flag.Bool("redirectOnRead", false, "whether proxy or redirect to volume server during file GET request")
+	f.cassandra_server = cmdFiler.Flag.String("cassandra.server", "", "host[:port] of the cassandra server")
+	f.cassandra_keyspace = cmdFiler.Flag.String("cassandra.keyspace", "seaweed", "keyspace of the cassandra server")
 }
 
 var cmdFiler = &Command{
 	UsageLine: "filer -port=8888 -dir=/tmp -master=<ip:port>",
 	Short:     "start a file server that points to a master server",
 	Long: `start a file server which accepts REST operation for any files.
-	
+
 	//create or overwrite the file, the directories /path/to will be automatically created
 	POST /path/to/file
 	//get the file content
@@ -47,10 +51,10 @@ var cmdFiler = &Command{
 	POST /path/to/
 	//return a json format subdirectory and files listing
 	GET /path/to/
-  
+
   Current <fullpath~fileid> mapping metadata store is local embedded leveldb.
   It should be highly scalable to hundreds of millions of files on a modest machine.
-  
+
   Future we will ensure it can avoid of being SPOF.
 
   `,
@@ -63,8 +67,9 @@ func runFiler(cmd *Command, args []string) bool {
 	}
 
 	r := http.NewServeMux()
-	_, nfs_err := weed_server.NewFilerServer(r, *f.port, *f.master, *f.dir, *f.collection,
+	_, nfs_err := weed_server.NewEmbeddedFilerServer(r, *f.port, *f.master, *f.dir, *f.collection,
 		*f.defaultReplicaPlacement, *f.redirectOnRead,
+		*f.cassandra_server, *f.cassandra_keyspace,
 	)
 	if nfs_err != nil {
 		glog.Fatalf(nfs_err.Error())

go/weed/server.go

@@ -157,8 +157,9 @@ func runServer(cmd *Command, args []string) bool {
 	if *isStartingFiler {
 		go func() {
 			r := http.NewServeMux()
-			_, nfs_err := weed_server.NewFilerServer(r, *filerOptions.port, *filerOptions.master, *filerOptions.dir, *filerOptions.collection,
+			_, nfs_err := weed_server.NewEmbeddedFilerServer(r, *filerOptions.port, *filerOptions.master, *filerOptions.dir, *filerOptions.collection,
 				*filerOptions.defaultReplicaPlacement, *filerOptions.redirectOnRead,
+				"", "",
 			)
 			if nfs_err != nil {
 				glog.Fatalf(nfs_err.Error())

go/weed/weed_server/filer_server.go

@@ -5,7 +5,9 @@ import (
 	"strconv"
 
 	"github.com/chrislusf/weed-fs/go/filer"
+	"github.com/chrislusf/weed-fs/go/filer/cassandra_store"
 	"github.com/chrislusf/weed-fs/go/filer/embedded_filer"
+	"github.com/chrislusf/weed-fs/go/filer/flat_namespace"
 	"github.com/chrislusf/weed-fs/go/glog"
 )
 
@@ -18,8 +20,9 @@ type FilerServer struct {
 	filer              filer.Filer
 }
 
-func NewFilerServer(r *http.ServeMux, port int, master string, dir string, collection string,
+func NewEmbeddedFilerServer(r *http.ServeMux, port int, master string, dir string, collection string,
 	replication string, redirectOnRead bool,
+	cassandra_server string, cassandra_keyspace string,
 ) (fs *FilerServer, err error) {
 	fs = &FilerServer{
 		master:             master,
@@ -29,12 +32,21 @@ func NewFilerServer(r *http.ServeMux, port int, master string, dir string, colle
 		port:               ":" + strconv.Itoa(port),
 	}
 
-	if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
-		glog.Fatal("Can not start filer in dir", dir, ": ", err.Error())
-		return
+	if cassandra_server == "" {
+		if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
+			glog.Fatalf("Can not start filer in dir %s : %v", err)
+			return
+		}
+
+		r.HandleFunc("/admin/mv", fs.moveHandler)
+	} else {
+		cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server)
+		if err != nil {
+			glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err)
+		}
+		fs.filer = flat_namespace.NewFlatNamesapceFiler(master, cassandra_store)
 	}
 
-	r.HandleFunc("/admin/mv", fs.moveHandler)
 	r.HandleFunc("/", fs.filerHandler)
 
 	return fs, nil

go/weed/weed_server/filer_server_handlers.go

@@ -179,7 +179,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	glog.V(4).Infoln("saving", path, "=>", assignResult.Fid)
 	if db_err := fs.filer.CreateFile(path, assignResult.Fid); db_err != nil {
 		operation.DeleteFile(fs.master, assignResult.Fid) //clean up
-		glog.V(0).Infoln("failing to write to filer server", r.RequestURI, db_err.Error())
+		glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
 		writeJsonError(w, r, db_err)
 		return
 	}