Repository: cortexproject/cortex
Pull request #2190 (2020-02-27 12:20:20)

Pass logger with userID to per-user TSDB components.
**What this PR does**: this PR adds "org_id" information to logger passed to per-user TSDB components.



pstibrany: > Thanks @pstibrany, LGTM! May you fix the tests, please? We have some tests where we also asserts on logs 🎉

OK, that was unexpected :) Fixed now.

Patch:
diff --git a/MAINTAINERS b/MAINTAINERS
index 8d696c52fe..71e527524d 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -1,6 +1,5 @@
 Bryan Boreham, Weaveworks <bryan@weave.works> (@bboreham)
 Chris Marchbanks, Splunk <csmarchbanks@gmail.com> (@csmarchbanks)
-Cody Boggs, Splunk <strofcon@gmail.com> (@cboggs)
 Jacob Lisi, Grafana Labs <jacob.t.lisi@gmail.com> (@jtlisi)
 Ken Haines, Microsoft <khaines@microsoft.com> (@khaines)
 Marco Pracucci, Grafana Labs <marco@pracucci.com> (@pracucci)
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index 70119c40e0..9b169612b9 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -305,8 +305,10 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	reg := prometheus.NewRegistry()
 	defer c.syncerMetrics.gatherThanosSyncerMetrics(reg)
 
