diff --git a/.github/workflows/nodeagent-test.yml b/.github/workflows/nodeagent-test.yml index 00a29bbdc..c53b8d2a8 100644 --- a/.github/workflows/nodeagent-test.yml +++ b/.github/workflows/nodeagent-test.yml @@ -21,7 +21,7 @@ jobs: test: needs: changes if: needs.changes.outputs.relevant == 'true' - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 timeout-minutes: 20 steps: - uses: actions/checkout@v6 @@ -37,6 +37,12 @@ jobs: - name: Check formatting working-directory: components/nodeagent run: make fmt-check + - name: Verify syscall BPF bytecode + working-directory: components/nodeagent + run: | + sudo apt-get update + sudo apt-get install -y clang-18 libbpf-dev=1:1.3.0-2build2 + make verify-syscalls-bpf CLANG=clang-18 - name: Vet, test, and build working-directory: components/nodeagent run: | @@ -56,6 +62,35 @@ jobs: helm template nodeagent kubernetes/charts/opensandbox-node-agent \ --set 'config.sources={source-a,source-b}' \ > /tmp/nodeagent-non-log-sources.yaml + helm template nodeagent kubernetes/charts/opensandbox-node-agent \ + --set 'config.sources={syscalls}' \ + > /tmp/nodeagent-syscalls.yaml + grep -A8 'capabilities:' /tmp/nodeagent-syscalls.yaml | grep -q -- '- BPF' + grep -A8 'capabilities:' /tmp/nodeagent-syscalls.yaml | grep -q -- '- PERFMON' + grep -A1 'name: NODEAGENT_SYSCALL_CGROUP_ROOT' /tmp/nodeagent-syscalls.yaml \ + | grep -q 'value: /host/sys/fs/cgroup' + grep -A2 'mountPath: /host/sys/fs/cgroup' /tmp/nodeagent-syscalls.yaml | grep -q 'readOnly: true' + grep -A2 'mountPath: /sys/kernel/tracing' /tmp/nodeagent-syscalls.yaml | grep -q 'readOnly: true' + grep -q 'name: host-cgroup' /tmp/nodeagent-syscalls.yaml + grep -q 'name: host-tracing' /tmp/nodeagent-syscalls.yaml + grep -q 'path: "/sys/fs/cgroup"' /tmp/nodeagent-syscalls.yaml + grep -q 'path: "/sys/kernel/tracing"' /tmp/nodeagent-syscalls.yaml + helm template nodeagent kubernetes/charts/opensandbox-node-agent \ + --set 'config.sources={syscalls}' \ + --set containerSecurityContext=null \ + > /tmp/nodeagent-syscalls-no-secctx.yaml + grep -A8 'capabilities:' /tmp/nodeagent-syscalls-no-secctx.yaml | grep -q -- '- BPF' + grep -A8 'capabilities:' /tmp/nodeagent-syscalls-no-secctx.yaml | grep -q -- '- PERFMON' + if grep -Eq -- '- (BPF|PERFMON)' /tmp/nodeagent-default-sources.yaml; then + echo "eBPF capabilities rendered while the syscalls Source is disabled" >&2 + exit 1 + fi + for nodeagent_syscall_marker in NODEAGENT_SYSCALL_CGROUP_ROOT 'name: host-cgroup' 'name: host-tracing'; do + if grep -q "${nodeagent_syscall_marker}" /tmp/nodeagent-default-sources.yaml; then + echo "syscalls host access rendered while the Source is disabled: ${nodeagent_syscall_marker}" >&2 + exit 1 + fi + done helm template nodeagent kubernetes/charts/opensandbox-node-agent \ --set 'config.sources={container-logs,source-a}' \ > /tmp/nodeagent-mixed-sources.yaml @@ -98,6 +133,24 @@ jobs: echo "container-logs configuration failed for an unexpected reason: $nodeagent_source_schema_error" >&2 exit 1 fi + for nodeagent_syscall_host_path in cgroup tracing; do + nodeagent_source_schema_error="" + if nodeagent_source_schema_error="$(helm template nodeagent kubernetes/charts/opensandbox-node-agent \ + --set 'config.sources={syscalls}' \ + --set "hostPaths.${nodeagent_syscall_host_path}=null" 2>&1)"; then + echo "syscalls configuration without hostPaths.${nodeagent_syscall_host_path} passed schema validation" >&2 + exit 1 + fi + if ! grep -Eiq "(don't meet the specifications of the schema|values don't meet)" <<<"$nodeagent_source_schema_error" \ + || ! grep -Eiq "hostPaths.*${nodeagent_syscall_host_path}|missing property.*${nodeagent_syscall_host_path}" <<<"$nodeagent_source_schema_error"; then + echo "syscalls configuration failed for an unexpected reason: $nodeagent_source_schema_error" >&2 + exit 1 + fi + done + helm template nodeagent kubernetes/charts/opensandbox-node-agent \ + --set 'config.sources={source-a,source-b}' \ + --set hostPaths.cgroup=null \ + --set hostPaths.tracing=null >/dev/null for nodeagent_required_setting in maxLineBytes partialTimeout endedStateRetention; do nodeagent_source_schema_error="" if nodeagent_source_schema_error="$(helm template nodeagent kubernetes/charts/opensandbox-node-agent \ diff --git a/components/nodeagent/Makefile b/components/nodeagent/Makefile index 0368a7ae4..ce3325e4f 100644 --- a/components/nodeagent/Makefile +++ b/components/nodeagent/Makefile @@ -2,6 +2,8 @@ VERSION ?= $(shell git describe --tags --always --dirty 2>/dev/null || echo dev) GIT_COMMIT ?= $(shell git rev-parse HEAD 2>/dev/null || echo unknown) BUILD_TIME ?= $(shell if [ -n "$$SOURCE_DATE_EPOCH" ]; then date -u -d "@$$SOURCE_DATE_EPOCH" +"%Y-%m-%dT%H:%M:%SZ" 2>/dev/null || date -u -r "$$SOURCE_DATE_EPOCH" +"%Y-%m-%dT%H:%M:%SZ" 2>/dev/null; else date -u +"%Y-%m-%dT%H:%M:%SZ"; fi) GO ?= go +CLANG_DEFAULT := $(shell command -v clang-18 2>/dev/null || command -v clang 2>/dev/null) +CLANG ?= $(CLANG_DEFAULT) GOFMT_DEFAULT := $(shell command -v "$$($(GO) env GOROOT 2>/dev/null)/bin/gofmt" 2>/dev/null || command -v gofmt 2>/dev/null) GOFMT ?= $(GOFMT_DEFAULT) PROJECT_GOFLAGS := -trimpath -buildvcs=false @@ -42,6 +44,24 @@ build: CGO_ENABLED=0 $(GO) build $(GO_BUILD_FLAGS) -ldflags "$(GO_LDFLAGS)" -o bin/nodeagent . CGO_ENABLED=0 $(GO) build $(GO_BUILD_FLAGS) -ldflags "$(GO_LDFLAGS)" -o bin/nodeagent-oss-cleanup ./cmd/oss-cleanup +.PHONY: generate-syscalls-bpf +generate-syscalls-bpf: + @case "$$(uname -s 2>/dev/null)" in Linux) ;; *) echo "syscall BPF generation requires Linux" >&2; exit 1;; esac + @case "$$(uname -m 2>/dev/null)" in x86_64) ;; *) echo "syscall BPF generation requires Linux x86_64" >&2; exit 1;; esac + @command -v "$(CLANG)" >/dev/null 2>&1 || { echo "clang not found: $(CLANG)" >&2; exit 1; } + @"$(CLANG)" --version | head -n 1 | grep -Eq 'clang version 18\.' || { echo "syscall BPF generation requires clang 18: $(CLANG)" >&2; exit 1; } + $(CLANG) -target bpfel -O2 -g -fdebug-prefix-map="$(CURDIR)"=. \ + -I/usr/include/$$(uname -m)-linux-gnu \ + -c pkg/source/syscalls/bpf/syscalls.bpf.c \ + -o "$(or $(BPF_OUTPUT),pkg/source/syscalls/syscalls_bpfel.o)" + +.PHONY: verify-syscalls-bpf +verify-syscalls-bpf: + @set -e; output="$$(mktemp)"; trap 'rm -f "$$output"' EXIT; \ + $(MAKE) --no-print-directory generate-syscalls-bpf BPF_OUTPUT="$$output"; \ + cmp pkg/source/syscalls/syscalls_bpfel.o "$$output" \ + || { echo "pkg/source/syscalls/syscalls_bpfel.o is stale; run 'make generate-syscalls-bpf'" >&2; exit 1; } + .PHONY: check check: fmt-check vet test integration-compile build diff --git a/components/nodeagent/README.md b/components/nodeagent/README.md index ef97ebbf9..0f6eeda91 100644 --- a/components/nodeagent/README.md +++ b/components/nodeagent/README.md @@ -2,8 +2,9 @@ Node Agent runs once per Linux Kubernetes node. It merges one or more Sources into a common pipeline, preserves order within each stream, and writes sandbox -records to a file or Alibaba Cloud OSS Sink. The stock binary currently ships -the `container-logs` Source for CRI stdout/stderr. +records to a file or Alibaba Cloud OSS Sink. The stock binary ships the +`container-logs` Source for CRI stdout/stderr and the opt-in `syscalls` Source +for cgroup-scoped Linux system-call records. ## Status @@ -22,6 +23,26 @@ binary and defaults to `container-logs`. Every Source owns its StreamRef namespace and private state handle. Each emitted RecordKind has a registered storage format that defines its encoding and object layout. +The `syscalls` Source attaches one eBPF program to +`raw_syscalls/sys_enter`, filters events by sandbox-container cgroup, and emits +NDJSON. It requires Linux kernel 5.11 or newer, cgroup v2, tracefs, and the `BPF` and `PERFMON` +capabilities. The Helm chart adds these mounts and capabilities only when +`syscalls` is enabled. The Source persists active stream identity and outcome +metadata so a restart can reattach a live container or finalize a stream whose +Pod disappeared while the Agent was down. It does not persist eBPF event +payloads, so the restart interval remains an unobservable gap and is reported +as `syscall-agent-restart`. Its bounded Source data queue continues processing +lifecycle events under output backpressure and reports discarded events as +`syscall-source-backpressure`. It also attaches only after Kubernetes reports +the container ID, so its finalization marker is `incomplete` with +`syscall-attach-after-container-start` rather than claiming full-lifecycle +coverage. The filter tracks the runtime's container cgroup itself; processes +moved into delegated descendant cgroups are outside this first implementation. + +The checked-in eBPF object is generated on Linux x86-64 with clang 18 and the +Ubuntu 24.04 libbpf headers. Run `make generate-syscalls-bpf CLANG=clang-18` +in this directory after changing the BPF C source. + Every Source also receives an isolated view of the node-local sandbox Pod store. It must call `Store.Forget` after it no longer needs a terminated Pod; the Store keeps that identity until every enabled Source has released it. diff --git a/components/nodeagent/go.mod b/components/nodeagent/go.mod index 000fbc50b..30b935cef 100644 --- a/components/nodeagent/go.mod +++ b/components/nodeagent/go.mod @@ -5,6 +5,7 @@ go 1.25.0 require ( github.com/alibaba/opensandbox/internal v0.0.0 github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible + github.com/cilium/ebpf v0.16.0 github.com/fsnotify/fsnotify v1.10.1 github.com/google/uuid v1.6.0 go.etcd.io/bbolt v1.5.0 @@ -51,6 +52,7 @@ require ( go.uber.org/zap v1.27.1 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/exp v0.0.0-20230224173230-c95f2b4c22f2 // indirect golang.org/x/net v0.55.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect golang.org/x/term v0.43.0 // indirect diff --git a/components/nodeagent/go.sum b/components/nodeagent/go.sum index 20921aba2..02c812e4a 100644 --- a/components/nodeagent/go.sum +++ b/components/nodeagent/go.sum @@ -4,6 +4,8 @@ github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1x github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cilium/ebpf v0.16.0 h1:+BiEnHL6Z7lXnlGUsXQPPAE7+kenAd4ES8MQ5min0Ok= +github.com/cilium/ebpf v0.16.0/go.mod h1:L7u2Blt2jMM/vLAVgjxluxtBKlz3/GWjB0dMOEngfwE= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= @@ -27,6 +29,8 @@ github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14= github.com/go-openapi/swag v0.23.0 h1:vsEVJDUo2hPJ2tu0/Xc+4noaxyEffXNIs3cOULZ+GrE= github.com/go-openapi/swag v0.23.0/go.mod h1:esZ8ITTYEsH1V2trKHjAN8Ai7xHb8RV+YSZ577vPjgQ= +github.com/go-quicktest/qt v1.101.0 h1:O1K29Txy5P2OK0dGo59b7b0LR6wKfIhttaAhHUyn7eI= +github.com/go-quicktest/qt v1.101.0/go.mod h1:14Bz/f7NwaXPtdYEgzsx46kqSxVwTbzVZsDC26tQJow= github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= @@ -46,6 +50,10 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= +github.com/josharian/native v1.1.0 h1:uuaP0hAbW7Y4l0ZRQ6C9zfb7Mg1mbFKry/xzDAfmtLA= +github.com/josharian/native v1.1.0/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w= +github.com/jsimonetti/rtnetlink/v2 v2.0.1 h1:xda7qaHDSVOsADNouv7ukSuicKZO7GgVUCXxpaIEIlM= +github.com/jsimonetti/rtnetlink/v2 v2.0.1/go.mod h1:7MoNYNbb3UaDHtF8udiJo/RH6VsTKP1pqKLUTVCvToE= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= @@ -59,6 +67,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= +github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g= +github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw= +github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U= +github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -125,6 +137,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/exp v0.0.0-20230224173230-c95f2b4c22f2 h1:Jvc7gsqn21cJHCmAWx0LiimpP18LZmUxkT5Mp7EZ1mI= +golang.org/x/exp v0.0.0-20230224173230-c95f2b4c22f2/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= diff --git a/components/nodeagent/main.go b/components/nodeagent/main.go index d706e6d87..e39d6e979 100644 --- a/components/nodeagent/main.go +++ b/components/nodeagent/main.go @@ -39,6 +39,7 @@ import ( _ "github.com/alibaba/opensandbox/nodeagent/pkg/sink/oss" sourcegroup "github.com/alibaba/opensandbox/nodeagent/pkg/source" _ "github.com/alibaba/opensandbox/nodeagent/pkg/source/containerlogs" + _ "github.com/alibaba/opensandbox/nodeagent/pkg/source/syscalls" "github.com/alibaba/opensandbox/nodeagent/pkg/state" "github.com/alibaba/opensandbox/nodeagent/pkg/store" "k8s.io/client-go/kubernetes" diff --git a/components/nodeagent/pkg/api/types.go b/components/nodeagent/pkg/api/types.go index 915b85826..6b2dfc8ea 100644 --- a/components/nodeagent/pkg/api/types.go +++ b/components/nodeagent/pkg/api/types.go @@ -28,7 +28,9 @@ type RecordKind string const ( SourceNameContainerLogs = "container-logs" + SourceNameSyscalls = "syscalls" RecordKindContainerLog RecordKind = "container-log" + RecordKindSyscall RecordKind = "syscall" ) type Capabilities struct { diff --git a/components/nodeagent/pkg/config/config.go b/components/nodeagent/pkg/config/config.go index ada6fd334..198c16d00 100644 --- a/components/nodeagent/pkg/config/config.go +++ b/components/nodeagent/pkg/config/config.go @@ -47,6 +47,7 @@ type Config struct { Sources []string Sink string LogRoot string + SyscallCgroupRoot string StateDir string StateMaxBytes int64 FilePath string @@ -86,6 +87,7 @@ func Load() (Config, error) { Sources: parseSources(envDefault("NODEAGENT_SOURCES", api.SourceNameContainerLogs), &errs), Sink: envDefault("NODEAGENT_SINKS", SinkFile), LogRoot: envDefault("NODEAGENT_LOG_ROOT", "/var/log/pods"), + SyscallCgroupRoot: envDefault("NODEAGENT_SYSCALL_CGROUP_ROOT", "/host/sys/fs/cgroup"), StateDir: envDefault("NODEAGENT_STATE_DIR", "/var/lib/opensandbox/nodeagent"), FilePath: strings.TrimSpace(os.Getenv("NODEAGENT_FILE_PATH")), OSSEndpoint: strings.TrimSpace(os.Getenv("NODEAGENT_OSS_ENDPOINT")), @@ -154,13 +156,21 @@ func (c Config) validate() []error { errs = append(errs, errors.New("NODEAGENT_STATE_DIR must not overlap NODEAGENT_LOG_ROOT")) } } + if c.HasSource(api.SourceNameSyscalls) { + if err := validateAbsolutePath(c.SyscallCgroupRoot); err != nil { + errs = append(errs, fmt.Errorf("NODEAGENT_SYSCALL_CGROUP_ROOT: %w", err)) + } + if pathsOverlap(c.StateDir, c.SyscallCgroupRoot) { + errs = append(errs, errors.New("NODEAGENT_STATE_DIR must not overlap NODEAGENT_SYSCALL_CGROUP_ROOT")) + } + } switch c.Sink { case SinkFile: if c.FilePath != "" { if err := validateAbsolutePath(c.FilePath); err != nil { errs = append(errs, fmt.Errorf("NODEAGENT_FILE_PATH: %w", err)) } - if pathsOverlap(c.FilePath, c.StateDir) || containerLogsEnabled && pathsOverlap(c.FilePath, c.LogRoot) { + if pathsOverlap(c.FilePath, c.StateDir) || containerLogsEnabled && pathsOverlap(c.FilePath, c.LogRoot) || c.HasSource(api.SourceNameSyscalls) && pathsOverlap(c.FilePath, c.SyscallCgroupRoot) { errs = append(errs, errors.New("NODEAGENT_FILE_PATH must not overlap active state or source paths")) } if c.FileMaxTotalBytes < c.FileMaxBytes { diff --git a/components/nodeagent/pkg/config/config_test.go b/components/nodeagent/pkg/config/config_test.go index f3f3d513f..5a14ff958 100644 --- a/components/nodeagent/pkg/config/config_test.go +++ b/components/nodeagent/pkg/config/config_test.go @@ -17,6 +17,8 @@ package config import ( "strings" "testing" + + "github.com/alibaba/opensandbox/nodeagent/pkg/api" ) func TestLoadFileConfig(t *testing.T) { @@ -86,6 +88,32 @@ func TestLoadDoesNotValidateDisabledContainerLogSettings(t *testing.T) { } } +func TestLoadValidatesOnlyEnabledSyscallSettings(t *testing.T) { + t.Setenv("NODE_NAME", "node-1") + t.Setenv("NODEAGENT_CLUSTER_ID", "prod-a") + t.Setenv("NODEAGENT_SOURCES", "custom-source") + t.Setenv("NODEAGENT_STATE_DIR", t.TempDir()) + t.Setenv("NODEAGENT_SYSCALL_CGROUP_ROOT", "relative") + if _, err := Load(); err != nil { + t.Fatalf("Load() validated a disabled syscalls Source: %v", err) + } + + t.Setenv("NODEAGENT_SOURCES", api.SourceNameSyscalls) + if _, err := Load(); err == nil || !strings.Contains(err.Error(), "NODEAGENT_SYSCALL_CGROUP_ROOT") { + t.Fatalf("Load() error = %v, want syscall cgroup root error", err) + } +} + +func TestValidateRejectsFileSinkInsideSyscallCgroupRoot(t *testing.T) { + cfg := validConfig() + cfg.Sources = []string{api.SourceNameSyscalls} + cfg.SyscallCgroupRoot = "/host/sys/fs/cgroup" + cfg.FilePath = "/host/sys/fs/cgroup/output" + if err := errorsContaining(cfg.validate(), "must not overlap active state or source paths"); err == "" { + t.Fatal("validate() accepted a file sink inside the cgroup source root") + } +} + func TestLoadRejectsInvalidIdentityAndBudget(t *testing.T) { t.Setenv("NODE_NAME", "node-1") t.Setenv("NODEAGENT_CLUSTER_ID", "INVALID") diff --git a/components/nodeagent/pkg/source/syscalls/bpf/syscalls.bpf.c b/components/nodeagent/pkg/source/syscalls/bpf/syscalls.bpf.c new file mode 100644 index 000000000..7c4b2f5c1 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/bpf/syscalls.bpf.c @@ -0,0 +1,87 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +#include +#include + +char LICENSE[] SEC("license") = "GPL"; + +struct { + __uint(type, BPF_MAP_TYPE_HASH); + __uint(max_entries, 65536); + __type(key, __u64); + __type(value, __u64); +} tracked_cgroups SEC(".maps"); + +struct { + __uint(type, BPF_MAP_TYPE_RINGBUF); + __uint(max_entries, 1 << 20); +} events SEC(".maps"); + +struct { + __uint(type, BPF_MAP_TYPE_HASH); + __uint(max_entries, 65536); + __type(key, __u64); + __type(value, __u64); +} lost_events SEC(".maps"); + +struct syscall_event { + __u64 ktime_ns; + __u64 cgroup_id; + __u64 handle; + __u32 host_pid; + __u32 host_tid; + __s64 syscall_nr; + char comm[16]; +} __attribute__((packed)); + +struct syscall_enter_context { + __u64 _tracepoint_header; + long id; + unsigned long args[6]; +}; + +SEC("tracepoint/raw_syscalls/sys_enter") +int collect_syscall(struct syscall_enter_context *ctx) +{ + __u64 cgroup_id = bpf_get_current_cgroup_id(); + __u64 *handle = bpf_map_lookup_elem(&tracked_cgroups, &cgroup_id); + __u64 pid_tgid; + + if (!handle) + return 0; + + struct syscall_event event = {}; + event.ktime_ns = bpf_ktime_get_ns(); + event.cgroup_id = cgroup_id; + event.handle = *handle; + pid_tgid = bpf_get_current_pid_tgid(); + event.host_pid = pid_tgid >> 32; + event.host_tid = (__u32)pid_tgid; + event.syscall_nr = ctx->id; + bpf_get_current_comm(&event.comm, sizeof(event.comm)); + + if (bpf_ringbuf_output(&events, &event, sizeof(event), 0)) { + __u64 zero = 0; + __u64 *lost = bpf_map_lookup_elem(&lost_events, handle); + + if (!lost) { + bpf_map_update_elem(&lost_events, handle, &zero, BPF_NOEXIST); + lost = bpf_map_lookup_elem(&lost_events, handle); + } + if (lost) + __sync_fetch_and_add(lost, 1); + } + return 0; +} diff --git a/components/nodeagent/pkg/source/syscalls/cgroup.go b/components/nodeagent/pkg/source/syscalls/cgroup.go new file mode 100644 index 000000000..ad999d36a --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/cgroup.go @@ -0,0 +1,128 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "errors" + "fmt" + "io/fs" + "path/filepath" + "strings" + + "github.com/alibaba/opensandbox/nodeagent/pkg/store" +) + +type resolvedCgroup struct { + id uint64 + path string +} + +func resolveCgroups(root string, resources []store.Resource) (map[string]resolvedCgroup, error) { + return resolveCgroupsWithInode(root, resources, directoryInode) +} + +func resolveCgroupsWithInode(root string, resources []store.Resource, inode func(string) (uint64, error)) (map[string]resolvedCgroup, error) { + byLeaf := make(map[string][]store.Resource) + for _, resource := range resources { + if resource.Terminated || resource.ContainerID == "" { + continue + } + for _, leaf := range containerCgroupNames(resource.ContainerRuntime, resource.ContainerID) { + byLeaf[leaf] = append(byLeaf[leaf], resource) + } + } + if len(byLeaf) == 0 { + return map[string]resolvedCgroup{}, nil + } + resolved := make(map[string]resolvedCgroup) + invalid := make(map[string]bool) + var resolutionErrors []error + err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { + if walkErr != nil { + if errors.Is(walkErr, fs.ErrNotExist) { + return nil + } + return walkErr + } + if !entry.IsDir() || path == root { + return nil + } + candidates := byLeaf[entry.Name()] + if len(candidates) == 0 { + return nil + } + for _, resource := range candidates { + if invalid[resource.PodUID] { + continue + } + if resource.PodUID != "" && !pathContainsPodUID(path, resource.PodUID) { + continue + } + if previous, exists := resolved[resource.PodUID]; exists && previous.path != path { + delete(resolved, resource.PodUID) + invalid[resource.PodUID] = true + resolutionErrors = append(resolutionErrors, fmt.Errorf("container cgroup for Pod %s is ambiguous: %s and %s", resource.PodUID, previous.path, path)) + continue + } + id, err := inode(path) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + continue + } + invalid[resource.PodUID] = true + resolutionErrors = append(resolutionErrors, fmt.Errorf("read container cgroup %s: %w", path, err)) + continue + } + resolved[resource.PodUID] = resolvedCgroup{id: id, path: path} + } + return filepath.SkipDir + }) + if err != nil { + return nil, err + } + return resolved, errors.Join(resolutionErrors...) +} + +func containerCgroupNames(runtimeName, containerID string) []string { + if containerID == "" { + return nil + } + names := []string{containerID} + prefixes := []string{"cri-containerd-", "crio-", "docker-"} + if runtimeName != "" { + switch runtimeName { + case "containerd": + prefixes = []string{"cri-containerd-"} + case "cri-o", "crio": + prefixes = []string{"crio-"} + case "docker": + prefixes = []string{"docker-"} + } + } + for _, prefix := range prefixes { + names = append(names, prefix+containerID, prefix+containerID+".scope") + } + return names +} + +func pathContainsPodUID(path, podUID string) bool { + normalizedUID := strings.ReplaceAll(podUID, "-", "_") + for _, segment := range strings.Split(filepath.ToSlash(path), "/") { + if segment == "pod"+podUID || strings.HasSuffix(segment, "-pod"+normalizedUID+".slice") { + return true + } + } + return false +} diff --git a/components/nodeagent/pkg/source/syscalls/cgroup_test.go b/components/nodeagent/pkg/source/syscalls/cgroup_test.go new file mode 100644 index 000000000..771597822 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/cgroup_test.go @@ -0,0 +1,136 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "io/fs" + "os" + "path/filepath" + "slices" + "testing" + + "github.com/alibaba/opensandbox/nodeagent/pkg/api" + "github.com/alibaba/opensandbox/nodeagent/pkg/store" +) + +func TestContainerCgroupNamesIncludeSystemdAndCgroupfsForms(t *testing.T) { + containerID := "abcdef" + for _, want := range []string{"crio-abcdef", "crio-abcdef.scope"} { + if !slices.Contains(containerCgroupNames("cri-o", containerID), want) { + t.Fatalf("CRI-O cgroup names do not contain %q", want) + } + } + for _, want := range []string{"docker-abcdef", "docker-abcdef.scope"} { + if !slices.Contains(containerCgroupNames("docker", containerID), want) { + t.Fatalf("Docker cgroup names do not contain %q", want) + } + } +} + +func TestResolveCgroupsRequiresFullContainerIDAndPodAncestor(t *testing.T) { + root := t.TempDir() + podUID := "12345678-abcd-4321-abcd-1234567890ab" + containerID := "0123456789abcdef" + valid := filepath.Join(root, "kubepods.slice", "kubepods-burstable.slice", "kubepods-burstable-pod12345678_abcd_4321_abcd_1234567890ab.slice", "cri-containerd-"+containerID+".scope") + wrongPod := filepath.Join(root, "kubepods", "podother", containerID) + prefixOnly := filepath.Join(root, "kubepods", "pod"+podUID, containerID[:12]) + for _, path := range []string{valid, wrongPod, prefixOnly} { + if err := os.MkdirAll(path, 0o755); err != nil { + t.Fatal(err) + } + } + resource := store.Resource{Resource: api.Resource{PodUID: podUID}, ContainerRuntime: "containerd", ContainerID: containerID} + resolved, err := resolveCgroupsWithInode(root, []store.Resource{resource}, func(path string) (uint64, error) { + if path != valid { + t.Fatalf("resolved unexpected path %q", path) + } + return 42, nil + }) + if err != nil { + t.Fatal(err) + } + if got := resolved[podUID]; got.id != 42 || got.path != valid { + t.Fatalf("resolved=%+v", got) + } +} + +func TestResolveCgroupsRejectsAmbiguousCandidates(t *testing.T) { + root := t.TempDir() + podUID := "u1" + containerID := "abcdef" + for _, qos := range []string{"burstable", "besteffort"} { + path := filepath.Join(root, qos, "pod"+podUID, containerID) + if err := os.MkdirAll(path, 0o755); err != nil { + t.Fatal(err) + } + } + resource := store.Resource{Resource: api.Resource{PodUID: podUID}, ContainerID: containerID} + resolved, err := resolveCgroupsWithInode(root, []store.Resource{resource}, func(string) (uint64, error) { return 1, nil }) + if err == nil { + t.Fatal("ambiguous cgroup candidates were accepted") + } + if len(resolved) != 0 { + t.Fatalf("ambiguous cgroup was resolved: %+v", resolved) + } +} + +func TestResolveCgroupsKeepsOtherResultsAfterAmbiguousCandidate(t *testing.T) { + root := t.TempDir() + for _, path := range []string{ + filepath.Join(root, "a", "podu1", "container-1"), + filepath.Join(root, "b", "podu1", "container-1"), + filepath.Join(root, "c", "podu2", "container-2"), + } { + if err := os.MkdirAll(path, 0o755); err != nil { + t.Fatal(err) + } + } + resources := []store.Resource{ + {Resource: api.Resource{PodUID: "u1"}, ContainerID: "container-1"}, + {Resource: api.Resource{PodUID: "u2"}, ContainerID: "container-2"}, + } + resolved, err := resolveCgroupsWithInode(root, resources, func(path string) (uint64, error) { + if filepath.Base(path) == "container-2" { + return 2, nil + } + return 1, nil + }) + if err == nil { + t.Fatal("ambiguous cgroup was not reported") + } + if _, found := resolved["u1"]; found || resolved["u2"].id != 2 { + t.Fatalf("partial resolution=%+v", resolved) + } +} + +func TestResolveCgroupsIgnoresVanishedCandidate(t *testing.T) { + root := t.TempDir() + podUID := "u1" + containerID := "abcdef" + path := filepath.Join(root, "pod"+podUID, containerID) + if err := os.MkdirAll(path, 0o755); err != nil { + t.Fatal(err) + } + resource := store.Resource{Resource: api.Resource{PodUID: podUID}, ContainerID: containerID} + resolved, err := resolveCgroupsWithInode(root, []store.Resource{resource}, func(string) (uint64, error) { + return 0, fs.ErrNotExist + }) + if err != nil { + t.Fatal(err) + } + if len(resolved) != 0 { + t.Fatalf("resolved vanished candidate: %+v", resolved) + } +} diff --git a/components/nodeagent/pkg/source/syscalls/clock_linux.go b/components/nodeagent/pkg/source/syscalls/clock_linux.go new file mode 100644 index 000000000..b191a07ee --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/clock_linux.go @@ -0,0 +1,31 @@ +//go:build linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "time" + + "golang.org/x/sys/unix" +) + +func monotonicWallOffset() (int64, error) { + var monotonic unix.Timespec + if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &monotonic); err != nil { + return 0, err + } + return time.Now().UnixNano() - monotonic.Nano(), nil +} diff --git a/components/nodeagent/pkg/source/syscalls/clock_other.go b/components/nodeagent/pkg/source/syscalls/clock_other.go new file mode 100644 index 000000000..7b3074cf3 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/clock_other.go @@ -0,0 +1,23 @@ +//go:build !linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import "errors" + +func monotonicWallOffset() (int64, error) { + return 0, errors.New("monotonic syscall timestamps require Linux") +} diff --git a/components/nodeagent/pkg/source/syscalls/event.go b/components/nodeagent/pkg/source/syscalls/event.go new file mode 100644 index 000000000..fc2edb204 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/event.go @@ -0,0 +1,36 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "encoding/binary" + "fmt" + "strings" +) + +func decodeKernelEvent(raw []byte) (kernelEvent, error) { + if len(raw) != 56 { + return kernelEvent{}, fmt.Errorf("unexpected syscall event size %d", len(raw)) + } + return kernelEvent{ + MonotonicNS: binary.LittleEndian.Uint64(raw[0:8]), + CgroupID: binary.LittleEndian.Uint64(raw[8:16]), + Handle: binary.LittleEndian.Uint64(raw[16:24]), + HostPID: binary.LittleEndian.Uint32(raw[24:28]), + HostTID: binary.LittleEndian.Uint32(raw[28:32]), + SyscallNR: int64(binary.LittleEndian.Uint64(raw[32:40])), + Comm: strings.TrimRight(string(raw[40:56]), "\x00"), + }, nil +} diff --git a/components/nodeagent/pkg/source/syscalls/event_test.go b/components/nodeagent/pkg/source/syscalls/event_test.go new file mode 100644 index 000000000..ce46cdf74 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/event_test.go @@ -0,0 +1,38 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "encoding/binary" + "testing" +) + +func TestDecodeKernelEvent(t *testing.T) { + raw := make([]byte, 56) + binary.LittleEndian.PutUint64(raw[0:8], 100) + binary.LittleEndian.PutUint64(raw[8:16], 200) + binary.LittleEndian.PutUint64(raw[16:24], 300) + binary.LittleEndian.PutUint32(raw[24:28], 400) + binary.LittleEndian.PutUint32(raw[28:32], 401) + binary.LittleEndian.PutUint64(raw[32:40], 63) + copy(raw[40:56], "cat") + event, err := decodeKernelEvent(raw) + if err != nil { + t.Fatal(err) + } + if event.MonotonicNS != 100 || event.CgroupID != 200 || event.Handle != 300 || event.HostPID != 400 || event.HostTID != 401 || event.SyscallNR != 63 || event.Comm != "cat" { + t.Fatalf("event=%+v", event) + } +} diff --git a/components/nodeagent/pkg/source/syscalls/inode_linux.go b/components/nodeagent/pkg/source/syscalls/inode_linux.go new file mode 100644 index 000000000..b7a80658c --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/inode_linux.go @@ -0,0 +1,35 @@ +//go:build linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "errors" + "os" + "syscall" +) + +func directoryInode(path string) (uint64, error) { + info, err := os.Stat(path) + if err != nil { + return 0, err + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok { + return 0, errors.New("cgroup stat did not expose an inode") + } + return stat.Ino, nil +} diff --git a/components/nodeagent/pkg/source/syscalls/inode_other.go b/components/nodeagent/pkg/source/syscalls/inode_other.go new file mode 100644 index 000000000..54d485766 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/inode_other.go @@ -0,0 +1,23 @@ +//go:build !linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import "errors" + +func directoryInode(string) (uint64, error) { + return 0, errors.New("cgroup inode resolution requires Linux") +} diff --git a/components/nodeagent/pkg/source/syscalls/source.go b/components/nodeagent/pkg/source/syscalls/source.go new file mode 100644 index 000000000..9aac99bd7 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/source.go @@ -0,0 +1,597 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "time" + + "github.com/alibaba/opensandbox/internal/logger" + "github.com/alibaba/opensandbox/nodeagent/pkg/api" + "github.com/alibaba/opensandbox/nodeagent/pkg/registry" + "github.com/alibaba/opensandbox/nodeagent/pkg/store" + "github.com/alibaba/opensandbox/nodeagent/pkg/streamformat" + "github.com/google/uuid" +) + +const ( + sourceName = api.SourceNameSyscalls + reconcileInterval = 5 * time.Second + sourceQueueSize = 1024 + lossReasonOverflow = "ring-buffer-overflow" + lossReasonSourceBackpressure = "syscall-source-backpressure" + lossReasonLateAttach = "syscall-attach-after-container-start" + lossReasonRestart = "syscall-agent-restart" +) + +type kernelEvent struct { + MonotonicNS uint64 + CgroupID uint64 + Handle uint64 + HostPID uint32 + HostTID uint32 + SyscallNR int64 + Comm string +} + +type kernelMessage struct { + Event kernelEvent + Drained bool +} + +type kernelTracer interface { + Track(cgroupID, handle uint64) error + Untrack(cgroupID uint64) error + Flush() error + Forget(handle uint64) error + Messages() <-chan kernelMessage + Errors() <-chan error + Lost() (map[uint64]uint64, error) + Close() error +} + +type syscallRecord struct { + SchemaVersion int `json:"schema_version"` + Timestamp time.Time `json:"timestamp"` + MonotonicNS uint64 `json:"monotonic_ns"` + SyscallNR int64 `json:"syscall_nr"` + NodeArch string `json:"node_arch"` + HostPID uint32 `json:"host_pid"` + HostTID uint32 `json:"host_tid"` + Comm string `json:"comm"` + ContainerRestartCount int32 `json:"container_restart_count"` +} + +type streamRuntime struct { + resource store.Resource + handle uint64 + cgroupID uint64 + sequence uint64 + coverageStartedAt time.Time + outcome api.SourceOutcome + endDrain uint64 + endReady bool +} + +type streamBinding struct { + stream *streamRuntime + cgroupID uint64 + restartCount int32 + lost uint64 + drainAfter uint64 +} + +type source struct { + cgroupRoot string + store store.View + state registry.SourceState + log logger.Logger + onError func(error) + runID string + + mu sync.Mutex + cancel context.CancelFunc + done chan struct{} + out chan<- api.SourceEvent + tracer kernelTracer +} + +func init() { + registry.RegisterSource(sourceName, func(dependencies registry.SourceDependencies) (api.Source, error) { + return &source{cgroupRoot: dependencies.Config.SyscallCgroupRoot, store: dependencies.Store, state: dependencies.State, log: dependencies.Logger.Named(sourceName), onError: dependencies.OnError, runID: uuid.NewString(), done: make(chan struct{})}, nil + }) +} + +func (s *source) Capabilities() api.Capabilities { + return api.Capabilities{RecordKinds: []api.RecordKind{api.RecordKindSyscall}} +} + +func (s *source) Start(ctx context.Context, out chan<- api.SourceEvent) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.cancel != nil { + return errors.New("syscalls source already started") + } + if _, err := os.Stat(filepath.Join(s.cgroupRoot, "cgroup.controllers")); err != nil { + return fmt.Errorf("syscalls source requires a cgroup v2 root at %s: %w", s.cgroupRoot, err) + } + streams, err := s.loadStreams() + if err != nil { + return fmt.Errorf("restore syscall streams: %w", err) + } + tracer, err := newKernelTracer() + if err != nil { + return fmt.Errorf("start syscall tracer: %w", err) + } + runCtx, cancel := context.WithCancel(ctx) + s.cancel = cancel + s.out = out + s.tracer = tracer + go s.run(runCtx, streams) + return nil +} + +func (s *source) Stop(ctx context.Context) error { + s.mu.Lock() + cancel := s.cancel + s.mu.Unlock() + if cancel == nil { + return nil + } + cancel() + select { + case <-s.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (s *source) Acknowledge(_ context.Context, results []api.AckResult) error { + for _, result := range results { + if err := validateToken(result.Token.Source, result.Token.StreamRef, result.Token.Value); err != nil { + return api.Permanent(err) + } + } + return nil +} + +func (s *source) AcknowledgeEnd(_ context.Context, token api.EndToken) error { + if err := validateToken(token.Source, token.StreamRef, token.Value); err != nil { + return api.Permanent(err) + } + if err := s.deleteStream(token.StreamRef); err != nil { + return fmt.Errorf("delete finalized syscall stream state: %w", err) + } + return nil +} + +func validateToken(source string, streamRef api.StreamRef, value []byte) error { + if source != sourceName || streamRef.Kind != api.RecordKindSyscall || !strings.HasPrefix(streamRef.ID, sourceName+"/") { + return errors.New("syscalls token identity is invalid") + } + if len(value) != 8 { + return errors.New("syscalls token value is invalid") + } + return nil +} + +func (s *source) run(ctx context.Context, streams map[string]*streamRuntime) { + defer close(s.done) + defer close(s.out) + defer func() { + if err := s.tracer.Close(); err != nil && s.onError != nil { + s.onError(fmt.Errorf("close syscall tracer: %w", err)) + } + }() + + byHandle := make(map[uint64]*streamBinding) + drains := drainCoordinator{tracer: s.tracer} + pending := make([]api.SourceEvent, 0, sourceQueueSize) + var nextHandle uint64 + clockOffset, err := monotonicWallOffset() + if err != nil { + s.reportError(fmt.Errorf("read monotonic clock: %w", err)) + return + } + if err := s.reconcile(streams, byHandle, &nextHandle, &drains); err != nil { + s.reportError(err) + } + if err := s.finishTerminated(streams, &pending); err != nil { + s.reportError(err) + } + + reconcileTicker := time.NewTicker(reconcileInterval) + defer reconcileTicker.Stop() + for { + var output chan<- api.SourceEvent + var next api.SourceEvent + if len(pending) != 0 { + output = s.out + next = pending[0] + } + select { + case <-ctx.Done(): + return + case output <- next: + pending[0] = api.SourceEvent{} + if len(pending) == 1 { + pending = pending[:0] + } else { + pending = pending[1:] + } + case message, ok := <-s.tracer.Messages(): + if !ok { + s.reportError(errors.New("syscall tracer event stream closed")) + return + } + if message.Drained { + s.handleDrain(streams, byHandle, &drains, &pending) + continue + } + if err := s.enqueueEvent(&pending, byHandle[message.Event.Handle], message.Event, clockOffset); err != nil { + s.reportError(err) + } + case err, ok := <-s.tracer.Errors(): + if !ok { + s.reportError(errors.New("syscall tracer error stream closed")) + return + } + if err != nil { + s.reportError(fmt.Errorf("read syscall tracer: %w", err)) + return + } + case <-s.store.Changes(): + if err := s.reconcile(streams, byHandle, &nextHandle, &drains); err != nil { + s.reportError(err) + } + if err := s.finishTerminated(streams, &pending); err != nil { + s.reportError(err) + } + case <-reconcileTicker.C: + if err := s.reconcile(streams, byHandle, &nextHandle, &drains); err != nil { + s.reportError(err) + } + if err := s.collectLoss(byHandle); err != nil { + s.reportError(err) + } + if nextOffset, err := monotonicWallOffset(); err != nil { + s.reportError(fmt.Errorf("refresh monotonic clock: %w", err)) + } else { + clockOffset = nextOffset + } + if err := s.finishTerminated(streams, &pending); err != nil { + s.reportError(err) + } + if err := drains.start(); err != nil { + s.reportError(fmt.Errorf("flush syscall ring buffer: %w", err)) + } + } + } +} + +type drainCoordinator struct { + tracer kernelTracer + generation uint64 + inFlight uint64 + queued bool +} + +func (d *drainCoordinator) schedule() (uint64, error) { + target := d.generation + 1 + d.queued = true + if err := d.start(); err != nil { + return target, err + } + return target, nil +} + +func (d *drainCoordinator) start() error { + if d.inFlight != 0 || !d.queued { + return nil + } + if err := d.tracer.Flush(); err != nil { + return err + } + d.generation++ + d.inFlight = d.generation + d.queued = false + return nil +} + +func (d *drainCoordinator) complete() (uint64, error) { + if d.inFlight == 0 { + return 0, errors.New("unexpected syscall drain barrier") + } + completed := d.inFlight + d.inFlight = 0 + return completed, nil +} + +func (s *source) reconcile(streams map[string]*streamRuntime, byHandle map[uint64]*streamBinding, nextHandle *uint64, drains *drainCoordinator) error { + resources := s.store.List() + seen := make(map[string]bool, len(resources)) + toResolve := make([]store.Resource, 0, len(resources)) + for _, resource := range resources { + seen[resource.PodUID] = true + stream := streams[resource.PodUID] + if !resource.Terminated && resource.ContainerID != "" && (stream == nil || !stream.endReady && (stream.handle == 0 || stream.resource.ContainerID != resource.ContainerID || stream.resource.ContainerRestartCount != resource.ContainerRestartCount)) { + toResolve = append(toResolve, resource) + } + } + resolved, resolveErr := resolveCgroups(s.cgroupRoot, toResolve) + for _, resource := range resources { + stream := streams[resource.PodUID] + if stream != nil && stream.endReady { + continue + } + if resource.Terminated { + if stream == nil { + s.store.Forget(resource.PodUID) + continue + } + if err := s.prepareEnd(stream, byHandle[stream.handle], drains); err != nil { + return errors.Join(resolveErr, err) + } + continue + } + if resource.ContainerID == "" { + continue + } + if stream != nil && stream.handle != 0 && stream.resource.ContainerID == resource.ContainerID && stream.resource.ContainerRestartCount == resource.ContainerRestartCount { + stream.resource = resource + continue + } + cgroup, found := resolved[resource.PodUID] + if !found { + continue + } + cgroupID := cgroup.id + if stream == nil { + stream = &streamRuntime{ + resource: resource, + coverageStartedAt: time.Now().UTC().Truncate(time.Second), + outcome: api.SourceOutcome{ + HadSourceGaps: true, + LossReasons: []string{lossReasonLateAttach}, + }, + } + streams[resource.PodUID] = stream + if err := s.persistStream(stream); err != nil { + delete(streams, resource.PodUID) + return errors.Join(resolveErr, err) + } + } + if stream.handle == 0 { + if err := s.attach(stream, resource, cgroupID, nextHandle, byHandle); err != nil { + return errors.Join(resolveErr, err) + } + continue + } + previousContainerID := stream.resource.ContainerID + previousRestartCount := stream.resource.ContainerRestartCount + if stream.cgroupID != cgroupID || previousContainerID != resource.ContainerID || previousRestartCount != resource.ContainerRestartCount { + previousBinding := byHandle[stream.handle] + if previousBinding != nil && previousBinding.drainAfter != 0 { + continue + } + if err := s.tracer.Untrack(stream.cgroupID); err != nil { + return errors.Join(resolveErr, fmt.Errorf("untrack replaced cgroup %d: %w", stream.cgroupID, err)) + } + generation, err := drains.schedule() + if previousBinding != nil { + previousBinding.drainAfter = generation + } + if err != nil { + return errors.Join(resolveErr, fmt.Errorf("flush replaced cgroup %d: %w", stream.cgroupID, err)) + } + (*nextHandle)++ + if err := s.tracer.Track(cgroupID, *nextHandle); err != nil { + return errors.Join(resolveErr, fmt.Errorf("track replacement cgroup %d: %w", cgroupID, err)) + } + stream.handle = *nextHandle + stream.cgroupID = cgroupID + byHandle[stream.handle] = &streamBinding{stream: stream, cgroupID: cgroupID, restartCount: resource.ContainerRestartCount} + } + stream.resource = resource + } + for podUID, stream := range streams { + if seen[podUID] || stream.endReady || stream.endDrain != 0 { + continue + } + if err := s.prepareEnd(stream, byHandle[stream.handle], drains); err != nil { + return errors.Join(resolveErr, err) + } + } + return resolveErr +} + +func (s *source) attach(stream *streamRuntime, resource store.Resource, cgroupID uint64, nextHandle *uint64, byHandle map[uint64]*streamBinding) error { + (*nextHandle)++ + stream.resource = resource + stream.handle = *nextHandle + stream.cgroupID = cgroupID + if err := s.tracer.Track(cgroupID, stream.handle); err != nil { + stream.handle = 0 + stream.cgroupID = 0 + return fmt.Errorf("track cgroup %d: %w", cgroupID, err) + } + byHandle[stream.handle] = &streamBinding{stream: stream, cgroupID: cgroupID, restartCount: resource.ContainerRestartCount} + return nil +} + +func (s *source) prepareEnd(stream *streamRuntime, binding *streamBinding, drains *drainCoordinator) error { + if stream.endDrain != 0 || binding != nil && binding.drainAfter != 0 { + return nil + } + if stream.handle == 0 { + stream.endReady = true + return nil + } + if err := s.tracer.Untrack(stream.cgroupID); err != nil { + return fmt.Errorf("untrack cgroup %d before stream end: %w", stream.cgroupID, err) + } + generation, err := drains.schedule() + if binding != nil { + binding.drainAfter = generation + } + stream.endDrain = generation + if err != nil { + return fmt.Errorf("flush cgroup %d before stream end: %w", stream.cgroupID, err) + } + return nil +} + +func (s *source) enqueueEvent(pending *[]api.SourceEvent, binding *streamBinding, event kernelEvent, clockOffset int64) error { + if binding == nil || binding.cgroupID != event.CgroupID { + return nil + } + if len(*pending) >= sourceQueueSize { + stream := binding.stream + stream.outcome.HadSourceGaps = true + if contains(stream.outcome.LossReasons, lossReasonSourceBackpressure) { + return nil + } + stream.outcome.LossReasons = append(stream.outcome.LossReasons, lossReasonSourceBackpressure) + s.log.Warnf("syscall source queue is full; dropping events for sandbox %s", stream.resource.SandboxID) + if err := s.persistStream(stream); err != nil { + return err + } + return nil + } + stream := binding.stream + stream.sequence++ + timestamp := time.Unix(0, clockOffset+int64(event.MonotonicNS)).UTC() + body, err := json.Marshal(syscallRecord{SchemaVersion: 1, Timestamp: timestamp, MonotonicNS: event.MonotonicNS, SyscallNR: event.SyscallNR, NodeArch: runtime.GOARCH, HostPID: event.HostPID, HostTID: event.HostTID, Comm: event.Comm, ContainerRestartCount: binding.restartCount}) + if err != nil { + return fmt.Errorf("encode syscall record: %w", err) + } + streamRef := api.StreamRef{ID: streamformat.SyscallStreamID(stream.resource.PodUID, stream.resource.Container), Kind: api.RecordKindSyscall} + value := make([]byte, 8) + binary.LittleEndian.PutUint64(value, stream.sequence) + eventID := streamRef.ID + ":" + s.runID + ":" + strconv.FormatUint(stream.sequence, 10) + *pending = append(*pending, api.SourceEvent{Delivery: &api.Delivery{ + Record: api.Record{Kind: api.RecordKindSyscall, Timestamp: timestamp, Body: body, Resource: stream.resource.Resource}, + StreamRef: streamRef, + AckToken: api.AckToken{ID: eventID, Source: sourceName, StreamRef: streamRef, Value: value}, + RecordID: eventID, + }}) + return nil +} + +func (s *source) handleDrain(streams map[string]*streamRuntime, byHandle map[uint64]*streamBinding, drains *drainCoordinator, pending *[]api.SourceEvent) { + completed, err := drains.complete() + if err != nil { + s.reportError(err) + return + } + if err := s.collectLoss(byHandle); err != nil { + s.reportError(err) + } + for handle, binding := range byHandle { + if binding.drainAfter == 0 || binding.drainAfter > completed { + continue + } + if err := s.tracer.Forget(handle); err != nil { + s.reportError(fmt.Errorf("forget drained syscall handle %d: %w", handle, err)) + } + delete(byHandle, handle) + if binding.stream.handle == handle { + binding.stream.handle = 0 + binding.stream.cgroupID = 0 + } + } + for _, stream := range streams { + if stream.endDrain != 0 && stream.endDrain <= completed { + stream.endDrain = 0 + stream.endReady = true + } + } + if err := s.finishTerminated(streams, pending); err != nil { + s.reportError(err) + } + if err := drains.start(); err != nil { + s.reportError(fmt.Errorf("flush queued syscall ring buffer: %w", err)) + } +} + +func (s *source) collectLoss(byHandle map[uint64]*streamBinding) error { + lost, err := s.tracer.Lost() + if err != nil { + return fmt.Errorf("read syscall loss counters: %w", err) + } + for handle, total := range lost { + binding := byHandle[handle] + if binding == nil || total <= binding.lost { + continue + } + stream := binding.stream + delta := total - binding.lost + stream.outcome.HadSourceGaps = true + if !contains(stream.outcome.LossReasons, lossReasonOverflow) { + stream.outcome.LossReasons = append(stream.outcome.LossReasons, lossReasonOverflow) + } + s.log.Warnf("syscall ring buffer dropped %d events for sandbox %s", delta, stream.resource.SandboxID) + if err := s.persistStream(stream); err != nil { + return err + } + binding.lost = total + } + return nil +} + +func (s *source) finishTerminated(streams map[string]*streamRuntime, pending *[]api.SourceEvent) error { + for podUID, stream := range streams { + if !stream.endReady { + continue + } + if err := s.persistStream(stream); err != nil { + return err + } + streamRef := api.StreamRef{ID: streamformat.SyscallStreamID(stream.resource.PodUID, stream.resource.Container), Kind: api.RecordKindSyscall} + value := make([]byte, 8) + binary.LittleEndian.PutUint64(value, 1) + end := &api.StreamEnd{StreamRef: streamRef, EndToken: api.EndToken{ID: streamRef.ID + ":end:1", Source: sourceName, StreamRef: streamRef, Value: value}, Revision: 1, CoverageStartedAt: stream.coverageStartedAt, Resource: stream.resource.Resource, Outcome: stream.outcome} + *pending = append(*pending, api.SourceEvent{End: end}) + delete(streams, podUID) + s.store.Forget(podUID) + } + return nil +} + +func (s *source) reportError(err error) { + if s.onError != nil { + s.onError(err) + } +} + +func contains(values []string, value string) bool { + for _, candidate := range values { + if candidate == value { + return true + } + } + return false +} diff --git a/components/nodeagent/pkg/source/syscalls/source_test.go b/components/nodeagent/pkg/source/syscalls/source_test.go new file mode 100644 index 000000000..a6e900886 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/source_test.go @@ -0,0 +1,253 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "context" + "errors" + "os" + "path/filepath" + "runtime" + "testing" + "time" + + "github.com/alibaba/opensandbox/internal/logger" + "github.com/alibaba/opensandbox/nodeagent/pkg/api" + checkpointstate "github.com/alibaba/opensandbox/nodeagent/pkg/state" + "github.com/alibaba/opensandbox/nodeagent/pkg/store" +) + +type fakeTracer struct { + lost map[uint64]uint64 + trackErr error + trackCalls int + flushCalls int +} + +func (t *fakeTracer) Track(uint64, uint64) error { + t.trackCalls++ + return t.trackErr +} +func (*fakeTracer) Untrack(uint64) error { return nil } +func (t *fakeTracer) Flush() error { t.flushCalls++; return nil } +func (*fakeTracer) Forget(uint64) error { return nil } +func (*fakeTracer) Messages() <-chan kernelMessage { return nil } +func (*fakeTracer) Errors() <-chan error { return nil } +func (t *fakeTracer) Lost() (map[uint64]uint64, error) { return t.lost, nil } +func (*fakeTracer) Close() error { return nil } + +type fakeStoreView struct { + resources []store.Resource +} + +func (v *fakeStoreView) List() []store.Resource { return v.resources } +func (*fakeStoreView) GetByUID(string) (store.Resource, bool) { return store.Resource{}, false } +func (*fakeStoreView) Forget(string) {} +func (*fakeStoreView) Changes() <-chan struct{} { return nil } + +func testSourceState(t *testing.T) *checkpointstate.SourceState { + t.Helper() + db, err := checkpointstate.Open(t.TempDir(), "target", 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + private, err := db.SourceState(sourceName) + if err != nil { + t.Fatal(err) + } + return private +} + +func testAPIResource(podUID string) api.Resource { + return api.Resource{SandboxID: "sb-1", ClusterName: "cluster", Namespace: "ns", PodName: "pod", PodUID: podUID, NodeName: "node", Container: "sandbox"} +} + +func testStream(podUID string) *streamRuntime { + return &streamRuntime{ + resource: store.Resource{Resource: testAPIResource(podUID), ContainerID: "container-1", ContainerRuntime: "containerd"}, + coverageStartedAt: time.Unix(1, 0).UTC(), + outcome: api.SourceOutcome{HadSourceGaps: true, LossReasons: []string{lossReasonLateAttach}}, + } +} + +func TestCollectLossMarksSourceGapOnce(t *testing.T) { + tracer := &fakeTracer{lost: map[uint64]uint64{7: 3}} + source := &source{tracer: tracer, state: testSourceState(t), log: logger.MustNew(logger.Config{Level: "error"})} + stream := testStream("u1") + stream.handle = 7 + stream.outcome = api.SourceOutcome{} + binding := &streamBinding{stream: stream} + streams := map[uint64]*streamBinding{7: binding} + if err := source.collectLoss(streams); err != nil { + t.Fatal(err) + } + tracer.lost[7] = 5 + if err := source.collectLoss(streams); err != nil { + t.Fatal(err) + } + if !stream.outcome.HadSourceGaps || len(stream.outcome.LossReasons) != 1 || stream.outcome.LossReasons[0] != lossReasonOverflow || binding.lost != 5 { + t.Fatalf("outcome=%+v lost=%d", stream.outcome, binding.lost) + } +} + +func TestValidateTokenRejectsForeignIdentity(t *testing.T) { + ref := api.StreamRef{ID: "syscalls/u1/sandbox", Kind: api.RecordKindSyscall} + if err := validateToken(sourceName, ref, make([]byte, 8)); err != nil { + t.Fatal(err) + } + if err := validateToken("other", ref, make([]byte, 8)); err == nil { + t.Fatal("foreign Source token was accepted") + } +} + +func TestReconcileRetriesReplacementAfterTrackFailure(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("cgroup inode resolution requires Linux") + } + root := t.TempDir() + resource := store.Resource{ + Resource: testAPIResource("u1"), + ContainerRuntime: "containerd", + ContainerID: "new-container", + } + if err := os.MkdirAll(filepath.Join(root, "podu1", "cri-containerd-new-container.scope"), 0o755); err != nil { + t.Fatal(err) + } + tracer := &fakeTracer{trackErr: errors.New("temporary track failure")} + source := &source{cgroupRoot: root, store: &fakeStoreView{resources: []store.Resource{resource}}, state: testSourceState(t), tracer: tracer, log: logger.MustNew(logger.Config{Level: "error"})} + stream := testStream("u1") + stream.resource.ContainerID = "old-container" + stream.handle, stream.cgroupID = 1, 1 + streams := map[string]*streamRuntime{"u1": stream} + bindings := map[uint64]*streamBinding{1: {stream: stream, cgroupID: 1}} + nextHandle := uint64(1) + drains := drainCoordinator{tracer: tracer} + + if err := source.reconcile(streams, bindings, &nextHandle, &drains); err == nil { + t.Fatal("replacement track failure was not returned") + } + if stream.resource.ContainerID != "old-container" { + t.Fatalf("failed replacement committed container ID %q", stream.resource.ContainerID) + } + + pending := []api.SourceEvent{} + source.handleDrain(streams, bindings, &drains, &pending) + tracer.trackErr = nil + if err := source.reconcile(streams, bindings, &nextHandle, &drains); err != nil { + t.Fatal(err) + } + if stream.resource.ContainerID != "new-container" || tracer.trackCalls != 2 { + t.Fatalf("replacement was not retried: resource=%+v trackCalls=%d", stream.resource, tracer.trackCalls) + } +} + +func TestRecoveredMissingStreamFinalizesUntilEndAcknowledged(t *testing.T) { + private := testSourceState(t) + active := testStream("u1") + writer := &source{state: private} + if err := writer.persistStream(active); err != nil { + t.Fatal(err) + } + + recovered := &source{state: private, store: &fakeStoreView{}, tracer: &fakeTracer{}} + streams, err := recovered.loadStreams() + if err != nil { + t.Fatal(err) + } + stream := streams["u1"] + if stream == nil || !contains(stream.outcome.LossReasons, lossReasonRestart) { + t.Fatalf("recovered stream=%+v", stream) + } + drains := drainCoordinator{tracer: recovered.tracer} + if err := recovered.reconcile(streams, map[uint64]*streamBinding{}, new(uint64), &drains); err != nil { + t.Fatal(err) + } + pending := []api.SourceEvent{} + if err := recovered.finishTerminated(streams, &pending); err != nil { + t.Fatal(err) + } + if len(pending) != 1 { + t.Fatalf("pending events=%d, want 1", len(pending)) + } + event := pending[0] + if event.End == nil || !contains(event.End.Outcome.LossReasons, lossReasonRestart) { + t.Fatalf("end=%+v", event.End) + } + replayed, err := recovered.loadStreams() + if err != nil { + t.Fatal(err) + } + if got := replayed["u1"]; got == nil || !got.endReady || len(got.outcome.LossReasons) != len(event.End.Outcome.LossReasons) { + t.Fatalf("replayed stream=%+v", got) + } + if err := recovered.AcknowledgeEnd(context.Background(), event.End.EndToken); err != nil { + t.Fatal(err) + } + remaining, err := recovered.loadStreams() + if err != nil { + t.Fatal(err) + } + if len(remaining) != 0 { + t.Fatalf("streams after end ACK=%v", remaining) + } +} + +func TestTerminatedStreamEmitsQueuedEventBeforeEnd(t *testing.T) { + private := testSourceState(t) + resource := store.Resource{Resource: testAPIResource("u1"), Terminated: true, ContainerRuntime: "containerd", ContainerID: "container-1"} + tracer := &fakeTracer{} + source := &source{store: &fakeStoreView{resources: []store.Resource{resource}}, state: private, tracer: tracer, runID: "run-1", log: logger.MustNew(logger.Config{Level: "error"})} + stream := testStream("u1") + stream.resource = resource + stream.handle, stream.cgroupID = 1, 11 + if err := source.persistStream(stream); err != nil { + t.Fatal(err) + } + streams := map[string]*streamRuntime{"u1": stream} + bindings := map[uint64]*streamBinding{1: {stream: stream, cgroupID: 11}} + drains := drainCoordinator{tracer: tracer} + if err := source.reconcile(streams, bindings, new(uint64), &drains); err != nil { + t.Fatal(err) + } + if tracer.flushCalls != 1 || stream.endReady { + t.Fatalf("flushCalls=%d endReady=%t", tracer.flushCalls, stream.endReady) + } + full := make([]api.SourceEvent, sourceQueueSize) + if err := source.enqueueEvent(&full, bindings[1], kernelEvent{CgroupID: 11, Handle: 1}, 0); err != nil { + t.Fatal(err) + } + if len(full) != sourceQueueSize || !contains(stream.outcome.LossReasons, lossReasonSourceBackpressure) { + t.Fatalf("full queue=%d outcome=%+v", len(full), stream.outcome) + } + pending := []api.SourceEvent{} + if err := source.finishTerminated(streams, &pending); err != nil { + t.Fatal(err) + } + if len(pending) != 0 { + t.Fatal("stream ended before the drain barrier") + } + if err := source.enqueueEvent(&pending, bindings[1], kernelEvent{MonotonicNS: 2, CgroupID: 11, Handle: 1, SyscallNR: 1}, 0); err != nil { + t.Fatal(err) + } + source.handleDrain(streams, bindings, &drains, &pending) + if len(pending) != 2 { + t.Fatalf("pending events=%d, want 2", len(pending)) + } + first, second := pending[0], pending[1] + if first.Delivery == nil || second.End == nil { + t.Fatalf("events arrived out of order: first=%+v second=%+v", first, second) + } +} diff --git a/components/nodeagent/pkg/source/syscalls/state.go b/components/nodeagent/pkg/source/syscalls/state.go new file mode 100644 index 000000000..e654bffd3 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/state.go @@ -0,0 +1,126 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "bytes" + "encoding/json" + "fmt" + "time" + + "github.com/alibaba/opensandbox/nodeagent/pkg/api" + "github.com/alibaba/opensandbox/nodeagent/pkg/state" + "github.com/alibaba/opensandbox/nodeagent/pkg/store" + "github.com/alibaba/opensandbox/nodeagent/pkg/streamformat" +) + +const persistedStreamSchemaVersion = 1 + +var persistedStreamPrefix = []byte("stream/") + +type persistedStream struct { + SchemaVersion int `json:"schema_version"` + Resource api.Resource `json:"resource"` + CoverageStartedAt time.Time `json:"coverage_started_at"` + Outcome api.SourceOutcome `json:"outcome"` + EndEmitted bool `json:"end_emitted,omitempty"` +} + +func (s *source) loadStreams() (map[string]*streamRuntime, error) { + streams := make(map[string]*streamRuntime) + err := s.state.View(func(reader state.SourceStateReader) error { + return reader.ForEach(func(key, value []byte) error { + if !bytes.HasPrefix(key, persistedStreamPrefix) { + return fmt.Errorf("unexpected syscall state key %q", key) + } + var persisted persistedStream + if err := json.Unmarshal(value, &persisted); err != nil { + return fmt.Errorf("decode syscall stream state %q: %w", key, err) + } + if err := validatePersistedStream(key, persisted); err != nil { + return err + } + outcome := persisted.Outcome + if !persisted.EndEmitted && !contains(outcome.LossReasons, lossReasonRestart) { + outcome.HadSourceGaps = true + outcome.LossReasons = append(outcome.LossReasons, lossReasonRestart) + } + stream := &streamRuntime{ + resource: store.Resource{Resource: persisted.Resource}, + coverageStartedAt: persisted.CoverageStartedAt, + outcome: outcome, + endReady: persisted.EndEmitted, + } + streams[persisted.Resource.PodUID] = stream + return nil + }) + }) + if err != nil { + return nil, err + } + return streams, nil +} + +func (s *source) persistStream(stream *streamRuntime) error { + persisted := persistedStream{ + SchemaVersion: persistedStreamSchemaVersion, + Resource: stream.resource.Resource, + CoverageStartedAt: stream.coverageStartedAt, + Outcome: stream.outcome, + EndEmitted: stream.endReady, + } + key := persistedStreamKey(streamformat.SyscallStreamID(stream.resource.PodUID, stream.resource.Container)) + raw, err := json.Marshal(persisted) + if err != nil { + return fmt.Errorf("encode syscall stream state: %w", err) + } + if err := s.state.Update(func(writer state.SourceStateWriter) error { + return writer.Put(key, raw) + }); err != nil { + return fmt.Errorf("persist syscall stream %s: %w", stream.resource.PodUID, err) + } + return nil +} + +func (s *source) deleteStream(streamRef api.StreamRef) error { + return s.state.Update(func(writer state.SourceStateWriter) error { + return writer.Delete(persistedStreamKey(streamRef.ID)) + }) +} + +func validatePersistedStream(key []byte, persisted persistedStream) error { + if persisted.SchemaVersion != persistedStreamSchemaVersion { + return fmt.Errorf("unsupported syscall stream state version %d", persisted.SchemaVersion) + } + resource := persisted.Resource + if resource.SandboxID == "" || resource.ClusterName == "" || resource.Namespace == "" || resource.PodName == "" || resource.PodUID == "" || resource.NodeName == "" || resource.Container == "" { + return fmt.Errorf("persisted syscall stream %q has incomplete resource identity", key) + } + if persisted.CoverageStartedAt.IsZero() || persisted.CoverageStartedAt.Location() != time.UTC || persisted.CoverageStartedAt.Nanosecond() != 0 { + return fmt.Errorf("persisted syscall stream %q has invalid coverage boundary", key) + } + if !persisted.Outcome.HadSourceGaps { + return fmt.Errorf("persisted syscall stream %q must retain incomplete coverage", key) + } + wantKey := persistedStreamKey(streamformat.SyscallStreamID(resource.PodUID, resource.Container)) + if !bytes.Equal(key, wantKey) { + return fmt.Errorf("persisted syscall stream key %q does not match resource identity", key) + } + return nil +} + +func persistedStreamKey(streamID string) []byte { + return append(append([]byte(nil), persistedStreamPrefix...), streamID...) +} diff --git a/components/nodeagent/pkg/source/syscalls/syscalls_bpfel.o b/components/nodeagent/pkg/source/syscalls/syscalls_bpfel.o new file mode 100644 index 000000000..148ddebc2 Binary files /dev/null and b/components/nodeagent/pkg/source/syscalls/syscalls_bpfel.o differ diff --git a/components/nodeagent/pkg/source/syscalls/tracer_linux.go b/components/nodeagent/pkg/source/syscalls/tracer_linux.go new file mode 100644 index 000000000..9adcc70ea --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/tracer_linux.go @@ -0,0 +1,181 @@ +//go:build linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import ( + "bytes" + _ "embed" + "errors" + "fmt" + "sync" + + "github.com/cilium/ebpf" + "github.com/cilium/ebpf/link" + "github.com/cilium/ebpf/ringbuf" + "github.com/cilium/ebpf/rlimit" +) + +//go:embed syscalls_bpfel.o +var syscallBPF []byte + +type bpfObjects struct { + CollectSyscall *ebpf.Program `ebpf:"collect_syscall"` + TrackedCgroups *ebpf.Map `ebpf:"tracked_cgroups"` + Events *ebpf.Map `ebpf:"events"` + LostEvents *ebpf.Map `ebpf:"lost_events"` +} + +func (o *bpfObjects) Close() error { + var errs []error + if o.CollectSyscall != nil { + errs = append(errs, o.CollectSyscall.Close()) + } + if o.TrackedCgroups != nil { + errs = append(errs, o.TrackedCgroups.Close()) + } + if o.Events != nil { + errs = append(errs, o.Events.Close()) + } + if o.LostEvents != nil { + errs = append(errs, o.LostEvents.Close()) + } + return errors.Join(errs...) +} + +type bpfTracer struct { + objects bpfObjects + link link.Link + reader *ringbuf.Reader + messages chan kernelMessage + errors chan error + done chan struct{} + closing chan struct{} + once sync.Once +} + +func newKernelTracer() (kernelTracer, error) { + // Kernels before 5.11 charge BPF map memory against RLIMIT_MEMLOCK. + _ = rlimit.RemoveMemlock() + spec, err := ebpf.LoadCollectionSpecFromReader(bytes.NewReader(syscallBPF)) + if err != nil { + return nil, fmt.Errorf("read embedded BPF object: %w", err) + } + var objects bpfObjects + if err := spec.LoadAndAssign(&objects, nil); err != nil { + _ = objects.Close() + return nil, fmt.Errorf("load BPF maps and program: %w", err) + } + attached, err := link.Tracepoint("raw_syscalls", "sys_enter", objects.CollectSyscall, nil) + if err != nil { + _ = objects.Close() + return nil, fmt.Errorf("attach raw_syscalls/sys_enter: %w", err) + } + reader, err := ringbuf.NewReader(objects.Events) + if err != nil { + _ = attached.Close() + _ = objects.Close() + return nil, fmt.Errorf("open syscall ring buffer: %w", err) + } + tracer := &bpfTracer{objects: objects, link: attached, reader: reader, messages: make(chan kernelMessage, 1024), errors: make(chan error, 1), done: make(chan struct{}), closing: make(chan struct{})} + go tracer.read() + return tracer, nil +} + +func (t *bpfTracer) Track(cgroupID, handle uint64) error { + return t.objects.TrackedCgroups.Update(cgroupID, handle, ebpf.UpdateAny) +} + +func (t *bpfTracer) Untrack(cgroupID uint64) error { + err := t.objects.TrackedCgroups.Delete(cgroupID) + if errors.Is(err, ebpf.ErrKeyNotExist) { + return nil + } + return err +} + +func (t *bpfTracer) Flush() error { return t.reader.Flush() } + +func (t *bpfTracer) Forget(handle uint64) error { + err := t.objects.LostEvents.Delete(handle) + if errors.Is(err, ebpf.ErrKeyNotExist) { + return nil + } + return err +} + +func (t *bpfTracer) Messages() <-chan kernelMessage { return t.messages } + +func (t *bpfTracer) Errors() <-chan error { return t.errors } + +func (t *bpfTracer) Lost() (map[uint64]uint64, error) { + lost := make(map[uint64]uint64) + iterator := t.objects.LostEvents.Iterate() + var handle, count uint64 + for iterator.Next(&handle, &count) { + lost[handle] = count + } + return lost, iterator.Err() +} + +func (t *bpfTracer) Close() error { + var closeErr error + t.once.Do(func() { + close(t.closing) + closeErr = errors.Join(t.link.Close(), t.reader.Close()) + <-t.done + closeErr = errors.Join(closeErr, t.objects.Close()) + }) + return closeErr +} + +func (t *bpfTracer) read() { + defer close(t.done) + var record ringbuf.Record + for { + err := t.reader.ReadInto(&record) + if err != nil { + if errors.Is(err, ringbuf.ErrFlushed) { + select { + case t.messages <- kernelMessage{Drained: true}: + case <-t.closing: + return + } + continue + } + if !errors.Is(err, ringbuf.ErrClosed) { + select { + case t.errors <- err: + case <-t.closing: + } + } + return + } + event, err := decodeKernelEvent(record.RawSample) + if err != nil { + select { + case t.errors <- err: + case <-t.closing: + } + return + } + select { + case t.messages <- kernelMessage{Event: event}: + case <-t.closing: + return + } + } +} diff --git a/components/nodeagent/pkg/source/syscalls/tracer_other.go b/components/nodeagent/pkg/source/syscalls/tracer_other.go new file mode 100644 index 000000000..808a06258 --- /dev/null +++ b/components/nodeagent/pkg/source/syscalls/tracer_other.go @@ -0,0 +1,23 @@ +//go:build !linux + +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package syscalls + +import "errors" + +func newKernelTracer() (kernelTracer, error) { + return nil, errors.New("syscall collection requires Linux") +} diff --git a/components/nodeagent/pkg/store/store.go b/components/nodeagent/pkg/store/store.go index f260c87f8..18b510b74 100644 --- a/components/nodeagent/pkg/store/store.go +++ b/components/nodeagent/pkg/store/store.go @@ -19,6 +19,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "time" @@ -49,7 +50,10 @@ type View interface { // Kubernetes lifecycle state used only by the Store and Sources. type Resource struct { api.Resource - Terminated bool + Terminated bool + ContainerID string + ContainerRuntime string + ContainerRestartCount int32 } type Store struct { @@ -226,6 +230,7 @@ func (s *Store) upsert(obj any) { }, Terminated: terminated, } + resource.ContainerRuntime, resource.ContainerID, resource.ContainerRestartCount = containerStatus(pod, ContainerName) s.mu.Lock() if previous, exists := s.resources[resource.PodUID]; exists && previous.SandboxID != resource.SandboxID { previous.Terminated = true @@ -284,3 +289,17 @@ func hasContainer(pod *corev1.Pod, name string) bool { } return false } + +func containerStatus(pod *corev1.Pod, name string) (runtimeName, containerID string, restartCount int32) { + for _, status := range pod.Status.ContainerStatuses { + if status.Name != name { + continue + } + runtimeName, containerID, _ = strings.Cut(status.ContainerID, "://") + if containerID == "" { + runtimeName = "" + } + return runtimeName, containerID, status.RestartCount + } + return "", "", 0 +} diff --git a/components/nodeagent/pkg/store/store_test.go b/components/nodeagent/pkg/store/store_test.go index ce1d7a49b..4f7ff3ee9 100644 --- a/components/nodeagent/pkg/store/store_test.go +++ b/components/nodeagent/pkg/store/store_test.go @@ -43,6 +43,34 @@ func TestStoreFiltersAndRetainsIdentity(t *testing.T) { } } +func TestStoreExposesSandboxContainerRuntimeIdentity(t *testing.T) { + s := New(fake.NewSimpleClientset(), "node-1", "prod-a") + view, err := s.ForSource("syscalls") + if err != nil { + t.Fatal(err) + } + s.upsert(&corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "plain", Namespace: "team-a", UID: types.UID("u1"), Labels: map[string]string{SandboxIDLabel: "sb-1"}}, + Spec: corev1.PodSpec{ + NodeName: "node-1", + Containers: []corev1.Container{{Name: ContainerName}}, + }, + Status: corev1.PodStatus{ContainerStatuses: []corev1.ContainerStatus{{ + Name: ContainerName, + ContainerID: "containerd://0123456789abcdef", + RestartCount: 3, + }}}, + }) + + resource, found := view.GetByUID("u1") + if !found { + t.Fatal("sandbox resource was not stored") + } + if resource.ContainerRuntime != "containerd" || resource.ContainerID != "0123456789abcdef" || resource.ContainerRestartCount != 3 { + t.Fatalf("runtime identity=%+v", resource) + } +} + func TestStoreStaleOnlyAfterThresholdAndClearsOnRelist(t *testing.T) { s := New(fake.NewSimpleClientset(), "node-1", "prod-a") s.markWatchFailed() diff --git a/components/nodeagent/pkg/streamformat/syscall.go b/components/nodeagent/pkg/streamformat/syscall.go new file mode 100644 index 000000000..901e52bd4 --- /dev/null +++ b/components/nodeagent/pkg/streamformat/syscall.go @@ -0,0 +1,61 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package streamformat + +import ( + "bytes" + "encoding/json" + "errors" + "path" + + "github.com/alibaba/opensandbox/nodeagent/pkg/api" + "github.com/alibaba/opensandbox/nodeagent/pkg/objectlayout" +) + +type syscallFormat struct{} + +func init() { Register(syscallFormat{}) } + +func SyscallStreamID(podUID, container string) string { + return path.Join(api.SourceNameSyscalls, podUID, container) +} + +func (syscallFormat) Kind() api.RecordKind { return api.RecordKindSyscall } + +func (syscallFormat) ContentType() string { return "application/x-ndjson" } + +func (syscallFormat) EncodeBatch(batch api.Batch) ([]byte, error) { + var out bytes.Buffer + for _, item := range batch.Items { + body := item.Record.Body + if len(body) == 0 || bytes.IndexByte(body, '\n') >= 0 || !json.Valid(body) || body[0] != '{' { + return nil, errors.New("syscall record body must be one JSON object without a newline") + } + out.Write(body) + out.WriteByte('\n') + } + return out.Bytes(), nil +} + +func (syscallFormat) ObjectFamily(streamRef api.StreamRef, resource api.Resource, _ api.StreamMetadata) (objectlayout.Family, error) { + if streamRef.ID != SyscallStreamID(resource.PodUID, resource.Container) { + return objectlayout.Family{}, errors.New("syscall stream reference does not match its resource identity") + } + return objectlayout.NewFamily("", []string{resource.ClusterName, "_streams", string(api.RecordKindSyscall), resource.Namespace, resource.SandboxID, resource.PodUID}, resource.Container+".syscalls", ".jsonl") +} + +func (syscallFormat) ObjectMetadata(_ api.Resource, _ api.StreamMetadata) (map[string]string, error) { + return nil, nil +} diff --git a/components/nodeagent/pkg/streamformat/syscall_test.go b/components/nodeagent/pkg/streamformat/syscall_test.go new file mode 100644 index 000000000..7efedf1f6 --- /dev/null +++ b/components/nodeagent/pkg/streamformat/syscall_test.go @@ -0,0 +1,49 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// 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. + +package streamformat + +import ( + "testing" + + "github.com/alibaba/opensandbox/nodeagent/pkg/api" +) + +func TestSyscallFormatEncodesNDJSONAndLayout(t *testing.T) { + resource := api.Resource{ClusterName: "prod-a", Namespace: "team-a", SandboxID: "sb-1", PodUID: "u1", Container: "sandbox"} + streamRef := api.StreamRef{ID: SyscallStreamID(resource.PodUID, resource.Container), Kind: api.RecordKindSyscall} + format, _, encoded, err := EncodeBatch(api.Batch{StreamRef: streamRef, Items: []api.BatchItem{{Record: api.Record{Kind: api.RecordKindSyscall, Resource: resource, Body: []byte(`{"schema_version":1,"syscall_nr":63}`)}}}}) + if err != nil { + t.Fatal(err) + } + if got, want := string(encoded), "{\"schema_version\":1,\"syscall_nr\":63}\n"; got != want { + t.Fatalf("encoded=%q, want %q", got, want) + } + family, err := ResolveFamily(format, "", streamRef, resource, nil) + if err != nil { + t.Fatal(err) + } + if got, want := family.DataKey(0), "prod-a/_streams/syscall/team-a/sb-1/u1/sandbox.syscalls.jsonl"; got != want { + t.Fatalf("data key=%q, want %q", got, want) + } +} + +func TestSyscallFormatRejectsNonObjectBody(t *testing.T) { + resource := api.Resource{ClusterName: "prod-a", Namespace: "team-a", SandboxID: "sb-1", PodUID: "u1", Container: "sandbox"} + streamRef := api.StreamRef{ID: SyscallStreamID(resource.PodUID, resource.Container), Kind: api.RecordKindSyscall} + _, _, _, err := EncodeBatch(api.Batch{StreamRef: streamRef, Items: []api.BatchItem{{Record: api.Record{Kind: api.RecordKindSyscall, Resource: resource, Body: []byte(`[1]`)}}}}) + if err == nil { + t.Fatal("non-object syscall body was accepted") + } +} diff --git a/kubernetes/charts/opensandbox-node-agent/README.md b/kubernetes/charts/opensandbox-node-agent/README.md index 2656fba44..68b88759c 100644 --- a/kubernetes/charts/opensandbox-node-agent/README.md +++ b/kubernetes/charts/opensandbox-node-agent/README.md @@ -2,8 +2,13 @@ This chart deploys one Node Agent per Linux node. It runs one or more Sources compiled into the image and sends sandbox records to one configured file or -Alibaba Cloud OSS sink. The published image currently includes the -`container-logs` Source for CRI stdout/stderr from non-pooled OpenSandbox Pods. +Alibaba Cloud OSS sink. The published image includes the `container-logs` +Source for CRI stdout/stderr and the opt-in `syscalls` Source for cgroup-scoped +Linux system-call records from non-pooled OpenSandbox Pods. + +The `syscalls` Source requires Linux kernel 5.11 or newer, cgroup v2, and a +tracefs mount. It tracks the runtime's container cgroup; delegated descendant +cgroups are not covered by this first implementation. The chart is disabled by default when used through the umbrella OpenSandbox chart. For OSS, create a Secret containing `access-key-id`, @@ -41,10 +46,12 @@ The following table lists the configurable parameters of the chart and their def | enabled | bool | `true` | Whether the node-agent is enabled (used by the umbrella opensandbox chart). | | extraEnv | list | `[]` | Additional environment variables for the node agent container. | | fullnameOverride | string | `""` | Override the full name of the chart. | -| hostPaths | object | `{"fileData":"/var/lib/opensandbox/nodeagent-data","logs":"/var/log/pods","state":"/var/lib/opensandbox/nodeagent"}` | Host paths available to the node agent; enabled Sources and Sinks select mounts. | +| hostPaths | object | `{"cgroup":"/sys/fs/cgroup","fileData":"/var/lib/opensandbox/nodeagent-data","logs":"/var/log/pods","state":"/var/lib/opensandbox/nodeagent","tracing":"/sys/kernel/tracing"}` | Host paths available to the node agent; enabled Sources and Sinks select mounts. | +| hostPaths.cgroup | string | `"/sys/fs/cgroup"` | Host cgroup v2 hierarchy used only by the syscalls Source. | | hostPaths.fileData | string | `"/var/lib/opensandbox/nodeagent-data"` | Host path for file sink data (must match sink.file.path). | | hostPaths.logs | string | `"/var/log/pods"` | Pod-log host path used only by the container-logs Source. | | hostPaths.state | string | `"/var/lib/opensandbox/nodeagent"` | Host path for checkpoint state (must match config.stateDir). | +| hostPaths.tracing | string | `"/sys/kernel/tracing"` | Host tracefs used only by the syscalls Source. | | image | object | `{"pullPolicy":"IfNotPresent","repository":"sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/nodeagent","tag":""}` | Node agent image configuration. | | image.pullPolicy | string | `"IfNotPresent"` | Image pull policy. | | image.repository | string | `"sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/nodeagent"` | Node agent image repository. | diff --git a/kubernetes/charts/opensandbox-node-agent/README.md.gotmpl b/kubernetes/charts/opensandbox-node-agent/README.md.gotmpl index 1ab6b6311..a22ac9301 100644 --- a/kubernetes/charts/opensandbox-node-agent/README.md.gotmpl +++ b/kubernetes/charts/opensandbox-node-agent/README.md.gotmpl @@ -2,8 +2,13 @@ This chart deploys one Node Agent per Linux node. It runs one or more Sources compiled into the image and sends sandbox records to one configured file or -Alibaba Cloud OSS sink. The published image currently includes the -`container-logs` Source for CRI stdout/stderr from non-pooled OpenSandbox Pods. +Alibaba Cloud OSS sink. The published image includes the `container-logs` +Source for CRI stdout/stderr and the opt-in `syscalls` Source for cgroup-scoped +Linux system-call records from non-pooled OpenSandbox Pods. + +The `syscalls` Source requires Linux kernel 5.11 or newer, cgroup v2, and a +tracefs mount. It tracks the runtime's container cgroup; delegated descendant +cgroups are not covered by this first implementation. The chart is disabled by default when used through the umbrella OpenSandbox chart. For OSS, create a Secret containing `access-key-id`, diff --git a/kubernetes/charts/opensandbox-node-agent/templates/daemonset.yaml b/kubernetes/charts/opensandbox-node-agent/templates/daemonset.yaml index cbe7a1768..017940a7a 100644 --- a/kubernetes/charts/opensandbox-node-agent/templates/daemonset.yaml +++ b/kubernetes/charts/opensandbox-node-agent/templates/daemonset.yaml @@ -2,6 +2,13 @@ {{- $nodeSelector := mergeOverwrite (deepCopy (default dict .Values.nodeSelector)) (dict "kubernetes.io/os" "linux") }} {{- $sources := .Values.config.sources }} {{- $containerLogsEnabled := has "container-logs" $sources }} +{{- $syscallsEnabled := has "syscalls" $sources }} +{{- $containerSecurityContext := deepCopy (default dict .Values.containerSecurityContext) }} +{{- if $syscallsEnabled }} +{{- $capabilities := deepCopy (default dict $containerSecurityContext.capabilities) }} +{{- $_ := set $capabilities "add" (uniq (concat (default list $capabilities.add) (list "BPF" "PERFMON"))) }} +{{- $_ := set $containerSecurityContext "capabilities" $capabilities }} +{{- end }} apiVersion: apps/v1 kind: DaemonSet metadata: @@ -40,7 +47,7 @@ spec: image: {{ include "opensandbox-node-agent.image" . }} imagePullPolicy: {{ .Values.image.pullPolicy }} securityContext: - {{- toYaml .Values.containerSecurityContext | nindent 12 }} + {{- toYaml $containerSecurityContext | nindent 12 }} env: - name: NODE_NAME valueFrom: @@ -80,6 +87,10 @@ spec: - name: NODEAGENT_ENDED_STATE_RETENTION value: {{ .Values.config.endedStateRetention | quote }} {{- end }} + {{- if $syscallsEnabled }} + - name: NODEAGENT_SYSCALL_CGROUP_ROOT + value: /host/sys/fs/cgroup + {{- end }} {{- if .Values.config.pprofAddr }} - name: NODEAGENT_PPROF_ADDR value: {{ .Values.config.pprofAddr | quote }} @@ -141,6 +152,14 @@ spec: mountPath: {{ required "hostPaths.logs is required when the container-logs Source is enabled" .Values.hostPaths.logs | quote }} readOnly: true {{- end }} + {{- if $syscallsEnabled }} + - name: host-cgroup + mountPath: /host/sys/fs/cgroup + readOnly: true + - name: host-tracing + mountPath: /sys/kernel/tracing + readOnly: true + {{- end }} - name: state mountPath: {{ .Values.config.stateDir | quote }} {{- if and (eq .Values.sink.type "file") .Values.sink.file.path }} @@ -154,6 +173,16 @@ spec: path: {{ required "hostPaths.logs is required when the container-logs Source is enabled" .Values.hostPaths.logs | quote }} type: Directory {{- end }} + {{- if $syscallsEnabled }} + - name: host-cgroup + hostPath: + path: {{ required "hostPaths.cgroup is required when the syscalls Source is enabled" .Values.hostPaths.cgroup | quote }} + type: Directory + - name: host-tracing + hostPath: + path: {{ required "hostPaths.tracing is required when the syscalls Source is enabled" .Values.hostPaths.tracing | quote }} + type: Directory + {{- end }} - name: state hostPath: path: {{ required "hostPaths.state is required" .Values.hostPaths.state | quote }} diff --git a/kubernetes/charts/opensandbox-node-agent/values.schema.json b/kubernetes/charts/opensandbox-node-agent/values.schema.json index 8043cb55e..0f45146ef 100644 --- a/kubernetes/charts/opensandbox-node-agent/values.schema.json +++ b/kubernetes/charts/opensandbox-node-agent/values.schema.json @@ -17,7 +17,9 @@ "type": "object", "required": ["state", "fileData"], "properties": { + "cgroup": { "$ref": "#/definitions/absolutePath" }, "logs": { "$ref": "#/definitions/absolutePath" }, + "tracing": { "$ref": "#/definitions/absolutePath" }, "state": { "$ref": "#/definitions/absolutePath" }, "fileData": { "$ref": "#/definitions/absolutePath" } } @@ -101,6 +103,22 @@ } } }, + { + "if": { + "required": ["config"], + "properties": { + "config": { + "required": ["sources"], + "properties": { "sources": { "contains": { "const": "syscalls" } } } + } + } + }, + "then": { + "properties": { + "hostPaths": { "required": ["cgroup", "tracing"] } + } + } + }, { "if": { "required": ["sink"], "properties": { "sink": { "required": ["type"], "properties": { "type": { "const": "oss" } } } } }, "then": { diff --git a/kubernetes/charts/opensandbox-node-agent/values.yaml b/kubernetes/charts/opensandbox-node-agent/values.yaml index c9e59abbc..5d940d737 100644 --- a/kubernetes/charts/opensandbox-node-agent/values.yaml +++ b/kubernetes/charts/opensandbox-node-agent/values.yaml @@ -91,6 +91,10 @@ sink: hostPaths: # -- Pod-log host path used only by the container-logs Source. logs: /var/log/pods + # -- Host cgroup v2 hierarchy used only by the syscalls Source. + cgroup: /sys/fs/cgroup + # -- Host tracefs used only by the syscalls Source. + tracing: /sys/kernel/tracing # -- Host path for checkpoint state (must match config.stateDir). state: /var/lib/opensandbox/nodeagent # -- Host path for file sink data (must match sink.file.path).