From bcb5fca49355c5d8cc7f183f0af2f3c71d30307c Mon Sep 17 00:00:00 2001 From: mpl Date: Wed, 31 Aug 2016 17:51:43 +0200 Subject: [PATCH] pkg/sorted: log and skip on too large a key/value So far we were returning an error, which appears to be a little too strict, so we now just ignore when a key or value is too large, log it, and go on. Fixes #849 Change-Id: Iadd4eaab7459643e22ab3043d1f45e3eab662b30 --- pkg/sorted/buffer/buffer.go | 7 +++++-- pkg/sorted/kvfile/kvfile.go | 6 ++++-- pkg/sorted/kvtest/kvtest.go | 18 ++++++++++++++---- pkg/sorted/leveldb/leveldb.go | 10 ++++------ pkg/sorted/mem.go | 9 +++++++-- pkg/sorted/mongo/mongokv.go | 7 +++++-- pkg/sorted/mysql/mysqlkv_test.go | 9 +++++---- pkg/sorted/sqlkv/sqlkv.go | 9 +++------ 8 files changed, 47 insertions(+), 28 deletions(-) diff --git a/pkg/sorted/buffer/buffer.go b/pkg/sorted/buffer/buffer.go index 8ecb19b80..09be66c62 100644 --- a/pkg/sorted/buffer/buffer.go +++ b/pkg/sorted/buffer/buffer.go @@ -22,6 +22,7 @@ package buffer // import "camlistore.org/pkg/sorted/buffer" import ( "fmt" + "log" "sync" "camlistore.org/pkg/sorted" @@ -101,7 +102,8 @@ func (kv *KeyValue) Get(key string) (string, error) { func (kv *KeyValue) Set(key, value string) error { if err := sorted.CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } kv.mu.RLock() err := kv.buf.Set(key, value) @@ -163,7 +165,8 @@ func (kv *KeyValue) CommitBatch(bm sorted.BatchMutation) error { continue } else { if err := sorted.CheckSizes(m.key, m.value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", m.key, m.value, err) + continue } } bmbuf.Set(m.key, m.value) diff --git a/pkg/sorted/kvfile/kvfile.go b/pkg/sorted/kvfile/kvfile.go index 5a7dc8819..7ce3c6abf 100644 --- a/pkg/sorted/kvfile/kvfile.go +++ b/pkg/sorted/kvfile/kvfile.go @@ -93,7 +93,8 @@ func (is *kvis) Get(key string) (string, error) { func (is *kvis) Set(key, value string) error { if err := sorted.CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } return is.db.Set([]byte(key), []byte(value)) } @@ -162,7 +163,8 @@ func (is *kvis) CommitBatch(bm sorted.BatchMutation) error { } } else { if err := sorted.CheckSizes(m.Key(), m.Value()); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", m.Key(), m.Value(), err) + continue } if err := is.db.Set([]byte(m.Key()), []byte(m.Value())); err != nil { return err diff --git a/pkg/sorted/kvtest/kvtest.go b/pkg/sorted/kvtest/kvtest.go index 3f4c2862f..115b1655a 100644 --- a/pkg/sorted/kvtest/kvtest.go +++ b/pkg/sorted/kvtest/kvtest.go @@ -167,11 +167,21 @@ func testInsertLarge(t *testing.T, kv sorted.KeyValue) { func testInsertTooLarge(t *testing.T, kv sorted.KeyValue) { largeKey := make([]byte, sorted.MaxKeySize+1) largeValue := make([]byte, sorted.MaxValueSize+1) - if err := kv.Set(string(largeKey), "whatever"); err == nil || err != sorted.ErrKeyTooLarge { - t.Fatalf("Insertion of too large a key should have failed, but err was %v", err) + err := kv.Set(string(largeKey), "whatever") + if err != nil { + t.Fatalf("Insertion of too large a key should have skipped with some logging, but err was %v", err) } - if err := kv.Set("whatever", string(largeValue)); err == nil || err != sorted.ErrValueTooLarge { - t.Fatalf("Insertion of too large a value should have failed, but err was %v", err) + _, err = kv.Get(string(largeKey)) + if err == nil { + t.Fatal("large key should not have been inserted") + } + err = kv.Set("testInsertTooLarge", string(largeValue)) + if err != nil { + t.Fatalf("Insertion of too large a value should have skipped with some logging, but err was %v", err) + } + _, err = kv.Get("testInsertTooLarge") + if err == nil { + t.Fatal("large value should not have been inserted") } } diff --git a/pkg/sorted/leveldb/leveldb.go b/pkg/sorted/leveldb/leveldb.go index b1d8e147e..e72d4b5bd 100644 --- a/pkg/sorted/leveldb/leveldb.go +++ b/pkg/sorted/leveldb/leveldb.go @@ -22,6 +22,7 @@ package leveldb // import "camlistore.org/pkg/sorted/leveldb" import ( "errors" "fmt" + "log" "os" "sync" @@ -109,7 +110,8 @@ func (is *kvis) Get(key string) (string, error) { func (is *kvis) Set(key, value string) error { if err := sorted.CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } return is.db.Put([]byte(key), []byte(value), is.writeOpts) } @@ -172,11 +174,7 @@ func (lvb *lvbatch) Set(key, value string) { return } if err := sorted.CheckSizes(key, value); err != nil { - if err == sorted.ErrKeyTooLarge { - lvb.err = fmt.Errorf("%v: %v", err, key) - } else { - lvb.err = fmt.Errorf("%v: %v", err, value) - } + log.Printf("Skipping storing (%q:%q): %v", key, value, err) return } lvb.batch.Put([]byte(key), []byte(value)) diff --git a/pkg/sorted/mem.go b/pkg/sorted/mem.go index f6f7c069d..30db7ab26 100644 --- a/pkg/sorted/mem.go +++ b/pkg/sorted/mem.go @@ -18,6 +18,7 @@ package sorted import ( "errors" + "log" "sync" "github.com/syndtr/goleveldb/leveldb/comparer" @@ -111,7 +112,8 @@ func (mk *memKeys) Find(start, end string) Iterator { func (mk *memKeys) Set(key, value string) error { if err := CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } mk.mu.Lock() defer mk.mu.Unlock() @@ -145,8 +147,11 @@ func (mk *memKeys) CommitBatch(bm BatchMutation) error { return err } } else { + // TODO(mpl): we need to force a reindex when we have a proper solution + // for storing these too large attributes, if ever. if err := CheckSizes(m.Key(), m.Value()); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", m.Key(), m.Value(), err) + continue } if err := mk.db.Put([]byte(m.Key()), []byte(m.Value())); err != nil { return err diff --git a/pkg/sorted/mongo/mongokv.go b/pkg/sorted/mongo/mongokv.go index 910bf4cdc..3fdf6072e 100644 --- a/pkg/sorted/mongo/mongokv.go +++ b/pkg/sorted/mongo/mongokv.go @@ -21,6 +21,7 @@ package mongo // import "camlistore.org/pkg/sorted/mongo" import ( "bytes" "errors" + "log" "sync" "time" @@ -146,7 +147,8 @@ func (kv *keyValue) Find(start, end string) sorted.Iterator { func (kv *keyValue) Set(key, value string) error { if err := sorted.CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } kv.mu.Lock() defer kv.mu.Unlock() @@ -196,7 +198,8 @@ func (kv *keyValue) CommitBatch(bm sorted.BatchMutation) error { } } else { if err := sorted.CheckSizes(m.Key(), m.Value()); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", m.Key(), m.Value(), err) + continue } if _, err := kv.db.Upsert(&bson.M{mgoKey: m.Key()}, &bson.M{mgoKey: m.Key(), mgoValue: m.Value()}); err != nil { return err diff --git a/pkg/sorted/mysql/mysqlkv_test.go b/pkg/sorted/mysql/mysqlkv_test.go index e2a533897..281a96fdb 100644 --- a/pkg/sorted/mysql/mysqlkv_test.go +++ b/pkg/sorted/mysql/mysqlkv_test.go @@ -17,6 +17,8 @@ limitations under the License. package mysql import ( + "database/sql" + "errors" "testing" "time" @@ -65,9 +67,8 @@ func TestRollback(t *testing.T) { t.Fatalf("mysql.NewKeyValue = %v", err) } - tooLargeAKey := make([]byte, sorted.MaxKeySize+10) - for i := range tooLargeAKey { - tooLargeAKey[i] = 'L' + kv.(*keyValue).KeyValue.BatchSetFunc = func(*sql.Tx, string, string) error { + return errors.New("Forced failure to trigger a rollback") } nbConnections := 2 @@ -82,7 +83,7 @@ func TestRollback(t *testing.T) { b := kv.BeginBatch() // Making the transaction fail, to force a rollback // -> this whole test fails before we introduce the rollback in CommitBatch. - b.Set(string(tooLargeAKey), "whatever") + b.Set("foo", "bar") if err := kv.CommitBatch(b); err == nil { t.Fatal("wanted failed commit because too large a key") } diff --git a/pkg/sorted/sqlkv/sqlkv.go b/pkg/sorted/sqlkv/sqlkv.go index 091388399..19c4f5828 100644 --- a/pkg/sorted/sqlkv/sqlkv.go +++ b/pkg/sorted/sqlkv/sqlkv.go @@ -105,11 +105,7 @@ func (b *batchTx) Set(key, value string) { return } if err := sorted.CheckSizes(key, value); err != nil { - if err == sorted.ErrKeyTooLarge { - b.err = fmt.Errorf("%v: %v", err, key) - } else { - b.err = fmt.Errorf("%v: %v", err, value) - } + log.Printf("Skipping storing (%q:%q): %v", key, value, err) return } if b.kv.BatchSetFunc != nil { @@ -172,7 +168,8 @@ func (kv *KeyValue) Get(key string) (value string, err error) { func (kv *KeyValue) Set(key, value string) error { if err := sorted.CheckSizes(key, value); err != nil { - return err + log.Printf("Skipping storing (%q:%q): %v", key, value, err) + return nil } if kv.Gate != nil { kv.Gate.Start()