Support distributed tracing (#11)

* Pass request trace ID to rep client

Signed-off-by: Maria Shaldybin <[email protected]>

* Move loggerWithTrace to trace package

* Pass trace Id to bbs client

---------

Signed-off-by: Maria Shaldybin <[email protected]>
Co-authored-by: Nick Rohn <[email protected]>

Author: mariash

auctionrunner/auction_runner.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"time"
 
+	"code.cloudfoundry.org/bbs/trace"
 	"code.cloudfoundry.org/clock"
 	"code.cloudfoundry.org/lager/v3"
 
@@ -51,21 +52,21 @@ func New(
 func (a *auctionRunner) Run(signals <-chan os.Signal, ready chan<- struct{}) error {
 	close(ready)
 
-	var hasWork chan struct{}
+	var hasWork chan Work
 	hasWork = a.batch.HasWork
 
 	for {
 		select {
-		case <-hasWork:
-			logger := a.logger.Session("auction")
+		case work := <-hasWork:
+			logger := trace.LoggerWithTraceInfo(a.logger, work.TraceID).Session("auction")
 
 			logger.Info("fetching-cell-reps")
-			clients, err := a.delegate.FetchCellReps()
+			clients, err := a.delegate.FetchCellReps(logger, work.TraceID)
 			if err != nil {
 				logger.Error("failed-to-fetch-reps", err)
 				time.Sleep(time.Second)
-				hasWork = make(chan struct{}, 1)
-				hasWork <- struct{}{}
+				hasWork = make(chan Work, 1)
+				hasWork <- work
 				break
 			}
 			logger.Info("fetched-cell-reps", lager.Data{"cell-reps-count": len(clients)})
@@ -119,17 +120,17 @@ func (a *auctionRunner) Run(signals <-chan os.Signal, ready chan<- struct{}) err
 			})
 
 			a.metricEmitter.AuctionCompleted(auctionResults)
-			a.delegate.AuctionCompleted(auctionResults)
+			a.delegate.AuctionCompleted(logger, work.TraceID, auctionResults)
 		case <-signals:
 			return nil
 		}
 	}
 }
 
