-
Notifications
You must be signed in to change notification settings - Fork 282
/
field_store.go
151 lines (132 loc) · 4.68 KB
/
field_store.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
package memdb
import (
"sort"
"github.com/eleme/lindb/pkg/field"
"github.com/eleme/lindb/pkg/logger"
"github.com/eleme/lindb/pkg/timeutil"
pb "github.com/eleme/lindb/rpc/proto/field"
"github.com/eleme/lindb/tsdb/metrictbl"
)
//go:generate mockgen -source ./field_store.go -destination=./field_store_mock_test.go -package memdb
// fStoreINTF abstracts a field-store
type fStoreINTF interface {
// getFieldType returns the field-type
getFieldType() field.Type
// write writes the metric's field with writeContext
write(f *pb.Field, writeCtx writeContext)
// flushFieldTo flushes field data of the specific familyTime
// return false if there is no data related of familyTime
flushFieldTo(tableFlusher metrictbl.TableFlusher, familyTime int64) (flushed bool)
// timeRange returns the start-time and end-time of fStore's data
// ok means data is available
timeRange(interval int64) (timeRange timeutil.TimeRange, ok bool)
}
// sStoreNodes implements the sort.Interface
type sStoreNodes []sStoreNode
func (s sStoreNodes) Len() int { return len(s) }
func (s sStoreNodes) Less(i, j int) bool { return s[i].familyTime < s[j].familyTime }
func (s sStoreNodes) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
// sStoreNode wraps sStoreINTF with familyTime
type sStoreNode struct {
familyTime int64
sStoreINTF
}
// fieldStore holds the relation of familyStartTime and segmentStore.
// there are only a few familyTimes in the segments,
// add delete operation occurs every one hour
// so slice is more cheaper than the map
type fieldStore struct {
fieldType field.Type // sum, gauge, min, max
fieldID uint16 // generated by id generator
sStoreNodes sStoreNodes // sorted sStore list by family-time
}
// newFieldStore returns a new fieldStore.
func newFieldStore(fieldID uint16, fieldType field.Type) fStoreINTF {
return &fieldStore{
fieldID: fieldID,
fieldType: fieldType}
}
// getSStore gets the sStore from list by familyTime.
func (fs *fieldStore) getSStore(familyTime int64) (sStoreINTF, bool) {
idx := sort.Search(len(fs.sStoreNodes), func(i int) bool {
return fs.sStoreNodes[i].familyTime >= familyTime
})
if idx >= len(fs.sStoreNodes) || fs.sStoreNodes[idx].familyTime != familyTime {
return nil, false
}
return fs.sStoreNodes[idx], true
}
// removeSStore removes the sStore by familyTime.
func (fs *fieldStore) removeSStore(familyTime int64) {
idx := sort.Search(len(fs.sStoreNodes), func(i int) bool {
return fs.sStoreNodes[i].familyTime >= familyTime
})
// familyTime greater than existed
if idx == len(fs.sStoreNodes) {
return
}
// not match
if fs.sStoreNodes[idx].familyTime != familyTime {
return
}
copy(fs.sStoreNodes[idx:], fs.sStoreNodes[idx+1:])
// fills the tail with empty sStore node
fs.sStoreNodes[len(fs.sStoreNodes)-1] = sStoreNode{}
fs.sStoreNodes = fs.sStoreNodes[:len(fs.sStoreNodes)-1]
}
// insertSStore inserts a new sStore to segments.
func (fs *fieldStore) insertSStore(familyTime int64, sStore sStoreINTF) {
fs.sStoreNodes = append(fs.sStoreNodes, sStoreNode{familyTime: familyTime, sStoreINTF: sStore})
sort.Sort(fs.sStoreNodes)
}
// getFieldType returns field type for current field store
func (fs *fieldStore) getFieldType() field.Type {
return fs.fieldType
}
func (fs *fieldStore) write(f *pb.Field, writeCtx writeContext) {
sStore, ok := fs.getSStore(writeCtx.familyTime)
switch fields := f.Field.(type) {
case *pb.Field_Sum:
if !ok {
//TODO ???
sStore = newSimpleFieldStore(field.GetAggFunc(field.Sum))
fs.insertSStore(writeCtx.familyTime, sStore)
}
sStore.writeFloat(fields.Sum, writeCtx)
default:
memDBLogger.Warn("convert field error, unknown field type")
}
}
// flushFieldTo flushes segments' data to writer and reset the segments-map.
func (fs *fieldStore) flushFieldTo(tableFlusher metrictbl.TableFlusher, familyTime int64) (flushed bool) {
sStore, ok := fs.getSStore(familyTime)
if !ok {
return false
}
fs.removeSStore(familyTime)
data, startSlot, endSlot, err := sStore.bytes()
if err != nil {
memDBLogger.Error("read segment data error:", logger.Error(err))
return false
}
tableFlusher.FlushField(fs.fieldID, fs.fieldType, data, startSlot, endSlot)
return true
}
func (fs *fieldStore) timeRange(interval int64) (timeRange timeutil.TimeRange, ok bool) {
for _, sStoreNode := range fs.sStoreNodes {
startSlot, endSlot, err := sStoreNode.sStoreINTF.slotRange()
if err != nil {
continue
}
ok = true
startTime := sStoreNode.familyTime + int64(startSlot)*interval
endTime := sStoreNode.familyTime + int64(endSlot)*interval
if timeRange.Start == 0 || startTime < timeRange.Start {
timeRange.Start = startTime
}
if timeRange.End == 0 || timeRange.End < endTime {
timeRange.End = endTime
}
}
return
}