Skip to content

Commit d6e45c2

Browse files
torcolvintorcolvin
committed
CBG-4600 apply networkType to cbgt
1 parent 770a635 commit d6e45c2

6 files changed

Lines changed: 258 additions & 40 deletions

File tree

‎base/dcp_sharded.go‎

Lines changed: 42 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
"time"
2525

2626
"github.com/couchbase/cbgt"
27+
"github.com/couchbase/gocbcore/v10/connstr"
2728
)
2829

2930
const (
@@ -322,6 +323,42 @@ func getCBGTIndexUUID(manager *cbgt.Manager, indexName string) (previousUUID str
322323
}
323324
}
324325

326+
// cbgtManagerOptions builds the options map passed to cbgt.NewManagerEx for the given DCP connection string.
327+
func cbgtManagerOptions(ctx context.Context, serverURL string) (map[string]string, error) {
328+
// Specify one feed per pindex
329+
options := make(map[string]string)
330+
options[cbgt.FeedAllotmentOption] = cbgt.FeedAllotmentOnePerPIndex
331+
options["managerLoadDataDir"] = "false"
332+
// TLS is controlled by the connection string.
333+
// cbgt uses this parameter to run in mixed mode - non-TLS for CCCP but TLS for memcached. Sync Gateway does not need to set this parameter.
334+
options["feedInitialBootstrapNonTLS"] = "false"
335+
336+
// Since cbgt initializes a buffer per CBS node per partition in most cases (vbuckets in partitions can't be grouped by CBS node),
337+
// setting the small buffer size used in cbgt 1.3.2. (see CBG-3341 for potential optimization of this value)
338+
idealKvConnectionBufferSize := 16384
339+
// This value will be overridden by the BucketSpec connection string.
340+
options["kvConnectionBufferSize"] = strconv.Itoa(idealKvConnectionBufferSize)
341+
connSpec, err := connstr.Parse(serverURL)
342+
if err != nil {
343+
return nil, fmt.Errorf("failed to parse connection string: %w", err)
344+
}
345+
346+
connStrKvBufferSize, err := getConnSpecOption[int](&connSpec, kvBufferSizeKey)
347+
if err == nil && connStrKvBufferSize != nil && *connStrKvBufferSize > idealKvConnectionBufferSize {
348+
WarnfCtx(ctx, "DCP sharded import connection string includes %s=%d, which is more than the implicit %s=%d. This will result in increased memory usage.", kvBufferSizeKey, *connStrKvBufferSize, kvBufferSizeKey, idealKvConnectionBufferSize)
349+
}
350+
351+
networkType, err := getConnSpecOption[string](&connSpec, networkKey)
352+
if err != nil {
353+
return nil, fmt.Errorf("failed to determine network type from connection string: %w", err)
354+
}
355+
if networkType != nil {
356+
options["gocbcoreIOConfigNetworkType"] = *networkType
357+
}
358+
359+
return options, nil
360+
}
361+
325362
// createCBGTManager creates a new manager for a given bucket and bucketSpec
326363
// Inline comments below provide additional detail on how cbgt uses each manager
327364
// parameter, and the implications for SG
@@ -378,31 +415,18 @@ func initCBGTManager(ctx context.Context, bucket Bucket, spec BucketSpec, cfgSG
378415
// avoids file system usage, in conjunction with managerLoadDataDir=false in options.
379416
dataDir := ""
380417

418+
options, err := cbgtManagerOptions(ctx, serverURL)
419+
if err != nil {
420+
return nil, err
421+
}
422+
381423
eventHandlersCtx, eventHandlersCancel := context.WithCancelCause(ctx)
382424
eventHandlers := &sgMgrEventHandlers{
383425
ctx: eventHandlersCtx,
384426
ctxCancel: eventHandlersCancel,
385427
unregisterFeedCallback: unregisterCallback,
386428
}
387429

388-
// Specify one feed per pindex
389-
options := make(map[string]string)
390-
options[cbgt.FeedAllotmentOption] = cbgt.FeedAllotmentOnePerPIndex
391-
options["managerLoadDataDir"] = "false"
392-
// TLS is controlled by the connection string.
393-
// cbgt uses this parameter to run in mixed mode - non-TLS for CCCP but TLS for memcached. Sync Gateway does not need to set this parameter.
394-
options["feedInitialBootstrapNonTLS"] = "false"
395-
396-
// Since cbgt initializes a buffer per CBS node per partition in most cases (vbuckets in partitions can't be grouped by CBS node),
397-
// setting the small buffer size used in cbgt 1.3.2. (see CBG-3341 for potential optimization of this value)
398-
idealKvConnectionBufferSize := 16384
399-
// This value will be overridden by the BucketSpec connection string.
400-
options["kvConnectionBufferSize"] = strconv.Itoa(idealKvConnectionBufferSize)
401-
402-
connStrKvBufferSize, err := getIntFromConnStr(serverURL, kvBufferSizeKey)
403-
if err == nil && connStrKvBufferSize != nil && *connStrKvBufferSize > idealKvConnectionBufferSize {
404-
WarnfCtx(ctx, "DCP sharded import connection string includes %s=%d, which is more than the implicit %s=%d. This will result in increased memory usage.", kvBufferSizeKey, *connStrKvBufferSize, kvBufferSizeKey, idealKvConnectionBufferSize)
405-
}
406430
// Creates a new cbgt manager.
407431
mgr := cbgt.NewManagerEx(
408432
SGCbgtMetadataVersion, // cbgt metadata version, matching 3.0 clients

‎base/dcp_test.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -563,3 +563,107 @@ func TestCBGTKvPoolSize(t *testing.T) {
563563
defer cbgtContext.Stop(ctx)
564564
require.Contains(t, cbgtContext.Manager.Server(), "kv_pool_size=1")
565565
}
566+
567+
func TestCBGTManagerOptions(t *testing.T) {
568+
testCases := []struct {
569+
name string
570+
server string
571+
expectedOptions map[string]string
572+
}{
573+
{
574+
name: "no options",
575+
server: "couchbase://127.0.0.1",
576+
expectedOptions: map[string]string{
577+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
578+
"managerLoadDataDir": "false",
579+
"feedInitialBootstrapNonTLS": "false",
580+
"kvConnectionBufferSize": "16384",
581+
},
582+
},
583+
{
584+
name: "network=default",
585+
server: "couchbase://127.0.0.1?network=default",
586+
expectedOptions: map[string]string{
587+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
588+
"managerLoadDataDir": "false",
589+
"feedInitialBootstrapNonTLS": "false",
590+
"kvConnectionBufferSize": "16384",
591+
"gocbcoreIOConfigNetworkType": "default",
592+
},
593+
},
594+
{
595+
name: "network=external",
596+
server: "couchbase://127.0.0.1?network=external",
597+
expectedOptions: map[string]string{
598+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
599+
"managerLoadDataDir": "false",
600+
"feedInitialBootstrapNonTLS": "false",
601+
"kvConnectionBufferSize": "16384",
602+
"gocbcoreIOConfigNetworkType": "external",
603+
},
604+
},
605+
{
606+
name: "network=auto",
607+
server: "couchbase://127.0.0.1?network=auto",
608+
expectedOptions: map[string]string{
609+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
610+
"managerLoadDataDir": "false",
611+
"feedInitialBootstrapNonTLS": "false",
612+
"kvConnectionBufferSize": "16384",
613+
"gocbcoreIOConfigNetworkType": "auto",
614+
},
615+
},
616+
{
617+
// kv_buffer_size never overrides the returned kvConnectionBufferSize option - the connection string is
618+
// passed to cbgt.NewManagerEx separately and cbgt applies it there.
619+
name: "kv_buffer_size below implicit size",
620+
server: "couchbase://127.0.0.1?kv_buffer_size=100",
621+
expectedOptions: map[string]string{
622+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
623+
"managerLoadDataDir": "false",
624+
"feedInitialBootstrapNonTLS": "false",
625+
"kvConnectionBufferSize": "16384",
626+
},
627+
},
628+
{
629+
name: "kv_buffer_size above implicit size",
630+
server: "couchbase://127.0.0.1?kv_buffer_size=20000",
631+
expectedOptions: map[string]string{
632+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
633+
"managerLoadDataDir": "false",
634+
"feedInitialBootstrapNonTLS": "false",
635+
"kvConnectionBufferSize": "16384",
636+
},
637+
},
638+
{
639+
// an unparsable kv_buffer_size is silently ignored, rather than failing the manager options build.
640+
name: "kv_buffer_size not an int",
641+
server: "couchbase://127.0.0.1?kv_buffer_size=notanumber",
642+
expectedOptions: map[string]string{
643+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
644+
"managerLoadDataDir": "false",
645+
"feedInitialBootstrapNonTLS": "false",
646+
"kvConnectionBufferSize": "16384",
647+
},
648+
},
649+
{
650+
// multiple values for a single option is also silently ignored.
651+
name: "multiple kv_buffer_size values",
652+
server: "couchbase://127.0.0.1?kv_buffer_size=20000&kv_buffer_size=30000",
653+
expectedOptions: map[string]string{
654+
cbgt.FeedAllotmentOption: cbgt.FeedAllotmentOnePerPIndex,
655+
"managerLoadDataDir": "false",
656+
"feedInitialBootstrapNonTLS": "false",
657+
"kvConnectionBufferSize": "16384",
658+
},
659+
},
660+
}
661+
for _, testCase := range testCases {
662+
t.Run(testCase.name, func(t *testing.T) {
663+
ctx := TestCtx(t)
664+
options, err := cbgtManagerOptions(ctx, testCase.server)
665+
require.NoError(t, err)
666+
require.Equal(t, testCase.expectedOptions, options)
667+
})
668+
}
669+
}

‎base/gocb_connection_string.go‎

Lines changed: 32 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -13,13 +13,14 @@ import (
1313
"net/url"
1414
"strconv"
1515

16-
"github.com/couchbaselabs/gocbconnstr"
16+
"github.com/couchbase/gocbcore/v10/connstr"
1717
)
1818

1919
const (
2020
dcpBufferSizeKey = "dcp_buffer_size"
2121
kvBufferSizeKey = "kv_buffer_size"
2222
kvPoolSizeKey = "kv_pool_size"
23+
networkKey = "network"
2324
)
2425

2526
// GoCBConnStringParams represents parameters that are passed to gocb when creating a new connection string.
@@ -39,8 +40,8 @@ func DefaultGoCBConnStringParams() *GoCBConnStringParams {
3940
}
4041

4142
// getGoCBConnSpec returns a gocb connection spec based on the server string. The provided defaults will be used only when the corresponding property is not set in the connection string.
42-
func getGoCBConnSpec(server string, defaults *GoCBConnStringParams) (*gocbconnstr.ConnSpec, error) {
43-
connSpec, err := gocbconnstr.Parse(server)
43+
func getGoCBConnSpec(server string, defaults *GoCBConnStringParams) (*connstr.ConnSpec, error) {
44+
connSpec, err := connstr.Parse(server)
4445
if err != nil {
4546
return nil, err
4647
}
@@ -83,26 +84,38 @@ func GetGoCBConnStringWithDefaults(server string, defaults *GoCBConnStringParams
8384
return connSpec.String(), nil
8485
}
8586

86-
// getIntFromConnStr returns a query parameter from a connection string. If it doesn't exist, return nil and no error. If there's an error in parsing the connection string, return an error.
87-
func getIntFromConnStr(connstr, key string) (*int, error) {
88-
connSpec, err := getGoCBConnSpec(connstr, nil)
89-
if err != nil {
90-
return nil, err
91-
}
92-
93-
values := url.Values(connSpec.Options)
94-
95-
arg := values[key]
96-
87+
// getConnSpecOption returns a single query parameter value from a connstr.ConnSpec, converted to type T (string or
88+
// int). If the option isn't set, returns nil and no error. Returns an error if the option is set more than once, or
89+
// if the value can't be converted to T.
90+
func getConnSpecOption[T string | int](spec *connstr.ConnSpec, key string) (*T, error) {
91+
arg := spec.Options[key]
9792
if len(arg) == 0 {
9893
return nil, nil
99-
} else if len(arg) > 1 {
100-
return nil, fmt.Errorf("Multiple %s values found in connection string %s", key, connstr)
94+
}
95+
if len(arg) > 1 {
96+
return nil, fmt.Errorf("multiple %s values found in connection string %q", key, spec.String())
10197
}
10298

103-
i, err := strconv.Atoi(arg[0])
99+
var value any
100+
switch any(*new(T)).(type) {
101+
case int:
102+
i, err := strconv.Atoi(arg[0])
103+
if err != nil {
104+
return nil, fmt.Errorf("invalid %s value %s in connection string %q, must be int", key, arg[0], spec.String())
105+
}
106+
value = i
107+
case string:
108+
value = arg[0]
109+
}
110+
typed := value.(T)
111+
return &typed, nil
112+
}
113+
114+
// getIntFromConnStr returns a query parameter from a connection string. If it doesn't exist, return nil and no error. If there's an error in parsing the connection string, return an error.
115+
func getIntFromConnStr(server, key string) (*int, error) {
116+
connSpec, err := getGoCBConnSpec(server, nil)
104117
if err != nil {
105-
return nil, fmt.Errorf("Invalid %s value %s in connection string %s, must be int", key, arg[0], connstr)
118+
return nil, err
106119
}
107-
return &i, nil
120+
return getConnSpecOption[int](connSpec, key)
108121
}

‎base/gocb_connection_string_test.go‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ package base
1111
import (
1212
"testing"
1313

14+
"github.com/couchbase/gocbcore/v10/connstr"
1415
"github.com/couchbase/sync_gateway/testing/require"
1516
)
1617

@@ -75,6 +76,85 @@ func TestGetGoCBConnStringWithDefaults(t *testing.T) {
7576
}
7677
}
7778

79+
func TestGetConnSpecOption(t *testing.T) {
80+
t.Run("int", func(t *testing.T) {
81+
testCases := []struct {
82+
name string
83+
options map[string][]string
84+
expected *int
85+
expectedError bool
86+
}{
87+
{
88+
name: "not set",
89+
options: map[string][]string{},
90+
},
91+
{
92+
name: "single value",
93+
options: map[string][]string{"kv_pool_size": {"8"}},
94+
expected: Ptr(8),
95+
},
96+
{
97+
name: "multiple values",
98+
options: map[string][]string{"kv_pool_size": {"8", "4"}},
99+
expectedError: true,
100+
},
101+
{
102+
name: "non-int value",
103+
options: map[string][]string{"kv_pool_size": {"notanint"}},
104+
expectedError: true,
105+
},
106+
}
107+
for _, testCase := range testCases {
108+
t.Run(testCase.name, func(t *testing.T) {
109+
spec := &connstr.ConnSpec{Options: testCase.options}
110+
value, err := getConnSpecOption[int](spec, "kv_pool_size")
111+
if testCase.expectedError {
112+
require.Error(t, err)
113+
} else {
114+
require.NoError(t, err)
115+
require.Equal(t, testCase.expected, value)
116+
}
117+
})
118+
}
119+
})
120+
121+
t.Run("string", func(t *testing.T) {
122+
testCases := []struct {
123+
name string
124+
options map[string][]string
125+
expected *string
126+
expectedError bool
127+
}{
128+
{
129+
name: "not set",
130+
options: map[string][]string{},
131+
},
132+
{
133+
name: "single value",
134+
options: map[string][]string{networkKey: {"external"}},
135+
expected: Ptr("external"),
136+
},
137+
{
138+
name: "multiple values",
139+
options: map[string][]string{networkKey: {"external", "default"}},
140+
expectedError: true,
141+
},
142+
}
143+
for _, testCase := range testCases {
144+
t.Run(testCase.name, func(t *testing.T) {
145+
spec := &connstr.ConnSpec{Options: testCase.options}
146+
value, err := getConnSpecOption[string](spec, networkKey)
147+
if testCase.expectedError {
148+
require.Error(t, err)
149+
} else {
150+
require.NoError(t, err)
151+
require.Equal(t, testCase.expected, value)
152+
}
153+
})
154+
}
155+
})
156+
}
157+
78158
func TestGetIntFromConnStr(t *testing.T) {
79159
testCases := []struct {
80160
name string

‎go.mod‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ require (
1616
github.com/couchbase/sg-bucket v0.0.0-20260706130600-fc995d3ac4be
1717
github.com/couchbasedeps/fast-skiplist v0.0.0-20250722125747-e0dd031fe2ac
1818
github.com/couchbaselabs/go-fleecedelta v0.0.0-20220909152808-6d09efa7a338
19-
github.com/couchbaselabs/gocbconnstr v1.0.5
2019
github.com/couchbaselabs/rosmar v0.0.0-20260706134933-2f0dabde975c
2120
github.com/elastic/gosigar v0.14.4
2221
github.com/felixge/fgprof v0.9.5

‎go.sum‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,6 @@ github.com/couchbaselabs/go-fleecedelta v0.0.0-20220909152808-6d09efa7a338 h1:xM
6464
github.com/couchbaselabs/go-fleecedelta v0.0.0-20220909152808-6d09efa7a338/go.mod h1:0f+dmhfcTKK+4quAe6rwqQUVVWtHX/eztNB8cmBUniQ=
6565
github.com/couchbaselabs/gocaves/client v0.0.0-20250107114554-f96479220ae8 h1:MQfvw4BiLTuyR69FuA5Kex+tXUeLkH+/ucJfVL1/hkM=
6666
github.com/couchbaselabs/gocaves/client v0.0.0-20250107114554-f96479220ae8/go.mod h1:AVekAZwIY2stsJOMWLAS/0uA/+qdp7pjO8EHnl61QkY=
67-
github.com/couchbaselabs/gocbconnstr v1.0.5 h1:e0JokB5qbcz7rfnxEhNRTKz8q1svoRvDoZihsiwNigA=
68-
github.com/couchbaselabs/gocbconnstr v1.0.5/go.mod h1:KV3fnIKMi8/AzX0O9zOrO9rofEqrRF1d2rG7qqjxC7o=
6967
github.com/couchbaselabs/gocbconnstr/v2 v2.0.0 h1:HU9DlAYYWR69jQnLN6cpg0fh0hxW/8d5hnglCXXjW78=
7068
github.com/couchbaselabs/gocbconnstr/v2 v2.0.0/go.mod h1:o7T431UOfFVHDNvMBUmUxpHnhivwv7BziUao/nMl81E=
7169
github.com/couchbaselabs/rosmar v0.0.0-20260706134933-2f0dabde975c h1:rTggaJZ5WvEizK3YV6B+wFvqlC3pUxHRCqH4oYzff9M=

0 commit comments

Comments
 (0)