Trigger compaction prior retention.

pkg/storage/stores/shipper/compactor/compactor_test.go

@@ -19,7 +19,7 @@ func TestIsDefaults(t *testing.T) {
 		{&Config{}, false},
 		{&Config{
 			SharedStoreKeyPrefix:     "index/",
-			CompactionInterval:       2 * time.Hour,
+			CompactionInterval:       10 * time.Minute,
 			RetentionDeleteDelay:     2 * time.Hour,
 			RetentionDeleteWorkCount: 150,
 		}, true},

pkg/storage/stores/shipper/compactor/table_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.etcd.io/bbolt"
 
+	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/retention"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/testutil"
 )
 
@@ -70,6 +71,103 @@ func TestTable_Compaction(t *testing.T) {
 	compareCompactedDB(t, filepath.Join(tablePathInStorage, files[0].Name()), filepath.Join(objectStoragePath, "test-copy"))
 }
 
+type TableMarkerFunc func(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error)
+
+func (t TableMarkerFunc) MarkForDelete(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error) {
+	return t(ctx, tableName, db)
+}
+
+func TestTable_CompactionRetention(t *testing.T) {
+	for name, tt := range map[string]struct {
+		dbCount     int
+		assert      func(t *testing.T, storagePath string)
+		tableMarker retention.TableMarker
+	}{
+		"emptied table": {
+			dbCount: 2,
+			assert: func(t *testing.T, storagePath string) {
+				_, err := os.Stat(storagePath)
+				require.True(t, os.IsNotExist(err))
+			},
+			tableMarker: TableMarkerFunc(func(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error) {
+				return true, 100, nil
+			}),
+		},
+		"marked table": {
+			dbCount: 2,
+			assert: func(t *testing.T, storagePath string) {
+				files, err := ioutil.ReadDir(storagePath)
+				require.NoError(t, err)
+				require.Len(t, files, 1)
+				require.True(t, strings.HasSuffix(files[0].Name(), ".gz"))
+			},
+			tableMarker: TableMarkerFunc(func(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error) {
+				return false, 100, nil
+			}),
+		},
+		"already compacted table": {
+			dbCount: 1,
+			assert: func(t *testing.T, storagePath string) {
+				files, err := ioutil.ReadDir(storagePath)
+				require.NoError(t, err)
+				require.Len(t, files, 1)
+				require.True(t, strings.HasSuffix(files[0].Name(), ".gz"))
+			},
+			tableMarker: TableMarkerFunc(func(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error) {
+				return false, 100, nil
+			}),
+		},
+		"not modified": {
+			dbCount: 1,
+			assert: func(t *testing.T, storagePath string) {
+				files, err := ioutil.ReadDir(storagePath)
+				require.NoError(t, err)
+				require.Len(t, files, 1)
+				require.True(t, strings.HasSuffix(files[0].Name(), ".gz"))
+			},
+			tableMarker: TableMarkerFunc(func(ctx context.Context, tableName string, db *bbolt.DB) (bool, int64, error) {
+				return false, 0, nil
+			}),
+		},
+	} {
+		tt := tt
+		t.Run(name, func(t *testing.T) {
+			tempDir := t.TempDir()
+
+			tableName := "test"
+			objectStoragePath := filepath.Join(tempDir, objectsStorageDirName)
+			tableWorkingDirectory := filepath.Join(tempDir, workingDirName, tableName)
+
+			// setup some dbs
+			numDBs := tt.dbCount
+			numRecordsPerDB := 100
+
+			dbsToSetup := make(map[string]testutil.DBRecords)
+			for i := 0; i < numDBs; i++ {
+				dbsToSetup[fmt.Sprint(i)] = testutil.DBRecords{
+					Start:      i * numRecordsPerDB,
+					NumRecords: (i + 1) * numRecordsPerDB,
+				}
+			}
+
+			testutil.SetupDBTablesAtPath(t, tableName, objectStoragePath, dbsToSetup, true)
+
+			// setup exact same copy of dbs for comparison.
+			testutil.SetupDBTablesAtPath(t, "test-copy", objectStoragePath, dbsToSetup, false)
+
+			// do the compaction
+			objectClient, err := local.NewFSObjectClient(local.FSConfig{Directory: objectStoragePath})
+			require.NoError(t, err)
+
+			table, err := newTable(context.Background(), tableWorkingDirectory, objectClient, true, tt.tableMarker)
+			require.NoError(t, err)
+
+			require.NoError(t, table.compact())
+			tt.assert(t, filepath.Join(objectStoragePath, tableName))
+		})
+	}
+}
+
 func TestTable_CompactionFailure(t *testing.T) {
 	tempDir, err := ioutil.TempDir("", "table-compaction-failure")
 	require.NoError(t, err)