Copilot commented on code in PR #1214: URL: https://github.com/apache/skywalking-banyandb/pull/1214#discussion_r3556410647
########## pkg/pipeline/sdk/encode.go: ########## @@ -0,0 +1,114 @@ +// 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 sdk + +import ( + "fmt" + + "github.com/apache/skywalking-banyandb/pkg/convert" + "github.com/apache/skywalking-banyandb/pkg/encoding/vararray" + "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" +) + +// int64Width is the byte width of one encoded int64 (matches convert.Int64ToBytes). +const int64Width = 8 + +// EncodeTagValue marshals a raw Go value into the native tag-value byte +// layout that DecodeTagValue decodes — the inverse of DecodeTagValue. It +// deliberately takes a plain Go value (any), not a Value: Value's fields are +// unexported and this package exports no constructor for it, so an encoder +// built around Value would be unconstructible from outside this package +// (e.g. from pkg/pipeline/sdk/sdktest, a different package, or a plugin +// author's own test) — defeating the purpose of a shared, reusable encoder. +// +// Only a genuinely nil v (an untyped nil `any`) encodes to a nil raw value — +// the native "tag absent on this row" representation. A present-but-empty +// collection ([]byte{}, []string{}, []int64{}) encodes to a NON-nil empty +// raw value, so it decodes back as an empty value rather than as null, +// consistently across all three collection types. +// +// Supported (vt, Go type) pairs: +// - ValueTypeStr: string +// - ValueTypeInt64: int64 +// - ValueTypeTimestamp: int64 (unix nanoseconds; same 8-byte wire layout as Int64) +// - ValueTypeFloat64: float64 +// - ValueTypeBinaryData: []byte +// - ValueTypeStrArr: []string +// - ValueTypeInt64Arr: []int64 +// +// A mismatch between vt and v's concrete Go type, or an unsupported vt, +// returns an error. +func EncodeTagValue(vt valuetype.ValueType, v any) ([]byte, error) { + if v == nil { + return nil, nil + } + switch vt { + case valuetype.ValueTypeStr: + s, ok := v.(string) + if !ok { + return nil, fmt.Errorf("EncodeTagValue: ValueTypeStr requires a string, got %T", v) + } + return convert.StringToBytes(s), nil Review Comment: EncodeTagValue returns convert.StringToBytes(s) for ValueTypeStr, which uses an unsafe, zero-copy string→[]byte conversion. Because the returned slice is writable, a caller could accidentally mutate read-only string backing memory (undefined behavior). Also, the function’s docs/round-trip semantics rely on empty strings encoding to a non-nil raw value (nil means null), which a plain string→[]byte conversion may not preserve. Prefer returning an allocated copy, while explicitly keeping empty string as a non-nil empty slice. ########## test/plugin-sidecar/run.sh: ########## @@ -0,0 +1,244 @@ +#!/usr/bin/env bash +# 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. +# +# SHIP GATE: prove the CGO "-plugins" host image + carrier-mount design works +# in REAL Kubernetes (kind). banyandb images ONLY — no OAP, no producer, no +# traffic generator. +# +# Flow: create kind cluster -> (build if needed) -> kind load BOTH images -> +# kubectl apply standalone (carrier initContainer -> emptyDir /plugins) +# -> wait Ready -> assertions -> teardown (always). +# +# Required assertions (the gate): +# (a) banyand pod reaches Ready (CGO/dynamic host boots in-cluster) +# (b) /plugins/latencystatussampler.so present in the banyand container +# (c) register a group whose pipeline references latencystatussampler.so and +# assert the data node LOADS it with NO failure (plugin.Open succeeded): +# - the fail-open ERROR is ABSENT, AND +# - sampler_active_count{group}>0 (a positive load-success signal). +# Stretch: write a drop-eligible + a keep trace, await merge, assert the +# eligible one is dropped (best-effort; does not fail the gate). +# +# Env knobs: +# TAG image tag (default: latest) +# HOST_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins +# CARRIER_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins-carrier +# SKIP_BUILD if "true", do not build images (assume present / to be loaded) +# KEEP_CLUSTER if "true", do not delete the kind cluster on exit (debugging) +# +# NOTE on the "stretch" (write a drop-eligible + keep trace, merge, assert the +# eligible one is dropped): that end-to-end DROP behavior is already proven by +# the in-process trace_pipeline integration suite +# (test/integration/standalone/pipeline), which boots the SAME "-plugins" host +# binary + carrier .so via --trace-pipeline-trusted-plugin-dir and asserts the +# sampler drops the eligible trace on merge. This kind gate deliberately scopes +# to what only real Kubernetes can prove — the CGO host boots as a pod and the +# carrier-mounted .so is loadable in-cluster (assertions a/b/c) — rather than +# re-deriving the drop logic (which needs the trace-write path, not available +# via bydbctl) more flakily over a port-forward. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" + +CLUSTER="${CLUSTER:-banyandb-plugin-sidecar}" +TAG="${TAG:-latest}" +HOST_IMAGE="${HOST_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins}" +CARRIER_IMAGE="${CARRIER_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins-carrier}" +SKIP_BUILD="${SKIP_BUILD:-false}" +KEEP_CLUSTER="${KEEP_CLUSTER:-false}" +RUN_STRETCH="${RUN_STRETCH:-true}" + +POD="banyandb-plugin" +GROUP="test-trace-pipeline" +KCTX="kind-${CLUSTER}" +PF_PID="" + +log() { echo -e "\n=== $* ==="; } +fail() { echo "GATE FAILURE: $*" >&2; exit 1; } + +cleanup() { + local rc=$? + if [[ -n "${PF_PID}" ]]; then kill "${PF_PID}" 2>/dev/null || true; fi + if [[ "${rc}" -ne 0 ]]; then + log "FAILURE diagnostics (rc=${rc})" + kubectl --context "${KCTX}" get pods -o wide 2>/dev/null || true + kubectl --context "${KCTX}" describe pod "${POD}" 2>/dev/null | tail -40 || true + echo "--- banyandb container logs (tail) ---" + kubectl --context "${KCTX}" logs "${POD}" -c banyandb 2>/dev/null | tail -60 || true + echo "--- install-plugins initContainer logs ---" + kubectl --context "${KCTX}" logs "${POD}" -c install-plugins 2>/dev/null | tail -20 || true + fi + if [[ "${KEEP_CLUSTER}" != "true" ]]; then + log "Deleting kind cluster ${CLUSTER}" + kind delete cluster --name "${CLUSTER}" 2>/dev/null || true + fi + exit "${rc}" +} +trap cleanup EXIT + +build_images() { + if [[ "${SKIP_BUILD}" == "true" ]]; then + log "SKIP_BUILD=true — assuming ${HOST_IMAGE} and ${CARRIER_IMAGE} already exist" + return + fi + if docker image inspect "${HOST_IMAGE}" >/dev/null 2>&1 && \ + docker image inspect "${CARRIER_IMAGE}" >/dev/null 2>&1; then + log "Reusing existing images ${HOST_IMAGE} and ${CARRIER_IMAGE}" + return + fi + log "Building companion static binaries + plugin host + carrier images (amd64)" + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 make -C banyand release ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" BINARYTYPE=plugins make -C banyand docker ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" make -C banyand docker.plugins-carrier ) +} + +main() { + command -v kind >/dev/null || fail "kind not found on PATH" + command -v kubectl >/dev/null || fail "kubectl not found on PATH" + command -v envsubst >/dev/null || fail "envsubst not found on PATH (install gettext-base)" + command -v go >/dev/null || fail "go not found on PATH (needed to run the pipeline register helper)" + + # Defensive: kill any stale kubectl port-forwards left by a prior run so our + # ephemeral-port forwards below can always bind (the wedge this revision fixes). + for p in $(pgrep -x kubectl 2>/dev/null || true); do + tr '\0' ' ' < "/proc/$p/cmdline" 2>/dev/null | grep -q port-forward && kill "$p" 2>/dev/null || true + done + + build_images + + log "Creating kind cluster ${CLUSTER}" + kind create cluster --name "${CLUSTER}" --config "${SCRIPT_DIR}/kind.yaml" --wait 180s + + log "Loading BOTH plugin images into kind (lockstep, same tag)" + kind load docker-image "${HOST_IMAGE}" --name "${CLUSTER}" + kind load docker-image "${CARRIER_IMAGE}" --name "${CLUSTER}" + + log "Deploying standalone banyandb (carrier initContainer -> emptyDir /plugins)" + HOST_IMAGE="${HOST_IMAGE}" CARRIER_IMAGE="${CARRIER_IMAGE}" \ + envsubst '${HOST_IMAGE} ${CARRIER_IMAGE}' < "${SCRIPT_DIR}/banyandb-standalone.yaml" \ + | kubectl --context "${KCTX}" apply -f - + + # ---- Assertion (a): pod reaches Ready (CGO/dynamic host boots) ---- + log "Assertion (a): waiting for pod/${POD} to become Ready" + kubectl --context "${KCTX}" wait --for=condition=Ready "pod/${POD}" --timeout=240s \ + || fail "(a) pod ${POD} did not become Ready — the CGO host image failed to boot in-cluster" + kubectl --context "${KCTX}" get pods -o wide + echo "(a) PASS: banyandb pod (CGO/dynamic -plugins host image) is Ready" + + # ---- Assertion (b): the carrier delivered the .so into the shared /plugins ---- + # The host (banyandb) container is distroless — it has NO shell, so we cannot + # `kubectl exec ... ls` into it. Instead verify delivery from the carrier + # initContainer's own log: it runs `cp /plugins/*.so /shared/ && ls -l /shared` + # into the emptyDir the host mounts at /plugins, so its log lists the .so it + # placed there. + log "Assertion (b): latencystatussampler.so delivered to the shared /plugins by the carrier initContainer" + local initlog + initlog="$(kubectl --context "${KCTX}" logs "${POD}" -c install-plugins 2>/dev/null || true)" + echo "${initlog}" + echo "${initlog}" | grep -q "latencystatussampler.so" \ + || fail "(b) install-plugins initContainer log does not show latencystatussampler.so — the carrier did not populate the shared volume" + echo "(b) PASS: latencystatussampler.so delivered to the shared /plugins volume (per the carrier initContainer log)" + + # ---- Assertion (c): register the pipeline; assert the data node LOADS it ---- + # Registered through the property schema registry (the store the data node's + # trace schemaRepo watches), NOT the GroupRegistryService HTTP/gRPC endpoint + # (whose store the pipeline reconcile does not observe, and whose HTTP gateway + # drops the nested pipeline field). Each port-forward uses a fresh EPHEMERAL + # local port and is torn down after use, so a stale forward from a prior run + # can never block the bind (the failure that previously wedged the poll). + local reg_lport=$(( (RANDOM % 10000) + 40000 )) + local met_lport=$(( (RANDOM % 10000) + 30000 )) + + log "Assertion (c): registering group + pipeline referencing latencystatussampler.so (property schema :17916 via local ${reg_lport})" + kubectl --context "${KCTX}" port-forward "pod/${POD}" "${reg_lport}:17916" >/tmp/pf-reg.log 2>&1 & + local reg_pf=$! + sleep 4 Review Comment: The property-schema `kubectl port-forward` PID is stored in a local variable, but it is not wired into the EXIT cleanup trap. If the script exits between starting the port-forward and the explicit `kill` (e.g., `go build` fails under `set -e`), the port-forward can be left running and wedge subsequent runs. Assign the reg port-forward PID to PF_PID while it’s active so cleanup always reaps it on early exit. ########## test/plugin-sidecar/run.sh: ########## @@ -0,0 +1,244 @@ +#!/usr/bin/env bash +# 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. +# +# SHIP GATE: prove the CGO "-plugins" host image + carrier-mount design works +# in REAL Kubernetes (kind). banyandb images ONLY — no OAP, no producer, no +# traffic generator. +# +# Flow: create kind cluster -> (build if needed) -> kind load BOTH images -> +# kubectl apply standalone (carrier initContainer -> emptyDir /plugins) +# -> wait Ready -> assertions -> teardown (always). +# +# Required assertions (the gate): +# (a) banyand pod reaches Ready (CGO/dynamic host boots in-cluster) +# (b) /plugins/latencystatussampler.so present in the banyand container +# (c) register a group whose pipeline references latencystatussampler.so and +# assert the data node LOADS it with NO failure (plugin.Open succeeded): +# - the fail-open ERROR is ABSENT, AND +# - sampler_active_count{group}>0 (a positive load-success signal). +# Stretch: write a drop-eligible + a keep trace, await merge, assert the +# eligible one is dropped (best-effort; does not fail the gate). +# +# Env knobs: +# TAG image tag (default: latest) +# HOST_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins +# CARRIER_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins-carrier +# SKIP_BUILD if "true", do not build images (assume present / to be loaded) +# KEEP_CLUSTER if "true", do not delete the kind cluster on exit (debugging) +# +# NOTE on the "stretch" (write a drop-eligible + keep trace, merge, assert the +# eligible one is dropped): that end-to-end DROP behavior is already proven by +# the in-process trace_pipeline integration suite +# (test/integration/standalone/pipeline), which boots the SAME "-plugins" host +# binary + carrier .so via --trace-pipeline-trusted-plugin-dir and asserts the +# sampler drops the eligible trace on merge. This kind gate deliberately scopes +# to what only real Kubernetes can prove — the CGO host boots as a pod and the +# carrier-mounted .so is loadable in-cluster (assertions a/b/c) — rather than +# re-deriving the drop logic (which needs the trace-write path, not available +# via bydbctl) more flakily over a port-forward. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" + +CLUSTER="${CLUSTER:-banyandb-plugin-sidecar}" +TAG="${TAG:-latest}" +HOST_IMAGE="${HOST_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins}" +CARRIER_IMAGE="${CARRIER_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins-carrier}" +SKIP_BUILD="${SKIP_BUILD:-false}" +KEEP_CLUSTER="${KEEP_CLUSTER:-false}" +RUN_STRETCH="${RUN_STRETCH:-true}" + +POD="banyandb-plugin" +GROUP="test-trace-pipeline" +KCTX="kind-${CLUSTER}" +PF_PID="" + +log() { echo -e "\n=== $* ==="; } +fail() { echo "GATE FAILURE: $*" >&2; exit 1; } + +cleanup() { + local rc=$? + if [[ -n "${PF_PID}" ]]; then kill "${PF_PID}" 2>/dev/null || true; fi + if [[ "${rc}" -ne 0 ]]; then + log "FAILURE diagnostics (rc=${rc})" + kubectl --context "${KCTX}" get pods -o wide 2>/dev/null || true + kubectl --context "${KCTX}" describe pod "${POD}" 2>/dev/null | tail -40 || true + echo "--- banyandb container logs (tail) ---" + kubectl --context "${KCTX}" logs "${POD}" -c banyandb 2>/dev/null | tail -60 || true + echo "--- install-plugins initContainer logs ---" + kubectl --context "${KCTX}" logs "${POD}" -c install-plugins 2>/dev/null | tail -20 || true + fi + if [[ "${KEEP_CLUSTER}" != "true" ]]; then + log "Deleting kind cluster ${CLUSTER}" + kind delete cluster --name "${CLUSTER}" 2>/dev/null || true + fi + exit "${rc}" +} +trap cleanup EXIT + +build_images() { + if [[ "${SKIP_BUILD}" == "true" ]]; then + log "SKIP_BUILD=true — assuming ${HOST_IMAGE} and ${CARRIER_IMAGE} already exist" + return + fi + if docker image inspect "${HOST_IMAGE}" >/dev/null 2>&1 && \ + docker image inspect "${CARRIER_IMAGE}" >/dev/null 2>&1; then + log "Reusing existing images ${HOST_IMAGE} and ${CARRIER_IMAGE}" + return + fi + log "Building companion static binaries + plugin host + carrier images (amd64)" + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 make -C banyand release ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" BINARYTYPE=plugins make -C banyand docker ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" make -C banyand docker.plugins-carrier ) +} + +main() { + command -v kind >/dev/null || fail "kind not found on PATH" + command -v kubectl >/dev/null || fail "kubectl not found on PATH" + command -v envsubst >/dev/null || fail "envsubst not found on PATH (install gettext-base)" + command -v go >/dev/null || fail "go not found on PATH (needed to run the pipeline register helper)" + + # Defensive: kill any stale kubectl port-forwards left by a prior run so our + # ephemeral-port forwards below can always bind (the wedge this revision fixes). + for p in $(pgrep -x kubectl 2>/dev/null || true); do + tr '\0' ' ' < "/proc/$p/cmdline" 2>/dev/null | grep -q port-forward && kill "$p" 2>/dev/null || true + done Review Comment: This loop kills *any* running `kubectl port-forward` process, including ones unrelated to this ship-gate (e.g., a developer’s other cluster work). It’s safer to only terminate port-forwards that target this test’s pod/context. Consider narrowing the match to `port-forward` commands for `pod/${POD}` under `${KCTX}` before killing. ########## test/plugin-sidecar/run.sh: ########## @@ -0,0 +1,244 @@ +#!/usr/bin/env bash +# 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. +# +# SHIP GATE: prove the CGO "-plugins" host image + carrier-mount design works +# in REAL Kubernetes (kind). banyandb images ONLY — no OAP, no producer, no +# traffic generator. +# +# Flow: create kind cluster -> (build if needed) -> kind load BOTH images -> +# kubectl apply standalone (carrier initContainer -> emptyDir /plugins) +# -> wait Ready -> assertions -> teardown (always). +# +# Required assertions (the gate): +# (a) banyand pod reaches Ready (CGO/dynamic host boots in-cluster) +# (b) /plugins/latencystatussampler.so present in the banyand container +# (c) register a group whose pipeline references latencystatussampler.so and +# assert the data node LOADS it with NO failure (plugin.Open succeeded): +# - the fail-open ERROR is ABSENT, AND +# - sampler_active_count{group}>0 (a positive load-success signal). +# Stretch: write a drop-eligible + a keep trace, await merge, assert the +# eligible one is dropped (best-effort; does not fail the gate). +# +# Env knobs: +# TAG image tag (default: latest) +# HOST_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins +# CARRIER_IMAGE default: apache/skywalking-banyandb:${TAG}-plugins-carrier +# SKIP_BUILD if "true", do not build images (assume present / to be loaded) +# KEEP_CLUSTER if "true", do not delete the kind cluster on exit (debugging) +# +# NOTE on the "stretch" (write a drop-eligible + keep trace, merge, assert the +# eligible one is dropped): that end-to-end DROP behavior is already proven by +# the in-process trace_pipeline integration suite +# (test/integration/standalone/pipeline), which boots the SAME "-plugins" host +# binary + carrier .so via --trace-pipeline-trusted-plugin-dir and asserts the +# sampler drops the eligible trace on merge. This kind gate deliberately scopes +# to what only real Kubernetes can prove — the CGO host boots as a pod and the +# carrier-mounted .so is loadable in-cluster (assertions a/b/c) — rather than +# re-deriving the drop logic (which needs the trace-write path, not available +# via bydbctl) more flakily over a port-forward. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" + +CLUSTER="${CLUSTER:-banyandb-plugin-sidecar}" +TAG="${TAG:-latest}" +HOST_IMAGE="${HOST_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins}" +CARRIER_IMAGE="${CARRIER_IMAGE:-apache/skywalking-banyandb:${TAG}-plugins-carrier}" +SKIP_BUILD="${SKIP_BUILD:-false}" +KEEP_CLUSTER="${KEEP_CLUSTER:-false}" +RUN_STRETCH="${RUN_STRETCH:-true}" + +POD="banyandb-plugin" +GROUP="test-trace-pipeline" +KCTX="kind-${CLUSTER}" +PF_PID="" + +log() { echo -e "\n=== $* ==="; } +fail() { echo "GATE FAILURE: $*" >&2; exit 1; } + +cleanup() { + local rc=$? + if [[ -n "${PF_PID}" ]]; then kill "${PF_PID}" 2>/dev/null || true; fi + if [[ "${rc}" -ne 0 ]]; then + log "FAILURE diagnostics (rc=${rc})" + kubectl --context "${KCTX}" get pods -o wide 2>/dev/null || true + kubectl --context "${KCTX}" describe pod "${POD}" 2>/dev/null | tail -40 || true + echo "--- banyandb container logs (tail) ---" + kubectl --context "${KCTX}" logs "${POD}" -c banyandb 2>/dev/null | tail -60 || true + echo "--- install-plugins initContainer logs ---" + kubectl --context "${KCTX}" logs "${POD}" -c install-plugins 2>/dev/null | tail -20 || true + fi + if [[ "${KEEP_CLUSTER}" != "true" ]]; then + log "Deleting kind cluster ${CLUSTER}" + kind delete cluster --name "${CLUSTER}" 2>/dev/null || true + fi + exit "${rc}" +} +trap cleanup EXIT + +build_images() { + if [[ "${SKIP_BUILD}" == "true" ]]; then + log "SKIP_BUILD=true — assuming ${HOST_IMAGE} and ${CARRIER_IMAGE} already exist" + return + fi + if docker image inspect "${HOST_IMAGE}" >/dev/null 2>&1 && \ + docker image inspect "${CARRIER_IMAGE}" >/dev/null 2>&1; then + log "Reusing existing images ${HOST_IMAGE} and ${CARRIER_IMAGE}" + return + fi + log "Building companion static binaries + plugin host + carrier images (amd64)" + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 make -C banyand release ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" BINARYTYPE=plugins make -C banyand docker ) + ( cd "${REPO_ROOT}" && PLATFORMS=linux/amd64 TAG="${TAG}" make -C banyand docker.plugins-carrier ) +} + +main() { + command -v kind >/dev/null || fail "kind not found on PATH" + command -v kubectl >/dev/null || fail "kubectl not found on PATH" + command -v envsubst >/dev/null || fail "envsubst not found on PATH (install gettext-base)" + command -v go >/dev/null || fail "go not found on PATH (needed to run the pipeline register helper)" + + # Defensive: kill any stale kubectl port-forwards left by a prior run so our + # ephemeral-port forwards below can always bind (the wedge this revision fixes). + for p in $(pgrep -x kubectl 2>/dev/null || true); do + tr '\0' ' ' < "/proc/$p/cmdline" 2>/dev/null | grep -q port-forward && kill "$p" 2>/dev/null || true + done + + build_images + + log "Creating kind cluster ${CLUSTER}" + kind create cluster --name "${CLUSTER}" --config "${SCRIPT_DIR}/kind.yaml" --wait 180s + + log "Loading BOTH plugin images into kind (lockstep, same tag)" + kind load docker-image "${HOST_IMAGE}" --name "${CLUSTER}" + kind load docker-image "${CARRIER_IMAGE}" --name "${CLUSTER}" + + log "Deploying standalone banyandb (carrier initContainer -> emptyDir /plugins)" + HOST_IMAGE="${HOST_IMAGE}" CARRIER_IMAGE="${CARRIER_IMAGE}" \ + envsubst '${HOST_IMAGE} ${CARRIER_IMAGE}' < "${SCRIPT_DIR}/banyandb-standalone.yaml" \ + | kubectl --context "${KCTX}" apply -f - + + # ---- Assertion (a): pod reaches Ready (CGO/dynamic host boots) ---- + log "Assertion (a): waiting for pod/${POD} to become Ready" + kubectl --context "${KCTX}" wait --for=condition=Ready "pod/${POD}" --timeout=240s \ + || fail "(a) pod ${POD} did not become Ready — the CGO host image failed to boot in-cluster" + kubectl --context "${KCTX}" get pods -o wide + echo "(a) PASS: banyandb pod (CGO/dynamic -plugins host image) is Ready" + + # ---- Assertion (b): the carrier delivered the .so into the shared /plugins ---- + # The host (banyandb) container is distroless — it has NO shell, so we cannot + # `kubectl exec ... ls` into it. Instead verify delivery from the carrier + # initContainer's own log: it runs `cp /plugins/*.so /shared/ && ls -l /shared` + # into the emptyDir the host mounts at /plugins, so its log lists the .so it + # placed there. + log "Assertion (b): latencystatussampler.so delivered to the shared /plugins by the carrier initContainer" + local initlog + initlog="$(kubectl --context "${KCTX}" logs "${POD}" -c install-plugins 2>/dev/null || true)" + echo "${initlog}" + echo "${initlog}" | grep -q "latencystatussampler.so" \ + || fail "(b) install-plugins initContainer log does not show latencystatussampler.so — the carrier did not populate the shared volume" + echo "(b) PASS: latencystatussampler.so delivered to the shared /plugins volume (per the carrier initContainer log)" + + # ---- Assertion (c): register the pipeline; assert the data node LOADS it ---- + # Registered through the property schema registry (the store the data node's + # trace schemaRepo watches), NOT the GroupRegistryService HTTP/gRPC endpoint + # (whose store the pipeline reconcile does not observe, and whose HTTP gateway + # drops the nested pipeline field). Each port-forward uses a fresh EPHEMERAL + # local port and is torn down after use, so a stale forward from a prior run + # can never block the bind (the failure that previously wedged the poll). + local reg_lport=$(( (RANDOM % 10000) + 40000 )) + local met_lport=$(( (RANDOM % 10000) + 30000 )) + + log "Assertion (c): registering group + pipeline referencing latencystatussampler.so (property schema :17916 via local ${reg_lport})" + kubectl --context "${KCTX}" port-forward "pod/${POD}" "${reg_lport}:17916" >/tmp/pf-reg.log 2>&1 & + local reg_pf=$! + sleep 4 + local register_bin="${RUNNER_TEMP:-/tmp}/plugin-sidecar-register" + ( cd "${REPO_ROOT}" && go build -o "${register_bin}" ./test/plugin-sidecar/register ) + "${register_bin}" \ + --property-schema-addr "localhost:${reg_lport}" \ + --node-name "${POD}:17912" \ + --group "${GROUP}" \ + --so latencystatussampler.so \ + || { kill "${reg_pf}" 2>/dev/null || true; fail "register helper failed"; } + kill "${reg_pf}" 2>/dev/null || true Review Comment: After the registration port-forward is killed, PF_PID should be cleared so the cleanup trap doesn’t treat a stale PID as the active forward (and to avoid confusing diagnostics if later steps fail). ########## test/plugin-sidecar/register/main.go: ########## @@ -0,0 +1,214 @@ +// 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. + +// Command register creates the trace group + schema and attaches a trace +// pipeline (a single latency-status sampler) so a standalone data node +// reconciles it and calls plugin.Open on the carrier-mounted .so. It is used +// by the plugin-sidecar kind ship gate to drive assertion (c). +// +// It writes through the PROPERTY schema registry client (the same path +// test/cases/tracepipeline's PreloadSchemaViaProperty uses), NOT the +// GroupRegistryService HTTP/gRPC endpoint: only the property schema store is +// watched by the data node's trace schemaRepo, so only writes through it fire +// the KindGroup reconcile that loads the plugin. The property store also +// round-trips the nested pipeline field (the HTTP gateway does not). +package main + +import ( + "context" + "flag" + "fmt" + "os" + "time" + + "github.com/pkg/errors" + "google.golang.org/protobuf/types/known/structpb" + + commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1" + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" + "github.com/apache/skywalking-banyandb/banyand/metadata/schema" + "github.com/apache/skywalking-banyandb/banyand/metadata/schema/property" + "github.com/apache/skywalking-banyandb/banyand/observability" +) + +func main() { + propAddr := flag.String("property-schema-addr", "localhost:17916", + "banyandb property-schema gRPC address (PropertySchemaGrpcAddress)") + nodeName := flag.String("node-name", "banyandb-plugin:17912", "target data node name (schema server identity)") + group := flag.String("group", "test-trace-pipeline", "target trace group") + soName := flag.String("so", "latencystatussampler.so", "trusted-dir-relative .so path") + threshold := flag.Float64("threshold-ms", 500, "latency sampler thresholdMs") + successValue := flag.String("success-value", "success", "latency sampler successValue") + flag.Parse() + + if err := run(*propAddr, *nodeName, *group, *soName, *threshold, *successValue); err != nil { + fmt.Fprintf(os.Stderr, "register: %v\n", err) + os.Exit(1) + } + fmt.Printf("registered pipeline on group %q referencing %q (via property schema registry)\n", *group, *soName) +} + +// nodeRegistry implements schema.Node: it advertises exactly one ROLE_META +// schema server at the given property-schema gRPC address. +type nodeRegistry struct { + nodes []*databasev1.Node +} + +func (r *nodeRegistry) ListNode(_ context.Context, role databasev1.Role) ([]*databasev1.Node, error) { + var out []*databasev1.Node + for _, n := range r.nodes { + for _, nr := range n.GetRoles() { + if nr == role { + out = append(out, n) + break + } + } + } + return out, nil +} +func (r *nodeRegistry) RegisterNode(_ context.Context, _ *databasev1.Node, _ bool) error { return nil } +func (r *nodeRegistry) GetNode(_ context.Context, _ string) (*databasev1.Node, error) { + return nil, nil +} +func (r *nodeRegistry) UpdateNode(_ context.Context, _ *databasev1.Node) error { return nil } + +func run(propAddr, nodeName, group, soName string, threshold float64, successValue string) error { + reg, regErr := property.NewSchemaRegistryClient(&property.ClientConfig{ + OMR: observability.BypassRegistry, + GRPCTimeout: 10 * time.Second, + NodeRegistry: &nodeRegistry{nodes: []*databasev1.Node{{ + Metadata: &commonv1.Metadata{Name: nodeName}, + Roles: []databasev1.Role{databasev1.Role_ROLE_META}, + PropertySchemaGrpcAddress: propAddr, + }}}, + }) + if regErr != nil { + return errors.Wrap(regErr, "new property schema registry client") + } + defer func() { _ = reg.Close() }() + + // Wait for the schema server node to become active. + deadline := time.Now().Add(30 * time.Second) + for len(reg.ActiveNodeNames()) == 0 { + if time.Now().After(deadline) { + return errors.New("no active property schema server node within 30s") + } + time.Sleep(200 * time.Millisecond) + } + + ctx := context.Background() + + // Mirror the proven integration-test sequence + // (PreloadSchemaViaProperty + RegisterSamplerRuntime): first create the + // group WITHOUT the pipeline, then attach the pipeline via UpdateGroup. The + // pipeline is delivered on the group-UPDATE watch event, which is the path + // that reliably fires the data node's reconcilePipeline (create-with-pipeline + // does not). + base := &commonv1.Group{ + Metadata: &commonv1.Metadata{Name: group}, + Catalog: commonv1.Catalog_CATALOG_TRACE, + ResourceOpts: &commonv1.ResourceOpts{ + ShardNum: 1, + SegmentInterval: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 1}, + Ttl: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 3}, + }, + } + if _, createErr := reg.CreateGroup(ctx, base); createErr != nil && !errors.Is(createErr, schema.ErrGRPCAlreadyExists) { + return errors.Wrap(createErr, "create group") + } + + // Create the trace schema (harmless if already present); mirrors + // test/cases/tracepipeline/testdata/traces/filter.json. + trace := traceSchema(group) + if _, traceErr := reg.CreateTrace(ctx, trace); traceErr != nil && !errors.Is(traceErr, schema.ErrGRPCAlreadyExists) { + return errors.Wrap(traceErr, "create trace schema") + } + + // Wait for the freshly-created group to propagate to (and be established + // on) the data node before attaching the pipeline — mirrors the + // integration suite's waitForSchemaSync between PreloadSchemaViaProperty + // and RegisterSamplerRuntime. Without this settle, the pipeline UpdateGroup + // races the group create and the data node's reconcilePipeline does not + // fire for it. + syncDeadline := time.Now().Add(30 * time.Second) + for { + got, getErr := reg.GetGroup(ctx, group) + if getErr == nil && got != nil { + break + } + if time.Now().After(syncDeadline) { + return errors.New("group did not become visible within 30s") + } + time.Sleep(500 * time.Millisecond) + } + time.Sleep(3 * time.Second) + + // Attach the pipeline via UpdateGroup → fires reconcilePipeline → plugin.Open. + withPipeline := &commonv1.Group{ + Metadata: &commonv1.Metadata{Name: group}, + Catalog: commonv1.Catalog_CATALOG_TRACE, + ResourceOpts: base.ResourceOpts, + Pipeline: pipelineConfig(soName, threshold, successValue), + } + if _, updateErr := reg.UpdateGroup(ctx, withPipeline); updateErr != nil { + return errors.Wrap(updateErr, "update group with pipeline") + } + return nil +} + +func pipelineConfig(soName string, threshold float64, successValue string) *commonv1.TracePipelineConfig { + cfgStruct, _ := structpb.NewStruct(map[string]any{ + "thresholdMs": threshold, + "successValue": successValue, + }) Review Comment: The error from structpb.NewStruct is currently ignored. NewStruct can fail for non-finite numbers (NaN/Inf), and ignoring the error would silently produce a nil config Struct, changing plugin behavior (it would fall back to defaults) without any signal. Fail fast on invalid config construction so the kind gate surfaces the issue clearly. ########## test/plugin-sidecar/register/main.go: ########## @@ -0,0 +1,214 @@ +// 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. + +// Command register creates the trace group + schema and attaches a trace +// pipeline (a single latency-status sampler) so a standalone data node +// reconciles it and calls plugin.Open on the carrier-mounted .so. It is used +// by the plugin-sidecar kind ship gate to drive assertion (c). +// +// It writes through the PROPERTY schema registry client (the same path +// test/cases/tracepipeline's PreloadSchemaViaProperty uses), NOT the +// GroupRegistryService HTTP/gRPC endpoint: only the property schema store is +// watched by the data node's trace schemaRepo, so only writes through it fire +// the KindGroup reconcile that loads the plugin. The property store also +// round-trips the nested pipeline field (the HTTP gateway does not). +package main + +import ( + "context" + "flag" + "fmt" + "os" + "time" + + "github.com/pkg/errors" + "google.golang.org/protobuf/types/known/structpb" + + commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1" + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" + "github.com/apache/skywalking-banyandb/banyand/metadata/schema" + "github.com/apache/skywalking-banyandb/banyand/metadata/schema/property" + "github.com/apache/skywalking-banyandb/banyand/observability" +) + +func main() { + propAddr := flag.String("property-schema-addr", "localhost:17916", + "banyandb property-schema gRPC address (PropertySchemaGrpcAddress)") + nodeName := flag.String("node-name", "banyandb-plugin:17912", "target data node name (schema server identity)") + group := flag.String("group", "test-trace-pipeline", "target trace group") + soName := flag.String("so", "latencystatussampler.so", "trusted-dir-relative .so path") + threshold := flag.Float64("threshold-ms", 500, "latency sampler thresholdMs") + successValue := flag.String("success-value", "success", "latency sampler successValue") + flag.Parse() + + if err := run(*propAddr, *nodeName, *group, *soName, *threshold, *successValue); err != nil { + fmt.Fprintf(os.Stderr, "register: %v\n", err) + os.Exit(1) + } + fmt.Printf("registered pipeline on group %q referencing %q (via property schema registry)\n", *group, *soName) +} + +// nodeRegistry implements schema.Node: it advertises exactly one ROLE_META +// schema server at the given property-schema gRPC address. +type nodeRegistry struct { + nodes []*databasev1.Node +} + +func (r *nodeRegistry) ListNode(_ context.Context, role databasev1.Role) ([]*databasev1.Node, error) { + var out []*databasev1.Node + for _, n := range r.nodes { + for _, nr := range n.GetRoles() { + if nr == role { + out = append(out, n) + break + } + } + } + return out, nil +} +func (r *nodeRegistry) RegisterNode(_ context.Context, _ *databasev1.Node, _ bool) error { return nil } +func (r *nodeRegistry) GetNode(_ context.Context, _ string) (*databasev1.Node, error) { + return nil, nil +} Review Comment: GetNode currently returns (nil, nil), which is ambiguous to callers (it looks like a successful lookup that happened to return a nil node). Returning an explicit error makes unexpected future call paths fail fast instead of behaving as a silent "success". Even though this helper currently only needs ListNode, returning a non-nil error here avoids fragile behavior if SchemaRegistryClient starts calling GetNode later. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
