feat(metadata): add renew and GC policy for app-level metadata (#3371)

* feat(metadata): add renew and GC policy for app-level metadata

* fix ci

* fix ci

* fix review

* fix nil

* rebase develop, resolve conflicts, add units

Author: eye-gu

common/constant/key.go

@@ -427,6 +427,11 @@ const (
 	MetadataServiceURLParamsPropertyName   = MetadataServicePrefix + "url-params"
 	MetadataServiceURLsPropertyName        = MetadataServicePrefix + "urls"
 	ServiceDiscoveryKey                    = "service_discovery" // indicate which service discovery instance will be used
+
+	// metadata GC configuration keys
+	MetadataGCEnabledKey      = "metadata.gc.enabled"
+	MetadataGCWindowKey       = "metadata.gc.window"        // GC window in days, aligned with daily renew cycle
+	MetadataRenewOnStartupKey = "metadata.renew-on-startup" // whether to run renewAppMetadata once on startup
 )
 
 // Generic Filter

metadata/info/metadata_info.go

@@ -67,6 +67,7 @@ type MetadataInfo struct {
 	exportedServiceURLs   map[string][]*common.URL `hessian:"-"` // server exported service urls
 	subscribedServiceURLs map[string][]*common.URL `hessian:"-"` // client subscribed service urls
 	mu                    sync.RWMutex             `json:"-" hessian:"-"`
+	LastUpdatedTime       int64                    `json:"lastUpdatedTime,omitempty" hessian:"-"`
 }
 
 func NewAppMetadataInfo(app string) *MetadataInfo {
@@ -221,6 +222,25 @@ func (info *MetadataInfo) ReplaceExportedServices(urls []*common.URL) {
 	}
 }
 
+// Snapshot creates a deep copy of the MetadataInfo for safe concurrent access.
+// The caller can modify the snapshot without affecting the original.
+func (info *MetadataInfo) Snapshot() MetadataInfo {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
+	services := make(map[string]*ServiceInfo, len(info.Services))
+	for k, v := range info.Services {
+		si := *v
+		services[k] = &si
+	}
+	return MetadataInfo{
+		App:      info.App,
+		Revision: info.Revision,
+		Tag:      info.Tag,
+		Services: services,
+	}
+}
+
 func (info *MetadataInfo) findExportedServiceURL(matchKey string) *common.URL {
 	for _, urls := range info.exportedServiceURLs {
 		for _, serviceURL := range urls {

metadata/mapping/metadata/service_name_mapping_concurrency_test.go

@@ -30,6 +30,7 @@ import (
 )
 
 import (
+	"dubbo.apache.org/dubbo-go/v3/common"
 	"dubbo.apache.org/dubbo-go/v3/metadata/info"
 	"dubbo.apache.org/dubbo-go/v3/metadata/mapping"
 	"dubbo.apache.org/dubbo-go/v3/metadata/report"
@@ -86,6 +87,11 @@ func (stubReport) GetServiceAppMapping(string, string, mapping.MappingListener)
 	return nil, nil
 }
 func (stubReport) RemoveServiceAppMappingListener(string, string) error { return nil }
+func (stubReport) UnPublishAppMetadata(string, string) error            { return nil }
+func (stubReport) ListAppRevisions(string) ([]report.AppRevision, error) {
+	return nil, nil
+}
+func (stubReport) URL() *common.URL { return nil }
 
 // casReport registers mappings with optimistic concurrency against a versionedStore, exactly
 // as the etcd/zk/nacos reports now do, returning report.ErrMappingCASConflict on conflict.

metadata/mapping/metadata/service_name_mapping_test.go

@@ -289,3 +289,18 @@ func (m *mockMetadataReport) RemoveServiceAppMappingListener(string, string) err
 	args := m.Called()
 	return args.Error(0)
 }
+
+func (m *mockMetadataReport) UnPublishAppMetadata(string, string) error {
+	args := m.Called()
+	return args.Error(0)
+}
+
+func (m *mockMetadataReport) ListAppRevisions(string) ([]report.AppRevision, error) {
+	args := m.Called()
+	return args.Get(0).([]report.AppRevision), args.Error(1)
+}
+
+func (m *mockMetadataReport) URL() *common.URL {
+	u, _ := common.NewURL("mock://127.0.0.1:8848")
+	return u
+}

metadata/report/etcd/report.go

@@ -19,6 +19,7 @@ package etcd
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 )
@@ -29,6 +30,8 @@ import (
 	"github.com/dubbogo/gost/log/logger"
 
 	perrors "github.com/pkg/errors"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 import (
@@ -40,6 +43,26 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/metadata/report"
 )
 
+// etcdClient abstracts the etcd Client operations used by etcdMetadataReport.
+type etcdClient interface {
+	Get(key string) (string, error)
+	Put(key, value string) error
+	Delete(key string) error
+	GetChildren(key string) ([]string, []string, error)
+	GetValAndRev(key string) (string, int64, error)
+	Create(key, value string) error
+	UpdateWithRev(key, value string, rev int64, opts ...clientv3.OpOption) error
+}
+
+// etcdClientWrapper wraps *gxetcd.Client to implement etcdClient.
+type etcdClientWrapper struct {
+	*gxetcd.Client
+}
+
+func (w etcdClientWrapper) Put(key, value string) error {
+	return w.Client.Put(key, value)
+}
+
 const DEFAULT_ROOT = "dubbo"
 
 func init() {
@@ -50,13 +73,19 @@ func init() {
 
 // etcdMetadataReport is the implementation of MetadataReport based etcd
 type etcdMetadataReport struct {
-	client  *gxetcd.Client
+	client  etcdClient
 	rootDir string
+	url     *common.URL
+}
+
+// URL returns the URL used to create this metadata report.
+func (e *etcdMetadataReport) URL() *common.URL {
+	return e.url
 }
 
 // GetAppMetadata get metadata info from etcd
 func (e *etcdMetadataReport) GetAppMetadata(application, revision string) (*info.MetadataInfo, error) {
-	key := e.rootDir + application + constant.PathSeparator + revision
+	key := e.rootDir + constant.PathSeparator + application + constant.PathSeparator + revision
 	data, err := e.client.Get(key)
 	if err != nil {
 		return nil, err
@@ -68,7 +97,7 @@ func (e *etcdMetadataReport) GetAppMetadata(application, revision string) (*info
 
 // PublishAppMetadata publish metadata info to etcd
 func (e *etcdMetadataReport) PublishAppMetadata(application, revision string, info *info.MetadataInfo) error {
-	key := e.rootDir + application + constant.PathSeparator + revision
+	key := e.rootDir + constant.PathSeparator + application + constant.PathSeparator + revision
 	value, err := json.Marshal(info)
 	if err == nil {
 		err = e.client.Put(key, string(value))
@@ -125,6 +154,35 @@ func (e *etcdMetadataReport) RemoveServiceAppMappingListener(key string, group s
 	return nil
 }
 
+// UnPublishAppMetadata removes metadata for a specific revision from etcd.
+// This operation is idempotent.
+func (e *etcdMetadataReport) UnPublishAppMetadata(application, revision string) error {
+	key := e.rootDir + constant.PathSeparator + application + constant.PathSeparator + revision
+	return e.client.Delete(key)
+}
+
+func (e *etcdMetadataReport) ListAppRevisions(application string) ([]report.AppRevision, error) {
+	prefix := e.rootDir + constant.PathSeparator + application + constant.PathSeparator
+	keys, values, err := e.client.GetChildren(prefix)
+	if err != nil {
+		if errors.Is(perrors.Cause(err), gxetcd.ErrKVPairNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	result := make([]report.AppRevision, 0, len(keys))
+	for i, key := range keys {
+		// Extract revision from key suffix (key is full path, revision is last segment)
+		revision := key[strings.LastIndex(key, constant.PathSeparator)+1:]
+		result = append(result, report.AppRevision{
+			Revision:   revision,
+			ModifyTime: report.ParseMetadataLastUpdatedTime([]byte(values[i])),
+		})
+	}
+	return result, nil
+}
+
 type etcdMetadataReportFactory struct{}
 
 // CreateMetadataReport get the MetadataReport instance of etcd
@@ -138,5 +196,5 @@ func (e *etcdMetadataReportFactory) CreateMetadataReport(url *common.URL) report
 	}
 	group := url.GetParam(constant.MetadataReportGroupKey, DEFAULT_ROOT)
 	group = constant.PathSeparator + strings.TrimPrefix(group, constant.PathSeparator)
-	return &etcdMetadataReport{client: client, rootDir: group}
+	return &etcdMetadataReport{client: etcdClientWrapper{client}, rootDir: group, url: url}
 }