diff --git a/engine/resources/net.go b/engine/resources/net.go index 248e3e51..98f99a23 100644 --- a/engine/resources/net.go +++ b/engine/resources/net.go @@ -42,6 +42,7 @@ import ( // do not clean up spawned goroutines. Should be replaced when a suitable // alternative is available. "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" ) func init() { @@ -195,7 +196,7 @@ func (obj *NetRes) Watch() error { defer wg.Wait() // create a netlink socket for receiving network interface events - conn, err := socketset.NewSocketSet(rtmGrps, obj.socketFile) + conn, err := socketset.NewSocketSet(rtmGrps, obj.socketFile, unix.NETLINK_ROUTE) if err != nil { return errwrap.Wrapf(err, "error creating socket set") } @@ -223,7 +224,7 @@ func (obj *NetRes) Watch() error { defer close(nlChan) for { // receive messages from the socket set - msgs, err := conn.Receive() + msgs, err := conn.ReceiveNetlinkMessages() if err != nil { select { case nlChan <- &nlChanStruct{ diff --git a/util/socketset/socketset.go b/util/socketset/socketset.go index e7a9453a..ccaa9e15 100644 --- a/util/socketset/socketset.go +++ b/util/socketset/socketset.go @@ -15,14 +15,16 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -// Package socketset is in API for creating a select style netlink -// socket to receive events from the kernel. +// Package socketset is in API for creating a select style netlink socket to +// receive events from the kernel. package socketset import ( + "bytes" "fmt" "os" "path" + "strings" "syscall" errwrap "github.com/pkg/errors" @@ -30,8 +32,8 @@ import ( ) // SocketSet is used to receive events from a socket and shut it down cleanly -// when asked. It contains a socket for events and a pipe socket to unblock -// receive on shutdown. +// when asked. It contains a socket for events and a pipe socket to unblock receive +// on shutdown. type SocketSet struct { fdEvents int fdPipe int @@ -39,63 +41,45 @@ type SocketSet struct { } // NewSocketSet returns a socketSet, initialized with the given parameters. -func NewSocketSet(groups uint32, file string) (*SocketSet, error) { - // make a netlink socket file descriptor - fdEvents, err := unix.Socket(unix.AF_NETLINK, unix.SOCK_RAW, unix.NETLINK_ROUTE) +func NewSocketSet(groups uint32, name string, proto int) (*SocketSet, error) { + fdEvents, err := unix.Socket(unix.AF_NETLINK, unix.SOCK_RAW, proto) if err != nil { return nil, errwrap.Wrapf(err, "error creating netlink socket") } + // bind to the socket and add add the netlink groups we need to get events if err := unix.Bind(fdEvents, &unix.SockaddrNetlink{ Family: unix.AF_NETLINK, Groups: groups, + Pid: uint32(os.Getpid()), // set PID to our process }); err != nil { return nil, errwrap.Wrapf(err, "error binding netlink socket") } - // create a pipe socket to unblock unix.Select when we close + // this pipe unblocks unix.Select upon closing fdPipe, err := unix.Socket(unix.AF_UNIX, unix.SOCK_RAW, unix.PROT_NONE) if err != nil { return nil, errwrap.Wrapf(err, "error creating pipe socket") } + // bind the pipe to a file if err = unix.Bind(fdPipe, &unix.SockaddrUnix{ - Name: file, + Name: name, }); err != nil { return nil, errwrap.Wrapf(err, "error binding pipe socket") } + return &SocketSet{ fdEvents: fdEvents, fdPipe: fdPipe, - pipeFile: file, + pipeFile: name, }, nil } -// Shutdown closes the event file descriptor and unblocks receive by sending -// a message to the pipe file descriptor. It must be called before close, and -// should only be called once. -func (obj *SocketSet) Shutdown() error { - // close the event socket so no more events are produced - if err := unix.Close(obj.fdEvents); err != nil { - return err - } - // send a message to the pipe to unblock select - return unix.Sendto(obj.fdPipe, nil, 0, &unix.SockaddrUnix{ - Name: path.Join(obj.pipeFile), - }) -} - -// Close closes the pipe file descriptor. It must only be called after -// shutdown has closed fdEvents, and unblocked receive. It should only be -// called once. -func (obj *SocketSet) Close() error { - return unix.Close(obj.fdPipe) -} - -// Receive waits for bytes from fdEvents and parses them into a slice of -// netlink messages. It will block until an event is produced, or shutdown -// is called. -func (obj *SocketSet) Receive() ([]syscall.NetlinkMessage, error) { +// ReceiveBytes waits for bytes from fdEvents and return a byte array truncated +// to the message length. It will block until an event is produced, or shutdown is +// called. +func (obj *SocketSet) ReceiveBytes() ([]byte, error) { // Select will return when any fd in fdSet (fdEvents and fdPipe) is ready // to read. _, err := unix.Select(obj.nfd(), obj.fdSet(), nil, nil, nil) @@ -121,8 +105,95 @@ func (obj *SocketSet) Receive() ([]syscall.NetlinkMessage, error) { return nil, fmt.Errorf("received short header") } b = b[:n] // truncate b to message length + return b, nil +} + +// ReceiveNetlinkMessages is a wrapper around ReceiveBytes that parses the bytes +// from the event and returns an array of NetlinkMessages. +func (obj *SocketSet) ReceiveNetlinkMessages() ([]syscall.NetlinkMessage, error) { + msgBytes, err := obj.ReceiveBytes() + if err != nil { + return nil, err + } // use syscall to parse, as func does not exist in x/sys/unix - return syscall.ParseNetlinkMessage(b) + return syscall.ParseNetlinkMessage(msgBytes) +} + +// UEvent struct has attributes for KOBJECT_NETWORK_UEVENT, passed from the +// kernel. +type UEvent struct { + // default keys, as per https://github.com/torvalds/linux/blob/master/lib/kobject_uevent.c + Action string + Devpath string + Subsystem string + + // every other KV pair + Data map[string]string +} + +// ReceiveUEvent is a wrapper around ReceiveBytes. Parses the UEvent data +// receieved from the socket and puts it into a UEvent struct. +func (obj *SocketSet) ReceiveUEvent() (*UEvent, error) { + // TODO: can multiple events come in the same socket? + event := &UEvent{Data: map[string]string{}} + + msgBytes, err := obj.ReceiveBytes() + if err != nil { + return nil, err + } + + submsg := msgBytes[:] + i := 0 +Loop: + for { + submsg = submsg[i:] + n := bytes.IndexByte(submsg, 0x0) + if n == -1 { + break Loop + } + i = n + 1 + + attrLine := string(submsg[:n]) + split := strings.SplitN(attrLine, "=", 2) + if len(split) < 2 { + continue + } + switch split[0] { + case "ACTION": + event.Action = split[1] + case "DEVPATH": + event.Devpath = split[1] + case "SUBSYSTEM": + event.Subsystem = split[1] + default: + event.Data[split[0]] = split[1] + } + } + + return event, nil +} + +// Shutdown closes the event file descriptor and unblocks receive by sending a +// message to the pipe file descriptor. It must be called before close, and +// should only be called once. +func (obj *SocketSet) Shutdown() error { + // close the event socket so no more events are produced + if err := unix.Close(obj.fdEvents); err != nil { + return err + } + // send a message to the pipe to unblock select + return unix.Sendto(obj.fdPipe, nil, 0, &unix.SockaddrUnix{ + Name: path.Join(obj.pipeFile), + }) +} + +// Close closes the pipe file descriptor. It must only be called after Shutdown +// has closed fdEvents, and unblocked receive. It should only be called once. +func (obj *SocketSet) Close() error { + if err := unix.Unlink(obj.pipeFile); err != nil { + return errwrap.Wrapf(err, "could not unbind %s", obj.pipeFile) + } + return unix.Close(obj.fdPipe) } // nfd returns one more than the highest fd value in the struct, for use as as @@ -135,11 +206,11 @@ func (obj *SocketSet) nfd() int { return obj.fdPipe + 1 } -// fdSet returns a bitmask representation of the integer values of fdEvents -// and fdPipe. See man select for more info. +// fdSet returns a bitmask representation of the integer values of fdEvents and +// fdPipe. See man select for more info. func (obj *SocketSet) fdSet() *unix.FdSet { fdSet := &unix.FdSet{} - // Generate the bitmask representing the file descriptors in the socketSet. + // Generate the bitmask representing the file descriptors in the SocketSet. // The rightmost bit corresponds to file descriptor zero, and each bit to // the left represents the next file descriptor number in the sequence of // all real numbers. E.g. the FdSet containing containing 0 and 4 is 10001.