Create table and related cache for additional instance data

apiserver/eurekaserver/write_test.go

@@ -108,6 +108,10 @@ func TestEurekaServer_renew(t *testing.T) {
 			insId:            ins,
 			disableBeatInsId: disableBeatIns,
 		}, nil)
+	mockStore.EXPECT().
+		GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		AnyTimes().
+		Return(nil, nil)
 	mockStore.EXPECT().StartReadTx().Return(mockTx, nil).AnyTimes()
 	mockStore.EXPECT().
 		GetMoreServices(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).

auth/defaultauth/common_test.go

@@ -109,6 +109,14 @@ func initCache(ctrl *gomock.Controller) (*cache.Config, *storemock.MockStore) {
 			Valid: true,
 		},
 	}, nil).AnyTimes()
+	storage.EXPECT().GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(map[string]*model.InstanceConsole{
+		"123": {
+			Id:       "123",
+			Isolate:  false,
+			Weight:   100,
+			Metadata: "{a: 1}",
+		},
+	}, nil).AnyTimes()
 	storage.EXPECT().GetUnixSecond(gomock.Any()).AnyTimes().Return(time.Now().Unix(), nil)
 
 	return cfg, storage

cache/service/instance.go

@@ -51,7 +51,9 @@ type instanceCache struct {
 	// service id -> [instanceid ->instance]
 	services *utils.SyncMap[string, *ServiceInstances]
 	// service id -> [instanceCount]
-	instanceCounts   *utils.SyncMap[string, *model.InstanceCount]
+	instanceCounts *utils.SyncMap[string, *model.InstanceCount]
+	// service id -> [instanceConsole]
+	instanceConsoles *utils.SyncMap[string, *model.InstanceConsole]
 	instancePorts    *instancePorts
 	disableBusiness  bool
 	needMeta         bool
@@ -76,6 +78,7 @@ func (ic *instanceCache) Initialize(opt map[string]interface{}) error {
 	ic.ids = utils.NewSyncMap[string, *model.Instance]()
 	ic.services = utils.NewSyncMap[string, *ServiceInstances]()
 	ic.instanceCounts = utils.NewSyncMap[string, *model.InstanceCount]()
+	ic.instanceConsoles = utils.NewSyncMap[string, *model.InstanceConsole]()
 	ic.instancePorts = newInstancePorts()
 	if opt == nil {
 		return nil
@@ -191,6 +194,17 @@ func (ic *instanceCache) handleUpdate(start time.Time, tx store.Tx) ([]*eventhub
 		return nil, nil, -1, err
 	}
 
+	instanceConsoles, err := ic.storage.GetMoreInstanceConsoles(tx, ic.LastFetchTime(), ic.IsFirstUpdate(),
+		ic.needMeta, ic.systemServiceID)
+	if err != nil {
+		log.Error("[Cache][InstanceConsole] update get storage more", zap.Error(err))
+		return nil, nil, -1, err
+	}
+	for _, item := range instanceConsoles {
+		//Todo: check validation
+		ic.instanceConsoles.Store(item.Id, item)
+	}
+
 	events, lastMtimes, update, del := ic.setInstances(instances)
 	log.Info("[Cache][Instance] get more instances",
 		zap.Int("pull-from-store", len(instances)), zap.Int("update", update), zap.Int("delete", del),
@@ -204,6 +218,7 @@ func (ic *instanceCache) Clear() error {
 	ic.ids = utils.NewSyncMap[string, *model.Instance]()
 	ic.services = utils.NewSyncMap[string, *ServiceInstances]()
 	ic.instanceCounts = utils.NewSyncMap[string, *model.InstanceCount]()
+	ic.instanceConsoles = utils.NewSyncMap[string, *model.InstanceConsole]()
 	ic.instancePorts.reset()
 	ic.instanceCount = 0
 	return nil
@@ -430,6 +445,20 @@ func (ic *instanceCache) GetInstance(instanceID string) *model.Instance {
 	return value
 }
 
+// GetInstanceConsole 根据实例ID获取实例数据
+func (ic *instanceCache) GetInstanceConsole(instanceConsoleID string) *model.InstanceConsole {
+	if instanceConsoleID == "" {
+		return nil
+	}
+
+	value, ok := ic.instanceConsoles.Load(instanceConsoleID)
+	if !ok {
+		return nil
+	}
+
+	return value
+}
+
 // GetInstancesByServiceID 根据ServiceID获取实例数据
 func (ic *instanceCache) GetInstancesByServiceID(serviceID string) []*model.Instance {
 	if serviceID == "" {

cache/service/instance_test.go

@@ -92,6 +92,21 @@ func genModelInstances(label string, total int) map[string]*model.Instance {
 	return out
 }
 
+func genModelInstancesConsole(label string, total int) map[string]*model.InstanceConsole {
+	out := make(map[string]*model.InstanceConsole)
+	for i := 0; i < total; i++ {
+		entry := &model.InstanceConsole{
+			Id:       fmt.Sprintf("InstanceConsole-%s-%d", label, i),
+			Isolate:  false,
+			Weight:   100,
+			Metadata: "Metadata",
+		}
+		out[entry.Id] = entry
+	}
+
+	return out
+}
+
 // 对instanceCache的缓存数据进行计数统计
 func iteratorInstances(ic *instanceCache) (int, int) {
 	instancesCount := 0
@@ -116,6 +131,7 @@ func TestInstanceCache_Update(t *testing.T) {
 		ret := make(map[string]*model.Instance)
 		instances1 := genModelInstances("service1", 10) // 每次gen为一个服务的
 		instances2 := genModelInstances("service2", 5)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 
 		for id, instance := range instances1 {
 			ret[id] = instance
@@ -127,13 +143,17 @@ func TestInstanceCache_Update(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(ret, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		gomock.InOrder(storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(15), nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
 		}
 
 		servicesCount, instancesCount := iteratorInstances(ic)
-		if servicesCount == 2 && instancesCount == 10+5 { // gen两次，有两个不同服务
+		instanceConsoleCounts := ic.instanceConsoles.Len()
+		if servicesCount == 2 && instancesCount == 10+5 && instanceConsoleCounts == 3 { // gen两次，有两个不同服务
 			t.Logf("pass")
 		} else {
 			t.Fatalf("error: %d, %d", servicesCount, instancesCount)
@@ -145,26 +165,34 @@ func TestInstanceCache_Update(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(nil, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(nil, nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
 		}
 
 		servicesCount, instancesCount := iteratorInstances(ic)
-		if servicesCount != 0 || instancesCount != 0 {
+		instanceConsoleCounts := ic.instanceConsoles.Len()
+		if servicesCount != 0 || instancesCount != 0 || instanceConsoleCounts != 0 {
 			t.Fatalf("error: %d %d", servicesCount, instancesCount)
 		}
 	})
 
 	t.Run("lastMtime可以正常更新", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("services", 10)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 		maxMtime := time.Now()
 		instances[fmt.Sprintf("instanceID-%s-%d", "services", 5)].ModifyTime = maxMtime
 
 		gomock.InOrder(
 			storage.EXPECT().
 				GetMoreInstances(gomock.Any(), gomock.Any(), gomock.Any(), ic.needMeta, ic.systemServiceID).
 				Return(instances, nil),
+			storage.EXPECT().
+				GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+				Return(instanceConsoles, nil),
 			storage.EXPECT().GetUnixSecond(gomock.Any()).Return(maxMtime.Unix(), nil).AnyTimes(),
 		)
 		if err := ic.Update(); err != nil {
@@ -186,6 +214,9 @@ func TestInstanceCache_Update2(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(nil, fmt.Errorf("storage get error")))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(nil, nil))
 		gomock.InOrder(storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(0), fmt.Errorf("storage get error")))
 		if err := ic.Update(); err != nil {
 			t.Logf("pass: %s", err.Error())
@@ -197,9 +228,13 @@ func TestInstanceCache_Update2(t *testing.T) {
 	t.Run("更新数据，再删除部分数据，缓存正常", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("service-a", 20)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(instances, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
 		}
@@ -216,6 +251,9 @@ func TestInstanceCache_Update2(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(instances, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
 		}
@@ -229,6 +267,7 @@ func TestInstanceCache_Update2(t *testing.T) {
 	t.Run("对账发现缓存数据数量和存储层不一致", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("service-a", 20)
+
 		queryCount := int32(0)
 		storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(0), nil).AnyTimes()
 		storage.EXPECT().
@@ -254,9 +293,14 @@ func TestInstanceCache_GetInstance(t *testing.T) {
 	t.Run("缓存有数据，可以正常获取到数据", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("my-services", 10)
+		instanceConsoles := genModelInstancesConsole("console", 3)
+
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(instances, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		gomock.InOrder(storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(10), nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
@@ -269,6 +313,14 @@ func TestInstanceCache_GetInstance(t *testing.T) {
 		if instance := ic.GetInstance("test-instance-xx"); instance != nil {
 			t.Fatalf("error")
 		}
+
+		if instanceConsole := ic.GetInstanceConsole(instanceConsoles[fmt.Sprintf("InstanceConsole-%s-%d", "console", 2)].Id); instanceConsole == nil {
+			t.Fatalf("error")
+		}
+
+		if instanceConsole := ic.GetInstance("test-instanceConsole-xx"); instanceConsole != nil {
+			t.Fatalf("error")
+		}
 	})
 }
 
@@ -278,6 +330,7 @@ func TestInstanceCache_GetServicePorts(t *testing.T) {
 	t.Run("缓存有数据，可以正常获取到服务的端口列表", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("my-services", 10)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 
 		ports := make(map[string][]*model.ServicePort)
 
@@ -309,6 +362,9 @@ func TestInstanceCache_GetServicePorts(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(instances, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		gomock.InOrder(storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(10), nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
@@ -331,6 +387,7 @@ func TestInstanceCache_fillIntrnalLabels(t *testing.T) {
 	t.Run("向实例Metadata中自动注入北极星默认label信息", func(t *testing.T) {
 		_ = ic.Clear()
 		instances := genModelInstances("inject-internal-label", 10)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 
 		ports := make(map[string][]string)
 
@@ -360,6 +417,9 @@ func TestInstanceCache_fillIntrnalLabels(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(instances, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		gomock.InOrder(storage.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(10), nil))
 		if err := ic.Update(); err != nil {
 			t.Fatalf("error: %s", err.Error())
@@ -391,6 +451,7 @@ func TestGetInstancesByServiceID(t *testing.T) {
 		instances1 := genModelInstances("my-services", instances1Count)
 		instances2 := genModelInstances("my-services-a", instances2Count)
 		// instances2 = append(instances2, instances1...)
+		instanceConsoles := genModelInstancesConsole("console", 3)
 
 		ret := make(map[string]*model.Instance)
 		for id, instance := range instances1 {
@@ -403,6 +464,9 @@ func TestGetInstancesByServiceID(t *testing.T) {
 		gomock.InOrder(storage.EXPECT().
 			GetMoreInstances(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
 			Return(ret, nil))
+		gomock.InOrder(storage.EXPECT().
+			GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), ic.IsFirstUpdate(), ic.needMeta, ic.systemServiceID).
+			Return(instanceConsoles, nil))
 		gomock.InOrder(storage.EXPECT().
 			GetInstancesCountTx(gomock.Any()).
 			Return(uint32(instances1Count+instances2Count), nil))

cache/service/service_test.go

@@ -402,6 +402,7 @@ func TestServiceCache_GetServicesByFilter(t *testing.T) {
 		mockStore.EXPECT().GetInstancesCountTx(gomock.Any()).Return(uint32(len(instances)), nil).AnyTimes()
 		mockStore.EXPECT().GetMoreServices(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(services, nil).AnyTimes()
 		mockStore.EXPECT().GetMoreInstances(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(instances, nil).AnyTimes()
+		mockStore.EXPECT().GetMoreInstanceConsoles(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 		ic.setInstances(instances)
 
 		hostToService := make(map[string]string)

common/model/naming.go

@@ -864,6 +864,14 @@ type InstanceEvent struct {
 	MetaData   map[string]string
 }
 
+// instance info from console
+type InstanceConsole struct {
+	Id       string
+	Isolate  bool
+	Weight   int8
+	Metadata string
+}
+
 // InjectMetadata 从context中获取metadata并注入到事件对象
 func (i *InstanceEvent) InjectMetadata(ctx context.Context) {
 	value := ctx.Value(CtxEventKeyMetadata)

go.mod

@@ -83,7 +83,7 @@ require (
 require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/agiledragon/gomonkey/v2 v2.10.1
-	github.com/polarismesh/specification v1.4.1-alpha
+	github.com/polarismesh/specification v1.4.1
 )
 
 require (

go.sum

@@ -325,6 +325,8 @@ github.com/polarismesh/go-restful-openapi/v2 v2.0.0-20220928152401-083908d10219
 github.com/polarismesh/go-restful-openapi/v2 v2.0.0-20220928152401-083908d10219/go.mod h1:4WhwBysTom9Eoy0hQ4W69I0FmO+T0EpjEW9/5sgHoUk=
 github.com/polarismesh/specification v1.4.1-alpha h1:4y7/cDN50o0JrFyEu6j78u4trEsessKAjS/7YPtlwVo=
 github.com/polarismesh/specification v1.4.1-alpha/go.mod h1:rDvMMtl5qebPmqiBLNa5Ps0XtwkP31ZLirbH4kXA0YU=
+github.com/polarismesh/specification v1.4.1 h1:lTZqeyUhhWuKyr6NDKBwmUrNfcUDvKLxWT/uOq71T5A=
+github.com/polarismesh/specification v1.4.1/go.mod h1:rDvMMtl5qebPmqiBLNa5Ps0XtwkP31ZLirbH4kXA0YU=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=

store/boltdb/instance.go

@@ -572,6 +572,31 @@ func (i *instanceStore) GetMoreInstances(tx store.Tx, mtime time.Time, firstUpda
 	return toInstance(instances), nil
 }
 
+// GetMoreInstancesConsoles 获取增量修改数据
+func (ins *instanceStore) GetMoreInstanceConsoles(tx store.Tx, mtime time.Time, firstUpdate, needMeta bool,
+	serviceID []string) (map[string]*model.InstanceConsole, error) {
+	//Todo: get from boltdb
+	return nil, nil
+}
+
+// DeleteInstanceConsole 逻辑删除instanceConsole
+func (ins *instanceStore) DeleteInstanceConsole(instanceConsoleID string) error {
+	//Todo: delete from boltdb
+	return nil
+}
+
+// CleanInstanceConsole 物理删除instanceConsole
+func (ins *instanceStore) CleanInstanceConsole(instanceConsoleID string) error {
+	//Todo: clean from boltdb
+	return nil
+}
+
+// UpdateInstanceConsole 更新instanceConsole
+func (ins *instanceStore) UpdateInstanceConsole(instanceConsole *model.InstanceConsole) error {
+	//Todo: update from boltdb
+	return nil
+}
+
 // BatchSetInstanceHealthStatus 批量设置实例的健康状态
 func (i *instanceStore) BatchSetInstanceHealthStatus(ids []interface{}, healthy int, revision string) error {
 	for _, id := range ids {

store/discover_api.go

@@ -97,12 +97,15 @@ type InstanceStore interface {
 	BatchAddInstances(instances []*model.Instance) error
 	// UpdateInstance 更新实例
 	UpdateInstance(instance *model.Instance) error
+	UpdateInstanceConsole(instanceConsole *model.InstanceConsole) error
 	// DeleteInstance 删除一个实例，实际是把valid置为false
 	DeleteInstance(instanceID string) error
+	DeleteInstanceConsole(instanceConsoleID string) error
 	// BatchDeleteInstances 批量删除实例，flag=1
 	BatchDeleteInstances(ids []interface{}) error
 	// CleanInstance 清空一个实例，真正删除
 	CleanInstance(instanceID string) error
+	CleanInstanceConsole(instanceConsoleID string) error
 	// BatchGetInstanceIsolate 检查ID是否存在，并且返回存在的ID，以及ID的隔离状态
 	BatchGetInstanceIsolate(ids map[string]bool) (map[string]bool, error)
 	// GetInstancesBrief 获取实例关联的token
@@ -121,6 +124,7 @@ type InstanceStore interface {
 	// GetMoreInstances 根据mtime获取增量instances，返回所有store的变更信息
 	// 此方法用于 cache 增量更新，需要注意 mtime 应为数据库时间戳
 	GetMoreInstances(tx Tx, mtime time.Time, firstUpdate, needMeta bool, serviceID []string) (map[string]*model.Instance, error)
+	GetMoreInstanceConsoles(tx Tx, mtime time.Time, firstUpdate, needMeta bool, serviceID []string) (map[string]*model.InstanceConsole, error)
 	// SetInstanceHealthStatus 设置实例的健康状态
 	SetInstanceHealthStatus(instanceID string, flag int, revision string) error
 	// BatchSetInstanceHealthStatus 批量设置实例的健康状态