ING-1203: Refactored KV client management

Author: brett19

agent.go

@@ -19,13 +19,11 @@ import (
 var buildVersion string = buildversion.GetVersion("github.com/couchbase/gocbcorex")
 
 type agentState struct {
-	bucket             string
-	tlsConfig          *tls.Config
-	authenticator      Authenticator
-	numPoolConnections uint
-	httpTransport      *http.Transport
+	bucket        string
+	tlsConfig     *tls.Config
+	authenticator Authenticator
+	httpTransport *http.Transport
 
-	lastClients  map[string]*KvClientConfig
 	latestConfig *ParsedConfig
 }
 
@@ -37,7 +35,8 @@ type Agent struct {
 	state agentState
 
 	cfgWatcher  ConfigWatcher
-	connMgr     KvClientManager
+	connMgr     KvEndpointClientManager
+	mconnMgr    MultiKvEndpointClientManager
 	collections CollectionResolver
 	retries     RetryManager
 	vbRouter    VbucketRouter
@@ -148,12 +147,11 @@ func CreateAgent(ctx context.Context, opts AgentOptions) (*Agent, error) {
 		networkType: networkType,
 
 		state: agentState{
-			bucket:             opts.BucketName,
-			tlsConfig:          opts.TLSConfig,
-			authenticator:      opts.Authenticator,
-			numPoolConnections: connectionPoolSize,
-			latestConfig:       bootstrapConfig,
-			httpTransport:      httpTransport,
+			bucket:        opts.BucketName,
+			tlsConfig:     opts.TLSConfig,
+			authenticator: opts.Authenticator,
+			latestConfig:  bootstrapConfig,
+			httpTransport: httpTransport,
 		},
 	}
 	if opts.RetryManager == nil {
@@ -164,20 +162,29 @@ func CreateAgent(ctx context.Context, opts AgentOptions) (*Agent, error) {
 
 	agentComponentConfigs := agent.genAgentComponentConfigsLocked()
 
-	connMgr, err := NewKvClientManager(&KvClientManagerConfig{
-		NumPoolConnections: agent.state.numPoolConnections,
-		Clients:            agentComponentConfigs.KvClientManagerClients,
-	}, &KvClientManagerOptions{
-		Logger: agent.logger.Named("client-manager"),
+	mconnMgr, err := NewMultiKvEndpointClientManager(&MultiKvEndpointClientManagerOptions{
+		Logger:    agent.logger.Named("multi-client-manager"),
+		Endpoints: agentComponentConfigs.KvTargets,
+		Auth:      agentComponentConfigs.KvAuth,
+	})
+	if err != nil {
+		return nil, handleAgentCreateErr(err)
+	}
+	agent.mconnMgr = mconnMgr
+
+	connMgr, err := mconnMgr.NewManager(NewManagerOptions{
+		NumPoolConnections: connectionPoolSize,
+		OnDemandConnect:    false,
+		SelectedBucket:     agentComponentConfigs.KvSelectedBucket,
 	})
 	if err != nil {
 		return nil, handleAgentCreateErr(err)
 	}
 	agent.connMgr = connMgr
 
 	coreCollections, err := NewCollectionResolverMemd(&CollectionResolverMemdOptions{
-		Logger:  agent.logger,
-		ConnMgr: agent.connMgr,
+		Logger:       agent.logger,
+		ConnProvider: agent.connMgr,
 	})
 	if err != nil {
 		return nil, handleAgentCreateErr(err)
@@ -213,9 +220,9 @@ func CreateAgent(ctx context.Context, opts AgentOptions) (*Agent, error) {
 		configWatcher, err := NewConfigWatcherMemd(
 			&agentComponentConfigs.ConfigWatcherMemdConfig,
 			&ConfigWatcherMemdOptions{
-				Logger:          logger.Named("memd-config-watcher"),
-				KvClientManager: connMgr,
-				PollingPeriod:   2500 * time.Millisecond,
+				Logger:         logger.Named("memd-config-watcher"),
+				ClientProvider: connMgr,
+				PollingPeriod:  2500 * time.Millisecond,
 			},
 		)
 		if err != nil {
@@ -258,12 +265,12 @@ func CreateAgent(ctx context.Context, opts AgentOptions) (*Agent, error) {
 	}
 
 	agent.crud = &CrudComponent{
-		logger:      agent.logger,
-		collections: agent.collections,
-		retries:     consistencyRetryMgr,
-		connManager: agent.connMgr,
-		nmvHandler:  &agentNmvHandler{agent},
-		vbs:         agent.vbRouter,
+		logger:       agent.logger,
+		collections:  agent.collections,
+		retries:      consistencyRetryMgr,
+		connProvider: agent.connMgr,
+		nmvHandler:   &agentNmvHandler{agent},
+		vbs:          agent.vbRouter,
 		compression: &CompressionManagerDefault{
 			disableCompression:   !useCompression,
 			compressionMinSize:   compressionMinSize,
@@ -370,8 +377,8 @@ func (agent *Agent) NumVbuckets() int {
 }
 
 func (agent *Agent) Close() error {
-	if err := agent.connMgr.Close(); err != nil {
-		agent.logger.Debug("Failed to close conn mgr", zap.Error(err))
+	if err := agent.mconnMgr.Close(); err != nil {
+		agent.logger.Debug("Failed to close multi conn mgr", zap.Error(err))
 	}
 
 	agent.cfgWatcherCancel()
@@ -412,41 +419,23 @@ func (agent *Agent) updateStateLocked() {
 	// the routing table.  Then go back and remove the old entries from
 	// the connection manager list.
 
-	oldClients := make(map[string]*KvClientConfig)
-	if agent.state.lastClients != nil {
-		for clientName, client := range agent.state.lastClients {
-			oldClients[clientName] = client
-		}
-	}
-	for clientName, client := range agentComponentConfigs.KvClientManagerClients {
-		if oldClients[clientName] == nil {
-			oldClients[clientName] = client
-		}
-	}
-
-	err := agent.connMgr.Reconfigure(&KvClientManagerConfig{
-		NumPoolConnections: agent.state.numPoolConnections,
-		Clients:            oldClients,
-	})
+	err := agent.mconnMgr.UpdateEndpoints(agentComponentConfigs.KvTargets, true)
 	if err != nil {
-		agent.logger.Error("failed to reconfigure connection manager (old clients)", zap.Error(err))
+		agent.logger.Error("failed to add-only update kv connection manager endpoint", zap.Error(err))
 	}
 
 	agent.vbRouter.UpdateRoutingInfo(agentComponentConfigs.VbucketRoutingInfo)
 
 	if agent.memdCfgWatcher != nil {
-		err = agent.memdCfgWatcher.Reconfigure(&agentComponentConfigs.ConfigWatcherMemdConfig)
+		err := agent.memdCfgWatcher.Reconfigure(&agentComponentConfigs.ConfigWatcherMemdConfig)
 		if err != nil {
 			agent.logger.Error("failed to reconfigure memd config watcher component", zap.Error(err))
 		}
 	}
 
-	err = agent.connMgr.Reconfigure(&KvClientManagerConfig{
-		NumPoolConnections: agent.state.numPoolConnections,
-		Clients:            agentComponentConfigs.KvClientManagerClients,
-	})
+	err = agent.mconnMgr.UpdateEndpoints(agentComponentConfigs.KvTargets, false)
 	if err != nil {
-		agent.logger.Error("failed to reconfigure connection manager (updated clients)", zap.Error(err))
+		agent.logger.Error("failed to update kv connection manager endpoint", zap.Error(err))
 	}
 
 	err = agent.query.Reconfigure(&agentComponentConfigs.QueryComponentConfig)

authenticator.go

@@ -25,3 +25,23 @@ func (a *PasswordAuthenticator) GetCredentials(
 ) (string, string, error) {
 	return a.Username, a.Password, nil
 }
+
+type kvClientAuth struct {
+	Authenticator Authenticator
+}
+
+var _ KvClientAuth = (*kvClientAuth)(nil)
+
+func (v *kvClientAuth) GetAuth(address string) (username, password string, clientCert *tls.Certificate, err error) {
+	clientCert, err = v.Authenticator.GetClientCertificate(ServiceTypeMemd, address)
+	if err != nil {
+		return "", "", nil, err
+	}
+
+	username, password, err = v.Authenticator.GetCredentials(ServiceTypeMemd, address)
+	if err != nil {
+		return "", "", nil, err
+	}
+
+	return username, password, clientCert, nil
+}

collectionresolver_memd.go

@@ -8,13 +8,13 @@ import (
 )
 
 type CollectionResolverMemdOptions struct {
-	Logger  *zap.Logger
-	ConnMgr KvClientManager
+	Logger       *zap.Logger
+	ConnProvider KvClientProvider
 }
 
 type CollectionResolverMemd struct {
-	logger  *zap.Logger
-	connMgr KvClientManager
+	logger       *zap.Logger
+	connProvider KvClientProvider
 }
 
 var _ CollectionResolver = (*CollectionResolverMemd)(nil)
@@ -25,16 +25,16 @@ func NewCollectionResolverMemd(opts *CollectionResolverMemdOptions) (*Collection
 	}
 
 	return &CollectionResolverMemd{
-		logger:  loggerOrNop(opts.Logger),
-		connMgr: opts.ConnMgr,
+		logger:       loggerOrNop(opts.Logger),
+		connProvider: opts.ConnProvider,
 	}, nil
 }
 
 func (cr *CollectionResolverMemd) ResolveCollectionID(
 	ctx context.Context, scopeName, collectionName string,
 ) (collectionId uint32, manifestRev uint64, err error) {
-	resp, err := OrchestrateRandomMemdClient(
-		ctx, cr.connMgr,
+	resp, err := OrchestrateKvClient(
+		ctx, cr.connProvider,
 		func(client KvClient) (*memdx.GetCollectionIDResponse, error) {
 			return client.GetCollectionID(ctx, &memdx.GetCollectionIDRequest{
 				ScopeName:      scopeName,

componentconfigs.go

@@ -9,7 +9,9 @@ import (
 type AgentComponentConfigs struct {
 	ConfigWatcherHttpConfig         ConfigWatcherHttpConfig
 	ConfigWatcherMemdConfig         ConfigWatcherMemdConfig
-	KvClientManagerClients          map[string]*KvClientConfig
+	KvTargets                       map[string]KvTarget
+	KvAuth                          KvClientAuth
+	KvSelectedBucket                string
 	VbucketRoutingInfo              *VbucketRoutingInfo
 	QueryComponentConfig            QueryComponentConfig
 	MgmtComponentConfig             MgmtComponentConfig
@@ -83,14 +85,20 @@ func GenerateComponentConfigsFromConfig(
 		}
 	}
 
-	clients := make(map[string]*KvClientConfig)
+	var kvTlsConfig *KvTargetTlsConfig
+	if tlsConfig != nil {
+		kvTlsConfig = &KvTargetTlsConfig{
+			InsecureSkipVerify: tlsConfig.InsecureSkipVerify,
+			RootCAs:            tlsConfig.RootCAs,
+			CipherSuites:       tlsConfig.CipherSuites,
+		}
+	}
+
+	kvTargets := make(map[string]KvTarget, len(kvDataHosts))
 	for nodeId, addr := range kvDataHosts {
-		clients[nodeId] = &KvClientConfig{
-			Address:        addr,
-			TlsConfig:      tlsConfig,
-			ClientName:     clientName,
-			SelectedBucket: bucketName,
-			Authenticator:  authenticator,
+		kvTargets[nodeId] = KvTarget{
+			Address:   addr,
+			TLSConfig: kvTlsConfig,
 		}
 	}
 
@@ -110,7 +118,11 @@ func GenerateComponentConfigsFromConfig(
 		ConfigWatcherMemdConfig: ConfigWatcherMemdConfig{
 			Endpoints: kvDataNodeIds,
 		},
-		KvClientManagerClients: clients,
+		KvTargets: kvTargets,
+		KvAuth: &kvClientAuth{
+			Authenticator: authenticator,
+		},
+		KvSelectedBucket: bucketName,
 		VbucketRoutingInfo: &VbucketRoutingInfo{
 			VbMap:      config.VbucketMap,
 			ServerList: kvDataNodeIds,

configwatcher_memd.go

@@ -16,29 +16,29 @@ type ConfigWatcherMemdConfig struct {
 }
 
 type ConfigWatcherMemdOptions struct {
-	Logger          *zap.Logger
-	KvClientManager KvClientManager
-	PollingPeriod   time.Duration
+	Logger         *zap.Logger
+	ClientProvider KvEndpointClientProvider
+	PollingPeriod  time.Duration
 }
 
 type configWatcherMemdState struct {
 	endpoints []string
 }
 
 type ConfigWatcherMemd struct {
-	logger          *zap.Logger
-	kvClientManager KvClientManager
-	pollingPeriod   time.Duration
+	logger         *zap.Logger
+	clientProvider KvEndpointClientProvider
+	pollingPeriod  time.Duration
 
 	lock  sync.Mutex
 	state *configWatcherMemdState
 }
 
 func NewConfigWatcherMemd(config *ConfigWatcherMemdConfig, opts *ConfigWatcherMemdOptions) (*ConfigWatcherMemd, error) {
 	return &ConfigWatcherMemd{
-		logger:          opts.Logger,
-		kvClientManager: opts.KvClientManager,
-		pollingPeriod:   opts.PollingPeriod,
+		logger:         opts.Logger,
+		clientProvider: opts.ClientProvider,
+		pollingPeriod:  opts.PollingPeriod,
 		state: &configWatcherMemdState{
 			endpoints: config.Endpoints,
 		},
@@ -57,10 +57,10 @@ func (w *ConfigWatcherMemd) Reconfigure(config *ConfigWatcherMemdConfig) error {
 func configWatcherMemd_pollOne(
 	ctx context.Context,
 	logger *zap.Logger,
-	kvClientManager KvClientManager,
+	clientProvider KvEndpointClientProvider,
 	endpoint string,
 ) (*ParsedConfig, error) {
-	client, err := kvClientManager.GetClient(ctx, endpoint)
+	client, err := clientProvider.GetEndpointClient(ctx, endpoint)
 	if err != nil {
 		return nil, err
 	}
@@ -145,7 +145,7 @@ func (w *ConfigWatcherMemd) watchThread(ctx context.Context, outCh chan<- *Parse
 		parsedConfig, err := configWatcherMemd_pollOne(
 			pollCtx,
 			w.logger,
-			w.kvClientManager,
+			w.clientProvider,
 			endpoint)
 		cancel()
 		if err != nil {