Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add some tests in compactor and fix a bug in IntervalHasExpiredChunks check in retention with tests #3969

Merged
merged 2 commits into from
Jul 8, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 98 additions & 1 deletion pkg/storage/stores/shipper/compactor/compactor_test.go
Original file line number Diff line number Diff line change
@@ -1,16 +1,40 @@
package compactor

import (
"context"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strings"
"testing"
"time"

"github.com/cortexproject/cortex/pkg/chunk"
"github.com/prometheus/common/model"
"github.com/cortexproject/cortex/pkg/chunk/local"
"github.com/cortexproject/cortex/pkg/chunk/storage"
"github.com/cortexproject/cortex/pkg/util/flagext"

"github.com/prometheus/common/model"
"github.com/stretchr/testify/require"

loki_storage "github.com/grafana/loki/pkg/storage"
"github.com/grafana/loki/pkg/storage/stores/shipper/testutil"
)

func setupTestCompactor(t *testing.T, tempDir string) *Compactor {
cfg := Config{}
flagext.DefaultValues(&cfg)
cfg.WorkingDirectory = filepath.Join(tempDir, workingDirName)
cfg.SharedStoreType = "filesystem"
cfg.RetentionEnabled = false

c, err := NewCompactor(cfg, storage.Config{FSConfig: local.FSConfig{Directory: tempDir}}, loki_storage.SchemaConfig{}, nil, nil)
require.NoError(t, err)

return c
}

func TestIsDefaults(t *testing.T) {
for i, tc := range []struct {
in *Config
Expand Down Expand Up @@ -71,3 +95,76 @@ func TestExtractIntervalFromTableName(t *testing.T) {
}

}

func TestCompactor_RunCompaction(t *testing.T) {
tempDir, err := ioutil.TempDir("", "compactor-run-compaction")
require.NoError(t, err)

defer func() {
require.NoError(t, os.RemoveAll(tempDir))
}()

tablesPath := filepath.Join(tempDir, "index")
tablesCopyPath := filepath.Join(tempDir, "index-copy")

tables := map[string]map[string]testutil.DBRecords{
"table1": {
"db1": {
Start: 0,
NumRecords: 10,
},
"db2": {
Start: 10,
NumRecords: 10,
},
"db3": {
Start: 20,
NumRecords: 10,
},
"db4": {
Start: 30,
NumRecords: 10,
},
},
"table2": {
"db1": {
Start: 40,
NumRecords: 10,
},
"db2": {
Start: 50,
NumRecords: 10,
},
"db3": {
Start: 60,
NumRecords: 10,
},
"db4": {
Start: 70,
NumRecords: 10,
},
},
}

for name, dbs := range tables {
testutil.SetupDBTablesAtPath(t, name, tablesPath, dbs, false)

// setup exact same copy of dbs for comparison.
testutil.SetupDBTablesAtPath(t, name, tablesCopyPath, dbs, false)
}

compactor := setupTestCompactor(t, tempDir)
err = compactor.RunCompaction(context.Background())
require.NoError(t, err)

for name := range tables {
// verify that we have only 1 file left in storage after compaction.
files, err := ioutil.ReadDir(filepath.Join(tablesPath, name))
require.NoError(t, err)
require.Len(t, files, 1)
require.True(t, strings.HasSuffix(files[0].Name(), ".gz"))

// verify we have all the kvs in compacted db which were there in source dbs.
compareCompactedDB(t, filepath.Join(tablesPath, name, files[0].Name()), filepath.Join(tablesCopyPath, name))
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ func (e *expirationChecker) MarkPhaseFailed() {}
func (e *expirationChecker) MarkPhaseFinished() {}

func (e *expirationChecker) IntervalHasExpiredChunks(interval model.Interval) bool {
return e.earliestRetentionStartTime.Before(interval.Start) || e.earliestRetentionStartTime.Before(interval.End)
return interval.Start.Before(e.earliestRetentionStartTime)
}

type TenantsRetention struct {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -208,3 +208,49 @@ func TestFindEarliestRetentionStartTime(t *testing.T) {
})
}
}

func TestExpirationChecker_IntervalHasExpiredChunks(t *testing.T) {
for _, tc := range []struct {
name string
expirationChecker expirationChecker
interval model.Interval
hasExpiredChunks bool
}{
{
name: "not expired",
expirationChecker: expirationChecker{
earliestRetentionStartTime: model.Now().Add(-24 * time.Hour),
},
interval: model.Interval{
Start: model.Now().Add(-time.Hour),
End: model.Now(),
},
},
{
name: "partially expired",
expirationChecker: expirationChecker{
earliestRetentionStartTime: model.Now().Add(-24 * time.Hour),
},
interval: model.Interval{
Start: model.Now().Add(-25 * time.Hour),
End: model.Now().Add(-22 * time.Hour),
},
hasExpiredChunks: true,
},
{
name: "fully expired",
expirationChecker: expirationChecker{
earliestRetentionStartTime: model.Now().Add(-24 * time.Hour),
},
interval: model.Interval{
Start: model.Now().Add(-26 * time.Hour),
End: model.Now().Add(-25 * time.Hour),
},
hasExpiredChunks: true,
},
} {
t.Run(tc.name, func(t *testing.T) {
require.Equal(t, tc.hasExpiredChunks, tc.expirationChecker.IntervalHasExpiredChunks(tc.interval))
})
}
}