Skip to content

Commit

Permalink
Rename MetadataManager to DomainManager
Browse files Browse the repository at this point in the history
  • Loading branch information
@longquanzheng
longquanzheng authored Jun 6, 2021
1 parent 3eda53b commit 0b058c0
Show file tree
Hide file tree
Showing 32 changed files with 190 additions and 190 deletions.
12 changes: 6 additions & 6 deletions common/cache/domainCache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ type (
shutdownChan chan struct{}
cacheNameToID *atomic.Value
cacheByID *atomic.Value
metadataMgr persistence.MetadataManager
domainManager persistence.DomainManager
clusterMetadata cluster.Metadata
timeSource clock.TimeSource
metricsClient metrics.Client
Expand Down Expand Up @@ -145,7 +145,7 @@ type (

// NewDomainCache creates a new instance of cache for holding onto domain information to reduce the load on persistence
func NewDomainCache(
metadataMgr persistence.MetadataManager,
domainManager persistence.DomainManager,
clusterMetadata cluster.Metadata,
metricsClient metrics.Client,
logger log.Logger,
Expand All @@ -156,7 +156,7 @@ func NewDomainCache(
shutdownChan: make(chan struct{}),
cacheNameToID: &atomic.Value{},
cacheByID: &atomic.Value{},
metadataMgr: metadataMgr,
domainManager: domainManager,
clusterMetadata: clusterMetadata,
timeSource: clock.NewRealTimeSource(),
metricsClient: metricsClient,
Expand Down Expand Up @@ -450,7 +450,7 @@ func (c *domainCache) refreshDomainsLocked() error {
// this can guarantee that domains in the cache are not updated more than metadata record
ctx, cancel := context.WithTimeout(context.Background(), domainCachePersistenceTimeout)
defer cancel()
metadata, err := c.metadataMgr.GetMetadata(ctx)
metadata, err := c.domainManager.GetMetadata(ctx)
if err != nil {
return err
}
Expand All @@ -463,7 +463,7 @@ func (c *domainCache) refreshDomainsLocked() error {
for continuePage {
ctx, cancel := context.WithTimeout(context.Background(), domainCachePersistenceTimeout)
request.NextPageToken = token
response, err := c.metadataMgr.ListDomains(ctx, request)
response, err := c.domainManager.ListDomains(ctx, request)
cancel()
if err != nil {
return err
Expand Down Expand Up @@ -541,7 +541,7 @@ func (c *domainCache) checkDomainExists(
ctx, cancel := context.WithTimeout(context.Background(), domainCachePersistenceTimeout)
defer cancel()

_, err := c.metadataMgr.GetDomain(ctx, &persistence.GetDomainRequest{Name: name, ID: id})
_, err := c.domainManager.GetDomain(ctx, &persistence.GetDomainRequest{Name: name, ID: id})
return err
}

Expand Down
24 changes: 12 additions & 12 deletions common/domain/failover_watcher.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,11 +57,11 @@ type (
refreshJitter dynamicconfig.FloatPropertyFn
retryPolicy backoff.RetryPolicy

metadataMgr persistence.MetadataManager
domainCache cache.DomainCache
timeSource clock.TimeSource
metrics metrics.Client
logger log.Logger
domainManager persistence.DomainManager
domainCache cache.DomainCache
timeSource clock.TimeSource
metrics metrics.Client
logger log.Logger
}
)

Expand All @@ -70,7 +70,7 @@ var _ FailoverWatcher = (*failoverWatcherImpl)(nil)
// NewFailoverWatcher initializes domain failover processor
func NewFailoverWatcher(
domainCache cache.DomainCache,
metadataMgr persistence.MetadataManager,
domainManager persistence.DomainManager,
timeSource clock.TimeSource,
refreshInterval dynamicconfig.DurationPropertyFn,
refreshJitter dynamicconfig.FloatPropertyFn,
Expand All @@ -90,7 +90,7 @@ func NewFailoverWatcher(
refreshJitter: refreshJitter,
retryPolicy: retryPolicy,
domainCache: domainCache,
metadataMgr: metadataMgr,
domainManager: domainManager,
timeSource: timeSource,
metrics: metrics,
logger: logger,
Expand Down Expand Up @@ -157,7 +157,7 @@ func (p *failoverWatcherImpl) handleFailoverTimeout(
domainID := domain.GetInfo().ID
// force failover the domain without setting the failover timeout
if err := CleanPendingActiveState(
p.metadataMgr,
p.domainManager,
domainID,
domain.GetFailoverVersion(),
p.retryPolicy,
Expand All @@ -170,7 +170,7 @@ func (p *failoverWatcherImpl) handleFailoverTimeout(

// CleanPendingActiveState removes the pending active state from the domain
func CleanPendingActiveState(
metadataMgr persistence.MetadataManager,
domainManager persistence.DomainManager,
domainID string,
failoverVersion int64,
policy backoff.RetryPolicy,
Expand All @@ -180,13 +180,13 @@ func CleanPendingActiveState(
// this version can be regarded as the lock on the v2 domain table
// and since we do not know which table will return the domain afterwards
// this call has to be made
metadata, err := metadataMgr.GetMetadata(context.Background())
metadata, err := domainManager.GetMetadata(context.Background())
if err != nil {
return err
}
notificationVersion := metadata.NotificationVersion

getResponse, err := metadataMgr.GetDomain(context.Background(), &persistence.GetDomainRequest{ID: domainID})
getResponse, err := domainManager.GetDomain(context.Background(), &persistence.GetDomainRequest{ID: domainID})
if err != nil {
return err
}
Expand All @@ -207,7 +207,7 @@ func CleanPendingActiveState(
NotificationVersion: notificationVersion,
}
op := func() error {
return metadataMgr.UpdateDomain(context.Background(), updateReq)
return domainManager.UpdateDomain(context.Background(), updateReq)
}
if err := backoff.Retry(
op,
Expand Down
26 changes: 13 additions & 13 deletions common/domain/handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ type (

// handlerImpl is the domain operation handler implementation
handlerImpl struct {
metadataMgr persistence.MetadataManager
domainManager persistence.DomainManager
clusterMetadata cluster.Metadata
domainReplicator Replicator
domainAttrValidator *AttrValidatorImpl
Expand All @@ -99,7 +99,7 @@ var _ Handler = (*handlerImpl)(nil)
func NewHandler(
config Config,
logger log.Logger,
metadataMgr persistence.MetadataManager,
domainManager persistence.DomainManager,
clusterMetadata cluster.Metadata,
domainReplicator Replicator,
archivalMetadata archiver.ArchivalMetadata,
Expand All @@ -108,7 +108,7 @@ func NewHandler(
) Handler {
return &handlerImpl{
logger: logger,
metadataMgr: metadataMgr,
domainManager: domainManager,
clusterMetadata: clusterMetadata,
domainReplicator: domainReplicator,
domainAttrValidator: newAttrValidator(clusterMetadata, int32(config.MinRetentionDays())),
Expand Down Expand Up @@ -137,7 +137,7 @@ func (d *handlerImpl) RegisterDomain(
}

// first check if the name is already registered as the local domain
_, err := d.metadataMgr.GetDomain(ctx, &persistence.GetDomainRequest{Name: registerRequest.GetName()})
_, err := d.domainManager.GetDomain(ctx, &persistence.GetDomainRequest{Name: registerRequest.GetName()})
switch err.(type) {
case nil:
// domain already exists, cannot proceed
Expand Down Expand Up @@ -256,7 +256,7 @@ func (d *handlerImpl) RegisterDomain(
LastUpdatedTime: d.timeSource.Now().UnixNano(),
}

domainResponse, err := d.metadataMgr.CreateDomain(ctx, domainRequest)
domainResponse, err := d.domainManager.CreateDomain(ctx, domainRequest)
if err != nil {
return err
}
Expand Down Expand Up @@ -297,7 +297,7 @@ func (d *handlerImpl) ListDomains(
pageSize = int(listRequest.GetPageSize())
}

resp, err := d.metadataMgr.ListDomains(ctx, &persistence.ListDomainsRequest{
resp, err := d.domainManager.ListDomains(ctx, &persistence.ListDomainsRequest{
PageSize: pageSize,
NextPageToken: listRequest.NextPageToken,
})
Expand Down Expand Up @@ -335,7 +335,7 @@ func (d *handlerImpl) DescribeDomain(
Name: describeRequest.GetName(),
ID: describeRequest.GetUUID(),
}
resp, err := d.metadataMgr.GetDomain(ctx, req)
resp, err := d.domainManager.GetDomain(ctx, req)
if err != nil {
return nil, err
}
Expand All @@ -358,12 +358,12 @@ func (d *handlerImpl) UpdateDomain(
// this version can be regarded as the lock on the v2 domain table
// and since we do not know which table will return the domain afterwards
// this call has to be made
metadata, err := d.metadataMgr.GetMetadata(ctx)
metadata, err := d.domainManager.GetMetadata(ctx)
if err != nil {
return nil, err
}
notificationVersion := metadata.NotificationVersion
getResponse, err := d.metadataMgr.GetDomain(ctx, &persistence.GetDomainRequest{Name: updateRequest.GetName()})
getResponse, err := d.domainManager.GetDomain(ctx, &persistence.GetDomainRequest{Name: updateRequest.GetName()})
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -534,7 +534,7 @@ func (d *handlerImpl) UpdateDomain(
LastUpdatedTime: lastUpdatedTime.UnixNano(),
NotificationVersion: notificationVersion,
}
err = d.metadataMgr.UpdateDomain(ctx, updateReq)
err = d.domainManager.UpdateDomain(ctx, updateReq)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -579,12 +579,12 @@ func (d *handlerImpl) DeprecateDomain(
// this version can be regarded as the lock on the v2 domain table
// and since we do not know which table will return the domain afterwards
// this call has to be made
metadata, err := d.metadataMgr.GetMetadata(ctx)
metadata, err := d.domainManager.GetMetadata(ctx)
if err != nil {
return err
}
notificationVersion := metadata.NotificationVersion
getResponse, err := d.metadataMgr.GetDomain(ctx, &persistence.GetDomainRequest{Name: deprecateRequest.GetName()})
getResponse, err := d.domainManager.GetDomain(ctx, &persistence.GetDomainRequest{Name: deprecateRequest.GetName()})
if err != nil {
return err
}
Expand All @@ -608,7 +608,7 @@ func (d *handlerImpl) DeprecateDomain(
LastUpdatedTime: d.timeSource.Now().UnixNano(),
NotificationVersion: notificationVersion,
}
err = d.metadataMgr.UpdateDomain(ctx, updateReq)
err = d.domainManager.UpdateDomain(ctx, updateReq)
if err != nil {
return err
}
Expand Down
6 changes: 3 additions & 3 deletions common/domain/handler_GlobalDomainDisabled_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ type (

minRetentionDays int
maxBadBinaryCount int
metadataMgr persistence.MetadataManager
domainManager persistence.DomainManager
mockProducer *mocks.KafkaProducer
mockDomainReplicator Replicator
archivalMetadata archiver.ArchivalMetadata
Expand Down Expand Up @@ -87,7 +87,7 @@ func (s *domainHandlerGlobalDomainDisabledSuite) SetupTest() {
dcCollection := dc.NewCollection(dc.NewNopClient(), logger)
s.minRetentionDays = 1
s.maxBadBinaryCount = 10
s.metadataMgr = s.TestBase.MetadataManager
s.domainManager = s.TestBase.DomainManager
s.mockProducer = &mocks.KafkaProducer{}
s.mockDomainReplicator = NewDomainReplicator(s.mockProducer, logger)
s.archivalMetadata = archiver.NewArchivalMetadata(
Expand All @@ -107,7 +107,7 @@ func (s *domainHandlerGlobalDomainDisabledSuite) SetupTest() {
s.handler = NewHandler(
domainConfig,
logger,
s.metadataMgr,
s.domainManager,
s.ClusterMetadata,
s.mockDomainReplicator,
s.archivalMetadata,
Expand Down
8 changes: 4 additions & 4 deletions common/domain/handler_GlobalDomainEnabled_MasterCluster_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ type (

minRetentionDays int
maxBadBinaryCount int
metadataMgr persistence.MetadataManager
domainManager persistence.DomainManager
mockProducer *mocks.KafkaProducer
mockDomainReplicator Replicator
archivalMetadata archiver.ArchivalMetadata
Expand Down Expand Up @@ -88,7 +88,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) SetupTest() {
dcCollection := dc.NewCollection(dc.NewNopClient(), logger)
s.minRetentionDays = 1
s.maxBadBinaryCount = 10
s.metadataMgr = s.TestBase.MetadataManager
s.domainManager = s.TestBase.DomainManager
s.mockProducer = &mocks.KafkaProducer{}
s.mockDomainReplicator = NewDomainReplicator(s.mockProducer, logger)
s.archivalMetadata = archiver.NewArchivalMetadata(
Expand All @@ -108,7 +108,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) SetupTest() {
s.handler = NewHandler(
domainConfig,
logger,
s.metadataMgr,
s.domainManager,
s.ClusterMetadata,
s.mockDomainReplicator,
s.archivalMetadata,
Expand Down Expand Up @@ -870,7 +870,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestUpdateDomain_C
s.handler = NewHandler(
domainConfig,
loggerimpl.NewNopLogger(),
s.metadataMgr,
s.domainManager,
s.ClusterMetadata,
s.mockDomainReplicator,
s.archivalMetadata,
Expand Down
18 changes: 9 additions & 9 deletions common/domain/handler_GlobalDomainEnabled_NotMasterCluster_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ type (

minRetentionDays int
maxBadBinaryCount int
metadataMgr persistence.MetadataManager
domainManager persistence.DomainManager
mockProducer *mocks.KafkaProducer
mockDomainReplicator Replicator
archivalMetadata archiver.ArchivalMetadata
Expand Down Expand Up @@ -88,7 +88,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) SetupTest() {
dcCollection := dc.NewCollection(dc.NewNopClient(), logger)
s.minRetentionDays = 1
s.maxBadBinaryCount = 10
s.metadataMgr = s.TestBase.MetadataManager
s.domainManager = s.TestBase.DomainManager
s.mockProducer = &mocks.KafkaProducer{}
s.mockDomainReplicator = NewDomainReplicator(s.mockProducer, logger)
s.archivalMetadata = archiver.NewArchivalMetadata(
Expand All @@ -108,7 +108,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) SetupTest() {
s.handler = NewHandler(
domainConfig,
logger,
s.metadataMgr,
s.domainManager,
s.ClusterMetadata,
s.mockDomainReplicator,
s.archivalMetadata,
Expand Down Expand Up @@ -514,7 +514,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
data := map[string]string{"some random key": "some random value"}
isGlobalDomain := true

_, err := s.MetadataManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
_, err := s.DomainManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
Info: &persistence.DomainInfo{
ID: uuid.New(),
Name: domainName,
Expand Down Expand Up @@ -574,7 +574,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
data := map[string]string{"some random key": "some random value"}
isGlobalDomain := true

_, err := s.MetadataManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
_, err := s.DomainManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
Info: &persistence.DomainInfo{
ID: uuid.New(),
Name: domainName,
Expand Down Expand Up @@ -647,7 +647,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
data := map[string]string{"some random key": "some random value"}
isGlobalDomain := true

_, err := s.MetadataManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
_, err := s.DomainManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
Info: &persistence.DomainInfo{
ID: uuid.New(),
Name: domainName,
Expand Down Expand Up @@ -753,10 +753,10 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) setupLocalDomai
}

func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) setupGlobalDomainWithMetadataManager(domainName string) *types.DescribeDomainResponse {
return setupGlobalDomainWithMetadataManager(s.Suite, s.handler, s.ClusterMetadata, s.MetadataManager, domainName)
return setupGlobalDomainWithMetadataManager(s.Suite, s.handler, s.ClusterMetadata, s.DomainManager, domainName)
}

func setupGlobalDomainWithMetadataManager(s suite.Suite, handler *handlerImpl, clusterMetadata cluster.Metadata, metadataManager persistence.MetadataManager, domainName string) *types.DescribeDomainResponse {
func setupGlobalDomainWithMetadataManager(s suite.Suite, handler *handlerImpl, clusterMetadata cluster.Metadata, domainManager persistence.DomainManager, domainName string) *types.DescribeDomainResponse {
description := "some random description"
email := "some random email"
retention := int32(7)
Expand All @@ -776,7 +776,7 @@ func setupGlobalDomainWithMetadataManager(s suite.Suite, handler *handlerImpl, c
data := map[string]string{"some random key": "some random value"}
isGlobalDomain := true

_, err := metadataManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
_, err := domainManager.CreateDomain(context.Background(), &persistence.CreateDomainRequest{
Info: &persistence.DomainInfo{
ID: uuid.New(),
Name: domainName,
Expand Down
Loading

0 comments on commit 0b058c0

Please sign in to comment.