From bea101a624a3e761adaeb30e16e595d6f70854fd Mon Sep 17 00:00:00 2001 From: Dmitry Golovin Date: Sun, 2 Aug 2026 16:36:56 +0300 Subject: [PATCH 1/3] processcollector: add network byte counters on Darwin Implements process_network_receive_bytes_total and process_network_transmit_bytes_total on macOS by talking to the undocumented "com.apple.network.statistics" kernel control socket (the same mechanism used by nettop/netstat), filtered to the current process's pid. No cgo or third-party dependency is required: the protocol is implemented directly on golang.org/x/sys/unix, using the same PF_SYSTEM/SYSPROTO_CONTROL socket mechanism already used elsewhere for utun. Struct layouts mirror bsd/net/ntstat.h from Apple's XNU source. Closes #1590. Signed-off-by: Dmitry Golovin --- CHANGELOG.md | 1 + prometheus/process_collector_darwin.go | 9 +- prometheus/process_collector_darwin_test.go | 4 + .../process_collector_mem_cgo_darwin.go | 3 - .../process_collector_mem_nocgo_darwin.go | 4 +- .../process_collector_netstat_darwin.go | 317 ++++++++++++++++++ 6 files changed, 331 insertions(+), 7 deletions(-) create mode 100644 prometheus/process_collector_netstat_darwin.go diff --git a/CHANGELOG.md b/CHANGELOG.md index e7a47c66e..cab2014a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ ## Unreleased * [FEATURE] testutil: Add GatherAndFormat to encode a subset of metrics from a Gatherer. #2091 +* [FEATURE] prometheus: `NewProcessCollector` on Darwin now reports `process_network_receive_bytes_total` and `process_network_transmit_bytes_total`, read via the `com.apple.network.statistics` kernel control socket (no cgo, no third-party dependency). Closes #1590. #2083 ## 1.24.1 / 2026-07-23 diff --git a/prometheus/process_collector_darwin.go b/prometheus/process_collector_darwin.go index 2b16298f4..c3498e9ca 100644 --- a/prometheus/process_collector_darwin.go +++ b/prometheus/process_collector_darwin.go @@ -136,6 +136,11 @@ func (c *processCollector) processCollect(ch chan<- Metric) { c.reportError(ch, c.maxVsize, err) } - // TODO: socket(PF_SYSTEM) to fetch "com.apple.network.statistics" might - // be able to get the per-process network send/receive counts. + if rxBytes, txBytes, err := getNetworkBytes(); err == nil { + ch <- MustNewConstMetric(c.inBytes, CounterValue, float64(rxBytes)) + ch <- MustNewConstMetric(c.outBytes, CounterValue, float64(txBytes)) + } else { + c.reportError(ch, c.inBytes, err) + c.reportError(ch, c.outBytes, err) + } } diff --git a/prometheus/process_collector_darwin_test.go b/prometheus/process_collector_darwin_test.go index 023cafce3..83df0145e 100644 --- a/prometheus/process_collector_darwin_test.go +++ b/prometheus/process_collector_darwin_test.go @@ -56,11 +56,15 @@ func TestDarwinProcessCollector(t *testing.T) { regexp.MustCompile("\nprocess_open_fds [1-9]"), regexp.MustCompile("\nprocess_virtual_memory_max_bytes (-1|[1-9])"), regexp.MustCompile("\nprocess_start_time_seconds [0-9]"), + regexp.MustCompile("\nprocess_network_receive_bytes_total [0-9]"), + regexp.MustCompile("\nprocess_network_transmit_bytes_total [0-9]"), regexp.MustCompile("\nfoobar_process_cpu_seconds_total [0-9]"), regexp.MustCompile("\nfoobar_process_max_fds [1-9]"), regexp.MustCompile("\nfoobar_process_open_fds [1-9]"), regexp.MustCompile("\nfoobar_process_virtual_memory_max_bytes (-1|[1-9])"), regexp.MustCompile("\nfoobar_process_start_time_seconds [0-9]"), + regexp.MustCompile("\nfoobar_process_network_receive_bytes_total [0-9]"), + regexp.MustCompile("\nfoobar_process_network_transmit_bytes_total [0-9]"), } { if !re.Match(buf.Bytes()) { t.Errorf("want body to match %s\n%s", re, buf.String()) diff --git a/prometheus/process_collector_mem_cgo_darwin.go b/prometheus/process_collector_mem_cgo_darwin.go index 9ac53f999..a77690a1c 100644 --- a/prometheus/process_collector_mem_cgo_darwin.go +++ b/prometheus/process_collector_mem_cgo_darwin.go @@ -43,9 +43,6 @@ func (c *processCollector) describe(ch chan<- *Desc) { ch <- c.startTime ch <- c.rss ch <- c.vsize - - /* the process could be collected but not implemented yet ch <- c.inBytes ch <- c.outBytes - */ } diff --git a/prometheus/process_collector_mem_nocgo_darwin.go b/prometheus/process_collector_mem_nocgo_darwin.go index 378865129..7476401b4 100644 --- a/prometheus/process_collector_mem_nocgo_darwin.go +++ b/prometheus/process_collector_mem_nocgo_darwin.go @@ -29,11 +29,11 @@ func (c *processCollector) describe(ch chan<- *Desc) { ch <- c.maxFDs ch <- c.maxVsize ch <- c.startTime + ch <- c.inBytes + ch <- c.outBytes /* the process could be collected but not implemented yet ch <- c.rss ch <- c.vsize - ch <- c.inBytes - ch <- c.outBytes */ } diff --git a/prometheus/process_collector_netstat_darwin.go b/prometheus/process_collector_netstat_darwin.go new file mode 100644 index 000000000..8205cc11a --- /dev/null +++ b/prometheus/process_collector_netstat_darwin.go @@ -0,0 +1,317 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build darwin && !ios + +package prometheus + +import ( + "bytes" + "encoding/binary" + "fmt" + "os" + "time" + + "golang.org/x/sys/unix" +) + +// This file implements a client for the undocumented "com.apple.network.statistics" +// kernel control socket, which is what Apple's own `nettop`/`netstat` tools use to +// report per-process network byte counters. There is no public Apple API for this +// and no cgo or third-party dependency is used: the protocol is implemented directly +// on top of golang.org/x/sys/unix, using the same PF_SYSTEM/SYSPROTO_CONTROL socket +// mechanism as the utun driver. Struct layouts below mirror bsd/net/ntstat.h from +// Apple's XNU source (https://github.com/apple/darwin-xnu/blob/main/bsd/net/ntstat.h). + +const ( + netStatControlName = "com.apple.network.statistics" + + nstatProviderTCPUserland uint32 = 3 + nstatProviderUDPUserland uint32 = 5 + + nstatMsgTypeAddAllSrcs uint32 = 1002 + nstatMsgTypeRemSrc uint32 = 1003 + nstatMsgTypeQuerySrc uint32 = 1004 + + nstatMsgTypeSuccess uint32 = 0 + nstatMsgTypeError uint32 = 1 + nstatMsgTypeSrcAdded uint32 = 10001 + nstatMsgTypeSrcCounts uint32 = 10004 + + // Restrict subscription to sources owned by a single pid, rather than + // system-wide (which would require elevated privileges anyway). + nstatFilterSpecificUserByPid uint64 = 0x01000000 + + nstatReadTimeout = 2 * time.Second +) + +type nstatMsgHdr struct { + Context uint64 + Type uint32 + Length uint16 + Flags uint16 +} + +type nstatMsgAddAllSrcs struct { + Hdr nstatMsgHdr + Filter uint64 + Events uint64 + Provider uint32 + TargetPid int32 + TargetUUID [16]byte +} + +type nstatMsgSrcAdded struct { + Hdr nstatMsgHdr + SrcRef uint64 + Provider uint32 + Reserved [4]byte +} + +type nstatMsgQuerySrcReq struct { + Hdr nstatMsgHdr + SrcRef uint64 +} + +// nstatCounts mirrors struct nstat_counts. Only the first four fields are used +// here; the rest are read (to consume the full wire message) but not exposed. +type nstatCounts struct { + RxPackets uint64 + RxBytes uint64 + TxPackets uint64 + TxBytes uint64 + CellRxBytes uint64 + CellTxBytes uint64 + WifiRxBytes uint64 + WifiTxBytes uint64 + WiredRxBytes uint64 + WiredTxBytes uint64 + RxDuplicateBytes uint32 + RxOutOfOrderBytes uint32 + TxRetransmit uint32 + ConnectAttempts uint32 + ConnectSuccesses uint32 + MinRtt uint32 + AvgRtt uint32 + VarRtt uint32 +} + +type nstatMsgSrcCounts struct { + Hdr nstatMsgHdr + SrcRef uint64 + EventFlags uint64 + Counts nstatCounts +} + +type nstatMsgErr struct { + Hdr nstatMsgHdr + Error uint32 + Reserved [4]byte +} + +// getNetworkBytes returns the total bytes received and sent over the network +// by the current process, summed across its TCP and UDP sockets. +func getNetworkBytes() (rxBytes, txBytes uint64, err error) { + fd, err := openNstatSocket() + if err != nil { + return 0, 0, err + } + defer unix.Close(fd) + + pid := int32(os.Getpid()) + for _, provider := range []uint32{nstatProviderTCPUserland, nstatProviderUDPUserland} { + refs, err := nstatCollectSrcRefs(fd, provider, pid) + if err != nil { + return 0, 0, fmt.Errorf("nstat: enumerating sources for provider %d: %w", provider, err) + } + for _, ref := range refs { + rx, tx, err := nstatQueryCounts(fd, ref) + if err != nil { + // The source may have been torn down between enumeration and + // query (e.g. a connection just closed); skip it rather than + // failing the whole collection. + continue + } + rxBytes += rx + txBytes += tx + } + } + + return rxBytes, txBytes, nil +} + +func openNstatSocket() (int, error) { + fd, err := unix.Socket(unix.AF_SYSTEM, unix.SOCK_DGRAM, 2 /* SYSPROTO_CONTROL */) + if err != nil { + return -1, fmt.Errorf("nstat: socket: %w", err) + } + + ctlInfo := &unix.CtlInfo{} + copy(ctlInfo.Name[:], netStatControlName) + if err := unix.IoctlCtlInfo(fd, ctlInfo); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("nstat: IoctlCtlInfo: %w", err) + } + + if err := unix.Connect(fd, &unix.SockaddrCtl{ID: ctlInfo.Id}); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("nstat: connect: %w", err) + } + + return fd, nil +} + +// nstatCollectSrcRefs subscribes to all sources of the given provider owned by +// pid, and returns the srcrefs the kernel reports. The kernel replies with zero +// or more SRC_ADDED messages followed by a SUCCESS message carrying the same +// context, which marks the end of enumeration. +func nstatCollectSrcRefs(fd int, provider uint32, pid int32) ([]uint64, error) { + const ctx = 1 + + req := nstatMsgAddAllSrcs{ + Hdr: nstatMsgHdr{Context: ctx, Type: nstatMsgTypeAddAllSrcs}, + Filter: nstatFilterSpecificUserByPid, + Provider: provider, + TargetPid: pid, + } + req.Hdr.Length = uint16(binary.Size(req)) + if err := nstatSend(fd, req); err != nil { + return nil, err + } + + deadline := time.Now().Add(nstatReadTimeout) + var refs []uint64 + buf := make([]byte, 4096) + for { + n, err := nstatRead(fd, buf, deadline) + if err != nil { + return nil, err + } + + hdr, err := nstatReadHdr(buf[:n]) + if err != nil { + return nil, err + } + + switch hdr.Type { + case nstatMsgTypeSrcAdded: + var m nstatMsgSrcAdded + if err := binary.Read(bytes.NewReader(buf[:n]), binary.LittleEndian, &m); err != nil { + return nil, fmt.Errorf("nstat: decoding SRC_ADDED: %w", err) + } + refs = append(refs, m.SrcRef) + case nstatMsgTypeSuccess: + if hdr.Context == ctx { + return refs, nil + } + case nstatMsgTypeError: + nErr, err := nstatReadErr(buf[:n]) + if err != nil { + return nil, err + } + return nil, fmt.Errorf("kernel returned errno %d", nErr) + } + } +} + +func nstatQueryCounts(fd int, srcref uint64) (rxBytes, txBytes uint64, err error) { + const ctx = 2 + + req := nstatMsgQuerySrcReq{ + Hdr: nstatMsgHdr{Context: ctx, Type: nstatMsgTypeQuerySrc}, + SrcRef: srcref, + } + req.Hdr.Length = uint16(binary.Size(req)) + if err := nstatSend(fd, req); err != nil { + return 0, 0, err + } + + deadline := time.Now().Add(nstatReadTimeout) + buf := make([]byte, 4096) + for { + n, err := nstatRead(fd, buf, deadline) + if err != nil { + return 0, 0, err + } + + hdr, err := nstatReadHdr(buf[:n]) + if err != nil { + return 0, 0, err + } + + switch hdr.Type { + case nstatMsgTypeSrcCounts: + var m nstatMsgSrcCounts + if err := binary.Read(bytes.NewReader(buf[:n]), binary.LittleEndian, &m); err != nil { + return 0, 0, fmt.Errorf("nstat: decoding SRC_COUNTS: %w", err) + } + return m.Counts.RxBytes, m.Counts.TxBytes, nil + case nstatMsgTypeError: + nErr, err := nstatReadErr(buf[:n]) + if err != nil { + return 0, 0, err + } + return 0, 0, fmt.Errorf("kernel returned errno %d", nErr) + } + } +} + +func nstatSend(fd int, msg any) error { + buf := &bytes.Buffer{} + if err := binary.Write(buf, binary.LittleEndian, msg); err != nil { + return fmt.Errorf("nstat: encoding request: %w", err) + } + if _, err := unix.Write(fd, buf.Bytes()); err != nil { + return fmt.Errorf("nstat: write: %w", err) + } + return nil +} + +// nstatRead reads one datagram from the control socket, applying deadline as a +// per-call receive timeout. +func nstatRead(fd int, buf []byte, deadline time.Time) (int, error) { + d := time.Until(deadline) + if d < 0 { + d = 0 + } + tv := unix.NsecToTimeval(d.Nanoseconds()) + if err := unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil { + return 0, fmt.Errorf("nstat: SetsockoptTimeval: %w", err) + } + n, err := unix.Read(fd, buf) + if err != nil { + return 0, fmt.Errorf("nstat: read: %w", err) + } + if n < int(unsafeSizeofNstatMsgHdr) { + return 0, fmt.Errorf("nstat: short read (%d bytes)", n) + } + return n, nil +} + +const unsafeSizeofNstatMsgHdr = 16 + +func nstatReadHdr(buf []byte) (nstatMsgHdr, error) { + var hdr nstatMsgHdr + if err := binary.Read(bytes.NewReader(buf), binary.LittleEndian, &hdr); err != nil { + return hdr, fmt.Errorf("nstat: decoding header: %w", err) + } + return hdr, nil +} + +func nstatReadErr(buf []byte) (uint32, error) { + var m nstatMsgErr + if err := binary.Read(bytes.NewReader(buf), binary.LittleEndian, &m); err != nil { + return 0, fmt.Errorf("nstat: decoding ERROR: %w", err) + } + return m.Error, nil +} From 839658f691d9913e95275d70b53a64cd435db76c Mon Sep 17 00:00:00 2001 From: Dmitry Golovin Date: Sun, 2 Aug 2026 16:48:13 +0300 Subject: [PATCH 2/3] fix license header year Signed-off-by: Dmitry Golovin --- prometheus/process_collector_netstat_darwin.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/prometheus/process_collector_netstat_darwin.go b/prometheus/process_collector_netstat_darwin.go index 8205cc11a..04f284d49 100644 --- a/prometheus/process_collector_netstat_darwin.go +++ b/prometheus/process_collector_netstat_darwin.go @@ -1,4 +1,4 @@ -// Copyright 2026 The Prometheus Authors +// Copyright The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at From 2bbbd1df4da731db8b022803c2504ad9df58abc9 Mon Sep 17 00:00:00 2001 From: Dmitry Golovin Date: Wed, 5 Aug 2026 15:29:34 +0300 Subject: [PATCH 3/3] process_collector_netstat_darwin: address review feedback - Trim unused trailing fields from nstatCounts, nstatMsgSrcAdded, and nstatMsgErr. binary.Read only reads sizeof(struct) bytes and leaves the rest of the datagram unread, so the Go structs only need to cover the fields we actually use. - Set SO_RCVTIMEO once in openNstatSocket instead of on every read. Did not adopt the suggestion to also query the TCP_KERNEL/UDP_KERNEL providers: verified via GET_SRC_DESC that NSTAT_FILTER_SPECIFIC_USER_BY_PID is not honored by those providers on this machine (they return every socket on the system, not just this process's), which would attribute other processes' traffic to this one. Tested with and without a local VPN active to rule that out as a factor; behavior was identical. Signed-off-by: Dmitry Golovin --- .../process_collector_netstat_darwin.go | 75 ++++++++----------- 1 file changed, 31 insertions(+), 44 deletions(-) diff --git a/prometheus/process_collector_netstat_darwin.go b/prometheus/process_collector_netstat_darwin.go index 04f284d49..68e0a1e21 100644 --- a/prometheus/process_collector_netstat_darwin.go +++ b/prometheus/process_collector_netstat_darwin.go @@ -71,11 +71,12 @@ type nstatMsgAddAllSrcs struct { TargetUUID [16]byte } +// nstatMsgSrcAdded mirrors the leading fields of struct nstat_msg_src_added; +// trailing Provider/Reserved fields are present on the wire but unused here, +// so binary.Read simply leaves them unread. type nstatMsgSrcAdded struct { - Hdr nstatMsgHdr - SrcRef uint64 - Provider uint32 - Reserved [4]byte + Hdr nstatMsgHdr + SrcRef uint64 } type nstatMsgQuerySrcReq struct { @@ -83,27 +84,15 @@ type nstatMsgQuerySrcReq struct { SrcRef uint64 } -// nstatCounts mirrors struct nstat_counts. Only the first four fields are used -// here; the rest are read (to consume the full wire message) but not exposed. +// nstatCounts mirrors the leading fields of struct nstat_counts. The real +// struct has more trailing fields (retransmits, RTT estimates, etc.); we only +// need the byte counters, and binary.Read leaves the rest of the message +// unread, so trimming here avoids coupling to the full XNU layout. type nstatCounts struct { - RxPackets uint64 - RxBytes uint64 - TxPackets uint64 - TxBytes uint64 - CellRxBytes uint64 - CellTxBytes uint64 - WifiRxBytes uint64 - WifiTxBytes uint64 - WiredRxBytes uint64 - WiredTxBytes uint64 - RxDuplicateBytes uint32 - RxOutOfOrderBytes uint32 - TxRetransmit uint32 - ConnectAttempts uint32 - ConnectSuccesses uint32 - MinRtt uint32 - AvgRtt uint32 - VarRtt uint32 + RxPackets uint64 + RxBytes uint64 + TxPackets uint64 + TxBytes uint64 } type nstatMsgSrcCounts struct { @@ -113,10 +102,11 @@ type nstatMsgSrcCounts struct { Counts nstatCounts } +// nstatMsgErr mirrors the leading fields of struct nstat_msg_error; the +// trailing Reserved field is unused here. type nstatMsgErr struct { - Hdr nstatMsgHdr - Error uint32 - Reserved [4]byte + Hdr nstatMsgHdr + Error uint32 } // getNetworkBytes returns the total bytes received and sent over the network @@ -168,6 +158,12 @@ func openNstatSocket() (int, error) { return -1, fmt.Errorf("nstat: connect: %w", err) } + tv := unix.NsecToTimeval(nstatReadTimeout.Nanoseconds()) + if err := unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("nstat: SetsockoptTimeval: %w", err) + } + return fd, nil } @@ -189,11 +185,10 @@ func nstatCollectSrcRefs(fd int, provider uint32, pid int32) ([]uint64, error) { return nil, err } - deadline := time.Now().Add(nstatReadTimeout) var refs []uint64 buf := make([]byte, 4096) for { - n, err := nstatRead(fd, buf, deadline) + n, err := nstatRead(fd, buf) if err != nil { return nil, err } @@ -236,10 +231,9 @@ func nstatQueryCounts(fd int, srcref uint64) (rxBytes, txBytes uint64, err error return 0, 0, err } - deadline := time.Now().Add(nstatReadTimeout) buf := make([]byte, 4096) for { - n, err := nstatRead(fd, buf, deadline) + n, err := nstatRead(fd, buf) if err != nil { return 0, 0, err } @@ -277,28 +271,21 @@ func nstatSend(fd int, msg any) error { return nil } -// nstatRead reads one datagram from the control socket, applying deadline as a -// per-call receive timeout. -func nstatRead(fd int, buf []byte, deadline time.Time) (int, error) { - d := time.Until(deadline) - if d < 0 { - d = 0 - } - tv := unix.NsecToTimeval(d.Nanoseconds()) - if err := unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil { - return 0, fmt.Errorf("nstat: SetsockoptTimeval: %w", err) - } +// nstatRead reads one datagram from the control socket. SO_RCVTIMEO is set +// once on the socket in openNstatSocket, so a stalled kernel response can't +// hang collection forever. +func nstatRead(fd int, buf []byte) (int, error) { n, err := unix.Read(fd, buf) if err != nil { return 0, fmt.Errorf("nstat: read: %w", err) } - if n < int(unsafeSizeofNstatMsgHdr) { + if n < nstatMsgHdrSize { return 0, fmt.Errorf("nstat: short read (%d bytes)", n) } return n, nil } -const unsafeSizeofNstatMsgHdr = 16 +const nstatMsgHdrSize = 16 func nstatReadHdr(buf []byte) (nstatMsgHdr, error) { var hdr nstatMsgHdr