hilimd
4 years ago
committed by
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
92 changed files with 853 additions and 337 deletions
-
4unmaintained/see_log_entry/see_log_entry.go
-
38weed/filer/abstract_sql/abstract_sql_store.go
-
87weed/filer/abstract_sql/abstract_sql_store_kv.go
-
0weed/filer/cassandra/README.txt
-
21weed/filer/cassandra/cassandra_store.go
-
18weed/filer/cassandra/cassandra_store_kv.go
-
2weed/filer/configuration.go
-
2weed/filer/entry.go
-
2weed/filer/entry_codec.go
-
22weed/filer/etcd/etcd_store.go
-
44weed/filer/etcd/etcd_store_kv.go
-
2weed/filer/filechunk_manifest.go
-
2weed/filer/filechunk_manifest_test.go
-
2weed/filer/filechunks.go
-
2weed/filer/filechunks2_test.go
-
2weed/filer/filechunks_test.go
-
12weed/filer/filer.go
-
4weed/filer/filer_buckets.go
-
4weed/filer/filer_delete_entry.go
-
2weed/filer/filer_deletion.go
-
6weed/filer/filer_notify.go
-
2weed/filer/filer_notify_append.go
-
2weed/filer/filer_notify_test.go
-
79weed/filer/filerstore.go
-
25weed/filer/leveldb/leveldb_store.go
-
45weed/filer/leveldb/leveldb_store_kv.go
-
24weed/filer/leveldb/leveldb_store_test.go
-
4weed/filer/leveldb2/leveldb2_local_store.go
-
29weed/filer/leveldb2/leveldb2_store.go
-
56weed/filer/leveldb2/leveldb2_store_kv.go
-
24weed/filer/leveldb2/leveldb2_store_test.go
-
2weed/filer/meta_aggregator.go
-
2weed/filer/meta_replay.go
-
25weed/filer/mongodb/mongodb_store.go
-
72weed/filer/mongodb/mongodb_store_kv.go
-
12weed/filer/mysql/mysql_store.go
-
2weed/filer/permission.go
-
0weed/filer/postgres/README.txt
-
10weed/filer/postgres/postgres_store.go
-
2weed/filer/reader_at.go
-
2weed/filer/reader_at_test.go
-
4weed/filer/redis/redis_cluster_store.go
-
4weed/filer/redis/redis_store.go
-
16weed/filer/redis/universal_redis_store.go
-
42weed/filer/redis/universal_redis_store_kv.go
-
4weed/filer/redis2/redis_cluster_store.go
-
4weed/filer/redis2/redis_store.go
-
16weed/filer/redis2/universal_redis_store.go
-
42weed/filer/redis2/universal_redis_store_kv.go
-
2weed/filer/stream.go
-
2weed/filer/topics.go
-
8weed/filesys/dir.go
-
4weed/filesys/dir_link.go
-
14weed/filesys/file.go
-
20weed/filesys/filehandle.go
-
20weed/filesys/meta_cache/meta_cache.go
-
4weed/filesys/meta_cache/meta_cache_init.go
-
6weed/filesys/meta_cache/meta_cache_subscribe.go
-
6weed/filesys/wfs_deletion.go
-
4weed/filesys/wfs_write.go
-
6weed/messaging/broker/broker_grpc_server.go
-
4weed/messaging/broker/broker_grpc_server_publish.go
-
6weed/messaging/broker/broker_grpc_server_subscribe.go
-
4weed/messaging/broker/topic_manager.go
-
6weed/replication/sink/azuresink/azure_sink.go
-
6weed/replication/sink/b2sink/b2_sink.go
-
24weed/replication/sink/filersink/filer_sink.go
-
6weed/replication/sink/gcssink/gcs_sink.go
-
8weed/replication/sink/s3sink/s3_sink.go
-
6weed/replication/sink/s3sink/s3_write.go
-
8weed/s3api/filer_multipart.go
-
6weed/s3api/s3api_objects_list_handlers.go
-
43weed/server/filer_grpc_server.go
-
16weed/server/filer_grpc_server_rename.go
-
6weed/server/filer_grpc_server_sub_meta.go
-
24weed/server/filer_server.go
-
8weed/server/filer_server_handlers_read.go
-
2weed/server/filer_server_handlers_read_dir.go
-
12weed/server/filer_server_handlers_write_autochunk.go
-
6weed/server/filer_server_handlers_write_cipher.go
-
2weed/server/master_grpc_server.go
-
2weed/server/master_server.go
-
9weed/server/volume_grpc_copy.go
-
20weed/server/webdav_server.go
-
4weed/shell/command_fs_cat.go
-
4weed/shell/command_fs_du.go
-
4weed/shell/command_fs_ls.go
-
4weed/shell/command_volume_fsck.go
-
3weed/storage/disk_location.go
-
5weed/storage/needle/volume_ttl_test.go
-
2weed/storage/store.go
-
5weed/storage/store_vacuum.go
@ -0,0 +1,87 @@ |
|||
package abstract_sql |
|||
|
|||
import ( |
|||
"context" |
|||
"database/sql" |
|||
"fmt" |
|||
"strings" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/glog" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
) |
|||
|
|||
func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
dirStr, dirHash, name := genDirAndName(key) |
|||
|
|||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, dirHash, name, dirStr, value) |
|||
if err != nil { |
|||
if !strings.Contains(strings.ToLower(err.Error()), "duplicate") { |
|||
return fmt.Errorf("kv insert: %s", err) |
|||
} |
|||
} |
|||
|
|||
// now the insert failed possibly due to duplication constraints
|
|||
glog.V(1).Infof("kv insert falls back to update: %s", err) |
|||
|
|||
res, err = store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, value, dirHash, name, dirStr) |
|||
if err != nil { |
|||
return fmt.Errorf("kv upsert: %s", err) |
|||
} |
|||
|
|||
_, err = res.RowsAffected() |
|||
if err != nil { |
|||
return fmt.Errorf("kv upsert no rows affected: %s", err) |
|||
} |
|||
return nil |
|||
|
|||
} |
|||
|
|||
func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
dirStr, dirHash, name := genDirAndName(key) |
|||
row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, dirHash, name, dirStr) |
|||
|
|||
err = row.Scan(&value) |
|||
|
|||
if err == sql.ErrNoRows { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
if err != nil { |
|||
return nil, fmt.Errorf("kv get: %v", err) |
|||
} |
|||
|
|||
return |
|||
} |
|||
|
|||
func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
dirStr, dirHash, name := genDirAndName(key) |
|||
|
|||
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, dirHash, name, dirStr) |
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %s", err) |
|||
} |
|||
|
|||
_, err = res.RowsAffected() |
|||
if err != nil { |
|||
return fmt.Errorf("kv delete no rows affected: %s", err) |
|||
} |
|||
|
|||
return nil |
|||
|
|||
} |
|||
|
|||
func genDirAndName(key []byte) (dirStr string, dirHash int64, name string) { |
|||
for len(key) < 8 { |
|||
key = append(key, 0) |
|||
} |
|||
|
|||
dirHash = int64(util.BytesToUint64(key[:8])) |
|||
dirStr = string(key[:8]) |
|||
name = string(key[8:]) |
|||
|
|||
return |
|||
} |
@ -0,0 +1,18 @@ |
|||
package cassandra |
|||
|
|||
import ( |
|||
"context" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
) |
|||
|
|||
func (store *CassandraStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
return filer.ErrKvNotImplemented |
|||
} |
|||
|
|||
func (store *CassandraStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
return nil, filer.ErrKvNotImplemented |
|||
} |
|||
|
|||
func (store *CassandraStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
return filer.ErrKvNotImplemented |
|||
} |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"os" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"os" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"bytes" |
@ -0,0 +1,44 @@ |
|||
package etcd |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
) |
|||
|
|||
func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
_, err = store.client.Put(ctx, string(key), string(value)) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
resp, err := store.client.Get(ctx, string(key), nil) |
|||
|
|||
if err != nil { |
|||
return nil, fmt.Errorf("kv get: %v", err) |
|||
} |
|||
|
|||
if len(resp.Kvs) == 0 { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
return resp.Kvs[0].Value, nil |
|||
} |
|||
|
|||
func (store *EtcdStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
_, err = store.client.Delete(ctx, string(key)) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"bytes" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"bytes" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"fmt" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"sort" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"fmt" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"strings" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"context" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"testing" |
@ -0,0 +1,45 @@ |
|||
package leveldb |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/syndtr/goleveldb/leveldb" |
|||
) |
|||
|
|||
func (store *LevelDBStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
err = store.db.Put(key, value, nil) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *LevelDBStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
value, err = store.db.Get(key, nil) |
|||
|
|||
if err == leveldb.ErrNotFound { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
if err != nil { |
|||
return nil, fmt.Errorf("kv get: %v", err) |
|||
} |
|||
|
|||
return |
|||
} |
|||
|
|||
func (store *LevelDBStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
err = store.db.Delete(key, nil) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
@ -0,0 +1,56 @@ |
|||
package leveldb |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/syndtr/goleveldb/leveldb" |
|||
) |
|||
|
|||
func (store *LevelDB2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
partitionId := bucketKvKey(key, store.dbCount) |
|||
|
|||
err = store.dbs[partitionId].Put(key, value, nil) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv bucket %d put: %v", partitionId, err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *LevelDB2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
partitionId := bucketKvKey(key, store.dbCount) |
|||
|
|||
value, err = store.dbs[partitionId].Get(key, nil) |
|||
|
|||
if err == leveldb.ErrNotFound { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
if err != nil { |
|||
return nil, fmt.Errorf("kv bucket %d get: %v", partitionId, err) |
|||
} |
|||
|
|||
return |
|||
} |
|||
|
|||
func (store *LevelDB2Store) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
partitionId := bucketKvKey(key, store.dbCount) |
|||
|
|||
err = store.dbs[partitionId].Delete(key, nil) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv bucket %d delete: %v", partitionId, err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func bucketKvKey(key []byte, dbCount int) (partitionId int) { |
|||
return int(key[len(key)-1]) % dbCount |
|||
} |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"context" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"context" |
@ -0,0 +1,72 @@ |
|||
package mongodb |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/glog" |
|||
"go.mongodb.org/mongo-driver/bson" |
|||
"go.mongodb.org/mongo-driver/mongo" |
|||
) |
|||
|
|||
func (store *MongodbStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
dir, name := genDirAndName(key) |
|||
|
|||
c := store.connect.Database(store.database).Collection(store.collectionName) |
|||
|
|||
_, err = c.InsertOne(ctx, Model{ |
|||
Directory: dir, |
|||
Name: name, |
|||
Meta: value, |
|||
}) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *MongodbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
dir, name := genDirAndName(key) |
|||
|
|||
var data Model |
|||
|
|||
var where = bson.M{"directory": dir, "name": name} |
|||
err = store.connect.Database(store.database).Collection(store.collectionName).FindOne(ctx, where).Decode(&data) |
|||
if err != mongo.ErrNoDocuments && err != nil { |
|||
glog.Error("kv get: %v", err) |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
if len(data.Meta) == 0 { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
return data.Meta, nil |
|||
} |
|||
|
|||
func (store *MongodbStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
dir, name := genDirAndName(key) |
|||
|
|||
where := bson.M{"directory": dir, "name": name} |
|||
_, err = store.connect.Database(store.database).Collection(store.collectionName).DeleteOne(ctx, where) |
|||
if err != nil { |
|||
return fmt.Errorf("kv delete %s : %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func genDirAndName(key []byte) (dir string, name string) { |
|||
for len(key) < 8 { |
|||
key = append(key, 0) |
|||
} |
|||
|
|||
dir = string(key[:8]) |
|||
name = string(key[8:]) |
|||
|
|||
return |
|||
} |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
func hasWritePermission(dir *Entry, entry *Entry) bool { |
|||
|
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"context" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"fmt" |
@ -1,13 +1,13 @@ |
|||
package redis |
|||
|
|||
import ( |
|||
"github.com/chrislusf/seaweedfs/weed/filer2" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func init() { |
|||
filer2.Stores = append(filer2.Stores, &RedisClusterStore{}) |
|||
filer.Stores = append(filer.Stores, &RedisClusterStore{}) |
|||
} |
|||
|
|||
type RedisClusterStore struct { |
@ -1,13 +1,13 @@ |
|||
package redis |
|||
|
|||
import ( |
|||
"github.com/chrislusf/seaweedfs/weed/filer2" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func init() { |
|||
filer2.Stores = append(filer2.Stores, &RedisStore{}) |
|||
filer.Stores = append(filer.Stores, &RedisStore{}) |
|||
} |
|||
|
|||
type RedisStore struct { |
@ -0,0 +1,42 @@ |
|||
package redis |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func (store *UniversalRedisStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
_, err = store.Client.Set(string(key), value, 0).Result() |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *UniversalRedisStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
data, err := store.Client.Get(string(key)).Result() |
|||
|
|||
if err == redis.Nil { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
return []byte(data), err |
|||
} |
|||
|
|||
func (store *UniversalRedisStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
_, err = store.Client.Del(string(key)).Result() |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
@ -1,13 +1,13 @@ |
|||
package redis2 |
|||
|
|||
import ( |
|||
"github.com/chrislusf/seaweedfs/weed/filer2" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func init() { |
|||
filer2.Stores = append(filer2.Stores, &RedisCluster2Store{}) |
|||
filer.Stores = append(filer.Stores, &RedisCluster2Store{}) |
|||
} |
|||
|
|||
type RedisCluster2Store struct { |
@ -1,13 +1,13 @@ |
|||
package redis2 |
|||
|
|||
import ( |
|||
"github.com/chrislusf/seaweedfs/weed/filer2" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func init() { |
|||
filer2.Stores = append(filer2.Stores, &Redis2Store{}) |
|||
filer.Stores = append(filer.Stores, &Redis2Store{}) |
|||
} |
|||
|
|||
type Redis2Store struct { |
@ -0,0 +1,42 @@ |
|||
package redis2 |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/go-redis/redis" |
|||
) |
|||
|
|||
func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
_, err = store.Client.Set(string(key), value, 0).Result() |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
data, err := store.Client.Get(string(key)).Result() |
|||
|
|||
if err == redis.Nil { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
return []byte(data), err |
|||
} |
|||
|
|||
func (store *UniversalRedis2Store) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
_, err = store.Client.Del(string(key)).Result() |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
import ( |
|||
"bytes" |
@ -1,4 +1,4 @@ |
|||
package filer2 |
|||
package filer |
|||
|
|||
const ( |
|||
TopicsDir = "/topics" |
Write
Preview
Loading…
Cancel
Save
Reference in new issue