forked from grafana/loki
/
store.go
157 lines (137 loc) · 4.18 KB
/
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
152
153
154
155
156
157
package tsdb
import (
"fmt"
"sync"
"github.com/go-kit/log/level"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/pkg/storage/chunk/client"
"github.com/grafana/loki/pkg/storage/chunk/fetcher"
"github.com/grafana/loki/pkg/storage/config"
"github.com/grafana/loki/pkg/storage/stores"
"github.com/grafana/loki/pkg/storage/stores/indexshipper"
"github.com/grafana/loki/pkg/storage/stores/indexshipper/downloads"
"github.com/grafana/loki/pkg/storage/stores/series"
"github.com/grafana/loki/pkg/storage/stores/tsdb/index"
util_log "github.com/grafana/loki/pkg/util/log"
)
type store struct {
indexShipper indexshipper.IndexShipper
indexWriter IndexWriter
indexStore series.IndexStore
stopOnce sync.Once
}
type newStoreFactoryFunc func(
indexShipperCfg indexshipper.Config,
p config.PeriodConfig,
f *fetcher.Fetcher,
objectClient client.ObjectClient,
limits downloads.Limits,
tableRanges config.TableRanges,
reg prometheus.Registerer,
) (
chunkWriter stores.ChunkWriter,
indexStore series.IndexStore,
stopFunc func(),
err error,
)
// NewStore creates a new store if not initialized already.
// Each call to NewStore will always build a new stores.ChunkWriter even if the store was already initialized since
// fetcher.Fetcher instances could be different due to periodic configs having different types of object storage configured
// for storing chunks.
// It also helps us make tsdb store a singleton because
// we do not need to build store for each schema config since we do not do any schema specific handling yet.
// If we do need to do schema specific handling, it would be a good idea to abstract away the handling since
// running multiple head managers would be complicated and wasteful.
var NewStore = func() newStoreFactoryFunc {
var storeInstance *store
return func(
indexShipperCfg indexshipper.Config,
p config.PeriodConfig,
f *fetcher.Fetcher,
objectClient client.ObjectClient,
limits downloads.Limits,
tableRanges config.TableRanges,
reg prometheus.Registerer,
) (
stores.ChunkWriter,
series.IndexStore,
func(),
error,
) {
if storeInstance == nil {
storeInstance = &store{}
err := storeInstance.init(indexShipperCfg, objectClient, limits, tableRanges, reg)
if err != nil {
return nil, nil, nil, err
}
}
return NewChunkWriter(f, p, storeInstance.indexWriter), storeInstance.indexStore, storeInstance.Stop, nil
}
}()
func (s *store) init(indexShipperCfg indexshipper.Config, objectClient client.ObjectClient,
limits downloads.Limits, tableRanges config.TableRanges, reg prometheus.Registerer) error {
var err error
s.indexShipper, err = indexshipper.NewIndexShipper(
indexShipperCfg,
objectClient,
limits,
nil,
OpenShippableTSDB,
tableRanges,
prometheus.WrapRegistererWithPrefix("loki_tsdb_shipper_", reg),
)
if err != nil {
return err
}
var indices []Index
if indexShipperCfg.Mode != indexshipper.ModeReadOnly {
var (
nodeName = indexShipperCfg.IngesterName
dir = indexShipperCfg.ActiveIndexDirectory
)
tsdbMetrics := NewMetrics(reg)
tsdbManager := NewTSDBManager(
nodeName,
dir,
s.indexShipper,
tableRanges,
util_log.Logger,
tsdbMetrics,
)
headManager := NewHeadManager(
util_log.Logger,
dir,
tsdbMetrics,
tsdbManager,
)
if err := headManager.Start(); err != nil {
return err
}
s.indexWriter = headManager
indices = append(indices, headManager)
} else {
s.indexWriter = failingIndexWriter{}
}
indices = append(indices, newIndexShipperQuerier(s.indexShipper, tableRanges))
multiIndex, err := NewMultiIndex(indices...)
if err != nil {
return err
}
s.indexStore = NewIndexClient(multiIndex)
return nil
}
func (s *store) Stop() {
s.stopOnce.Do(func() {
if hm, ok := s.indexWriter.(*HeadManager); ok {
if err := hm.Stop(); err != nil {
level.Error(util_log.Logger).Log("msg", "failed to stop head manager", "err", err)
}
}
s.indexShipper.Stop()
})
}
type failingIndexWriter struct{}
func (f failingIndexWriter) Append(_ string, _ labels.Labels, _ index.ChunkMetas) error {
return fmt.Errorf("index writer is not initialized due to tsdb store being initialized in read-only mode")
}