mirror of https://github.com/grafana/loki
Index shipper move code (#6704)
* move compactor code to generic indexshipper * move table client to generic indexshipper * move storage package and relevant code for managing files on storage to generic indexshipper * add missing files * lintpull/6728/head^2
Sandeep Sukhani
3 years ago
committed by
GitHub
parent
0d99d80acd
commit
882010a810
@ -0,0 +1,114 @@ |
||||
package storage |
||||
|
||||
import ( |
||||
"errors" |
||||
"fmt" |
||||
"io" |
||||
"os" |
||||
"strings" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/go-kit/log" |
||||
"github.com/go-kit/log/level" |
||||
gzip "github.com/klauspost/pgzip" |
||||
) |
||||
|
||||
var ( |
||||
gzipReader = sync.Pool{} |
||||
) |
||||
|
||||
// getGzipReader gets or creates a new CompressionReader and reset it to read from src
|
||||
func getGzipReader(src io.Reader) (io.Reader, error) { |
||||
if r := gzipReader.Get(); r != nil { |
||||
reader := r.(*gzip.Reader) |
||||
err := reader.Reset(src) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return reader, nil |
||||
} |
||||
reader, err := gzip.NewReader(src) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return reader, nil |
||||
} |
||||
|
||||
// putGzipReader places back in the pool a CompressionReader
|
||||
func putGzipReader(reader io.Reader) { |
||||
gzipReader.Put(reader) |
||||
} |
||||
|
||||
type GetFileFunc func() (io.ReadCloser, error) |
||||
|
||||
// DownloadFileFromStorage downloads a file from storage to given location.
|
||||
func DownloadFileFromStorage(destination string, decompressFile bool, sync bool, logger log.Logger, getFileFunc GetFileFunc) error { |
||||
start := time.Now() |
||||
readCloser, err := getFileFunc() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
defer func() { |
||||
if err := readCloser.Close(); err != nil { |
||||
level.Error(logger).Log("msg", "failed to close read closer", "err", err) |
||||
} |
||||
}() |
||||
|
||||
f, err := os.Create(destination) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
defer func() { |
||||
if err := f.Close(); err != nil { |
||||
level.Warn(logger).Log("msg", "failed to close file", "file", destination) |
||||
} |
||||
}() |
||||
var objectReader io.Reader = readCloser |
||||
if decompressFile { |
||||
decompressedReader, err := getGzipReader(readCloser) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer putGzipReader(decompressedReader) |
||||
|
||||
objectReader = decompressedReader |
||||
} |
||||
|
||||
_, err = io.Copy(f, objectReader) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
level.Info(logger).Log("msg", "downloaded file", "total_time", time.Since(start)) |
||||
if sync { |
||||
return f.Sync() |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func IsCompressedFile(filename string) bool { |
||||
return strings.HasSuffix(filename, ".gz") |
||||
} |
||||
|
||||
func LoggerWithFilename(logger log.Logger, filename string) log.Logger { |
||||
return log.With(logger, "file-name", filename) |
||||
} |
||||
|
||||
func ValidateSharedStoreKeyPrefix(prefix string) error { |
||||
if prefix == "" { |
||||
return errors.New("shared store key prefix must be set") |
||||
} else if strings.Contains(prefix, "\\") { |
||||
// When using windows filesystem as object store the implementation of ObjectClient in Cortex takes care of conversion of separator.
|
||||
// We just need to always use `/` as a path separator.
|
||||
return fmt.Errorf("shared store key prefix should only have '%s' as a path separator", delimiter) |
||||
} else if strings.HasPrefix(prefix, delimiter) { |
||||
return errors.New("shared store key prefix should never start with a path separator i.e '/'") |
||||
} else if !strings.HasSuffix(prefix, delimiter) { |
||||
return errors.New("shared store key prefix should end with a path separator i.e '/'") |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
@ -0,0 +1,59 @@ |
||||
package indexshipper |
||||
|
||||
import ( |
||||
"context" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk/client" |
||||
"github.com/grafana/loki/pkg/storage/config" |
||||
"github.com/grafana/loki/pkg/storage/stores/indexshipper/storage" |
||||
"github.com/grafana/loki/pkg/storage/stores/series/index" |
||||
) |
||||
|
||||
type tableClient struct { |
||||
indexStorageClient storage.Client |
||||
} |
||||
|
||||
// NewTableClient creates a client for managing tables in object storage based index store.
|
||||
// It is typically used when running a table manager.
|
||||
func NewTableClient(objectClient client.ObjectClient, storageKeyPrefix string) index.TableClient { |
||||
return &tableClient{storage.NewIndexStorageClient(objectClient, storageKeyPrefix)} |
||||
} |
||||
|
||||
func (b *tableClient) ListTables(ctx context.Context) ([]string, error) { |
||||
b.indexStorageClient.RefreshIndexListCache(ctx) |
||||
return b.indexStorageClient.ListTables(ctx) |
||||
} |
||||
|
||||
func (b *tableClient) CreateTable(ctx context.Context, desc config.TableDesc) error { |
||||
return nil |
||||
} |
||||
|
||||
func (b *tableClient) Stop() { |
||||
b.indexStorageClient.Stop() |
||||
} |
||||
|
||||
func (b *tableClient) DeleteTable(ctx context.Context, tableName string) error { |
||||
files, _, err := b.indexStorageClient.ListFiles(ctx, tableName, true) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
for _, file := range files { |
||||
err := b.indexStorageClient.DeleteFile(ctx, tableName, file.Name) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
func (b *tableClient) DescribeTable(ctx context.Context, name string) (desc config.TableDesc, isActive bool, err error) { |
||||
return config.TableDesc{ |
||||
Name: name, |
||||
}, true, nil |
||||
} |
||||
|
||||
func (b *tableClient) UpdateTable(ctx context.Context, current, expected config.TableDesc) error { |
||||
return nil |
||||
} |
||||
@ -1,57 +0,0 @@ |
||||
package shipper |
||||
|
||||
import ( |
||||
"context" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk/client" |
||||
"github.com/grafana/loki/pkg/storage/config" |
||||
"github.com/grafana/loki/pkg/storage/stores/series/index" |
||||
"github.com/grafana/loki/pkg/storage/stores/shipper/storage" |
||||
) |
||||
|
||||
type boltDBShipperTableClient struct { |
||||
indexStorageClient storage.Client |
||||
} |
||||
|
||||
func NewBoltDBShipperTableClient(objectClient client.ObjectClient, storageKeyPrefix string) index.TableClient { |
||||
return &boltDBShipperTableClient{storage.NewIndexStorageClient(objectClient, storageKeyPrefix)} |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) ListTables(ctx context.Context) ([]string, error) { |
||||
b.indexStorageClient.RefreshIndexListCache(ctx) |
||||
return b.indexStorageClient.ListTables(ctx) |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) CreateTable(ctx context.Context, desc config.TableDesc) error { |
||||
return nil |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) Stop() { |
||||
b.indexStorageClient.Stop() |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) DeleteTable(ctx context.Context, tableName string) error { |
||||
files, _, err := b.indexStorageClient.ListFiles(ctx, tableName, true) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
for _, file := range files { |
||||
err := b.indexStorageClient.DeleteFile(ctx, tableName, file.Name) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) DescribeTable(ctx context.Context, name string) (desc config.TableDesc, isActive bool, err error) { |
||||
return config.TableDesc{ |
||||
Name: name, |
||||
}, true, nil |
||||
} |
||||
|
||||
func (b *boltDBShipperTableClient) UpdateTable(ctx context.Context, current, expected config.TableDesc) error { |
||||
return nil |
||||
} |
||||
Loading…
Reference in new issue