[PR] Support node registry by DNS [skywalking-banyandb]

["mrproliu (via GitHub)" <gi...@apache.org>]

mrproliu opened a new pull request, #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920

   1. Support enabling/disabling the DNS node registry. If disabled, it will still use `etcd` as storage. 
   2. Add a new gRPC protocol to support querying the self node info. 
   3. Wrap the oberverbility as a sub package(`service`) to avoid package dependency issue. 
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["wu-sheng (via GitHub)" <gi...@apache.org>]

wu-sheng commented on PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#issuecomment-3703162342

   > 1. Support enabling/disabling the DNS node registry. If disabled, it will still use `etcd` as storage.
   
   Please update document about service discovery, and indicate which is the default option.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["hanahmily (via GitHub)" <gi...@apache.org>]

hanahmily commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2659284442


##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)

Review Comment:
   Avoid defining the interface that is accessed within the same package.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()

Review Comment:
   The variable `lastSuccessMutex` appears to be redundant, as it is accessed only by the discovery loop.



##########
banyand/metadata/client.go:
##########
@@ -96,13 +108,35 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// DNS-based node discovery configuration
+	fs.BoolVar(&s.dnsRegistryEnabled, "node-registry-dns-enabled", false,

Review Comment:
   Use a single flag: "node-discovery-mode" with three options: "dns," "file," and "etcd." Replace "registry" with "discovery," which is more appropriate.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")

Review Comment:
   ```suggestion
   		svc.log.Info().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
   ```



##########
banyand/metadata/client.go:
##########
@@ -96,13 +108,35 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// DNS-based node discovery configuration
+	fs.BoolVar(&s.dnsRegistryEnabled, "node-registry-dns-enabled", false,
+		"Is DNS registry enabled")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-registry-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")

Review Comment:
   ```suggestion
   		"DNS SRV addresses for node discovery (e.g., banyandb.svc.cluster.local)")
   ```
   
   The "_grpc._tcp" part should be reserved and not exposed through the flag."



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)

Review Comment:
   You did not start the reloader. It cannot monitor the file unless it is activated.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	s.cacheMutex.Lock()
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			// update cache and notify the handlers
+			delete(s.nodeCache, addr)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("Node removed from cache (no longer in DNS)")
+
+			s.cacheMutex.Unlock()
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, false)
+			s.cacheMutex.Lock()
+		}
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()

Review Comment:
   Copilot is correct, fix this bug.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC

Review Comment:
   fix it.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	s.cacheMutex.Lock()
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			// update cache and notify the handlers
+			delete(s.nodeCache, addr)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("Node removed from cache (no longer in DNS)")
+
+			s.cacheMutex.Unlock()
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, false)
+			s.cacheMutex.Lock()
+		}
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// Record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting matadata
+	dialOpts, err := s.getTLSDialOptions()
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.ConnWithAuth(address, s.grpcTimeout, "", "", dialOpts...)

Review Comment:
   How to handle the basic auth(username/password) if the liaison node's authN is activated?



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)

Review Comment:
   Use "time.Timer" instead of "Sleep". Place the timer in a "select" block to ensure that the close event is not blocked by the sleep.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2656039263


##########
banyand/queue/sub/server.go:
##########
@@ -139,6 +144,27 @@ func (s *server) PreRun(_ context.Context) error {
 		s.log.Info().Str("certFile", s.certFile).Str("keyFile", s.keyFile).Msg("Initialized TLS reloader for queue server")
 	}
 
+	nodeVal := ctx.Value(common.ContextNodeKey)
+	roleVal := ctx.Value(common.ContextNodeRolesKey)
+	if nodeVal == nil || roleVal == nil {
+		s.log.Warn().Msg("node or role value not found in context")
+		return nil
+	}
+	nodeRoles := roleVal.([]databasev1.Role)
+	node := nodeVal.(common.Node)

Review Comment:
   Type assertion without checking may panic if the context value is not the expected type. Add type assertion checks or document that these values are guaranteed to be the correct types.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	s.cacheMutex.Lock()
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			// update cache and notify the handlers
+			delete(s.nodeCache, addr)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("Node removed from cache (no longer in DNS)")
+
+			s.cacheMutex.Unlock()
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, false)
+			s.cacheMutex.Lock()
+		}
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// Record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting matadata
+	dialOpts, err := s.getTLSDialOptions()
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.ConnWithAuth(address, s.grpcTimeout, "", "", dialOpts...)

Review Comment:
   Using grpchelper.ConnWithAuth with ctxTimeout parameter, but passing s.grpcTimeout instead. The connection should respect the context deadline from ctxTimeout, not pass a separate timeout duration.
   ```suggestion
   	conn, connErr := grpchelper.ConnWithAuth(address, ctxTimeout, "", "", dialOpts...)
   ```



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	s.cacheMutex.Lock()
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			// update cache and notify the handlers
+			delete(s.nodeCache, addr)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("Node removed from cache (no longer in DNS)")
+
+			s.cacheMutex.Unlock()
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, false)
+			s.cacheMutex.Lock()
+		}
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// Record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting matadata

Review Comment:
   Typo in comment: "matadata" should be "metadata"
   ```suggestion
   	// for TLS connections with other nodes to getting metadata
   ```



##########
banyand/metadata/client.go:
##########
@@ -96,13 +108,35 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// DNS-based node discovery configuration
+	fs.BoolVar(&s.dnsRegistryEnabled, "node-registry-dns-enabled", false,
+		"Is DNS registry enabled")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-registry-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")
+	fs.DurationVar(&s.dnsFetchInitInterval, "node-registry-dns-fetch-init-interval", 1*time.Second,
+		"DNS query interval during initialization phase")
+	fs.DurationVar(&s.dnsFetchInitDuration, "node-registry-dns-fetch-init-duration", 30*time.Second,
+		"Duration of the initialization phase for DNS discovery")
+	fs.DurationVar(&s.dnsFetchInterval, "node-registry-dns-fetch-interval", 10*time.Second,
+		"DNS query interval after initialization phase")
+	fs.DurationVar(&s.grpcTimeout, "node-registry-grpc-timeout", 5*time.Second,
+		"Timeout for gRPC calls to fetch node metadata")
+	fs.BoolVar(&s.dnsTLSEnabled, "node-registry-dns-tls", false,
+		"Enable TLS for DNS discovery gRPC connections")
+	fs.StringVar(&s.dnsCACertPath, "node-registry-dns-ca-cert", "",
+		"CA certificate file to verify DNS discovered nodes")
+
 	return fs
 }
 
 func (s *clientService) Validate() error {
 	if s.endpoints == nil {
 		return errors.New("endpoints is empty")
 	}
+	if s.dnsRegistryEnabled && s.dnsTLSEnabled && s.dnsCACertPath == "" {
+		return fmt.Errorf("DNS TLS is enabled, but no CA certificate file was provided")
+	}
 	return nil

Review Comment:
   DNS discovery is initialized but validation only checks dnsCACertPath when TLS is enabled. Should also validate that dnsSRVAddresses is not empty when dnsRegistryEnabled is true, to fail fast rather than during runtime.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC

Review Comment:
   Comment says "fetch metrics from gRPC" but the code actually fetches node metadata, not metrics. The comment should be corrected.
   ```suggestion
   			// fetch node metadata from gRPC
   ```



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// Validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	svc := &Service{
+		srvAddresses:      cfg.SRVAddresses,
+		initInterval:      cfg.InitInterval,
+		initDuration:      cfg.InitDuration,
+		pollInterval:      cfg.PollInterval,
+		grpcTimeout:       cfg.GRPCTimeout,
+		tlsEnabled:        cfg.TLSEnabled,
+		caCertPath:        cfg.CACertPath,
+		nodeCache:         make(map[string]*databasev1.Node),
+		handlers:          make(map[string]schema.EventHandler),
+		lastSuccessfulDNS: []string{},
+		closer:            run.NewCloser(1),
+		log:               logger.GetLogger("metadata-discovery-dns"),
+		resolver:          &defaultResolver{},
+	}
+
+	if svc.tlsEnabled && svc.caCertPath != "" {
+		var err error
+		svc.caCertReloader, err = pkgtls.NewClientCertReloader(svc.caCertPath, svc.log)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize CA certificate reloader: %w", err)
+		}
+		svc.log.Debug().Str("caCertPath", svc.caCertPath).Msg("Initialized DNS CA certificate reloader")
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions() ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	if s.caCertReloader != nil {
+		tlsConfig, err := s.caCertReloader.GetClientTLSConfig("")
+		if err != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader: %w", err)
+		}
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, s.caCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.closer.CloseNotify():
+			return
+		default:
+		}
+
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+		time.Sleep(interval)
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+
+		// Use last successful cache
+		s.lastSuccessMutex.RLock()
+		addresses = s.lastSuccessfulDNS
+		s.lastSuccessMutex.RUnlock()
+
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		// Update last successful cache
+		s.lastSuccessMutex.Lock()
+		s.lastSuccessfulDNS = addresses
+		s.lastSuccessMutex.Unlock()
+
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	var queryErrors []error
+
+	for _, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch metrics from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	s.cacheMutex.Lock()
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			// update cache and notify the handlers
+			delete(s.nodeCache, addr)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("Node removed from cache (no longer in DNS)")
+
+			s.cacheMutex.Unlock()
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, false)
+			s.cacheMutex.Lock()
+		}
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()

