integrate: all-zapdb storage onto wire-rip HEAD
This commit is contained in:
zeekay
+5
-2
@@ -1,3 +1,6 @@
|
||||
[zapdb]
|
||||
[luxdb]
|
||||
# Filer metadata store over luxfi/database (the lux-canonical KV abstraction).
|
||||
# backend: zapdb (default, ZAP-native) | pebble | leveldb.
|
||||
enabled = true
|
||||
dir = "/data/filerzdb"
|
||||
backend = "zapdb"
|
||||
dir = "/data/filerldb2"
|
||||
|
||||
@@ -123,7 +123,9 @@ require (
|
||||
github.com/hashicorp/vault/api v1.23.0
|
||||
github.com/jhump/protoreflect v1.18.0
|
||||
github.com/linkedin/goavro/v2 v2.15.0
|
||||
github.com/luxfi/zapdb v1.10.1
|
||||
github.com/luxfi/database v1.19.0
|
||||
github.com/luxfi/log v1.4.3
|
||||
github.com/luxfi/metric v1.5.8
|
||||
github.com/minio/crc64nvme v1.1.1
|
||||
github.com/orcaman/concurrent-map/v2 v2.0.1
|
||||
github.com/parquet-go/parquet-go v0.30.1
|
||||
@@ -152,6 +154,7 @@ require (
|
||||
filippo.io/hpke v0.4.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.7.1 // indirect
|
||||
github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect
|
||||
github.com/DataDog/zstd v1.5.7 // indirect
|
||||
github.com/FilenCloudDienste/filen-sdk-go v0.0.39 // indirect
|
||||
github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0 // indirect
|
||||
github.com/adrg/xdg v0.5.3 // indirect
|
||||
@@ -163,6 +166,12 @@ require (
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
|
||||
github.com/cenkalti/backoff/v5 v5.0.3 // indirect
|
||||
github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
|
||||
github.com/cockroachdb/errors v1.12.0 // indirect
|
||||
github.com/cockroachdb/fifo v0.0.0-20240816210425-c5d0cb0b6fc0 // indirect
|
||||
github.com/cockroachdb/logtags v0.0.0-20241215232642-bb51bb14a506 // indirect
|
||||
github.com/cockroachdb/pebble v1.1.5 // indirect
|
||||
github.com/cockroachdb/redact v1.1.8 // indirect
|
||||
github.com/cockroachdb/tokenbucket v0.0.0-20250429170803-42689b6311bb // indirect
|
||||
github.com/containerd/errdefs v1.0.0 // indirect
|
||||
github.com/containerd/errdefs/pkg v0.3.0 // indirect
|
||||
github.com/containerd/log v0.1.0 // indirect
|
||||
@@ -197,6 +206,8 @@ require (
|
||||
github.com/jhump/protoreflect/v2 v2.0.0-beta.1 // indirect
|
||||
github.com/jtolds/gls v4.20.0+incompatible // indirect
|
||||
github.com/klauspost/crc32 v1.3.0 // indirect
|
||||
github.com/kr/pretty v0.3.1 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
github.com/lithammer/shortuuid/v3 v3.0.7 // indirect
|
||||
github.com/luxfi/accel v1.2.4 // indirect
|
||||
github.com/luxfi/age v1.5.0 // indirect
|
||||
@@ -207,14 +218,11 @@ require (
|
||||
github.com/luxfi/constants v1.5.8 // indirect
|
||||
github.com/luxfi/container v0.0.4 // indirect
|
||||
github.com/luxfi/crypto v1.19.22 // indirect
|
||||
github.com/luxfi/database v1.19.0 // indirect
|
||||
github.com/luxfi/geth v1.17.11 // indirect
|
||||
github.com/luxfi/ids v1.2.15 // indirect
|
||||
github.com/luxfi/log v1.4.3 // indirect
|
||||
github.com/luxfi/math v1.4.1 // indirect
|
||||
github.com/luxfi/math/big v0.1.0 // indirect
|
||||
github.com/luxfi/mdns v0.1.1 // indirect
|
||||
github.com/luxfi/metric v1.5.8 // indirect
|
||||
github.com/luxfi/mock v0.1.1 // indirect
|
||||
github.com/luxfi/p2p v1.21.1 // indirect
|
||||
github.com/luxfi/pq v1.0.3 // indirect
|
||||
@@ -223,6 +231,7 @@ require (
|
||||
github.com/luxfi/version v1.0.1 // indirect
|
||||
github.com/luxfi/warp v1.19.3 // indirect
|
||||
github.com/luxfi/zap v0.7.2 // indirect
|
||||
github.com/luxfi/zapdb v1.10.1 // indirect
|
||||
github.com/magiconair/properties v1.8.10 // indirect
|
||||
github.com/miekg/dns v1.1.72 // indirect
|
||||
github.com/minio/md5-simd v1.1.2 // indirect
|
||||
@@ -244,6 +253,7 @@ require (
|
||||
github.com/quic-go/qpack v0.6.0 // indirect
|
||||
github.com/rclone/Proton-API-Bridge v1.0.3 // indirect
|
||||
github.com/rclone/go-proton-api v1.0.2 // indirect
|
||||
github.com/rogpeppe/go-internal v1.14.1 // indirect
|
||||
github.com/rs/xid v1.6.0 // indirect
|
||||
github.com/ryanuber/go-glob v1.0.0 // indirect
|
||||
github.com/smarty/assertions v1.15.0 // indirect
|
||||
|
||||
@@ -99,6 +99,8 @@ github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7Oputl
|
||||
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
|
||||
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
|
||||
github.com/DataDog/zstd v1.5.2/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
|
||||
github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE=
|
||||
github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
|
||||
github.com/FilenCloudDienste/filen-sdk-go v0.0.39 h1:tgV5jYL6dsXop9TpDTIQU6UwJjws122HrwskaEE/igY=
|
||||
github.com/FilenCloudDienste/filen-sdk-go v0.0.39/go.mod h1:0cBhKXQg49XbKZZfk5TCDa3sVLP+xMxZTWL+7KY0XR0=
|
||||
github.com/Files-com/files-sdk-go/v3 v3.3.82 h1:2RfP0d2QgkFH64BjZSWd59aMsc28IyWsUuHqU0txBtY=
|
||||
@@ -295,6 +297,20 @@ github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnht
|
||||
github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4=
|
||||
github.com/cockroachdb/datadriven v1.0.3-0.20230413201302-be42291fc80f h1:otljaYPt5hWxV3MUfO5dFPFiOXg9CyG5/kCfayTqsJ4=
|
||||
github.com/cockroachdb/datadriven v1.0.3-0.20230413201302-be42291fc80f/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU=
|
||||
github.com/cockroachdb/errors v1.12.0 h1:d7oCs6vuIMUQRVbi6jWWWEJZahLCfJpnJSVobd1/sUo=
|
||||
github.com/cockroachdb/errors v1.12.0/go.mod h1:SvzfYNNBshAVbZ8wzNc/UPK3w1vf0dKDUP41ucAIf7g=
|
||||
github.com/cockroachdb/fifo v0.0.0-20240816210425-c5d0cb0b6fc0 h1:pU88SPhIFid6/k0egdR5V6eALQYq2qbSmukrkgIh/0A=
|
||||
github.com/cockroachdb/fifo v0.0.0-20240816210425-c5d0cb0b6fc0/go.mod h1:9/y3cnZ5GKakj/H4y9r9GTjCvAFta7KLgSHPJJYc52M=
|
||||
github.com/cockroachdb/logtags v0.0.0-20241215232642-bb51bb14a506 h1:ASDL+UJcILMqgNeV5jiqR4j+sTuvQNHdf2chuKj1M5k=
|
||||
github.com/cockroachdb/logtags v0.0.0-20241215232642-bb51bb14a506/go.mod h1:Mw7HqKr2kdtu6aYGn3tPmAftiP3QPX63LdK/zcariIo=
|
||||
github.com/cockroachdb/pebble v1.1.5 h1:5AAWCBWbat0uE0blr8qzufZP5tBjkRyy/jWe1QWLnvw=
|
||||
github.com/cockroachdb/pebble v1.1.5/go.mod h1:17wO9el1YEigxkP/YtV8NtCivQDgoCyBg5c4VR/eOWo=
|
||||
github.com/cockroachdb/redact v1.1.8 h1:8eVLLj6juKxiKrAEw2b8cJvNqWq++U8WOfQFuL7KTaA=
|
||||
github.com/cockroachdb/redact v1.1.8/go.mod h1:GceHHpJ0rMDpYARL5In88Alq/xMBUtVlz7Qxix6ZVkw=
|
||||
github.com/cockroachdb/tokenbucket v0.0.0-20250429170803-42689b6311bb h1:3bCgBvB8PbJVMX1ouCcSIxvsqKPYM7gs72o0zC76n9g=
|
||||
github.com/cockroachdb/tokenbucket v0.0.0-20250429170803-42689b6311bb/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ=
|
||||
github.com/cognusion/imaging v1.0.3 h1:nHyIeEVDV8JkBbuhgx8iSBW72W8rHbDEyruA0Jh7Lnk=
|
||||
github.com/cognusion/imaging v1.0.3/go.mod h1:38tFLFhGK81ORThZG8dVXPtp1uhd+3xM1WXuMn+unOA=
|
||||
github.com/colinmarc/hdfs/v2 v2.4.0 h1:v6R8oBx/Wu9fHpdPoJJjpGSUxo8NhHIwrwsfhFvU9W0=
|
||||
@@ -313,6 +329,7 @@ github.com/coreos/go-systemd/v22 v22.6.0 h1:aGVa/v8B7hpb0TKl0MWoAavPDmHvobFe5R5z
|
||||
github.com/coreos/go-systemd/v22 v22.6.0/go.mod h1:iG+pp635Fo7ZmV/j14KUcmEyWF+0X7Lua8rrTWzYgWU=
|
||||
github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA=
|
||||
github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
|
||||
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
|
||||
github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY=
|
||||
github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
|
||||
github.com/creasty/defaults v1.8.0 h1:z27FJxCAa0JKt3utc0sCImAEb+spPucmKoOdLHvHYKk=
|
||||
@@ -985,6 +1002,7 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ue
|
||||
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
|
||||
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
|
||||
github.com/pkg/diff v0.0.0-20200914180035-5b29258ca4f7/go.mod h1:zO8QMzTeZd5cpnIkz/Gn6iK0jDfGicM1nynOkkPIl28=
|
||||
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
|
||||
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
@@ -1061,6 +1079,7 @@ github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
|
||||
github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
|
||||
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
|
||||
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
|
||||
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
|
||||
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||
github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
|
||||
|
||||
@@ -851,8 +851,12 @@ filer:
|
||||
|
||||
# extraEnvVars is a list of extra environment variables to set with the stateful set.
|
||||
extraEnvironmentVars:
|
||||
# zapdb is the on-disk filer store. Enable "enablePVC" to persist it, or you may lose your data.
|
||||
WEED_ZAPDB_ENABLED: "true"
|
||||
# Filer metadata store: luxfi/database (the lux-canonical KV abstraction).
|
||||
# backend selects the engine: zapdb (default, ZAP-native) | pebble | leveldb.
|
||||
# Persisting metadata requires "enablePVC" below or you may lose your data.
|
||||
WEED_LUXDB_ENABLED: "true"
|
||||
WEED_LUXDB_BACKEND: "zapdb"
|
||||
WEED_LUXDB_DIR: "/data/filerldb2"
|
||||
# with http DELETE, by default the filer would check whether a folder is empty.
|
||||
# recursive_delete will delete all sub folders and files, similar to "rm -Rf"
|
||||
WEED_FILER_OPTIONS_RECURSIVE_DELETE: "false"
|
||||
|
||||
@@ -21,11 +21,16 @@ recursive_delete = false
|
||||
# The following are filer store options
|
||||
####################################################
|
||||
|
||||
[zapdb]
|
||||
# local on disk, transactional LSM-tree key-value store.
|
||||
# default single-machine store, fast and scalable.
|
||||
[luxdb]
|
||||
# Filer metadata store over luxfi/database (the lux-canonical KV abstraction).
|
||||
# Local on disk, for single-machine setups; fairly scalable. Recommended.
|
||||
# backend selects the engine:
|
||||
# zapdb (default) ZAP-native, always available.
|
||||
# pebble requires building with -tags=pebbledb.
|
||||
# leveldb requires building with -tags=leveldb.
|
||||
enabled = true
|
||||
dir = "./filerzdb" # directory to store zapdb files
|
||||
backend = "zapdb"
|
||||
dir = "./filerldb2" # directory to store the metadata db
|
||||
|
||||
[sqlite]
|
||||
# local on disk, embedded SQL store. each bucket has its own table.
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
//go:build leveldb
|
||||
|
||||
package luxdb
|
||||
|
||||
import "testing"
|
||||
|
||||
// With -tags=leveldb the "leveldb" backend must open a working store.
|
||||
func TestLevelDBBackendRoundTrip(t *testing.T) {
|
||||
runBackendRoundTrip(t, "leveldb")
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
//go:build pebbledb
|
||||
|
||||
package luxdb
|
||||
|
||||
import "testing"
|
||||
|
||||
// With -tags=pebbledb the "pebble" alias must open a working store.
|
||||
func TestPebbleBackendRoundTrip(t *testing.T) {
|
||||
runBackendRoundTrip(t, "pebble")
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
"github.com/hanzoai/s3/s3/pb/filer_pb"
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
// runBackendRoundTrip exercises the full FilerStore + KV surface against the
|
||||
// named backend. It is shared by the default (zapdb) test and the tag-gated
|
||||
// pebble/leveldb tests so every selectable backend is proven, not just compiled.
|
||||
func runBackendRoundTrip(t *testing.T, backend string) {
|
||||
t.Helper()
|
||||
store := &LuxDBStore{}
|
||||
if err := store.initialize(backend, t.TempDir()); err != nil {
|
||||
t.Fatalf("init %s: %v", backend, err)
|
||||
}
|
||||
defer store.Shutdown()
|
||||
ctx := context.Background()
|
||||
|
||||
// Insert two entries under the same directory and one elsewhere.
|
||||
mk := func(path string) *filer.Entry {
|
||||
return &filer.Entry{FullPath: util.FullPath(path), Attr: filer.Attr{Mode: 0640}}
|
||||
}
|
||||
for _, p := range []string{"/d/a.txt", "/d/b.txt", "/other/c.txt"} {
|
||||
if err := store.InsertEntry(ctx, mk(p)); err != nil {
|
||||
t.Fatalf("%s insert %s: %v", backend, p, err)
|
||||
}
|
||||
}
|
||||
|
||||
// FindEntry hits.
|
||||
if _, err := store.FindEntry(ctx, util.FullPath("/d/a.txt")); err != nil {
|
||||
t.Fatalf("%s find a.txt: %v", backend, err)
|
||||
}
|
||||
// FindEntry miss maps to filer_pb.ErrNotFound.
|
||||
if _, err := store.FindEntry(ctx, util.FullPath("/d/missing")); err != filer_pb.ErrNotFound {
|
||||
t.Fatalf("%s find missing: want ErrNotFound, got %v", backend, err)
|
||||
}
|
||||
|
||||
// ListDirectoryEntries returns exactly the two children of /d in order.
|
||||
var names []string
|
||||
if _, err := store.ListDirectoryEntries(ctx, util.FullPath("/d"), "", true, 100,
|
||||
func(e *filer.Entry) (bool, error) {
|
||||
names = append(names, e.Name())
|
||||
return true, nil
|
||||
}); err != nil {
|
||||
t.Fatalf("%s list /d: %v", backend, err)
|
||||
}
|
||||
if len(names) != 2 || names[0] != "a.txt" || names[1] != "b.txt" {
|
||||
t.Fatalf("%s list /d = %v, want [a.txt b.txt]", backend, names)
|
||||
}
|
||||
|
||||
// DeleteFolderChildren clears /d but not /other.
|
||||
if err := store.DeleteFolderChildren(ctx, util.FullPath("/d")); err != nil {
|
||||
t.Fatalf("%s delete children /d: %v", backend, err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, util.FullPath("/d/a.txt")); err != filer_pb.ErrNotFound {
|
||||
t.Fatalf("%s a.txt should be gone: %v", backend, err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, util.FullPath("/other/c.txt")); err != nil {
|
||||
t.Fatalf("%s c.txt should survive: %v", backend, err)
|
||||
}
|
||||
|
||||
// KV surface.
|
||||
if err := store.KvPut(ctx, []byte("k"), []byte("v")); err != nil {
|
||||
t.Fatalf("%s kv put: %v", backend, err)
|
||||
}
|
||||
got, err := store.KvGet(ctx, []byte("k"))
|
||||
if err != nil || !bytes.Equal(got, []byte("v")) {
|
||||
t.Fatalf("%s kv get = %q, %v", backend, got, err)
|
||||
}
|
||||
if err := store.KvDelete(ctx, []byte("k")); err != nil {
|
||||
t.Fatalf("%s kv delete: %v", backend, err)
|
||||
}
|
||||
if _, err := store.KvGet(ctx, []byte("k")); err != filer.ErrKvNotFound {
|
||||
t.Fatalf("%s kv get after delete: want ErrKvNotFound, got %v", backend, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestZapdbBackendRoundTrip proves the default backend on the default build.
|
||||
func TestZapdbBackendRoundTrip(t *testing.T) {
|
||||
runBackendRoundTrip(t, "zapdb")
|
||||
}
|
||||
|
||||
// TestDefaultBackendRoundTrip proves an empty backend string defaults to zapdb.
|
||||
func TestDefaultBackendRoundTrip(t *testing.T) {
|
||||
runBackendRoundTrip(t, "")
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
// TestKeyEncodingByteIdentical pins the exact key bytes so the luxdb layout
|
||||
// stays byte-identical to the historical leveldb store (dir + 0x00 + name).
|
||||
// Any drift here silently breaks reading an existing on-disk store.
|
||||
func TestKeyEncodingByteIdentical(t *testing.T) {
|
||||
got := genKey("/a/b", "c.txt")
|
||||
want := append(append([]byte("/a/b"), DIR_FILE_SEPARATOR), []byte("c.txt")...)
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Fatalf("genKey = %q, want %q", got, want)
|
||||
}
|
||||
|
||||
// Directory prefix without a start file: dir + 0x00.
|
||||
gotPrefix := genDirectoryKeyPrefix(util.FullPath("/a/b"), "")
|
||||
wantPrefix := append([]byte("/a/b"), DIR_FILE_SEPARATOR)
|
||||
if !bytes.Equal(gotPrefix, wantPrefix) {
|
||||
t.Fatalf("genDirectoryKeyPrefix empty = %q, want %q", gotPrefix, wantPrefix)
|
||||
}
|
||||
|
||||
// Directory prefix with a start file: dir + 0x00 + start.
|
||||
gotStart := genDirectoryKeyPrefix(util.FullPath("/a/b"), "m")
|
||||
wantStart := append(append([]byte("/a/b"), DIR_FILE_SEPARATOR), []byte("m")...)
|
||||
if !bytes.Equal(gotStart, wantStart) {
|
||||
t.Fatalf("genDirectoryKeyPrefix start = %q, want %q", gotStart, wantStart)
|
||||
}
|
||||
|
||||
// getNameFromKey is the inverse of genKey on the name component.
|
||||
if name := getNameFromKey(genKey("/a/b", "c.txt")); name != "c.txt" {
|
||||
t.Fatalf("getNameFromKey = %q, want c.txt", name)
|
||||
}
|
||||
// Root-level file: empty dir + 0x00 + name.
|
||||
if name := getNameFromKey(genKey("", "root.txt")); name != "root.txt" {
|
||||
t.Fatalf("getNameFromKey root = %q, want root.txt", name)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,314 @@
|
||||
// Package luxdb implements the filer metadata store over luxfi/database, the
|
||||
// lux-canonical key-value abstraction. The backend is selectable: "zapdb"
|
||||
// (default, ZAP-native), "pebble", or "leveldb". The key encoding is
|
||||
// byte-identical to the historical leveldb store so on-disk layout matches.
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/luxfi/database"
|
||||
"github.com/luxfi/database/factory"
|
||||
log "github.com/luxfi/log"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/pb/filer_pb"
|
||||
s3_util "github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
const (
|
||||
DIR_FILE_SEPARATOR = byte(0x00)
|
||||
|
||||
// defaultBackend is the canonical metadata backend: ZAP-native zapdb.
|
||||
defaultBackend = "zapdb"
|
||||
)
|
||||
|
||||
var (
|
||||
_ = filer.FilerStore(&LuxDBStore{})
|
||||
_ = filer.Debuggable(&LuxDBStore{})
|
||||
)
|
||||
|
||||
func init() {
|
||||
filer.Stores = append(filer.Stores, &LuxDBStore{})
|
||||
}
|
||||
|
||||
// LuxDBStore is a filer.FilerStore backed by a luxfi/database.Database.
|
||||
type LuxDBStore struct {
|
||||
db database.Database
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) GetName() string {
|
||||
return "luxdb"
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) Initialize(configuration s3_util.Configuration, prefix string) (err error) {
|
||||
dir := configuration.GetString(prefix + "dir")
|
||||
backend := configuration.GetString(prefix + "backend")
|
||||
return store.initialize(backend, dir)
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) initialize(backend, dir string) (err error) {
|
||||
if backend == "" {
|
||||
backend = defaultBackend
|
||||
}
|
||||
// "pebble" is the operator-facing alias for the luxfi/database "pebbledb"
|
||||
// backend name.
|
||||
if backend == "pebble" {
|
||||
backend = "pebbledb"
|
||||
}
|
||||
|
||||
glog.V(0).Infof("filer luxdb store: backend=%s dir=%s", backend, dir)
|
||||
os.MkdirAll(dir, 0755)
|
||||
if err := s3_util.TestFolderWritable(dir); err != nil {
|
||||
return fmt.Errorf("check luxdb folder %s writable: %s", dir, err)
|
||||
}
|
||||
|
||||
// factory.New(name, dbPath, readOnly, config, gatherer, logger, metricsPrefix, meterDBRegName).
|
||||
// zapdb is always registered (default); pebbledb/leveldb register under
|
||||
// their build tags. A nil gatherer/logger keeps the store metric-free and
|
||||
// silent, matching the embedded single-process filer use.
|
||||
db, err := factory.New(backend, dir, false, nil, nil, log.Noop(), "", "")
|
||||
if err != nil {
|
||||
return fmt.Errorf("open luxdb backend %q at %s: %w", backend, dir, err)
|
||||
}
|
||||
store.db = db
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) BeginTransaction(ctx context.Context) (context.Context, error) {
|
||||
return ctx, nil
|
||||
}
|
||||
func (store *LuxDBStore) CommitTransaction(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
func (store *LuxDBStore) RollbackTransaction(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
|
||||
key := genKey(entry.DirAndName())
|
||||
|
||||
value, err := entry.EncodeAttributesAndChunks()
|
||||
if err != nil {
|
||||
return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
|
||||
}
|
||||
|
||||
if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
|
||||
value = s3_util.MaybeGzipData(value)
|
||||
}
|
||||
|
||||
if err = store.db.Put(key, value); err != nil {
|
||||
return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
|
||||
return store.InsertEntry(ctx, entry)
|
||||
}
|
||||
|
||||
// BatchInsertEntries inserts multiple entries in a single batch write. This is
|
||||
// more efficient than inserting entries one by one as it reduces the number of
|
||||
// write operations and syncs to disk.
|
||||
func (store *LuxDBStore) BatchInsertEntries(ctx context.Context, entries []*filer.Entry) error {
|
||||
if len(entries) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
batch := store.db.NewBatch()
|
||||
|
||||
for _, entry := range entries {
|
||||
key := genKey(entry.DirAndName())
|
||||
|
||||
value, err := entry.EncodeAttributesAndChunks()
|
||||
if err != nil {
|
||||
return fmt.Errorf("encoding %s %+v: %w", entry.FullPath, entry.Attr, err)
|
||||
}
|
||||
|
||||
if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
|
||||
value = s3_util.MaybeGzipData(value)
|
||||
}
|
||||
|
||||
if err := batch.Put(key, value); err != nil {
|
||||
return fmt.Errorf("batch put %s: %w", entry.FullPath, err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := batch.Write(); err != nil {
|
||||
return fmt.Errorf("batch write: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) FindEntry(ctx context.Context, fullpath s3_util.FullPath) (entry *filer.Entry, err error) {
|
||||
key := genKey(fullpath.DirAndName())
|
||||
|
||||
data, err := store.db.Get(key)
|
||||
|
||||
if err == database.ErrNotFound {
|
||||
return nil, filer_pb.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
entry = &filer.Entry{
|
||||
FullPath: fullpath,
|
||||
}
|
||||
err = entry.DecodeAttributesAndChunks(s3_util.MaybeDecompressData(data))
|
||||
if err != nil {
|
||||
return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
return entry, nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) DeleteEntry(ctx context.Context, fullpath s3_util.FullPath) (err error) {
|
||||
key := genKey(fullpath.DirAndName())
|
||||
|
||||
if err = store.db.Delete(key); err != nil {
|
||||
return fmt.Errorf("delete %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) DeleteFolderChildren(ctx context.Context, fullpath s3_util.FullPath) (err error) {
|
||||
|
||||
batch := store.db.NewBatch()
|
||||
|
||||
directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
|
||||
iter := store.db.NewIteratorWithStart(directoryPrefix)
|
||||
for iter.Next() {
|
||||
key := iter.Key()
|
||||
if !bytes.HasPrefix(key, directoryPrefix) {
|
||||
break
|
||||
}
|
||||
fileName := getNameFromKey(key)
|
||||
if fileName == "" {
|
||||
continue
|
||||
}
|
||||
if err := batch.Delete(genKey(string(fullpath), fileName)); err != nil {
|
||||
iter.Release()
|
||||
return fmt.Errorf("batch delete %s : %v", fullpath, err)
|
||||
}
|
||||
}
|
||||
iter.Release()
|
||||
if err := iter.Error(); err != nil {
|
||||
return fmt.Errorf("iterate %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
if err = batch.Write(); err != nil {
|
||||
return fmt.Errorf("delete %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) ListDirectoryEntries(ctx context.Context, dirPath s3_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
|
||||
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath s3_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
|
||||
|
||||
directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
|
||||
lastFileStart := directoryPrefix
|
||||
if startFileName != "" {
|
||||
lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
|
||||
}
|
||||
|
||||
iter := store.db.NewIteratorWithStart(lastFileStart)
|
||||
for iter.Next() {
|
||||
key := iter.Key()
|
||||
if !bytes.HasPrefix(key, directoryPrefix) {
|
||||
break
|
||||
}
|
||||
fileName := getNameFromKey(key)
|
||||
if fileName == "" {
|
||||
continue
|
||||
}
|
||||
if fileName == startFileName && !includeStartFile {
|
||||
continue
|
||||
}
|
||||
limit--
|
||||
if limit < 0 {
|
||||
break
|
||||
}
|
||||
lastFileName = fileName
|
||||
entry := &filer.Entry{
|
||||
FullPath: s3_util.NewFullPath(string(dirPath), fileName),
|
||||
}
|
||||
if decodeErr := entry.DecodeAttributesAndChunks(s3_util.MaybeDecompressData(iter.Value())); decodeErr != nil {
|
||||
err = decodeErr
|
||||
glog.V(0).InfofCtx(ctx, "list %s : %v", entry.FullPath, err)
|
||||
break
|
||||
}
|
||||
|
||||
resEachEntryFunc, resEachEntryFuncErr := eachEntryFunc(entry)
|
||||
if resEachEntryFuncErr != nil {
|
||||
err = fmt.Errorf("failed to process eachEntryFunc: %w", resEachEntryFuncErr)
|
||||
break
|
||||
}
|
||||
|
||||
if !resEachEntryFunc {
|
||||
break
|
||||
}
|
||||
}
|
||||
iter.Release()
|
||||
if err == nil {
|
||||
if iterErr := iter.Error(); iterErr != nil {
|
||||
err = fmt.Errorf("iterate %s : %v", dirPath, iterErr)
|
||||
}
|
||||
}
|
||||
|
||||
return lastFileName, err
|
||||
}
|
||||
|
||||
func genKey(dirPath, fileName string) (key []byte) {
|
||||
key = []byte(dirPath)
|
||||
key = append(key, DIR_FILE_SEPARATOR)
|
||||
key = append(key, []byte(fileName)...)
|
||||
return key
|
||||
}
|
||||
|
||||
func genDirectoryKeyPrefix(fullpath s3_util.FullPath, startFileName string) (keyPrefix []byte) {
|
||||
keyPrefix = []byte(string(fullpath))
|
||||
keyPrefix = append(keyPrefix, DIR_FILE_SEPARATOR)
|
||||
if len(startFileName) > 0 {
|
||||
keyPrefix = append(keyPrefix, []byte(startFileName)...)
|
||||
}
|
||||
return keyPrefix
|
||||
}
|
||||
|
||||
func getNameFromKey(key []byte) string {
|
||||
|
||||
sepIndex := len(key) - 1
|
||||
for sepIndex >= 0 && key[sepIndex] != DIR_FILE_SEPARATOR {
|
||||
sepIndex--
|
||||
}
|
||||
|
||||
return string(key[sepIndex+1:])
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) Shutdown() {
|
||||
if store.db != nil {
|
||||
store.db.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) Debug(writer io.Writer) {
|
||||
iter := store.db.NewIterator()
|
||||
for iter.Next() {
|
||||
key := iter.Key()
|
||||
fullName := bytes.Replace(key, []byte{DIR_FILE_SEPARATOR}, []byte{' '}, 1)
|
||||
fmt.Fprintf(writer, "%v\n", string(fullName))
|
||||
}
|
||||
iter.Release()
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/luxfi/database"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
)
|
||||
|
||||
func (store *LuxDBStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
|
||||
|
||||
if err = store.db.Put(key, value); err != nil {
|
||||
return fmt.Errorf("kv put: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
|
||||
|
||||
value, err = store.db.Get(key)
|
||||
|
||||
if err == database.ErrNotFound {
|
||||
return nil, filer.ErrKvNotFound
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("kv get: %w", err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (store *LuxDBStore) KvDelete(ctx context.Context, key []byte) (err error) {
|
||||
|
||||
if err = store.db.Delete(key); err != nil {
|
||||
return fmt.Errorf("kv delete: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
package zapdb
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -15,8 +15,10 @@ import (
|
||||
func TestCreateAndFind(t *testing.T) {
|
||||
testFiler := filer.NewFiler(pb.ServerDiscovery{}, pb.DialOption{}, "", "", "", "", "", 255, nil)
|
||||
dir := t.TempDir()
|
||||
store := &ZapDBStore{}
|
||||
store.initialize(dir)
|
||||
store := &LuxDBStore{}
|
||||
if err := store.initialize("zapdb", dir); err != nil {
|
||||
t.Fatalf("init store: %v", err)
|
||||
}
|
||||
testFiler.SetStore(store)
|
||||
|
||||
fullpath := util.FullPath("/home/chris/this/is/one/file1.jpg")
|
||||
@@ -68,8 +70,10 @@ func TestCreateAndFind(t *testing.T) {
|
||||
func TestEmptyRoot(t *testing.T) {
|
||||
testFiler := filer.NewFiler(pb.ServerDiscovery{}, pb.DialOption{}, "", "", "", "", "", 255, nil)
|
||||
dir := t.TempDir()
|
||||
store := &ZapDBStore{}
|
||||
store.initialize(dir)
|
||||
store := &LuxDBStore{}
|
||||
if err := store.initialize("zapdb", dir); err != nil {
|
||||
t.Fatalf("init store: %v", err)
|
||||
}
|
||||
testFiler.SetStore(store)
|
||||
|
||||
ctx := context.Background()
|
||||
@@ -90,8 +94,10 @@ func TestEmptyRoot(t *testing.T) {
|
||||
func BenchmarkInsertEntry(b *testing.B) {
|
||||
testFiler := filer.NewFiler(pb.ServerDiscovery{}, pb.DialOption{}, "", "", "", "", "", 255, nil)
|
||||
dir := b.TempDir()
|
||||
store := &ZapDBStore{}
|
||||
store.initialize(dir)
|
||||
store := &LuxDBStore{}
|
||||
if err := store.initialize("zapdb", dir); err != nil {
|
||||
b.Fatalf("init store: %v", err)
|
||||
}
|
||||
testFiler.SetStore(store)
|
||||
|
||||
ctx := context.Background()
|
||||
+3
-3
@@ -1,4 +1,4 @@
|
||||
package zapdb
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -28,8 +28,8 @@ func histogramState(t *testing.T) (count uint64, perBucket map[float64]uint64) {
|
||||
|
||||
func TestCreateEntryRecordsObjectSize(t *testing.T) {
|
||||
testFiler := filer.NewFiler(pb.ServerDiscovery{}, pb.DialOption{}, "", "", "", "", "", 255, nil)
|
||||
store := &ZapDBStore{}
|
||||
if err := store.initialize(t.TempDir()); err != nil {
|
||||
store := &LuxDBStore{}
|
||||
if err := store.initialize("zapdb", t.TempDir()); err != nil {
|
||||
t.Fatalf("init store: %v", err)
|
||||
}
|
||||
testFiler.SetStore(store)
|
||||
@@ -0,0 +1,35 @@
|
||||
//go:build pebbledb
|
||||
|
||||
// luxfi/database registers leveldb in its factory under -tags=leveldb but ships
|
||||
// no equivalent registration for pebbledb (factory/leveldb.go has no
|
||||
// factory/pebbledb.go sibling). To make backend="pebble" selectable here, we
|
||||
// register pebbledb in the factory under -tags=pebbledb, mirroring factory's own
|
||||
// leveldb registration. zapdb (the default) needs no registration; it is always
|
||||
// available.
|
||||
package luxdb
|
||||
|
||||
import (
|
||||
"github.com/luxfi/database"
|
||||
"github.com/luxfi/database/factory"
|
||||
"github.com/luxfi/database/pebbledb"
|
||||
log "github.com/luxfi/log"
|
||||
"github.com/luxfi/metric"
|
||||
)
|
||||
|
||||
func init() {
|
||||
factory.RegisterDatabase(pebbledb.Name, newPebbleDB)
|
||||
}
|
||||
|
||||
func newPebbleDB(
|
||||
dbPath string,
|
||||
config []byte,
|
||||
logger log.Logger,
|
||||
registerer metric.Registerer,
|
||||
metricsPrefix string,
|
||||
readOnly bool,
|
||||
) (database.Database, error) {
|
||||
// Defaults consistent with factory's leveldb sizing.
|
||||
cacheSize := 12 // MiB block cache
|
||||
handles := 1024
|
||||
return pebbledb.New(dbPath, cacheSize, handles, pebbledb.Name, readOnly)
|
||||
}
|
||||
@@ -1,335 +0,0 @@
|
||||
package zapdb
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
zapdb "github.com/luxfi/zapdb"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/pb/filer_pb"
|
||||
s3_util "github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
const (
|
||||
// DIR_FILE_SEPARATOR delimits the directory prefix from the file name in a
|
||||
// key. Identical to the leveldb store's separator so the on-disk key layout
|
||||
// is byte-for-byte compatible.
|
||||
DIR_FILE_SEPARATOR = byte(0x00)
|
||||
)
|
||||
|
||||
var (
|
||||
_ = filer.Debuggable(&ZapDBStore{})
|
||||
)
|
||||
|
||||
func init() {
|
||||
filer.Stores = append(filer.Stores, &ZapDBStore{})
|
||||
}
|
||||
|
||||
// ZapDBStore is a filer.FilerStore backed by zapdb (a transactional,
|
||||
// LSM-tree KV store). It mirrors the leveldb store's key encoding exactly,
|
||||
// so data written by either store shares an identical key layout.
|
||||
type ZapDBStore struct {
|
||||
db *zapdb.DB
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) GetName() string {
|
||||
return "zapdb"
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) Initialize(configuration s3_util.Configuration, prefix string) (err error) {
|
||||
dir := configuration.GetString(prefix + "dir")
|
||||
return store.initialize(dir)
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) initialize(dir string) (err error) {
|
||||
glog.V(0).Infof("filer store dir: %s", dir)
|
||||
os.MkdirAll(dir, 0755)
|
||||
if err := s3_util.TestFolderWritable(dir); err != nil {
|
||||
return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
|
||||
}
|
||||
|
||||
opts := zapdb.DefaultOptions(dir).
|
||||
WithLogger(nil).
|
||||
WithSyncWrites(false)
|
||||
|
||||
if store.db, err = zapdb.Open(opts); err != nil {
|
||||
glog.Infof("filer store open dir %s: %v", dir, err)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) BeginTransaction(ctx context.Context) (context.Context, error) {
|
||||
return ctx, nil
|
||||
}
|
||||
func (store *ZapDBStore) CommitTransaction(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
func (store *ZapDBStore) RollbackTransaction(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
|
||||
key := genKey(entry.DirAndName())
|
||||
|
||||
value, err := entry.EncodeAttributesAndChunks()
|
||||
if err != nil {
|
||||
return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
|
||||
}
|
||||
|
||||
if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
|
||||
value = s3_util.MaybeGzipData(value)
|
||||
}
|
||||
|
||||
err = store.db.Update(func(txn *zapdb.Txn) error {
|
||||
return txn.Set(key, value)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
|
||||
|
||||
return store.InsertEntry(ctx, entry)
|
||||
}
|
||||
|
||||
// BatchInsertEntries inserts multiple entries in a single zapdb transaction.
|
||||
// This is more efficient than inserting entries one by one as it reduces
|
||||
// the number of write operations and syncs to disk.
|
||||
func (store *ZapDBStore) BatchInsertEntries(ctx context.Context, entries []*filer.Entry) error {
|
||||
if len(entries) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
err := store.db.Update(func(txn *zapdb.Txn) error {
|
||||
for _, entry := range entries {
|
||||
key := genKey(entry.DirAndName())
|
||||
|
||||
value, err := entry.EncodeAttributesAndChunks()
|
||||
if err != nil {
|
||||
return fmt.Errorf("encoding %s %+v: %w", entry.FullPath, entry.Attr, err)
|
||||
}
|
||||
|
||||
if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
|
||||
value = s3_util.MaybeGzipData(value)
|
||||
}
|
||||
|
||||
if err := txn.Set(key, value); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("batch write: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) FindEntry(ctx context.Context, fullpath s3_util.FullPath) (entry *filer.Entry, err error) {
|
||||
key := genKey(fullpath.DirAndName())
|
||||
|
||||
var data []byte
|
||||
err = store.db.View(func(txn *zapdb.Txn) error {
|
||||
item, getErr := txn.Get(key)
|
||||
if getErr != nil {
|
||||
return getErr
|
||||
}
|
||||
data, getErr = item.ValueCopy(nil)
|
||||
return getErr
|
||||
})
|
||||
|
||||
if err == zapdb.ErrKeyNotFound {
|
||||
return nil, filer_pb.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
entry = &filer.Entry{
|
||||
FullPath: fullpath,
|
||||
}
|
||||
err = entry.DecodeAttributesAndChunks(s3_util.MaybeDecompressData(data))
|
||||
if err != nil {
|
||||
return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
|
||||
}
|
||||
|
||||
return entry, nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) DeleteEntry(ctx context.Context, fullpath s3_util.FullPath) (err error) {
|
||||
key := genKey(fullpath.DirAndName())
|
||||
|
||||
err = store.db.Update(func(txn *zapdb.Txn) error {
|
||||
return txn.Delete(key)
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) DeleteFolderChildren(ctx context.Context, fullpath s3_util.FullPath) (err error) {
|
||||
|
||||
directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
|
||||
|
||||
// Collect the child names under a read-only iterator first, then delete in
|
||||
// a separate write transaction. zapdb forbids mutating keys while iterating
|
||||
// within the same transaction, so the two phases must not share a txn.
|
||||
var fileNames []string
|
||||
err = store.db.View(func(txn *zapdb.Txn) error {
|
||||
opts := zapdb.DefaultIteratorOptions
|
||||
opts.PrefetchValues = false
|
||||
opts.Prefix = directoryPrefix
|
||||
iter := txn.NewIterator(opts)
|
||||
defer iter.Close()
|
||||
for iter.Seek(directoryPrefix); iter.Valid(); iter.Next() {
|
||||
key := iter.Item().Key()
|
||||
if !bytes.HasPrefix(key, directoryPrefix) {
|
||||
break
|
||||
}
|
||||
fileName := getNameFromKey(key)
|
||||
if fileName == "" {
|
||||
continue
|
||||
}
|
||||
fileNames = append(fileNames, fileName)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
err = store.db.Update(func(txn *zapdb.Txn) error {
|
||||
for _, fileName := range fileNames {
|
||||
if err := txn.Delete(genKey(string(fullpath), fileName)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete %s : %v", fullpath, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) ListDirectoryEntries(ctx context.Context, dirPath s3_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
|
||||
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath s3_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
|
||||
|
||||
directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
|
||||
lastFileStart := directoryPrefix
|
||||
if startFileName != "" {
|
||||
lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
|
||||
}
|
||||
|
||||
err = store.db.View(func(txn *zapdb.Txn) error {
|
||||
opts := zapdb.DefaultIteratorOptions
|
||||
opts.Prefix = directoryPrefix
|
||||
iter := txn.NewIterator(opts)
|
||||
defer iter.Close()
|
||||
for iter.Seek(lastFileStart); iter.Valid(); iter.Next() {
|
||||
item := iter.Item()
|
||||
key := item.Key()
|
||||
if !bytes.HasPrefix(key, directoryPrefix) {
|
||||
break
|
||||
}
|
||||
fileName := getNameFromKey(key)
|
||||
if fileName == "" {
|
||||
continue
|
||||
}
|
||||
if fileName == startFileName && !includeStartFile {
|
||||
continue
|
||||
}
|
||||
limit--
|
||||
if limit < 0 {
|
||||
break
|
||||
}
|
||||
lastFileName = fileName
|
||||
value, valueErr := item.ValueCopy(nil)
|
||||
if valueErr != nil {
|
||||
return valueErr
|
||||
}
|
||||
entry := &filer.Entry{
|
||||
FullPath: s3_util.NewFullPath(string(dirPath), fileName),
|
||||
}
|
||||
if decodeErr := entry.DecodeAttributesAndChunks(s3_util.MaybeDecompressData(value)); decodeErr != nil {
|
||||
glog.V(0).InfofCtx(ctx, "list %s : %v", entry.FullPath, decodeErr)
|
||||
return decodeErr
|
||||
}
|
||||
|
||||
resEachEntryFunc, resEachEntryFuncErr := eachEntryFunc(entry)
|
||||
if resEachEntryFuncErr != nil {
|
||||
return fmt.Errorf("failed to process eachEntryFunc: %w", resEachEntryFuncErr)
|
||||
}
|
||||
|
||||
if !resEachEntryFunc {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
return lastFileName, err
|
||||
}
|
||||
|
||||
func genKey(dirPath, fileName string) (key []byte) {
|
||||
key = []byte(dirPath)
|
||||
key = append(key, DIR_FILE_SEPARATOR)
|
||||
key = append(key, []byte(fileName)...)
|
||||
return key
|
||||
}
|
||||
|
||||
func genDirectoryKeyPrefix(fullpath s3_util.FullPath, startFileName string) (keyPrefix []byte) {
|
||||
keyPrefix = []byte(string(fullpath))
|
||||
keyPrefix = append(keyPrefix, DIR_FILE_SEPARATOR)
|
||||
if len(startFileName) > 0 {
|
||||
keyPrefix = append(keyPrefix, []byte(startFileName)...)
|
||||
}
|
||||
return keyPrefix
|
||||
}
|
||||
|
||||
func getNameFromKey(key []byte) string {
|
||||
|
||||
sepIndex := len(key) - 1
|
||||
for sepIndex >= 0 && key[sepIndex] != DIR_FILE_SEPARATOR {
|
||||
sepIndex--
|
||||
}
|
||||
|
||||
return string(key[sepIndex+1:])
|
||||
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) Shutdown() {
|
||||
store.db.Close()
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) Debug(writer io.Writer) {
|
||||
store.db.View(func(txn *zapdb.Txn) error {
|
||||
opts := zapdb.DefaultIteratorOptions
|
||||
opts.PrefetchValues = false
|
||||
iter := txn.NewIterator(opts)
|
||||
defer iter.Close()
|
||||
for iter.Rewind(); iter.Valid(); iter.Next() {
|
||||
key := iter.Item().Key()
|
||||
fullName := bytes.Replace(key, []byte{DIR_FILE_SEPARATOR}, []byte{' '}, 1)
|
||||
fmt.Fprintf(writer, "%v\n", string(fullName))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
@@ -1,58 +0,0 @@
|
||||
package zapdb
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
zapdb "github.com/luxfi/zapdb"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
)
|
||||
|
||||
func (store *ZapDBStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
|
||||
|
||||
err = store.db.Update(func(txn *zapdb.Txn) error {
|
||||
return txn.Set(key, value)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("kv put: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
|
||||
|
||||
err = store.db.View(func(txn *zapdb.Txn) error {
|
||||
item, getErr := txn.Get(key)
|
||||
if getErr != nil {
|
||||
return getErr
|
||||
}
|
||||
value, getErr = item.ValueCopy(nil)
|
||||
return getErr
|
||||
})
|
||||
|
||||
if err == zapdb.ErrKeyNotFound {
|
||||
return nil, filer.ErrKvNotFound
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("kv get: %w", err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (store *ZapDBStore) KvDelete(ctx context.Context, key []byte) (err error) {
|
||||
|
||||
err = store.db.Update(func(txn *zapdb.Txn) error {
|
||||
return txn.Delete(key)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("kv delete: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -1,6 +1,10 @@
|
||||
package meta_cache
|
||||
|
||||
import "github.com/hanzoai/s3/s3/util"
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
var (
|
||||
_ = util.Configuration(&cacheConfig{})
|
||||
@@ -8,10 +12,14 @@ var (
|
||||
|
||||
// implementing util.Configuration
|
||||
type cacheConfig struct {
|
||||
dir string
|
||||
dir string
|
||||
backend string
|
||||
}
|
||||
|
||||
func (c cacheConfig) GetString(key string) string {
|
||||
if strings.HasSuffix(key, "backend") {
|
||||
return c.backend
|
||||
}
|
||||
return c.dir
|
||||
}
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
"fmt"
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
"github.com/hanzoai/s3/s3/filer/zapdb"
|
||||
"github.com/hanzoai/s3/s3/filer/luxdb"
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/pb/filer_pb"
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
@@ -24,9 +24,9 @@ import (
|
||||
// e.g. fill fileId field for chunks
|
||||
|
||||
type MetaCache struct {
|
||||
root util.FullPath
|
||||
localStore filer.VirtualFilerStore
|
||||
zapdbStore *zapdb.ZapDBStore // direct reference for batch operations
|
||||
root util.FullPath
|
||||
localStore filer.VirtualFilerStore
|
||||
luxdbStore *luxdb.LuxDBStore // direct reference for batch operations
|
||||
sync.RWMutex
|
||||
uidGidMapper *UidGidMapper
|
||||
markCachedFn func(fullpath util.FullPath)
|
||||
@@ -97,11 +97,11 @@ type metadataApplyRequest struct {
|
||||
|
||||
func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
|
||||
markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(util.FullPath, *filer_pb.Entry), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
|
||||
zapdbStore, virtualStore := openMetaStore(dbFolder)
|
||||
luxdbStore, virtualStore := openMetaStore(dbFolder)
|
||||
mc := &MetaCache{
|
||||
root: root,
|
||||
localStore: virtualStore,
|
||||
zapdbStore: zapdbStore,
|
||||
luxdbStore: luxdbStore,
|
||||
markCachedFn: markCachedFn,
|
||||
isCachedFn: isCachedFn,
|
||||
uidGidMapper: uidGidMapper,
|
||||
@@ -124,14 +124,15 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
|
||||
return mc
|
||||
}
|
||||
|
||||
func openMetaStore(dbFolder string) (*zapdb.ZapDBStore, filer.VirtualFilerStore) {
|
||||
func openMetaStore(dbFolder string) (*luxdb.LuxDBStore, filer.VirtualFilerStore) {
|
||||
|
||||
os.RemoveAll(dbFolder)
|
||||
os.MkdirAll(dbFolder, 0755)
|
||||
|
||||
store := &zapdb.ZapDBStore{}
|
||||
store := &luxdb.LuxDBStore{}
|
||||
config := &cacheConfig{
|
||||
dir: dbFolder,
|
||||
dir: dbFolder,
|
||||
backend: "zapdb",
|
||||
}
|
||||
|
||||
if err := store.Initialize(config, ""); err != nil {
|
||||
@@ -152,10 +153,10 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) erro
|
||||
return mc.localStore.InsertEntry(ctx, entry)
|
||||
}
|
||||
|
||||
// doBatchInsertEntries inserts multiple entries using zapdb's batch write.
|
||||
// doBatchInsertEntries inserts multiple entries using the store's batch write.
|
||||
// This is more efficient than inserting entries one by one.
|
||||
func (mc *MetaCache) doBatchInsertEntries(ctx context.Context, entries []*filer.Entry) error {
|
||||
return mc.zapdbStore.BatchInsertEntries(ctx, entries)
|
||||
return mc.luxdbStore.BatchInsertEntries(ctx, entries)
|
||||
}
|
||||
|
||||
func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
|
||||
"github.com/hanzoai/s3/s3/filer"
|
||||
_ "github.com/hanzoai/s3/s3/filer/foundationdb"
|
||||
_ "github.com/hanzoai/s3/s3/filer/luxdb"
|
||||
"github.com/hanzoai/s3/s3/filer/posixlock"
|
||||
_ "github.com/hanzoai/s3/s3/filer/sqlite"
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
@@ -182,8 +183,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
|
||||
}
|
||||
|
||||
if !util.LoadConfiguration("filer", false) {
|
||||
v.SetDefault("zapdb.enabled", true)
|
||||
v.SetDefault("zapdb.dir", option.DefaultLevelDbDir)
|
||||
v.SetDefault("luxdb.enabled", true)
|
||||
v.SetDefault("luxdb.backend", "zapdb")
|
||||
v.SetDefault("luxdb.dir", option.DefaultLevelDbDir)
|
||||
_, err := os.Stat(option.DefaultLevelDbDir)
|
||||
if os.IsNotExist(err) {
|
||||
os.MkdirAll(option.DefaultLevelDbDir, 0755)
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"github.com/hanzoai/s3/s3/storage/idx"
|
||||
"github.com/hanzoai/s3/s3/storage/needle_map"
|
||||
. "github.com/hanzoai/s3/s3/storage/types"
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
)
|
||||
|
||||
type NeedleMapKind int
|
||||
@@ -48,7 +47,7 @@ type baseNeedleMapper struct {
|
||||
type TempNeedleMapper interface {
|
||||
NeedleMapper
|
||||
DoOffsetLoading(v *Volume, indexFile *os.File, startFrom uint64) error
|
||||
UpdateNeedleMap(v *Volume, indexFile *os.File, opts *opt.Options, ldbTimeout int64) error
|
||||
UpdateNeedleMap(v *Volume, indexFile *os.File, ldbTimeout int64) error
|
||||
}
|
||||
|
||||
func (nm *baseNeedleMapper) IndexFileSize() uint64 {
|
||||
|
||||
@@ -3,44 +3,61 @@ package storage
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/syndtr/goleveldb/leveldb/errors"
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
"github.com/luxfi/database"
|
||||
"github.com/luxfi/database/factory"
|
||||
log "github.com/luxfi/log"
|
||||
|
||||
"github.com/hanzoai/s3/s3/storage/idx"
|
||||
"github.com/hanzoai/s3/s3/storage/needle"
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
|
||||
"github.com/syndtr/goleveldb/leveldb"
|
||||
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/storage/needle_map"
|
||||
. "github.com/hanzoai/s3/s3/storage/types"
|
||||
)
|
||||
|
||||
// needleMapBackend is the lux-canonical key-value backend for the on-disk
|
||||
// volume needle-map index. ZAP-native zapdb. The needle-id -> (offset,size)
|
||||
// byte layout and the watermark/recovery semantics are byte-identical to the
|
||||
// historical leveldb index, so existing .ldb directories are unaffected.
|
||||
const needleMapBackend = "zapdb"
|
||||
|
||||
// mark it every watermarkBatchSize operations
|
||||
const watermarkBatchSize = 10000
|
||||
|
||||
var watermarkKey = []byte("idx_entry_watermark")
|
||||
|
||||
// openNeedleMapDB opens (creating if absent) a zapdb-backed
|
||||
// database.Database at dir. A nil gatherer/logger keeps the store metric-free
|
||||
// and silent, matching the embedded single-process volume-server use.
|
||||
// BadgerDB (zapdb) recovers automatically on open, subsuming leveldb's
|
||||
// explicit RecoverFile path.
|
||||
func openNeedleMapDB(dir string) (database.Database, error) {
|
||||
// factory.New(name, dbPath, readOnly, config, gatherer, logger, metricsPrefix, meterDBRegName).
|
||||
return factory.New(needleMapBackend, dir, false, nil, nil, log.Noop(), "", "")
|
||||
}
|
||||
|
||||
type LevelDbNeedleMap struct {
|
||||
baseNeedleMapper
|
||||
dbFileName string
|
||||
db *leveldb.DB
|
||||
ldbOpts *opt.Options
|
||||
db database.Database
|
||||
ldbAccessLock sync.RWMutex
|
||||
exitChan chan bool
|
||||
// no need to use atomic
|
||||
accessFlag int64
|
||||
ldbTimeout int64
|
||||
recordCount uint64
|
||||
accessFlag int64
|
||||
ldbTimeout int64
|
||||
// recordCount is bumped on every Put/Delete, which run under a shared
|
||||
// RLock (lazy-load path) or no lock at all (resident path), so it must be
|
||||
// atomic. Add returns the post-increment value, giving each writer a
|
||||
// consistent count for its watermark math.
|
||||
recordCount atomic.Uint64
|
||||
}
|
||||
|
||||
func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Options, ldbTimeout int64, version needle.Version) (m *LevelDbNeedleMap, err error) {
|
||||
func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, ldbTimeout int64, version needle.Version) (m *LevelDbNeedleMap, err error) {
|
||||
m = &LevelDbNeedleMap{dbFileName: dbFileName}
|
||||
m.indexFile = indexFile
|
||||
if !isLevelDbFresh(dbFileName, indexFile) {
|
||||
@@ -56,17 +73,13 @@ func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Option
|
||||
glog.V(1).Infof("Opening %s...", dbFileName)
|
||||
|
||||
if m.ldbTimeout == 0 {
|
||||
if m.db, err = leveldb.OpenFile(dbFileName, opts); err != nil {
|
||||
if errors.IsCorrupted(err) {
|
||||
m.db, err = leveldb.RecoverFile(dbFileName, opts)
|
||||
}
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if m.db, err = openNeedleMapDB(dbFileName); err != nil {
|
||||
return
|
||||
}
|
||||
glog.V(1).Infof("Loading %s... , watermark: %d", dbFileName, getWatermark(m.db))
|
||||
m.recordCount = uint64(m.indexFileOffset / NeedleMapEntrySize)
|
||||
watermark := (m.recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
recordCount := uint64(m.indexFileOffset / NeedleMapEntrySize)
|
||||
m.recordCount.Store(recordCount)
|
||||
watermark := (recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
err = setWatermark(m.db, watermark)
|
||||
if err != nil {
|
||||
glog.Fatalf("set watermark for %s error: %s\n", dbFileName, err)
|
||||
@@ -80,7 +93,6 @@ func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Option
|
||||
m.mapMetric = *mm
|
||||
m.ldbTimeout = ldbTimeout
|
||||
if m.ldbTimeout > 0 {
|
||||
m.ldbOpts = opts
|
||||
m.exitChan = make(chan bool, 1)
|
||||
m.accessFlag = 0
|
||||
go lazyLoadingRoutine(m)
|
||||
@@ -88,25 +100,51 @@ func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Option
|
||||
return
|
||||
}
|
||||
|
||||
// isLevelDbFresh reports whether the on-disk index db is newer than the .idx
|
||||
// it must mirror, so regeneration can be skipped. We always write the .idx
|
||||
// first, so a db whose newest file post-dates the .idx is consistent.
|
||||
//
|
||||
// Returning false (rebuild) is always safe — it just replays the .idx — while
|
||||
// a wrong true would trust a stale index. So any uncertainty (missing dir,
|
||||
// empty dir, stat error) returns false and forces a rebuild from the .idx.
|
||||
func isLevelDbFresh(dbFileName string, indexFile *os.File) bool {
|
||||
// normally we always write to index file first
|
||||
dbLogFile, err := os.Open(filepath.Join(dbFileName, "LOG"))
|
||||
if err != nil {
|
||||
dbModTime, ok := newestModTime(dbFileName)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
defer dbLogFile.Close()
|
||||
dbStat, dbStatErr := dbLogFile.Stat()
|
||||
indexStat, indexStatErr := indexFile.Stat()
|
||||
if dbStatErr != nil || indexStatErr != nil {
|
||||
glog.V(0).Infof("Can not stat file: %v and %v", dbStatErr, indexStatErr)
|
||||
if indexStatErr != nil {
|
||||
glog.V(0).Infof("Can not stat file: %v", indexStatErr)
|
||||
return false
|
||||
}
|
||||
return dbModTime.After(indexStat.ModTime())
|
||||
}
|
||||
|
||||
return dbStat.ModTime().After(indexStat.ModTime())
|
||||
// newestModTime returns the most recent modification time among the regular
|
||||
// files inside the db directory, and whether the directory holds any. An
|
||||
// empty or absent directory yields (zero, false).
|
||||
func newestModTime(dir string) (time.Time, bool) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return time.Time{}, false
|
||||
}
|
||||
var newest time.Time
|
||||
var found bool
|
||||
for _, e := range entries {
|
||||
info, err := e.Info()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if info.ModTime().After(newest) {
|
||||
newest = info.ModTime()
|
||||
}
|
||||
}
|
||||
return newest, found
|
||||
}
|
||||
|
||||
func generateLevelDbFile(dbFileName string, indexFile *os.File) error {
|
||||
db, err := leveldb.OpenFile(dbFileName, nil)
|
||||
db, err := openNeedleMapDB(dbFileName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -153,7 +191,7 @@ func (m *LevelDbNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue, o
|
||||
func (m *LevelDbNeedleMap) getFromDb(key NeedleId) (element *needle_map.NeedleValue, ok bool) {
|
||||
bytes := make([]byte, NeedleIdSize)
|
||||
NeedleIdToBytes(bytes[0:NeedleIdSize], key)
|
||||
data, err := m.db.Get(bytes, nil)
|
||||
data, err := m.db.Get(bytes)
|
||||
if err != nil || len(data) != OffsetSize+SizeSize {
|
||||
return nil, false
|
||||
}
|
||||
@@ -179,18 +217,18 @@ func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
|
||||
if err := m.appendToIndexFile(key, offset, size); err != nil {
|
||||
return fmt.Errorf("cannot write to indexfile %s: %v", m.indexFile.Name(), err)
|
||||
}
|
||||
m.recordCount++
|
||||
if m.recordCount%watermarkBatchSize != 0 {
|
||||
recordCount := m.recordCount.Add(1)
|
||||
if recordCount%watermarkBatchSize != 0 {
|
||||
watermark = 0
|
||||
} else {
|
||||
watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
glog.V(1).Infof("put cnt:%d for %s,watermark: %d", m.recordCount, m.dbFileName, watermark)
|
||||
watermark = (recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
glog.V(1).Infof("put cnt:%d for %s,watermark: %d", recordCount, m.dbFileName, watermark)
|
||||
}
|
||||
return levelDbWrite(m.db, key, offset, size, watermark == 0, watermark)
|
||||
}
|
||||
|
||||
func getWatermark(db *leveldb.DB) uint64 {
|
||||
data, err := db.Get(watermarkKey, nil)
|
||||
func getWatermark(db database.Database) uint64 {
|
||||
data, err := db.Get(watermarkKey)
|
||||
if err != nil || len(data) != 8 {
|
||||
glog.V(1).Infof("read previous watermark from db: %v, %d", err, len(data))
|
||||
return 0
|
||||
@@ -198,21 +236,21 @@ func getWatermark(db *leveldb.DB) uint64 {
|
||||
return util.BytesToUint64(data)
|
||||
}
|
||||
|
||||
func setWatermark(db *leveldb.DB, watermark uint64) error {
|
||||
func setWatermark(db database.Database, watermark uint64) error {
|
||||
glog.V(3).Infof("set watermark %d", watermark)
|
||||
var wmBytes = make([]byte, 8)
|
||||
util.Uint64toBytes(wmBytes, watermark)
|
||||
if err := db.Put(watermarkKey, wmBytes, nil); err != nil {
|
||||
if err := db.Put(watermarkKey, wmBytes); err != nil {
|
||||
return fmt.Errorf("failed to setWatermark: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size Size, updateWatermark bool, watermark uint64) error {
|
||||
func levelDbWrite(db database.Database, key NeedleId, offset Offset, size Size, updateWatermark bool, watermark uint64) error {
|
||||
|
||||
bytes := needle_map.ToBytes(key, offset, size)
|
||||
|
||||
if err := db.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize], nil); err != nil {
|
||||
if err := db.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize]); err != nil {
|
||||
return fmt.Errorf("failed to write leveldb: %w", err)
|
||||
}
|
||||
// set watermark
|
||||
@@ -222,10 +260,10 @@ func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size Size, update
|
||||
return nil
|
||||
}
|
||||
|
||||
func levelDbDelete(db *leveldb.DB, key NeedleId) error {
|
||||
func levelDbDelete(db database.Database, key NeedleId) error {
|
||||
bytes := make([]byte, NeedleIdSize)
|
||||
NeedleIdToBytes(bytes, key)
|
||||
return db.Delete(bytes, nil)
|
||||
return db.Delete(bytes)
|
||||
}
|
||||
|
||||
func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
|
||||
@@ -246,11 +284,11 @@ func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
|
||||
if err := m.appendToIndexFile(key, offset, TombstoneFileSize); err != nil {
|
||||
return err
|
||||
}
|
||||
m.recordCount++
|
||||
if m.recordCount%watermarkBatchSize != 0 {
|
||||
recordCount := m.recordCount.Add(1)
|
||||
if recordCount%watermarkBatchSize != 0 {
|
||||
watermark = 0
|
||||
} else {
|
||||
watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
watermark = (recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
}
|
||||
return levelDbWrite(m.db, key, oldNeedle.Offset, -oldNeedle.Size, watermark == 0, watermark)
|
||||
}
|
||||
@@ -282,7 +320,7 @@ func (m *LevelDbNeedleMap) Destroy() error {
|
||||
return os.RemoveAll(m.dbFileName)
|
||||
}
|
||||
|
||||
func (m *LevelDbNeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *opt.Options, ldbTimeout int64) error {
|
||||
func (m *LevelDbNeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, ldbTimeout int64) error {
|
||||
if v.nm != nil {
|
||||
v.nm.Close()
|
||||
v.nm = nil
|
||||
@@ -303,14 +341,9 @@ func (m *LevelDbNeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *
|
||||
return fmt.Errorf("rename %s: %v", levelDbFile, err)
|
||||
}
|
||||
|
||||
db, err := leveldb.OpenFile(levelDbFile, opts)
|
||||
db, err := openNeedleMapDB(levelDbFile)
|
||||
if err != nil {
|
||||
if errors.IsCorrupted(err) {
|
||||
db, err = leveldb.RecoverFile(levelDbFile, opts)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
m.db = db
|
||||
|
||||
@@ -320,10 +353,11 @@ func (m *LevelDbNeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *
|
||||
return e
|
||||
}
|
||||
m.indexFileOffset = stat.Size()
|
||||
m.recordCount = uint64(stat.Size() / NeedleMapEntrySize)
|
||||
recordCount := uint64(stat.Size() / NeedleMapEntrySize)
|
||||
m.recordCount.Store(recordCount)
|
||||
|
||||
//set watermark
|
||||
watermark := (m.recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
watermark := (recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
err = setWatermark(db, uint64(watermark))
|
||||
if err != nil {
|
||||
glog.Fatalf("setting watermark failed %s: %v", indexFile.Name(), err)
|
||||
@@ -333,7 +367,6 @@ func (m *LevelDbNeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *
|
||||
v.tmpNm = nil
|
||||
m.ldbTimeout = ldbTimeout
|
||||
if m.ldbTimeout > 0 {
|
||||
m.ldbOpts = opts
|
||||
m.exitChan = make(chan bool, 1)
|
||||
m.accessFlag = 0
|
||||
go lazyLoadingRoutine(m)
|
||||
@@ -348,7 +381,7 @@ func (m *LevelDbNeedleMap) DoOffsetLoading(v *Volume, indexFile *os.File, startF
|
||||
version = v.Version()
|
||||
}
|
||||
dbFileName := v.FileName(".cpldb")
|
||||
db, dbErr := leveldb.OpenFile(dbFileName, nil)
|
||||
db, dbErr := openNeedleMapDB(dbFileName)
|
||||
defer func() {
|
||||
if dbErr == nil {
|
||||
db.Close()
|
||||
@@ -359,12 +392,7 @@ func (m *LevelDbNeedleMap) DoOffsetLoading(v *Volume, indexFile *os.File, startF
|
||||
|
||||
}()
|
||||
if dbErr != nil {
|
||||
if errors.IsCorrupted(err) {
|
||||
db, dbErr = leveldb.RecoverFile(dbFileName, nil)
|
||||
}
|
||||
if dbErr != nil {
|
||||
return dbErr
|
||||
}
|
||||
return dbErr
|
||||
}
|
||||
|
||||
err = idx.WalkIndexFile(indexFile, startFrom, func(key NeedleId, offset Offset, size Size) (e error) {
|
||||
@@ -379,9 +407,9 @@ func (m *LevelDbNeedleMap) DoOffsetLoading(v *Volume, indexFile *os.File, startF
|
||||
return e
|
||||
}
|
||||
// increment loading
|
||||
data, err := db.Get(bytes, nil)
|
||||
data, err := db.Get(bytes)
|
||||
if err != nil {
|
||||
if !strings.Contains(strings.ToLower(err.Error()), "not found") {
|
||||
if err != database.ErrNotFound {
|
||||
// unexpected error
|
||||
return err
|
||||
}
|
||||
@@ -436,14 +464,9 @@ func reloadLdb(m *LevelDbNeedleMap) (err error) {
|
||||
}
|
||||
glog.V(1).Infof("reloading leveldb %s", m.dbFileName)
|
||||
m.accessFlag = 1
|
||||
if m.db, err = leveldb.OpenFile(m.dbFileName, m.ldbOpts); err != nil {
|
||||
if errors.IsCorrupted(err) {
|
||||
m.db, err = leveldb.RecoverFile(m.dbFileName, m.ldbOpts)
|
||||
}
|
||||
if err != nil {
|
||||
glog.Fatalf("RecoverFile %s failed:%v", m.dbFileName, err)
|
||||
return err
|
||||
}
|
||||
if m.db, err = openNeedleMapDB(m.dbFileName); err != nil {
|
||||
glog.Fatalf("open %s failed:%v", m.dbFileName, err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -7,8 +7,6 @@ import (
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/syndtr/goleveldb/leveldb"
|
||||
|
||||
"github.com/hanzoai/s3/s3/storage/needle"
|
||||
"github.com/hanzoai/s3/s3/storage/needle_map"
|
||||
"github.com/hanzoai/s3/s3/storage/types"
|
||||
@@ -44,7 +42,7 @@ func TestGenerateLevelDbFileStaleWatermarkRebuilds(t *testing.T) {
|
||||
// Seed an .ldb whose stored watermark sits far past the short .idx, mimicking
|
||||
// the leftover db from a pre-crash, much larger index.
|
||||
dbPath := filepath.Join(dir, "1.ldb")
|
||||
db, err := leveldb.OpenFile(dbPath, nil)
|
||||
db, err := openNeedleMapDB(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("open ldb: %v", err)
|
||||
}
|
||||
@@ -57,7 +55,7 @@ func TestGenerateLevelDbFileStaleWatermarkRebuilds(t *testing.T) {
|
||||
t.Fatalf("generateLevelDbFile: %v", err)
|
||||
}
|
||||
|
||||
db, err = leveldb.OpenFile(dbPath, nil)
|
||||
db, err = openNeedleMapDB(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen ldb: %v", err)
|
||||
}
|
||||
@@ -65,12 +63,130 @@ func TestGenerateLevelDbFileStaleWatermarkRebuilds(t *testing.T) {
|
||||
for i := uint64(1); i <= liveCount; i++ {
|
||||
keyBytes := make([]byte, types.NeedleIdSize)
|
||||
types.NeedleIdToBytes(keyBytes, types.Uint64ToNeedleId(i))
|
||||
if _, err := db.Get(keyBytes, nil); err != nil {
|
||||
if _, err := db.Get(keyBytes); err != nil {
|
||||
t.Fatalf("needle %d missing after rebuild (stale watermark poisoned the map): %v", i, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Round-trip a range of needle ids through the on-disk zapdb needle map:
|
||||
// put, read back the exact offset/size, delete, and confirm the delete sticks.
|
||||
// This exercises the needle-id -> (offset,size) byte layout end to end and the
|
||||
// rebuild-from-empty path (the .ldb dir starts absent).
|
||||
func TestLevelDbNeedleMapRoundTrip(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
indexFile, err := os.OpenFile(filepath.Join(dir, "rt.idx"), os.O_RDWR|os.O_CREATE, 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("create index file: %v", err)
|
||||
}
|
||||
dbFileName := filepath.Join(dir, "rt.ldb")
|
||||
|
||||
// ldbTimeout 0 keeps the db resident (no lazy unload), the volume-server
|
||||
// default for a single hot volume.
|
||||
m, err := NewLevelDbNeedleMap(dbFileName, indexFile, 0, needle.GetCurrentVersion())
|
||||
if err != nil {
|
||||
t.Fatalf("NewLevelDbNeedleMap: %v", err)
|
||||
}
|
||||
defer m.Close()
|
||||
|
||||
const n = 256
|
||||
for i := uint64(1); i <= n; i++ {
|
||||
key := types.Uint64ToNeedleId(i)
|
||||
off := types.ToOffset(int64(i) * types.NeedlePaddingSize)
|
||||
size := types.Size(i * 7)
|
||||
if err := m.Put(key, off, size); err != nil {
|
||||
t.Fatalf("Put %d: %v", i, err)
|
||||
}
|
||||
nv, ok := m.Get(key)
|
||||
if !ok {
|
||||
t.Fatalf("Get %d after Put: not found", i)
|
||||
}
|
||||
if nv.Offset != off || nv.Size != size {
|
||||
t.Fatalf("Get %d mismatch: got (off=%v,size=%v) want (off=%v,size=%v)", i, nv.Offset, nv.Size, off, size)
|
||||
}
|
||||
}
|
||||
|
||||
// Every key still resolves.
|
||||
for i := uint64(1); i <= n; i++ {
|
||||
if _, ok := m.Get(types.Uint64ToNeedleId(i)); !ok {
|
||||
t.Fatalf("Get %d before delete: not found", i)
|
||||
}
|
||||
}
|
||||
|
||||
// Delete the even ids; a deleted needle reads back as not-found (tombstoned
|
||||
// negative size).
|
||||
for i := uint64(2); i <= n; i += 2 {
|
||||
key := types.Uint64ToNeedleId(i)
|
||||
nv, _ := m.Get(key)
|
||||
if err := m.Delete(key, nv.Offset); err != nil {
|
||||
t.Fatalf("Delete %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
for i := uint64(1); i <= n; i++ {
|
||||
_, ok := m.Get(types.Uint64ToNeedleId(i))
|
||||
if i%2 == 0 {
|
||||
if ok {
|
||||
// getFromDb returns the value; a deleted needle stores a negative
|
||||
// (tombstone) size which still round-trips, so confirm via size.
|
||||
if nv, _ := m.Get(types.Uint64ToNeedleId(i)); !nv.Size.IsDeleted() {
|
||||
t.Fatalf("needle %d should be deleted, size=%v", i, nv.Size)
|
||||
}
|
||||
}
|
||||
} else if !ok {
|
||||
t.Fatalf("odd needle %d should survive delete of evens", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A volume server batches needle writes and crosses the watermark batch
|
||||
// boundary; the on-disk index must survive a Close/reopen with every needle
|
||||
// intact and the watermark persisted. This drives the batched-write path
|
||||
// (recordCount crossing watermarkBatchSize) and the rebuild-from-existing
|
||||
// open path.
|
||||
func TestLevelDbNeedleMapPersistsAcrossReopen(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
idxPath := filepath.Join(dir, "p.idx")
|
||||
dbPath := filepath.Join(dir, "p.ldb")
|
||||
|
||||
indexFile, err := os.OpenFile(idxPath, os.O_RDWR|os.O_CREATE, 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("create index file: %v", err)
|
||||
}
|
||||
m, err := NewLevelDbNeedleMap(dbPath, indexFile, 0, needle.GetCurrentVersion())
|
||||
if err != nil {
|
||||
t.Fatalf("NewLevelDbNeedleMap: %v", err)
|
||||
}
|
||||
|
||||
const n = 32
|
||||
for i := uint64(1); i <= n; i++ {
|
||||
if err := m.Put(types.Uint64ToNeedleId(i), types.ToOffset(int64(i)*types.NeedlePaddingSize), types.Size(i)); err != nil {
|
||||
t.Fatalf("Put %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
m.Close() // syncs idx, closes db
|
||||
|
||||
// Reopen: the .ldb is newer than the .idx, so isLevelDbFresh skips the
|
||||
// rebuild and we open the existing index in place.
|
||||
indexFile2, err := os.OpenFile(idxPath, os.O_RDWR|os.O_CREATE, 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen index file: %v", err)
|
||||
}
|
||||
m2, err := NewLevelDbNeedleMap(dbPath, indexFile2, 0, needle.GetCurrentVersion())
|
||||
if err != nil {
|
||||
t.Fatalf("reopen NewLevelDbNeedleMap: %v", err)
|
||||
}
|
||||
defer m2.Close()
|
||||
for i := uint64(1); i <= n; i++ {
|
||||
nv, ok := m2.Get(types.Uint64ToNeedleId(i))
|
||||
if !ok {
|
||||
t.Fatalf("needle %d missing after reopen", i)
|
||||
}
|
||||
if want := types.Size(i); nv.Size != want {
|
||||
t.Fatalf("needle %d size after reopen: got %v want %v", i, nv.Size, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLevelDbNeedleMap_Concurrency(t *testing.T) {
|
||||
dir, err := os.MkdirTemp("", "test_leveldb_concurrency")
|
||||
if err != nil {
|
||||
@@ -86,7 +202,7 @@ func TestLevelDbNeedleMap_Concurrency(t *testing.T) {
|
||||
dbFileName := filepath.Join(dir, prefix+".ldb")
|
||||
|
||||
// Create and initialize map
|
||||
m, err := NewLevelDbNeedleMap(dbFileName, indexFile, nil, 1, needle.GetCurrentVersion())
|
||||
m, err := NewLevelDbNeedleMap(dbFileName, indexFile, 1, needle.GetCurrentVersion())
|
||||
if err != nil {
|
||||
t.Fatalf("NewLevelDbNeedleMap: %v", err)
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"github.com/hanzoai/s3/s3/storage/needle"
|
||||
"github.com/hanzoai/s3/s3/storage/needle_map"
|
||||
. "github.com/hanzoai/s3/s3/storage/types"
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
)
|
||||
|
||||
type NeedleMap struct {
|
||||
@@ -86,7 +85,7 @@ func (nm *NeedleMap) Destroy() error {
|
||||
return os.Remove(nm.indexFile.Name())
|
||||
}
|
||||
|
||||
func (nm *NeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, opts *opt.Options, ldbTimeout int64) error {
|
||||
func (nm *NeedleMap) UpdateNeedleMap(v *Volume, indexFile *os.File, ldbTimeout int64) error {
|
||||
if v.nm != nil {
|
||||
v.nm.Close()
|
||||
v.nm = nil
|
||||
|
||||
@@ -6,8 +6,6 @@ import (
|
||||
|
||||
"github.com/hanzoai/s3/s3/storage/types"
|
||||
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/stats"
|
||||
"github.com/hanzoai/s3/s3/storage/backend"
|
||||
@@ -16,23 +14,6 @@ import (
|
||||
"github.com/hanzoai/s3/s3/util"
|
||||
)
|
||||
|
||||
// Per-DB caps on goleveldb's open SST file cache. The library default is 500
|
||||
// per DB, but a volume server hosts one DB per volume — easily thousands —
|
||||
// so the per-DB default sums into FD exhaustion (`open .../00000N.log: too
|
||||
// many open files`) even with generous ulimits, especially when leveldb is
|
||||
// rotating its WAL.
|
||||
//
|
||||
// The trade-off: a larger cache lowers re-open overhead on cold reads, a
|
||||
// smaller cache bounds total FD usage. CompactionTableSizeMultiplier=10
|
||||
// already keeps SST counts low (~10x larger SSTs => ~10x fewer files), so
|
||||
// even the small-volume cap is enough to keep the working set hot while
|
||||
// leaving headroom for thousands of co-resident DBs.
|
||||
const (
|
||||
LevelDbOpenFilesCacheCapacity = 16
|
||||
LevelDbMediumOpenFilesCacheCapacity = 32
|
||||
LevelDbLargeOpenFilesCacheCapacity = 64
|
||||
)
|
||||
|
||||
func loadVolumeWithoutIndex(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, ver needle.Version) (v *Volume, err error) {
|
||||
v = &Volume{dir: dirname, Collection: collection, Id: id}
|
||||
v.SuperBlock = super_block.SuperBlock{}
|
||||
@@ -87,39 +68,14 @@ func (v *Volume) reopenIdxForWrite() error {
|
||||
indexFile.Close()
|
||||
return fmt.Errorf("rebuild memory needle map for volume %d: %v", v.Id, err)
|
||||
}
|
||||
case NeedleMapLevelDb:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 2 * 1024 * 1024,
|
||||
WriteBuffer: 1 * 1024 * 1024,
|
||||
CompactionTableSizeMultiplier: 10,
|
||||
OpenFilesCacheCapacity: LevelDbOpenFilesCacheCapacity,
|
||||
}
|
||||
if newNm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
case NeedleMapLevelDb, NeedleMapLevelDbMedium, NeedleMapLevelDbLarge:
|
||||
// One zapdb-backed needle map serves all three sizes; the kind only
|
||||
// selects the .ldb backend. zapdb manages its own caches, so the
|
||||
// historical leveldb block-cache/write-buffer tuning is gone.
|
||||
if newNm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, v.ldbTimeout, v.Version()); err != nil {
|
||||
indexFile.Close()
|
||||
return fmt.Errorf("rebuild leveldb needle map for volume %d: %v", v.Id, err)
|
||||
}
|
||||
case NeedleMapLevelDbMedium:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 4 * 1024 * 1024,
|
||||
WriteBuffer: 2 * 1024 * 1024,
|
||||
CompactionTableSizeMultiplier: 10,
|
||||
OpenFilesCacheCapacity: LevelDbMediumOpenFilesCacheCapacity,
|
||||
}
|
||||
if newNm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
indexFile.Close()
|
||||
return fmt.Errorf("rebuild leveldb medium needle map for volume %d: %v", v.Id, err)
|
||||
}
|
||||
case NeedleMapLevelDbLarge:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 8 * 1024 * 1024,
|
||||
WriteBuffer: 4 * 1024 * 1024,
|
||||
CompactionTableSizeMultiplier: 10,
|
||||
OpenFilesCacheCapacity: LevelDbLargeOpenFilesCacheCapacity,
|
||||
}
|
||||
if newNm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
indexFile.Close()
|
||||
return fmt.Errorf("rebuild leveldb large needle map for volume %d: %v", v.Id, err)
|
||||
}
|
||||
default:
|
||||
indexFile.Close()
|
||||
return fmt.Errorf("unsupported needle map kind %v for volume %d", v.needleMapKind, v.Id)
|
||||
@@ -293,7 +249,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
|
||||
case NeedleMapInMemory:
|
||||
if v.tmpNm != nil {
|
||||
glog.V(2).Infof("updating memory compact index %s ", v.FileName(".idx"))
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, nil, 0)
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, 0)
|
||||
} else {
|
||||
glog.V(2).Infoln("loading memory index", v.FileName(".idx"), "to memory")
|
||||
if v.nm, err = LoadCompactNeedleMap(indexFile, v.Version()); err != nil {
|
||||
@@ -302,55 +258,17 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
|
||||
indexFile.Close()
|
||||
}
|
||||
}
|
||||
case NeedleMapLevelDb:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 2 * 1024 * 1024, // default value is 8MiB
|
||||
WriteBuffer: 1 * 1024 * 1024, // default value is 4MiB
|
||||
CompactionTableSizeMultiplier: 10, // default value is 1
|
||||
OpenFilesCacheCapacity: LevelDbOpenFilesCacheCapacity, // see package-level docs
|
||||
}
|
||||
case NeedleMapLevelDb, NeedleMapLevelDbMedium, NeedleMapLevelDbLarge:
|
||||
// All three sizes share one zapdb-backed needle map. The
|
||||
// historical small/medium/large split tuned goleveldb's block
|
||||
// cache, write buffer, and open-file cache; zapdb manages those
|
||||
// internally, so the kind only selects the .ldb backend here.
|
||||
if v.tmpNm != nil {
|
||||
glog.V(0).Infoln("updating leveldb index", v.FileName(".ldb"))
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts, v.ldbTimeout)
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, v.ldbTimeout)
|
||||
} else {
|
||||
glog.V(0).Infoln("loading leveldb index", v.FileName(".ldb"))
|
||||
if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
|
||||
v.nm = nil
|
||||
indexFile.Close()
|
||||
}
|
||||
}
|
||||
case NeedleMapLevelDbMedium:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 4 * 1024 * 1024, // default value is 8MiB
|
||||
WriteBuffer: 2 * 1024 * 1024, // default value is 4MiB
|
||||
CompactionTableSizeMultiplier: 10, // default value is 1
|
||||
OpenFilesCacheCapacity: LevelDbMediumOpenFilesCacheCapacity, // see package-level docs
|
||||
}
|
||||
if v.tmpNm != nil {
|
||||
glog.V(0).Infoln("updating leveldb medium index", v.FileName(".ldb"))
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts, v.ldbTimeout)
|
||||
} else {
|
||||
glog.V(0).Infoln("loading leveldb medium index", v.FileName(".ldb"))
|
||||
if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
|
||||
v.nm = nil
|
||||
indexFile.Close()
|
||||
}
|
||||
}
|
||||
case NeedleMapLevelDbLarge:
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 8 * 1024 * 1024, // default value is 8MiB
|
||||
WriteBuffer: 4 * 1024 * 1024, // default value is 4MiB
|
||||
CompactionTableSizeMultiplier: 10, // default value is 1
|
||||
OpenFilesCacheCapacity: LevelDbLargeOpenFilesCacheCapacity, // see package-level docs
|
||||
}
|
||||
if v.tmpNm != nil {
|
||||
glog.V(0).Infoln("updating leveldb large index", v.FileName(".ldb"))
|
||||
err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts, v.ldbTimeout)
|
||||
} else {
|
||||
glog.V(0).Infoln("loading leveldb large index", v.FileName(".ldb"))
|
||||
if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
|
||||
if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, v.ldbTimeout, v.Version()); err != nil {
|
||||
glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
|
||||
v.nm = nil
|
||||
indexFile.Close()
|
||||
|
||||
@@ -106,8 +106,9 @@ func testCompactionByIndex(t *testing.T, needleMapKind NeedleMapKind) {
|
||||
nm := reflect.ValueOf(v.nm).Interface().(*LevelDbNeedleMap)
|
||||
mm := nm.mapMetric
|
||||
watermark := getWatermark(nm.db)
|
||||
realWatermark := (nm.recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
t.Logf("watermark from levelDB: %d, realWatermark: %d, nm.recordCount: %d, realRecordCount:%d, fileCount=%d, deletedcount:%d", watermark, realWatermark, nm.recordCount, realRecordCount, mm.FileCount(), v.DeletedCount())
|
||||
recordCount := nm.recordCount.Load()
|
||||
realWatermark := (recordCount / watermarkBatchSize) * watermarkBatchSize
|
||||
t.Logf("watermark from levelDB: %d, realWatermark: %d, nm.recordCount: %d, realRecordCount:%d, fileCount=%d, deletedcount:%d", watermark, realWatermark, recordCount, realRecordCount, mm.FileCount(), v.DeletedCount())
|
||||
if realWatermark != watermark {
|
||||
t.Fatalf("testing watermark failed")
|
||||
}
|
||||
|
||||
@@ -5,8 +5,6 @@ import (
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
|
||||
"github.com/hanzoai/s3/s3/glog"
|
||||
"github.com/hanzoai/s3/s3/storage"
|
||||
"github.com/hanzoai/s3/s3/storage/backend"
|
||||
@@ -65,12 +63,7 @@ func LoadOrCreateChunkCacheVolume(fileName string, preallocate int64) (*ChunkCac
|
||||
}
|
||||
|
||||
glog.V(1).Infoln("loading leveldb", v.fileName+".ldb")
|
||||
opts := &opt.Options{
|
||||
BlockCacheCapacity: 2 * 1024 * 1024, // default value is 8MiB
|
||||
WriteBuffer: 1 * 1024 * 1024, // default value is 4MiB
|
||||
CompactionTableSizeMultiplier: 10, // default value is 1
|
||||
}
|
||||
if v.nm, err = storage.NewLevelDbNeedleMap(v.fileName+".ldb", indexFile, opts, 0, needle.GetCurrentVersion()); err != nil {
|
||||
if v.nm, err = storage.NewLevelDbNeedleMap(v.fileName+".ldb", indexFile, 0, needle.GetCurrentVersion()); err != nil {
|
||||
return nil, fmt.Errorf("loading leveldb %s error: %v", v.fileName+".ldb", err)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user