You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/03/04 03:00:01 UTC
(skywalking-banyandb) branch main updated: feat(fodc): Add eBPF-based KTM I/O monitor to FODC agent (#919)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 7568a326b feat(fodc): Add eBPF-based KTM I/O monitor to FODC agent (#919)
7568a326b is described below
commit 7568a326bb7b10b6aa804bf0f4239904c347d9d5
Author: Lihan Zhou <13...@users.noreply.github.com>
AuthorDate: Tue Mar 3 18:59:55 2026 -0800
feat(fodc): Add eBPF-based KTM I/O monitor to FODC agent (#919)
---
.github/workflows/ci.yml | 45 +-
.github/workflows/test-fodc-e2e.yml | 133 ++++++
.gitignore | 8 +
CHANGES.md | 1 +
dist/LICENSE | 2 +
.../license-github.com-SkyAPM-ktm-ebpf.txt | 201 ++++++++
dist/licenses/license-github.com-cilium-ebpf.txt | 23 +
docs/design/ktm.md | 72 +--
docs/operation/fodc/ktm_metrics.md | 5 +-
fodc/agent/Makefile | 5 +-
fodc/agent/internal/cmd/root.go | 165 +++++--
.../integration/basic_metrics_buffering_test.go | 2 +-
.../internal/integration/buffer_overflow_test.go | 2 +-
.../internal/integration/metrics_export_test.go | 218 ++++++++-
fodc/agent/internal/ktm/config.go | 41 ++
fodc/agent/internal/ktm/iomonitor/collector.go | 213 +++++++++
.../ktm/iomonitor/ebpf/cgroup_discovery.go | 72 +++
.../internal/ktm/iomonitor/ebpf/cgroup_filter.go | 39 ++
fodc/agent/internal/ktm/iomonitor/ebpf/filter.go | 224 +++++++++
.../internal/ktm/iomonitor/ebpf/kernel_compat.go | 230 +++++++++
fodc/agent/internal/ktm/iomonitor/ebpf/loader.go | 333 +++++++++++++
fodc/agent/internal/ktm/iomonitor/module_linux.go | 435 +++++++++++++++++
fodc/agent/internal/ktm/iomonitor/module_other.go | 59 +++
fodc/agent/internal/ktm/ktm.go | 105 ++++
fodc/agent/internal/server/server.go | 4 +-
fodc/agent/internal/server/server_test.go | 11 +-
fodc/proxy/internal/api/server.go | 85 +++-
fodc/proxy/internal/api/server_test.go | 531 ++++++++++++++++++++-
go.mod | 2 +
go.sum | 14 +
pkg/test/flags/flags.go | 14 +
fodc/agent/Makefile => scripts/build/bpf2go.mk | 36 +-
fodc/agent/Makefile => scripts/build/bpftool.mk | 32 +-
scripts/build/version.mk | 4 +
test/fodc/README.md | 99 ++++
test/fodc/check.sh | 151 ++++++
test/fodc/kind.yaml | 19 +
test/fodc/pod-sysadmin.yaml | 79 +++
test/fodc/pod.yaml | 78 +++
test/fodc/write-load.sh | 75 +++
40 files changed, 3739 insertions(+), 128 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 64635debe..fbe7e865a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -74,6 +74,43 @@ jobs:
run: make license-dep
- name: Check
run: make check
+ detect_fodc_e2e_changes:
+ name: Detect FODC E2E Changes
+ runs-on: ubuntu-latest
+ needs: [check]
+ outputs:
+ should_run: ${{ steps.filter.outputs.should_run }}
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+ with:
+ fetch-depth: 0
+ - name: Detect changed files for FODC E2E
+ id: filter
+ run: |
+ set -euo pipefail
+
+ should_run=true
+ if [ "${{ github.event_name }}" = "pull_request" ]; then
+ should_run=false
+ base_sha="${{ github.event.pull_request.base.sha }}"
+ head_sha="${{ github.event.pull_request.head.sha }}"
+ changed_files="$(git diff --name-only "${base_sha}" "${head_sha}")"
+ while IFS= read -r changed_file; do
+ case "${changed_file}" in
+ fodc/*|test/fodc/*|banyand/*|api/proto/*|scripts/build/*)
+ should_run=true
+ break
+ ;;
+ .github/actions/setup-build-env/*|.github/workflows/test-fodc-e2e.yml|.github/workflows/prepare.yml|.github/workflows/ci.yml)
+ should_run=true
+ break
+ ;;
+ esac
+ done <<< "${changed_files}"
+ fi
+
+ echo "should_run=${should_run}" >> "${GITHUB_OUTPUT}"
test-banyand:
name: Test Banyand
needs: [check]
@@ -86,6 +123,11 @@ jobs:
name: Test FODC
needs: [check]
uses: ./.github/workflows/test-fodc.yml
+ test-fodc-e2e:
+ name: Test FODC E2E
+ if: ${{ github.event_name != 'pull_request' || needs.detect_fodc_e2e_changes.outputs.should_run == 'true' }}
+ needs: [check, detect_fodc_e2e_changes]
+ uses: ./.github/workflows/test-fodc-e2e.yml
test-pkg:
name: Test Pkg
needs: [check]
@@ -106,6 +148,7 @@ jobs:
result:
name: Continuous Integration
runs-on: ubuntu-24.04
- needs: [test-banyand, test-bydbctl, test-fodc, test-pkg, test-integration-standalone, test-integration-distributed, e2e]
+ needs:
+ [detect_fodc_e2e_changes, test-banyand, test-bydbctl, test-fodc, test-fodc-e2e, test-pkg, test-integration-standalone, test-integration-distributed, e2e]
steps:
- run: echo 'success'
diff --git a/.github/workflows/test-fodc-e2e.yml b/.github/workflows/test-fodc-e2e.yml
new file mode 100644
index 000000000..99d04c8c6
--- /dev/null
+++ b/.github/workflows/test-fodc-e2e.yml
@@ -0,0 +1,133 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+name: FODC E2E (Kind)
+
+on:
+ workflow_call:
+ inputs:
+ enable_sys_admin_fallback:
+ required: false
+ type: boolean
+ default: false
+ workflow_dispatch:
+ inputs:
+ enable_sys_admin_fallback:
+ description: Enable fallback to pod-sysadmin.yaml when strict check returns exit code 2.
+ required: false
+ type: boolean
+ default: false
+
+jobs:
+ fodc-kind:
+ runs-on: ubuntu-latest
+ timeout-minutes: 60
+ continue-on-error: true
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+ with:
+ fetch-tags: true
+
+ - name: Setup build environment
+ uses: ./.github/actions/setup-build-env
+ with:
+ download-artifacts: 'true'
+ setup-docker: 'true'
+
+ - name: Setup kubectl
+ uses: azure/setup-kubectl@v4
+ with:
+ version: v1.30.0
+
+ - name: Install kind
+ run: |
+ mkdir -p "${HOME}/.local/bin"
+ curl -fsSL -o "${RUNNER_TEMP}/kind" "https://kind.sigs.k8s.io/dl/v0.23.0/kind-linux-amd64"
+ install -m 0755 "${RUNNER_TEMP}/kind" "${HOME}/.local/bin/kind"
+ echo "${HOME}/.local/bin" >> "$GITHUB_PATH"
+
+ - name: Create Kind cluster
+ run: kind create cluster --name banyand-ktm --config test/fodc/kind.yaml --wait 180s
+
+ - name: Build BanyanDB image
+ run: |
+ make -C banyand release
+ make -C banyand docker
+
+ - name: Build FODC agent image
+ run: |
+ make -C fodc/agent release
+ make -C fodc/agent docker
+
+ - name: Load images into Kind
+ run: |
+ kind load docker-image apache/skywalking-banyandb:latest --name banyand-ktm
+ kind load docker-image apache/skywalking-banyandb-fodc-agent:latest --name banyand-ktm
+
+ - name: Deploy FODC pod
+ run: kubectl apply -f test/fodc/pod.yaml
+
+ - name: Start write load
+ run: |
+ chmod +x test/fodc/write-load.sh
+ nohup ./test/fodc/write-load.sh > /tmp/write-load.log 2>&1 &
+ echo $! > /tmp/write-load.pid
+ echo "write-load PID: $(cat /tmp/write-load.pid)"
+
+ - name: Verify KTM metrics
+ env:
+ ENABLE_SYS_ADMIN_FALLBACK: ${{ inputs.enable_sys_admin_fallback && 'true' || 'false' }}
+ run: |
+ set -euo pipefail
+ export PRINT_KTM_SAMPLES_ON_SUCCESS=true
+ export MAX_KTM_SAMPLE_LINES=40
+
+ set +e
+ ./test/fodc/check.sh
+ status=$?
+ set -e
+
+ if [ "$status" -eq 2 ]; then
+ if [ "${ENABLE_SYS_ADMIN_FALLBACK}" = "true" ]; then
+ echo "Strict-mode mismatch (exit 2), retrying with CAP_SYS_ADMIN fallback..."
+ # Kill the old write-load before redeploying
+ kill "$(cat /tmp/write-load.pid)" 2>/dev/null || true
+ kubectl delete -f test/fodc/pod.yaml
+ kubectl apply -f test/fodc/pod-sysadmin.yaml
+ nohup ./test/fodc/write-load.sh > /tmp/write-load.log 2>&1 &
+ echo $! > /tmp/write-load.pid
+ ./test/fodc/check.sh
+ else
+ echo "Strict-mode mismatch (exit 2), CAP_SYS_ADMIN fallback is disabled."
+ exit "$status"
+ fi
+ elif [ "$status" -ne 0 ]; then
+ echo "First attempt failed (exit $status), not retrying with CAP_SYS_ADMIN."
+ exit "$status"
+ fi
+
+ - name: Stop write load
+ if: always()
+ run: |
+ if [ -f /tmp/write-load.pid ]; then
+ kill "$(cat /tmp/write-load.pid)" 2>/dev/null || true
+ fi
+ if [ -f /tmp/write-load.log ]; then
+ echo "::group::write-load output"
+ cat /tmp/write-load.log
+ echo "::endgroup::"
+ fi
diff --git a/.gitignore b/.gitignore
index 84d7020e1..e433fb61c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -75,4 +75,12 @@ gomock_reflect*
.cursor/
# Claude
+Claude.md
.claude/
+CLAUDE.md
+
+# eBPF generated files and binaries
+fodc/agent/internal/ktm/iomonitor/ebpf/generated/vmlinux.h
+
+# Codex
+AGENTS.md
diff --git a/CHANGES.md b/CHANGES.md
index 35ca8ce87..48c7daebc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -24,6 +24,7 @@ Release Notes.
- To: `<data-dir>/property/data/<group>/shard-<id>/...`
- Add a generic snapshot coordination package for atomic snapshot transitions across trace and sidx.
- Support map-reduce aggregation for measure queries: map phase (partial aggregation on data nodes) and reduce phase (final aggregation on liaison).
+- Add eBPF-based KTM I/O monitor for FODC agent.
### Bug Fixes
diff --git a/dist/LICENSE b/dist/LICENSE
index 658600aa7..5223a2c54 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -191,6 +191,7 @@ Apache-2.0 licenses
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 Apache-2.0
github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01 Apache-2.0
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3 Apache-2.0
+ github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff Apache-2.0
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe Apache-2.0
github.com/aws/aws-sdk-go-v2 v1.40.0 Apache-2.0
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.3 Apache-2.0
@@ -390,6 +391,7 @@ MIT licenses
github.com/cenkalti/backoff/v4 v4.3.0 MIT
github.com/cenkalti/backoff/v5 v5.0.3 MIT
github.com/cespare/xxhash/v2 v2.3.0 MIT
+ github.com/cilium/ebpf v0.19.0 MIT
github.com/cpuguy83/go-md2man/v2 v2.0.7 MIT
github.com/dgryski/go-metro v0.0.0-20250106013310-edb8663e5e33 MIT
github.com/dustin/go-humanize v1.0.1 MIT
diff --git a/dist/licenses/license-github.com-SkyAPM-ktm-ebpf.txt b/dist/licenses/license-github.com-SkyAPM-ktm-ebpf.txt
new file mode 100644
index 000000000..261eeb9e9
--- /dev/null
+++ b/dist/licenses/license-github.com-SkyAPM-ktm-ebpf.txt
@@ -0,0 +1,201 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ 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.
diff --git a/dist/licenses/license-github.com-cilium-ebpf.txt b/dist/licenses/license-github.com-cilium-ebpf.txt
new file mode 100644
index 000000000..c637ae99c
--- /dev/null
+++ b/dist/licenses/license-github.com-cilium-ebpf.txt
@@ -0,0 +1,23 @@
+MIT License
+
+Copyright (c) 2017 Nathan Sweet
+Copyright (c) 2018, 2019 Cloudflare
+Copyright (c) 2019 Authors of Cilium
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/docs/design/ktm.md b/docs/design/ktm.md
index 047c6de4f..4e083b26b 100644
--- a/docs/design/ktm.md
+++ b/docs/design/ktm.md
@@ -3,7 +3,7 @@
## Overview
-Kernel Telemetry Module (KTM) is an optional, modular kernel observability component embedded inside the BanyanDB First Occurrence Data Collection (FODC) sidecar. The first built-in module is an eBPF-based I/O monitor ("iomonitor") that focuses on page cache behavior, fadvise() effectiveness, and memory pressure signals and their impact on BanyanDB performance. KTM is not a standalone agent or network-facing service; it runs as a sub-component of the FODC sidecar ("black box") and expose [...]
+Kernel Telemetry Module (KTM) is an optional, modular kernel observability component embedded inside the BanyanDB First Occurrence Data Collection (FODC) sidecar. The first built-in module is an eBPF-based I/O monitor ("iomonitor") that focuses on page cache behavior, fadvise() effectiveness, and memory pressure signals and their impact on BanyanDB performance. KTM is not a standalone agent or network-facing service; it runs as a sub-component of the FODC sidecar ("black box") and expose [...]
## Architecture
@@ -50,7 +50,7 @@ Notes:
- Focus: page cache add/delete, fadvise() calls, I/O counters, and memory reclaim signals.
- Attachment points: stable tracepoints where possible; fentry/fexit preferred on newer kernels.
- Data path: kernel events -> BPF maps (monotonic counters) -> userspace collector -> exporters.
-- Scoping: Fixed to the single, co-located BanyanDB process within the same container/pod, using cgroup membership first and a configurable comm-prefix fallback (default `banyand`).
+- Scoping: Fixed to the single, co-located BanyanDB process within the same container/pod, using cgroup membership first and a hardcoded comm fallback (`banyand`).
## Metrics Model and Collection Strategy
@@ -64,9 +64,7 @@ Notes:
Configuration surface (current):
- `collector.interval`: Controls the periodic push interval for metrics to Flight Recorder. Defaults to 10s.
-- `collector.ebpf.cgroup_path` (optional): absolute or `/sys/fs/cgroup`-relative path to the BanyanDB cgroup v2; if unset, KTM autodetects by scanning `/proc/*/comm` for the configured prefix (default `banyand`).
-- `collector.ebpf.comm_prefix` (optional): comm name prefix used for process discovery and fallback filtering; defaults to `banyand`.
-- Target discovery heuristic: match `/proc/<pid>/comm` against the configured prefix (default `banyand`) to locate BanyanDB and derive its cgroup; this also serves as the runtime fallback if the cgroup filter is unset.
+- `collector.ebpf.cgroup_path` (optional): absolute or `/sys/fs/cgroup`-relative path to the BanyanDB cgroup v2. If not explicitly configured, KTM attempts to auto-detect the target cgroup. Only when auto-detection fails does it fall back to comm-only filtering (degraded mode with cross-pod risk).
- Cleanup strategy is monotonic counters only; downstream derives rates. KTM does not clear BPF maps during collection.
- Configuration is applied via the FODC sidecar; KTM does not define its own standalone process-level configuration surface.
@@ -74,39 +72,41 @@ Configuration surface (current):
## Scoping and Filtering
- Scoping is not optional; KTM is designed exclusively to monitor the single BanyanDB process it is co-located with in a sidecar deployment.
-- The target container is identified at startup; eBPF programs filter events by cgroup membership first. If the cgroup filter is absent or misses, a configurable comm-prefix match (default `banyand`) is used as a narrow fallback.
+- **Kernel-side filtering**: eBPF programs always enforce `comm="banyand"` in addition to any cgroup filtering. This is hardcoded and not configurable to ensure only BanyanDB processes are monitored.
+- **Filtering strategy**:
+ - Preferred mode (strict): cgroup membership + comm="banyand" (both enforced)
+ - Degraded mode (comm-only): comm="banyand" only (when cgroup unavailable)
+ - **Warning**: comm-only mode may mix metrics across pods if multiple processes match the comm name.
- The design intentionally avoids multi-process or node-level (DaemonSet) monitoring to keep the implementation simple and overhead minimal.
### Target Process Discovery (Pod / VM)
-KTM needs to resolve the single “target” BanyanDB process before enabling filters and attaching eBPF programs. In both Kubernetes pods and VM/bare-metal deployments, KTM uses a **process matcher** based on a configurable comm-prefix match (default `banyand`).
+KTM resolves the target cgroup before enabling filters and attaching eBPF programs. If `cgroup_path` is not provided, it derives a Pod-level cgroup from `/proc/self/cgroup` (and falls back to the self cgroup for non-Kubernetes environments). Kernel-side filtering always uses `comm="banyand"` and is not configurable.
#### Kubernetes Pod (sidecar)
Preconditions:
-- The pod must be configured with `shareProcessNamespace: true` so the monitor sidecar can see the target container’s `/proc` entries.
- cgroup v2 mounted (typically at `/sys/fs/cgroup`) to enable the primary cgroup filter.
-- If the target process cannot be discovered (for example, `shareProcessNamespace` is off), KTM logs a warning and keeps periodically checking; once the target process appears, KTM enables the module automatically.
+- If the cgroup path cannot be resolved at startup, KTM falls back to comm-only filtering (degraded mode).
Discovery flow (high level):
-- Scan `/proc` for candidate processes.
-- For each PID, read `/proc/<pid>/comm` and match it against the configured comm prefix (default `banyand`) (or an explicitly provided cgroup path).
-- Once matched, derive the target cgroup from `/proc/<pid>/cgroup` (cgroup v2) and program the eBPF cgroup filter. The comm match remains as runtime fallback if the cgroup check does not fire.
-- If no matching process is found at startup, KTM continues periodic probing and activates once it is found.
+- If `cgroup_path` is explicitly configured, use it directly.
+- Otherwise, derive a Pod-level cgroup from `/proc/self/cgroup`.
+- Program the eBPF cgroup filter with the resolved cgroup ID.
#### VM / bare metal
Discovery flow (high level):
-- Scan `/proc` for candidate processes.
-- Match `/proc/<pid>/comm` against the configured prefix (default `banyand`).
-- Use the discovered PID to derive cgroup v2 path and program the filter; keep the comm match as runtime fallback if cgroup filtering is unavailable.
-- If no matching process is found at startup, KTM continues periodic probing and activates once it is found.
+- If `cgroup_path` is explicitly configured, use it directly.
+- Otherwise, derive a Pod-level cgroup from `/proc/self/cgroup`.
+- Program the eBPF cgroup filter with the resolved cgroup ID.
+- If resolution fails, fall back to comm-only filtering (degraded mode).
### Scoping Semantics
- The BPF maps use a single-slot structure (e.g., a BPF array map with a single entry) to store global monotonic counters for the target process.
- This approach eliminates the need for per-pid hash maps, key eviction logic, and complexities related to tracking multiple processes.
-- All kernel events are filtered by the target container’s cgroup ID when available; if the cgroup filter misses (for example, map not populated), a comm-prefix match (configurable; default `banyand`) is used before any counters are updated.
+- **Kernel-side filtering**: All kernel events are filtered by checking `comm="banyand"` (hardcoded, non-configurable). When cgroup filtering is available, both cgroup ID and comm are enforced. When cgroup is unavailable (degraded mode), only comm filtering is used.
Example (YAML):
```yaml
@@ -116,9 +116,9 @@ collector:
- iomonitor
ebpf:
# Optional: absolute or /sys/fs/cgroup-relative path to the BanyanDB cgroup v2.
+ # If not set, KTM attempts auto-detection by scanning /proc for comm="banyand".
+ # Falls back to comm-only mode on failure.
# cgroup_path: /sys/fs/cgroup/<banyandb-cgroup>
- # Optional: comm name prefix used for process discovery and fallback filtering.
- # comm_prefix: banyand
```
@@ -144,26 +144,30 @@ KTM itself has no direct network exporters. All metrics collected by KTM are per
Prefix: metrics are currently emitted under the `ktm_` namespace to reflect their kernel eBPF origin (e.g., `ktm_cache_misses_total`).
- I/O & Cache
- - `ktm_cache_read_attempts_total`
- - `ktm_cache_misses_total`
- - `ktm_page_cache_adds_total`
+ - `ktm_cache_lookups_total`
+ - `ktm_cache_fills_total`
+ - `ktm_cache_deletes_total`
- I/O latency (syscall-level)
- `ktm_sys_read_latency_seconds` (histogram family: exposes `_bucket`, `_count`, `_sum`)
- `ktm_sys_pread_latency_seconds` (histogram family: exposes `_bucket`, `_count`, `_sum`)
+ - `ktm_sys_read_bytes_total`
+ - `ktm_sys_pread_bytes_total`
- fadvise()
- `ktm_fadvise_calls_total`
- - `ktm_fadvise_advice_total{advice="..."}`
- - `ktm_fadvise_success_total`
+ - `ktm_fadvise_dontneed_total`
- Memory
- `ktm_memory_lru_pages_scanned_total`
- `ktm_memory_lru_pages_reclaimed_total`
- - `ktm_memory_direct_reclaim_processes`
+ - `ktm_memory_lru_shrink_events_total`
+ - `ktm_memory_direct_reclaim_begin_total`
+- Status
+ - `ktm_degraded` (gauge; 1 if KTM is running in comm-only degraded mode, 0 otherwise)
Semantics: all counters are monotonic; latency metrics are exported as Prometheus histograms (`_bucket`, `_count`, `_sum`); use Prometheus functions for rates/derivatives; no map clearing between scrapes. KTM does not emit ratio/percentage metrics; derive them upstream.
## Safety & Overhead Boundary
-KTM is strictly passive: no kernel modifications, no syscall blocking, and only bounded-size monotonic maps. Probes attach to stable tracepoints or fentry/fexit paths with kprobe fallbacks, and expected CPU overhead remains <1% under typical BanyanDB workloads.
+KTM is strictly passive: no kernel modifications, no syscall blocking, and only bounded-size monotonic maps. Probes attach to stable tracepoints, and expected CPU overhead remains <1% under typical BanyanDB workloads.
## Security and Permissions
@@ -177,7 +181,7 @@ The sidecar should be configured with the minimal set of capabilities required f
## Failure Modes
-KTM is designed to fail gracefully. If the eBPF programs fail to load at startup for any reason (e.g., kernel incompatibility, insufficient permissions, BTF information unavailable), the KTM module will be disabled.
+KTM is designed to fail gracefully. If the eBPF programs fail to load at startup for any reason (e.g., kernel incompatibility, missing tracefs/tracepoints, insufficient permissions, BTF information unavailable), the KTM module will be disabled.
In this state:
- An error will be logged to indicate that KTM could not be initialized and is therefore inactive.
@@ -189,15 +193,17 @@ This approach ensures that a failure within the observability module does not im
## Restart Semantics
- On sidecar restart, BPF maps are recreated and all counters reset to zero. Downstream systems (e.g., Prometheus via FODC integrations) should treat this as a new counter lifecycle and continue deriving rates/derivatives normally.
-- If BanyanDB restarts (PID changes), the cgroup filter continues to match as long as the container does not change; the comm fallback still matches the configured prefix (default `banyand`).
-- If the pod/container is recreated (cgroup path changes), KTM re-runs target discovery, re-programs the cgroup filter, and starts counters from zero; metrics from the old container are discarded without reconciliation.
-- KTM performs a lightweight health check during collection to ensure the cgroup filter is still populated; if it is missing (for example, container crash/restart), KTM re-detects and re-programs the filter automatically.
+- If BanyanDB restarts (PID changes), the cgroup filter continues to match as long as the container does not change; the comm fallback still matches `banyand`.
+- If the pod/container is recreated (cgroup path changes), KTM re-runs target discovery on the next sidecar start, re-programs the cgroup filter, and starts counters from zero; metrics from the old container are discarded without reconciliation.
+- KTM does not perform periodic health checks or re-detection. If the cgroup path changes, the sidecar/Pod must be restarted.
## Kernel Attachment Points (Current)
+All attachment points are tracepoints.
+
- `sys_enter_read`, `sys_exit_read`, `sys_enter_pread64`, `sys_exit_pread64` (syscall-level I/O latency).
-- `mm_filemap_add_to_page_cache`, `filemap_get_read_batch` (page cache add/churn).
-- `ksys_fadvise64_64` (fadvise policy actions; fentry/fexit preferred).
+- `sys_enter_fadvise64`, `sys_exit_fadvise64` (fadvise policy actions).
+- `mm_filemap_get_pages`, `mm_filemap_add_to_page_cache`, `mm_filemap_delete_from_page_cache` (page cache add/churn).
- `mm_vmscan_lru_shrink_inactive`, `mm_vmscan_direct_reclaim_begin` (memory reclaim/pressure).
## Limitations
diff --git a/docs/operation/fodc/ktm_metrics.md b/docs/operation/fodc/ktm_metrics.md
index fc1117ab1..1355f8c3a 100644
--- a/docs/operation/fodc/ktm_metrics.md
+++ b/docs/operation/fodc/ktm_metrics.md
@@ -322,6 +322,7 @@ graph TD
## 8. Operational Prerequisites and Observability
-- BTF availability and bpffs mounted are expected for fentry/fexit and map pinning where used.
-- Kernel versions must support the chosen tracepoints/fentry paths; kprobe fallbacks apply otherwise.
+- Tracefs must be mounted (usually at `/sys/kernel/tracing`) and syscall tracepoints enabled; KTM attaches via tracepoints.
+- Kernel versions must expose the required tracepoints (syscalls, filemap, vmscan); there is no fentry/kprobe fallback in the current implementation.
+- BTF availability is expected for CO-RE program loading; missing BTF prevents KTM from starting.
- On failure to load/attach, KTM logs an error and disables itself (see Failure Modes in the design document).
diff --git a/fodc/agent/Makefile b/fodc/agent/Makefile
index 8a5672a26..7ec7b702d 100644
--- a/fodc/agent/Makefile
+++ b/fodc/agent/Makefile
@@ -26,7 +26,6 @@ SOURCE_DIR := fodc/agent
include ../../scripts/build/version.mk
include ../../scripts/build/base.mk
-include ../../scripts/build/generate_go.mk
include ../../scripts/build/build.mk
include ../../scripts/build/docker.mk
include ../../scripts/build/test.mk
@@ -35,3 +34,7 @@ include ../../scripts/build/vendor.mk
include ../../scripts/build/help.mk
prepare-build: generate
+
+.PHONY: generate
+generate:
+ @echo "Skipping KTM eBPF generation (using github.com/SkyAPM/ktm-ebpf prebuilt artifacts)"
diff --git a/fodc/agent/internal/cmd/root.go b/fodc/agent/internal/cmd/root.go
index c4aaab190..fa7a09167 100644
--- a/fodc/agent/internal/cmd/root.go
+++ b/fodc/agent/internal/cmd/root.go
@@ -29,11 +29,14 @@ import (
"time"
"github.com/prometheus/client_golang/prometheus"
+ "github.com/rs/zerolog"
"github.com/spf13/cobra"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/cluster"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/exporter"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/flightrecorder"
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/ktm"
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/proxy"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/server"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/watchdog"
@@ -58,6 +61,8 @@ var (
pollMetricsPorts []string
maxMetricsMemoryUsagePercent int
prometheusListenAddr string
+ ktmEnabled bool
+ ktmInterval time.Duration
proxyAddr string
podName string
containerNames []string
@@ -100,12 +105,75 @@ func init() {
"Interval for sending heartbeats to Proxy. Note: The Proxy may override this value in RegisterAgentResponse.")
rootCmd.Flags().DurationVar(&reconnectInterval, "reconnect-interval", defaultReconnectInterval,
"Interval for reconnection attempts when connection to Proxy is lost")
+ rootCmd.Flags().BoolVar(&ktmEnabled, "ktm-enabled", false, "Enable Kernel Trace Module (eBPF)")
+ rootCmd.Flags().DurationVar(&ktmInterval, "ktm-interval", 10*time.Second, "Interval for KTM metrics collection")
rootCmd.Flags().StringSliceVar(&clusterStatePorts, "cluster-state-ports", []string{},
"Ports of the BanyanDB node's gRPC endpoints to poll cluster state from. If empty, cluster state polling is disabled.")
rootCmd.Flags().DurationVar(&clusterStatePollInterval, "cluster-state-poll-interval", defaultClusterStatePollInterval,
"Interval at which to poll cluster state from the BanyanDB nodes")
}
+func calculateCapacity(log *logger.Logger) int64 {
+ memoryLimit, memErr := cgroups.MemoryLimit()
+ if memErr != nil {
+ if runtime.GOOS != "linux" || strings.Contains(memErr.Error(), "mountinfo") {
+ log.Debug().Err(memErr).Msg("Cgroup memory limit not available (expected on non-Linux systems), using default capacity")
+ } else {
+ log.Warn().Err(memErr).Msg("Failed to get cgroup memory limit, using default capacity")
+ }
+ memoryLimit = 1024 * 1024 * 1024
+ }
+ var capacitySize int64
+ if memoryLimit > 0 {
+ capacitySize = (memoryLimit * int64(maxMetricsMemoryUsagePercent)) / 100
+ } else {
+ capacitySize = 100 * 1024 * 1024
+ log.Info().Msg("Memory limit is unlimited or invalid, using default capacity of 100MB")
+ }
+ log.Info().
+ Int64("memory-limit-bytes", memoryLimit).
+ Int("memory-usage-percent", maxMetricsMemoryUsagePercent).
+ Int64("capacity-size-bytes", capacitySize).
+ Msg("Flight Recorder capacity configured")
+ return capacitySize
+}
+
+func initializeKTM(ctx context.Context, log *logger.Logger, fr *flightrecorder.FlightRecorder) func() {
+ if !ktmEnabled {
+ return nil
+ }
+
+ if runtime.GOOS != "linux" {
+ log.Info().Str("os", runtime.GOOS).Msg("KTM is only supported on Linux, skipping initialization")
+ if updateErr := fr.Update([]fodcmetrics.RawMetric{
+ {
+ Name: "ktm_status",
+ Value: 0,
+ Desc: "KTM status: 0=Disabled, 1=Degraded (comm-only), 2=Full (cgroup+comm)",
+ },
+ }); updateErr != nil {
+ log.Warn().Err(updateErr).Msg("Failed to write ktm_status metric")
+ }
+ return nil
+ }
+
+ startStop, startErr := startKTM(ctx, *log.Logger, fr)
+ if startErr != nil {
+ log.Warn().Err(startErr).Msg("KTM startup failed, continuing without kernel tracing")
+ if updateErr := fr.Update([]fodcmetrics.RawMetric{
+ {
+ Name: "ktm_status",
+ Value: 0,
+ Desc: "KTM status: 0=Disabled, 1=Degraded (comm-only), 2=Full (cgroup+comm)",
+ },
+ }); updateErr != nil {
+ log.Warn().Err(updateErr).Msg("Failed to write ktm_status metric")
+ }
+ return nil
+ }
+ return startStop
+}
+
// runFODC is the main function for the FODC agent.
func runFODC(_ *cobra.Command, _ []string) error {
if initErr := logger.Init(logger.Logging{Env: "prod", Level: "info"}); initErr != nil {
@@ -138,7 +206,7 @@ func runFODC(_ *cobra.Command, _ []string) error {
ctx := context.Background()
clusterCollector, clusterErr := cluster.StartCollector(ctx, log, clusterStatePorts, clusterStatePollInterval, podName)
if clusterErr != nil {
- _ = metricsServer.Stop()
+ _ = metricsServer.Stop(ctx)
return clusterErr
}
@@ -150,9 +218,15 @@ func runFODC(_ *cobra.Command, _ []string) error {
}
wd := watchdog.NewWatchdogWithConfig(fr, metricsEndpoints, metricsPollInterval, nodeRole, podName, containerNames)
+
+ stopKTM := initializeKTM(ctx, log, fr)
+
if preRunErr := wd.PreRun(ctx); preRunErr != nil {
+ if stopKTM != nil {
+ stopKTM()
+ }
cluster.StopCollector(clusterCollector)
- _ = metricsServer.Stop()
+ _ = metricsServer.Stop(ctx)
return fmt.Errorf("failed to initialize watchdog: %w", preRunErr)
}
stopCh := wd.Serve()
@@ -171,6 +245,11 @@ func runFODC(_ *cobra.Command, _ []string) error {
}
wd.GracefulStop()
+
+ if stopKTM != nil {
+ stopKTM()
+ }
+
if proxyClient != nil {
if disconnectErr := proxyClient.Disconnect(); disconnectErr != nil {
log.Warn().Err(disconnectErr).Msg("Error disconnecting proxy client")
@@ -178,12 +257,67 @@ func runFODC(_ *cobra.Command, _ []string) error {
} else {
cluster.StopCollector(clusterCollector)
}
- if shutdownErr := metricsServer.Stop(); shutdownErr != nil {
+ if shutdownErr := metricsServer.Stop(ctx); shutdownErr != nil {
log.Warn().Err(shutdownErr).Msg("Error shutting down metrics server")
}
return nil
}
+func startKTM(ctx context.Context, log zerolog.Logger, fr *flightrecorder.FlightRecorder) (func(), error) {
+ if ktmInterval <= 0 {
+ return nil, fmt.Errorf("ktm-interval must be positive, got %v", ktmInterval)
+ }
+
+ ktmCfg := ktm.Config{
+ Enabled: ktmEnabled,
+ Interval: ktmInterval,
+ Modules: []string{"iomonitor"},
+ }
+
+ ktmSvc, createErr := ktm.NewKTM(ktmCfg, log)
+ if createErr != nil {
+ return nil, fmt.Errorf("failed to create KTM: %w", createErr)
+ }
+
+ if startErr := ktmSvc.Start(ctx); startErr != nil {
+ return nil, fmt.Errorf("failed to start KTM: %w", startErr)
+ }
+
+ stopBridgeCh := make(chan struct{})
+ go func() {
+ ticker := time.NewTicker(ktmInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-stopBridgeCh:
+ return
+ case <-ticker.C:
+ rawMetrics := ktmSvc.GetMetrics()
+ ktmStatus := 2.0
+ if ktmSvc.IsDegraded() {
+ ktmStatus = 1.0
+ }
+ rawMetrics = append(rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_status",
+ Value: ktmStatus,
+ Desc: "KTM status: 0=Disabled, 1=Degraded (comm-only), 2=Full (cgroup+comm)",
+ })
+ if updateErr := fr.Update(rawMetrics); updateErr != nil {
+ log.Warn().Err(updateErr).Msg("Failed to update FlightRecorder with KTM metrics")
+ }
+ }
+ }
+ }()
+
+ return func() {
+ close(stopBridgeCh)
+ _ = ktmSvc.Stop()
+ }, nil
+}
+
func validateFlags() error {
if metricsPollInterval <= 0 {
return fmt.Errorf("poll-metrics-interval must be greater than 0")
@@ -200,31 +334,6 @@ func validateFlags() error {
return nil
}
-func calculateCapacity(log *logger.Logger) int64 {
- memoryLimit, memErr := cgroups.MemoryLimit()
- if memErr != nil {
- if runtime.GOOS != "linux" || strings.Contains(memErr.Error(), "mountinfo") {
- log.Debug().Err(memErr).Msg("Cgroup memory limit not available, using default")
- } else {
- log.Warn().Err(memErr).Msg("Failed to get cgroup memory limit, using default")
- }
- memoryLimit = 1024 * 1024 * 1024
- }
- var capacitySize int64
- if memoryLimit > 0 {
- capacitySize = (memoryLimit * int64(maxMetricsMemoryUsagePercent)) / 100
- } else {
- capacitySize = 100 * 1024 * 1024
- log.Info().Msg("Memory limit is unlimited or invalid, using default capacity of 100MB")
- }
- log.Info().
- Int64("memory-limit-bytes", memoryLimit).
- Int("memory-usage-percent", maxMetricsMemoryUsagePercent).
- Int64("capacity-size-bytes", capacitySize).
- Msg("Flight Recorder capacity configured")
- return capacitySize
-}
-
func startProxyClient(ctx context.Context, log *logger.Logger, fr *flightrecorder.FlightRecorder,
nodeRole string, nodeLabels map[string]string, collector *cluster.Collector,
) *proxy.Client {
diff --git a/fodc/agent/internal/integration/basic_metrics_buffering_test.go b/fodc/agent/internal/integration/basic_metrics_buffering_test.go
index d6eba7e15..80b86d931 100644
--- a/fodc/agent/internal/integration/basic_metrics_buffering_test.go
+++ b/fodc/agent/internal/integration/basic_metrics_buffering_test.go
@@ -116,7 +116,7 @@ var _ = Describe("Test Case 1: Basic Metrics Buffering", func() {
// Stop metrics server
if metricsServer != nil {
- stopErr := metricsServer.Stop()
+ stopErr := metricsServer.Stop(context.Background())
Expect(stopErr).NotTo(HaveOccurred())
}
})
diff --git a/fodc/agent/internal/integration/buffer_overflow_test.go b/fodc/agent/internal/integration/buffer_overflow_test.go
index b904b38ec..cdf974dc4 100644
--- a/fodc/agent/internal/integration/buffer_overflow_test.go
+++ b/fodc/agent/internal/integration/buffer_overflow_test.go
@@ -115,7 +115,7 @@ var _ = Describe("Test Case 2: Buffer Overflow Handling", func() {
// Stop metrics server
if metricsServer != nil {
- stopErr := metricsServer.Stop()
+ stopErr := metricsServer.Stop(context.Background())
Expect(stopErr).NotTo(HaveOccurred())
}
})
diff --git a/fodc/agent/internal/integration/metrics_export_test.go b/fodc/agent/internal/integration/metrics_export_test.go
index 3ff1af8cd..0336bd15d 100644
--- a/fodc/agent/internal/integration/metrics_export_test.go
+++ b/fodc/agent/internal/integration/metrics_export_test.go
@@ -23,6 +23,8 @@ import (
"io"
"net"
"net/http"
+ "os"
+ "runtime"
"strings"
"sync"
"time"
@@ -30,12 +32,15 @@ import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
"github.com/prometheus/client_golang/prometheus"
+ "github.com/rs/zerolog"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/exporter"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/flightrecorder"
- "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/ktm"
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/server"
"github.com/apache/skywalking-banyandb/fodc/agent/internal/watchdog"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
)
// volatileMetricNames contains names of metrics that are known to be volatile (gauges that change frequently).
@@ -103,6 +108,15 @@ func calculateTolerance(metricKeyStr string, bufferedValue float64) float64 {
return absTolerance
}
+func skipOrFailKTM(reason string) {
+ if flags.RequireKTM {
+ Fail(reason)
+ return
+ }
+ GinkgoWriter.Printf("Skipping KTM integration test: %s\n", reason)
+ Skip(reason)
+}
+
var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
var (
metricsEndpoint string
@@ -185,7 +199,7 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
// Stop metrics server
if metricsServer != nil {
- stopErr := metricsServer.Stop()
+ stopErr := metricsServer.Stop(context.Background())
Expect(stopErr).NotTo(HaveOccurred())
}
})
@@ -280,7 +294,7 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
// Step 4: Parse Prometheus format and verify exported metrics match buffered metrics
// Strip timestamps from metric lines before parsing (Prometheus format includes optional timestamps)
bodyWithoutTimestamps := stripTimestampsFromPrometheusFormat(body)
- parsedMetrics, parseErr := metrics.ParseWithAgentLabels(bodyWithoutTimestamps, "", "", "")
+ parsedMetrics, parseErr := fodcmetrics.ParseWithAgentLabels(bodyWithoutTimestamps, "", "", "")
Expect(parseErr).NotTo(HaveOccurred(), "Should be able to parse Prometheus format")
Expect(len(parsedMetrics)).To(BeNumerically(">", 0), "Should have parsed at least one metric")
@@ -301,7 +315,7 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
exportedMetricsMap := make(map[string]float64)
for _, parsedMetric := range parsedMetrics {
// Reconstruct metric key from parsed metric
- metricKey := metrics.MetricKey{
+ metricKey := fodcmetrics.MetricKey{
Name: parsedMetric.Name,
Labels: parsedMetric.Labels,
}
@@ -333,7 +347,7 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
// Step 6: Verify metric labels are preserved correctly
for _, parsedMetric := range parsedMetrics {
- metricKey := metrics.MetricKey{
+ metricKey := fodcmetrics.MetricKey{
Name: parsedMetric.Name,
Labels: parsedMetric.Labels,
}
@@ -546,7 +560,7 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
if result != "" {
// Strip timestamps before parsing (Prometheus format includes optional timestamps)
resultWithoutTimestamps := stripTimestampsFromPrometheusFormat(result)
- _, parseErr := metrics.ParseWithAgentLabels(resultWithoutTimestamps, "", "", "")
+ _, parseErr := fodcmetrics.ParseWithAgentLabels(resultWithoutTimestamps, "", "", "")
Expect(parseErr).NotTo(HaveOccurred(),
fmt.Sprintf("Scrape result %d should be valid Prometheus format", i))
}
@@ -561,8 +575,8 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
if firstResult != "" && lastResult != "" {
firstResultWithoutTimestamps := stripTimestampsFromPrometheusFormat(firstResult)
lastResultWithoutTimestamps := stripTimestampsFromPrometheusFormat(lastResult)
- firstMetrics, firstErr := metrics.ParseWithAgentLabels(firstResultWithoutTimestamps, "", "", "")
- lastMetrics, lastErr := metrics.ParseWithAgentLabels(lastResultWithoutTimestamps, "", "", "")
+ firstMetrics, firstErr := fodcmetrics.ParseWithAgentLabels(firstResultWithoutTimestamps, "", "", "")
+ lastMetrics, lastErr := fodcmetrics.ParseWithAgentLabels(lastResultWithoutTimestamps, "", "", "")
if firstErr == nil && lastErr == nil {
// Both should have metrics (may have different values)
@@ -574,6 +588,194 @@ var _ = Describe("Test Case 3: Metrics Export to Prometheus", func() {
}
}
})
+
+ It("should capture and export KTM metrics when enabled", func() {
+ if runtime.GOOS != "linux" {
+ skipOrFailKTM("KTM is only supported on Linux")
+ return
+ }
+
+ // Initialize KTM for this test
+ ctx := context.Background()
+ ktmInterval := 2 * time.Second
+
+ ktmCfg := ktm.Config{
+ Enabled: true,
+ Interval: ktmInterval,
+ Modules: []string{"iomonitor"},
+ }
+
+ ktmLog := zerolog.New(io.Discard)
+ ktmSvc, createErr := ktm.NewKTM(ktmCfg, ktmLog)
+ if createErr != nil {
+ skipOrFailKTM(fmt.Sprintf("KTM initialization failed (may lack permissions): %v", createErr))
+ return
+ }
+
+ startErr := ktmSvc.Start(ctx)
+ if startErr != nil {
+ skipOrFailKTM(fmt.Sprintf("KTM start failed (may lack CAP_BPF or root): %v", startErr))
+ return
+ }
+ defer ktmSvc.Stop()
+
+ // Start metrics bridge goroutine
+ stopBridgeCh := make(chan struct{})
+ defer close(stopBridgeCh)
+
+ go func() {
+ ticker := time.NewTicker(ktmInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-stopBridgeCh:
+ return
+ case <-ticker.C:
+ rawMetrics := ktmSvc.GetMetrics()
+ if rawMetrics == nil {
+ continue
+ }
+ ktmStatus := 2.0
+ if ktmSvc.IsDegraded() {
+ ktmStatus = 1.0
+ }
+ rawMetrics = append(rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_status",
+ Value: ktmStatus,
+ Desc: "KTM status: 1=Degraded (comm-only), 2=Full (cgroup+comm)",
+ })
+ if len(rawMetrics) > 0 {
+ _ = fr.Update(rawMetrics)
+ }
+ }
+ }
+ }()
+
+ // Create a temporary file to generate I/O operations for KTM to capture
+ tmpFile, tmpErr := os.CreateTemp("", "ktm-test-*.dat")
+ Expect(tmpErr).NotTo(HaveOccurred())
+ tmpFilePath := tmpFile.Name()
+ defer os.Remove(tmpFilePath)
+
+ // Write some data to generate I/O
+ testData := make([]byte, 4096)
+ for i := range testData {
+ testData[i] = byte(i % 256)
+ }
+ _, writeErr := tmpFile.Write(testData)
+ Expect(writeErr).NotTo(HaveOccurred())
+ tmpFile.Close()
+
+ // Read the file multiple times to generate read I/O
+ for i := 0; i < 10; i++ {
+ readFile, readErr := os.Open(tmpFilePath)
+ Expect(readErr).NotTo(HaveOccurred())
+ readData := make([]byte, 4096)
+ _, _ = readFile.Read(readData)
+ readFile.Close()
+ time.Sleep(50 * time.Millisecond)
+ }
+
+ // Wait for KTM to collect metrics
+ Eventually(func() bool {
+ datasources := fr.GetDatasources()
+ if len(datasources) == 0 {
+ return false
+ }
+ ds := datasources[0]
+ metricsMap := ds.GetMetrics()
+
+ // Check if any KTM metrics are present
+ for metricKey := range metricsMap {
+ if strings.Contains(metricKey, "ktm_") {
+ return true
+ }
+ }
+ return false
+ }, 10*time.Second, 500*time.Millisecond).Should(BeTrue(), "KTM metrics should be buffered in FlightRecorder")
+
+ // Verify KTM metrics are in FlightRecorder
+ datasources := fr.GetDatasources()
+ Expect(datasources).NotTo(BeEmpty())
+ ds := datasources[0]
+ metricsMap := ds.GetMetrics()
+
+ // Check for ktm_status metric
+ ktmStatusFound := false
+ for metricKey := range metricsMap {
+ if strings.Contains(metricKey, "ktm_status") {
+ ktmStatusFound = true
+ metricBuffer := metricsMap[metricKey]
+ currentValue := metricBuffer.GetCurrentValue()
+ // ktm_status should be 1 (Degraded) or 2 (Full), not 0 (Disabled)
+ Expect(currentValue).To(BeNumerically(">=", 1.0), "KTM should be enabled (status >= 1)")
+ Expect(currentValue).To(BeNumerically("<=", 2.0), "KTM status should be valid (status <= 2)")
+ GinkgoWriter.Printf("KTM status: %.0f\n", currentValue)
+ break
+ }
+ }
+ Expect(ktmStatusFound).To(BeTrue(), "ktm_status metric should be present")
+
+ // Scrape Prometheus endpoint and verify KTM metrics are exported
+ fodcMetricsURL := fmt.Sprintf("http://%s/metrics", metricsServerAddr)
+ scrapeClient := &http.Client{Timeout: 5 * time.Second}
+
+ resp, scrapeErr := scrapeClient.Get(fodcMetricsURL)
+ Expect(scrapeErr).NotTo(HaveOccurred())
+ Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+ bodyBytes, readErr := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ Expect(readErr).NotTo(HaveOccurred())
+
+ body := string(bodyBytes)
+ GinkgoWriter.Printf("Scraped metrics (KTM test):\n%s\n", body)
+
+ // Parse metrics
+ bodyWithoutTimestamps := stripTimestampsFromPrometheusFormat(body)
+ parsedMetrics, parseErr := fodcmetrics.ParseWithAgentLabels(bodyWithoutTimestamps, "", "", "")
+ Expect(parseErr).NotTo(HaveOccurred())
+
+ // Verify KTM metrics are present in exported format
+ ktmMetricsFound := make(map[string]bool)
+ expectedKTMMetrics := []string{
+ "ktm_status",
+ "ktm_degraded",
+ }
+
+ for _, parsedMetric := range parsedMetrics {
+ for _, expectedMetric := range expectedKTMMetrics {
+ if parsedMetric.Name == expectedMetric {
+ ktmMetricsFound[expectedMetric] = true
+ GinkgoWriter.Printf("Found KTM metric: %s = %.2f\n", parsedMetric.Name, parsedMetric.Value)
+ }
+ }
+ // Also check for any other ktm_ prefixed metrics
+ if strings.HasPrefix(parsedMetric.Name, "ktm_") {
+ GinkgoWriter.Printf("Additional KTM metric: %s = %.2f\n", parsedMetric.Name, parsedMetric.Value)
+ }
+ }
+
+ // Verify at least the core KTM metrics are present
+ Expect(ktmMetricsFound["ktm_status"]).To(BeTrue(), "ktm_status should be exported")
+ // ktm_degraded is optional - it may not be present in the first collection cycle
+ if ktmMetricsFound["ktm_degraded"] {
+ GinkgoWriter.Printf("ktm_degraded metric is present\n")
+ } else {
+ GinkgoWriter.Printf("ktm_degraded metric not yet collected (this is acceptable)\n")
+ }
+
+ // Verify ktm_status value in exported metrics matches FlightRecorder
+ for _, parsedMetric := range parsedMetrics {
+ if parsedMetric.Name == "ktm_status" {
+ Expect(parsedMetric.Value).To(BeNumerically(">=", 1.0), "Exported ktm_status should indicate KTM is enabled")
+ Expect(parsedMetric.Value).To(BeNumerically("<=", 2.0), "Exported ktm_status should be valid")
+ }
+ }
+ })
})
// stripTimestampsFromPrometheusFormat removes timestamps from Prometheus metric lines.
diff --git a/fodc/agent/internal/ktm/config.go b/fodc/agent/internal/ktm/config.go
new file mode 100644
index 000000000..aea8de49d
--- /dev/null
+++ b/fodc/agent/internal/ktm/config.go
@@ -0,0 +1,41 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 ktm
+
+import (
+ "time"
+
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/ktm/iomonitor"
+)
+
+// Config represents the configuration for the KTM module.
+type Config struct {
+ EBPF iomonitor.EBPFConfig `mapstructure:"ebpf"`
+ Modules []string `mapstructure:"modules"`
+ Interval time.Duration `mapstructure:"interval"`
+ Enabled bool `mapstructure:"enabled"`
+}
+
+// DefaultConfig returns the default configuration for KTM.
+func DefaultConfig() Config {
+ return Config{
+ Enabled: false,
+ Interval: 10 * time.Second,
+ Modules: []string{"iomonitor"},
+ }
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/collector.go b/fodc/agent/internal/ktm/iomonitor/collector.go
new file mode 100644
index 000000000..28ee4f3f7
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/collector.go
@@ -0,0 +1,213 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 iomonitor implements I/O monitoring using eBPF.
+package iomonitor
+
+import (
+ "context"
+ "fmt"
+ "sync"
+ "time"
+
+ "github.com/rs/zerolog"
+
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
+)
+
+// CollectorConfig defines the collector configuration.
+// Field order chosen to minimize padding (large -> small).
+type CollectorConfig struct {
+ EBPF EBPFConfig `mapstructure:"ebpf"`
+ Modules []string `mapstructure:"modules"`
+ Interval time.Duration `mapstructure:"interval"`
+}
+
+// EBPFConfig defines eBPF-specific configuration.
+type EBPFConfig struct {
+ CgroupPath string `mapstructure:"cgroup_path"` // Optional cgroup v2 path to filter PIDs
+}
+
+// Collector manages eBPF program lifecycle and metrics collection.
+type Collector struct {
+ logger zerolog.Logger
+ modules map[string]Module
+ lastMetrics []fodcmetrics.RawMetric
+ ticker *time.Ticker
+ stopChan chan struct{}
+ config CollectorConfig
+ wg sync.WaitGroup
+ mu sync.RWMutex
+}
+
+// Module represents an eBPF monitoring module.
+type Module interface {
+ Name() string
+ Start() error
+ Stop() error
+ Collect() ([]fodcmetrics.RawMetric, error)
+}
+
+// New creates a new collector instance.
+func New(cfg CollectorConfig, log zerolog.Logger) (*Collector, error) {
+ if cfg.Interval <= 0 {
+ return nil, fmt.Errorf("collector interval must be positive, got %v", cfg.Interval)
+ }
+
+ c := &Collector{
+ config: cfg,
+ logger: log,
+ modules: make(map[string]Module),
+ lastMetrics: make([]fodcmetrics.RawMetric, 0),
+ stopChan: make(chan struct{}),
+ }
+
+ // Initialize modules based on configuration
+ for _, moduleName := range cfg.Modules {
+ module, err := c.createModule(moduleName, cfg.EBPF)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create module %s: %w", moduleName, err)
+ }
+ c.modules[moduleName] = module
+ }
+
+ return c, nil
+}
+
+// createModule creates a module instance by name.
+func (c *Collector) createModule(name string, ebpfCfg EBPFConfig) (Module, error) {
+ switch name {
+ case "iomonitor":
+ // Create the comprehensive I/O monitor module
+ module, err := newModule(c.logger, ebpfCfg)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create iomonitor module: %w", err)
+ }
+ return module, nil
+ default:
+ return nil, fmt.Errorf("unknown module: %s", name)
+ }
+}
+
+// Start starts the collector.
+func (c *Collector) Start(ctx context.Context) error {
+ c.logger.Info().Msg("Starting collector")
+
+ // Start all modules
+ for name, module := range c.modules {
+ if err := module.Start(); err != nil {
+ return fmt.Errorf("failed to start module %s: %w", name, err)
+ }
+ c.logger.Info().Str("module", name).Msg("Started module")
+ }
+
+ // Start collection ticker
+ c.ticker = time.NewTicker(c.config.Interval)
+ c.wg.Add(1)
+ go c.collectLoop(ctx)
+
+ return nil
+}
+
+// collectLoop runs the collection loop.
+func (c *Collector) collectLoop(ctx context.Context) {
+ defer c.wg.Done()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-c.stopChan:
+ return
+ case <-c.ticker.C:
+ c.Collect()
+ }
+ }
+}
+
+// Collect collects metrics from all modules.
+func (c *Collector) Collect() {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ allMetrics := make([]fodcmetrics.RawMetric, 0)
+
+ for name, module := range c.modules {
+ rawMetrics, err := module.Collect()
+ if err != nil {
+ c.logger.Error().Str("module", name).Err(err).Msg("Failed to collect metrics")
+ continue
+ }
+
+ // Aggregate metrics
+ allMetrics = append(allMetrics, rawMetrics...)
+ c.logger.Debug().Str("module", name).Int("count", len(rawMetrics)).Msg("Collected metrics")
+ }
+
+ c.lastMetrics = allMetrics
+}
+
+// GetMetrics returns current metrics.
+func (c *Collector) GetMetrics() []fodcmetrics.RawMetric {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+
+ // Return a copy to avoid race conditions
+ result := make([]fodcmetrics.RawMetric, len(c.lastMetrics))
+ copy(result, c.lastMetrics)
+ return result
+}
+
+// IsDegraded returns whether any module is running in degraded mode.
+func (c *Collector) IsDegraded() bool {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+
+ for _, mod := range c.modules {
+ if ioMod, ok := mod.(interface{ Degraded() bool }); ok {
+ if ioMod.Degraded() {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+// Close stops the collector and cleans up resources.
+func (c *Collector) Close() error {
+ c.logger.Info().Msg("Stopping collector")
+
+ // Stop ticker
+ if c.ticker != nil {
+ c.ticker.Stop()
+ }
+
+ // Signal stop
+ close(c.stopChan)
+
+ // Wait for collection loop to finish
+ c.wg.Wait()
+
+ // Stop all modules
+ for name, module := range c.modules {
+ if err := module.Stop(); err != nil {
+ c.logger.Error().Str("module", name).Err(err).Msg("Failed to stop module")
+ }
+ }
+
+ return nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_discovery.go b/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_discovery.go
new file mode 100644
index 000000000..6a6adec6f
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_discovery.go
@@ -0,0 +1,72 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+)
+
+// findCgroup2Mount locates the cgroup v2 unified mount point.
+func findCgroup2Mount() (string, error) {
+ data, readErr := os.ReadFile("/proc/mounts")
+ if readErr != nil {
+ return "", readErr
+ }
+ for _, line := range strings.Split(string(data), "\n") {
+ fields := strings.Fields(line)
+ if len(fields) >= 3 && fields[2] == "cgroup2" {
+ if _, statErr := os.Stat(filepath.Join(fields[1], "cgroup.controllers")); statErr == nil {
+ return fields[1], nil
+ }
+ }
+ }
+ return "", errors.New("cgroup2 mount not found")
+}
+
+// findContainerCgroupInTree walks the host cgroup tree rooted at cgroupRoot
+// looking for a cgroup directory that contains a process whose comm matches
+// commName. The walk stops at the first match.
+//
+// This is used as a fallback when /proc/self/cgroup returns "/" due to
+// cgroup namespace isolation, making pod-scoped discovery impossible.
+func findContainerCgroupInTree(cgroupRoot, commName string) (string, error) {
+ var result string
+ walkErr := filepath.WalkDir(cgroupRoot, func(path string, d os.DirEntry, walkDirErr error) error {
+ if walkDirErr != nil || !d.IsDir() {
+ return nil
+ }
+ if cgroupContainsComm(path, commName) {
+ result = path
+ return filepath.SkipAll
+ }
+ return nil
+ })
+ if walkErr != nil {
+ return "", fmt.Errorf("failed to walk cgroup tree: %w", walkErr)
+ }
+ if result == "" {
+ return "", fmt.Errorf("%w: comm=%q not found in cgroup tree %s", ErrProcessNotFound, commName, cgroupRoot)
+ }
+ return result, nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_filter.go b/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_filter.go
new file mode 100644
index 000000000..8a76e8704
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/ebpf/cgroup_filter.go
@@ -0,0 +1,39 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+)
+
+// resolveCgroupPath validates that the given path exists and looks like a
+// cgroup v2 directory (contains cgroup.procs).
+func resolveCgroupPath(path string) (string, error) {
+ abs, absErr := filepath.Abs(path)
+ if absErr != nil {
+ return "", absErr
+ }
+ if _, statErr := os.Stat(filepath.Join(abs, "cgroup.procs")); statErr != nil {
+ return "", fmt.Errorf("not a valid cgroup directory: %w", statErr)
+ }
+ return abs, nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/ebpf/filter.go b/fodc/agent/internal/ktm/iomonitor/ebpf/filter.go
new file mode 100644
index 000000000..7905d26ac
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/ebpf/filter.go
@@ -0,0 +1,224 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+
+ generated "github.com/SkyAPM/ktm-ebpf/iomonitor/ebpf/generated"
+ "github.com/cilium/ebpf"
+ "golang.org/x/sys/unix"
+)
+
+const targetComm = "banyand"
+
+// ErrProcessNotFound is returned when no process with the given comm name is found.
+var ErrProcessNotFound = errors.New("process not found")
+
+// configureFilters sets up the cgroup filtering strategy.
+// Returns (degraded bool, degradedReason string, error) where degraded=true means cgroup filtering is disabled.
+func configureFilters(objs *generated.IomonitorObjects, cgroupPath string) (bool, string, error) {
+ targetPath, resolveErr := resolveTargetCgroupPath(cgroupPath)
+ if resolveErr != nil {
+ if updateErr := updateConfigMap(objs.ConfigMap, 0); updateErr != nil {
+ return false, "", fmt.Errorf("failed to configure degraded mode: %w", updateErr)
+ }
+ return true, fmt.Sprintf("cgroup path resolution failed: %v", resolveErr), nil
+ }
+
+ cgID, cgErr := cgroupID(targetPath)
+ if cgErr != nil {
+ if updateErr := updateConfigMap(objs.ConfigMap, 0); updateErr != nil {
+ return false, "", fmt.Errorf("failed to configure degraded mode: %w", updateErr)
+ }
+ return true, fmt.Sprintf("failed to get cgroup ID from %s: %v", targetPath, cgErr), nil
+ }
+
+ if updateErr := updateConfigMap(objs.ConfigMap, cgID); updateErr != nil {
+ return false, "", fmt.Errorf("failed to program config map: %w", updateErr)
+ }
+
+ return false, "", nil
+}
+
+// resolveTargetCgroupPath returns the absolute cgroup v2 directory path of the
+// banyand container to use as the eBPF cgroup filter target.
+//
+// If cfgPath is non-empty it is used directly (after validation).
+// Otherwise the function performs pod-scoped discovery:
+// 1. Read fodc-agent's own cgroup from /proc/self/cgroup.
+// 2. Extract the pod-level cgroup (strip container suffix).
+// 3. Scan child cgroups under the pod cgroup for the banyand container.
+// 4. Return the banyand container's cgroup path so that
+// bpf_get_current_cgroup_id() in the kernel produces an exact match.
+func resolveTargetCgroupPath(cfgPath string) (string, error) {
+ if cfgPath != "" {
+ return resolveCgroupPath(cfgPath)
+ }
+
+ cgMount, mountErr := findCgroup2Mount()
+ if mountErr != nil {
+ return "", fmt.Errorf("failed to find cgroup2 mount: %w", mountErr)
+ }
+
+ selfCgRel, selfErr := readSelfCgroupV2Path()
+ if selfErr != nil {
+ return "", fmt.Errorf("failed to read self cgroup: %w", selfErr)
+ }
+
+ // Inside a cgroup-namespaced container /proc/self/cgroup returns "/".
+ // Fall back to walking the host cgroup tree to find the banyand container.
+ if selfCgRel == "/" {
+ return findContainerCgroupInTree(cgMount, targetComm)
+ }
+
+ podCgRel := extractPodLevelCgroup(selfCgRel)
+ if podCgRel == "" {
+ return "", fmt.Errorf("failed to extract pod-level cgroup from %q", selfCgRel)
+ }
+
+ podCgAbs := filepath.Join(cgMount, podCgRel)
+ banyandCg, findErr := findContainerCgroupByComm(podCgAbs, targetComm)
+ if findErr != nil {
+ return "", fmt.Errorf("failed to find %s container cgroup under %s: %w", targetComm, podCgAbs, findErr)
+ }
+
+ return banyandCg, nil
+}
+
+// findContainerCgroupByComm scans child cgroup directories under podCgroupDir
+// and returns the absolute path of the first child cgroup that contains a
+// process whose comm matches the given name.
+func findContainerCgroupByComm(podCgroupDir, commName string) (string, error) {
+ children, readErr := os.ReadDir(podCgroupDir)
+ if readErr != nil {
+ return "", fmt.Errorf("failed to read pod cgroup dir %s: %w", podCgroupDir, readErr)
+ }
+
+ for _, child := range children {
+ if !child.IsDir() {
+ continue
+ }
+ childPath := filepath.Join(podCgroupDir, child.Name())
+ if cgroupContainsComm(childPath, commName) {
+ return childPath, nil
+ }
+ }
+
+ return "", fmt.Errorf("%w: comm=%q not found in any child cgroup of %s", ErrProcessNotFound, commName, podCgroupDir)
+}
+
+// cgroupContainsComm checks whether any PID listed in the cgroup's
+// cgroup.procs has the given comm name.
+func cgroupContainsComm(cgroupDir, commName string) bool {
+ procsPath := filepath.Join(cgroupDir, "cgroup.procs")
+ data, readErr := os.ReadFile(procsPath)
+ if readErr != nil {
+ return false
+ }
+ for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
+ line = strings.TrimSpace(line)
+ if line == "" {
+ continue
+ }
+ pid, parseErr := strconv.Atoi(line)
+ if parseErr != nil {
+ continue
+ }
+ comm, commErr := os.ReadFile(fmt.Sprintf("/proc/%d/comm", pid))
+ if commErr != nil {
+ continue
+ }
+ if strings.TrimSpace(string(comm)) == commName {
+ return true
+ }
+ }
+ return false
+}
+
+// extractPodLevelCgroup extracts the pod-level cgroup path from a container-level path.
+//
+// Example:
+//
+// Input: /kubepods.slice/kubepods-pod<uuid>.slice/cri-containerd-<id>.scope
+// Output: /kubepods.slice/kubepods-pod<uuid>.slice
+func extractPodLevelCgroup(containerPath string) string {
+ parts := strings.Split(strings.TrimPrefix(containerPath, "/"), "/")
+ for idx := len(parts) - 1; idx >= 0; idx-- {
+ part := parts[idx]
+ // Kubernetes cgroup patterns:
+ // - kubepods-pod<uuid>.slice (systemd driver)
+ // - pod<uuid> (cgroupfs driver)
+ if strings.HasPrefix(part, "kubepods-pod") ||
+ (strings.HasPrefix(part, "pod") && len(part) > 3) {
+ return "/" + strings.Join(parts[:idx+1], "/")
+ }
+ }
+ return ""
+}
+
+// readSelfCgroupV2Path reads /proc/self/cgroup and returns the cgroup v2
+// relative path for the current process.
+func readSelfCgroupV2Path() (string, error) {
+ data, readErr := os.ReadFile("/proc/self/cgroup")
+ if readErr != nil {
+ return "", readErr
+ }
+ for _, line := range strings.Split(string(data), "\n") {
+ line = strings.TrimSpace(line)
+ if line == "" {
+ continue
+ }
+ parts := strings.SplitN(line, ":", 3)
+ if len(parts) != 3 || parts[0] != "0" {
+ continue
+ }
+ cgPath := parts[2]
+ if cgPath == "" {
+ cgPath = "/"
+ }
+ return cgPath, nil
+ }
+ return "", errors.New("cgroup v2 entry not found in /proc/self/cgroup")
+}
+
+func updateConfigMap(configMap *ebpf.Map, cgroupID uint64) error {
+ if configMap == nil {
+ return fmt.Errorf("config_map not available")
+ }
+
+ key := uint32(0)
+ return configMap.Update(key, cgroupID, ebpf.UpdateAny)
+}
+
+// cgroupID returns the inode number (used as cgroup ID by the kernel) for the
+// given cgroup directory path.
+func cgroupID(path string) (uint64, error) {
+ var st unix.Stat_t
+ if statErr := unix.Stat(path, &st); statErr != nil {
+ return 0, statErr
+ }
+ return st.Ino, nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/ebpf/kernel_compat.go b/fodc/agent/internal/ktm/iomonitor/ebpf/kernel_compat.go
new file mode 100644
index 000000000..8aa23c762
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/ebpf/kernel_compat.go
@@ -0,0 +1,230 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+// Package ebpf provides eBPF program loading and management functionality.
+package ebpf
+
+import (
+ "fmt"
+ "os"
+ "strconv"
+ "strings"
+
+ "github.com/cilium/ebpf/btf"
+)
+
+// KernelFeatures represents available kernel features for eBPF.
+type KernelFeatures struct {
+ KernelVersion KernelVersion
+ HasBTF bool
+ HasFentry bool
+ HasTracepoints bool
+ HasKprobes bool
+}
+
+// KernelVersion represents a parsed kernel version.
+type KernelVersion struct {
+ Major int
+ Minor int
+ Patch int
+}
+
+// DetectKernelFeatures detects available kernel features for eBPF attachment.
+func DetectKernelFeatures() (*KernelFeatures, error) {
+ features := &KernelFeatures{
+ HasTracepoints: true, // Available since 4.7
+ HasKprobes: true, // Available since ancient times
+ }
+
+ // Get kernel version
+ version, err := GetKernelVersion()
+ if err != nil {
+ return nil, fmt.Errorf("failed to get kernel version: %w", err)
+ }
+ features.KernelVersion = version
+
+ // Check BTF support (required for fentry/fexit)
+ features.HasBTF = checkBTFSupport()
+
+ // fentry/fexit requires kernel >= 5.5 and BTF support
+ if version.Major > 5 || (version.Major == 5 && version.Minor >= 5) {
+ features.HasFentry = features.HasBTF
+ }
+
+ return features, nil
+}
+
+// GetKernelVersion parses the kernel version from /proc/version_signature or uname.
+func GetKernelVersion() (KernelVersion, error) {
+ // Try Ubuntu's version_signature first (more reliable)
+ if data, err := os.ReadFile("/proc/version_signature"); err == nil {
+ parts := strings.Fields(string(data))
+ if len(parts) >= 2 {
+ return parseVersionString(parts[1])
+ }
+ }
+
+ // Fallback to /proc/sys/kernel/osrelease
+ data, err := os.ReadFile("/proc/sys/kernel/osrelease")
+ if err != nil {
+ return KernelVersion{}, fmt.Errorf("failed to read kernel version: %w", err)
+ }
+
+ versionStr := strings.TrimSpace(string(data))
+ return parseVersionString(versionStr)
+}
+
+// parseVersionString parses a kernel version string like "5.15.0-91-generic".
+func parseVersionString(versionStr string) (KernelVersion, error) {
+ // Remove any suffix after the version numbers
+ if idx := strings.IndexAny(versionStr, "-_"); idx != -1 {
+ versionStr = versionStr[:idx]
+ }
+
+ parts := strings.Split(versionStr, ".")
+ if len(parts) < 2 {
+ return KernelVersion{}, fmt.Errorf("invalid kernel version format: %s", versionStr)
+ }
+
+ var version KernelVersion
+ var err error
+
+ version.Major, err = strconv.Atoi(parts[0])
+ if err != nil {
+ return KernelVersion{}, fmt.Errorf("invalid major version: %w", err)
+ }
+
+ version.Minor, err = strconv.Atoi(parts[1])
+ if err != nil {
+ return KernelVersion{}, fmt.Errorf("invalid minor version: %w", err)
+ }
+
+ if len(parts) >= 3 {
+ version.Patch, _ = strconv.Atoi(parts[2]) // Ignore error for patch
+ }
+
+ return version, nil
+}
+
+// checkBTFSupport checks if the kernel has BTF support.
+func checkBTFSupport() bool {
+ // Check if BTF is available in /sys/kernel/btf/vmlinux
+ if _, err := os.Stat("/sys/kernel/btf/vmlinux"); err == nil {
+ return true
+ }
+
+ // Try to load BTF from kernel
+ _, err := btf.LoadKernelSpec()
+ return err == nil
+}
+
+// GetFadviseFunctionNames returns possible function names for fadvise based on kernel version.
+func GetFadviseFunctionNames(version KernelVersion) []string {
+ // Kernel >= 5.11: ksys_fadvise64_64 is the primary internal function
+ if version.Major > 5 || (version.Major == 5 && version.Minor >= 11) {
+ return []string{
+ "ksys_fadvise64_64",
+ "__x64_sys_fadvise64",
+ "__do_sys_fadvise64", // Some kernels use this
+ "sys_fadvise64",
+ }
+ }
+
+ // Kernel 5.0 - 5.10
+ if version.Major == 5 {
+ return []string{
+ "ksys_fadvise64_64",
+ "__x64_sys_fadvise64",
+ "sys_fadvise64",
+ }
+ }
+
+ // Kernel 4.x
+ return []string{
+ "sys_fadvise64",
+ "SyS_fadvise64", // Older naming convention
+ "sys_fadvise64_64",
+ }
+}
+
+// GetFilemapFunctionNames returns possible function names for filemap operations.
+func GetFilemapFunctionNames(version KernelVersion) (readFuncs []string, addFuncs []string) {
+ // Read batch functions (varies by kernel version)
+ if version.Major > 5 || (version.Major == 5 && version.Minor >= 15) {
+ readFuncs = []string{
+ "filemap_get_read_batch",
+ "filemap_read_folio", // Newer kernels use folio
+ "filemap_read_page",
+ }
+ } else {
+ readFuncs = []string{
+ "filemap_get_read_batch",
+ "do_generic_file_read",
+ "generic_file_buffered_read",
+ }
+ }
+
+ // Page cache add functions
+ if version.Major > 5 || (version.Major == 5 && version.Minor >= 16) {
+ addFuncs = []string{
+ "filemap_add_folio", // Newer kernels use folio
+ "__filemap_add_folio",
+ "add_to_page_cache_lru",
+ }
+ } else {
+ addFuncs = []string{
+ "add_to_page_cache_lru",
+ "add_to_page_cache_locked",
+ "__add_to_page_cache_locked",
+ }
+ }
+
+ return readFuncs, addFuncs
+}
+
+// ShouldUseFentry determines if fentry should be used for a given function.
+func ShouldUseFentry(features *KernelFeatures, funcName string) bool {
+ if !features.HasFentry {
+ return false
+ }
+
+ // Check if the function exists in BTF
+ spec, err := btf.LoadKernelSpec()
+ if err != nil {
+ return false
+ }
+
+ var found bool
+ err = spec.TypeByName(funcName, &found)
+ return err == nil && found
+}
+
+// AttachmentModeString returns a string representation of the best available attachment mode.
+func AttachmentModeString(features *KernelFeatures) string {
+ if features.HasFentry {
+ return "fentry/fexit (optimal)"
+ }
+ if features.HasTracepoints {
+ return "tracepoints (stable)"
+ }
+ if features.HasKprobes {
+ return "kprobes (fallback)"
+ }
+ return "none available"
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/ebpf/loader.go b/fodc/agent/internal/ktm/iomonitor/ebpf/loader.go
new file mode 100644
index 000000000..d9df4e7dd
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/ebpf/loader.go
@@ -0,0 +1,333 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+// Enhanced loader with tracepoint-based eBPF program attachment.
+
+import (
+ "fmt"
+ "os"
+
+ generated "github.com/SkyAPM/ktm-ebpf/iomonitor/ebpf/generated"
+ "github.com/cilium/ebpf"
+ "github.com/cilium/ebpf/link"
+ "github.com/cilium/ebpf/rlimit"
+ "github.com/rs/zerolog"
+)
+
+// Attachment mode constants for tracking how programs are attached.
+const (
+ // attachModeTracepoint indicates attachment via tracepoint.
+ attachModeTracepoint = "tracepoint"
+)
+
+// EnhancedLoader handles loading and managing eBPF programs with intelligent fallback.
+type EnhancedLoader struct {
+ attachmentModes map[string]string
+ spec *ebpf.CollectionSpec
+ objects *generated.IomonitorObjects
+ features *KernelFeatures
+ logger zerolog.Logger
+ cgroupPath string
+ links []link.Link
+ degraded bool
+}
+
+// NewEnhancedLoader creates a new enhanced eBPF program loader.
+func NewEnhancedLoader(log zerolog.Logger) (*EnhancedLoader, error) {
+ // Remove memory limit for eBPF
+ if err := rlimit.RemoveMemlock(); err != nil {
+ return nil, fmt.Errorf("failed to remove memlock limit: %w", err)
+ }
+
+ // Detect kernel features
+ features, err := DetectKernelFeatures()
+ if err != nil {
+ return nil, fmt.Errorf("failed to detect kernel features: %w", err)
+ }
+
+ log.Info().
+ Str("version", fmt.Sprintf("%d.%d.%d",
+ features.KernelVersion.Major,
+ features.KernelVersion.Minor,
+ features.KernelVersion.Patch)).
+ Bool("BTF", features.HasBTF).
+ Msg("Detected kernel features")
+
+ return &EnhancedLoader{
+ links: make([]link.Link, 0),
+ features: features,
+ logger: log,
+ attachmentModes: make(map[string]string),
+ cgroupPath: "",
+ }, nil
+}
+
+// SetCgroupPath enables cgroup-based PID filtering using a cgroup v2 path.
+// An empty path disables the filter.
+func (l *EnhancedLoader) SetCgroupPath(path string) {
+ l.cgroupPath = path
+}
+
+// LoadPrograms loads the eBPF programs.
+func (l *EnhancedLoader) LoadPrograms() error {
+ var err error
+
+ // Load the eBPF program collection
+ l.spec, err = generated.LoadIomonitor()
+ if err != nil {
+ return fmt.Errorf("failed to load eBPF spec: %w", err)
+ }
+
+ // Load eBPF objects
+ l.objects = &generated.IomonitorObjects{}
+ if err = l.spec.LoadAndAssign(l.objects, nil); err != nil {
+ return fmt.Errorf("failed to load eBPF objects: %w", err)
+ }
+
+ // Configure cgroup filter if requested.
+ degraded, degradedReason, err := l.ConfigureFilters()
+ if err != nil {
+ return fmt.Errorf("failed to configure filters: %w", err)
+ }
+ l.setDegradedState(degraded, degradedReason)
+
+ return nil
+}
+
+// ConfigureFilters sets cgroup id for filtering.
+// Returns (degraded bool, degradedReason string, error) where degraded=true means cgroup filtering is disabled.
+func (l *EnhancedLoader) ConfigureFilters() (bool, string, error) {
+ if l.objects == nil {
+ return false, "", fmt.Errorf("eBPF objects not loaded")
+ }
+
+ return configureFilters(l.objects, l.cgroupPath)
+}
+
+// Degraded reports whether KTM is running in comm-only (degraded) mode.
+func (l *EnhancedLoader) Degraded() bool {
+ return l.degraded
+}
+
+// AttachPrograms attaches the eBPF programs with intelligent fallback.
+func (l *EnhancedLoader) AttachPrograms() error {
+ if l.objects == nil {
+ return fmt.Errorf("eBPF objects not loaded")
+ }
+
+ // Attach fadvise monitoring
+ if err := l.attachFadviseWithFallback(); err != nil {
+ l.logger.Error().Err(err).Msg("Failed to attach fadvise monitoring")
+ return err
+ }
+
+ // Attach read latency monitoring
+ if err := l.attachReadLatencyWithFallback(); err != nil {
+ l.logger.Error().Err(err).Msg("Failed to attach read latency monitoring")
+ return err
+ }
+
+ // Attach memory monitoring (optional, continue on failure)
+ l.attachMemoryTracepoints()
+
+ // Attach cache monitoring
+ l.attachCacheWithFallback()
+
+ // Log attachment summary
+ l.logger.Info().Any("attachment_modes", l.attachmentModes).Msg("eBPF programs attached successfully")
+
+ return nil
+}
+
+// attachFadviseWithFallback attaches fadvise monitoring using tracepoints.
+func (l *EnhancedLoader) attachFadviseWithFallback() error {
+ funcName := "fadvise64"
+
+ if l.objects.TraceEnterFadvise64 == nil || l.objects.TraceExitFadvise64 == nil {
+ return fmt.Errorf("fadvise tracepoint programs not loaded")
+ }
+
+ tpEnter, err := link.Tracepoint("syscalls", "sys_enter_fadvise64",
+ l.objects.TraceEnterFadvise64, nil)
+ if err != nil {
+ return fmt.Errorf("failed to attach sys_enter_fadvise64: %w", err)
+ }
+ l.links = append(l.links, tpEnter)
+
+ tpExit, err := link.Tracepoint("syscalls", "sys_exit_fadvise64",
+ l.objects.TraceExitFadvise64, nil)
+ if err != nil {
+ // Cleanup enter if exit fails
+ tpEnter.Close()
+ l.links = l.links[:len(l.links)-1]
+ return fmt.Errorf("failed to attach sys_exit_fadvise64: %w", err)
+ }
+ l.links = append(l.links, tpExit)
+ l.attachmentModes[funcName] = attachModeTracepoint
+ l.logger.Info().Msg("Attached fadvise monitoring using tracepoints")
+ return nil
+}
+
+// attachReadLatencyWithFallback attaches tracepoints for read and pread64.
+func (l *EnhancedLoader) attachReadLatencyWithFallback() error {
+ if err := l.attachSyscallLatency("read",
+ l.objects.TraceEnterReadTp, l.objects.TraceExitReadTp); err != nil {
+ return err
+ }
+
+ return l.attachSyscallLatency("pread64",
+ l.objects.TraceEnterPread64Tp, l.objects.TraceExitPread64Tp)
+}
+
+func (l *EnhancedLoader) attachSyscallLatency(syscallName string, tpEnterProg, tpExitProg *ebpf.Program) error {
+ attachmentKey := "syscall_" + syscallName
+
+ if tpEnterProg == nil || tpExitProg == nil {
+ return fmt.Errorf("tracepoint programs for %s not loaded", syscallName)
+ }
+
+ tpEnter, err := link.Tracepoint("syscalls", "sys_enter_"+syscallName, tpEnterProg, nil)
+ if err != nil {
+ return fmt.Errorf("failed to attach sys_enter_%s: %w", syscallName, err)
+ }
+ l.links = append(l.links, tpEnter)
+
+ tpExit, err := link.Tracepoint("syscalls", "sys_exit_"+syscallName, tpExitProg, nil)
+ if err != nil {
+ _ = tpEnter.Close()
+ l.links = l.links[:len(l.links)-1]
+ return fmt.Errorf("failed to attach sys_exit_%s: %w", syscallName, err)
+ }
+ l.links = append(l.links, tpExit)
+ l.attachmentModes[attachmentKey] = attachModeTracepoint
+ l.logger.Info().Str("syscall", syscallName).Msg("Attached syscall latency using tracepoints")
+ return nil
+}
+
+// attachCacheWithFallback attaches cache monitoring tracepoints.
+func (l *EnhancedLoader) attachCacheWithFallback() {
+ if l.objects.TraceMmFilemapGetPages != nil {
+ tp, err := link.Tracepoint("filemap", "mm_filemap_get_pages", l.objects.TraceMmFilemapGetPages, nil)
+ if err == nil {
+ l.links = append(l.links, tp)
+ l.attachmentModes["filemap_lookup"] = attachModeTracepoint
+ l.logger.Info().Msg("Attached cache lookup monitoring using tracepoint")
+ } else {
+ l.logger.Debug().Err(err).Msg("Failed to attach mm_filemap_get_pages")
+ }
+ }
+
+ if l.objects.TraceMmFilemapAddToPageCache != nil {
+ tp, err := link.Tracepoint("filemap", "mm_filemap_add_to_page_cache", l.objects.TraceMmFilemapAddToPageCache, nil)
+ if err == nil {
+ l.links = append(l.links, tp)
+ l.attachmentModes["page_cache_add"] = attachModeTracepoint
+ l.logger.Info().Msg("Attached cache fill monitoring using tracepoint")
+ } else {
+ l.logger.Debug().Err(err).Msg("Failed to attach mm_filemap_add_to_page_cache")
+ }
+ }
+
+ if l.objects.TraceMmFilemapDeleteFromPageCache != nil {
+ tp, err := link.Tracepoint("filemap", "mm_filemap_delete_from_page_cache", l.objects.TraceMmFilemapDeleteFromPageCache, nil)
+ if err == nil {
+ l.links = append(l.links, tp)
+ l.attachmentModes["page_cache_delete"] = attachModeTracepoint
+ l.logger.Info().Msg("Attached cache delete monitoring using tracepoint")
+ } else {
+ l.logger.Debug().Err(err).Msg("Failed to attach mm_filemap_delete_from_page_cache")
+ }
+ }
+}
+
+// attachMemoryTracepoints attaches memory-related tracepoints (no fallback needed).
+func (l *EnhancedLoader) attachMemoryTracepoints() {
+ // LRU shrink tracepoint
+ if l.objects.TraceLruShrinkInactive != nil {
+ tpLru, err := link.Tracepoint("vmscan", "mm_vmscan_lru_shrink_inactive",
+ l.objects.TraceLruShrinkInactive, nil)
+ if err == nil {
+ l.links = append(l.links, tpLru)
+ l.attachmentModes["lru_shrink"] = attachModeTracepoint
+ }
+ }
+
+ // Direct reclaim tracepoint
+ if l.objects.TraceDirectReclaimBegin != nil {
+ tpReclaim, err := link.Tracepoint("vmscan", "mm_vmscan_direct_reclaim_begin",
+ l.objects.TraceDirectReclaimBegin, nil)
+ if err == nil {
+ l.links = append(l.links, tpReclaim)
+ l.attachmentModes["direct_reclaim"] = attachModeTracepoint
+ }
+ }
+}
+
+// GetObjects returns the loaded eBPF objects.
+func (l *EnhancedLoader) GetObjects() *generated.IomonitorObjects {
+ return l.objects
+}
+
+// GetAttachmentModes returns the attachment modes used for each function.
+func (l *EnhancedLoader) GetAttachmentModes() map[string]string {
+ return l.attachmentModes
+}
+
+// GetKernelFeatures returns detected kernel features.
+func (l *EnhancedLoader) GetKernelFeatures() *KernelFeatures {
+ return l.features
+}
+
+func (l *EnhancedLoader) setDegradedState(degraded bool, reason string) {
+ if degraded == l.degraded {
+ return
+ }
+ l.degraded = degraded
+ if degraded {
+ l.logger.Warn().
+ Str("cgroup_path", l.cgroupPath).
+ Str("reason", reason).
+ Msg("KTM is running in Degraded mode (comm-only) due to Cgroup detection failure. Data isolation in K8s pods cannot be guaranteed.")
+ return
+ }
+ l.logger.Info().Msg("KTM recovered: cgroup scoping restored")
+}
+
+// Close cleans up all resources.
+func (l *EnhancedLoader) Close() error {
+ // Close all links
+ for _, lnk := range l.links {
+ if err := lnk.Close(); err != nil {
+ fmt.Fprintf(os.Stderr, "Warning: failed to close link: %v\n", err)
+ }
+ }
+ l.links = nil
+
+ // Close eBPF objects
+ if l.objects != nil {
+ if err := l.objects.Close(); err != nil {
+ fmt.Fprintf(os.Stderr, "Warning: failed to close eBPF objects: %v\n", err)
+ }
+ l.objects = nil
+ }
+
+ return nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/module_linux.go b/fodc/agent/internal/ktm/iomonitor/module_linux.go
new file mode 100644
index 000000000..ba589151d
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/module_linux.go
@@ -0,0 +1,435 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux && (amd64 || arm64 || 386)
+
+package iomonitor
+
+import (
+ "fmt"
+ "math"
+
+ "github.com/SkyAPM/ktm-ebpf/iomonitor/ebpf/generated"
+ "github.com/rs/zerolog"
+
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/ktm/iomonitor/ebpf"
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
+)
+
+type module struct {
+ logger zerolog.Logger
+ loader *ebpf.EnhancedLoader
+ objs *generated.IomonitorObjects
+ name string
+ cgroupPath string
+}
+
+func newModule(log zerolog.Logger, ebpfCfg EBPFConfig) (*module, error) {
+ ebpfLoader, err := ebpf.NewEnhancedLoader(log)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create eBPF loader: %w", err)
+ }
+
+ return &module{
+ name: "iomonitor",
+ logger: log,
+ loader: ebpfLoader,
+ cgroupPath: ebpfCfg.CgroupPath,
+ }, nil
+}
+
+// Name returns the module name.
+func (m *module) Name() string {
+ return m.name
+}
+
+// Start loads and attaches the eBPF programs.
+func (m *module) Start() error {
+ m.logger.Info().Msg("Starting I/O monitor module")
+
+ // Configure cgroup filtering if requested.
+ if m.cgroupPath != "" {
+ m.logger.Info().Str("cgroup_path", m.cgroupPath).Msg("Enabling cgroup filter for eBPF programs")
+ m.loader.SetCgroupPath(m.cgroupPath)
+ }
+
+ // Load eBPF programs
+ if err := m.loader.LoadPrograms(); err != nil {
+ return fmt.Errorf("failed to load eBPF programs: %w", err)
+ }
+
+ // Attach to tracepoints/kprobes
+ if err := m.loader.AttachPrograms(); err != nil {
+ return fmt.Errorf("failed to attach programs: %w", err)
+ }
+
+ // Get loaded objects
+ m.objs = m.loader.GetObjects()
+ if m.objs == nil {
+ return fmt.Errorf("failed to get eBPF objects")
+ }
+
+ m.logger.Info().Msg("I/O monitor module started successfully")
+ return nil
+}
+
+// Stop cleans up the module.
+func (m *module) Stop() error {
+ m.logger.Info().Msg("Stopping I/O monitor module")
+
+ if m.loader != nil {
+ if err := m.loader.Close(); err != nil {
+ m.logger.Error().Err(err).Msg("Failed to close eBPF loader")
+ }
+ }
+
+ return nil
+}
+
+// Degraded returns whether the module is running in degraded mode.
+func (m *module) Degraded() bool {
+ if m.loader == nil {
+ return false
+ }
+ return m.loader.Degraded()
+}
+
+// Collect gathers all metrics from eBPF maps.
+func (m *module) Collect() ([]fodcmetrics.RawMetric, error) {
+ if m.objs == nil {
+ return nil, fmt.Errorf("eBPF objects not initialized")
+ }
+
+ rawMetrics := make([]fodcmetrics.RawMetric, 0)
+
+ // Collect metrics (cumulative)
+ m.collectMetrics(&rawMetrics)
+ m.addDegradedMetric(&rawMetrics)
+
+ return rawMetrics, nil
+}
+
+func (m *module) addDegradedMetric(rawMetrics *[]fodcmetrics.RawMetric) {
+ degraded := 0.0
+ if m.loader != nil && m.loader.Degraded() {
+ degraded = 1.0
+ }
+ *rawMetrics = append(*rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_degraded",
+ Value: degraded,
+ Desc: "KTM degraded mode indicator: 0=normal, 1=degraded",
+ })
+}
+
+// collectMetrics collects metrics from all maps without clearing them.
+func (m *module) collectMetrics(rawMetrics *[]fodcmetrics.RawMetric) {
+ if err := m.collectFadviseStats(rawMetrics); err != nil {
+ m.logger.Debug().Err(err).Msg("Failed to collect fadvise stats")
+ }
+
+ if err := m.collectCacheStats(rawMetrics); err != nil {
+ m.logger.Debug().Err(err).Msg("Failed to collect cache stats")
+ }
+
+ m.collectMemoryStats(rawMetrics)
+
+ if err := m.collectReadLatencyStats(rawMetrics); err != nil {
+ m.logger.Debug().Err(err).Msg("Failed to collect read latency stats")
+ }
+
+ if err := m.collectPreadLatencyStats(rawMetrics); err != nil {
+ m.logger.Debug().Err(err).Msg("Failed to collect pread latency stats")
+ }
+}
+
+func (m *module) collectFadviseStats(rawMetrics *[]fodcmetrics.RawMetric) error {
+ var pid uint32
+ var perCPUStats []generated.IomonitorFadviseStatsT
+ iter := m.objs.FadviseStatsMap.Iterate()
+
+ var totalCalls uint64
+ var dontneed uint64
+
+ for iter.Next(&pid, &perCPUStats) {
+ // Fold per-CPU values
+ for _, cpuStats := range perCPUStats {
+ totalCalls += cpuStats.TotalCalls
+ dontneed += cpuStats.AdviceDontneed
+ }
+ }
+
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_fadvise_calls_total",
+ Value: float64(totalCalls),
+ Desc: "Total number of fadvise calls",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_fadvise_dontneed_total",
+ Value: float64(dontneed),
+ Desc: "Total number of fadvise DONTNEED calls",
+ },
+ )
+
+ return iter.Err()
+}
+
+func (m *module) collectCacheStats(rawMetrics *[]fodcmetrics.RawMetric) error {
+ var pid uint32
+ var perCPUStats []generated.IomonitorCacheStatsT
+ iter := m.objs.CacheStatsMap.Iterate()
+
+ var lookups uint64
+ var adds uint64
+ var deletes uint64
+
+ for iter.Next(&pid, &perCPUStats) {
+ // Fold per-CPU values
+ for _, cpuStats := range perCPUStats {
+ lookups += cpuStats.Lookups
+ adds += cpuStats.Adds
+ deletes += cpuStats.Deletes
+ }
+ }
+
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_cache_lookups_total",
+ Value: float64(lookups),
+ Desc: "Total number of page cache lookups",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_cache_fills_total",
+ Value: float64(adds),
+ Desc: "Total number of page cache fills",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_cache_deletes_total",
+ Value: float64(deletes),
+ Desc: "Total number of page cache deletions",
+ },
+ )
+
+ return iter.Err()
+}
+
+func (m *module) collectMemoryStats(rawMetrics *[]fodcmetrics.RawMetric) {
+ // LRU shrink stats (per-CPU)
+ var key uint32
+ var perCPUShrink []generated.IomonitorShrinkCountersT
+
+ if err := m.objs.ShrinkStatsMap.Lookup(key, &perCPUShrink); err == nil {
+ var totalScanned, totalReclaimed, totalEvents uint64
+ for _, cpuCounters := range perCPUShrink {
+ totalScanned += cpuCounters.NrScannedTotal
+ totalReclaimed += cpuCounters.NrReclaimedTotal
+ totalEvents += cpuCounters.EventsTotal
+ }
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_memory_lru_pages_scanned_total",
+ Value: float64(totalScanned),
+ Desc: "Total number of LRU pages scanned",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_memory_lru_pages_reclaimed_total",
+ Value: float64(totalReclaimed),
+ Desc: "Total number of LRU pages reclaimed",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_memory_lru_shrink_events_total",
+ Value: float64(totalEvents),
+ Desc: "Total number of LRU shrink events",
+ },
+ )
+ }
+
+ // Direct reclaim stats (per-CPU)
+ var reclaimKey uint32
+ var perCPUReclaim []generated.IomonitorReclaimCountersT
+ if err := m.objs.ReclaimCountersMap.Lookup(reclaimKey, &perCPUReclaim); err == nil {
+ var totalBegin uint64
+ for _, cpuCounters := range perCPUReclaim {
+ totalBegin += cpuCounters.DirectReclaimBeginTotal
+ }
+ *rawMetrics = append(*rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_memory_direct_reclaim_begin_total",
+ Value: float64(totalBegin),
+ Desc: "Total number of direct reclaim begin events",
+ })
+ }
+}
+
+func (m *module) collectReadLatencyStats(rawMetrics *[]fodcmetrics.RawMetric) error {
+ var pid uint32
+ var perCPUStats []generated.IomonitorReadLatencyStatsT
+ iter := m.objs.ReadLatencyStatsMap.Iterate()
+
+ var aggBuckets [32]uint64
+ var totalSumNs uint64
+ var totalCount uint64
+ var totalBytes uint64
+
+ for iter.Next(&pid, &perCPUStats) {
+ // Fold per-CPU values
+ for _, cpuStats := range perCPUStats {
+ totalCount += cpuStats.Count
+ totalSumNs += cpuStats.SumLatencyNs
+ totalBytes += cpuStats.ReadBytesTotal
+ for i, v := range cpuStats.Buckets {
+ aggBuckets[i] += v
+ }
+ }
+ }
+
+ if err := iter.Err(); err != nil {
+ return err
+ }
+
+ // Add histogram sum and count
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_read_latency_seconds_sum",
+ Value: float64(totalSumNs) / 1e9,
+ Desc: "Total sum of read() latency in seconds",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_read_latency_seconds_count",
+ Value: float64(totalCount),
+ Desc: "Total count of read() calls",
+ },
+ )
+
+ // Add histogram buckets
+ var cumulative uint64
+ for i := 0; i < 32; i++ {
+ count := aggBuckets[i]
+ cumulative += count
+
+ // Upper bound in seconds
+ // Bucket i upper bound: 2^i microseconds
+ upperBoundUs := math.Pow(2, float64(i))
+ upperBoundSec := upperBoundUs / 1e6
+
+ *rawMetrics = append(*rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_sys_read_latency_seconds_bucket",
+ Value: float64(cumulative),
+ Labels: []fodcmetrics.Label{
+ {Name: "le", Value: fmt.Sprintf("%g", upperBoundSec)},
+ },
+ Desc: "Histogram buckets for read() latency",
+ })
+ }
+ // Add +Inf bucket and bytes total
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_read_latency_seconds_bucket",
+ Value: float64(cumulative),
+ Labels: []fodcmetrics.Label{
+ {Name: "le", Value: "+Inf"},
+ },
+ Desc: "Histogram buckets for read() latency",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_read_bytes_total",
+ Value: float64(totalBytes),
+ Desc: "Total bytes read via read() syscall",
+ },
+ )
+
+ return nil
+}
+
+func (m *module) collectPreadLatencyStats(rawMetrics *[]fodcmetrics.RawMetric) error {
+ var pid uint32
+ var perCPUStats []generated.IomonitorReadLatencyStatsT
+ iter := m.objs.PreadLatencyStatsMap.Iterate()
+
+ var aggBuckets [32]uint64
+ var totalSumNs uint64
+ var totalCount uint64
+ var totalBytes uint64
+
+ for iter.Next(&pid, &perCPUStats) {
+ // Fold per-CPU values
+ for _, cpuStats := range perCPUStats {
+ totalCount += cpuStats.Count
+ totalSumNs += cpuStats.SumLatencyNs
+ totalBytes += cpuStats.ReadBytesTotal
+ for i, v := range cpuStats.Buckets {
+ aggBuckets[i] += v
+ }
+ }
+ }
+
+ if err := iter.Err(); err != nil {
+ return err
+ }
+
+ // Add histogram sum and count
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_pread_latency_seconds_sum",
+ Value: float64(totalSumNs) / 1e9,
+ Desc: "Total sum of pread() latency in seconds",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_pread_latency_seconds_count",
+ Value: float64(totalCount),
+ Desc: "Total count of pread() calls",
+ },
+ )
+
+ // Add histogram buckets
+ var cumulative uint64
+ for i := 0; i < 32; i++ {
+ count := aggBuckets[i]
+ cumulative += count
+
+ // Upper bound in seconds
+ // Bucket i upper bound: 2^i microseconds
+ upperBoundUs := math.Pow(2, float64(i))
+ upperBoundSec := upperBoundUs / 1e6
+
+ *rawMetrics = append(*rawMetrics, fodcmetrics.RawMetric{
+ Name: "ktm_sys_pread_latency_seconds_bucket",
+ Value: float64(cumulative),
+ Labels: []fodcmetrics.Label{
+ {Name: "le", Value: fmt.Sprintf("%g", upperBoundSec)},
+ },
+ Desc: "Histogram buckets for pread() latency",
+ })
+ }
+ // Add +Inf bucket and bytes total
+ *rawMetrics = append(*rawMetrics,
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_pread_latency_seconds_bucket",
+ Value: float64(cumulative),
+ Labels: []fodcmetrics.Label{
+ {Name: "le", Value: "+Inf"},
+ },
+ Desc: "Histogram buckets for pread() latency",
+ },
+ fodcmetrics.RawMetric{
+ Name: "ktm_sys_pread_bytes_total",
+ Value: float64(totalBytes),
+ Desc: "Total bytes read via pread() syscall",
+ },
+ )
+
+ return nil
+}
diff --git a/fodc/agent/internal/ktm/iomonitor/module_other.go b/fodc/agent/internal/ktm/iomonitor/module_other.go
new file mode 100644
index 000000000..7404cef7d
--- /dev/null
+++ b/fodc/agent/internal/ktm/iomonitor/module_other.go
@@ -0,0 +1,59 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build !linux || !(amd64 || arm64 || 386)
+
+package iomonitor
+
+import (
+ "github.com/rs/zerolog"
+
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
+)
+
+type module struct {
+ logger zerolog.Logger
+ name string
+}
+
+func newModule(log zerolog.Logger, _ EBPFConfig) (*module, error) {
+ return &module{
+ logger: log,
+ name: "iomonitor",
+ }, nil
+}
+
+func (m *module) Name() string {
+ return m.name
+}
+
+func (m *module) Start() error {
+ m.logger.Info().Str("module", m.name).Msg("I/O monitor module runs in no-op mode on this platform")
+ return nil
+}
+
+func (m *module) Stop() error {
+ return nil
+}
+
+func (m *module) Degraded() bool {
+ return false
+}
+
+func (m *module) Collect() ([]fodcmetrics.RawMetric, error) {
+ return []fodcmetrics.RawMetric{}, nil
+}
diff --git a/fodc/agent/internal/ktm/ktm.go b/fodc/agent/internal/ktm/ktm.go
new file mode 100644
index 000000000..b1caa73a0
--- /dev/null
+++ b/fodc/agent/internal/ktm/ktm.go
@@ -0,0 +1,105 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 ktm is a module that monitors I/O operations in the kernel.
+package ktm
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/rs/zerolog"
+
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/ktm/iomonitor"
+ fodcmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
+)
+
+// KTM represents the Kernel Trace Module.
+type KTM struct {
+ logger zerolog.Logger
+ collector *iomonitor.Collector
+ stopCh chan struct{}
+ config Config
+}
+
+// NewKTM creates a new KTM instance.
+func NewKTM(cfg Config, log zerolog.Logger) (*KTM, error) {
+ if !cfg.Enabled {
+ return &KTM{config: cfg, logger: log}, nil
+ }
+
+ // Convert KTM config to Collector config
+ collectorConfig := iomonitor.CollectorConfig{
+ Modules: cfg.Modules,
+ EBPF: cfg.EBPF,
+ Interval: cfg.Interval,
+ }
+
+ col, err := iomonitor.New(collectorConfig, log)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create collector: %w", err)
+ }
+
+ return &KTM{
+ config: cfg,
+ logger: log,
+ collector: col,
+ stopCh: make(chan struct{}),
+ }, nil
+}
+
+// Start starts the KTM module.
+func (k *KTM) Start(ctx context.Context) error {
+ if !k.config.Enabled {
+ k.logger.Info().Msg("KTM module is disabled")
+ return nil
+ }
+
+ k.logger.Info().Msg("Starting KTM module")
+ if err := k.collector.Start(ctx); err != nil {
+ return fmt.Errorf("failed to start collector: %w", err)
+ }
+
+ return nil
+}
+
+// Stop stops the KTM module.
+func (k *KTM) Stop() error {
+ if !k.config.Enabled {
+ return nil
+ }
+
+ k.logger.Info().Msg("Stopping KTM module")
+ close(k.stopCh)
+ return k.collector.Close()
+}
+
+// GetMetrics returns the collected metrics.
+func (k *KTM) GetMetrics() []fodcmetrics.RawMetric {
+ if !k.config.Enabled || k.collector == nil {
+ return nil
+ }
+ return k.collector.GetMetrics()
+}
+
+// IsDegraded returns whether KTM is running in degraded (comm-only) mode.
+func (k *KTM) IsDegraded() bool {
+ if !k.config.Enabled || k.collector == nil {
+ return false
+ }
+ return k.collector.IsDegraded()
+}
diff --git a/fodc/agent/internal/server/server.go b/fodc/agent/internal/server/server.go
index f9bfec973..9be7d52b3 100644
--- a/fodc/agent/internal/server/server.go
+++ b/fodc/agent/internal/server/server.go
@@ -110,14 +110,14 @@ func (s *Server) Start(registry *prometheus.Registry, datasourceCollector *expor
// Stop gracefully stops the Prometheus metrics server.
// It waits for ongoing requests to complete within the configured shutdown timeout.
-func (s *Server) Stop() error {
+func (s *Server) Stop(ctx context.Context) error {
if !s.started {
return fmt.Errorf("server is not started")
}
s.logger.Info().Msg("Stopping Prometheus metrics server")
- shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), s.config.ShutdownTimeout)
+ shutdownCtx, shutdownCancel := context.WithTimeout(ctx, s.config.ShutdownTimeout)
defer shutdownCancel()
if shutdownErr := s.server.Shutdown(shutdownCtx); shutdownErr != nil {
diff --git a/fodc/agent/internal/server/server_test.go b/fodc/agent/internal/server/server_test.go
index 0b7716cb6..9a5968136 100644
--- a/fodc/agent/internal/server/server_test.go
+++ b/fodc/agent/internal/server/server_test.go
@@ -18,6 +18,7 @@
package server
import (
+ "context"
"testing"
"time"
@@ -96,7 +97,7 @@ func TestServer_StartStop(t *testing.T) {
time.Sleep(100 * time.Millisecond)
// Stop the server
- stopErr := server.Stop()
+ stopErr := server.Stop(context.Background())
assert.NoError(t, stopErr)
assert.False(t, server.IsStarted())
@@ -133,7 +134,7 @@ func TestServer_Start_AlreadyStarted(t *testing.T) {
assert.Contains(t, startErr2.Error(), "server is already started")
// Clean up
- server.Stop()
+ server.Stop(context.Background())
}
func TestServer_Stop_NotStarted(t *testing.T) {
@@ -146,7 +147,7 @@ func TestServer_Stop_NotStarted(t *testing.T) {
assert.False(t, server.IsStarted())
// Try to stop without starting - should fail
- stopErr := server.Stop()
+ stopErr := server.Stop(context.Background())
assert.Error(t, stopErr)
assert.Contains(t, stopErr.Error(), "server is not started")
}
@@ -170,7 +171,7 @@ func TestServer_Start_RegistryAlreadyHasCollector(t *testing.T) {
assert.True(t, server.IsStarted())
// Stop the server
- require.NoError(t, server.Stop())
+ require.NoError(t, server.Stop(context.Background()))
assert.False(t, server.IsStarted())
// Check that no errors were sent on the error channel
@@ -215,7 +216,7 @@ func TestServer_IsStarted(t *testing.T) {
assert.True(t, server.IsStarted())
// Stop the server
- stopErr := server.Stop()
+ stopErr := server.Stop(context.Background())
require.NoError(t, stopErr)
assert.False(t, server.IsStarted())
}
diff --git a/fodc/proxy/internal/api/server.go b/fodc/proxy/internal/api/server.go
index 545347878..7f3af5d5e 100644
--- a/fodc/proxy/internal/api/server.go
+++ b/fodc/proxy/internal/api/server.go
@@ -254,6 +254,8 @@ func (s *Server) formatPrometheusText(aggregatedMetrics []*metrics.AggregatedMet
}
metricMap := make(map[string]*metricGroup)
+ histogramBases := make(map[string]bool)
+
for _, metric := range aggregatedMetrics {
key := metric.Name
group, exists := metricMap[key]
@@ -266,17 +268,73 @@ func (s *Server) formatPrometheusText(aggregatedMetrics []*metrics.AggregatedMet
metricMap[key] = group
}
group.metrics = append(group.metrics, metric)
+
+ if strings.HasSuffix(metric.Name, "_bucket") ||
+ strings.HasSuffix(metric.Name, "_sum") ||
+ strings.HasSuffix(metric.Name, "_count") {
+ baseName := getHistogramBaseName(metric.Name)
+ if baseName != "" {
+ histogramBases[baseName] = true
+ }
+ }
+ }
+
+ histogramMetrics := make(map[string][]*metrics.AggregatedMetric)
+ regularMetrics := make(map[string]*metricGroup)
+
+ for name, group := range metricMap {
+ baseName := getHistogramBaseName(name)
+ if baseName != "" && histogramBases[baseName] {
+ histogramMetrics[baseName] = append(histogramMetrics[baseName], group.metrics...)
+ } else {
+ regularMetrics[name] = group
+ }
}
var builder strings.Builder
- metricNames := make([]string, 0, len(metricMap))
- for name := range metricMap {
- metricNames = append(metricNames, name)
+
+ histogramNames := make([]string, 0, len(histogramBases))
+ for baseName := range histogramBases {
+ histogramNames = append(histogramNames, baseName)
+ }
+ sort.Strings(histogramNames)
+
+ for _, baseName := range histogramNames {
+ allMetrics := histogramMetrics[baseName]
+ if len(allMetrics) == 0 {
+ continue
+ }
+
+ description := allMetrics[0].Description
+ if description != "" {
+ builder.WriteString(fmt.Sprintf("# HELP %s %s\n", baseName, description))
+ }
+ builder.WriteString(fmt.Sprintf("# TYPE %s histogram\n", baseName))
+
+ for _, metric := range allMetrics {
+ labelParts := make([]string, 0, len(metric.Labels))
+ for key, value := range metric.Labels {
+ labelParts = append(labelParts, fmt.Sprintf(`%s="%s"`, key, value))
+ }
+ sort.Strings(labelParts)
+
+ labelStr := ""
+ if len(labelParts) > 0 {
+ labelStr = "{" + strings.Join(labelParts, ",") + "}"
+ }
+
+ builder.WriteString(fmt.Sprintf("%s%s %s\n", metric.Name, labelStr, formatFloat(metric.Value)))
+ }
+ }
+
+ regularNames := make([]string, 0, len(regularMetrics))
+ for name := range regularMetrics {
+ regularNames = append(regularNames, name)
}
- sort.Strings(metricNames)
+ sort.Strings(regularNames)
- for _, name := range metricNames {
- group := metricMap[name]
+ for _, name := range regularNames {
+ group := regularMetrics[name]
if group.description != "" {
builder.WriteString(fmt.Sprintf("# HELP %s %s\n", group.name, group.description))
}
@@ -382,6 +440,21 @@ func formatFloat(value float64) string {
return strconv.FormatFloat(value, 'f', -1, 64)
}
+// getHistogramBaseName extracts the base name from histogram metric names.
+// Returns empty string if the metric is not a histogram component.
+func getHistogramBaseName(metricName string) string {
+ if strings.HasSuffix(metricName, "_bucket") {
+ return strings.TrimSuffix(metricName, "_bucket")
+ }
+ if strings.HasSuffix(metricName, "_sum") {
+ return strings.TrimSuffix(metricName, "_sum")
+ }
+ if strings.HasSuffix(metricName, "_count") {
+ return strings.TrimSuffix(metricName, "_count")
+ }
+ return ""
+}
+
type metricGroup struct {
name string
description string
diff --git a/fodc/proxy/internal/api/server_test.go b/fodc/proxy/internal/api/server_test.go
index 4ab25793b..113d6c857 100644
--- a/fodc/proxy/internal/api/server_test.go
+++ b/fodc/proxy/internal/api/server_test.go
@@ -22,6 +22,8 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
+ "net/url"
+ "strings"
"testing"
"time"
@@ -63,7 +65,6 @@ func (m *mockClusterDataRequester) RequestClusterData(agentID string) error {
if m.requestErr != nil {
return m.requestErr
}
- // Simulate agent responding with topology
if topology, exists := m.topologies[agentID]; exists && m.clusterMgr != nil {
m.clusterMgr.UpdateClusterTopology(agentID, topology)
}
@@ -77,7 +78,7 @@ func (m *mockClusterDataRequester) setTopology(agentID string, topology *fodcv1.
m.topologies[agentID] = topology
}
-func newTestServer(t *testing.T) (*Server, *registry.AgentRegistry, *cluster.Manager, *mockClusterDataRequester) {
+func newTestServer(t *testing.T) (*Server, *registry.AgentRegistry, *mockClusterDataRequester) {
t.Helper()
initTestLogger(t)
testLogger := logger.GetLogger("test", "api")
@@ -88,11 +89,33 @@ func newTestServer(t *testing.T) (*Server, *registry.AgentRegistry, *cluster.Man
clusterMgr := cluster.NewManager(testRegistry, mockRequester, testLogger)
mockRequester.clusterMgr = clusterMgr
server := NewServer(aggregator, clusterMgr, testRegistry, testLogger)
- return server, testRegistry, clusterMgr, mockRequester
+ return server, testRegistry, mockRequester
+}
+
+func newTestServerOnly(t *testing.T) *Server {
+ t.Helper()
+ server, testRegistry, mockRequester := newTestServer(t)
+ _ = testRegistry
+ _ = mockRequester
+ return server
+}
+
+func createTestMetric(name string, value float64, labels map[string]string, timestamp time.Time) *metrics.AggregatedMetric {
+ if labels == nil {
+ labels = make(map[string]string)
+ }
+ return &metrics.AggregatedMetric{
+ Name: name,
+ Value: value,
+ Labels: labels,
+ Timestamp: timestamp,
+ AgentID: "test-agent-1",
+ Description: "Test metric description",
+ }
}
func TestHandleClusterTopology_Success(t *testing.T) {
- server, testRegistry, _, mockRequester := newTestServer(t)
+ server, testRegistry, mockRequester := newTestServer(t)
ctx := context.Background()
identity := registry.AgentIdentity{
PodName: "test-pod",
@@ -170,3 +193,503 @@ func TestHandleClusterTopology_RequestClusterDataError(t *testing.T) {
server.handleClusterTopology(resp, req)
assert.Equal(t, http.StatusOK, resp.Code)
}
+
+func TestNewServer(t *testing.T) {
+ initTestLogger(t)
+ testLogger := logger.GetLogger("test", "api")
+ testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
+ mockSender := &mockRequestSender{}
+ aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
+
+ server := NewServer(aggregator, nil, testRegistry, testLogger)
+
+ assert.NotNil(t, server)
+ assert.Equal(t, aggregator, server.metricsAggregator)
+ assert.Equal(t, testRegistry, server.registry)
+ assert.WithinDuration(t, time.Now(), server.startTime, time.Second)
+}
+
+func TestServer_StartStop(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ err := server.Start("localhost:0", 5*time.Second, 5*time.Second)
+ require.NoError(t, err)
+ assert.NotNil(t, server.server)
+
+ time.Sleep(50 * time.Millisecond)
+
+ stopErr := server.Stop()
+ assert.NoError(t, stopErr)
+}
+
+func TestServer_Stop_NotStarted(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ err := server.Stop()
+ assert.NoError(t, err)
+}
+
+func TestHandleMetrics_Success(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetrics(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+ assert.Contains(t, w.Header().Get("Content-Type"), "text/plain")
+}
+
+func TestHandleMetrics_MethodNotAllowed(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodPost, "/metrics", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetrics(w, req)
+
+ assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
+}
+
+func TestHandleMetrics_WithRoleFilter(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodGet, "/metrics?role=datanode-cold", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetrics(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+}
+
+func TestHandleMetrics_WithAddressFilter(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodGet, "/metrics?address=192.168.1.1", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetrics(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+}
+
+func TestHandleMetrics_CollectionError(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ req = req.WithContext(ctx)
+ w := httptest.NewRecorder()
+
+ server.handleMetrics(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+}
+
+func TestHandleMetricsWindows_Success(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)
+ endTime := now.Format(time.RFC3339)
+
+ u, urlErr := url.Parse("/metrics-windows")
+ require.NoError(t, urlErr)
+ q := u.Query()
+ q.Set("start_time", startTime)
+ q.Set("end_time", endTime)
+ u.RawQuery = q.Encode()
+
+ req := httptest.NewRequest(http.MethodGet, u.String(), nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+ assert.Equal(t, "application/json", w.Header().Get("Content-Type"))
+
+ var response []map[string]interface{}
+ decodeErr := json.NewDecoder(w.Body).Decode(&response)
+ require.NoError(t, decodeErr)
+}
+
+func TestHandleMetricsWindows_InvalidStartTime(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ u, urlErr := url.Parse("/metrics-windows")
+ require.NoError(t, urlErr)
+ q := u.Query()
+ q.Set("start_time", "invalid")
+ u.RawQuery = q.Encode()
+
+ req := httptest.NewRequest(http.MethodGet, u.String(), nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusBadRequest, w.Code)
+ assert.Contains(t, w.Body.String(), "Invalid start_time format")
+}
+
+func TestHandleMetricsWindows_InvalidEndTime(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)
+
+ u, urlErr := url.Parse("/metrics-windows")
+ require.NoError(t, urlErr)
+ q := u.Query()
+ q.Set("start_time", startTime)
+ q.Set("end_time", "invalid")
+ u.RawQuery = q.Encode()
+
+ req := httptest.NewRequest(http.MethodGet, u.String(), nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusBadRequest, w.Code)
+ assert.Contains(t, w.Body.String(), "Invalid end_time format")
+}
+
+func TestHandleMetricsWindows_MethodNotAllowed(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodPost, "/metrics-windows", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
+}
+
+func TestHandleMetricsWindows_WithoutTimeWindow(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodGet, "/metrics-windows", nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+ var response []map[string]interface{}
+ decodeErr := json.NewDecoder(w.Body).Decode(&response)
+ require.NoError(t, decodeErr)
+}
+
+func TestHandleHealth_Success(t *testing.T) {
+ server, testRegistry, _ := newTestServer(t)
+
+ ctx := context.Background()
+ identity1 := registry.AgentIdentity{PodName: "192.168.1.1", Role: "datanode-cold", ContainerNames: []string{"data"}}
+ identity2 := registry.AgentIdentity{PodName: "192.168.1.2", Role: "liaison", ContainerNames: []string{"liaison"}}
+
+ _, err1 := testRegistry.RegisterAgent(ctx, identity1)
+ require.NoError(t, err1)
+ _, err2 := testRegistry.RegisterAgent(ctx, identity2)
+ require.NoError(t, err2)
+
+ req := httptest.NewRequest(http.MethodGet, "/health", nil)
+ w := httptest.NewRecorder()
+
+ server.handleHealth(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+ assert.Equal(t, "application/json", w.Header().Get("Content-Type"))
+
+ var response map[string]interface{}
+ decodeErr := json.NewDecoder(w.Body).Decode(&response)
+ require.NoError(t, decodeErr)
+ assert.Equal(t, "healthy", response["status"])
+ assert.Equal(t, float64(2), response["agents_total"])
+ assert.GreaterOrEqual(t, response["uptime_seconds"], float64(0))
+}
+
+func TestHandleHealth_MethodNotAllowed(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ req := httptest.NewRequest(http.MethodPost, "/health", nil)
+ w := httptest.NewRecorder()
+
+ server.handleHealth(w, req)
+
+ assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
+}
+
+func TestFormatPrometheusText_Empty(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ result := server.formatPrometheusText([]*metrics.AggregatedMetric{})
+
+ assert.Empty(t, result)
+}
+
+func TestFormatPrometheusText_SingleMetric(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ assert.Contains(t, result, "# HELP cpu_usage")
+ assert.Contains(t, result, "# TYPE cpu_usage gauge")
+ assert.Contains(t, result, "cpu_usage")
+ assert.Contains(t, result, "75.5")
+}
+
+func TestFormatPrometheusText_MultipleMetrics(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
+ createTestMetric("cpu_usage", 80.0, map[string]string{"cpu": "1", "pod_name": "192.168.1.1"}, now),
+ createTestMetric("memory_usage", 50.0, map[string]string{"type": "heap", "pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ assert.NotEmpty(t, result)
+ assert.Contains(t, result, "cpu_usage")
+ assert.Contains(t, result, "memory_usage")
+}
+
+func TestFormatPrometheusText_WithLabels(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "env": "test", "pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ assert.Contains(t, result, "cpu_usage{")
+ assert.Contains(t, result, "cpu=\"0\"")
+ assert.Contains(t, result, "env=\"test\"")
+}
+
+func TestFormatPrometheusText_Histogram(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("request_latency_bucket", 10, map[string]string{"le": "0.1", "ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_bucket", 25, map[string]string{"le": "0.5", "ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_bucket", 30, map[string]string{"le": "+Inf", "ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_sum", 12.5, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_count", 30, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ assert.Contains(t, result, "# TYPE request_latency histogram")
+ assert.Contains(t, result, "request_latency_bucket{")
+ assert.Contains(t, result, "request_latency_sum{")
+ assert.Contains(t, result, "request_latency_count{")
+ assert.Contains(t, result, "le=\"0.1\"")
+ assert.Contains(t, result, "le=\"0.5\"")
+ assert.Contains(t, result, "le=\"+Inf\"")
+ assert.NotContains(t, result, "# TYPE request_latency_bucket")
+ assert.NotContains(t, result, "# TYPE request_latency_sum")
+ assert.NotContains(t, result, "# TYPE request_latency_count")
+}
+
+func TestFormatPrometheusText_HistogramAndGauge(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("request_latency_bucket", 10, map[string]string{"le": "0.1", "ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_sum", 12.5, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("request_latency_count", 30, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "ip": "192.168.1.1", "port": "8080"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ assert.Contains(t, result, "# TYPE request_latency histogram")
+ assert.Contains(t, result, "# TYPE cpu_usage gauge")
+ assert.Contains(t, result, "request_latency_bucket{")
+ assert.Contains(t, result, "cpu_usage{")
+}
+
+func TestGetHistogramBaseName(t *testing.T) {
+ tests := []struct {
+ name string
+ input string
+ expected string
+ }{
+ {"bucket suffix", "request_latency_bucket", "request_latency"},
+ {"sum suffix", "request_latency_sum", "request_latency"},
+ {"count suffix", "request_latency_count", "request_latency"},
+ {"no histogram suffix", "cpu_usage", ""},
+ {"empty string", "", ""},
+ {"only suffix", "_bucket", ""},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ result := getHistogramBaseName(tt.input)
+ assert.Equal(t, tt.expected, result)
+ })
+ }
+}
+
+func TestFormatMetricsWindowJSON_Empty(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ result := server.formatMetricsWindowJSON([]*metrics.AggregatedMetric{})
+
+ assert.Equal(t, 0, len(result))
+}
+
+func TestFormatMetricsWindowJSON_SingleMetric(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatMetricsWindowJSON(metricsList)
+
+ assert.Equal(t, 1, len(result))
+ item := result[0]
+ assert.Equal(t, "cpu_usage", item["name"])
+ assert.Equal(t, "test-agent-1", item["agent_id"])
+ assert.Equal(t, "192.168.1.1", item["pod_name"])
+ assert.Contains(t, item, "data")
+}
+
+func TestFormatMetricsWindowJSON_MultipleMetrics(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
+ createTestMetric("memory_usage", 50.0, map[string]string{"type": "heap", "pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatMetricsWindowJSON(metricsList)
+
+ assert.Equal(t, 2, len(result))
+ metricNames := make(map[string]bool)
+ for _, item := range result {
+ metricNames[item["name"].(string)] = true
+ }
+ assert.True(t, metricNames["cpu_usage"])
+ assert.True(t, metricNames["memory_usage"])
+}
+
+func TestGetMetricKey(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ metric := createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "env": "test", "pod_name": "192.168.1.1"}, time.Now())
+
+ key := server.getMetricKey(metric)
+
+ assert.Contains(t, key, "cpu_usage")
+ assert.Contains(t, key, "test-agent-1")
+ assert.Contains(t, key, "cpu=0")
+ assert.Contains(t, key, "env=test")
+ assert.Contains(t, key, "pod_name=192.168.1.1")
+}
+
+func TestFormatFloat(t *testing.T) {
+ tests := []struct {
+ name string
+ expected string
+ value float64
+ }{
+ {"integer", "100", 100.0},
+ {"decimal", "75.5", 75.5},
+ {"small", "0.001", 0.001},
+ {"large", "1000000", 1000000.0},
+ {"negative", "-50.5", -50.5},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ result := formatFloat(tt.value)
+ assert.Equal(t, tt.expected, result)
+ })
+ }
+}
+
+func TestServer_Integration(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/metrics", server.handleMetrics)
+ mux.HandleFunc("/metrics-windows", server.handleMetricsWindows)
+ mux.HandleFunc("/health", server.handleHealth)
+
+ testServer := httptest.NewServer(mux)
+ defer testServer.Close()
+
+ client := &http.Client{Timeout: 500 * time.Millisecond}
+
+ healthResp, healthErr := client.Get(testServer.URL + "/health")
+ require.NoError(t, healthErr)
+ assert.Equal(t, http.StatusOK, healthResp.StatusCode)
+ healthResp.Body.Close()
+
+ metricsResp, metricsErr := client.Get(testServer.URL + "/metrics")
+ if metricsErr == nil {
+ assert.Equal(t, http.StatusOK, metricsResp.StatusCode)
+ metricsResp.Body.Close()
+ }
+}
+
+func TestHandleMetricsWindows_WithFilters(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)
+ endTime := now.Format(time.RFC3339)
+
+ u, urlErr := url.Parse("/metrics-windows")
+ require.NoError(t, urlErr)
+ q := u.Query()
+ q.Set("start_time", startTime)
+ q.Set("end_time", endTime)
+ q.Set("role", "datanode-cold")
+ q.Set("address", "192.168.1.1")
+ u.RawQuery = q.Encode()
+
+ req := httptest.NewRequest(http.MethodGet, u.String(), nil)
+ w := httptest.NewRecorder()
+
+ server.handleMetricsWindows(w, req)
+
+ assert.Equal(t, http.StatusOK, w.Code)
+ var response []map[string]interface{}
+ decodeErr := json.NewDecoder(w.Body).Decode(&response)
+ require.NoError(t, decodeErr)
+}
+
+func TestFormatPrometheusText_SortedMetrics(t *testing.T) {
+ server := newTestServerOnly(t)
+
+ now := time.Now()
+ metricsList := []*metrics.AggregatedMetric{
+ createTestMetric("z_metric", 1.0, map[string]string{"pod_name": "192.168.1.1"}, now),
+ createTestMetric("a_metric", 2.0, map[string]string{"pod_name": "192.168.1.1"}, now),
+ createTestMetric("m_metric", 3.0, map[string]string{"pod_name": "192.168.1.1"}, now),
+ }
+
+ result := server.formatPrometheusText(metricsList)
+
+ aIndex := strings.Index(result, "a_metric")
+ mIndex := strings.Index(result, "m_metric")
+ zIndex := strings.Index(result, "z_metric")
+
+ assert.True(t, aIndex < mIndex && mIndex < zIndex, "Metrics should be sorted alphabetically")
+}
diff --git a/go.mod b/go.mod
index 0c6d646fb..c03ebd8a2 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.20.0
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.3
github.com/RoaringBitmap/roaring v1.9.4
+ github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff
github.com/alecthomas/participle/v2 v2.1.4
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe
github.com/aws/aws-sdk-go-v2 v1.40.0
@@ -15,6 +16,7 @@ require (
github.com/benbjohnson/clock v1.3.5
github.com/blugelabs/bluge v0.2.2
github.com/cespare/xxhash/v2 v2.3.0
+ github.com/cilium/ebpf v0.19.0
github.com/emirpasic/gods v1.18.1
github.com/envoyproxy/protoc-gen-validate v1.2.1
github.com/go-chi/chi/v5 v5.2.3
diff --git a/go.sum b/go.sum
index fb6bc7203..b9ed85b60 100644
--- a/go.sum
+++ b/go.sum
@@ -59,6 +59,8 @@ github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01 h1:f56Ybipl9Yh9wFCZc3
github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01/go.mod h1:VfQRtJQEwpooobYuSBadb8QYjTUICH+CX42IJYrfSiM=
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3 h1:bZvLPihpC1q1AJkmR6ere/aJOr3ev+JCytvjKuL+gE8=
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3/go.mod h1:DoQeb0Ee86LyruZSL77Ddscfk/THJ38x453CRCnGEPI=
+github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff h1:pjcdkEAe+NL2bA0J/mu/Vs+eG+w+KN/BX9yGkOC1f60=
+github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff/go.mod h1:oUklUxzpQkBmJeQcA6ObchxOUcUYw0Gl80NLJzIqVrk=
github.com/VictoriaMetrics/fastcache v1.13.2 h1:2XTB49aLSuCex7e9P5rqrfQcMkzGjh5Vq3GMFa8YpCA=
github.com/VictoriaMetrics/fastcache v1.13.2/go.mod h1:hHXhl4DA2fTL2HTZDJFXWgW0LNjo6B+4aj2Wmng3TjU=
github.com/alecthomas/assert/v2 v2.11.0 h1:2Q9r3ki8+JYXvGsDyBXwH3LcJ+WK5D0gc5E8vS6K3D0=
@@ -137,6 +139,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.19.0 h1:Ro/rE64RmFBeA9FGjcTc+KmCeY6jXmryu6FfnzPRIao=
+github.com/cilium/ebpf v0.19.0/go.mod h1:fLCgMo3l8tZmAdM3B2XqdFzXBpwkcSTroaVqN08OWVY=
github.com/cncf/xds/go v0.0.0-20251110193048-8bfbf64dc13e h1:gt7U1Igw0xbJdyaCM5H2CnlAlPSkzrhsebQB6WQWjLA=
github.com/cncf/xds/go v0.0.0-20251110193048-8bfbf64dc13e/go.mod h1:KdCmV+x/BuvyMxRnYBlmVaq4OLiKW6iRQfvC62cvdkI=
github.com/cockroachdb/datadriven v1.0.2 h1:H9MtNqVoVhvd9nCBwOyDjUEdZCREqbIdCJD93PBm/jA=
@@ -209,6 +213,8 @@ github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
+github.com/go-quicktest/qt v1.101.1-0.20240301121107-c6c8733fa1e6 h1:teYtXy9B7y5lHTp8V9KPxpYRAVA7dozigQcMiBust1s=
+github.com/go-quicktest/qt v1.101.1-0.20240301121107-c6c8733fa1e6/go.mod h1:p4lGIVX+8Wa6ZPNDvqcxq36XpUDLh42FLetFU7odllI=
github.com/go-resty/resty/v2 v2.16.5 h1:hBKqmWrr7uRc3euHVqmh1HTHcKn99Smr7o5spptdhTM=
github.com/go-resty/resty/v2 v2.16.5/go.mod h1:hkJtXbA2iKHzJheXYvQ8snQES5ZLGKMwQ07xAwp/fiA=
github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
@@ -263,8 +269,12 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbdFz6I=
github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60=
+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/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE=
github.com/joshdk/go-junit v1.0.0/go.mod h1:TiiV0PqkaNfFXjEiyjWM3XXrhVyCa1K4Zfga6W52ung=
+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/kamstrup/intmap v0.5.1 h1:ENGAowczZA+PJPYYlreoqJvWgQVtAmX1l899WfYFVK0=
github.com/kamstrup/intmap v0.5.1/go.mod h1:gWUVWHKzWj8xpJVFf5GC0O26bWmv3GqdnIX/LMT6Aq4=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
@@ -301,6 +311,10 @@ github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+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/mfridman/tparse v0.18.0 h1:wh6dzOKaIwkUGyKgOntDW4liXSo37qg5AXbIhkMV3vE=
github.com/mfridman/tparse v0.18.0/go.mod h1:gEvqZTuCgEhPbYk/2lS3Kcxg1GmTxxU7kTC8DvP0i/A=
github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI=
diff --git a/pkg/test/flags/flags.go b/pkg/test/flags/flags.go
index a870d66fe..eb306f97b 100644
--- a/pkg/test/flags/flags.go
+++ b/pkg/test/flags/flags.go
@@ -19,6 +19,7 @@
package flags
import (
+ "strconv"
"time"
)
@@ -29,6 +30,8 @@ var (
consistentlyTimeout string
+ requireKTM string
+
// EventuallyTimeout is the timeout of async time cases execution.
EventuallyTimeout time.Duration
@@ -40,6 +43,9 @@ var (
// LogLevel is the log level of test cases.
LogLevel = "debug"
+
+ // RequireKTM indicates whether KTM integration tests must run and should fail instead of skipping when requirements are missing.
+ RequireKTM bool
)
func init() {
@@ -52,6 +58,14 @@ func init() {
if consistentlyTimeout == "" {
consistentlyTimeout = "5s"
}
+ if requireKTM == "" {
+ requireKTM = "false"
+ }
+ requireKTMParsed, requireKTMParseErr := strconv.ParseBool(requireKTM)
+ if requireKTMParseErr != nil {
+ panic(requireKTMParseErr)
+ }
+ RequireKTM = requireKTMParsed
d, err := time.ParseDuration(eventuallyTimeout)
if err != nil {
panic(err)
diff --git a/fodc/agent/Makefile b/scripts/build/bpf2go.mk
similarity index 54%
copy from fodc/agent/Makefile
copy to scripts/build/bpf2go.mk
index 8a5672a26..dea4d33bc 100644
--- a/fodc/agent/Makefile
+++ b/scripts/build/bpf2go.mk
@@ -16,22 +16,22 @@
# under the License.
#
-NAME := fodc
-BINARIES := $(NAME)-agent
+BPF2GO := $(tool_bin)/bpf2go
+IS_LINUX := $(shell uname -s | grep -i linux)
-IMG_NAME := skywalking-banyandb-fodc-agent
-
-# Override SOURCE_DIR to match the actual Go module path structure
-SOURCE_DIR := fodc/agent
-
-include ../../scripts/build/version.mk
-include ../../scripts/build/base.mk
-include ../../scripts/build/generate_go.mk
-include ../../scripts/build/build.mk
-include ../../scripts/build/docker.mk
-include ../../scripts/build/test.mk
-include ../../scripts/build/lint.mk
-include ../../scripts/build/vendor.mk
-include ../../scripts/build/help.mk
-
-prepare-build: generate
+$(BPF2GO):
+ @echo "Installing bpf2go..."
+ @mkdir -p $(tool_bin)
+ @GOBIN=$(tool_bin) go install github.com/cilium/ebpf/cmd/bpf2go@$(BPF2GO_VERSION)
+ @if [ ! -f "$(BPF2GO)" ]; then \
+ echo "WARNING: Failed to install bpf2go at $(BPF2GO)"; \
+ if [ "$(IS_LINUX)" = "Linux" ]; then \
+ echo "ERROR: bpf2go is required on Linux"; \
+ exit 1; \
+ fi; \
+ echo "Creating a stub bpf2go executable for non-Linux platform"; \
+ echo '#!/bin/sh' > $(BPF2GO); \
+ echo 'echo "bpf2go stub: eBPF code generation only works on Linux"' >> $(BPF2GO); \
+ echo 'exit 0' >> $(BPF2GO); \
+ chmod +x $(BPF2GO); \
+ fi
\ No newline at end of file
diff --git a/fodc/agent/Makefile b/scripts/build/bpftool.mk
similarity index 56%
copy from fodc/agent/Makefile
copy to scripts/build/bpftool.mk
index 8a5672a26..afe438496 100644
--- a/fodc/agent/Makefile
+++ b/scripts/build/bpftool.mk
@@ -16,22 +16,18 @@
# under the License.
#
-NAME := fodc
-BINARIES := $(NAME)-agent
+# Define the bpftool location - use 'which bpftool' if available on system
+BPFTL := $(shell which bpftool 2>/dev/null)
+IS_LINUX := $(shell uname -s | grep -i linux)
-IMG_NAME := skywalking-banyandb-fodc-agent
-
-# Override SOURCE_DIR to match the actual Go module path structure
-SOURCE_DIR := fodc/agent
-
-include ../../scripts/build/version.mk
-include ../../scripts/build/base.mk
-include ../../scripts/build/generate_go.mk
-include ../../scripts/build/build.mk
-include ../../scripts/build/docker.mk
-include ../../scripts/build/test.mk
-include ../../scripts/build/lint.mk
-include ../../scripts/build/vendor.mk
-include ../../scripts/build/help.mk
-
-prepare-build: generate
+$(BPFTL):
+ifeq ($(IS_LINUX),Linux)
+ @echo "Installing bpftool (Linux detected)..."
+ @sudo apt-get update && sudo apt-get install -y bpftool clang llvm libelf-dev
+ @echo "Verifying bpftool installation..."
+ @which bpftool || (echo "Failed to install bpftool" && exit 1)
+ $(eval BPFTL := $(shell which bpftool))
+else
+ @echo "Non-Linux OS detected, switching to Docker for eBPF generation..."
+ @echo "Please ensure Docker is installed and running"
+endif
diff --git a/scripts/build/version.mk b/scripts/build/version.mk
index 6c741d185..d6e9626d1 100644
--- a/scripts/build/version.mk
+++ b/scripts/build/version.mk
@@ -32,3 +32,7 @@ MOCKGEN_VERSION := v0.6.0
GINKGO_VERSION := v2.27.2
GOVULNCHECK_VERSION := v1.1.4
+
+# Version for bpf2go tool used for eBPF code generation. Keep in sync with
+# pkg/fs/fadvismonitor/Dockerfile ARG BPF2GO_VERSION
+BPF2GO_VERSION := v0.13.0
\ No newline at end of file
diff --git a/test/fodc/README.md b/test/fodc/README.md
new file mode 100644
index 000000000..6d0364d4d
--- /dev/null
+++ b/test/fodc/README.md
@@ -0,0 +1,99 @@
+# FODC KTM Integration Test (Kind)
+
+This test boots a Kind cluster, runs BanyanDB with a FODC sidecar that has KTM enabled, and verifies KTM metrics on the FODC `/metrics` endpoint.
+
+## Prerequisites
+- `kind`, `kubectl`, and `docker` installed.
+- eBPF build dependencies for the FODC agent: `clang`, `llvm-strip`, `bpftool`, and a readable `/sys/kernel/btf/vmlinux` on the host.
+
+## Steps
+1. Create the Kind cluster.
+
+```bash
+kind create cluster --name banyand-ktm --config kind.yaml
+```
+
+2. Build images.
+
+```bash
+make -C banyand release
+make -C banyand docker
+make -C fodc/agent release
+make -C fodc/agent docker
+```
+
+3. Load images into Kind.
+
+```bash
+kind load docker-image apache/skywalking-banyandb:latest --name banyand-ktm
+kind load docker-image apache/skywalking-banyandb-fodc-agent:latest --name banyand-ktm
+```
+
+4. Deploy the test pod.
+
+```bash
+kubectl apply -f pod.yaml
+```
+
+If your CI runner uses an older kernel and KTM fails to load eBPF programs, use the fallback manifest that includes `CAP_SYS_ADMIN`:
+
+```bash
+kubectl apply -f pod-sysadmin.yaml
+```
+
+5. Start the write-load generator (background) to produce I/O on BanyanDB so KTM metrics become non-zero.
+
+```bash
+./write-load.sh &
+```
+
+6. Verify KTM metrics from the FODC agent.
+
+```bash
+./check.sh
+```
+
+The check script waits until `ktm_status >= 1` (degraded comm-only or full cgroup+comm mode), all required metric names are present, **and** at least one histogram sum (`ktm_sys_read_latency_seconds_sum` or `ktm_sys_pread_latency_seconds_sum`) is > 0, proving that eBPF tracepoints are actually measuring I/O latency.
+
+> **Note:** The pod manifests intentionally omit the `/sys/fs/cgroup` mount. This forces KTM into comm-only (degraded) mode, which avoids a known cgroup ID mismatch between `bpf_get_current_cgroup_id()` (leaf container cgroup) and the pod-level cgroup configured by the Go code.
+
+If you want a CI fallback from minimal permissions to the sysadmin manifest, you can use the following pattern:
+
+```bash
+set -euo pipefail
+
+kubectl apply -f pod.yaml
+./write-load.sh &
+LOAD_PID=$!
+set +e
+./check.sh
+status=$?
+set -e
+
+if [ "$status" -ne 0 ]; then
+ kill "$LOAD_PID" 2>/dev/null || true
+ kubectl delete -f pod.yaml
+ kubectl apply -f pod-sysadmin.yaml
+ ./write-load.sh &
+ LOAD_PID=$!
+ ./check.sh
+fi
+kill "$LOAD_PID" 2>/dev/null || true
+```
+
+## Cleanup
+```bash
+kubectl delete -f pod.yaml
+kubectl delete -f pod-sysadmin.yaml
+kind delete cluster --name banyand-ktm
+```
+
+## Troubleshooting
+- If `ktm_status` is `0`, the KTM module did not start. Check that the FODC container has `CAP_BPF`, `CAP_PERFMON` (or `CAP_SYS_ADMIN` on older kernels),
+ and `CAP_SYS_RESOURCE`, and that the host kernel exposes `/sys/kernel/btf/vmlinux`.
+- If KTM fails to attach tracepoints, ensure tracefs is mounted into the container. The pod manifests mount `/sys/kernel/tracing` and `/sys/kernel/debug` from the host.
+- Inspect logs for startup errors.
+
+```bash
+kubectl logs banyand-fodc-ktm -c fodc-agent
+```
diff --git a/test/fodc/check.sh b/test/fodc/check.sh
new file mode 100755
index 000000000..12b4ce7cd
--- /dev/null
+++ b/test/fodc/check.sh
@@ -0,0 +1,151 @@
+#!/bin/bash
+
+# Licensed to Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+set -euo pipefail
+
+POD_NAME=${POD_NAME:-banyand-fodc-ktm}
+CONTAINER_NAME=${CONTAINER_NAME:-fodc-agent}
+METRICS_URL=${METRICS_URL:-http://127.0.0.1:9090/metrics}
+WAIT_TIMEOUT=${WAIT_TIMEOUT:-300s}
+MAX_WAIT_SECONDS=${MAX_WAIT_SECONDS:-120}
+SLEEP_SECONDS=${SLEEP_SECONDS:-2}
+PRINT_KTM_ON_SUCCESS=${PRINT_KTM_ON_SUCCESS:-true}
+PRINT_KTM_SAMPLES_ON_SUCCESS=${PRINT_KTM_SAMPLES_ON_SUCCESS:-false}
+MAX_KTM_SAMPLE_LINES=${MAX_KTM_SAMPLE_LINES:-40}
+REQUIRED_KTM_METRIC_COUNT=${REQUIRED_KTM_METRIC_COUNT:-8}
+REQUIRED_KTM_METRICS=${REQUIRED_KTM_METRICS:-"ktm_status ktm_degraded \
+ktm_fadvise_calls_total ktm_cache_lookups_total \
+ktm_sys_read_latency_seconds_count ktm_sys_pread_latency_seconds_count \
+ktm_sys_read_latency_seconds_sum ktm_sys_pread_latency_seconds_sum"}
+FAILURE_LOG_TAIL_LINES=${FAILURE_LOG_TAIL_LINES:-200}
+NONZERO_KTM_METRICS=${NONZERO_KTM_METRICS:-"ktm_sys_read_latency_seconds_sum ktm_sys_pread_latency_seconds_sum"}
+
+echo "Preflight: Host kernel and tracing info"
+echo "Kernel: $(uname -r)"
+if [ -r /sys/kernel/btf/vmlinux ]; then
+ echo "BTF: /sys/kernel/btf/vmlinux is readable"
+else
+ echo "BTF: /sys/kernel/btf/vmlinux is NOT readable"
+fi
+if [ -d /sys/kernel/tracing ]; then
+ echo "Tracefs: /sys/kernel/tracing exists"
+else
+ echo "Tracefs: /sys/kernel/tracing is missing"
+fi
+
+kubectl wait --for=condition=Ready "pod/${POD_NAME}" --timeout="${WAIT_TIMEOUT}"
+
+deadline=$((SECONDS + MAX_WAIT_SECONDS))
+last_metrics=""
+ktm_status_inactive=""
+last_ktm_metric_count=0
+last_missing_metrics=""
+
+while [ "${SECONDS}" -lt "${deadline}" ]; do
+ metrics="$(kubectl exec "${POD_NAME}" -c "${CONTAINER_NAME}" -- sh -c "wget -qO- ${METRICS_URL}" || true)"
+ if [ -n "${metrics}" ]; then
+ last_metrics="${metrics}"
+ fi
+
+ ktm_metric_names="$(printf '%s\n' "${metrics}" | awk '/^ktm_[a-zA-Z0-9_]+([[:space:]]|\{)/ {name = $1; sub(/\{.*/, "", name); print name}' | sort -u)"
+ ktm_metric_count="$(printf '%s\n' "${ktm_metric_names}" | awk 'NF {count++} END {print count + 0}')"
+ last_ktm_metric_count="${ktm_metric_count}"
+
+ missing_metrics=""
+ for required_metric in ${REQUIRED_KTM_METRICS}; do
+ if ! printf '%s\n' "${ktm_metric_names}" | awk -v metric="${required_metric}" '$0 == metric {found = 1} END {exit found ? 0 : 1}'; then
+ missing_metrics="${missing_metrics} ${required_metric}"
+ fi
+ done
+ last_missing_metrics="$(echo "${missing_metrics}" | xargs)"
+
+ ktm_status="$(printf '%s\n' "${metrics}" | awk '/^ktm_status[[:space:]]/ {print $2; exit}')"
+ if [ -n "${ktm_status}" ]; then
+ if awk -v v="${ktm_status}" 'BEGIN {exit (v == 2.0) ? 0 : 1}'; then
+ if [ "${ktm_metric_count}" -ge "${REQUIRED_KTM_METRIC_COUNT}" ] && [ -z "${last_missing_metrics}" ]; then
+ # Verify at least one I/O metric has a value > 0
+ has_nonzero=false
+ for nz_metric in ${NONZERO_KTM_METRICS}; do
+ nz_val="$(printf '%s\n' "${metrics}" | awk -v m="${nz_metric}" '$0 ~ "^"m"[[:space:]]" {print $2; exit}')"
+ if [ -n "${nz_val}" ] && awk -v v="${nz_val}" 'BEGIN {exit (v > 0) ? 0 : 1}'; then
+ has_nonzero=true
+ break
+ fi
+ done
+ if ! ${has_nonzero}; then
+ sleep "${SLEEP_SECONDS}"
+ continue
+ fi
+ echo "KTM smoke check passed. ktm_status=${ktm_status}, ktm_metric_count=${ktm_metric_count}"
+ if [ "${PRINT_KTM_ON_SUCCESS}" = "true" ]; then
+ echo "::group::KTM success details"
+ echo "Detected unique ktm_* metric names:"
+ printf '%s\n' "${ktm_metric_names}"
+ if [ "${PRINT_KTM_SAMPLES_ON_SUCCESS}" = "true" ]; then
+ echo "KTM metric samples (first unique entries):"
+ printf '%s\n' "${metrics}" | awk -v max_lines="${MAX_KTM_SAMPLE_LINES}" '
+ /^ktm_[a-zA-Z0-9_]+([[:space:]]|\{)/ {
+ metric_name = $1
+ sub(/\{.*/, "", metric_name)
+ if (!(metric_name in shown)) {
+ print $0
+ shown[metric_name] = 1
+ printed++
+ }
+ if (printed >= max_lines) {
+ exit
+ }
+ }
+ '
+ fi
+ echo "::endgroup::"
+ fi
+ exit 0
+ fi
+ else
+ ktm_status_inactive="${ktm_status}"
+ fi
+ fi
+
+ sleep "${SLEEP_SECONDS}"
+done
+
+echo "KTM metrics check failed."
+if [ -n "${ktm_status_inactive}" ]; then
+ echo "Found ktm_status but KTM is not in strict mode (value ${ktm_status_inactive}, expected == 2)."
+else
+ echo "ktm_status metric not found."
+fi
+echo "Found unique ktm_* metrics: ${last_ktm_metric_count} (required >= ${REQUIRED_KTM_METRIC_COUNT})"
+if [ -n "${last_missing_metrics}" ]; then
+ echo "Missing required ktm metrics:${last_missing_metrics}"
+fi
+
+if [ -n "${last_metrics}" ]; then
+ echo "::group::KTM metrics (ktm_ lines only)"
+ printf '%s\n' "${last_metrics}" | awk '/^ktm_/ {print}'
+ echo "::endgroup::"
+ echo "::group::KTM metrics (full scrape)"
+ printf '%s\n' "${last_metrics}"
+ echo "::endgroup::"
+fi
+
+echo "::group::FODC logs (tail)"
+kubectl logs "${POD_NAME}" -c "${CONTAINER_NAME}" --tail="${FAILURE_LOG_TAIL_LINES}"
+echo "::endgroup::"
+
+exit 1
diff --git a/test/fodc/kind.yaml b/test/fodc/kind.yaml
new file mode 100644
index 000000000..1081c5157
--- /dev/null
+++ b/test/fodc/kind.yaml
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+ - role: control-plane
diff --git a/test/fodc/pod-sysadmin.yaml b/test/fodc/pod-sysadmin.yaml
new file mode 100644
index 000000000..c312a5f96
--- /dev/null
+++ b/test/fodc/pod-sysadmin.yaml
@@ -0,0 +1,79 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+apiVersion: v1
+kind: Pod
+metadata:
+ name: banyand-fodc-ktm
+ labels:
+ app: banyand-fodc-ktm
+spec:
+ shareProcessNamespace: true
+ containers:
+ - name: banyand
+ image: apache/skywalking-banyandb:latest
+ imagePullPolicy: IfNotPresent
+ args:
+ - standalone
+ ports:
+ - name: banyand-http
+ containerPort: 17913
+ - name: banyand-metrics
+ containerPort: 2121
+ - name: fodc-agent
+ image: apache/skywalking-banyandb-fodc-agent:latest
+ imagePullPolicy: IfNotPresent
+ args:
+ - --ktm-enabled
+ - --ktm-interval=2s
+ - --poll-metrics-interval=2s
+ - --poll-metrics-ports=2121
+ - --prometheus-listen-addr=:9090
+ ports:
+ - name: fodc-metrics
+ containerPort: 9090
+ volumeMounts:
+ - name: tracefs
+ mountPath: /sys/kernel/tracing
+ readOnly: true
+ - name: debugfs
+ mountPath: /sys/kernel/debug
+ readOnly: true
+ - name: cgroup
+ mountPath: /sys/fs/cgroup
+ readOnly: true
+ securityContext:
+ allowPrivilegeEscalation: true
+ seccompProfile:
+ type: Unconfined
+ capabilities:
+ add:
+ - BPF
+ - PERFMON
+ - SYS_ADMIN
+ - SYS_RESOURCE
+ volumes:
+ - name: tracefs
+ hostPath:
+ path: /sys/kernel/tracing
+ type: DirectoryOrCreate
+ - name: debugfs
+ hostPath:
+ path: /sys/kernel/debug
+ type: DirectoryOrCreate
+ - name: cgroup
+ hostPath:
+ path: /sys/fs/cgroup
+ type: DirectoryOrCreate
diff --git a/test/fodc/pod.yaml b/test/fodc/pod.yaml
new file mode 100644
index 000000000..4e1054aad
--- /dev/null
+++ b/test/fodc/pod.yaml
@@ -0,0 +1,78 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+apiVersion: v1
+kind: Pod
+metadata:
+ name: banyand-fodc-ktm
+ labels:
+ app: banyand-fodc-ktm
+spec:
+ shareProcessNamespace: true
+ containers:
+ - name: banyand
+ image: apache/skywalking-banyandb:latest
+ imagePullPolicy: IfNotPresent
+ args:
+ - standalone
+ ports:
+ - name: banyand-http
+ containerPort: 17913
+ - name: banyand-metrics
+ containerPort: 2121
+ - name: fodc-agent
+ image: apache/skywalking-banyandb-fodc-agent:latest
+ imagePullPolicy: IfNotPresent
+ args:
+ - --ktm-enabled
+ - --ktm-interval=2s
+ - --poll-metrics-interval=2s
+ - --poll-metrics-ports=2121
+ - --prometheus-listen-addr=:9090
+ ports:
+ - name: fodc-metrics
+ containerPort: 9090
+ volumeMounts:
+ - name: tracefs
+ mountPath: /sys/kernel/tracing
+ readOnly: true
+ - name: debugfs
+ mountPath: /sys/kernel/debug
+ readOnly: true
+ - name: cgroup
+ mountPath: /sys/fs/cgroup
+ readOnly: true
+ securityContext:
+ allowPrivilegeEscalation: true
+ seccompProfile:
+ type: Unconfined
+ capabilities:
+ add:
+ - BPF
+ - PERFMON
+ - SYS_RESOURCE
+ volumes:
+ - name: tracefs
+ hostPath:
+ path: /sys/kernel/tracing
+ type: DirectoryOrCreate
+ - name: debugfs
+ hostPath:
+ path: /sys/kernel/debug
+ type: DirectoryOrCreate
+ - name: cgroup
+ hostPath:
+ path: /sys/fs/cgroup
+ type: DirectoryOrCreate
diff --git a/test/fodc/write-load.sh b/test/fodc/write-load.sh
new file mode 100755
index 000000000..61c5e92f1
--- /dev/null
+++ b/test/fodc/write-load.sh
@@ -0,0 +1,75 @@
+#!/bin/bash
+
+# Licensed to Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+# write-load.sh generates I/O on BanyanDB by creating property groups and
+# querying them via the HTTP REST API. This ensures KTM eBPF tracepoints
+# observe real disk I/O so that ktm_* metrics become non-zero.
+#
+# The script runs in the background and keeps generating load until killed.
+
+set -uo pipefail
+
+POD_NAME=${POD_NAME:-banyand-fodc-ktm}
+BANYAND_CONTAINER=${BANYAND_CONTAINER:-banyand}
+BANYAND_API=${BANYAND_API:-http://127.0.0.1:17913/api}
+LOAD_INTERVAL=${LOAD_INTERVAL:-1}
+
+http_post() {
+ local path="$1"
+ local body="$2"
+ kubectl exec "${POD_NAME}" -c "${BANYAND_CONTAINER}" -- \
+ sh -c "wget -qO- --header 'Content-Type: application/json' \
+ --post-data '${body}' \
+ '${BANYAND_API}${path}'" 2>/dev/null || true
+}
+
+http_get() {
+ local path="$1"
+ kubectl exec "${POD_NAME}" -c "${BANYAND_CONTAINER}" -- \
+ sh -c "wget -qO- '${BANYAND_API}${path}'" 2>/dev/null || true
+}
+
+echo "write-load: waiting for pod to be ready..."
+kubectl wait --for=condition=Ready "pod/${POD_NAME}" --timeout=300s
+
+echo "write-load: creating property groups to generate write I/O..."
+for i in $(seq 1 5); do
+ http_post "/v1/group/schema" \
+ "{\"group\":{\"metadata\":{\"name\":\"e2e-fodc-${i}\"},\"catalog\":\"CATALOG_PROPERTY\",\"resource_opts\":{\"shard_num\":1}}}"
+done
+
+echo "write-load: entering continuous read/write loop..."
+round=0
+while true; do
+ round=$((round + 1))
+
+ # Create another group (write I/O)
+ http_post "/v1/group/schema" \
+ "{\"group\":{\"metadata\":{\"name\":\"e2e-fodc-loop-${round}\"},\"catalog\":\"CATALOG_PROPERTY\",\"resource_opts\":{\"shard_num\":1}}}"
+
+ # List all groups (read I/O)
+ http_get "/v1/group/schema/lists"
+
+ # Health check (read I/O)
+ http_get "/healthz"
+
+ if [ "$((round % 10))" -eq 0 ]; then
+ echo "write-load: completed ${round} rounds"
+ fi
+
+ sleep "${LOAD_INTERVAL}"
+done