Trigger connection established and closed events at switch layer

vyasgun · vyasgun · commit 7d21c394c49d · 2025-11-07T14:11:59.000+05:30
Send connection_established and connection_closed directly from the switch when MAC is learned or connection with the VM closes.

Assisted-by: Claude (Anthropic AI)
Signed-off-by: Gunjan Vyas &lt;vyasgun20@gmail.com&gt;
diff --git a/cmd/gvproxy/config.go b/cmd/gvproxy/config.go
@@ -12,6 +12,7 @@ import (
 	"slices"
 	"strings"
 
+	"github.com/containers/gvisor-tap-vsock/pkg/notification"
 	"github.com/containers/gvisor-tap-vsock/pkg/types"
 	log "github.com/sirupsen/logrus"
 	yaml "gopkg.in/yaml.v3"
@@ -65,6 +66,7 @@ type GvproxyConfig struct {
 	Services           string                 `yaml:"services,omitempty"`
 	Ec2MetadataAccess  bool                   `yaml:"ec2-metadata-access,omitempty"`
 	NotificationSocket string                 `yaml:"notification,omitempty"`
+	NotificationSender *notification.NotificationSender
 }
 
 type GvproxyConfigForward struct {
@@ -241,6 +243,7 @@ func GvproxyConfigure(config *GvproxyConfig, args *GvproxyArgs, version string)
 		config.PIDFile = args.pidFile
 	}
 	if args.notificationSocket != "" {
+		log.Debugf("notification socket: %s", args.notificationSocket)
 		uri, err := url.Parse(args.notificationSocket)
 		if err != nil {
 			return config, fmt.Errorf("invalid value for notification listen address: %w", err)
diff --git a/cmd/gvproxy/main.go b/cmd/gvproxy/main.go
@@ -128,13 +128,15 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 	}
 	log.Info("waiting for clients...")
 
+	// Start the notification sender in a goroutine
 	notificationSender := notification.NewNotificationSender(config.NotificationSocket)
 	if config.NotificationSocket != "" {
 		g.Go(func() error {
 			notificationSender.Start(ctx)
 			return nil
 		})
 	}
+	vn.SetNotificationSender(notificationSender)
 	for _, endpoint := range config.Listen {
 		log.Infof("listening %s", endpoint)
 		ln, err := transport.Listen(endpoint)
@@ -182,6 +184,7 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 	if config.Interfaces.VPNKit != "" {
 		vpnkitListener, err := transport.Listen(config.Interfaces.VPNKit)
 		if err != nil {
+			notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 			return fmt.Errorf("vpnkit listen error: %w", err)
 		}
 		g.Go(func() error {
@@ -195,10 +198,10 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 				}
 				conn, err := vpnkitListener.Accept()
 				if err != nil {
+					notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 					log.Errorf("vpnkit accept error: %s", err)
 					continue
 				}
-				notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
 				g.Go(func() error {
 					return vn.AcceptVpnKit(conn)
 				})
@@ -210,6 +213,7 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 	if config.Interfaces.Qemu != "" {
 		qemuListener, err := transport.Listen(config.Interfaces.Qemu)
 		if err != nil {
+			notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 			return fmt.Errorf("qemu listen error: %w", err)
 		}
 
@@ -227,14 +231,14 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 				notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 				return fmt.Errorf("qemu accept error: %w", err)
 			}
-			notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
 			return vn.AcceptQemu(ctx, conn)
 		})
 	}
 
 	if config.Interfaces.Bess != "" {
 		bessListener, err := transport.Listen(config.Interfaces.Bess)
 		if err != nil {
+			notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 			return fmt.Errorf("bess listen error: %w", err)
 		}
 
@@ -252,7 +256,6 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 				notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 				return fmt.Errorf("bess accept error: %w", err)
 			}
-			notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
 			return vn.AcceptBess(ctx, conn)
 		})
 	}
@@ -276,20 +279,21 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 		g.Go(func() error {
 			vfkitConn, err := transport.AcceptVfkit(conn)
 			if err != nil {
-				notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 				return fmt.Errorf("vfkit accept error: %w", err)
 			}
 
-			notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
 			return vn.AcceptVfkit(ctx, vfkitConn)
 		})
 	}
 
 	if config.Interfaces.Stdio != "" {
 		g.Go(func() error {
 			conn := stdio.GetStdioConn()
-			notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
-			return vn.AcceptStdio(ctx, conn)
+			err := vn.AcceptStdio(ctx, conn)
+			if err != nil {
+				notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
+			}
+			return err
 		})
 	}
 
