Create table and related cache for additional instance data

apiserver/eurekaserver/write_test.go

@@ -109,6 +109,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()).

cache/service/instance.go

@@ -51,7 +51,9 @@
 	// service id -> [instanceid ->instance]
 	services *utils.SyncMap[string, *model.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
@@ -78,6 +80,7 @@
 	ic.ids = utils.NewSyncMap[string, *model.Instance]()
 	ic.services = utils.NewSyncMap[string, *model.ServiceInstances]()
 	ic.instanceCounts = utils.NewSyncMap[string, *model.InstanceCount]()
+	ic.instanceConsoles = utils.NewSyncMap[string, *model.InstanceConsole]()
 	ic.instancePorts = newInstancePorts()
 	if opt == nil {
 		return nil
@@ -192,6 +195,17 @@
 		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),
@@ -205,6 +219,7 @@
 	ic.ids = utils.NewSyncMap[string, *model.Instance]()
 	ic.services = utils.NewSyncMap[string, *model.ServiceInstances]()
 	ic.instanceCounts = utils.NewSyncMap[string, *model.InstanceCount]()
+	ic.instanceConsoles = utils.NewSyncMap[string, *model.InstanceConsole]()
 	ic.instancePorts.reset()
 	ic.instanceCount = 0
 	return nil
@@ -434,6 +449,20 @@
 	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
+}
+
 // GetInstances 根据服务名获取实例，先查找服务名对应的服务ID，再找实例列表
 func (ic *instanceCache) GetInstances(serviceID string) *model.ServiceInstances {
 	if serviceID == "" {

cache/service/instance_test.go

@@ -94,6 +94,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
@@ -118,6 +133,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
@@ -129,13 +145,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)
@@ -147,26 +167,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 {
@@ -188,6 +216,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())
@@ -199,9 +230,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())
 		}
@@ -218,6 +253,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())
 		}
@@ -231,6 +269,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().
@@ -256,9 +295,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())
@@ -271,6 +315,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")
+		}
 	})
 }
 
@@ -280,6 +332,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)
 
@@ -311,6 +364,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())
@@ -333,6 +389,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)
 
@@ -362,6 +419,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())
@@ -393,6 +453,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 {
@@ -405,6 +466,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

@@ -406,6 +406,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

@@ -928,6 +928,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)

store/boltdb/instance.go

@@ -572,6 +572,31 @@
 	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

@@ -94,12 +94,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
@@ -118,6 +121,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 批量设置实例的健康状态