@cryptotaxi247 / kubo
Code Commits Issues Pulls Sessions CI/CD
master
master
go 195 lines 5.36 KB
Raw
1 package pebbleds
2
3 import (
4 "fmt"
5 "path/filepath"
6 "time"
7
8 "github.com/cockroachdb/pebble/v2"
9 pebbleds "github.com/ipfs/go-ds-pebble"
10 "github.com/ipfs/kubo/misc/fsutil"
11 "github.com/ipfs/kubo/plugin"
12 "github.com/ipfs/kubo/repo"
13 "github.com/ipfs/kubo/repo/fsrepo"
14 )
15
16 // Plugins is exported list of plugins that will be loaded.
17 var Plugins = []plugin.Plugin{
18 &pebbledsPlugin{},
19 }
20
21 type pebbledsPlugin struct{}
22
23 var _ plugin.PluginDatastore = (*pebbledsPlugin)(nil)
24
25 func (*pebbledsPlugin) Name() string {
26 return "ds-pebble"
27 }
28
29 func (*pebbledsPlugin) Version() string {
30 return "0.1.0"
31 }
32
33 func (*pebbledsPlugin) Init(_ *plugin.Environment) error {
34 return nil
35 }
36
37 func (*pebbledsPlugin) DatastoreTypeName() string {
38 return "pebbleds"
39 }
40
41 type datastoreConfig struct {
42 path string
43 cacheSize int64
44
45 // Documentation of these values: https://pkg.go.dev/github.com/cockroachdb/pebble@v1.1.2#Options
46 pebbleOpts *pebble.Options
47 }
48
49 // PebbleDatastoreConfig returns a configuration stub for a pebble datastore
50 // from the given parameters.
51 func (*pebbledsPlugin) DatastoreConfigParser() fsrepo.ConfigFromMap {
52 return func(params map[string]any) (fsrepo.DatastoreConfig, error) {
53 var c datastoreConfig
54 var ok bool
55
56 c.path, ok = params["path"].(string)
57 if !ok {
58 return nil, fmt.Errorf("'path' field is missing or not string")
59 }
60
61 cacheSize, err := getConfigInt("cacheSize", params)
62 if err != nil {
63 return nil, err
64 }
65 c.cacheSize = int64(cacheSize)
66
67 bytesPerSync, err := getConfigInt("bytesPerSync", params)
68 if err != nil {
69 return nil, err
70 }
71 disableWAL, err := getConfigBool("disableWAL", params)
72 if err != nil {
73 return nil, err
74 }
75 fmv, err := getConfigInt("formatMajorVersion", params)
76 if err != nil {
77 return nil, err
78 }
79 formatMajorVersion := pebble.FormatMajorVersion(fmv)
80 l0CompactionThreshold, err := getConfigInt("l0CompactionThreshold", params)
81 if err != nil {
82 return nil, err
83 }
84 l0StopWritesThreshold, err := getConfigInt("l0StopWritesThreshold", params)
85 if err != nil {
86 return nil, err
87 }
88 lBaseMaxBytes, err := getConfigInt("lBaseMaxBytes", params)
89 if err != nil {
90 return nil, err
91 }
92 maxConcurrentCompactions, err := getConfigInt("maxConcurrentCompactions", params)
93 if err != nil {
94 return nil, err
95 }
96 memTableSize, err := getConfigInt("memTableSize", params)
97 if err != nil {
98 return nil, err
99 }
100 memTableStopWritesThreshold, err := getConfigInt("memTableStopWritesThreshold", params)
101 if err != nil {
102 return nil, err
103 }
104 walBytesPerSync, err := getConfigInt("walBytesPerSync", params)
105 if err != nil {
106 return nil, err
107 }
108 walMinSyncSec, err := getConfigInt("walMinSyncIntervalSeconds", params)
109 if err != nil {
110 return nil, err
111 }
112
113 if formatMajorVersion == 0 {
114 // Pebble DB format not configured. Automatically ratchet the
115 // database to the latest format. This may prevent downgrade.
116 formatMajorVersion = pebble.FormatNewest
117 } else if formatMajorVersion < pebble.FormatNewest {
118 // Pebble DB format is configured, but is not the latest.
119 fmt.Println("⚠️ A newer pebble db format is available.")
120 fmt.Println(" To upgrade, set the following in the pebble datastore config:")
121 fmt.Println(" \"formatMajorVersion\":", int(pebble.FormatNewest))
122 }
123
124 if bytesPerSync != 0 || disableWAL || formatMajorVersion != 0 || l0CompactionThreshold != 0 || l0StopWritesThreshold != 0 || lBaseMaxBytes != 0 || maxConcurrentCompactions != 0 || memTableSize != 0 || memTableStopWritesThreshold != 0 || walBytesPerSync != 0 || walMinSyncSec != 0 {
125 c.pebbleOpts = &pebble.Options{
126 BytesPerSync: bytesPerSync,
127 DisableWAL: disableWAL,
128 FormatMajorVersion: formatMajorVersion,
129 L0CompactionThreshold: l0CompactionThreshold,
130 L0StopWritesThreshold: l0StopWritesThreshold,
131 LBaseMaxBytes: int64(lBaseMaxBytes),
132 MemTableSize: uint64(memTableSize),
133 MemTableStopWritesThreshold: memTableStopWritesThreshold,
134 WALBytesPerSync: walBytesPerSync,
135 }
136 if maxConcurrentCompactions != 0 {
137 c.pebbleOpts.CompactionConcurrencyRange = func() (int, int) { return 1, maxConcurrentCompactions }
138 }
139 if walMinSyncSec != 0 {
140 c.pebbleOpts.WALMinSyncInterval = func() time.Duration { return time.Duration(walMinSyncSec) * time.Second }
141 }
142 }
143
144 return &c, nil
145 }
146 }
147
148 func getConfigBool(name string, params map[string]any) (bool, error) {
149 val, ok := params[name]
150 if ok {
151 bval, ok := val.(bool)
152 if !ok {
153 return false, fmt.Errorf("%q field was not a bool", name)
154 }
155 return bval, nil
156 }
157 return false, nil
158 }
159
160 func getConfigInt(name string, params map[string]any) (int, error) {
161 val, ok := params[name]
162 if ok {
163 // TODO: see why val may be an int or a float64.
164 ival, ok := val.(int)
165 if !ok {
166 fval, ok := val.(float64)
167 if !ok {
168 return 0, fmt.Errorf("%q field was not an integer or a float64", name)
169 }
170 return int(fval), nil
171 }
172 return ival, nil
173 }
174 return 0, nil
175 }
176
177 func (c *datastoreConfig) DiskSpec() fsrepo.DiskSpec {
178 return map[string]any{
179 "type": "pebbleds",
180 "path": c.path,
181 }
182 }
183
184 func (c *datastoreConfig) Create(path string) (repo.Datastore, error) {
185 p := c.path
186 if !filepath.IsAbs(p) {
187 p = filepath.Join(path, p)
188 }
189
190 if err := fsutil.DirWritable(p); err != nil {
191 return nil, err
192 }
193
194 return pebbleds.NewDatastore(p, pebbleds.WithCacheSize(c.cacheSize), pebbleds.WithPebbleOpts(c.pebbleOpts))
195 }