Skip to content

Commit 73c2c14

Browse files
ZhengYa-0110ZhengYa-0110
committed
feat: modify index definition to be compatible with OB
1 parent b9ddf88 commit 73c2c14

5 files changed

Lines changed: 113 additions & 54 deletions

File tree

‎server/controller/prometheus/cache/label.go‎

Lines changed: 16 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -122,18 +122,24 @@ func (l *label) Add(entries []LabelCacheEntry) {
122122
}
123123

124124
func (l *label) refresh(args ...interface{}) error {
125-
data, err := l.load()
125+
rows, err := l.org.DB.Model(&metadbmodel.PrometheusLabel{}).Select("id", "name_id", "value_id").Rows()
126126
if err != nil {
127127
return err
128128
}
129-
l.processLoadedData(data)
130-
return nil
131-
}
132-
133-
func (l *label) processLoadedData(data []*metadbmodel.PrometheusLabel) {
134-
newActive := make(map[LabelKey]int, len(data))
135-
for _, item := range data {
136-
newActive[NewLabelKey(item.NameID, item.ValueID)] = item.ID
129+
defer rows.Close()
130+
131+
newActive := make(map[LabelKey]int)
132+
for rows.Next() {
133+
var id, nameID, valueID int
134+
if scanErr := rows.Scan(&id, &nameID, &valueID); scanErr != nil {
135+
log.Errorf("stream scan prometheus_label interrupted: %v", scanErr, l.org.LogPrefix)
136+
return scanErr
137+
}
138+
newActive[NewLabelKey(nameID, valueID)] = id
139+
}
140+
if err := rows.Err(); err != nil {
141+
log.Errorf("stream read prometheus_label error: %v", err, l.org.LogPrefix)
142+
return err
137143
}
138144

139145
l.mu.Lock()
@@ -145,10 +151,5 @@ func (l *label) processLoadedData(data []*metadbmodel.PrometheusLabel) {
145151
l.mu.Unlock()
146152

147153
l.replaceActive(newActive)
148-
}
149-
150-
func (l *label) load() ([]*metadbmodel.PrometheusLabel, error) {
151-
var labels []*metadbmodel.PrometheusLabel
152-
err := l.org.DB.Select("id", "name_id", "value_id").Find(&labels).Error
153-
return labels, err
154+
return nil
154155
}

‎server/controller/prometheus/cache/label_value.go‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -109,16 +109,27 @@ func (lv *labelValue) Add(batch []*controller.PrometheusLabelValue) {
109109
}
110110

111111
func (lv *labelValue) refresh(args ...interface{}) error {
112-
data, err := lv.load()
112+
rows, err := lv.org.DB.Model(&metadbmodel.PrometheusLabelValue{}).Select("id", "value").Rows()
113113
if err != nil {
114114
return err
115115
}
116-
117-
newActive := make(map[string]int, len(data))
118-
newActiveR := make(map[int]string, len(data))
119-
for _, item := range data {
120-
newActive[item.Value] = item.ID
121-
newActiveR[item.ID] = item.Value
116+
defer rows.Close()
117+
118+
newActive := make(map[string]int)
119+
newActiveR := make(map[int]string)
120+
for rows.Next() {
121+
var id int
122+
var value string
123+
if scanErr := rows.Scan(&id, &value); scanErr != nil {
124+
log.Errorf("stream scan prometheus_label_value interrupted: %v", scanErr, lv.org.LogPrefix)
125+
return scanErr
126+
}
127+
newActive[value] = id
128+
newActiveR[id] = value
129+
}
130+
if err := rows.Err(); err != nil {
131+
log.Errorf("stream read prometheus_label_value error: %v", err, lv.org.LogPrefix)
132+
return err
122133
}
123134

124135
lv.mu.Lock()
@@ -133,9 +144,3 @@ func (lv *labelValue) refresh(args ...interface{}) error {
133144
lv.replaceActive(newActive)
134145
return nil
135146
}
136-
137-
func (lv *labelValue) load() ([]*metadbmodel.PrometheusLabelValue, error) {
138-
var labelValues []*metadbmodel.PrometheusLabelValue
139-
err := lv.org.DB.Select("id", "value").Find(&labelValues).Error
140-
return labelValues, err
141-
}

‎server/controller/prometheus/encoder/label.go‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -66,15 +66,26 @@ func (l *label) refresh(args ...interface{}) error {
6666
}
6767
l.lock.Unlock()
6868

69-
var items []*metadbmodel.PrometheusLabel
70-
if err := l.org.DB.Select("id", "name_id", "value_id").Find(&items).Error; err != nil {
69+
log.Info("TODO start")
70+
rows, err := l.org.DB.Model(&metadbmodel.PrometheusLabel{}).Select("id", "name_id", "value_id").Rows()
71+
if err != nil {
7172
return err
7273
}
73-
newMap := make(map[cache.LabelKey]int, len(items))
74-
for _, item := range items {
75-
newMap[cache.NewLabelKey(item.NameID, item.ValueID)] = item.ID
74+
defer rows.Close()
75+
76+
newMap := make(map[cache.LabelKey]int)
77+
for rows.Next() {
78+
var id, nameID, valueID int
79+
if scanErr := rows.Scan(&id, &nameID, &valueID); scanErr != nil {
80+
return scanErr
81+
}
82+
newMap[cache.NewLabelKey(nameID, valueID)] = id
83+
}
84+
if err := rows.Err(); err != nil {
85+
return err
7686
}
7787

88+
log.Info("TODO end")
7889
l.lock.Lock()
7990
for k, v := range l.labelKeyToID {
8091
if _, wasInSnapshot := preKeys[k]; !wasInSnapshot {

‎server/controller/prometheus/encoder/label_value.go‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -52,16 +52,30 @@ func (lv *labelValue) refresh(args ...interface{}) error {
5252
}
5353
lv.lock.Unlock()
5454

55-
var items []*metadbmodel.PrometheusLabelValue
56-
if err := lv.org.DB.Select("id", "value").Find(&items).Error; err != nil {
55+
log.Info("TODO start")
56+
rows, err := lv.org.DB.Model(&metadbmodel.PrometheusLabelValue{}).Select("id", "value").Rows()
57+
if err != nil {
5758
log.Errorf("db query %s failed: %v", lv.resourceType, err, lv.org.LogPrefix)
5859
return err
5960
}
60-
newMap := make(map[string]int, len(items))
61-
for _, item := range items {
62-
newMap[item.Value] = item.ID
61+
defer rows.Close()
62+
63+
newMap := make(map[string]int)
64+
for rows.Next() {
65+
var id int
66+
var value string
67+
if scanErr := rows.Scan(&id, &value); scanErr != nil {
68+
log.Errorf("db stream scan %s interrupted: %v", lv.resourceType, scanErr, lv.org.LogPrefix)
69+
return scanErr
70+
}
71+
newMap[value] = id
72+
}
73+
if err := rows.Err(); err != nil {
74+
log.Errorf("db stream %s error: %v", lv.resourceType, err, lv.org.LogPrefix)
75+
return err
6376
}
6477

78+
log.Info("TODO end")
6579
lv.lock.Lock()
6680
for k, v := range lv.strToID {
6781
if _, wasInSnapshot := preKeys[k]; !wasInSnapshot {

‎server/controller/tagrecorder/ch_app_label.go‎

Lines changed: 44 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -40,13 +40,26 @@ func NewChAPPLabel() *ChAPPLabel {
4040

4141
func (l *ChAPPLabel) generateNewData(db *metadb.DB) (map[PrometheusAPPLabelKey]metadbmodel.ChAPPLabel, bool) {
4242
log.Infof("generate data for %s", l.resourceTypeName, db.LogPrefixORGID)
43-
var prometheusLabels []metadbmodel.PrometheusLabel
44-
err := db.Unscoped().Find(&prometheusLabels).Error
45-
43+
labelRows, err := db.Unscoped().Model(&metadbmodel.PrometheusLabel{}).Select("id", "name_id", "value_id").Rows()
4644
if err != nil {
4745
log.Errorf(dbQueryResourceFailed(l.resourceTypeName, err), db.LogPrefixORGID)
4846
return nil, false
4947
}
48+
defer labelRows.Close()
49+
50+
var prometheusLabels []metadbmodel.PrometheusLabel
51+
for labelRows.Next() {
52+
var item metadbmodel.PrometheusLabel
53+
if scanErr := labelRows.Scan(&item.ID, &item.NameID, &item.ValueID); scanErr != nil {
54+
log.Errorf("stream scan %s prometheus_label interrupted: %v", l.resourceTypeName, scanErr, db.LogPrefixORGID)
55+
return nil, false
56+
}
57+
prometheusLabels = append(prometheusLabels, item)
58+
}
59+
if err := labelRows.Err(); err != nil {
60+
log.Errorf("stream read %s prometheus_label error: %v", l.resourceTypeName, err, db.LogPrefixORGID)
61+
return nil, false
62+
}
5063

5164
_, ok := l.generateAPPLabelData(db)
5265

@@ -111,29 +124,44 @@ func (l *ChAPPLabel) generateAPPLabelData(db *metadb.DB) ([]string, bool) {
111124
func (l *ChAPPLabel) generateNameIDData(db *metadb.DB) (map[string]int, map[string]int, bool) {
112125
labelNameIDMap := make(map[string]int)
113126
valueNameIDMap := make(map[string]int)
114-
var prometheusLabelNames []metadbmodel.PrometheusLabelName
115-
var prometheusLabelValues []metadbmodel.PrometheusLabelValue
116-
117-
err := db.Unscoped().Find(&prometheusLabelNames).Error
118-
127+
nameRows, err := db.Unscoped().Model(&metadbmodel.PrometheusLabelName{}).Select("id", "name").Rows()
119128
if err != nil {
120129
log.Errorf(dbQueryResourceFailed(l.resourceTypeName, err), db.LogPrefixORGID)
121130
return nil, nil, false
122131
}
132+
defer nameRows.Close()
133+
for nameRows.Next() {
134+
var id int
135+
var name string
136+
if scanErr := nameRows.Scan(&id, &name); scanErr != nil {
137+
log.Errorf("stream scan %s prometheus_label_name interrupted: %v", l.resourceTypeName, scanErr, db.LogPrefixORGID)
138+
return nil, nil, false
139+
}
140+
labelNameIDMap[name] = id
141+
}
142+
if err := nameRows.Err(); err != nil {
143+
log.Errorf("stream read %s prometheus_label_name error: %v", l.resourceTypeName, err, db.LogPrefixORGID)
144+
return nil, nil, false
145+
}
123146

124-
err = db.Unscoped().Find(&prometheusLabelValues).Error
125-
147+
valueRows, err := db.Unscoped().Model(&metadbmodel.PrometheusLabelValue{}).Select("id", "value").Rows()
126148
if err != nil {
127149
log.Errorf(dbQueryResourceFailed(l.resourceTypeName, err), db.LogPrefixORGID)
128150
return nil, nil, false
129151
}
130-
131-
for _, prometheusLabelName := range prometheusLabelNames {
132-
labelNameIDMap[prometheusLabelName.Name] = prometheusLabelName.ID
152+
defer valueRows.Close()
153+
for valueRows.Next() {
154+
var id int
155+
var value string
156+
if scanErr := valueRows.Scan(&id, &value); scanErr != nil {
157+
log.Errorf("stream scan %s prometheus_label_value interrupted: %v", l.resourceTypeName, scanErr, db.LogPrefixORGID)
158+
return nil, nil, false
159+
}
160+
valueNameIDMap[value] = id
133161
}
134-
135-
for _, prometheusLabelValue := range prometheusLabelValues {
136-
valueNameIDMap[prometheusLabelValue.Value] = prometheusLabelValue.ID
162+
if err := valueRows.Err(); err != nil {
163+
log.Errorf("stream read %s prometheus_label_value error: %v", l.resourceTypeName, err, db.LogPrefixORGID)
164+
return nil, nil, false
137165
}
138166
return labelNameIDMap, valueNameIDMap, true
139167
}

0 commit comments

Comments
 (0)