feat: add notification mechanism for network lifecycle events

Introduces a channel-based notification sender that sends JSON messages
over Unix domain sockets for key events:
- Ready: network stack is ready for connections
- ConnectionEstablished: VM connection accepted
- HypervisorError: error during VM connection setup

Signed-off-by: Gunjan Vyas <[email protected]>

Author: vyasgun

cmd/gvproxy/config.go

@@ -26,25 +26,26 @@ const (
 )
 
 type GvproxyArgs struct {
-	config            string
-	endpoints         arrayFlags
-	debug             bool
-	mtu               int
-	sshPort           int
-	vpnkitSocket      string
-	qemuSocket        string
-	bessSocket        string
-	stdioSocket       string
-	vfkitSocket       string
-	forwardSocket     arrayFlags
-	forwardDest       arrayFlags
-	forwardUser       arrayFlags
-	forwardIdentify   arrayFlags
-	pidFile           string
-	pcapFile          string
-	logFile           string
-	servicesEndpoint  string
-	ec2MetadataAccess bool
+	config             string
+	endpoints          arrayFlags
+	debug              bool
+	mtu                int
+	sshPort            int
+	vpnkitSocket       string
+	qemuSocket         string
+	bessSocket         string
+	stdioSocket        string
+	vfkitSocket        string
+	notificationSocket string
+	forwardSocket      arrayFlags
+	forwardDest        arrayFlags
+	forwardUser        arrayFlags
+	forwardIdentify    arrayFlags
+	pidFile            string
+	pcapFile           string
+	logFile            string
+	servicesEndpoint   string
+	ec2MetadataAccess  bool
 }
 
 type GvproxyConfig struct {
@@ -58,11 +59,12 @@ type GvproxyConfig struct {
 		Stdio  string `yaml:"stdio,omitempty"`
 		Vfkit  string `yaml:"vfkit,omitempty"`
 	} `yaml:"interfaces,omitempty"`
-	Forwards          []GvproxyConfigForward `yaml:"forwards,omitempty"`
-	PIDFile           string                 `yaml:"pid-file,omitempty"`
-	LogFile           string                 `yaml:"log-file,omitempty"`
-	Services          string                 `yaml:"services,omitempty"`
-	Ec2MetadataAccess bool                   `yaml:"ec2-metadata-access,omitempty"`
+	Forwards           []GvproxyConfigForward `yaml:"forwards,omitempty"`
+	PIDFile            string                 `yaml:"pid-file,omitempty"`
+	LogFile            string                 `yaml:"log-file,omitempty"`
+	Services           string                 `yaml:"services,omitempty"`
+	Ec2MetadataAccess  bool                   `yaml:"ec2-metadata-access,omitempty"`
+	NotificationSocket string                 `yaml:"notification,omitempty"`
 }
 
 type GvproxyConfigForward struct {
@@ -130,6 +132,7 @@ func GvproxyArgParse(flagSet *flag.FlagSet, args *GvproxyArgs, argv []string) (*
 	flagSet.StringVar(&args.logFile, "log-file", "", "Output log messages (logrus) to a given file path")
 	flagSet.StringVar(&args.servicesEndpoint, "services", "", "Exposes the same HTTP API as the --listen flag, without the /connect endpoint")
 	flagSet.BoolVar(&args.ec2MetadataAccess, "ec2-metadata-access", false, "Permits access to EC2 Metadata Service (TCP only)")
+	flagSet.StringVar(&args.notificationSocket, "notification", "", "Socket to be used to send network-ready notifications")
 	if err := flagSet.Parse(argv); err != nil {
 		return nil, err
 	}
@@ -237,6 +240,16 @@ func GvproxyConfigure(config *GvproxyConfig, args *GvproxyArgs, version string)
 	if args.pidFile != "" {
 		config.PIDFile = args.pidFile
 	}
+	if args.notificationSocket != "" {
+		uri, err := url.Parse(args.notificationSocket)
+		if err != nil {
+			return config, fmt.Errorf("invalid value for notification listen address: %w", err)
+		}
+		if uri.Scheme != "unix" {
+			return config, errors.New("notification listen address must be unix:// address")
+		}
+		config.NotificationSocket = uri.Path
+	}
 	if len(args.endpoints) > 0 {
 		config.Listen = args.endpoints
 	}

cmd/gvproxy/main.go

@@ -18,8 +18,10 @@ import (
 	"time"
 
 	"github.com/containers/gvisor-tap-vsock/pkg/net/stdio"
+	"github.com/containers/gvisor-tap-vsock/pkg/notification"
 	"github.com/containers/gvisor-tap-vsock/pkg/sshclient"
 	"github.com/containers/gvisor-tap-vsock/pkg/transport"
+	"github.com/containers/gvisor-tap-vsock/pkg/types"
 	"github.com/containers/gvisor-tap-vsock/pkg/virtualnetwork"
 	"github.com/containers/winquit/pkg/winquit"
 	humanize "github.com/dustin/go-humanize"
@@ -126,6 +128,13 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 	}
 	log.Info("waiting for clients...")
 
+	notificationSender := notification.NewNotificationSender(config.NotificationSocket)
+	if config.NotificationSocket != "" {
+		g.Go(func() error {
+			notificationSender.Start(ctx)
+			return nil
+		})
+	}
 	for _, endpoint := range config.Listen {
 		log.Infof("listening %s", endpoint)
 		ln, err := transport.Listen(endpoint)
@@ -134,6 +143,7 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 		}
 		httpServe(ctx, g, ln, withProfiler(vn))
 	}
+	notificationSender.Send(types.NotificationMessage{NotificationType: types.Ready})
 
 	if config.Services != "" {
 		log.Infof("enabling services API. Listening %s", config.Services)
@@ -188,6 +198,7 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 					log.Errorf("vpnkit accept error: %s", err)
 					continue
 				}
+				notificationSender.Send(types.NotificationMessage{NotificationType: types.ConnectionEstablished})
 				g.Go(func() error {
 					return vn.AcceptVpnKit(conn)
 				})
@@ -213,8 +224,10 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 		g.Go(func() error {
 			conn, err := qemuListener.Accept()
 			if err != nil {
+				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)
 		})
 	}