-func (a *auctionRunner) ScheduleLRPsForAuctions(lrpStarts []auctioneer.LRPStartRequest) {
-	a.batch.AddLRPStarts(lrpStarts)
+func (a *auctionRunner) ScheduleLRPsForAuctions(lrpStarts []auctioneer.LRPStartRequest, traceID string) {
+	a.batch.AddLRPStarts(lrpStarts, traceID)
 }
 
-func (a *auctionRunner) ScheduleTasksForAuctions(tasks []auctioneer.TaskStartRequest) {
-	a.batch.AddTasks(tasks)
+func (a *auctionRunner) ScheduleTasksForAuctions(tasks []auctioneer.TaskStartRequest, traceID string) {
+	a.batch.AddTasks(tasks, traceID)
 }

auctionrunner/batch.go

@@ -11,11 +11,15 @@ import (
 	"code.cloudfoundry.org/clock"
 )
 
+type Work struct {
+	TraceID string
+}
+
 type Batch struct {
 	lrpAuctions  []auctiontypes.LRPAuction
 	taskAuctions []auctiontypes.TaskAuction
 	lock         *sync.Mutex
-	HasWork      chan struct{}
+	HasWork      chan Work
 	clock        clock.Clock
 }
 
@@ -24,11 +28,11 @@ func NewBatch(clock clock.Clock) *Batch {
 		lrpAuctions: []auctiontypes.LRPAuction{},
 		lock:        &sync.Mutex{},
 		clock:       clock,
-		HasWork:     make(chan struct{}, 1),
+		HasWork:     make(chan Work, 1),
 	}
 }
 
-func (b *Batch) AddLRPStarts(starts []auctioneer.LRPStartRequest) {
+func (b *Batch) AddLRPStarts(starts []auctioneer.LRPStartRequest, traceID string) {
 	auctions := make([]auctiontypes.LRPAuction, 0, len(starts))
 	now := b.clock.Now()
 	for i := range starts {
@@ -42,11 +46,11 @@ func (b *Batch) AddLRPStarts(starts []auctioneer.LRPStartRequest) {
 
 	b.lock.Lock()
 	b.lrpAuctions = append(b.lrpAuctions, auctions...)
-	b.claimToHaveWork()
+	b.claimToHaveWork(traceID)
 	b.lock.Unlock()
 }
 
-func (b *Batch) AddTasks(tasks []auctioneer.TaskStartRequest) {
+func (b *Batch) AddTasks(tasks []auctioneer.TaskStartRequest, traceID string) {
 	auctions := make([]auctiontypes.TaskAuction, 0, len(tasks))
 	now := b.clock.Now()
 	for i := range tasks {
@@ -55,7 +59,7 @@ func (b *Batch) AddTasks(tasks []auctioneer.TaskStartRequest) {
 
 	b.lock.Lock()
 	b.taskAuctions = append(b.taskAuctions, auctions...)
-	b.claimToHaveWork()
+	b.claimToHaveWork(traceID)
 	b.lock.Unlock()
 }
 
@@ -96,9 +100,9 @@ func (b *Batch) DedupeAndDrain() ([]auctiontypes.LRPAuction, []auctiontypes.Task
 	return dedupedLRPAuctions, dedupedTaskAuctions
 }
 
-func (b *Batch) claimToHaveWork() {
+func (b *Batch) claimToHaveWork(traceID string) {
 	select {
-	case b.HasWork <- struct{}{}:
+	case b.HasWork <- Work{TraceID: traceID}:
 	default:
 	}
 }

auctionrunner/batch_test.go

@@ -34,7 +34,7 @@ var _ = Describe("Batch", func() {
 		Context("when adding start auctions", func() {
 			BeforeEach(func() {
 				lrpStart = BuildLRPStartRequest("pg-1", "domain", []int{1}, "linux", 10, 10, 10, []string{}, []string{})
-				batch.AddLRPStarts([]auctioneer.LRPStartRequest{lrpStart})
+				batch.AddLRPStarts([]auctioneer.LRPStartRequest{lrpStart}, "some-trace-id")
 			})
 
 			It("makes the start auction available when drained", func() {
@@ -50,7 +50,7 @@ var _ = Describe("Batch", func() {
 		Context("when adding tasks", func() {
 			BeforeEach(func() {
 				task = BuildTaskStartRequest("tg-1", "domain", "linux", 10, 10, 10)
-				batch.AddTasks([]auctioneer.TaskStartRequest{task})
+				batch.AddTasks([]auctioneer.TaskStartRequest{task}, "some-trace-id")
 			})
 
 			It("makes the stop auction available when drained", func() {
@@ -70,12 +70,12 @@ var _ = Describe("Batch", func() {
 				BuildLRPStartRequest("pg-1", "domain", []int{1}, "linux", 10, 10, 10, []string{"driver-1"}, []string{"tag-1"}),
 				BuildLRPStartRequest("pg-1", "domain", []int{1}, "linux", 10, 10, 10, []string{"driver-1"}, []string{"tag-1"}),
 				BuildLRPStartRequest("pg-2", "domain", []int{2}, "linux", 10, 10, 10, []string{"driver-2"}, []string{"tag-2"}),
-			})
+			}, "some-trace-id")
 
 			batch.AddTasks([]auctioneer.TaskStartRequest{
 				BuildTaskStartRequest("tg-1", "domain", "linux", 10, 10, 10),
 				BuildTaskStartRequest("tg-1", "domain", "linux", 10, 10, 10),
-				BuildTaskStartRequest("tg-2", "domain", "linux", 10, 10, 10)})
+				BuildTaskStartRequest("tg-2", "domain", "linux", 10, 10, 10)}, "some-trace-id")
 		})
 
 		It("should dedupe any duplicate start auctions and stop auctions", func() {

auctiontypes/fakes/fake_auction_runner.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"code.cloudfoundry.org/auctioneer"
+	"code.cloudfoundry.org/lager/v3"
 	"code.cloudfoundry.org/rep"
 	"github.com/tedsuo/ifrit"
 )
@@ -47,13 +48,13 @@ var ErrorExceededInflightCreation = errors.New("waiting to start instance: reach
 //go:generate counterfeiter -o fakes/fake_auction_runner.go . AuctionRunner
 type AuctionRunner interface {
 	ifrit.Runner
-	ScheduleLRPsForAuctions([]auctioneer.LRPStartRequest)
-	ScheduleTasksForAuctions([]auctioneer.TaskStartRequest)
+	ScheduleLRPsForAuctions([]auctioneer.LRPStartRequest, string)
+	ScheduleTasksForAuctions([]auctioneer.TaskStartRequest, string)
 }
 
 type AuctionRunnerDelegate interface {
-	FetchCellReps() (map[string]rep.Client, error)
-	AuctionCompleted(AuctionResults)
+	FetchCellReps(lager.Logger, string) (map[string]rep.Client, error)
+	AuctionCompleted(lager.Logger, string, AuctionResults)
 }
 
 //go:generate counterfeiter -o fakes/fake_metric_emitter.go . AuctionMetricEmitterDelegate

auctiontypes/types.go

@@ -4,6 +4,7 @@ import (
 	"sync"
 
 	"code.cloudfoundry.org/auction/auctiontypes"
+	"code.cloudfoundry.org/lager/v3"
 	"code.cloudfoundry.org/rep"
 )
 
@@ -30,15 +31,15 @@ func (a *auctionRunnerDelegate) SetCellLimit(limit int) {
 	a.cellLimit = limit
 }
 
-func (a *auctionRunnerDelegate) FetchCellReps() (map[string]rep.Client, error) {
+func (a *auctionRunnerDelegate) FetchCellReps(lager.Logger, string) (map[string]rep.Client, error) {
 	subset := map[string]rep.Client{}
 	for i := 0; i < a.cellLimit; i++ {
 		subset[cellGuid(i)] = a.cells[cellGuid(i)]
 	}
 	return subset, nil
 }
 
-func (a *auctionRunnerDelegate) AuctionCompleted(work auctiontypes.AuctionResults) {
+func (a *auctionRunnerDelegate) AuctionCompleted(logger lager.Logger, traceID string, work auctiontypes.AuctionResults) {
 	a.lock.Lock()
 	defer a.lock.Unlock()
 	a.workResults.FailedLRPs = append(a.workResults.FailedLRPs, work.FailedLRPs...)

simulation/auction_runner_delegate_test.go

@@ -71,7 +71,7 @@ var _ = Describe("Auction", func() {
 
 	runStartAuction := func(lrpStartAuctions []auctioneer.LRPStartRequest, numCells int) {
 		runnerDelegate.SetCellLimit(numCells)
-		runner.ScheduleLRPsForAuctions(lrpStartAuctions)
+		runner.ScheduleLRPsForAuctions(lrpStartAuctions, "some-trace-id")
 
 		Eventually(runnerDelegate.ResultSize, time.Minute, 100*time.Millisecond).Should(Equal(len(lrpStartAuctions)))
 	}
@@ -83,7 +83,7 @@ var _ = Describe("Auction", func() {
 		Eventually(runnerDelegate.ResultSize, time.Minute, 100*time.Millisecond).Should(Equal(len(lrpStartAuctions)))
 		duration := time.Since(t)
 
-		cells, _ := runnerDelegate.FetchCellReps()
+		cells, _ := runnerDelegate.FetchCellReps(logger, "some-trace-id")
 		report := visualization.NewReport(len(lrpStartAuctions), cells, runnerDelegate.Results(), duration)
 
 		visualization.PrintReport(report)