feat: pass request context and node address to the events sink adapter

Context may be handy to cancel any subsequent calls in the adapter.
While node Addr is required in Sidero to match nodes.

Signed-off-by: Artem Chernyshev <artem.chernyshev@talos-systems.com>

cmd/siderolink-agent/event_sink.go

@@ -25,13 +25,13 @@ var eventSinkFlags struct {
 type adapter struct{}
 
 // HandleEvent implements events.Adapter.
-func (s *adapter) HandleEvent(e events.Event) error {
+func (s *adapter) HandleEvent(ctx context.Context, e events.Event) error {
 	data, err := yaml.Marshal(e.Payload)
 	if err != nil {
 		return err
 	}
 
-	log.Printf("Event: %s, ID: %s, Payload: \n\t%s", e.TypeURL, e.ID, strings.Join(strings.Split(string(data), "\n"), "\n\t"))
+	log.Printf("Node: %s, Event: %s, ID: %s, Payload: \n\t%s", e.Node, e.TypeURL, e.ID, strings.Join(strings.Split(string(data), "\n"), "\n\t"))
 
 	return nil
 }

pkg/events/event.go

@@ -12,4 +12,5 @@ type Event struct {
 	Payload proto.Message
 	TypeURL string
 	ID      string
+	Node    string
 }

pkg/events/sink.go

@@ -8,14 +8,15 @@ import (
 	"context"
 
 	"github.com/talos-systems/talos/pkg/machinery/api/machine"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/talos-systems/siderolink/api/events"
 )
 
 // Adapter is an abstract event stream receiver.
 type Adapter interface {
-	HandleEvent(event Event) error
+	HandleEvent(ctx context.Context, event Event) error
 }
 
 // Sink implements events.EventSinkServiceServer.
@@ -64,7 +65,15 @@ func (s *Sink) Publish(ctx context.Context, e *events.EventRequest) (*events.Eve
 		return res, err
 	}
 
-	return res, s.adapter.HandleEvent(Event{
+	var node string
+
+	peer, ok := peer.FromContext(ctx)
+	if ok {
+		node = peer.Addr.String()
+	}
+
+	return res, s.adapter.HandleEvent(ctx, Event{
+		Node:    node,
 		TypeURL: typeURL,
 		ID:      e.Id,
 		Payload: msg,

pkg/events/sink_test.go

@@ -36,10 +36,14 @@ type state struct {
 }
 
 // HandleEvent implements events.Adapter.
-func (s *state) HandleEvent(e events.Event) error {
+func (s *state) HandleEvent(ctx context.Context, e events.Event) error {
 	s.stateMu.Lock()
 	defer s.stateMu.Unlock()
 
+	if e.Node == "" {
+		return fmt.Errorf("node address information is empty")
+	}
+
 	switch msg := e.Payload.(type) {
 	case *machine.AddressEvent:
 		s.Addresses = msg.Addresses