mirror of https://github.com/grafana/loki
make tsdb store instance a singleton (#6273)
* make tsdb store instance a singleton * make it mandatory to have 24h index period config for tsdb * avoid initializing head and tsdb manager when running tsdb store in read only modepull/6302/head^2
Sandeep Sukhani
4 years ago
committed by
GitHub
parent
3c1accccb1
commit
fefc33debc
@ -0,0 +1,112 @@ |
||||
package tsdb |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"math" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"github.com/prometheus/prometheus/model/labels" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk" |
||||
"github.com/grafana/loki/pkg/storage/stores/indexshipper" |
||||
shipper_index "github.com/grafana/loki/pkg/storage/stores/indexshipper/index" |
||||
"github.com/grafana/loki/pkg/storage/stores/tsdb/index" |
||||
) |
||||
|
||||
// indexShipperQuerier is used for querying index from the shipper.
|
||||
type indexShipperQuerier struct { |
||||
shipper indexshipper.IndexShipper |
||||
chunkFilter chunk.RequestChunkFilterer |
||||
} |
||||
|
||||
func newIndexShipperQuerier(shipper indexshipper.IndexShipper) Index { |
||||
return &indexShipperQuerier{shipper: shipper} |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) indices(ctx context.Context, from, through model.Time, user string) (Index, error) { |
||||
var indices []Index |
||||
|
||||
// Ensure we query both per tenant and multitenant TSDBs
|
||||
|
||||
for _, bkt := range indexBuckets(from, through) { |
||||
if err := i.shipper.ForEach(ctx, fmt.Sprintf("%d", bkt), user, func(idx shipper_index.Index) error { |
||||
_, multitenant := parseMultitenantTSDBName(idx.Name()) |
||||
impl, ok := idx.(Index) |
||||
if !ok { |
||||
return fmt.Errorf("unexpected shipper index type: %T", idx) |
||||
} |
||||
if multitenant { |
||||
indices = append(indices, NewMultiTenantIndex(impl)) |
||||
} else { |
||||
indices = append(indices, impl) |
||||
} |
||||
return nil |
||||
}); err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
} |
||||
|
||||
if len(indices) == 0 { |
||||
return NoopIndex{}, nil |
||||
} |
||||
idx, err := NewMultiIndex(indices...) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
if i.chunkFilter != nil { |
||||
idx.SetChunkFilterer(i.chunkFilter) |
||||
} |
||||
return idx, nil |
||||
} |
||||
|
||||
// TODO(owen-d): how to better implement this?
|
||||
// setting 0->maxint will force the tsdbmanager to always query
|
||||
// underlying tsdbs, which is safe, but can we optimize this?
|
||||
func (i *indexShipperQuerier) Bounds() (model.Time, model.Time) { |
||||
return 0, math.MaxInt64 |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) SetChunkFilterer(chunkFilter chunk.RequestChunkFilterer) { |
||||
i.chunkFilter = chunkFilter |
||||
} |
||||
|
||||
// Close implements Index.Close, but we offload this responsibility
|
||||
// to the index shipper
|
||||
func (i *indexShipperQuerier) Close() error { |
||||
return nil |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, shard *index.ShardAnnotation, matchers ...*labels.Matcher) ([]ChunkRef, error) { |
||||
idx, err := i.indices(ctx, from, through, userID) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return idx.GetChunkRefs(ctx, userID, from, through, res, shard, matchers...) |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) Series(ctx context.Context, userID string, from, through model.Time, res []Series, shard *index.ShardAnnotation, matchers ...*labels.Matcher) ([]Series, error) { |
||||
idx, err := i.indices(ctx, from, through, userID) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return idx.Series(ctx, userID, from, through, res, shard, matchers...) |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) LabelNames(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([]string, error) { |
||||
idx, err := i.indices(ctx, from, through, userID) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return idx.LabelNames(ctx, userID, from, through, matchers...) |
||||
} |
||||
|
||||
func (i *indexShipperQuerier) LabelValues(ctx context.Context, userID string, from, through model.Time, name string, matchers ...*labels.Matcher) ([]string, error) { |
||||
idx, err := i.indices(ctx, from, through, userID) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return idx.LabelValues(ctx, userID, from, through, name, matchers...) |
||||
} |
||||
Loading…
Reference in new issue