@@ -236,15 +249,18 @@ func run(ctx context.Context, g *errgroup.Group, config *GvproxyConfig) error {
 		g.Go(func() error {
 			conn, err := bessListener.Accept()
 			if err != nil {
+				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)
 		})
 	}
 
 	if config.Interfaces.Vfkit != "" {
 		conn, err := transport.ListenUnixgram(config.Interfaces.Vfkit)
 		if err != nil {
+			notificationSender.Send(types.NotificationMessage{NotificationType: types.HypervisorError})
 			return fmt.Errorf("vfkit listen error: %w", err)
 		}
 
@@ -260,15 +276,19 @@ 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)
 		})
 	}

pkg/notification/sender.go

@@ -0,0 +1,75 @@
+package notification
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+
+	"github.com/containers/gvisor-tap-vsock/pkg/types"
+	log "github.com/sirupsen/logrus"
+)
+
+type NotificationSender struct {
+	notificationCh chan types.NotificationMessage
+	socket         string
+}
+
+func NewNotificationSender(socket string) *NotificationSender {
+	if socket == "" {
+		return &NotificationSender{
+			socket:         "",
+			notificationCh: nil,
+		}
+	}
+
+	return &NotificationSender{
+		socket:         socket,
+		notificationCh: make(chan types.NotificationMessage, 100),
+	}
+}
+
+func (s *NotificationSender) Send(notification types.NotificationMessage) {
+	if s.notificationCh == nil {
+		return
+	}
+	select {
+	case s.notificationCh <- notification:
+	default:
+		log.Warn("unable to send notification")
+	}
+}
+
+func (s *NotificationSender) Start(ctx context.Context) {
+	if s.notificationCh == nil {
+		return
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case notification := <-s.notificationCh:
+			if err := s.sendToSocket(notification); err != nil {
+				log.Errorf("failed to send notification: %v", err)
+				continue
+			}
+		}
+	}
+}
+
+func (s *NotificationSender) sendToSocket(notification types.NotificationMessage) error {
+	if s.socket == "" {
+		return nil
+	}
+	conn, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: s.socket, Net: "unix"})
+	if err != nil {
+		return fmt.Errorf("cannot dial notification socket: %w", err)
+	}
+	defer conn.Close()
+	enc := json.NewEncoder(conn)
+	if err := enc.Encode(notification); err != nil {
+		return fmt.Errorf("failed to encode notification: %w", err)
+	}
+	return nil
+}

pkg/types/handshake.go

@@ -19,3 +19,16 @@ type UnexposeRequest struct {
 	Local    string            `json:"local"`
 	Protocol TransportProtocol `json:"protocol"`
 }
+
+type NotificationMessage struct {
+	NotificationType NotificationType `json:"notification_type"`
+}
+
+type NotificationType string
+
+const (
+	Ready                 NotificationType = "ready"
+	ConnectionEstablished NotificationType = "connection_established"
+	HypervisorWarning     NotificationType = "hypervisor_warning"
+	HypervisorError       NotificationType = "hypervisor_error"
+)