diff --git a/pkg/mcs/discovery/register.go b/pkg/mcs/discovery/register.go index 1610c1a25a0..c6d2f111c8d 100644 --- a/pkg/mcs/discovery/register.go +++ b/pkg/mcs/discovery/register.go @@ -32,6 +32,8 @@ import ( // DefaultLeaseInSeconds is the default lease time in seconds. const DefaultLeaseInSeconds = 5 +const maxKeepAliveRetryInterval = time.Duration(DefaultLeaseInSeconds) * time.Second + // ServiceRegister is used to register the service to etcd. type ServiceRegister struct { ctx context.Context @@ -88,7 +90,7 @@ func (sr *ServiceRegister) Register() error { } func (sr *ServiceRegister) renewKeepalive() <-chan *clientv3.LeaseKeepAliveResponse { - t := time.NewTicker(time.Duration(sr.ttl) * time.Second / 2) + t := time.NewTicker(sr.keepAliveRetryInterval()) defer t.Stop() for { select { @@ -111,6 +113,14 @@ func (sr *ServiceRegister) renewKeepalive() <-chan *clientv3.LeaseKeepAliveRespo } } +func (sr *ServiceRegister) keepAliveRetryInterval() time.Duration { + interval := time.Duration(sr.ttl) * time.Second / 2 + if interval > maxKeepAliveRetryInterval { + return maxKeepAliveRetryInterval + } + return interval +} + func (sr *ServiceRegister) putWithTTL() (clientv3.LeaseID, error) { ctx, cancel := context.WithTimeout(sr.ctx, etcdutil.DefaultRequestTimeout) defer cancel() @@ -120,8 +130,12 @@ func (sr *ServiceRegister) putWithTTL() (clientv3.LeaseID, error) { // Deregister deregisters the service from etcd. func (sr *ServiceRegister) Deregister() error { sr.cancel() - ctx, cancel := context.WithTimeout(context.Background(), time.Duration(sr.ttl)*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), deregisterTimeout()) defer cancel() _, err := sr.cli.Delete(ctx, sr.key) return err } + +func deregisterTimeout() time.Duration { + return etcdutil.DefaultRequestTimeout +} diff --git a/pkg/mcs/discovery/register_test.go b/pkg/mcs/discovery/register_test.go index ec7dd73f3c7..7afa4174e77 100644 --- a/pkg/mcs/discovery/register_test.go +++ b/pkg/mcs/discovery/register_test.go @@ -93,6 +93,14 @@ func TestRegister(t *testing.T) { etcd.Close() } +func TestServiceRegisterLongTTLDoesNotDelayRecoveryOperations(t *testing.T) { + re := require.New(t) + sr := NewServiceRegister(context.Background(), nil, "test_service", "127.0.0.1:1", "127.0.0.1:1", 3600) + + re.Equal(time.Duration(DefaultLeaseInSeconds)*time.Second, sr.keepAliveRetryInterval()) + re.Equal(etcdutil.DefaultRequestTimeout, deregisterTimeout()) +} + func getKeyAfterLeaseExpired(ctx context.Context, re *require.Assertions, client *clientv3.Client, key string) string { time.Sleep(DefaultLeaseInSeconds * time.Second) // ensure that the lease is expired time.Sleep(500 * time.Millisecond) // wait for the etcd to clean up the expired keys diff --git a/pkg/mcs/resourcemanager/server/server.go b/pkg/mcs/resourcemanager/server/server.go index d0410852556..88ac713daf5 100644 --- a/pkg/mcs/resourcemanager/server/server.go +++ b/pkg/mcs/resourcemanager/server/server.go @@ -323,6 +323,11 @@ func (s *Server) GetConfig() *Config { return s.cfg } +// GetLeaderLease returns the configured leader lease in seconds. +func (s *Server) GetLeaderLease() int64 { + return s.cfg.LeaderLease +} + // GetParticipant returns the participant. func (s *Server) GetParticipant() *member.Participant { return s.participant diff --git a/pkg/mcs/router/server/server.go b/pkg/mcs/router/server/server.go index e612738c428..f67603291c8 100644 --- a/pkg/mcs/router/server/server.go +++ b/pkg/mcs/router/server/server.go @@ -94,6 +94,13 @@ func (s *Server) GetAddr() string { return s.cfg.ListenAddr } +// GetLeaderLease returns the configured leader lease in seconds. The router +// service has no lease config, so it returns 0 and the registry falls back to +// discovery.DefaultLeaseInSeconds. +func (*Server) GetLeaderLease() int64 { + return 0 +} + // GetAdvertiseListenAddr returns the advertise address of the server. func (s *Server) GetAdvertiseListenAddr() string { return s.cfg.AdvertiseListenAddr diff --git a/pkg/mcs/scheduling/server/server.go b/pkg/mcs/scheduling/server/server.go index ce621211a46..8ed73e3e79c 100644 --- a/pkg/mcs/scheduling/server/server.go +++ b/pkg/mcs/scheduling/server/server.go @@ -634,6 +634,11 @@ func (s *Server) GetConfig() *config.Config { return cfg } +// GetLeaderLease returns the configured leader lease in seconds. +func (s *Server) GetLeaderLease() int64 { + return s.cfg.LeaderLease +} + // CreateServer creates the Server func CreateServer(ctx context.Context, cfg *config.Config) *Server { svr := &Server{ diff --git a/pkg/mcs/tso/server/server.go b/pkg/mcs/tso/server/server.go index 8efe2a65ce9..7079bef51a8 100644 --- a/pkg/mcs/tso/server/server.go +++ b/pkg/mcs/tso/server/server.go @@ -342,6 +342,11 @@ func (s *Server) GetConfig() *Config { return s.cfg } +// GetLeaderLease returns the configured leader lease in seconds. +func (s *Server) GetLeaderLease() int64 { + return s.cfg.GetLease() +} + // GetTLSConfig gets the security config. func (s *Server) GetTLSConfig() *grpcutil.TLSConfig { return &s.cfg.Security.TLSConfig diff --git a/pkg/mcs/utils/util.go b/pkg/mcs/utils/util.go index 40581404006..ec7b9e187de 100644 --- a/pkg/mcs/utils/util.go +++ b/pkg/mcs/utils/util.go @@ -104,6 +104,11 @@ type server interface { diagnosticspb.DiagnosticsServer StartTimestamp() int64 Name() string + // GetLeaderLease returns the configured leader lease in seconds. It is + // reused as the service registry lease TTL. Services without a lease + // config should return 0, in which case discovery.DefaultLeaseInSeconds + // is used. + GetLeaderLease() int64 } // InitClient initializes the etcd and http clients. @@ -295,9 +300,12 @@ func Register(s server, serviceName string) (*discovery.ServiceRegistryEntry, *d if err != nil { return nil, nil, err } + // Reuse the configured leader lease as the registry lease TTL, but never + // go below discovery.DefaultLeaseInSeconds to keep the registry entry + // stable. + lease := max(s.GetLeaderLease(), discovery.DefaultLeaseInSeconds) serviceRegister := discovery.NewServiceRegister(s.Context(), s.GetEtcdClient(), - serviceName, s.GetAdvertiseListenAddr(), serializedEntry, - discovery.DefaultLeaseInSeconds) + serviceName, s.GetAdvertiseListenAddr(), serializedEntry, lease) if err := serviceRegister.Register(); err != nil { log.Error("failed to register the service", zap.String("service-name", serviceName), errs.ZapError(err)) return nil, nil, err