+	ulogger := util.WithUserID(userID, c.logger)
+
 	fetcher, err := block.NewMetaFetcher(
-		c.logger,
+		ulogger,
 		c.compactorCfg.MetaSyncConcurrency,
 		bucket,
 		// The fetcher stores cached metas in the "meta-syncer/" sub directory,
@@ -321,7 +323,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	}
 
 	syncer, err := compact.NewSyncer(
-		c.logger,
+		ulogger,
 		reg,
 		bucket,
 		fetcher,
@@ -334,7 +336,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	}
 
 	compactor, err := compact.NewBucketCompactor(
-		c.logger,
+		ulogger,
 		syncer,
 		c.tsdbCompactor,
 		path.Join(c.compactorCfg.DataDir, "compact"),
diff --git a/pkg/ingester/ingester_v2.go b/pkg/ingester/ingester_v2.go
index 62b6866e79..9895e355d1 100644
--- a/pkg/ingester/ingester_v2.go
+++ b/pkg/ingester/ingester_v2.go
@@ -599,11 +599,11 @@ func (i *Ingester) getOrCreateTSDB(userID string, force bool) (*userTSDB, error)
 // createTSDB creates a TSDB for a given userID, and returns the created db.
 func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	tsdbPromReg := prometheus.NewRegistry()
-
 	udir := i.cfg.TSDBConfig.BlocksDir(userID)
+	userLogger := util.WithUserID(userID, util.Logger)
 
 	// Create a new user database
-	db, err := tsdb.Open(udir, util.Logger, tsdbPromReg, &tsdb.Options{
+	db, err := tsdb.Open(udir, userLogger, tsdbPromReg, &tsdb.Options{
 		RetentionDuration: uint64(i.cfg.TSDBConfig.Retention / time.Millisecond),
 		BlockRanges:       i.cfg.TSDBConfig.BlockRanges.ToMilliseconds(),
 		NoLockfile:        true,
@@ -633,7 +633,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	// Create a new shipper for this database
 	if i.cfg.TSDBConfig.ShipInterval > 0 {
 		userDB.shipper = shipper.New(
-			util.Logger,
+			userLogger,
 			tsdbPromReg,
 			udir,
 			cortex_tsdb.NewUserBucketClient(userID, i.TSDBState.bucket),
diff --git a/pkg/querier/block_store.go b/pkg/querier/block_store.go
index fb1dc716c8..da299766d1 100644
--- a/pkg/querier/block_store.go
+++ b/pkg/querier/block_store.go
@@ -25,6 +25,7 @@ import (
 	"google.golang.org/grpc/metadata"
 
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
@@ -349,14 +350,16 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 		return bs, nil
 	}
 
-	level.Info(u.logger).Log("msg", "creating user bucket store", "user", userID)
+	userLogger := util.WithUserID(userID, u.logger)
+
+	level.Info(userLogger).Log("msg", "creating user bucket store")
 
 	userBkt := tsdb.NewUserBucketClient(userID, u.bucket)
 
 	reg := prometheus.NewRegistry()
 	indexCacheSizeBytes := u.cfg.BucketStore.IndexCacheSizeBytes
 	maxItemSizeBytes := indexCacheSizeBytes / 2
-	indexCache, err := storecache.NewInMemoryIndexCacheWithConfig(u.logger, reg, storecache.InMemoryIndexCacheConfig{
+	indexCache, err := storecache.NewInMemoryIndexCacheWithConfig(userLogger, reg, storecache.InMemoryIndexCacheConfig{
 		MaxSize:     storecache.Bytes(indexCacheSizeBytes),
 		MaxItemSize: storecache.Bytes(maxItemSizeBytes),
 	})
@@ -365,7 +368,7 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 	}
 
 	fetcher, err := block.NewMetaFetcher(
-		u.logger,
+		userLogger,
 		u.cfg.BucketStore.MetaSyncConcurrency,
 		userBkt,
 		filepath.Join(u.cfg.BucketStore.SyncDir, userID), // The fetcher stores cached metas in the "meta-syncer/" sub directory
@@ -377,7 +380,7 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 	}
 
 	bs, err = store.NewBucketStore(
-		u.logger,
+		userLogger,
 		reg,
 		userBkt,
 		fetcher,


Test patch:
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index f03e2bae69..5c8ac29e89 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -294,16 +294,16 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 		`level=info msg="discovering users from bucket"`,
 		`level=info msg="discovered users from bucket" users=2`,
 		`level=info msg="starting compaction of user blocks" user=user-1`,
-		`level=info msg="start sync of metas"`,
-		`level=info msg="start of GC"`,
-		`level=info msg="start of compaction"`,
-		`level=info msg="compaction iterations done"`,
+		`level=info org_id=user-1 msg="start sync of metas"`,
+		`level=info org_id=user-1 msg="start of GC"`,
+		`level=info org_id=user-1 msg="start of compaction"`,
+		`level=info org_id=user-1 msg="compaction iterations done"`,
 		`level=info msg="successfully compacted user blocks" user=user-1`,
 		`level=info msg="starting compaction of user blocks" user=user-2`,
-		`level=info msg="start sync of metas"`,
-		`level=info msg="start of GC"`,
-		`level=info msg="start of compaction"`,
-		`level=info msg="compaction iterations done"`,
+		`level=info org_id=user-2 msg="start sync of metas"`,
+		`level=info org_id=user-2 msg="start of GC"`,
+		`level=info org_id=user-2 msg="start of compaction"`,
+		`level=info org_id=user-2 msg="compaction iterations done"`,
 		`level=info msg="successfully compacted user blocks" user=user-2`,
 	}, removeMetaFetcherLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
 
@@ -368,16 +368,16 @@ func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunni
 		`level=info msg="discovering users from bucket"`,
 		`level=info msg="discovered users from bucket" users=2`,
 		`level=info msg="starting compaction of user blocks" user=user-1`,
-		`level=info msg="start sync of metas"`,
-		`level=info msg="start of GC"`,
-		`level=info msg="start of compaction"`,
-		`level=info msg="compaction iterations done"`,
+		`level=info org_id=user-1 msg="start sync of metas"`,
+		`level=info org_id=user-1 msg="start of GC"`,
+		`level=info org_id=user-1 msg="start of compaction"`,
+		`level=info org_id=user-1 msg="compaction iterations done"`,
 		`level=info msg="successfully compacted user blocks" user=user-1`,
 		`level=info msg="starting compaction of user blocks" user=user-2`,
-		`level=info msg="start sync of metas"`,
-		`level=info msg="start of GC"`,
-		`level=info msg="start of compaction"`,
-		`level=info msg="compaction iterations done"`,
+		`level=info org_id=user-2 msg="start sync of metas"`,
+		`level=info org_id=user-2 msg="start of GC"`,
+		`level=info org_id=user-2 msg="start of compaction"`,
+		`level=info org_id=user-2 msg="compaction iterations done"`,
 		`level=info msg="successfully compacted user blocks" user=user-2`,
 	}, removeMetaFetcherLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
 }