storj/storage/postgreskv/client.go
Egon Elbre 76fdb5d863 storage: add configurable lookup limits
Currently storage tests were tied to the default lookup limit.
By increasing the limits, the tests will take longer and sometimes
cause a large number of goroutines to be started.

This change adds configurable lookup limit to all storage backends.

Also remove boltdb.NewShared, since it's not used any more.

Change-Id: I1a052f149da471246fac5745da133c3cfc27582e
2020-01-22 21:35:56 +02:00

279 lines
6.8 KiB
Go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package postgreskv
import (
"context"
"database/sql"
"github.com/lib/pq"
"github.com/zeebo/errs"
"gopkg.in/spacemonkeygo/monkit.v2"
"storj.io/storj/private/dbutil"
"storj.io/storj/private/dbutil/pgutil"
"storj.io/storj/private/tagsql"
"storj.io/storj/storage"
"storj.io/storj/storage/postgreskv/schema"
)
var (
mon = monkit.Package()
)
// Client is the entrypoint into a postgreskv data store
type Client struct {
db tagsql.DB
lookupLimit int
}
// New instantiates a new postgreskv client given db URL
func New(dbURL string) (*Client, error) {
dbURL = pgutil.CheckApplicationName(dbURL)
db, err := tagsql.Open("postgres", dbURL)
if err != nil {
return nil, err
}
dbutil.Configure(db, mon)
//TODO: Fix the parameters!!
err = schema.PrepareDB(context.TODO(), db, dbURL)
if err != nil {
return nil, err
}
return NewWith(db), nil
}
// NewWith instantiates a new postgreskv client given db.
func NewWith(db tagsql.DB) *Client {
return &Client{db: db, lookupLimit: storage.DefaultLookupLimit}
}
// SetLookupLimit sets the lookup limit.
func (client *Client) SetLookupLimit(v int) { client.lookupLimit = v }
// LookupLimit returns the maximum limit that is allowed.
func (client *Client) LookupLimit() int { return client.lookupLimit }
// Close closes the client
func (client *Client) Close() error {
return client.db.Close()
}
// Put sets the value for the provided key.
func (client *Client) Put(ctx context.Context, key storage.Key, value storage.Value) (err error) {
defer mon.Task()(&ctx)(&err)
if key.IsZero() {
return storage.ErrEmptyKey.New("")
}
q := `
INSERT INTO pathdata (fullpath, metadata)
VALUES ($1::BYTEA, $2::BYTEA)
ON CONFLICT (fullpath) DO UPDATE SET metadata = EXCLUDED.metadata
`
_, err = client.db.Exec(ctx, q, []byte(key), []byte(value))
return Error.Wrap(err)
}
// Get looks up the provided key and returns its value (or an error).
func (client *Client) Get(ctx context.Context, key storage.Key) (_ storage.Value, err error) {
defer mon.Task()(&ctx)(&err)
if key.IsZero() {
return nil, storage.ErrEmptyKey.New("")
}
q := "SELECT metadata FROM pathdata WHERE fullpath = $1::BYTEA"
row := client.db.QueryRow(ctx, q, []byte(key))
var val []byte
err = row.Scan(&val)
if err == sql.ErrNoRows {
return nil, storage.ErrKeyNotFound.New("%q", key)
}
return val, Error.Wrap(err)
}
// GetAll finds all values for the provided keys (up to LookupLimit).
// If more keys are provided than the maximum, an error will be returned.
func (client *Client) GetAll(ctx context.Context, keys storage.Keys) (_ storage.Values, err error) {
defer mon.Task()(&ctx)(&err)
if len(keys) > client.lookupLimit {
return nil, storage.ErrLimitExceeded
}
q := `
SELECT metadata
FROM pathdata pd
RIGHT JOIN
unnest($1::BYTEA[]) WITH ORDINALITY pk(request, ord)
ON (pd.fullpath = pk.request)
ORDER BY pk.ord
`
rows, err := client.db.Query(ctx, q, pq.ByteaArray(keys.ByteSlices()))
if err != nil {
return nil, errs.Wrap(err)
}
defer func() { err = errs.Combine(err, Error.Wrap(rows.Close())) }()
values := make([]storage.Value, 0, len(keys))
for rows.Next() {
var value []byte
if err := rows.Scan(&value); err != nil {
return nil, Error.Wrap(err)
}
values = append(values, storage.Value(value))
}
return values, Error.Wrap(rows.Err())
}
// Delete deletes the given key and its associated value.
func (client *Client) Delete(ctx context.Context, key storage.Key) (err error) {
defer mon.Task()(&ctx)(&err)
if key.IsZero() {
return storage.ErrEmptyKey.New("")
}
q := "DELETE FROM pathdata WHERE fullpath = $1::BYTEA"
result, err := client.db.Exec(ctx, q, []byte(key))
if err != nil {
return err
}
numRows, err := result.RowsAffected()
if err != nil {
return err
}
if numRows == 0 {
return storage.ErrKeyNotFound.New("%q", key)
}
return nil
}
// List returns either a list of known keys, in order, or an error.
func (client *Client) List(ctx context.Context, first storage.Key, limit int) (_ storage.Keys, err error) {
defer mon.Task()(&ctx)(&err)
return storage.ListKeys(ctx, client, first, limit)
}
// Iterate calls the callback with an iterator over the keys.
func (client *Client) Iterate(ctx context.Context, opts storage.IterateOptions, fn func(context.Context, storage.Iterator) error) (err error) {
defer mon.Task()(&ctx)(&err)
if opts.Limit <= 0 || opts.Limit > client.lookupLimit {
opts.Limit = client.lookupLimit
}
opi, err := newOrderedPostgresIterator(ctx, client, opts)
if err != nil {
return err
}
defer func() {
err = errs.Combine(err, opi.Close())
}()
return fn(ctx, opi)
}
// CompareAndSwap atomically compares and swaps oldValue with newValue
func (client *Client) CompareAndSwap(ctx context.Context, key storage.Key, oldValue, newValue storage.Value) (err error) {
defer mon.Task()(&ctx)(&err)
if key.IsZero() {
return storage.ErrEmptyKey.New("")
}
if oldValue == nil && newValue == nil {
q := "SELECT metadata FROM pathdata WHERE fullpath = $1::BYTEA"
row := client.db.QueryRow(ctx, q, []byte(key))
var val []byte
err = row.Scan(&val)
if err == sql.ErrNoRows {
return nil
}
if err != nil {
return Error.Wrap(err)
}
return storage.ErrValueChanged.New("%q", key)
}
if oldValue == nil {
q := `
INSERT INTO pathdata (fullpath, metadata) VALUES ($1::BYTEA, $2::BYTEA)
ON CONFLICT DO NOTHING
RETURNING 1
`
row := client.db.QueryRow(ctx, q, []byte(key), []byte(newValue))
var val []byte
err = row.Scan(&val)
if err == sql.ErrNoRows {
return storage.ErrValueChanged.New("%q", key)
}
return Error.Wrap(err)
}
var row *sql.Row
if newValue == nil {
q := `
WITH matching_key AS (
SELECT * FROM pathdata WHERE fullpath = $1::BYTEA
), updated AS (
DELETE FROM pathdata
USING matching_key mk
WHERE pathdata.metadata = $2::BYTEA
AND pathdata.fullpath = mk.fullpath
RETURNING 1
)
SELECT EXISTS(SELECT 1 FROM matching_key) AS key_present, EXISTS(SELECT 1 FROM updated) AS value_updated
`
row = client.db.QueryRow(ctx, q, []byte(key), []byte(oldValue))
} else {
q := `
WITH matching_key AS (
SELECT * FROM pathdata WHERE fullpath = $1::BYTEA
), updated AS (
UPDATE pathdata
SET metadata = $3::BYTEA
FROM matching_key mk
WHERE pathdata.metadata = $2::BYTEA
AND pathdata.fullpath = mk.fullpath
RETURNING 1
)
SELECT EXISTS(SELECT 1 FROM matching_key) AS key_present, EXISTS(SELECT 1 FROM updated) AS value_updated;
`
row = client.db.QueryRow(ctx, q, []byte(key), []byte(oldValue), []byte(newValue))
}
var keyPresent, valueUpdated bool
err = row.Scan(&keyPresent, &valueUpdated)
if err != nil {
return Error.Wrap(err)
}
if !keyPresent {
return storage.ErrKeyNotFound.New("%q", key)
}
if !valueUpdated {
return storage.ErrValueChanged.New("%q", key)
}
return nil
}