Merge pull request concourse#8322 from evanchaoli/depop-avoid-duplica…

…ting-parallel-volume-streams

Depop avoid duplicating parallel volume streams

atc/db/lock/lock.go

@@ -60,6 +60,10 @@ func NewInMemoryCheckBuildTrackingLockID(checkableType string, checkableId int)
  return LockID{LockTypeInMemoryCheckBuildTracking, lockIDFromString(fmt.Sprintf("%s-%d", checkableType, checkableId))}
 }
 
+func NewVolumeStreamingLockID(resourceCacheID int, worker string) LockID {
+ return LockID{LockTypeJobScheduling, lockIDFromString(fmt.Sprintf("%d-%s", resourceCacheID, worker))}
+}
+
 //counterfeiter:generate . LockFactory
 type LockFactory interface {
  Acquire(logger lager.Logger, ids LockID) (Lock, bool, error)

atc/engine/build_step_delegate.go

@@ -229,6 +229,22 @@ func (delegate *buildStepDelegate) StreamingVolume(logger lager.Logger, volume s
  }
 }
 
+func (delegate *buildStepDelegate) WaitingForStreamedVolume(logger lager.Logger, volume string, destWorker string) {
+ err := delegate.build.SaveEvent(event.WaitingForStreamedVolume{
+ Time: time.Now().Unix(),
+ Origin: event.Origin{
+ ID: event.OriginID(delegate.planID),
+ },
+ Volume: volume,
+ DestWorker: destWorker,
+ })
+
+ if err != nil {
+ logger.Error("failed-to-save-waiting-for-streamed-volume-event", err)
+ return
+ }
+}
+
 func (delegate *buildStepDelegate) Errored(logger lager.Logger, message string) {
  err := delegate.build.SaveEvent(event.Error{
  Message: message,

atc/engine/build_step_delegate_test.go

@@ -1079,4 +1079,18 @@ var _ = Describe("BuildStepDelegate", func() {
  Expect(e.(event.StreamingVolume).DestWorker).To(Equal("dest-worker"))
  })
  })
+
+ Describe("WaitingForStreamedVolume", func() {
+ JustBeforeEach(func() {
+ delegate.WaitingForStreamedVolume(logger, "some-volume", "dest-worker")
+ })
+
+ It("saves an event", func() {
+ Expect(fakeBuild.SaveEventCallCount()).To(Equal(1))
+ e := fakeBuild.SaveEventArgsForCall(0)
+ Expect(e.EventType()).To(Equal(atc.EventType("waiting-for-streamed-volume")))
+ Expect(e.(event.WaitingForStreamedVolume).Volume).To(Equal("some-volume"))
+ Expect(e.(event.WaitingForStreamedVolume).DestWorker).To(Equal("dest-worker"))
+ })
+ })
 })

atc/event/events.go

@@ -120,6 +120,16 @@ type StreamingVolume struct {
 func (StreamingVolume) EventType() atc.EventType { return EventTypeStreamingVolume }
 func (StreamingVolume) Version() atc.EventVersion { return "1.0" }
 
+type WaitingForStreamedVolume struct {
+ Time int64 `json:"time"`
+ Origin Origin `json:"origin"`
+ Volume string `json:"volume"`
+ DestWorker string `json:"dest_worker"`
+}
+
+func (WaitingForStreamedVolume) EventType() atc.EventType { return EventTypeWaitingForStreamedVolume }
+func (WaitingForStreamedVolume) Version() atc.EventVersion { return "1.0" }
+
 type Log struct {
  Time int64 `json:"time"`
  Origin Origin `json:"origin"`

atc/event/parser.go

@@ -49,6 +49,7 @@ func init() {
  RegisterEvent(WaitingForWorker{})
  RegisterEvent(SelectedWorker{})
  RegisterEvent(StreamingVolume{})
+ RegisterEvent(WaitingForStreamedVolume{})
  RegisterEvent(Log{})
  RegisterEvent(Error{})
  RegisterEvent(ImageCheck{})

atc/event/types.go

@@ -18,6 +18,9 @@ const (
  // a step (get/put/task) is streaming a volume from another worker
  EventTypeStreamingVolume atc.EventType = "streaming-volume"
 
+ // a step (get/put/task) is waiting for another step to stream the volume to this worker
+ EventTypeWaitingForStreamedVolume atc.EventType = "waiting-for-streamed-volume"
+
  // task execution started
  EventTypeStartTask atc.EventType = "start-task"
 

atc/exec/build_step_delegate.go

@@ -36,6 +36,7 @@ type BuildStepDelegate interface {
  WaitingForWorker(lager.Logger)
  SelectedWorker(lager.Logger, string)
  StreamingVolume(lager.Logger, string, string, string)
+ WaitingForStreamedVolume(lager.Logger, string, string)
 
  ConstructAcrossSubsteps([]byte, []atc.AcrossVar, [][]interface{}) ([]atc.VarScopedPlan, error)
  ContainerOwner(planId atc.PlanID) db.ContainerOwner