diff --git a/pkg/tap/switch.go b/pkg/tap/switch.go
@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 	"syscall"
 
+	"github.com/containers/gvisor-tap-vsock/pkg/notification"
 	"github.com/containers/gvisor-tap-vsock/pkg/types"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
@@ -48,6 +49,8 @@ type Switch struct {
 	writeLock sync.Mutex
 
 	gateway VirtualDevice
+
+	notificationSender *notification.NotificationSender
 }
 
 func NewSwitch(debug bool, mtu int) *Switch {
@@ -197,6 +200,12 @@ func (e *Switch) disconnect(id int, conn net.Conn) {
 
 	for address, targetConn := range e.cam {
 		if targetConn == id {
+			if e.notificationSender != nil {
+				e.notificationSender.Send(types.NotificationMessage{
+					NotificationType: types.ConnectionClosed,
+					MacAddress:       address.String(),
+				})
+			}
 			delete(e.cam, address)
 		}
 	}
@@ -267,9 +276,17 @@ func (e *Switch) rxBuf(_ context.Context, id int, buf []byte) {
 	eth := header.Ethernet(buf)
 
 	e.camLock.Lock()
+	_, exists := e.cam[eth.SourceAddress()]
 	e.cam[eth.SourceAddress()] = id
 	e.camLock.Unlock()
 
+	if !exists && e.notificationSender != nil {
+		e.notificationSender.Send(types.NotificationMessage{
+			NotificationType: types.ConnectionEstablished,
+			MacAddress:       eth.SourceAddress().String(),
+		})
+	}
+
 	if eth.DestinationAddress() != e.gateway.LinkAddress() {
 		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 			Payload: buffer.MakeWithData(buf),
@@ -304,3 +321,7 @@ func protocolImplementation(protocol types.Protocol) protocol {
 		return &hyperkitProtocol{}
 	}
 }
+
+func (e *Switch) SetNotificationSender(notificationSender *notification.NotificationSender) {
+	e.notificationSender = notificationSender
+}
diff --git a/pkg/types/handshake.go b/pkg/types/handshake.go
@@ -22,6 +22,7 @@ type UnexposeRequest struct {
 
 type NotificationMessage struct {
 	NotificationType NotificationType `json:"notification_type"`
+	MacAddress       string           `json:"mac_address,omitempty"`
 }
 
 type NotificationType string
@@ -31,4 +32,5 @@ const (
 	ConnectionEstablished NotificationType = "connection_established"
 	HypervisorWarning     NotificationType = "hypervisor_warning"
 	HypervisorError       NotificationType = "hypervisor_error"
+	ConnectionClosed      NotificationType = "connection_closed"
 )
diff --git a/pkg/virtualnetwork/virtualnetwork.go b/pkg/virtualnetwork/virtualnetwork.go
@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"os"
 
+	"github.com/containers/gvisor-tap-vsock/pkg/notification"
 	"github.com/containers/gvisor-tap-vsock/pkg/tap"
 	"github.com/containers/gvisor-tap-vsock/pkg/types"
 	"gvisor.dev/gvisor/pkg/tcpip"
@@ -28,6 +29,10 @@ type VirtualNetwork struct {
 	ipPool        *tap.IPPool
 }
 
+func (n *VirtualNetwork) SetNotificationSender(notificationSender *notification.NotificationSender) {
+	n.networkSwitch.SetNotificationSender(notificationSender)
+}
+
 func New(configuration *types.Configuration) (*VirtualNetwork, error) {
 	_, subnet, err := net.ParseCIDR(configuration.Subnet)
 	if err != nil {

Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,7 @@ type UnexposeRequest struct {`
`22`	`22`
`23`	`23`	`type NotificationMessage struct {`
`24`	`24`	NotificationType NotificationType `json:"notification_type"`
	`25`	+ MacAddress string `json:"mac_address,omitempty"`
`25`	`26`	`}`
`26`	`27`
`27`	`28`	`type NotificationType string`
`@@ -31,4 +32,5 @@ const (`
`31`	`32`	`ConnectionEstablished NotificationType = "connection_established"`
`32`	`33`	`HypervisorWarning NotificationType = "hypervisor_warning"`
`33`	`34`	`HypervisorError NotificationType = "hypervisor_error"`
	`35`	`+ ConnectionClosed NotificationType = "connection_closed"`
`34`	`36`	`)`