mirror of https://github.com/grafana/loki
(chore) Bloomshipper: Separate store and client (#11865)
**What this PR does / why we need it**: This PR removes the `StoreAndClient` interface that was accepted by the `BloomShipper`. Since the `BloomStore` had to not only implement the `Store` interface, but also the `Client` interface, it caused re-implementation of the same methods in different ways. Now the shipper solely relies on the `Store` interface. See individual commit messages for more context. Tests have been rewritten from scratch and placed in their own respective test files for store and client. --------- Signed-off-by: Christian Haudum <christian.haudum@gmail.com>pull/11756/head
Christian Haudum
2 years ago
committed by
GitHub
parent
2e3fa3b861
commit
73edf7a943
@ -0,0 +1,268 @@ |
||||
package bloomshipper |
||||
|
||||
import ( |
||||
"bytes" |
||||
"context" |
||||
"encoding/json" |
||||
"os" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/go-kit/log" |
||||
"github.com/prometheus/client_golang/prometheus" |
||||
"github.com/prometheus/common/model" |
||||
"github.com/stretchr/testify/require" |
||||
|
||||
"github.com/grafana/loki/pkg/storage" |
||||
v1 "github.com/grafana/loki/pkg/storage/bloom/v1" |
||||
"github.com/grafana/loki/pkg/storage/chunk/cache" |
||||
storageconfig "github.com/grafana/loki/pkg/storage/config" |
||||
"github.com/grafana/loki/pkg/storage/stores/shipper/bloomshipper/config" |
||||
) |
||||
|
||||
func newMockBloomStore(t *testing.T) (*BloomStore, string) { |
||||
workDir := t.TempDir() |
||||
|
||||
periodicConfigs := []storageconfig.PeriodConfig{ |
||||
{ |
||||
ObjectType: storageconfig.StorageTypeInMemory, |
||||
From: parseDayTime("2024-01-01"), |
||||
IndexTables: storageconfig.IndexPeriodicTableConfig{ |
||||
PeriodicTableConfig: storageconfig.PeriodicTableConfig{ |
||||
Period: 24 * time.Hour, |
||||
// TODO(chaudum): Integrate {,Parse}MetaKey into schema config
|
||||
// Prefix: "schema_a_table_",
|
||||
}}, |
||||
}, |
||||
{ |
||||
ObjectType: storageconfig.StorageTypeInMemory, |
||||
From: parseDayTime("2024-02-01"), |
||||
IndexTables: storageconfig.IndexPeriodicTableConfig{ |
||||
PeriodicTableConfig: storageconfig.PeriodicTableConfig{ |
||||
Period: 24 * time.Hour, |
||||
// TODO(chaudum): Integrate {,Parse}MetaKey into schema config
|
||||
// Prefix: "schema_b_table_",
|
||||
}}, |
||||
}, |
||||
} |
||||
|
||||
storageConfig := storage.Config{ |
||||
BloomShipperConfig: config.Config{ |
||||
WorkingDirectory: workDir, |
||||
BlocksDownloadingQueue: config.DownloadingQueueConfig{ |
||||
WorkersCount: 1, |
||||
}, |
||||
BlocksCache: config.BlocksCacheConfig{ |
||||
EmbeddedCacheConfig: cache.EmbeddedCacheConfig{ |
||||
MaxSizeItems: 1000, |
||||
TTL: 1 * time.Hour, |
||||
}, |
||||
}, |
||||
}, |
||||
} |
||||
|
||||
metrics := storage.NewClientMetrics() |
||||
t.Cleanup(metrics.Unregister) |
||||
logger := log.NewLogfmtLogger(os.Stderr) |
||||
|
||||
metasCache := cache.NewMockCache() |
||||
blocksCache := NewBlocksCache(storageConfig.BloomShipperConfig, prometheus.NewPedanticRegistry(), logger) |
||||
store, err := NewBloomStore(periodicConfigs, storageConfig, metrics, metasCache, blocksCache, logger) |
||||
require.NoError(t, err) |
||||
t.Cleanup(store.Stop) |
||||
|
||||
return store, workDir |
||||
} |
||||
|
||||
func createMetaInStorage(store *BloomStore, tenant string, start model.Time, minFp, maxFp model.Fingerprint) (Meta, error) { |
||||
meta := Meta{ |
||||
MetaRef: MetaRef{ |
||||
Ref: Ref{ |
||||
TenantID: tenant, |
||||
Bounds: v1.NewBounds(minFp, maxFp), |
||||
// Unused
|
||||
// StartTimestamp: start,
|
||||
// EndTimestamp: start.Add(12 * time.Hour),
|
||||
}, |
||||
}, |
||||
Blocks: []BlockRef{}, |
||||
Tombstones: []BlockRef{}, |
||||
} |
||||
err := store.storeDo(start, func(s *bloomStoreEntry) error { |
||||
raw, _ := json.Marshal(meta) |
||||
meta.MetaRef.Ref.TableName = tablesForRange(s.cfg, NewInterval(start, start.Add(12*time.Hour)))[0] |
||||
return s.objectClient.PutObject(context.Background(), s.Meta(meta.MetaRef).Addr(), bytes.NewReader(raw)) |
||||
}) |
||||
return meta, err |
||||
} |
||||
|
||||
func createBlockInStorage(t *testing.T, store *BloomStore, tenant string, start model.Time, minFp, maxFp model.Fingerprint) (Block, error) { |
||||
tmpDir := t.TempDir() |
||||
fp, _ := os.CreateTemp(t.TempDir(), "*.tar.gz") |
||||
|
||||
blockWriter := v1.NewDirectoryBlockWriter(tmpDir) |
||||
err := blockWriter.Init() |
||||
require.NoError(t, err) |
||||
|
||||
err = v1.TarGz(fp, v1.NewDirectoryBlockReader(tmpDir)) |
||||
require.NoError(t, err) |
||||
|
||||
_, _ = fp.Seek(0, 0) |
||||
|
||||
block := Block{ |
||||
BlockRef: BlockRef{ |
||||
Ref: Ref{ |
||||
TenantID: tenant, |
||||
Bounds: v1.NewBounds(minFp, maxFp), |
||||
StartTimestamp: start, |
||||
EndTimestamp: start.Add(12 * time.Hour), |
||||
}, |
||||
}, |
||||
Data: fp, |
||||
} |
||||
err = store.storeDo(start, func(s *bloomStoreEntry) error { |
||||
block.BlockRef.Ref.TableName = tablesForRange(s.cfg, NewInterval(start, start.Add(12*time.Hour)))[0] |
||||
return s.objectClient.PutObject(context.Background(), s.Block(block.BlockRef).Addr(), block.Data) |
||||
}) |
||||
return block, err |
||||
} |
||||
|
||||
func TestBloomStore_ResolveMetas(t *testing.T) { |
||||
store, _ := newMockBloomStore(t) |
||||
|
||||
// schema 1
|
||||
// outside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-19 00:00"), 0x00010000, 0x0001ffff) |
||||
// outside of interval, inside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-19 00:00"), 0x00000000, 0x0000ffff) |
||||
// inside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-20 00:00"), 0x00010000, 0x0001ffff) |
||||
// inside of interval, inside of bounds
|
||||
m1, _ := createMetaInStorage(store, "tenant", parseTime("2024-01-20 00:00"), 0x00000000, 0x0000ffff) |
||||
|
||||
// schema 2
|
||||
// inside of interval, inside of bounds
|
||||
m2, _ := createMetaInStorage(store, "tenant", parseTime("2024-02-05 00:00"), 0x00000000, 0x0000ffff) |
||||
// inside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-05 00:00"), 0x00010000, 0x0001ffff) |
||||
// outside of interval, inside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-11 00:00"), 0x00000000, 0x0000ffff) |
||||
// outside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-11 00:00"), 0x00010000, 0x0001ffff) |
||||
|
||||
t.Run("tenant matches", func(t *testing.T) { |
||||
ctx := context.Background() |
||||
params := MetaSearchParams{ |
||||
"tenant", |
||||
NewInterval(parseTime("2024-01-20 00:00"), parseTime("2024-02-10 00:00")), |
||||
v1.NewBounds(0x00000000, 0x0000ffff), |
||||
} |
||||
|
||||
refs, fetchers, err := store.ResolveMetas(ctx, params) |
||||
require.NoError(t, err) |
||||
require.Len(t, refs, 2) |
||||
require.Len(t, fetchers, 2) |
||||
|
||||
require.Equal(t, [][]MetaRef{{m1.MetaRef}, {m2.MetaRef}}, refs) |
||||
}) |
||||
|
||||
t.Run("tenant does not match", func(t *testing.T) { |
||||
ctx := context.Background() |
||||
params := MetaSearchParams{ |
||||
"other", |
||||
NewInterval(parseTime("2024-01-20 00:00"), parseTime("2024-02-10 00:00")), |
||||
v1.NewBounds(0x00000000, 0x0000ffff), |
||||
} |
||||
|
||||
refs, fetchers, err := store.ResolveMetas(ctx, params) |
||||
require.NoError(t, err) |
||||
require.Len(t, refs, 0) |
||||
require.Len(t, fetchers, 0) |
||||
require.Equal(t, [][]MetaRef{}, refs) |
||||
}) |
||||
} |
||||
|
||||
func TestBloomStore_FetchMetas(t *testing.T) { |
||||
store, _ := newMockBloomStore(t) |
||||
|
||||
// schema 1
|
||||
// outside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-19 00:00"), 0x00010000, 0x0001ffff) |
||||
// outside of interval, inside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-19 00:00"), 0x00000000, 0x0000ffff) |
||||
// inside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-01-20 00:00"), 0x00010000, 0x0001ffff) |
||||
// inside of interval, inside of bounds
|
||||
m1, _ := createMetaInStorage(store, "tenant", parseTime("2024-01-20 00:00"), 0x00000000, 0x0000ffff) |
||||
|
||||
// schema 2
|
||||
// inside of interval, inside of bounds
|
||||
m2, _ := createMetaInStorage(store, "tenant", parseTime("2024-02-05 00:00"), 0x00000000, 0x0000ffff) |
||||
// inside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-05 00:00"), 0x00010000, 0x0001ffff) |
||||
// outside of interval, inside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-11 00:00"), 0x00000000, 0x0000ffff) |
||||
// outside of interval, outside of bounds
|
||||
_, _ = createMetaInStorage(store, "tenant", parseTime("2024-02-11 00:00"), 0x00010000, 0x0001ffff) |
||||
|
||||
t.Run("tenant matches", func(t *testing.T) { |
||||
ctx := context.Background() |
||||
params := MetaSearchParams{ |
||||
"tenant", |
||||
NewInterval(parseTime("2024-01-20 00:00"), parseTime("2024-02-10 00:00")), |
||||
v1.NewBounds(0x00000000, 0x0000ffff), |
||||
} |
||||
|
||||
metas, err := store.FetchMetas(ctx, params) |
||||
require.NoError(t, err) |
||||
require.Len(t, metas, 2) |
||||
|
||||
require.Equal(t, []Meta{m1, m2}, metas) |
||||
}) |
||||
|
||||
t.Run("tenant does not match", func(t *testing.T) { |
||||
ctx := context.Background() |
||||
params := MetaSearchParams{ |
||||
"other", |
||||
NewInterval(parseTime("2024-01-20 00:00"), parseTime("2024-02-10 00:00")), |
||||
v1.NewBounds(0x00000000, 0x0000ffff), |
||||
} |
||||
|
||||
metas, err := store.FetchMetas(ctx, params) |
||||
require.NoError(t, err) |
||||
require.Len(t, metas, 0) |
||||
require.Equal(t, []Meta{}, metas) |
||||
}) |
||||
} |
||||
|
||||
func TestBloomStore_FetchBlocks(t *testing.T) { |
||||
store, _ := newMockBloomStore(t) |
||||
|
||||
// schema 1
|
||||
b1, _ := createBlockInStorage(t, store, "tenant", parseTime("2024-01-20 00:00"), 0x00000000, 0x0000ffff) |
||||
b2, _ := createBlockInStorage(t, store, "tenant", parseTime("2024-01-20 00:00"), 0x00010000, 0x0001ffff) |
||||
// schema 2
|
||||
b3, _ := createBlockInStorage(t, store, "tenant", parseTime("2024-02-05 00:00"), 0x00000000, 0x0000ffff) |
||||
b4, _ := createBlockInStorage(t, store, "tenant", parseTime("2024-02-05 00:00"), 0x00000000, 0x0001ffff) |
||||
|
||||
ctx := context.Background() |
||||
|
||||
// first call fetches two blocks from cache
|
||||
blockDirs, err := store.FetchBlocks(ctx, []BlockRef{b1.BlockRef, b3.BlockRef}) |
||||
require.NoError(t, err) |
||||
require.Len(t, blockDirs, 2) |
||||
|
||||
require.ElementsMatch(t, []BlockRef{b1.BlockRef, b3.BlockRef}, []BlockRef{blockDirs[0].BlockRef, blockDirs[1].BlockRef}) |
||||
|
||||
// second call fetches two blocks from cache and two from storage
|
||||
blockDirs, err = store.FetchBlocks(ctx, []BlockRef{b1.BlockRef, b2.BlockRef, b3.BlockRef, b4.BlockRef}) |
||||
require.NoError(t, err) |
||||
require.Len(t, blockDirs, 4) |
||||
|
||||
// Note the order: b1 and b2 come from cache, so they are in the beginning of the response
|
||||
// Do we need to sort the response based on the request order of block refs?
|
||||
require.ElementsMatch(t, |
||||
[]BlockRef{b1.BlockRef, b3.BlockRef, b2.BlockRef, b4.BlockRef}, |
||||
[]BlockRef{blockDirs[0].BlockRef, blockDirs[1].BlockRef, blockDirs[2].BlockRef, blockDirs[3].BlockRef}, |
||||
) |
||||
} |
||||
Loading…
Reference in new issue