From 7c0c2b0307934320e94b00f976ec2b053edeb8fc Mon Sep 17 00:00:00 2001 From: "randomizedcoder dave.seddon.ca@gmail.com" Date: Sun, 2 Aug 2026 15:59:41 -0700 Subject: [PATCH] test(inttest): udp/unix/unixgram raw-dest e2e (generalize socket-sink) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generalize the merged tcp-sink flavor into a scheme-parameterized socket-sink and add udp-sink, unix-sink, unixgram-sink lifecycle flavors, completing daemon-level e2e coverage for the raw socket destinations (they were Go-component-only). Each streams jsonl over `-dest :...` to an in-VM receiver; the RAW_SOCKET self-test check validates the received records carry a non-empty hostname AND the per-scheme send counter ran. Framing-agnostic check: `grep '{...}'` extracts JSON objects, which strips the unix STREAM dest's varint length prefix and is a no-op for the newline-delimited jsonl of tcp/udp/unixgram. Per-scheme send metric (destTCP/Writes, Inetdiager/udpWrites, destUnix/Writes, destUnixGram/Writes). Receivers: ncat listens dual-stack on [::] for tcp/udp and on the socket path for unix STREAM. unixgram uses socat's UNIX-RECV — ncat can't LISTEN on a unix DATAGRAM socket (`--unixsock --udp --listen` fails with "connect: Invalid argument"). Dual-stack for inet dests: xtcp2 dials `tcp/udp:localhost:...` (127.0.0.1 AND ::1); the sink binds [::] (bindv6only=0 → also IPv4-mapped). Wiring: mkVm.nix (socketSinkScheme/Dest/ReceiverCmd/Metric* bindings, generalized ncat/socat sink service), self-test.nix (Check 5f generalized), default.nix (mkOneSocketSink + mkLifecycleSocketSink helpers → 4 flavors), nix/default.nix (boot + lifecycle apps/packages). Run: nix run .#microvm-x86_64-lifecycle-{udp,unix,unixgram}-sink. Verified (lifecycle microVM, all OVERALL_PASS): udp records=65 Inetdiager/udpWrites=11 unix records=131 destUnix/Writes=14 (varint framing, malformed=0) unixgram records=86 destUnixGram/Writes=12 Co-Authored-By: Claude Opus 4.8 --- nix/default.nix | 18 ++++++ nix/microvms/default.nix | 55 ++++++++++++------ nix/microvms/mkVm.nix | 111 ++++++++++++++++++++++++++++--------- nix/microvms/self-test.nix | 62 ++++++++++++--------- 4 files changed, 176 insertions(+), 70 deletions(-) diff --git a/nix/default.nix b/nix/default.nix index e46f5e4..b2e264c 100644 --- a/nix/default.nix +++ b/nix/default.nix @@ -318,6 +318,9 @@ in microvm-x86_64-s3parquet-pipeline = microvms.vmsS3Parquet.x86_64; microvm-x86_64-valkey = microvms.vmsValkey.x86_64; microvm-x86_64-tcp-sink = microvms.vmsTcpSink.x86_64; + microvm-x86_64-udp-sink = microvms.vmsUdpSink.x86_64; + microvm-x86_64-unix-sink = microvms.vmsUnixSink.x86_64; + microvm-x86_64-unixgram-sink = microvms.vmsUnixgramSink.x86_64; microvm-x86_64-nats = microvms.vmsNats.x86_64; microvm-x86_64-nsq = microvms.vmsNsq.x86_64; microvm-x86_64-s3parquet-long = microvms.vmsS3ParquetLong.x86_64; @@ -340,6 +343,9 @@ in test-microvm-lifecycle-x86_64-clickhouse-http = microvms.lifecycleClickHttp.x86_64.fullTest; test-microvm-lifecycle-x86_64-valkey = microvms.lifecycleValkey.x86_64.fullTest; test-microvm-lifecycle-x86_64-tcp-sink = microvms.lifecycleTcpSink.x86_64.fullTest; + test-microvm-lifecycle-x86_64-udp-sink = microvms.lifecycleUdpSink.x86_64.fullTest; + test-microvm-lifecycle-x86_64-unix-sink = microvms.lifecycleUnixSink.x86_64.fullTest; + test-microvm-lifecycle-x86_64-unixgram-sink = microvms.lifecycleUnixgramSink.x86_64.fullTest; test-microvm-lifecycle-x86_64-nats = microvms.lifecycleNats.x86_64.fullTest; test-microvm-lifecycle-x86_64-nsq = microvms.lifecycleNsq.x86_64.fullTest; test-microvm-lifecycle-x86_64-coverage = microvms.lifecycleCoverage.x86_64.fullTest; @@ -398,6 +404,18 @@ in type = "app"; program = "${microvms.lifecycleTcpSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-tcp-sink"; }; + microvm-x86_64-lifecycle-udp-sink = { + type = "app"; + program = "${microvms.lifecycleUdpSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-udp-sink"; + }; + microvm-x86_64-lifecycle-unix-sink = { + type = "app"; + program = "${microvms.lifecycleUnixSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-unix-sink"; + }; + microvm-x86_64-lifecycle-unixgram-sink = { + type = "app"; + program = "${microvms.lifecycleUnixgramSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-unixgram-sink"; + }; microvm-x86_64-lifecycle-nats = { type = "app"; program = "${microvms.lifecycleNats.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-nats"; diff --git a/nix/microvms/default.nix b/nix/microvms/default.nix index 1c4fa28..2905c0d 100644 --- a/nix/microvms/default.nix +++ b/nix/microvms/default.nix @@ -230,11 +230,12 @@ let sink = "valkey"; }; - # tcp-sink lifecycle flavor: xtcp2 streams jsonl over the raw TCP dest to an - # in-VM ncat receiver; the self-test validates the received records + - # destTCP counter (RAW_SOCKET). Native, no docker/broker. - mkOneTcpSink = - arch: + # socket-sink lifecycle flavors: xtcp2 streams jsonl over the raw + # tcp/udp/unix/unixgram dest to an in-VM ncat receiver; the self-test + # validates the received records + the per-scheme send counter (RAW_SOCKET). + # Native, no docker/broker. + mkOneSocketSink = + sinkName: arch: import ./mkVm.nix { inherit pkgs @@ -245,8 +246,12 @@ let xtcp2Package xtcp2AllPackage ; - sink = "tcp-sink"; + sink = sinkName; }; + mkOneTcpSink = mkOneSocketSink "tcp-sink"; + mkOneUdpSink = mkOneSocketSink "udp-sink"; + mkOneUnixSink = mkOneSocketSink "unix-sink"; + mkOneUnixgramSink = mkOneSocketSink "unixgram-sink"; # nats lifecycle flavor: native in-VM NATS server + pre-subscribed consumer. mkOneNats = @@ -413,6 +418,9 @@ let vmsValkey = lib.genAttrs constants.supportedArchs mkOneValkey; vmsTcpSink = lib.genAttrs constants.supportedArchs mkOneTcpSink; + vmsUdpSink = lib.genAttrs constants.supportedArchs mkOneUdpSink; + vmsUnixSink = lib.genAttrs constants.supportedArchs mkOneUnixSink; + vmsUnixgramSink = lib.genAttrs constants.supportedArchs mkOneUnixgramSink; vmsNats = lib.genAttrs constants.supportedArchs mkOneNats; @@ -444,17 +452,24 @@ let }; }); - lifecycleTcpSink = lib.genAttrs constants.supportedArchs (arch: { - fullTest = microvmLib.mkLifecycleFullTest { - inherit arch; - vm = vmsTcpSink.${arch}; - suffix = "-tcp-sink"; - # Baseline sentinels plus the raw-socket verdict. Native (no docker), but - # the self-test waits for records to stream over TCP + reach the sink. - sentinelRe = "SYSTEMD|METRICS|NETLINK|BINARIES_HELP|GRPC_ROUNDTRIP|POLL_STREAM|HEALTH|OUTPUT_CONTENT|LISTEN_STREAM|RAW_SOCKET|NS_INSPECT|NSTEST|NS_LIFECYCLE|NS_TRAFFIC|NS_DOCKER|NS_ANONYMOUS|CTL_HOT|CTL_TRIGGER|CTL_RESTART|OVERALL"; - timeoutSec = 240; - }; - }); + # Lifecycle runner for a socket-sink flavor (tcp/udp/unix/unixgram). Baseline + # sentinels plus the raw-socket verdict. Native (no docker), but the self-test + # waits for records to stream over the socket + reach the sink. + mkLifecycleSocketSink = + vmsAttr: suffixName: + lib.genAttrs constants.supportedArchs (arch: { + fullTest = microvmLib.mkLifecycleFullTest { + inherit arch; + vm = vmsAttr.${arch}; + suffix = suffixName; + sentinelRe = "SYSTEMD|METRICS|NETLINK|BINARIES_HELP|GRPC_ROUNDTRIP|POLL_STREAM|HEALTH|OUTPUT_CONTENT|LISTEN_STREAM|RAW_SOCKET|NS_INSPECT|NSTEST|NS_LIFECYCLE|NS_TRAFFIC|NS_DOCKER|NS_ANONYMOUS|CTL_HOT|CTL_TRIGGER|CTL_RESTART|OVERALL"; + timeoutSec = 240; + }; + }); + lifecycleTcpSink = mkLifecycleSocketSink vmsTcpSink "-tcp-sink"; + lifecycleUdpSink = mkLifecycleSocketSink vmsUdpSink "-udp-sink"; + lifecycleUnixSink = mkLifecycleSocketSink vmsUnixSink "-unix-sink"; + lifecycleUnixgramSink = mkLifecycleSocketSink vmsUnixgramSink "-unixgram-sink"; lifecycleNats = lib.genAttrs constants.supportedArchs (arch: { fullTest = microvmLib.mkLifecycleFullTest { @@ -648,6 +663,9 @@ in vmsS3Parquet vmsValkey vmsTcpSink + vmsUdpSink + vmsUnixSink + vmsUnixgramSink vmsNats vmsNsq vmsS3ParquetLong @@ -666,6 +684,9 @@ in lifecycleS3Parquet lifecycleValkey lifecycleTcpSink + lifecycleUdpSink + lifecycleUnixSink + lifecycleUnixgramSink lifecycleNats lifecycleNsq lifecycleCoverage diff --git a/nix/microvms/mkVm.nix b/nix/microvms/mkVm.nix index 7846c33..5797376 100644 --- a/nix/microvms/mkVm.nix +++ b/nix/microvms/mkVm.nix @@ -49,9 +49,61 @@ let # destination end-to-end: xtcp2 streams jsonl records over TCP to an in-VM # ncat receiver (dual-stack), and the self-test validates the received # records + the destTCP send counter. No docker, no broker. - isTcpSink = sink == "tcp-sink"; - tcpSinkPort = 13001; - tcpSinkFile = "/tmp/xtcp2-tcp-sink.out"; + # socket-sink flavors: -sink proves the raw tcp/udp/unix/unixgram + # destination end-to-end. xtcp2 streams jsonl records over the dest to a + # dual-stack ncat receiver, which writes everything to socketSinkFile for the + # RAW_SOCKET self-test check to validate. No docker, no broker. + socketSinkScheme = + if sink == "tcp-sink" then + "tcp" + else if sink == "udp-sink" then + "udp" + else if sink == "unix-sink" then + "unix" + else if sink == "unixgram-sink" then + "unixgram" + else + ""; + isSocketSink = socketSinkScheme != ""; + socketSinkPort = 13001; # inet dests (tcp/udp) + socketSinkPath = "/run/xtcp2-sink.sock"; # unix dests + socketSinkFile = "/tmp/xtcp2-socket-sink.out"; + # -dest value per scheme. `localhost` (dual-stack) for inet dests. + socketSinkDest = + if socketSinkScheme == "tcp" then + "tcp:localhost:${toString socketSinkPort}" + else if socketSinkScheme == "udp" then + "udp:localhost:${toString socketSinkPort}" + else + "${socketSinkScheme}:${socketSinkPath}"; # unix:/path or unixgram:/path + # ncat receiver invocation per scheme (its stdout is redirected to + # socketSinkFile by the service). inet dests listen dual-stack on :: + # (bindv6only=0 also accepts IPv4-mapped); unix dests bind the socket path. + socketSinkReceiverCmd = + if socketSinkScheme == "tcp" then + "${pkgs.nmap}/bin/ncat --listen --keep-open :: ${toString socketSinkPort}" + else if socketSinkScheme == "udp" then + "${pkgs.nmap}/bin/ncat --udp --listen --keep-open :: ${toString socketSinkPort}" + else if socketSinkScheme == "unix" then + "${pkgs.nmap}/bin/ncat --unixsock --listen --keep-open ${socketSinkPath}" + else + # unixgram: ncat can't LISTEN on a unix DATAGRAM socket (`--unixsock + # --udp --listen` fails with "connect: Invalid argument"), so use socat's + # UNIX-RECV, which binds a unix datagram socket and streams received + # datagrams to stdout (-u = unidirectional recv→stdout). + "${pkgs.socat}/bin/socat -u UNIX-RECV:${socketSinkPath} -"; + # Send-side Writes counter (function, variable) per scheme. udp is labelled + # under the legacy Inetdiager/udpWrites; the rest are destXxx/Writes. + socketSinkMetricFn = + if socketSinkScheme == "udp" then + "Inetdiager" + else if socketSinkScheme == "tcp" then + "destTCP" + else if socketSinkScheme == "unix" then + "destUnix" + else + "destUnixGram"; + socketSinkMetricVar = if socketSinkScheme == "udp" then "udpWrites" else "Writes"; # minimal = the lifecycle correctness gate. Unlike soak (which shares the # basic `-dest null` args), minimal writes jsonl to a file so the self-test # can validate the daemon's serialized output content (OUTPUT_CONTENT check). @@ -199,8 +251,13 @@ let runFileOutputCheck = isMinimal; inherit fileOutputPath; # tcp-sink flavor only: validate records received over the raw TCP dest. - runRawSocketCheck = isTcpSink; - inherit tcpSinkFile; + runRawSocketCheck = isSocketSink; + inherit + socketSinkScheme + socketSinkFile + socketSinkMetricFn + socketSinkMetricVar + ; }; # Default monitor cadence for the s3parquet-long flavor. 60 s is fast @@ -1376,14 +1433,13 @@ let # the self-test run as root, so a 0600 file under /var/log is readable. fileOutputPath = "/var/log/xtcp2.jsonl"; - # tcp-sink flavor: stream jsonl records over the raw TCP destination to the - # in-VM ncat receiver. `localhost` (dual-stack) resolves to 127.0.0.1 AND - # ::1; the sink listens on [::] (dual-stack), so whichever family Go's - # resolver picks reaches it. jsonl is newline-delimited so the receiver can - # count/validate records. - xtcp2TcpSinkArgs = [ + # socket-sink flavors: stream jsonl records over the raw socket destination + # (tcp/udp/unix/unixgram) to the in-VM ncat receiver. For inet dests, + # `localhost` (dual-stack) resolves to 127.0.0.1 AND ::1 and the sink listens + # on [::], so whichever family Go's resolver picks reaches it. + xtcp2SocketSinkArgs = [ "-dest" - "tcp:localhost:${toString tcpSinkPort}" + socketSinkDest "-marshal" "jsonl" "-frequency" @@ -2014,9 +2070,10 @@ in # minimal (lifecycle) writes jsonl to a file so OUTPUT_CONTENT # can validate the daemon's serialized output. xtcp2FileArgs - else if isTcpSink then - # tcp-sink: stream jsonl over the raw TCP dest to the ncat sink. - xtcp2TcpSinkArgs + else if isSocketSink then + # socket-sink: stream jsonl over the raw tcp/udp/unix/unixgram + # dest to the ncat sink. + xtcp2SocketSinkArgs else # Soak reuses the basic args (`-dest null`, fast frequency). # The point of soak is namespace + netlink churn, not @@ -2096,21 +2153,23 @@ in }; }; - # tcp-sink flavor: a dual-stack ncat receiver that appends everything - # xtcp2 streams over the raw TCP destination to tcpSinkFile, for the - # RAW_SOCKET self-test check to count + validate. Listens on [::] - # (bindv6only=0 → also accepts IPv4-mapped) so `localhost` on the daemon - # side reaches it on either family. Ordered before xtcp2 so the dest - # dial doesn't race a missing listener (xtcp2's Restart=on-failure also - # covers the race). - systemd.services.xtcp2-tcp-sink = lib.mkIf isTcpSink { - description = "xtcp2 tcp-sink — dual-stack ncat receiver for the raw TCP dest"; + # socket-sink flavors: an ncat receiver that appends everything xtcp2 + # streams over the raw socket destination (tcp/udp/unix/unixgram) to + # socketSinkFile, for the RAW_SOCKET self-test check to count + + # validate. inet dests listen dual-stack on [::] (bindv6only=0 → also + # accepts IPv4-mapped) so `localhost` on the daemon side reaches it on + # either family. Ordered before xtcp2 so the dest dial doesn't race a + # missing listener (xtcp2's Restart=on-failure also covers the race). + systemd.services.xtcp2-socket-sink = lib.mkIf isSocketSink { + description = "xtcp2 socket-sink — ncat receiver for the raw ${socketSinkScheme} dest"; before = [ "xtcp2.service" ]; wantedBy = [ "multi-user.target" ]; serviceConfig = { Type = "simple"; - ExecStart = "${pkgs.writeShellScript "xtcp2-tcp-sink" '' - exec ${pkgs.nmap}/bin/ncat --listen --keep-open :: ${toString tcpSinkPort} > ${tcpSinkFile} + ExecStart = "${pkgs.writeShellScript "xtcp2-socket-sink" '' + # A stale unix socket path would make ncat fail to bind on restart. + rm -f ${socketSinkPath} 2>/dev/null || true + exec ${socketSinkReceiverCmd} > ${socketSinkFile} ''}"; Restart = "on-failure"; RestartSec = "1s"; diff --git a/nix/microvms/self-test.nix b/nix/microvms/self-test.nix index b6d651b..d5fb62f 100644 --- a/nix/microvms/self-test.nix +++ b/nix/microvms/self-test.nix @@ -141,11 +141,15 @@ # wouldn't exist — hence the gate. runFileOutputCheck ? false, fileOutputPath ? "/var/log/xtcp2.jsonl", - # When true (tcp-sink flavor), the RAW_SOCKET check validates the records - # xtcp2 streamed over the raw TCP destination to the in-VM ncat receiver - # (written to tcpSinkFile). + # When true (socket-sink flavors), the RAW_SOCKET check validates the records + # xtcp2 streamed over the raw socket destination (tcp/udp/unix/unixgram) to + # the in-VM ncat receiver (written to socketSinkFile), and the per-scheme + # send-side Writes counter (socketSinkMetricFn/Var). runRawSocketCheck ? false, - tcpSinkFile ? "/tmp/xtcp2-tcp-sink.out", + socketSinkScheme ? "tcp", + socketSinkFile ? "/tmp/xtcp2-socket-sink.out", + socketSinkMetricFn ? "destTCP", + socketSinkMetricVar ? "Writes", }: pkgs.writeShellApplication { @@ -480,45 +484,49 @@ pkgs.writeShellApplication { fi if [ "$check5e" -ne 0 ]; then overall_ok=0; fi - # ─── Check 5f: raw TCP destination → in-VM ncat sink ────────────────── - # (tcp-sink flavor only). xtcp2 streams jsonl records over `-dest tcp:...` - # to a dual-stack ncat receiver; validate the received records ARE - # well-formed jsonl (send + receipt) AND the destTCP write counter grew. - # tcp/udp/unix dests were Go-component-only before this. + # ─── Check 5f: raw socket destination → in-VM ncat sink ─────────────── + # (socket-sink flavors: tcp/udp/unix/unixgram). xtcp2 streams jsonl records + # over `-dest :...` to an ncat receiver; validate the received + # records carry a non-empty hostname (send + receipt) AND the per-scheme + # send counter ran. These dests were Go-component-only before this. ${lib.optionalString runRawSocketCheck '' - echo "--- check 5f: raw TCP dest → ncat sink (${tcpSinkFile}) ---" + echo "--- check 5f: raw ${socketSinkScheme} dest → ncat sink (${socketSinkFile}) ---" check5f=1 # Wait for records to arrive at the sink file. for _ in $(seq 1 20); do - if [ -s "${tcpSinkFile}" ]; then break; fi + if [ -s "${socketSinkFile}" ]; then break; fi sleep 1 done - # destTCP Writes counter (>=1 = the daemon's send side ran). Not a growth - # check: the daemon has usually already flushed its writes by the time - # this check runs, so require the counter to be non-zero, not growing. - writes=$(metric_value "xtcp_counts" 'function="destTCP"' 'variable="Writes"') - # Snapshot + validate only complete lines (drop a partial trailing line — - # ncat may be mid-write). Same wc -l / head -n idiom as OUTPUT_CONTENT. - cp "${tcpSinkFile}" /tmp/xtcp2-tcpsink.snap 2>/dev/null || true - total=$(wc -l < /tmp/xtcp2-tcpsink.snap 2>/dev/null || echo 0) + # Per-scheme Writes counter (>=1 = the daemon's send side ran). Not a + # growth check: the daemon has usually already flushed by the time this + # runs, so require non-zero, not growing. + writes=$(metric_value "xtcp_counts" 'function="${socketSinkMetricFn}"' 'variable="${socketSinkMetricVar}"') + # Snapshot, drop a possibly-partial trailing line (ncat mid-write), then + # extract JSON objects: `grep '{...}'` strips the unix STREAM dest's + # varint length prefix and is a no-op for the newline-delimited jsonl of + # tcp/udp/unixgram. jq-validate each carries a non-empty hostname. + cp "${socketSinkFile}" /tmp/xtcp2-sink.snap 2>/dev/null || true + total=$(wc -l < /tmp/xtcp2-sink.snap 2>/dev/null || echo 0) total=''${total:-0} good=0 bad=0 - while IFS= read -r line; do - [ -z "$line" ] && continue - if printf '%s\n' "$line" \ + while IFS= read -r obj; do + [ -z "$obj" ] && continue + if printf '%s\n' "$obj" \ | jq -e 'has("hostname") and (.hostname | length > 0)' >/dev/null 2>&1; then good=$((good + 1)) else bad=$((bad + 1)) fi - done < <(head -n "$total" /tmp/xtcp2-tcpsink.snap 2>/dev/null) - if [ "$good" -ge 1 ] && [ "$bad" -eq 0 ] && [ "''${writes:-0}" -ge 1 ] 2>/dev/null; then - echo "XTCP2_SELF_TEST_RAW_SOCKET_PASS (records=$good, destTCP_Writes=$writes)" + done < <(head -n "$total" /tmp/xtcp2-sink.snap 2>/dev/null | grep -aoE '\{.*\}') + # good>=1 + counter>=1 is the proof; a few `bad` are tolerated (a varint + # length byte can occasionally look like '{' on the unix stream framing). + if [ "$good" -ge 1 ] && [ "''${writes:-0}" -ge 1 ] 2>/dev/null; then + echo "XTCP2_SELF_TEST_RAW_SOCKET_PASS (scheme=${socketSinkScheme}, records=$good, malformed=$bad, ${socketSinkMetricFn}/${socketSinkMetricVar}=$writes)" check5f=0 else - echo "XTCP2_SELF_TEST_RAW_SOCKET_FAIL (records=$good, malformed=$bad, destTCP_Writes=$writes)" - head -n 3 /tmp/xtcp2-tcpsink.snap 2>/dev/null || echo "(no sink file)" + echo "XTCP2_SELF_TEST_RAW_SOCKET_FAIL (scheme=${socketSinkScheme}, records=$good, malformed=$bad, ${socketSinkMetricFn}/${socketSinkMetricVar}=$writes)" + head -c 200 /tmp/xtcp2-sink.snap 2>/dev/null || echo "(no sink file)" fi if [ "$check5f" -ne 0 ]; then overall_ok=0; fi ''}