Add internal KVS pogreb package (#2302)

* feat: add internal pogreb kvs package Signed-off-by: hlts2 <[email protected]> * fix: typo Signed-off-by: hlts2 <[email protected]> * fix: tweak Signed-off-by: hlts2 <[email protected]> * fix: deepsource warning Signed-off-by: hlts2 <[email protected]> * feat: add test for kvs and small refactor Signed-off-by: hlts2 <[email protected]> * feat: add condition to check key existance Signed-off-by: hlts2 <[email protected]> * fix: small refactor Signed-off-by: hlts2 <[email protected]> * fix: tuning option value Signed-off-by: hlts2 <[email protected]> * fix: deepsource warning Signed-off-by: hlts2 <[email protected]> * fix: make format Signed-off-by: hlts2 <[email protected]> * Update internal/db/kvs/pogreb/pogreb.go Co-authored-by: datelier <[email protected]> * fix: error handling for option and update comment Signed-off-by: hlts2 <[email protected]> * fix: option setting Signed-off-by: hlts2 <[email protected]> --------- Signed-off-by: hlts2 <[email protected]> Co-authored-by: datelier <[email protected]>
vdaas · Jan 24, 2024 · 15ce01d · 15ce01d
1 parent 07570a1
commit 15ce01d
Show file tree

Hide file tree

Showing 7 changed files with 1,287 additions and 0 deletions.
diff --git a/go.mod b/go.mod
@@ -421,6 +421,7 @@ require (
 	github.com/Masterminds/goutils v1.1.1 // indirect
 	github.com/Masterminds/semver v1.5.0 // indirect
 	github.com/ajstarks/svgo v0.0.0-20211024235047-1546f124cd8b // indirect
+	github.com/akrylysov/pogreb v0.10.2 // indirect
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/campoy/embedmd v1.0.0 // indirect

diff --git a/go.sum b/go.sum
@@ -172,6 +172,8 @@ github.com/ajstarks/svgo v0.0.0-20211024235047-1546f124cd8b h1:slYM766cy2nI3BwyR
 github.com/ajstarks/svgo v0.0.0-20211024235047-1546f124cd8b/go.mod h1:1KcenG0jGWcpt8ov532z81sp/kMMUG485J2InIOyADM=
 github.com/akavel/rsrc v0.8.0/go.mod h1:uLoCtb9J+EyAqh+26kdrTgmzRBFPGOolLWKpdxkKq+c=
 github.com/akavel/rsrc v0.10.2/go.mod h1:uLoCtb9J+EyAqh+26kdrTgmzRBFPGOolLWKpdxkKq+c=
+github.com/akrylysov/pogreb v0.10.2 h1:e6PxmeyEhWyi2AKOBIJzAEi4HkiC+lKyCocRGlnDi78=
+github.com/akrylysov/pogreb v0.10.2/go.mod h1:pNs6QmpQ1UlTJKDezuRWmaqkgUE2TuU0YTWyqJZ7+lI=
 github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
 github.com/andybalholm/stroke v0.0.0-20221221101821-bd29b49d73f0/go.mod h1:ccdDYaY5+gO+cbnQdFxEXqfy0RkoV25H3jLXUDNM3wg=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=

diff --git a/hack/go.mod.default b/hack/go.mod.default
@@ -234,6 +234,7 @@ replace (
 	github.com/pkg/errors => github.com/pkg/errors upgrade
 	github.com/pkg/sftp => github.com/pkg/sftp upgrade
 	github.com/pmezard/go-difflib => github.com/pmezard/go-difflib upgrade
+	github.com/akrylysov/pogreb => github.com/akrylysov/pogreb upgrade
 	github.com/prashantv/gostub => github.com/prashantv/gostub upgrade
 	github.com/prometheus/client_golang => github.com/prometheus/client_golang upgrade
 	github.com/prometheus/client_model => github.com/prometheus/client_model upgrade

diff --git a/internal/db/kvs/pogreb/options.go b/internal/db/kvs/pogreb/options.go
@@ -0,0 +1,84 @@
+// Copyright (C) 2019-2024 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package pogreb
+
+import (
+	"time"
+
+	"github.com/akrylysov/pogreb"
+)
+
+var deafultOpts = []Option{
+	WithPath("pogreb.db"),
+	WithBackgroundSyncInterval("5s"),
+}
+
+// Option represents the functional option for database.
+type Option func(*db) error
+
+// WithPath returns the option to set path.
+func WithPath(path string) Option {
+	return func(d *db) error {
+		if path != "" {
+			d.path = path
+		}
+		return nil
+	}
+}
+
+// WithBackgroundSyncInterval returns the option to sets the amount of time between background Sync() calls.
+// Setting the value to 0 disables the automatic background synchronization.
+// Setting the value to -1 or less makes the DB call Sync() after every write operation.
+func WithBackgroundSyncInterval(s string) Option {
+	return func(d *db) error {
+		if s == "" {
+			return nil
+		}
+		dur, err := time.ParseDuration(s)
+		if err != nil {
+			return err
+		}
+		if d.opts == nil {
+			d.opts = new(pogreb.Options)
+		}
+		if dur < -1 {
+			dur = -1
+		}
+		d.opts.BackgroundSyncInterval = dur
+		return nil
+	}
+}
+
+// WithBackgroundCompactionInterval returns the option to sets the amount of time between background Compact() calls.
+// Setting the value to 0 or less disables the automatic background compaction.
+func WithBackgroundCompactionInterval(s string) Option {
+	return func(d *db) error {
+		if s == "" {
+			return nil
+		}
+		dur, err := time.ParseDuration(s)
+		if err != nil {
+			return err
+		}
+		if d.opts == nil {
+			d.opts = new(pogreb.Options)
+		}
+
+		if dur < 0 {
+			dur = 0
+		}
+		d.opts.BackgroundCompactionInterval = dur
+		return nil
+	}
+}
diff --git a/internal/db/kvs/pogreb/options_test.go b/internal/db/kvs/pogreb/options_test.go
@@ -0,0 +1,261 @@
+// Copyright (C) 2019-2024 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package pogreb
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/akrylysov/pogreb"
+	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/test/goleak"
+)
+
+func TestWithPath(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		path string
+	}
+	type want struct {
+		want Option
+		err  error
+		path string
+	}
+	type test struct {
+		name       string
+		args       args
+		want       want
+		checkFunc  func(want, *db, error) error
+		beforeFunc func(*testing.T, args)
+		afterFunc  func(*testing.T, args)
+	}
+	defaultCheckFunc := func(w want, got *db, err error) error {
+		if !errors.Is(err, w.err) {
+			return errors.Errorf("got_error: \"%#v\",\n\t\t\t\twant: \"%#v\"", err, w.err)
+		}
+		if err == nil {
+			if !reflect.DeepEqual(got.path, w.path) {
+				return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", got, w.want)
+			}
+		}
+		return nil
+	}
+	tests := []test{
+		func() test {
+			path := "db-dir"
+			return test{
+				name: "Succeeds to apply option",
+				args: args{
+					path: path,
+				},
+				want: want{
+					path: path,
+				},
+			}
+		}(),
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc(tt, test.args)
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc(tt, test.args)
+			}
+			checkFunc := test.checkFunc
+			if test.checkFunc == nil {
+				checkFunc = defaultCheckFunc
+			}
+
+			got := new(db)
+			err := WithPath(test.args.path)(got)
+			if err := checkFunc(test.want, got, err); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+		})
+	}
+}
+
+func TestWithBackgroundSyncInterval(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		s string
+	}
+	type want struct {
+		want Option
+		err  error
+		opts *pogreb.Options
+	}
+	type test struct {
+		name       string
+		args       args
+		want       want
+		checkFunc  func(want, *db, error) error
+		beforeFunc func(*testing.T, args)
+		afterFunc  func(*testing.T, args)
+	}
+	defaultCheckFunc := func(w want, got *db, err error) error {
+		if !errors.Is(err, w.err) {
+			return errors.Errorf("got_error: \"%#v\",\n\t\t\t\twant: \"%#v\"", err, w.err)
+		}
+		if err == nil {
+			if !reflect.DeepEqual(got.opts.BackgroundSyncInterval, w.opts.BackgroundSyncInterval) {
+				return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", got, w.want)
+			}
+		}
+		return nil
+	}
+	tests := []test{
+		func() test {
+			dur := "100ms"
+			return test{
+				name: "Succeeds to apply option",
+				args: args{
+					s: dur,
+				},
+				want: want{
+					opts: &pogreb.Options{
+						BackgroundSyncInterval: 100 * time.Millisecond,
+					},
+				},
+			}
+		}(),
+		func() test {
+			dur := "invalid"
+			return test{
+				name: "Fails to apply option with invalid value",
+				args: args{
+					s: dur,
+				},
+				want: want{
+					err: errors.New("time: invalid duration \"invalid\""),
+				},
+			}
+		}(),
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc(tt, test.args)
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc(tt, test.args)
+			}
+			checkFunc := test.checkFunc
+			if test.checkFunc == nil {
+				checkFunc = defaultCheckFunc
+			}
+
+			got := new(db)
+			err := WithBackgroundSyncInterval(test.args.s)(got)
+			if err := checkFunc(test.want, got, err); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+		})
+	}
+}
+
+func TestWithBackgroundCompactionInterval(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		s string
+	}
+	type want struct {
+		want Option
+		err  error
+		opts *pogreb.Options
+	}
+	type test struct {
+		name       string
+		args       args
+		want       want
+		checkFunc  func(want, *db, error) error
+		beforeFunc func(*testing.T, args)
+		afterFunc  func(*testing.T, args)
+	}
+	defaultCheckFunc := func(w want, got *db, err error) error {
+		if !errors.Is(err, w.err) {
+			return errors.Errorf("got_error: \"%#v\",\n\t\t\t\twant: \"%#v\"", err, w.err)
+		}
+		if err == nil {
+			if !reflect.DeepEqual(got.opts.BackgroundCompactionInterval, w.opts.BackgroundCompactionInterval) {
+				return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", got, w.want)
+			}
+		}
+		return nil
+	}
+	tests := []test{
+		func() test {
+			dur := "100ms"
+			return test{
+				name: "Succeeds to apply option",
+				args: args{
+					s: dur,
+				},
+				want: want{
+					opts: &pogreb.Options{
+						BackgroundCompactionInterval: 100 * time.Millisecond,
+					},
+				},
+			}
+		}(),
+		func() test {
+			dur := "invalid"
+			return test{
+				name: "Fails to apply option with invalid value",
+				args: args{
+					s: dur,
+				},
+				want: want{
+					err: errors.New("time: invalid duration \"invalid\""),
+				},
+			}
+		}(),
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc(tt, test.args)
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc(tt, test.args)
+			}
+			checkFunc := test.checkFunc
+			if test.checkFunc == nil {
+				checkFunc = defaultCheckFunc
+			}
+
+			got := new(db)
+			err := WithBackgroundCompactionInterval(test.args.s)(got)
+			if err := checkFunc(test.want, got, err); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+		})
+	}
+}
+
+// NOT IMPLEMENTED BELOW