feat(client): add allowed logstream list option to varlog client

kakao · Oct 6, 2022 · 1ce80e7 · 1ce80e7
1 parent 0b64855
commit 1ce80e7
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/pkg/varlog/operations.go b/pkg/varlog/operations.go
@@ -29,6 +29,11 @@ func (v *logImpl) append(ctx context.Context, tpid types.TopicID, lsid types.Log
 				result.Err = multierr.Append(result.Err, err)
 				continue
 			}
+			if _, ok = appendOpts.allowLogStreams[lsid]; appendOpts.allowLogStreams != nil && !ok {
+				err := fmt.Errorf("append: not allowed lsid %d", lsid)
+				result.Err = multierr.Append(result.Err, err)
+				continue
+			}
 		}
 		res, err := v.appendTo(ctx, tpid, lsid, data)
 		if err != nil {

diff --git a/pkg/varlog/options.go b/pkg/varlog/options.go
@@ -6,6 +6,8 @@ import (
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/kakao/varlog/pkg/types"
 )
 
 const (
@@ -152,6 +154,7 @@ func defaultAppendOptions() appendOptions {
 type appendOptions struct {
 	retryCount      int
 	selectLogStream bool
+	allowLogStreams map[types.LogStreamID]struct{}
 }
 
 type AppendOption interface {
@@ -182,6 +185,12 @@ func withoutSelectLogStream() AppendOption {
 	})
 }
 
+func WithAllowLogStreams(logStreams map[types.LogStreamID]struct{}) AppendOption {
+	return newAppendOption(func(opts *appendOptions) {
+		opts.allowLogStreams = logStreams
+	})
+}
+
 func defaultSubscribeOptions() subscribeOptions {
 	return subscribeOptions{
 		timeout: defaultSubscribeTimeout,