Merge pull request #117 from ydb-platform/traces

v3.10.0: Add trace stubs for discovery, scheme, scripting, coordiantion, ratelimiter clients + add options for define traces for specified client

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,16 @@
+## 3.10.0
+* Extended `trace.Details` constants for support per-service events
+* Added `trace.Discovery` struct for traces discovery events
+* Added `trace.Ratelimiter`, `trace.Coordination`, `trace.Scripting`, `trace.Scheme` stubs (will be implements in the future)
+* Added `ratelimiter/config`, `coordination/config`, `scripting/config`, `scheme/config`, `discovery/config` packages for specify per-service configs
+* Removed `trace.Driver.OnDiscovery` callback (moved to `trace.Discovery`)
+* Refactored initialization step (firstly makes discovery client)
+* Removed `internal/lazy.Discovery` (discovery client always initialized)
+* Fixed `trace.Table` event structs
+* Refactored grpc options for define dns-balancing configuration
+* Refactored `retry.Retry` signature (added `retry.WithID`, `retry.WithTrace` and `retry.WithIdempotent` opt-in args, required param `isIdempotentOperation` removed)
+* Refactored package `internal/repeater`
+
 ## 3.9.4
 * Fixed data race on closing session pool
 

config/config.go

@@ -14,10 +14,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
-const (
-	DefaultDiscoveryInterval = time.Minute
-)
-
 // Config contains driver configuration options.
 type Config interface {
 	// Endpoint is a required starting endpoint for connect
@@ -60,12 +56,6 @@ type Config interface {
 	// If OperationCancelAfter is zero then no timeout is used.
 	OperationCancelAfter() time.Duration
 
-	// DiscoveryInterval is the frequency of background tasks of ydb endpoints
-	// discovery.
-	// If DiscoveryInterval is zero then the DefaultDiscoveryInterval is used.
-	// If DiscoveryInterval is negative, then no background discovery prepared.
-	DiscoveryInterval() time.Duration
-
 	// Balancer is an optional configuration related to selected balancer.
 	// That is, some balancing methods allow to be configured.
 	Balancer() balancer.Balancer
@@ -102,7 +92,6 @@ type config struct {
 	streamTimeout        time.Duration
 	operationTimeout     time.Duration
 	operationCancelAfter time.Duration
-	discoveryInterval    time.Duration
 	dialTimeout          time.Duration
 	connectionTTL        time.Duration
 	balancer             balancer.Balancer
@@ -169,10 +158,6 @@ func (c *config) OperationCancelAfter() time.Duration {
 	return c.operationCancelAfter
 }
 
-func (c *config) DiscoveryInterval() time.Duration {
-	return c.discoveryInterval
-}
-
 func (c *config) Balancer() balancer.Balancer {
 	return c.balancer
 }
@@ -255,16 +240,6 @@ func WithOperationCancelAfter(operationCancelAfter time.Duration) Option {
 	}
 }
 
-func WithDiscoveryInterval(discoveryInterval time.Duration) Option {
-	return func(c *config) {
-		if discoveryInterval <= 0 {
-			c.discoveryInterval = 0
-		} else {
-			c.discoveryInterval = discoveryInterval
-		}
-	}
-}
-
 func WithDialTimeout(timeout time.Duration) Option {
 	return func(c *config) {
 		c.dialTimeout = timeout
@@ -303,9 +278,6 @@ func New(opts ...Option) Config {
 	if !c.secure {
 		c.tlsConfig = nil
 	}
-	if c.discoveryInterval == 0 {
-		c.balancer = balancers.SingleConn()
-	}
 	c.meta = meta.New(
 		c.database,
 		c.credentials,
@@ -333,9 +305,8 @@ func certPool() (certPool *x509.CertPool) {
 
 func defaultConfig() (c *config) {
 	return &config{
-		discoveryInterval: DefaultDiscoveryInterval,
-		balancer:          balancers.Default(),
-		secure:            true,
+		balancer: balancers.Default(),
+		secure:   true,
 		tlsConfig: &tls.Config{
 			MinVersion: tls.VersionTLS12,
 			RootCAs:    certPool(),

config/grpc_dial_options.go

@@ -21,11 +21,9 @@ func (c *config) GrpcDialOptions() (opts []grpc.DialOption) {
 		}),
 		grpc.WithKeepaliveParams(DefaultGrpcConnectionPolicy),
 		grpc.WithResolvers(
-			resolver.New(""), // for use this resolver by default
-			resolver.New("grpc"),
-			resolver.New("grpcs"),
+			resolver.New("ydb"),
 		),
-		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
+		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallRecvMsgSize(DefaultGRPCMsgSize),
 			grpc.MaxCallSendMsgSize(DefaultGRPCMsgSize),

connection.go

@@ -9,18 +9,23 @@ import (
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/coordination"
+	coordinationConfig "github.com/ydb-platform/ydb-go-sdk/v3/coordination/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/discovery"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/conn"
+	discoveryConfig "github.com/ydb-platform/ydb-go-sdk/v3/discovery/config"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/balancer/single"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/closer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/db"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/errors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/lazy"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/logger"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/meta"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/proxy"
-	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"github.com/ydb-platform/ydb-go-sdk/v3/ratelimiter"
+	ratelimiterConfig "github.com/ydb-platform/ydb-go-sdk/v3/ratelimiter/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
+	schemeConfig "github.com/ydb-platform/ydb-go-sdk/v3/scheme/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/scripting"
+	scriptingConfig "github.com/ydb-platform/ydb-go-sdk/v3/scripting/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
 	tableConfig "github.com/ydb-platform/ydb-go-sdk/v3/table/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
@@ -29,7 +34,8 @@ import (
 // Connection interface provide access to YDB service clients
 // Interface and list of clients may be changed in the future
 type Connection interface {
-	db.Connection
+	closer.Closer
+	db.ConnectionInfo
 
 	// Table returns table client with options from Connection instance.
 	// Options provide options replacement for requested table client
@@ -63,30 +69,35 @@ type Connection interface {
 }
 
 type connection struct {
-	config       config.Config
-	options      []config.Option
-	tableOptions []tableConfig.Option
-	scripting    scripting.Client
+	config  config.Config
+	options []config.Option
 
 	table        table.Client
-	scheme       scheme.Client
-	discovery    discovery.Client
-	coordination coordination.Client
-	rateLimiter  ratelimiter.Client
+	tableOptions []tableConfig.Option
+
+	scripting        scripting.Client
+	scriptingOptions []scriptingConfig.Option
+
+	scheme        scheme.Client
+	schemeOptions []schemeConfig.Option
+
+	discoveryOptions []discoveryConfig.Option
+
+	coordination        coordination.Client
+	coordinationOptions []coordinationConfig.Option
 
-	conns conn.Pool
-	mtx   sync.Mutex
-	db    db.Connection
+	ratelimiter        ratelimiter.Client
+	ratelimiterOptions []ratelimiterConfig.Option
+
+	mtx sync.Mutex
+	db  db.Connection
 }
 
 func (c *connection) Close(ctx context.Context) error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 	var issues []error
-	if err := c.discovery.Close(ctx); err != nil {
-		issues = append(issues, err)
-	}
-	if err := c.rateLimiter.Close(ctx); err != nil {
+	if err := c.ratelimiter.Close(ctx); err != nil {
 		issues = append(issues, err)
 	}
 	if err := c.coordination.Close(ctx); err != nil {
@@ -104,9 +115,6 @@ func (c *connection) Close(ctx context.Context) error {
 	if err := c.db.Close(ctx); err != nil {
 		issues = append(issues, err)
 	}
-	if err := c.conns.Close(ctx); err != nil {
-		issues = append(issues, err)
-	}
 	if len(issues) > 0 {
 		return errors.NewWithIssues("close failed", issues...)
 	}
@@ -167,16 +175,16 @@ func (c *connection) Coordination(opts ...CustomOption) coordination.Client {
 
 func (c *connection) Ratelimiter(opts ...CustomOption) ratelimiter.Client {
 	if len(opts) == 0 {
-		return c.rateLimiter
+		return c.ratelimiter
 	}
-	return proxy.Ratelimiter(c.rateLimiter, c.meta(opts...))
+	return proxy.Ratelimiter(c.ratelimiter, c.meta(opts...))
 }
 
 func (c *connection) Discovery(opts ...CustomOption) discovery.Client {
 	if len(opts) == 0 {
-		return c.discovery
+		return c.db.Discovery()
 	}
-	return proxy.Discovery(c.discovery, c.meta(opts...))
+	return proxy.Discovery(c.db.Discovery(), c.meta(opts...))
 }
 
 func (c *connection) Scripting(opts ...CustomOption) scripting.Client {
@@ -205,17 +213,14 @@ func New(ctx context.Context, opts ...Option) (_ Connection, err error) {
 	}
 	if logLevel, has := os.LookupEnv("YDB_LOG_SEVERITY_LEVEL"); has {
 		if l := logger.FromString(logLevel); l < logger.QUIET {
-			logger := logger.New(
-				logger.WithNamespace("ydb"),
-				logger.WithMinLevel(logger.FromString(logLevel)),
-				logger.WithNoColor(os.Getenv("YDB_LOG_NO_COLOR") != ""),
-			)
 			opts = append(
-				[]Option{
-					WithTraceDriver(log.Driver(logger, trace.DetailsAll)),
-					WithTraceTable(log.Table(logger, trace.DetailsAll)),
-				},
-				opts...,
+				opts,
+				WithLogger(
+					trace.DetailsAll,
+					WithNamespace("ydb"),
+					WithMinLevel(Level(logger.FromString(logLevel))),
+					WithNoColor(os.Getenv("YDB_LOG_NO_COLOR") != ""),
+				),
 			)
 		}
 	}
@@ -226,16 +231,42 @@ func New(ctx context.Context, opts ...Option) (_ Connection, err error) {
 		}
 	}
 	c.config = config.New(c.options...)
-	c.conns = conn.NewPool(ctx, c.config)
-	c.db, err = db.New(ctx, c.config, c.conns)
+	if c.config.Endpoint() == "" {
+		panic("empty dial address")
+	}
+	if c.config.Database() == "" {
+		panic("empty database")
+	}
+
+	if single.IsSingle(c.config.Balancer()) {
+		c.discoveryOptions = append(c.discoveryOptions, discoveryConfig.WithInterval(0))
+	}
+
+	c.db, err = db.New(
+		ctx,
+		c.config,
+		append(
+			[]discoveryConfig.Option{
+				discoveryConfig.WithEndpoint(c.Endpoint()),
+				discoveryConfig.WithDatabase(c.Name()),
+				discoveryConfig.WithSecure(c.Secure()),
+			},
+			c.discoveryOptions...,
+		)...,
+	)
 	if err != nil {
 		return nil, err
 	}
+
 	c.table = lazy.Table(c.db, c.tableOptions)
-	c.scheme = lazy.Scheme(c.db)
-	c.scripting = lazy.Scripting(c.db)
-	c.discovery = lazy.Discovery(c.db, c.config.Trace())
-	c.coordination = lazy.Coordination(c.db)
-	c.rateLimiter = lazy.Ratelimiter(c.db)
+
+	c.scheme = lazy.Scheme(c.db, c.schemeOptions)
+
+	c.scripting = lazy.Scripting(c.db, c.scriptingOptions)
+
+	c.coordination = lazy.Coordination(c.db, c.coordinationOptions)
+
+	c.ratelimiter = lazy.Ratelimiter(c.db, c.ratelimiterOptions)
+
 	return c, nil
 }

coordination/config.go

@@ -51,7 +51,7 @@ func (t RatelimiterCountersMode) String() string {
 	}
 }
 
-type Config struct {
+type NodeConfig struct {
 	Path                     string
 	SelfCheckPeriodMillis    uint32
 	SessionGracePeriodMillis uint32

coordination/config/config.go

@@ -0,0 +1,33 @@
+package config
+
+import (
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
+)
+
+type Config interface {
+	Trace() trace.Coordination
+}
+
+type config struct {
+	trace trace.Coordination
+}
+
+func (c *config) Trace() trace.Coordination {
+	return c.trace
+}
+
+type Option func(c *config)
+
+func WithTrace(trace trace.Coordination) Option {
+	return func(c *config) {
+		c.trace = trace
+	}
+}
+
+func New(opts ...Option) Config {
+	c := &config{}
+	for _, o := range opts {
+		o(c)
+	}
+	return c
+}

coordination/coordination.go

@@ -10,8 +10,8 @@ import (
 type Client interface {
 	closer.Closer
 
-	CreateNode(ctx context.Context, path string, config Config) (err error)
-	AlterNode(ctx context.Context, path string, config Config) (err error)
+	CreateNode(ctx context.Context, path string, config NodeConfig) (err error)
+	AlterNode(ctx context.Context, path string, config NodeConfig) (err error)
 	DropNode(ctx context.Context, path string) (err error)
-	DescribeNode(ctx context.Context, path string) (_ *scheme.Entry, _ *Config, err error)
+	DescribeNode(ctx context.Context, path string) (_ *scheme.Entry, _ *NodeConfig, err error)
 }