fix: prometheus

internal/api/init.go

@@ -94,9 +94,16 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 				return err
 			}
 
-			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
+			if !ok {
+				return errs.New("only etcd support autoSetPorts").Wrap()
+			}
 
-			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName, index), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
+			if err != nil {
+				return errs.Wrap(err)
+			}
+			err = etcdClient.SetWithLease(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName, index), target, prommetrics.TTL)
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}

internal/api/prometheus_discovery.go

@@ -11,7 +11,8 @@ import (
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/errs"
-	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/utils/datautil"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
@@ -40,33 +41,28 @@ func (p *PrometheusDiscoveryApi) Enable(c *gin.Context) {
 func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
 	eResp, err := p.client.Get(c, prommetrics.BuildDiscoveryKeyPrefix(key), clientv3.WithPrefix())
 	if err != nil {
-		// Log and respond with an error if preparation fails.
+		apiresp.GinError(c, errs.WrapMsg(err, "get key value"))
-		apiresp.GinError(c, errs.WrapMsg(err, "etcd get err"))
 		return
 	}
-	if len(eResp.Kvs) == 0 {
+	value := datautil.Batch(func(kv *mvccpb.KeyValue) []byte { return kv.Value }, eResp.Kvs)
-		c.JSON(http.StatusOK, []*prommetrics.Target{})
-	}
 
-	var (
+	if len(value) == 0 {
-		resp = &prommetrics.RespTarget{
+		c.JSON(http.StatusOK, []*prommetrics.RespTarget{})
-			Targets: make([]string, 0, len(eResp.Kvs)),
+		return
+	}
+	var resp prommetrics.RespTarget
+	for i := range value {
+		var tmp prommetrics.Target
+		if err = json.Unmarshal(value[i], &tmp); err != nil {
+			apiresp.GinError(c, errs.WrapMsg(err, "json unmarshal err"))
+			return
 		}
-	)
 
-	for i := range eResp.Kvs {
+		resp.Targets = append(resp.Targets, tmp.Target)
-		var target prommetrics.Target
+		resp.Labels = tmp.Labels // default label is fixed. See prommetrics.BuildDefaultTarget
-		err = json.Unmarshal(eResp.Kvs[i].Value, &target)
-		if err != nil {
-			log.ZError(c, "prometheus unmarshal err", errs.Wrap(err))
-		}
-		resp.Targets = append(resp.Targets, target.Target)
-		if resp.Labels == nil {
-			resp.Labels = target.Labels
-		}
 	}
 
-	c.JSON(200, []*prommetrics.RespTarget{resp})
+	c.JSON(http.StatusOK, []*prommetrics.RespTarget{&resp})
 }
 
 func (p *PrometheusDiscoveryApi) Api(c *gin.Context) {

internal/msgtransfer/init.go

@@ -176,9 +176,16 @@ func (m *MsgTransfer) Start(index int, cfg *Config) error {
 				return err
 			}
 
-			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
+			if !ok {
+				return errs.New("only etcd support autoSetPorts").Wrap()
+			}
 
-			_, err = etcdClient.Put(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName, index), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
+			if err != nil {
+				return errs.Wrap(err)
+			}
+			err = etcdClient.SetWithLease(context.TODO(), prommetrics.BuildDiscoveryKey(prommetrics.MessageTransferKeyName, index), target, prommetrics.TTL)
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}

pkg/common/prommetrics/discovery.go

@@ -5,6 +5,7 @@ import "fmt"
 const (
 	APIKeyName             = "api"
 	MessageTransferKeyName = "message-transfer"
+	TTL                    = 300
 )
 
 type Target struct {

pkg/common/startrpc/start.go

@@ -164,9 +164,16 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 				return err
 			}
 
-			etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+			etcdClient, ok := client.(*etcd.SvcDiscoveryRegistryImpl)
+			if !ok {
+				return errs.New("only etcd support autoSetPorts").Wrap()
+			}
 
-			_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName, index), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
+			target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
+			if err != nil {
+				return errs.Wrap(err)
+			}
+			err = etcdClient.SetWithLease(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName, index), target, prommetrics.TTL)
 			if err != nil {
 				return errs.WrapMsg(err, "etcd put err")
 			}