feat: add invocationId to plugin interface (#262)

Author: gregmagolan

pkg/aspect/lint/bep.go

@@ -132,7 +132,7 @@ func parseLinterMnemonicFromFilename(filename string) string {
 	return s[len(s)-2]
 }
 
-func (runner *LintBEPHandler) bepEventCallback(event *buildeventstream.BuildEvent, sn int64) error {
+func (runner *LintBEPHandler) bepEventCallback(event *buildeventstream.BuildEvent, sn int64, invocationId string) error {
 	switch event.Payload.(type) {
 
 	case *buildeventstream.BuildEvent_WorkspaceInfo:

pkg/plugin/sdk/v1alpha4/plugin/grpc.go

@@ -71,7 +71,7 @@ func (m *GRPCServer) BEPEventCallback(
 	ctx context.Context,
 	req *proto.BEPEventCallbackReq,
 ) (*proto.BEPEventCallbackRes, error) {
-	return &proto.BEPEventCallbackRes{}, m.Impl.BEPEventCallback(req.Event, req.SequenceNumber)
+	return &proto.BEPEventCallbackRes{}, m.Impl.BEPEventCallback(req.Event, req.SequenceNumber, req.InvocationId)
 }
 
 // Setup translates the gRPC call to the Plugin Setup implementation.
@@ -189,8 +189,8 @@ var _ Plugin = (*GRPCClient)(nil)
 
 // BEPEventCallback is called from the Core to execute the Plugin
 // BEPEventCallback.
-func (m *GRPCClient) BEPEventCallback(event *buildeventstream.BuildEvent, sn int64) error {
-	_, err := m.client.BEPEventCallback(context.Background(), &proto.BEPEventCallbackReq{Event: event, SequenceNumber: sn})
+func (m *GRPCClient) BEPEventCallback(event *buildeventstream.BuildEvent, sn int64, invocationId string) error {
+	_, err := m.client.BEPEventCallback(context.Background(), &proto.BEPEventCallbackReq{Event: event, SequenceNumber: sn, InvocationId: invocationId})
 	return err
 }
 

pkg/plugin/sdk/v1alpha4/plugin/interface.go

@@ -28,7 +28,7 @@ import (
 
 // Plugin determines how an aspect Plugin should be implemented.
 type Plugin interface {
-	BEPEventCallback(event *buildeventstream.BuildEvent, sn int64) error
+	BEPEventCallback(event *buildeventstream.BuildEvent, sn int64, invocationId string) error
 	CustomCommands() ([]*Command, error)
 	PostBuildHook(
 		isInteractiveMode bool,
@@ -88,7 +88,7 @@ func (*Base) Setup(*SetupConfig) error {
 }
 
 // BEPEventCallback satisfies Plugin.BEPEventCallback.
-func (*Base) BEPEventCallback(*buildeventstream.BuildEvent, int64) error {
+func (*Base) BEPEventCallback(*buildeventstream.BuildEvent, int64, string) error {
 	return nil
 }
 

pkg/plugin/sdk/v1alpha4/proto/plugin.pb.go

@@ -20,6 +20,7 @@ service Plugin {
 message BEPEventCallbackReq {
   build_event_stream.BuildEvent event = 1;
   int64 sequence_number = 2;
+  string invocation_id = 3;
 }
 
 message BEPEventCallbackRes {}

pkg/plugin/sdk/v1alpha4/proto/plugin.proto

@@ -212,7 +212,7 @@ func (bb *besBackend) RegisterBesProxy(ctx context.Context, p besproxy.BESProxy)
 
 // CallbackFn is the signature for the callback function used by the subscribers
 // of the Build Event Protocol events.
-type CallbackFn func(*buildeventstream.BuildEvent, int64) error
+type CallbackFn func(*buildeventstream.BuildEvent, int64, string) error
 
 // RegisterSubscriber registers a new subscriber callback function to the
 // Build Event Protocol events.
@@ -285,7 +285,7 @@ func (bb *besBackend) SendEventsToSubscribers(c <-chan *buildv1.PublishBuildTool
 				}
 				s := subscribers.head
 				for s != nil {
-					if err := s.callback(buildEvent, req.GetOrderedBuildEvent().GetSequenceNumber()); err != nil {
+					if err := s.callback(buildEvent, req.GetOrderedBuildEvent().GetSequenceNumber(), req.GetOrderedBuildEvent().GetStreamId().GetInvocationId()); err != nil {
 						bb.errorsMutex.Lock()
 						bb.errors.Insert(err)
 						bb.errorsMutex.Unlock()

pkg/plugin/system/bep/bes_backend.go

@@ -362,7 +362,7 @@ func TestPublishBuildToolEventStream(t *testing.T) {
 		var anyBuildEvent anypb.Any
 		anyBuildEvent.MarshalFrom(buildEvent)
 		event := &buildv1.BuildEvent{Event: &buildv1.BuildEvent_BazelEvent{BazelEvent: &anyBuildEvent}}
-		streamId := &buildv1.StreamId{BuildId: "1"}
+		streamId := &buildv1.StreamId{BuildId: "1", InvocationId: "1bdca2c1-fa12-4097-b538-4707ce8fe50d"}
 		orderedBuildEvent := &buildv1.OrderedBuildEvent{
 			StreamId:       streamId,
 			SequenceNumber: 1,
@@ -398,23 +398,26 @@ func TestPublishBuildToolEventStream(t *testing.T) {
 		}
 		close(besBackend.ready)
 		var calledSubscriber1, calledSubscriber2, calledSubscriber3 bool
-		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64) error {
+		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64, invocationId string) error {
 			// g.Expect(evt).To(Equal(buildEvent))
 			g.Expect(sn).To(Equal(orderedBuildEvent.SequenceNumber))
+			g.Expect(invocationId).To(Equal(orderedBuildEvent.StreamId.InvocationId))
 			calledSubscriber1 = true
 			return nil
 		}, false)
 		expectedSubscriber2Err := fmt.Errorf("error from subscriber 2")
-		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64) error {
+		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64, invocationId string) error {
 			// g.Expect(evt).To(Equal(buildEvent))
 			g.Expect(sn).To(Equal(orderedBuildEvent.SequenceNumber))
+			g.Expect(invocationId).To(Equal(orderedBuildEvent.StreamId.InvocationId))
 			calledSubscriber2 = true
 			return expectedSubscriber2Err
 		}, false)
 		expectedSubscriber3Err := fmt.Errorf("error from subscriber 3")
-		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64) error {
+		besBackend.RegisterSubscriber(func(evt *buildeventstream.BuildEvent, sn int64, invocationId string) error {
 			// g.Expect(evt).To(Equal(buildEvent))
 			g.Expect(sn).To(Equal(orderedBuildEvent.SequenceNumber))
+			g.Expect(invocationId).To(Equal(orderedBuildEvent.StreamId.InvocationId))
 			calledSubscriber3 = true
 			return expectedSubscriber3Err
 		}, false)

pkg/plugin/system/bep/bes_backend_test.go

@@ -228,7 +228,7 @@ func (bb *besPipe) streamBesEvents(ctx context.Context, conn *os.File) error {
 
 		seqId++
 
-		if err := bb.publishBesEvent(seqId, &event); err != nil {
+		if err := bb.publishBesEvent(seqId, bb.besInvocationId, &event); err != nil {
 			return fmt.Errorf("failed to publish BES event: %w", err)
 		}
 
@@ -243,14 +243,14 @@ func (bb *besPipe) streamBesEvents(ctx context.Context, conn *os.File) error {
 	return nil
 }
 
-func (bb *besPipe) publishBesEvent(seqId int64, event *buildeventstream.BuildEvent) error {
+func (bb *besPipe) publishBesEvent(seqId int64, invocationId string, event *buildeventstream.BuildEvent) error {
 	eg := errgroup.Group{}
 
 	for s := bb.subscribers.head; s != nil; s = s.next {
 		cb := s.callback
 		eg.Go(
 			func() error {
-				return cb(event, seqId)
+				return cb(event, seqId, invocationId)
 			},
 		)
 	}