Review Comment:
   The lock is released and re-acquired during handler notification, which could lead to inconsistent state if another goroutine modifies the cache between the unlock and lock. Consider collecting all nodes to delete first, then releasing the lock once before notifying handlers.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["codecov-commenter (via GitHub)" <gi...@apache.org>]

codecov-commenter commented on PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#issuecomment-3702865117

   ## [Codecov](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `0%` with `1 line` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 32.19%. Comparing base ([`3530dd9`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/3530dd9b5cefaa4fef60eec91ecbfbdfcaf243c5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`74385f0`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/74385f0c9eb0c04239e16dfcf93ac1c4d203f2ed?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 91 commits behind head on main.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [pkg/index/inverted/metrics.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920?src=pr&el=tree&filepath=pkg%2Findex%2Finverted%2Fmetrics.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2luZGV4L2ludmVydGVkL21ldHJpY3MuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   > :exclamation:  There is a different number of reports uploaded between BASE (3530dd9) and HEAD (74385f0). Click for more details.
   > 
   > <details><summary>HEAD has 1 upload less than BASE</summary>
   >
   >| Flag | BASE (3530dd9) | HEAD (74385f0) |
   >|------|------|------|
   >||1|0|
   ></details>
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff             @@
   ##             main     #920       +/-   ##
   ===========================================
   - Coverage   45.97%   32.19%   -13.79%     
   ===========================================
     Files         328      125      -203     
     Lines       55505    12836    -42669     
   ===========================================
   - Hits        25520     4132    -21388     
   + Misses      27909     8345    -19564     
   + Partials     2076      359     -1717     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [fodc](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.12% <ø> (?)` | |
   | [pkg](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `29.08% <0.00%> (?)` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/920?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["hanahmily (via GitHub)" <gi...@apache.org>]

hanahmily merged PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2662340011


##########
banyand/queue/sub/node.go:
##########
@@ -0,0 +1,30 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package sub
+
+import (
+	"context"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+func (s *server) GetCurrentNode(context.Context, *databasev1.GetCurrentNodeRequest) (*databasev1.GetCurrentNodeResponse, error) {
+	return &databasev1.GetCurrentNodeResponse{
+		Node: s.curNode,

Review Comment:
   The `GetCurrentNode` method returns `s.curNode` without any synchronization protection. Since `s.curNode` is set in `PreRun` and could potentially be read concurrently during server operation, this should use a mutex or atomic operations to ensure thread-safe access.



##########
banyand/metadata/client.go:
##########
@@ -96,13 +115,54 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// node discovery configuration
+	fs.StringVar(&s.nodeDiscoveryMode, "node-discovery-mode", NodeDiscoveryModeEtcd,
+		"Node discovery mode: 'etcd' for etcd-based discovery, 'dns' for DNS-based discovery")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-discovery-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")
+	fs.DurationVar(&s.dnsFetchInitInterval, "node-discovery-dns-fetch-init-interval", 5*time.Second,
+		"DNS query interval during initialization phase")
+	fs.DurationVar(&s.dnsFetchInitDuration, "node-discovery-dns-fetch-init-duration", 5*time.Minute,
+		"Duration of the initialization phase for DNS discovery")
+	fs.DurationVar(&s.dnsFetchInterval, "node-discovery-dns-fetch-interval", 15*time.Second,
+		"DNS query interval after initialization phase")
+	fs.DurationVar(&s.grpcTimeout, "node-discovery-grpc-timeout", 5*time.Second,
+		"Timeout for gRPC calls to fetch node metadata")
+	fs.BoolVar(&s.dnsTLSEnabled, "node-discovery-dns-tls", false,
+		"Enable TLS for DNS discovery gRPC connections")
+	fs.StringSliceVar(&s.dnsCACertPaths, "node-discovery-dns-ca-certs", []string{},
+		"Comma-separated list of CA certificate files to verify DNS discovered nodes (one per SRV address, in same order)")
+
 	return fs
 }
 
 func (s *clientService) Validate() error {
-	if s.endpoints == nil {
-		return errors.New("endpoints is empty")
+	if s.nodeDiscoveryMode != NodeDiscoveryModeEtcd && s.nodeDiscoveryMode != NodeDiscoveryModeDNS {
+		return fmt.Errorf("invalid node-discovery-mode: %s, must be '%s' or '%s'", s.nodeDiscoveryMode, NodeDiscoveryModeEtcd, NodeDiscoveryModeDNS)
 	}
+
+	// Validate etcd endpoints (required when using etcd-based node discovery for schema storage)
+	if s.nodeDiscoveryMode == NodeDiscoveryModeEtcd && len(s.endpoints) == 0 {

Review Comment:
   DNS mode still requires etcd for schema storage, but the validation comment on line 145 says "required when using etcd-based node discovery for schema storage" which is misleading. When DNS mode is used, etcd endpoints are still required for schema operations. The validation should enforce that etcd endpoints are always required (regardless of discovery mode), or the comment should be clarified.
   ```suggestion
   	// Validate etcd endpoints (always required for schema storage, regardless of node discovery mode)
   	if len(s.endpoints) == 0 {
   ```



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,605 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			s.cacheMutex.Lock()
+			if _, alreadyAdded := s.nodeCache[addr]; !alreadyAdded {
+				s.nodeCache[addr] = node
+
+				// notify handlers after releasing lock
+				s.notifyHandlers(schema.Metadata{
+					TypeMeta: schema.TypeMeta{
+						Kind: schema.KindNode,
+						Name: node.GetMetadata().GetName(),
+					},
+					Spec: node,
+				}, true)
+
+				s.log.Debug().
+					Str("address", addr).
+					Str("name", node.GetMetadata().GetName()).
+					Msg("New node discovered and added to cache")
+			}
+			s.cacheMutex.Unlock()
+		}
+	}
+
+	// collect nodes to delete first
+	s.cacheMutex.Lock()
+	nodesToDelete := make(map[string]*databasev1.Node)
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			nodesToDelete[addr] = node
+		}
+	}
+
+	// delete from cache while still holding lock
+	for addr, node := range nodesToDelete {
+		delete(s.nodeCache, addr)
+		s.log.Debug().
+			Str("address", addr).
+			Str("name", node.GetMetadata().GetName()).
+			Msg("Node removed from cache (no longer in DNS)")
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// Notify handlers after releasing lock
+	for _, node := range nodesToDelete {
+		s.notifyHandlers(schema.Metadata{
+			TypeMeta: schema.TypeMeta{
+				Kind: schema.KindNode,
+				Name: node.GetMetadata().GetName(),
+			},
+			Spec: node,
+		}, false)
+	}
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting metadata
+	dialOpts, err := s.getTLSDialOptions(address)
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.Conn(address, s.grpcTimeout, dialOpts...)

Review Comment:
   The context passed to `grpchelper.Conn` should be `ctxTimeout` instead of being ignored. This causes the gRPC connection to not respect the configured timeout. The `nolint:contextcheck` directive hides this issue but doesn't fix it.
   ```suggestion
   	conn, connErr := grpchelper.Conn(ctxTimeout, address, s.grpcTimeout, dialOpts...)
   ```



##########
banyand/metadata/client.go:
##########
@@ -151,10 +211,29 @@ func (s *clientService) PreRun(ctx context.Context) error {
 		}
 		return err
 	}
-	if !s.toRegisterNode {
-		return nil
+
+	if s.nodeDiscoveryMode == NodeDiscoveryModeDNS {
+		l.Info().Strs("srv-addresses", s.dnsSRVAddresses).Msg("Initializing DNS-based node discovery")
+
+		var createErr error
+		s.dnsDiscovery, createErr = dns.NewService(dns.Config{
+			SRVAddresses: s.dnsSRVAddresses,
+			InitInterval: s.dnsFetchInitInterval,
+			InitDuration: s.dnsFetchInitDuration,
+			PollInterval: s.dnsFetchInterval,
+			GRPCTimeout:  s.grpcTimeout,
+			TLSEnabled:   s.dnsTLSEnabled,
+			CACertPaths:  s.dnsCACertPaths,
+		})
+		if createErr != nil {
+			return fmt.Errorf("failed to create DNS discovery service: %w", createErr)
+		}
 	}

Review Comment:
   The DNS discovery service is created in PreRun but never started. Looking at the Serve method, `s.dnsDiscovery.Start()` is called there. However, if PreRun fails after creating the DNS discovery service (lines 219-230), the service and its associated resources (like TLS reloaders) are not cleaned up, leading to a resource leak.



##########
docs/operation/node-discovery.md:
##########
@@ -0,0 +1,296 @@
+# Node Discovery
+
+## Introduction
+
+Node discovery enables BanyanDB nodes to locate and communicate with each other in a distributed cluster. The service discovery mechanism is primarily applied in two scenarios:
+
+1. **Request Routing**: When liaison nodes need to send requests to data nodes for query/write execution.
+2. **Data Migration**: When the lifecycle service queries the node list to perform shard migration and rebalancing.
+
+BanyanDB supports two discovery mechanisms to accommodate different deployment environments:
+
+- **Etcd-based Discovery**: Traditional distributed consensus approach suitable for VM deployments and multi-cloud scenarios.
+- **DNS-based Discovery**: Cloud-native solution leveraging Kubernetes service discovery infrastructure.
+
+This document provides guidance on configuring and operating both discovery modes.
+
+## Etcd-Based Discovery
+
+### Overview
+
+Etcd-based discovery uses a distributed key-value store to maintain cluster membership. Each node registers itself in etcd with a time-limited lease and continuously renews the lease through heartbeat mechanisms.
+
+### Configuration Flags
+
+The following flags control etcd-based node discovery:
+
+```shell
+# Basic configuration
+--node-discovery-mode=etcd                    # Enable etcd mode (default)
+--namespace=banyandb                          # Namespace in etcd (default: banyandb)
+--etcd-endpoints=http://localhost:2379        # Comma-separated etcd endpoints
+
+# Authentication
+--etcd-username=myuser                        # Etcd username (optional)
+--etcd-password=mypass                        # Etcd password (optional)
+
+# TLS configuration
+--etcd-tls-ca-file=/path/to/ca.crt           # Trusted CA certificate
+--etcd-tls-cert-file=/path/to/client.crt     # Client certificate
+--etcd-tls-key-file=/path/to/client.key      # Client private key
+
+# Timeouts and intervals
+--node-registry-timeout=2m                    # Node registration timeout (default: 2m)
+--etcd-full-sync-interval=30m                 # Full state sync interval (default: 30m)
+```
+
+### Configuration Examples
+
+**Data Node with TLS:**
+
+```shell
+banyand data \
+  --node-discovery-mode=etcd \
+  --etcd-endpoints=https://etcd-1:2379,https://etcd-2:2379,https://etcd-3:2379 \
+  --etcd-tls-ca-file=/etc/banyandb/certs/etcd-ca.crt \
+  --etcd-tls-cert-file=/etc/banyandb/certs/etcd-client.crt \
+  --etcd-tls-key-file=/etc/banyandb/certs/etcd-client.key \
+  --namespace=production-banyandb
+```
+
+**Data Node with Basic Authentication:**
+
+```shell
+banyand data \
+  --node-discovery-mode=etcd \
+  --etcd-endpoints=http://etcd-1:2379,http://etcd-2:2379 \
+  --etcd-username=banyandb-client \
+  --etcd-password=${ETCD_PASSWORD} \
+  --namespace=production-banyandb
+```
+
+### Etcd Key Structure
+
+BanyanDB stores node registrations using the following pattern:
+
+```
+/{namespace}/nodes/{node-id}
+
+Example: /banyandb/nodes/data-node-1:17912
+```
+
+**Key Components:**
+
+- **Namespace**: Configurable via `--namespace` flag (default: `banyandb`). Allows multiple BanyanDB clusters to coexist in the same etcd instance
+- **Node ID**: Format `{hostname}:{port}` or `{ip}:{port}`, uniquely identifying each node
+
+The value stored is a Protocol Buffer serialized `databasev1.Node` message containing node metadata, addresses, roles, and labels.
+
+### Node Lifecycle
+
+#### Registration
+
+When a node starts, it performs the following sequence:
+
+1. **Lease Creation**: Creates a 5-second TTL lease with etcd, automatically renewed through keep-alive heartbeat.
+2. **Key-Value Registration**: Uses etcd transaction to ensure the key doesn't already exist, preventing registration conflicts.
+3. **Keep-Alive Loop**: Background goroutine continuously sends heartbeat with automatic reconnection and exponential backoff on network failures.
+
+#### Health Monitoring
+
+Node health is implicitly monitored through the lease mechanism:
+
+- Nodes with expired leases (>5 seconds without keep-alive) are automatically removed by etcd.
+- Presence in etcd indicates the node is alive.
+- Dead nodes disappear within 5 seconds of failure.
+
+#### Event Watching
+
+BanyanDB implements real-time cluster membership tracking:
+
+- Watches etcd prefix `/nodes/` for PUT (add/update) and DELETE events.
+- Revision-based streaming resumes from last known revision after reconnection.
+- Periodic full sync every 5-35 minutes (randomized) to detect missed events.

Review Comment:
   The documentation states that periodic full sync happens "every 5-35 minutes (randomized)" but the default value for `--etcd-full-sync-interval` flag is 30 minutes (non-randomized). This creates a discrepancy between the documentation and the actual implementation.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,605 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			s.cacheMutex.Lock()
+			if _, alreadyAdded := s.nodeCache[addr]; !alreadyAdded {
+				s.nodeCache[addr] = node
+
+				// notify handlers after releasing lock
+				s.notifyHandlers(schema.Metadata{
+					TypeMeta: schema.TypeMeta{
+						Kind: schema.KindNode,
+						Name: node.GetMetadata().GetName(),
+					},
+					Spec: node,
+				}, true)
+
+				s.log.Debug().
+					Str("address", addr).
+					Str("name", node.GetMetadata().GetName()).
+					Msg("New node discovered and added to cache")
+			}
+			s.cacheMutex.Unlock()
+		}
+	}
+
+	// collect nodes to delete first
+	s.cacheMutex.Lock()
+	nodesToDelete := make(map[string]*databasev1.Node)
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			nodesToDelete[addr] = node
+		}
+	}
+
+	// delete from cache while still holding lock
+	for addr, node := range nodesToDelete {
+		delete(s.nodeCache, addr)
+		s.log.Debug().
+			Str("address", addr).
+			Str("name", node.GetMetadata().GetName()).
+			Msg("Node removed from cache (no longer in DNS)")
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// Notify handlers after releasing lock
+	for _, node := range nodesToDelete {
+		s.notifyHandlers(schema.Metadata{
+			TypeMeta: schema.TypeMeta{
+				Kind: schema.KindNode,
+				Name: node.GetMetadata().GetName(),
+			},
+			Spec: node,
+		}, false)
+	}
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting metadata
+	dialOpts, err := s.getTLSDialOptions(address)
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.Conn(address, s.grpcTimeout, dialOpts...)
+	if connErr != nil {
+		grpcErr = fmt.Errorf("failed to connect to %s: %w", address, connErr)
+		return nil, grpcErr
+	}
+	defer conn.Close()
+
+	// query metadata of the node
+	client := databasev1.NewNodeQueryServiceClient(conn)
+	resp, callErr := client.GetCurrentNode(ctxTimeout, &databasev1.GetCurrentNodeRequest{})
+	if callErr != nil {
+		grpcErr = fmt.Errorf("failed to get current node from %s: %w", address, callErr)
+		return nil, grpcErr
+	}
+
+	return resp.GetNode(), nil
+}
+
+func (s *Service) notifyHandlers(metadata schema.Metadata, isAddOrUpdate bool) {
+	s.handlersMutex.RLock()
+	defer s.handlersMutex.RUnlock()
+
+	for _, handler := range s.handlers {
+		if isAddOrUpdate {
+			handler.OnAddOrUpdate(metadata)
+		} else {
+			handler.OnDelete(metadata)
+		}
+	}
+}
+
+// RegisterHandler registers an event handler for node changes.
+func (s *Service) RegisterHandler(name string, handler schema.EventHandler) {
+	s.handlersMutex.Lock()
+	defer s.handlersMutex.Unlock()
+
+	s.handlers[name] = handler
+	s.log.Debug().Str("handler", name).Msg("Registered DNS node discovery handler")
+}
+
+// SetMetrics set the OMR metrics.
+func (s *Service) SetMetrics(factory observability.Factory) {
+	s.metrics = newMetrics(factory)
+}
+
+// Close stops the DNS discovery service.
+func (s *Service) Close() error {
+	s.closer.Done()
+	s.closer.CloseThenWait()
+
+	// stop all Reloaders
+	for certPath, reloader := range s.pathToReloader {
+		reloader.Stop()
+		s.log.Debug().Str("certPath", certPath).Msg("Stopped CA certificate reloader")
+	}
+
+	return nil
+}
+
+// ListNode list all existing nodes from cache.
+func (s *Service) ListNode(ctx context.Context, role databasev1.Role) ([]*databasev1.Node, error) {
+	// if the service is haven't begun/finished, then try to query and update DNS first

Review Comment:
   The comment on line 550 has a grammar error: "haven't begun/finished" should be "hasn't begun/finished" since "service" is singular.
   ```suggestion
   	// if the service hasn't begun/finished, then try to query and update DNS first
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["mrproliu (via GitHub)" <gi...@apache.org>]

mrproliu commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2659314170


##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,508 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime     time.Time
+	resolver          Resolver
+	caCertReloader    *pkgtls.Reloader
+	nodeCache         map[string]*databasev1.Node
+	closer            *run.Closer
+	log               *logger.Logger
+	metrics           *metrics
+	handlers          map[string]schema.EventHandler
+	caCertPath        string
+	srvAddresses      []string
+	lastSuccessfulDNS []string
+	pollInterval      time.Duration
+	initInterval      time.Duration
+	initDuration      time.Duration
+	grpcTimeout       time.Duration
+	cacheMutex        sync.RWMutex
+	lastSuccessMutex  sync.RWMutex
+	handlersMutex     sync.RWMutex
+	lastQueryMutex    sync.RWMutex
+	tlsEnabled        bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPath   string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)

Review Comment:
   Because the UT still needs to mock the DNS result, this interface is necessary. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["mrproliu (via GitHub)" <gi...@apache.org>]

mrproliu commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2659314496


##########
banyand/metadata/client.go:
##########
@@ -96,13 +108,35 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// DNS-based node discovery configuration
+	fs.BoolVar(&s.dnsRegistryEnabled, "node-registry-dns-enabled", false,
+		"Is DNS registry enabled")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-registry-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")

Review Comment:
   It should be kept, because the port name will be defined in the Kubernetes service. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2660519022


##########
banyand/metadata/client.go:
##########
@@ -96,13 +115,54 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// node discovery configuration
+	fs.StringVar(&s.nodeDiscoveryMode, "node-discovery-mode", NodeDiscoveryModeEtcd,
+		"Node discovery mode: 'etcd' for etcd-based discovery, 'dns' for DNS-based discovery")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-discovery-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")
+	fs.DurationVar(&s.dnsFetchInitInterval, "node-discovery-dns-fetch-init-interval", 5*time.Second,
+		"DNS query interval during initialization phase")
+	fs.DurationVar(&s.dnsFetchInitDuration, "node-discovery-dns-fetch-init-duration", 5*time.Minute,
+		"Duration of the initialization phase for DNS discovery")
+	fs.DurationVar(&s.dnsFetchInterval, "node-discovery-dns-fetch-interval", 15*time.Second,
+		"DNS query interval after initialization phase")
+	fs.DurationVar(&s.grpcTimeout, "node-discovery-grpc-timeout", 5*time.Second,
+		"Timeout for gRPC calls to fetch node metadata")
+	fs.BoolVar(&s.dnsTLSEnabled, "node-discovery-dns-tls", false,
+		"Enable TLS for DNS discovery gRPC connections")
+	fs.StringSliceVar(&s.dnsCACertPaths, "node-discovery-dns-ca-certs", []string{},
+		"Comma-separated list of CA certificate files to verify DNS discovered nodes (one per SRV address, in same order)")
+
 	return fs
 }
 
 func (s *clientService) Validate() error {
-	if s.endpoints == nil {
-		return errors.New("endpoints is empty")
+	if s.nodeDiscoveryMode != NodeDiscoveryModeEtcd && s.nodeDiscoveryMode != NodeDiscoveryModeDNS {
+		return fmt.Errorf("invalid node-discovery-mode: %s, must be '%s' or '%s'", s.nodeDiscoveryMode, NodeDiscoveryModeEtcd, NodeDiscoveryModeDNS)
 	}
+
+	// Validate etcd endpoints (required for both modes for schema storage)

Review Comment:
   The validation only checks that etcd endpoints are not empty when mode is etcd, but the comment on line 145 states "required for both modes for schema storage". This suggests that even in DNS mode, etcd endpoints might be needed for schema storage. The validation should either require endpoints for both modes, or the comment should be corrected if DNS mode truly doesn't require etcd.
   ```suggestion
   	// Validate etcd endpoints (required when using etcd-based node discovery for schema storage)
   ```



##########
banyand/queue/sub/server.go:
##########
@@ -139,6 +144,27 @@ func (s *server) PreRun(_ context.Context) error {
 		s.log.Info().Str("certFile", s.certFile).Str("keyFile", s.keyFile).Msg("Initialized TLS reloader for queue server")
 	}
 
+	nodeVal := ctx.Value(common.ContextNodeKey)
+	roleVal := ctx.Value(common.ContextNodeRolesKey)
+	if nodeVal == nil || roleVal == nil {
+		s.log.Warn().Msg("node or role value not found in context")
+		return nil
+	}
+	nodeRoles := roleVal.([]databasev1.Role)
+	node := nodeVal.(common.Node)

Review Comment:
   Type assertions on lines 153-154 can panic if the context values are not of the expected types. While the nil check on line 149 handles missing values, if the values exist but are of wrong types, the server will panic. Add type assertion checks with ok pattern to handle type mismatches gracefully.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	// collect nodes to delete first
+	s.cacheMutex.Lock()
+	nodesToDelete := make(map[string]*databasev1.Node)
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			nodesToDelete[addr] = node
+		}
+	}
+
+	// delete from cache while still holding lock
+	for addr, node := range nodesToDelete {
+		delete(s.nodeCache, addr)
+		s.log.Debug().
+			Str("address", addr).
+			Str("name", node.GetMetadata().GetName()).
+			Msg("Node removed from cache (no longer in DNS)")
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// Notify handlers after releasing lock
+	for _, node := range nodesToDelete {
+		s.notifyHandlers(schema.Metadata{
+			TypeMeta: schema.TypeMeta{
+				Kind: schema.KindNode,
+				Name: node.GetMetadata().GetName(),
+			},
+			Spec: node,
+		}, false)
+	}
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting metadata
+	dialOpts, err := s.getTLSDialOptions(address)
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.ConnWithAuth(address, s.grpcTimeout, "", "", dialOpts...)

Review Comment:
   The nolint directive disables context checking, but a context with timeout (ctxTimeout) has already been created on line 477. The connection should use ctxTimeout instead of having its own timeout parameter, or if ConnWithAuth requires a separate timeout, it should respect the parent context's cancellation. This could lead to connections outliving their intended context lifetime.
   ```suggestion
   
   	// ensure the connection timeout does not exceed the remaining context deadline
   	connTimeout := s.grpcTimeout
   	if deadline, ok := ctxTimeout.Deadline(); ok {
   		remaining := time.Until(deadline)
   		if remaining <= 0 {
   			grpcErr = ctxTimeout.Err()
   			return nil, grpcErr
   		}
   		if remaining < connTimeout {
   			connTimeout = remaining
   		}
   	}
   
   	conn, connErr := grpchelper.ConnWithAuth(address, connTimeout, "", "", dialOpts...)
   ```



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses

Review Comment:
   When DNS query fails, the code updates lastSuccessfulDNS on line 295 only in the success path. However, if DNS fails but cached addresses are available and used (lines 287-293), lastSuccessfulDNS is not updated. This means if the cache is stale and DNS queries start working again but return different results, the old cache remains in lastSuccessfulDNS. Consider documenting this behavior or updating the cache assignment logic to be more explicit about when lastSuccessfulDNS should be refreshed.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()

Review Comment:
   There's a potential race condition in the cache check. Between releasing the RLock on line 384 and acquiring the Lock on line 399, another goroutine could have added the same node to the cache. This could lead to duplicate fetchNodeMetadata calls and duplicate handler notifications for the same address. Consider using a check-lock-check pattern or using a single Lock for the entire operation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["hanahmily (via GitHub)" <gi...@apache.org>]

hanahmily commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2661114046


##########
banyand/metadata/dns/dns_test.go:
##########
@@ -0,0 +1,1714 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dns_test
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+
+	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/dns"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+func TestDNS(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "DNS Discovery Suite", Label("integration"))
+}
+
+var _ = BeforeSuite(func() {
+	Expect(logger.Init(logger.Logging{
+		Env:   "dev",
+		Level: flags.LogLevel,
+	})).To(Succeed())
+})
+
+var _ = Describe("DNS Discovery Service", func() {
+	var (
+		ctx        context.Context
+		cancel     context.CancelFunc
+		mockServer *mockNodeQueryServer
+		grpcServer *grpc.Server
+		listener   net.Listener
+	)
+
+	BeforeEach(func() {
+		ctx, cancel = context.WithCancel(context.Background())
+
+		// Setup mock gRPC server
+		listener, grpcServer, mockServer = setupMockGRPCServer()
+
+		// Add test node to mock server
+		mockServer.node = createTestNode("node1", listener.Addr().String())
+	})
+
+	AfterEach(func() {
+		if grpcServer != nil {
+			grpcServer.Stop()
+		}
+		if listener != nil {
+			_ = listener.Close()
+		}
+		cancel()
+	})
+
+	Describe("NewService", func() {
+		It("should create service with valid config", func() {
+			config := dns.Config{
+				SRVAddresses: []string{"_grpc._tcp.test.local"},
+				InitInterval: 1 * time.Second,
+				InitDuration: 10 * time.Second,
+				PollInterval: 5 * time.Second,
+				GRPCTimeout:  3 * time.Second,
+				TLSEnabled:   false,
+			}
+
+			svc, err := dns.NewService(config)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(svc).NotTo(BeNil())
+			Expect(svc.Close()).To(Succeed())
+		})
+
+		It("should fail with empty SRV addresses", func() {
+			config := dns.Config{
+				SRVAddresses: []string{},
+			}
+
+			_, err := dns.NewService(config)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("SRV addresses cannot be empty"))
+		})
+
+		It("should fail with invalid CA cert path when TLS enabled", func() {
+			config := dns.Config{
+				SRVAddresses: []string{"_grpc._tcp.test.local"},
+				TLSEnabled:   true,
+				CACertPaths:  []string{"/non/existent/path/ca.crt"},
+			}
+
+			_, err := dns.NewService(config)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("Node Registry Interface", func() {
+		var (
+			svc              *dns.Service
+			mockResolver     *mockDNSResolver
+			grpcServer       *grpc.Server
+			registryListener net.Listener
+			mockServer       *mockNodeQueryServer
+		)
+
+		BeforeEach(func() {
+			mockResolver = newMockDNSResolver()
+
+			// Setup mock gRPC server
+			registryListener, grpcServer, mockServer = setupMockGRPCServer()
+
+			// Set up DNS to return our gRPC server
+			serverAddr := registryListener.Addr().String()
+			mockResolver.setResponse("_grpc._tcp.test.local", []*net.SRV{addrToSRV(serverAddr)})
+
+			// Configure mock server with test nodes
+			mockServer.node = createTestNode("registry-test-node", serverAddr,
+				databasev1.Role_ROLE_DATA, databasev1.Role_ROLE_LIAISON)
+
+			var err error
+			svc, err = dns.NewServiceWithResolver(createDefaultConfig(), mockResolver)
+			Expect(err).NotTo(HaveOccurred())
+
+			// Start the discovery service
+			err = svc.Start(ctx)
+			Expect(err).NotTo(HaveOccurred())
+
+			// Wait for initial DNS discovery to complete
+			time.Sleep(300 * time.Millisecond)
+		})
+
+		AfterEach(func() {
+			if svc != nil {
+				_ = svc.Close()
+				svc = nil
+			}
+			if grpcServer != nil {
+				grpcServer.Stop()
+				grpcServer = nil
+			}
+			if registryListener != nil {
+				_ = registryListener.Close()
+				registryListener = nil
+			}
+		})
+
+		It("should return error for RegisterNode", func() {
+			node := &databasev1.Node{
+				Metadata: &commonv1.Metadata{
+					Name: "test-node",
+				},
+			}
+
+			err := svc.RegisterNode(ctx, node, false)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("not supported in DNS discovery mode"))
+		})
+
+		It("should return error for UpdateNode", func() {
+			node := &databasev1.Node{
+				Metadata: &commonv1.Metadata{
+					Name: "test-node",
+				},
+			}
+
+			err := svc.UpdateNode(ctx, node)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("not supported in DNS discovery mode"))
+		})
+
+		It("should list nodes by role after DNS discovery", func() {
+			// Verify node was discovered
+			nodes, err := svc.ListNode(ctx, databasev1.Role_ROLE_DATA)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(nodes).To(HaveLen(1))
+			Expect(nodes[0].GetMetadata().GetName()).To(Equal("registry-test-node"))
+			Expect(nodes[0].GetRoles()).To(ContainElement(databasev1.Role_ROLE_DATA))
+
+			// Verify role filtering works
+			liaisonNodes, err := svc.ListNode(ctx, databasev1.Role_ROLE_LIAISON)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(liaisonNodes).To(HaveLen(1))
+
+			metadataNodes, err := svc.ListNode(ctx, databasev1.Role_ROLE_META)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(metadataNodes).To(HaveLen(0))
+
+			// Verify DNS resolver was called
+			Expect(mockResolver.getCallCount("_grpc._tcp.test.local")).To(BeNumerically(">=", 1))
+		})
+
+		It("should get node by name after DNS discovery", func() {
+			// Verify node can be retrieved by name
+			node, err := svc.GetNode(ctx, "registry-test-node")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(node).NotTo(BeNil())
+			Expect(node.GetMetadata().GetName()).To(Equal("registry-test-node"))
+			Expect(node.GetGrpcAddress()).NotTo(BeEmpty())
+
+			// Verify non-existent node returns error
+			_, err = svc.GetNode(ctx, "non-existent-node")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("not found"))
+		})
+	})
+
+	Describe("DNS Query Operations", func() {
+		var (
+			svc          *dns.Service
+			mockResolver *mockDNSResolver
+			config       dns.Config
+		)
+
+		BeforeEach(func() {
+			mockResolver = newMockDNSResolver()
+			config = createDefaultConfig()
+		})
+
+		AfterEach(func() {
+			if svc != nil {
+				Expect(svc.Close()).To(Succeed())
+			}
+		})
+
+		It("should successfully query single SRV record", func() {
+			mockResolver.setResponse("_grpc._tcp.test.local", []*net.SRV{
+				{Target: "node1.test.local", Port: 17912},
+			})
+
+			var queryErr error
+			svc, queryErr = dns.NewServiceWithResolver(config, mockResolver)
+			Expect(queryErr).NotTo(HaveOccurred())
+
+			addresses, queryErr := svc.QueryAllSRVRecords(ctx)
+			Expect(queryErr).NotTo(HaveOccurred())
+			Expect(addresses).To(HaveLen(1))
+			Expect(addresses).To(ContainElement("node1.test.local:17912"))
+
+			// Verify DNS resolver was called exactly once
+			Expect(mockResolver.getCallCount("_grpc._tcp.test.local")).To(Equal(1))
+		})
+
+		It("should successfully query and deduplicate multiple SRV records", func() {
+			config.SRVAddresses = []string{
+				"_grpc._tcp.zone1.local",
+				"_grpc._tcp.zone2.local",
+			}
+
+			mockResolver.setResponse("_grpc._tcp.zone1.local", []*net.SRV{
+				{Target: "node1.test.local", Port: 17912},
+				{Target: "node2.test.local", Port: 17912},
+			})
+			mockResolver.setResponse("_grpc._tcp.zone2.local", []*net.SRV{
+				{Target: "node1.test.local", Port: 17912}, // Duplicate
+				{Target: "node3.test.local", Port: 17912},
+			})
+
+			var queryErr error
+			svc, queryErr = dns.NewServiceWithResolver(config, mockResolver)
+			Expect(queryErr).NotTo(HaveOccurred())
+
+			addresses, queryErr := svc.QueryAllSRVRecords(ctx)
+			Expect(queryErr).NotTo(HaveOccurred())
+			Expect(addresses).To(HaveLen(3)) // Deduplicated
+			Expect(addresses).To(ContainElements(
+				"node1.test.local:17912",
+				"node2.test.local:17912",
+				"node3.test.local:17912",
+			))
+
+			// Verify each DNS address was queried exactly once
+			Expect(mockResolver.getCallCount("_grpc._tcp.zone1.local")).To(Equal(1))
+			Expect(mockResolver.getCallCount("_grpc._tcp.zone2.local")).To(Equal(1))
+		})
+
+		It("should fail when all DNS queries fail (🎯 critical scenario)", func() {
+			mockResolver.setError("_grpc._tcp.test.local", fmt.Errorf("DNS server unavailable"))
+
+			var queryErr error
+			svc, queryErr = dns.NewServiceWithResolver(config, mockResolver)
+			Expect(queryErr).NotTo(HaveOccurred())
+
+			addresses, queryErr := svc.QueryAllSRVRecords(ctx)
+			Expect(queryErr).To(HaveOccurred())
+			Expect(queryErr.Error()).To(ContainSubstring("DNS server unavailable"))
+			Expect(addresses).To(BeEmpty())
+
+			// Verify DNS was still called (and failed)
+			Expect(mockResolver.getCallCount("_grpc._tcp.test.local")).To(Equal(1))
+		})
+
+		It("should fail when any DNS query fails (🎯 critical scenario)", func() {

Review Comment:
   It should not fail when part of the DNS query fails.



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()
+
+			s.notifyHandlers(schema.Metadata{
+				TypeMeta: schema.TypeMeta{
+					Kind: schema.KindNode,
+					Name: node.GetMetadata().GetName(),
+				},
+				Spec: node,
+			}, true)
+
+			s.log.Debug().
+				Str("address", addr).
+				Str("name", node.GetMetadata().GetName()).
+				Msg("New node discovered and added to cache")
+		}
+	}
+
+	// collect nodes to delete first
+	s.cacheMutex.Lock()
+	nodesToDelete := make(map[string]*databasev1.Node)
+	for addr, node := range s.nodeCache {
+		if !addressSet[addr] {
+			nodesToDelete[addr] = node
+		}
+	}
+
+	// delete from cache while still holding lock
+	for addr, node := range nodesToDelete {
+		delete(s.nodeCache, addr)
+		s.log.Debug().
+			Str("address", addr).
+			Str("name", node.GetMetadata().GetName()).
+			Msg("Node removed from cache (no longer in DNS)")
+	}
+	cacheSize := len(s.nodeCache)
+	s.cacheMutex.Unlock()
+
+	// Notify handlers after releasing lock
+	for _, node := range nodesToDelete {
+		s.notifyHandlers(schema.Metadata{
+			TypeMeta: schema.TypeMeta{
+				Kind: schema.KindNode,
+				Name: node.GetMetadata().GetName(),
+			},
+			Spec: node,
+		}, false)
+	}
+
+	// update total nodes metric
+	if s.metrics != nil {
+		s.metrics.totalNodesCount.Set(float64(cacheSize))
+	}
+
+	if len(addErrors) > 0 {
+		return errors.Join(addErrors...)
+	}
+
+	return nil
+}
+
+func (s *Service) fetchNodeMetadata(ctx context.Context, address string) (*databasev1.Node, error) {
+	// record gRPC query metrics
+	startTime := time.Now()
+	var grpcErr error
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.grpcQueryCount.Inc(1)
+			s.metrics.grpcQueryDuration.Observe(duration.Seconds())
+			s.metrics.grpcQueryTotalDuration.Inc(duration.Seconds())
+			if grpcErr != nil {
+				s.metrics.grpcQueryFailedCount.Inc(1)
+			}
+		}
+	}()
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, s.grpcTimeout)
+	defer cancel()
+
+	// for TLS connections with other nodes to getting metadata
+	dialOpts, err := s.getTLSDialOptions(address)
+	if err != nil {
+		grpcErr = fmt.Errorf("failed to get TLS dial options: %w", err)
+		return nil, grpcErr
+	}
+	// nolint:contextcheck
+	conn, connErr := grpchelper.ConnWithAuth(address, s.grpcTimeout, "", "", dialOpts...)

Review Comment:
   ```suggestion
   	conn, connErr := grpchelper.Conn(address, s.grpcTimeout, dialOpts...)
   ```



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {
+		if s.metrics != nil {
+			s.metrics.dnsQueryFailedCount.Inc(1)
+		}
+		return nil, errors.Join(queryErrors...)
+	}
+
+	// update the resolved address to SRV index mapping
+	s.resolvedAddrMutex.Lock()
+	s.resolvedAddrToSRVIdx = newAddrToSRVIdx
+	s.resolvedAddrMutex.Unlock()
+
+	// convert map to slice
+	result := make([]string, 0, len(allAddresses))
+	for addr := range allAddresses {
+		result = append(result, addr)
+	}
+
+	return result, nil
+}
+
+func (s *Service) updateNodeCache(ctx context.Context, addresses []string) error {
+	addressSet := make(map[string]bool)
+	for _, addr := range addresses {
+		addressSet[addr] = true
+	}
+
+	var addErrors []error
+
+	for addr := range addressSet {
+		s.cacheMutex.RLock()
+		_, exists := s.nodeCache[addr]
+		s.cacheMutex.RUnlock()
+
+		if !exists {
+			// fetch node metadata from gRPC
+			node, fetchErr := s.fetchNodeMetadata(ctx, addr)
+			if fetchErr != nil {
+				s.log.Warn().
+					Err(fetchErr).
+					Str("address", addr).
+					Msg("Failed to fetch node metadata")
+				addErrors = append(addErrors, fetchErr)
+				continue
+			}
+
+			// update cache and notify handlers
+			s.cacheMutex.Lock()
+			s.nodeCache[addr] = node
+			s.cacheMutex.Unlock()

Review Comment:
   I prefer the check-lock-check pattern



##########
banyand/metadata/client.go:
##########
@@ -96,13 +115,54 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// node discovery configuration
+	fs.StringVar(&s.nodeDiscoveryMode, "node-discovery-mode", NodeDiscoveryModeEtcd,
+		"Node discovery mode: 'etcd' for etcd-based discovery, 'dns' for DNS-based discovery")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-discovery-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")
+	fs.DurationVar(&s.dnsFetchInitInterval, "node-discovery-dns-fetch-init-interval", 5*time.Second,
+		"DNS query interval during initialization phase")
+	fs.DurationVar(&s.dnsFetchInitDuration, "node-discovery-dns-fetch-init-duration", 5*time.Minute,
+		"Duration of the initialization phase for DNS discovery")
+	fs.DurationVar(&s.dnsFetchInterval, "node-discovery-dns-fetch-interval", 15*time.Second,
+		"DNS query interval after initialization phase")
+	fs.DurationVar(&s.grpcTimeout, "node-discovery-grpc-timeout", 5*time.Second,
+		"Timeout for gRPC calls to fetch node metadata")
+	fs.BoolVar(&s.dnsTLSEnabled, "node-discovery-dns-tls", false,
+		"Enable TLS for DNS discovery gRPC connections")
+	fs.StringSliceVar(&s.dnsCACertPaths, "node-discovery-dns-ca-certs", []string{},
+		"Comma-separated list of CA certificate files to verify DNS discovered nodes (one per SRV address, in same order)")
+
 	return fs
 }
 
 func (s *clientService) Validate() error {
-	if s.endpoints == nil {
-		return errors.New("endpoints is empty")
+	if s.nodeDiscoveryMode != NodeDiscoveryModeEtcd && s.nodeDiscoveryMode != NodeDiscoveryModeDNS {
+		return fmt.Errorf("invalid node-discovery-mode: %s, must be '%s' or '%s'", s.nodeDiscoveryMode, NodeDiscoveryModeEtcd, NodeDiscoveryModeDNS)
 	}
+
+	// Validate etcd endpoints (required for both modes for schema storage)

Review Comment:
   fix it



##########
banyand/metadata/dns/dns.go:
##########
@@ -0,0 +1,603 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package dns implements DNS-based node discovery for distributed metadata management.
+package dns
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+	"github.com/apache/skywalking-banyandb/pkg/run"
+	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
+)
+
+// Service implements DNS-based node discovery.
+type Service struct {
+	lastQueryTime        time.Time
+	resolver             Resolver
+	pathToReloader       map[string]*pkgtls.Reloader
+	srvIndexToPath       map[int]string
+	resolvedAddrToSRVIdx map[string]int
+	nodeCache            map[string]*databasev1.Node
+	closer               *run.Closer
+	log                  *logger.Logger
+	metrics              *metrics
+	handlers             map[string]schema.EventHandler
+	caCertPaths          []string
+	srvAddresses         []string
+	lastSuccessfulDNS    []string
+	pollInterval         time.Duration
+	initInterval         time.Duration
+	initDuration         time.Duration
+	grpcTimeout          time.Duration
+	cacheMutex           sync.RWMutex
+	handlersMutex        sync.RWMutex
+	lastQueryMutex       sync.RWMutex
+	resolvedAddrMutex    sync.RWMutex
+	tlsEnabled           bool
+}
+
+// Config holds configuration for DNS discovery service.
+type Config struct {
+	CACertPaths  []string
+	SRVAddresses []string
+	InitInterval time.Duration
+	InitDuration time.Duration
+	PollInterval time.Duration
+	GRPCTimeout  time.Duration
+	TLSEnabled   bool
+}
+
+// Resolver defines the interface for DNS SRV lookups.
+type Resolver interface {
+	LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error)
+}
+
+// defaultResolver wraps net.DefaultResolver to implement Resolver.
+type defaultResolver struct{}
+
+func (d *defaultResolver) LookupSRV(ctx context.Context, name string) (string, []*net.SRV, error) {
+	return net.DefaultResolver.LookupSRV(ctx, "", "", name)
+}
+
+// NewService creates a new DNS discovery service.
+func NewService(cfg Config) (*Service, error) {
+	// validation
+	if len(cfg.SRVAddresses) == 0 {
+		return nil, errors.New("DNS SRV addresses cannot be empty")
+	}
+
+	// validate CA cert paths match SRV addresses when TLS is enabled
+	if cfg.TLSEnabled {
+		if len(cfg.CACertPaths) != len(cfg.SRVAddresses) {
+			return nil, fmt.Errorf("number of CA cert paths (%d) must match number of SRV addresses (%d)",
+				len(cfg.CACertPaths), len(cfg.SRVAddresses))
+		}
+	}
+
+	svc := &Service{
+		srvAddresses:         cfg.SRVAddresses,
+		initInterval:         cfg.InitInterval,
+		initDuration:         cfg.InitDuration,
+		pollInterval:         cfg.PollInterval,
+		grpcTimeout:          cfg.GRPCTimeout,
+		tlsEnabled:           cfg.TLSEnabled,
+		caCertPaths:          cfg.CACertPaths,
+		nodeCache:            make(map[string]*databasev1.Node),
+		handlers:             make(map[string]schema.EventHandler),
+		lastSuccessfulDNS:    []string{},
+		pathToReloader:       make(map[string]*pkgtls.Reloader),
+		srvIndexToPath:       make(map[int]string),
+		resolvedAddrToSRVIdx: make(map[string]int),
+		closer:               run.NewCloser(1),
+		log:                  logger.GetLogger("metadata-discovery-dns"),
+		resolver:             &defaultResolver{},
+	}
+
+	// create shared reloaders for CA certificates
+	if svc.tlsEnabled {
+		for srvIdx, certPath := range cfg.CACertPaths {
+			// Store the SRV index → cert path mapping
+			svc.srvIndexToPath[srvIdx] = certPath
+
+			// check if we already have a Reloader for this path
+			if _, exists := svc.pathToReloader[certPath]; exists {
+				svc.log.Debug().Str("certPath", certPath).Int("srvIndex", srvIdx).
+					Msg("Reusing existing CA certificate reloader")
+				continue
+			}
+
+			// create new Reloader for this unique path
+			reloader, reloaderErr := pkgtls.NewClientCertReloader(certPath, svc.log)
+			if reloaderErr != nil {
+				// clean up any already-created reloaders
+				for _, r := range svc.pathToReloader {
+					r.Stop()
+				}
+				return nil, fmt.Errorf("failed to initialize CA certificate reloader for path %s (SRV index %d): %w",
+					certPath, srvIdx, reloaderErr)
+			}
+
+			svc.pathToReloader[certPath] = reloader
+			svc.log.Info().Str("certPath", certPath).Int("srvIndex", srvIdx).
+				Str("srvAddress", cfg.SRVAddresses[srvIdx]).Msg("Initialized DNS CA certificate reloader")
+		}
+	}
+
+	return svc, nil
+}
+
+// newServiceWithResolver creates a service with a custom resolver (for testing).
+func newServiceWithResolver(cfg Config, resolver Resolver) (*Service, error) {
+	svc, err := NewService(cfg)
+	if err != nil {
+		return nil, err
+	}
+	svc.resolver = resolver
+	return svc, nil
+}
+
+func (s *Service) getTLSDialOptions(address string) ([]grpc.DialOption, error) {
+	if !s.tlsEnabled {
+		return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+	}
+
+	// look up which Reloader to use for this address
+	if len(s.pathToReloader) > 0 {
+		// Find which SRV address this resolved address came from
+		s.resolvedAddrMutex.RLock()
+		srvIdx, addrExists := s.resolvedAddrToSRVIdx[address]
+		s.resolvedAddrMutex.RUnlock()
+
+		if !addrExists {
+			return nil, fmt.Errorf("no SRV mapping found for address %s", address)
+		}
+
+		// look up the cert path for this SRV index
+		certPath, pathExists := s.srvIndexToPath[srvIdx]
+		if !pathExists {
+			return nil, fmt.Errorf("no cert path found for SRV index %d (address %s)", srvIdx, address)
+		}
+
+		// get the Reloader for this cert path
+		reloader, reloaderExists := s.pathToReloader[certPath]
+		if !reloaderExists {
+			return nil, fmt.Errorf("no reloader found for cert path %s (address %s)", certPath, address)
+		}
+
+		// get fresh TLS config from the Reloader
+		tlsConfig, configErr := reloader.GetClientTLSConfig("")
+		if configErr != nil {
+			return nil, fmt.Errorf("failed to get TLS config from reloader for address %s: %w", address, configErr)
+		}
+
+		creds := credentials.NewTLS(tlsConfig)
+		return []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
+	}
+
+	// fallback to static TLS config (when no reloaders configured)
+	opts, err := grpchelper.SecureOptions(nil, s.tlsEnabled, false, "")
+	if err != nil {
+		return nil, fmt.Errorf("failed to load TLS config: %w", err)
+	}
+	return opts, nil
+}
+
+// Start begins the DNS discovery background process.
+func (s *Service) Start(ctx context.Context) error {
+	s.log.Debug().Msg("Starting DNS-based node discovery service")
+
+	// start all Reloaders
+	if len(s.pathToReloader) > 0 {
+		startedReloaders := make([]*pkgtls.Reloader, 0, len(s.pathToReloader))
+
+		for certPath, reloader := range s.pathToReloader {
+			if startErr := reloader.Start(); startErr != nil {
+				// stop any already-started reloaders
+				for _, r := range startedReloaders {
+					r.Stop()
+				}
+				return fmt.Errorf("failed to start CA certificate reloader for path %s: %w", certPath, startErr)
+			}
+			startedReloaders = append(startedReloaders, reloader)
+			s.log.Debug().Str("certPath", certPath).Msg("Started CA certificate reloader")
+		}
+	}
+
+	go s.discoveryLoop(ctx)
+
+	return nil
+}
+
+func (s *Service) discoveryLoop(ctx context.Context) {
+	// add the init phase finish time
+	initPhaseEnd := time.Now().Add(s.initDuration)
+
+	for {
+		if err := s.queryDNSAndUpdateNodes(ctx); err != nil {
+			s.log.Err(err).Msg("failed to query DNS and update nodes")
+		}
+
+		// wait for next interval
+		var interval time.Duration
+		if time.Now().Before(initPhaseEnd) {
+			interval = s.initInterval
+		} else {
+			interval = s.pollInterval
+		}
+
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.closer.CloseNotify():
+			timer.Stop()
+			return
+		case <-timer.C:
+			// continue to next iteration
+		}
+	}
+}
+
+func (s *Service) queryDNSAndUpdateNodes(ctx context.Context) error {
+	// Record summary metrics
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.discoveryCount.Inc(1)
+			s.metrics.discoveryDuration.Observe(duration.Seconds())
+			s.metrics.discoveryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	addresses, queryErr := s.queryAllSRVRecords(ctx)
+
+	if queryErr != nil {
+		s.log.Warn().Err(queryErr).Msg("DNS query failed, using last successful cache")
+		addresses = s.lastSuccessfulDNS
+		if len(addresses) == 0 {
+			if s.metrics != nil {
+				s.metrics.discoveryFailedCount.Inc(1)
+			}
+			return fmt.Errorf("DNS query failed and no cached addresses available: %w", queryErr)
+		}
+	} else {
+		s.lastSuccessfulDNS = addresses
+		if s.log.Debug().Enabled() {
+			s.log.Debug().
+				Int("count", len(addresses)).
+				Strs("addresses", addresses).
+				Strs("srv_addresses", s.srvAddresses).
+				Msg("DNS query successful")
+		}
+	}
+
+	// Update node cache based on DNS results
+	updateErr := s.updateNodeCache(ctx, addresses)
+	if updateErr != nil && s.metrics != nil {
+		s.metrics.discoveryFailedCount.Inc(1)
+	}
+	s.lastQueryMutex.Lock()
+	s.lastQueryTime = time.Now()
+	s.lastQueryMutex.Unlock()
+	return updateErr
+}
+
+func (s *Service) queryAllSRVRecords(ctx context.Context) ([]string, error) {
+	startTime := time.Now()
+	defer func() {
+		if s.metrics != nil {
+			duration := time.Since(startTime)
+			s.metrics.dnsQueryCount.Inc(1)
+			s.metrics.dnsQueryDuration.Observe(duration.Seconds())
+			s.metrics.dnsQueryTotalDuration.Inc(duration.Seconds())
+		}
+	}()
+
+	allAddresses := make(map[string]bool)
+	// track which SRV address (by index) each resolved address came from
+	newAddrToSRVIdx := make(map[string]int)
+	var queryErrors []error
+
+	for srvIdx, srvAddr := range s.srvAddresses {
+		_, addrs, lookupErr := s.resolver.LookupSRV(ctx, srvAddr)
+		if lookupErr != nil {
+			queryErrors = append(queryErrors, fmt.Errorf("lookup %s failed: %w", srvAddr, lookupErr))
+			continue
+		}
+
+		for _, srv := range addrs {
+			address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
+			allAddresses[address] = true
+
+			// track which SRV address this resolved to (first-wins strategy)
+			if _, exists := newAddrToSRVIdx[address]; !exists {
+				newAddrToSRVIdx[address] = srvIdx
+			}
+		}
+	}
+
+	// if there have any error occurred,
+	// then just return the query error to ignore the result to make sure the cache correct
+	if len(queryErrors) > 0 {

Review Comment:
   You should return resolved addresses, even if some cannot be resolved.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Support node registry by DNS [skywalking-banyandb]

["hanahmily (via GitHub)" <gi...@apache.org>]

hanahmily commented on code in PR #920:
URL: https://github.com/apache/skywalking-banyandb/pull/920#discussion_r2663166311


##########
banyand/metadata/client.go:
##########
@@ -96,13 +115,54 @@ func (s *clientService) FlagSet() *run.FlagSet {
 	fs.StringVar(&s.etcdTLSKeyFile, flagEtcdTLSKeyFile, "", "Private key for the etcd client certificate.")
 	fs.DurationVar(&s.registryTimeout, "node-registry-timeout", 2*time.Minute, "The timeout for the node registry")
 	fs.DurationVar(&s.etcdFullSyncInterval, "etcd-full-sync-interval", 30*time.Minute, "The interval for full sync etcd")
+
+	// node discovery configuration
+	fs.StringVar(&s.nodeDiscoveryMode, "node-discovery-mode", NodeDiscoveryModeEtcd,
+		"Node discovery mode: 'etcd' for etcd-based discovery, 'dns' for DNS-based discovery")
+	fs.StringSliceVar(&s.dnsSRVAddresses, "node-discovery-dns-srv-addresses", []string{},
+		"DNS SRV addresses for node discovery (e.g., _grpc._tcp.banyandb.svc.cluster.local)")
+	fs.DurationVar(&s.dnsFetchInitInterval, "node-discovery-dns-fetch-init-interval", 5*time.Second,
+		"DNS query interval during initialization phase")
+	fs.DurationVar(&s.dnsFetchInitDuration, "node-discovery-dns-fetch-init-duration", 5*time.Minute,
+		"Duration of the initialization phase for DNS discovery")
+	fs.DurationVar(&s.dnsFetchInterval, "node-discovery-dns-fetch-interval", 15*time.Second,
+		"DNS query interval after initialization phase")
+	fs.DurationVar(&s.grpcTimeout, "node-discovery-grpc-timeout", 5*time.Second,
+		"Timeout for gRPC calls to fetch node metadata")
+	fs.BoolVar(&s.dnsTLSEnabled, "node-discovery-dns-tls", false,
+		"Enable TLS for DNS discovery gRPC connections")
+	fs.StringSliceVar(&s.dnsCACertPaths, "node-discovery-dns-ca-certs", []string{},
+		"Comma-separated list of CA certificate files to verify DNS discovered nodes (one per SRV address, in same order)")
+
 	return fs
 }
 
 func (s *clientService) Validate() error {
-	if s.endpoints == nil {
-		return errors.New("endpoints is empty")
+	if s.nodeDiscoveryMode != NodeDiscoveryModeEtcd && s.nodeDiscoveryMode != NodeDiscoveryModeDNS {
+		return fmt.Errorf("invalid node-discovery-mode: %s, must be '%s' or '%s'", s.nodeDiscoveryMode, NodeDiscoveryModeEtcd, NodeDiscoveryModeDNS)
 	}
+
+	// Validate etcd endpoints (required when using etcd-based node discovery for schema storage)
+	if s.nodeDiscoveryMode == NodeDiscoveryModeEtcd && len(s.endpoints) == 0 {

Review Comment:
   It's a bug, pls fix it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org