From 0b92789fbc4c9ac2fc9f7d647a3373d5ca7dc564 Mon Sep 17 00:00:00 2001 From: Gaurav Chaudhary Date: Wed, 9 Sep 2026 12:43:46 +0530 Subject: [PATCH] test: speed up Pulsar integration tests Signed-off-by: Gaurav Chaudhary --- .github/workflows/pr.yaml | 8 +- .github/workflows/release.yaml | 9 +- README.md | 30 +++ go.mod | 49 +++- go.sum | 91 +++++--- pulsar/connector/basic_conn_test.go | 2 + pulsar/connector/consumer_test.go | 114 ++++++++++ pulsar/connector/mocks_test.go | 62 +++++ pulsar/connector/producer_test.go | 111 +++++++++ pulsar/connector/scripts/pulsar.sh | 20 -- pulsar/connector/scripts/pulsar_stop.sh | 13 -- pulsar/connector/suite_test.go | 289 +++--------------------- pulsar/internal/pulsartest/container.go | 141 ++++++++++++ pulsar/test/kill_port_process.go | 69 ------ pulsar/test/scripts/pulsar.sh | 20 -- pulsar/test/scripts/pulsar_stop.sh | 13 -- pulsar/test/suite.go | 165 ++++---------- pulsar/test/suite_integration_test.go | 40 ++++ 18 files changed, 694 insertions(+), 552 deletions(-) create mode 100644 pulsar/connector/consumer_test.go create mode 100644 pulsar/connector/mocks_test.go create mode 100644 pulsar/connector/producer_test.go delete mode 100755 pulsar/connector/scripts/pulsar.sh delete mode 100755 pulsar/connector/scripts/pulsar_stop.sh create mode 100644 pulsar/internal/pulsartest/container.go delete mode 100644 pulsar/test/kill_port_process.go delete mode 100755 pulsar/test/scripts/pulsar.sh delete mode 100755 pulsar/test/scripts/pulsar_stop.sh create mode 100644 pulsar/test/suite_integration_test.go diff --git a/.github/workflows/pr.yaml b/.github/workflows/pr.yaml index 73e7b57..a4818d8 100644 --- a/.github/workflows/pr.yaml +++ b/.github/workflows/pr.yaml @@ -10,7 +10,7 @@ jobs: pull-requests: write uses: kubescape/workflows/.github/workflows/go-basic-tests.yaml@main with: - GO_VERSION: "1.21" + GO_VERSION: "1.24.1" build: runs-on: ubuntu-latest steps: @@ -19,7 +19,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v3 with: - go-version: "1.21" + go-version: "1.24.1" - uses: actions/cache@v4 with: @@ -40,5 +40,5 @@ jobs: - name: Build run: go build -v ./... - - name: Test - run: go test -timeout 500s ./... \ No newline at end of file + - name: Integration tests + run: go test -tags=integration -p=1 -timeout 500s ./pulsar/connector ./pulsar/test diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index a996e9b..b6f416c 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -11,7 +11,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v2 with: - go-version: "1.21" + go-version: "1.24.1" - uses: actions/cache@v4 with: @@ -32,9 +32,12 @@ jobs: - name: Build run: go build -v ./... - - name: Test + - name: Unit tests run: go test -timeout 500s -v ./... + - name: Integration tests + run: go test -tags=integration -p=1 -timeout 500s -v ./pulsar/connector ./pulsar/test + - name: Create a release id: create_release uses: actions/create-release@v1 @@ -44,4 +47,4 @@ jobs: tag_name: v0.0.${{ github.run_number }} release_name: Release v0.0.${{ github.run_number }} draft: false - prerelease: false \ No newline at end of file + prerelease: false diff --git a/README.md b/README.md index 7ec9004..bdab133 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,33 @@ # Kubescape Messaging Package A collection of wrapper code around Pulsar to quickly and easily connect to Pulsar brokers, send and receive messages, and message queues and topics management. + +## Tests + +The default test suite uses in-process test doubles and does not require a +container runtime: + +```bash +go test ./... +``` + +Tests that verify real Pulsar behavior use Testcontainers and are selected with +the `integration` build tag: + +```bash +go test -tags=integration -p=1 -timeout=5m ./pulsar/connector ./pulsar/test +``` + +Integration tests require Docker or another Docker-compatible API. For rootless +Podman, enable its user socket and point Testcontainers at it: + +```bash +systemctl --user enable --now podman.socket +export DOCKER_HOST="unix://${XDG_RUNTIME_DIR}/podman/podman.sock" +export TESTCONTAINERS_RYUK_DISABLED=true +``` + +`PulsarTestSuite.AppPortStart` and `PulsarTestSuite.AdminPortStart` are retained +for source compatibility. Testcontainers now chooses collision-free ports and +replaces both fields with the resolved host ports during `SetupSuite`; values +assigned before setup no longer influence port selection. diff --git a/go.mod b/go.mod index 683f8c3..21ed10f 100644 --- a/go.mod +++ b/go.mod @@ -7,31 +7,49 @@ require ( github.com/armosec/armoapi-go v0.0.694 github.com/avast/retry-go v3.0.0+incompatible github.com/stretchr/testify v1.11.1 + github.com/testcontainers/testcontainers-go v0.40.0 + github.com/testcontainers/testcontainers-go/modules/pulsar v0.40.0 go.opentelemetry.io/otel v1.37.0 go.opentelemetry.io/otel/trace v1.37.0 go.uber.org/zap v1.27.0 - golang.org/x/sync v0.16.0 + golang.org/x/sync v0.17.0 ) require ( + dario.cat/mergo v1.0.2 // indirect github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect github.com/99designs/keyring v1.2.1 // indirect github.com/AthenZ/athenz v1.12.13 // indirect + github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect github.com/DataDog/zstd v1.5.7 // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect github.com/ardielle/ardielle-go v1.5.2 // indirect github.com/armosec/gojay v1.2.17 // indirect github.com/armosec/utils-k8s-go v0.0.30 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/bits-and-blooms/bitset v1.4.0 // indirect github.com/blang/semver/v4 v4.0.0 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect github.com/danieljoos/wincred v1.1.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/docker v28.5.1+incompatible // indirect + github.com/docker/go-connections v0.6.0 // indirect + github.com/docker/go-units v0.5.0 // indirect github.com/dvsekhvalnov/jose2go v1.6.0 // indirect + github.com/ebitengine/purego v0.8.4 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/go-jose/go-jose/v4 v4.0.5 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-jwt/jwt/v5 v5.2.2 // indirect @@ -44,36 +62,53 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.18.2 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/magiconair/properties v1.8.10 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/patternmatcher v0.6.0 // indirect + github.com/moby/sys/sequential v0.6.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/morikuni/aec v1.0.0 // indirect github.com/mtibben/percent v0.2.1 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/prometheus/client_golang v1.20.5 // indirect github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.61.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect + github.com/shirou/gopsutil/v4 v4.25.6 // indirect github.com/sirupsen/logrus v1.9.4-0.20230606125235-dd1b4c2e81af // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/spf13/cobra v1.9.1 // indirect github.com/spf13/pflag v1.0.7 // indirect github.com/stripe/stripe-go/v74 v74.30.0 // indirect + github.com/tklauser/go-sysconf v0.3.12 // indirect + github.com/tklauser/numcpus v0.6.1 // indirect github.com/x448/float16 v0.8.4 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 // indirect go.opentelemetry.io/otel/metric v1.37.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect - golang.org/x/crypto v0.41.0 // indirect - golang.org/x/mod v0.27.0 // indirect - golang.org/x/net v0.43.0 // indirect + golang.org/x/crypto v0.43.0 // indirect + golang.org/x/mod v0.28.0 // indirect + golang.org/x/net v0.45.0 // indirect golang.org/x/oauth2 v0.30.0 // indirect - golang.org/x/sys v0.35.0 // indirect - golang.org/x/term v0.34.0 // indirect - golang.org/x/text v0.28.0 // indirect + golang.org/x/sys v0.37.0 // indirect + golang.org/x/term v0.36.0 // indirect + golang.org/x/text v0.30.0 // indirect google.golang.org/protobuf v1.36.6 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 020e8f2..6c6e1f6 100644 --- a/go.sum +++ b/go.sum @@ -17,6 +17,8 @@ github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 h1:/vQbFIOMb github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4/go.mod h1:hN7oaIRCjzsZ2dE+yG5k+rsdt3qcwykqK6HVGcKwsw4= github.com/99designs/keyring v1.2.1 h1:tYLp1ULvO7i3fI5vE21ReQuj99QFSs7lGm0xWyJo87o= github.com/99designs/keyring v1.2.1/go.mod h1:fc+wB5KTk9wQ9sDx0kFXB3A0MaeGHM9AwRStKOQ5vOA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= github.com/AthenZ/athenz v1.12.13 h1:OhZNqZsoBXNrKBJobeUUEirPDnwt0HRo4kQMIO1UwwQ= github.com/AthenZ/athenz v1.12.13/go.mod h1:XXDXXgaQzXaBXnJX6x/bH4yF6eon2lkyzQZ0z/dxprE= github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= @@ -46,7 +48,6 @@ github.com/bits-and-blooms/bitset v1.4.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edY github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM= github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ= github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23/go.mod h1:bbYlZJ7hK1yFx9hf58LP0zeX7UjIGs20ufpu3evjr+s= -github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= @@ -63,6 +64,8 @@ github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7np github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY= +github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4= github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuAyr0= github.com/danieljoos/wincred v1.1.2/go.mod h1:GijpziifJoIBfYh+S7BbkdUTU4LfM+QnGqR5Vl2tAx0= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -73,14 +76,16 @@ github.com/dimfeld/httptreemux v5.0.1+incompatible h1:Qj3gVcDNoOthBAqftuD596rm4w github.com/dimfeld/httptreemux v5.0.1+incompatible/go.mod h1:rbUlSV+CCpv/SuqUTP/8Bk2O3LyUV436/yaRGkhP6Z0= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= -github.com/docker/docker v28.3.3+incompatible h1:Dypm25kh4rmk49v1eiVbsAtpAsYURjYkaKubwuBdxEI= -github.com/docker/docker v28.3.3+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/docker v28.5.1+incompatible h1:Bm8DchhSD2J6PsFzxC35TZo4TLGR2PdW/E69rU45NhM= +github.com/docker/docker v28.5.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/dvsekhvalnov/jose2go v1.6.0 h1:Y9gnSnP4qEI0+/uQkHvFXeD2PLPJeXEL+ySMEA2EjTY= github.com/dvsekhvalnov/jose2go v1.6.0/go.mod h1:QsHjhyTlD/lAVqn/NSbVZmSCGeDehTB/mPZadG+mhXU= +github.com/ebitengine/purego v0.8.4 h1:CF7LEKg5FFOsASUj0+QwaXf8Ht6TlFxg09+S9wz0omw= +github.com/ebitengine/purego v0.8.4/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/francoispqt/gojay v1.2.13 h1:d2m3sFjloqoIUQU3TsHBgj6qg/BVGlTBeHDUmyJnXKk= @@ -119,6 +124,7 @@ github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Z github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= @@ -135,6 +141,9 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg= github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk= +github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0 h1:TmHmbvxPmaegwhDubVz0lICL0J5Ka2vwTzhoePEXsGE= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0/go.mod h1:qztMSjm835F2bXf+5HKAPIS5qsmQDqZna/PgVt4rWtI= github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8ZofjG1Y75iExal34USq5p+wiN1tpie8IrU= github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0= github.com/hamba/avro/v2 v2.26.0 h1:IaT5l6W3zh7K67sMrT2+RreJyDTllBGVJm4+Hedk9qE= @@ -163,8 +172,8 @@ github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI= -github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= -github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY= github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= @@ -174,6 +183,8 @@ github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8 github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk= github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw= +github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs= github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= @@ -227,10 +238,8 @@ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4= -github.com/shirou/gopsutil/v3 v3.23.12/go.mod h1:1FrWgea594Jp7qmjHUUPlJDTPgcsb9mGnXDxavtikzM= -github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= -github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/shirou/gopsutil/v4 v4.25.6 h1:kLysI2JsKorfaFPcYmcJqbzROzsBWEOAtw6A7dIfqXs= +github.com/shirou/gopsutil/v4 v4.25.6/go.mod h1:PfybzyydfZcN+JMMjkF6Zb8Mq1A/VcogFFg7hj50W9c= github.com/sirupsen/logrus v1.9.4-0.20230606125235-dd1b4c2e81af h1:Sp5TG9f7K39yfB+If0vjp97vuT74F72r8hfRpP8jLU0= github.com/sirupsen/logrus v1.9.4-0.20230606125235-dd1b4c2e81af/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= @@ -250,8 +259,10 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/stripe/stripe-go/v74 v74.30.0 h1:0Kf0KkeFnY7iRhOwvTerX0Ia1BRw+eV1CVJ51mGYAUY= github.com/stripe/stripe-go/v74 v74.30.0/go.mod h1:f9L6LvaXa35ja7eyvP6GQswoaIPaBRvGAimAO+udbBw= -github.com/testcontainers/testcontainers-go v0.35.0 h1:uADsZpTKFAtp8SLK+hMwSaa+X+JiERHtd4sQAFmXeMo= -github.com/testcontainers/testcontainers-go v0.35.0/go.mod h1:oEVBj5zrfJTrgjwONs1SsRbnBtH9OKl+IGl3UMcr2B4= +github.com/testcontainers/testcontainers-go v0.40.0 h1:pSdJYLOVgLE8YdUY2FHQ1Fxu+aMnb6JfVz1mxk7OeMU= +github.com/testcontainers/testcontainers-go v0.40.0/go.mod h1:FSXV5KQtX2HAMlm7U3APNyLkkap35zNLxukw9oBi/MY= +github.com/testcontainers/testcontainers-go/modules/pulsar v0.40.0 h1:Xds52d07zwX31GqBLVxOlgFaq+/5r2ECJXNk/hRBjAM= +github.com/testcontainers/testcontainers-go/modules/pulsar v0.40.0/go.mod h1:B4EtUfhUI3kPhhuxFjjcVEoOl9U3LPHXKoR6w2xsXlg= github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU= github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI= github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk= @@ -262,8 +273,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= -github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw= -github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= @@ -272,10 +283,20 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 h1:Hf9xI/X go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0/go.mod h1:NfchwuyNoMcZ5MLHwPrODwUF1HWCXWrL31s8gSAdIKY= go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0 h1:Vh5HayB/0HHfOQA7Ctx69E/Y/DcQSMPpKANYVMQ7fBA= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0/go.mod h1:cpgtDBaqD/6ok/UG0jT15/uKjAY8mRA53diogHBg3UI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 h1:IeMeyr1aBvBiPVYihXIaeIZba6b8E1bYp7lbdxK8CQg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0/go.mod h1:oVdCUtjq9MK9BlS7TtucsQwUcXcymNiEDjgDD2jMtZU= go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE= go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= +go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI= +go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg= +go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc= +go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps= go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.opentelemetry.io/proto/otlp v1.3.1 h1:TrMUixzpM0yuc/znrFTP9MMRh8trP93mkCiDVeXrui0= +go.opentelemetry.io/proto/otlp v1.3.1/go.mod h1:0X1WI4de4ZsLrrJNLAQbFeLCm3T7yBkR0XqQ7niQU+8= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -289,8 +310,8 @@ golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4= -golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc= +golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= +golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -312,8 +333,8 @@ golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY= golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.27.0 h1:kb+q2PyFnEADO2IEF935ehFUXlWiNjJWtRNgBLSfbxQ= -golang.org/x/mod v0.27.0/go.mod h1:rWI627Fq0DEoudcK+MBkNkCe0EetEaDSwJJkCcjpazc= +golang.org/x/mod v0.28.0 h1:gQBtGhjxykdjY9YhZpSlZIsbnaE2+PgjfLWUQTnoZ1U= +golang.org/x/mod v0.28.0/go.mod h1:yfB/L0NOf/kmEbXjzCPOx1iK1fRutOydrCMsqRhEBxI= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -328,8 +349,8 @@ golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210520170846-37e1c6afe023/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= -golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +golang.org/x/net v0.45.0 h1:RLBg5JKixCy82FtLJpeNlVM0nrSqpCRYzVU1n8kj0tM= +golang.org/x/net v0.45.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -342,8 +363,8 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= -golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -353,25 +374,32 @@ golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210819135213-f52c844e1c1c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= -golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= +golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.34.0 h1:O/2T7POpk0ZZ7MAzMeWFSg6S5IpWd/RXDlM9hgM3DR4= -golang.org/x/term v0.34.0/go.mod h1:5jC53AEywhIVebHgPVeg0mj8OD3VO9OzclacVrqpaAw= +golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q= +golang.org/x/term v0.36.0/go.mod h1:Qu394IJq6V6dCBRgwqshf3mPF85AqzYEzofzRdZkWss= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= -golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= +golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= +golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= +golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= @@ -413,9 +441,16 @@ google.golang.org/genproto v0.0.0-20190801165951-fa694d86fc64/go.mod h1:DMBHOl98 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20190911173649-1774047e7e51/go.mod h1:IbNlFCBrqXvoKpeg0TB2l7cyZUmoaFKYIwrEpbDKLA8= google.golang.org/genproto v0.0.0-20191216164720-4f79533eabd1/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc= +google.golang.org/genproto v0.0.0-20250715232539-7130f93afb79 h1:Nt6z9UHqSlIdIGJdz6KhTIs2VRx/iOsA5iE8bmQNcxs= +google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79 h1:iOye66xuaAK0WnkPuhQPUFy8eJcmwUXqGGP3om6IxX8= +google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79/go.mod h1:HKJDgKsFUnv5VAGeQjz8kxcgDP0HoE0iZNp0OdZNlhE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79 h1:1ZwqphdOdWYXsUHgMpU/101nCtf/kSp9hOrcvFsnl10= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38= google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM= +google.golang.org/grpc v1.74.0 h1:sxRSkyLxlceWQiqDofxDot3d4u7DyoHPc7SBXMj8gGY= +google.golang.org/grpc v1.74.0/go.mod h1:NZUaK8dAMUfzhK6uxZ+9511LtOrk73UGWOFoNvz7z+s= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -432,6 +467,8 @@ gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/pulsar/connector/basic_conn_test.go b/pulsar/connector/basic_conn_test.go index 357b05b..cef002c 100644 --- a/pulsar/connector/basic_conn_test.go +++ b/pulsar/connector/basic_conn_test.go @@ -1,3 +1,5 @@ +//go:build integration + package connector import ( diff --git a/pulsar/connector/consumer_test.go b/pulsar/connector/consumer_test.go new file mode 100644 index 0000000..5aa2a2e --- /dev/null +++ b/pulsar/connector/consumer_test.go @@ -0,0 +1,114 @@ +package connector + +import ( + "testing" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/kubescape/messaging/pulsar/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewConsumerBuildsOptions(t *testing.T) { + messageChannel := make(chan pulsar.ConsumerMessage) + tests := []struct { + name string + options []CreateConsumerOption + wantTopic string + wantTopics []string + wantQueueSize int + wantMessageChan chan pulsar.ConsumerMessage + }{ + { + name: "single topic", + options: []CreateConsumerOption{ + WithTopic("test-topic"), + WithSubscriptionName("test-subscription"), + WithMessageChannel(messageChannel), + }, + wantTopic: "persistent://ca-messaging/test-namespace/test-topic", + wantMessageChan: messageChannel, + }, + { + name: "full topics", + options: []CreateConsumerOption{ + WithFullTopics([]TopicName{"persistent://other-tenant/other-namespace/test-topic"}), + WithSubscriptionName("test-subscription"), + }, + wantTopics: []string{"persistent://other-tenant/other-namespace/test-topic"}, + }, + { + name: "receiver queue size", + options: []CreateConsumerOption{ + WithTopic("test-topic"), + WithSubscriptionName("test-subscription"), + WithQueueSize(10), + }, + wantTopic: "persistent://ca-messaging/test-namespace/test-topic", + wantQueueSize: 10, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotOptions pulsar.ConsumerOptions + client := &mockClient{ + config: config.PulsarConfig{ + Tenant: "ca-messaging", + Namespace: "test-namespace", + MaxDeliveryAttempts: 2, + }, + subscribeFn: func(options pulsar.ConsumerOptions) (pulsar.Consumer, error) { + gotOptions = options + return &mockPulsarConsumer{}, nil + }, + } + + created, err := newConsumer(client, tt.options...) + require.NoError(t, err) + require.NotNil(t, created) + assert.Equal(t, tt.wantTopic, gotOptions.Topic) + if tt.wantTopics == nil { + assert.Empty(t, gotOptions.Topics) + } else { + assert.Equal(t, tt.wantTopics, gotOptions.Topics) + } + assert.Equal(t, "test-subscription", gotOptions.SubscriptionName) + assert.Equal(t, pulsar.Shared, gotOptions.Type) + assert.Equal(t, tt.wantQueueSize, gotOptions.ReceiverQueueSize) + assert.Equal(t, tt.wantMessageChan, gotOptions.MessageChannel) + require.NotNil(t, gotOptions.DLQ) + assert.Equal(t, uint32(2), gotOptions.DLQ.MaxDeliveries) + }) + } +} + +func TestConsumerReconsumeLaterPanicsWhenRetryDisabled(t *testing.T) { + wrapped := consumer{ + Consumer: &mockPulsarConsumer{}, + options: createConsumerOptions{}, + } + + assert.PanicsWithValue( + t, + "reconsumeLater called on consumer without retry enabled option set to true", + func() { wrapped.ReconsumeLater(&mockMessage{properties: map[string]string{}}, time.Millisecond) }, + ) +} + +func TestConsumerReconsumeLaterPanicsWhenSafeRetryRequired(t *testing.T) { + wrapped := consumer{ + Consumer: &mockPulsarConsumer{}, + options: createConsumerOptions{ + retryEnabled: true, + forceDLQSafeRetry: true, + }, + } + + assert.PanicsWithValue( + t, + "reconsumeLater: when forceDLQSafeRetry option is true ReconsumeLaterDLQSafe must be used", + func() { wrapped.ReconsumeLater(&mockMessage{properties: map[string]string{}}, time.Millisecond) }, + ) +} diff --git a/pulsar/connector/mocks_test.go b/pulsar/connector/mocks_test.go new file mode 100644 index 0000000..274528b --- /dev/null +++ b/pulsar/connector/mocks_test.go @@ -0,0 +1,62 @@ +package connector + +import ( + "context" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/kubescape/messaging/pulsar/config" +) + +type mockClient struct { + Client + config config.PulsarConfig + createProducerFn func(pulsar.ProducerOptions) (pulsar.Producer, error) + subscribeFn func(pulsar.ConsumerOptions) (pulsar.Consumer, error) +} + +func (m *mockClient) GetConfig() config.PulsarConfig { + return m.config +} + +func (m *mockClient) CreateProducer(options pulsar.ProducerOptions) (pulsar.Producer, error) { + return m.createProducerFn(options) +} + +func (m *mockClient) Subscribe(options pulsar.ConsumerOptions) (pulsar.Consumer, error) { + return m.subscribeFn(options) +} + +type mockProducer struct { + pulsar.Producer + sendFn func(context.Context, *pulsar.ProducerMessage) (pulsar.MessageID, error) +} + +func (m *mockProducer) Send(ctx context.Context, message *pulsar.ProducerMessage) (pulsar.MessageID, error) { + return m.sendFn(ctx, message) +} + +type mockPulsarConsumer struct { + pulsar.Consumer + reconsumeLaterFn func(pulsar.Message, time.Duration) +} + +func (m *mockPulsarConsumer) ReconsumeLater(message pulsar.Message, delay time.Duration) { + if m.reconsumeLaterFn != nil { + m.reconsumeLaterFn(message, delay) + } +} + +type mockMessage struct { + pulsar.Message + properties map[string]string + payload []byte +} + +func (m *mockMessage) Properties() map[string]string { + return m.properties +} + +func (m *mockMessage) Payload() []byte { + return m.payload +} diff --git a/pulsar/connector/producer_test.go b/pulsar/connector/producer_test.go new file mode 100644 index 0000000..fb53e6c --- /dev/null +++ b/pulsar/connector/producer_test.go @@ -0,0 +1,111 @@ +package connector + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/kubescape/messaging/pulsar/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewProducerBuildsTopic(t *testing.T) { + tests := []struct { + name string + options []CreateProducerOption + wantTopic string + }{ + { + name: "default namespace", + options: []CreateProducerOption{WithProducerTopic("test-topic")}, + wantTopic: "persistent://ca-messaging/test-namespace/test-topic", + }, + { + name: "overridden namespace", + options: []CreateProducerOption{ + WithProducerTopic("test-topic"), + WithProducerNamespace("other-tenant", "other-namespace"), + }, + wantTopic: "persistent://other-tenant/other-namespace/test-topic", + }, + { + name: "full persistent topic", + options: []CreateProducerOption{WithProducerFullTopic("persistent://tenant/namespace/topic")}, + wantTopic: "persistent://tenant/namespace/topic", + }, + { + name: "full non-persistent topic", + options: []CreateProducerOption{WithProducerFullTopic("non-persistent://tenant/namespace/topic")}, + wantTopic: "non-persistent://tenant/namespace/topic", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotOptions pulsar.ProducerOptions + pulsarProducer := &mockProducer{} + client := &mockClient{ + config: config.PulsarConfig{Tenant: "ca-messaging", Namespace: "test-namespace"}, + createProducerFn: func(options pulsar.ProducerOptions) (pulsar.Producer, error) { + gotOptions = options + return pulsarProducer, nil + }, + } + + created, err := newProducer(client, tt.options...) + require.NoError(t, err) + require.NotNil(t, created) + assert.Equal(t, tt.wantTopic, gotOptions.Topic) + }) + } +} + +func TestNewProducerWrapsClientError(t *testing.T) { + client := &mockClient{ + config: config.PulsarConfig{Tenant: "tenant", Namespace: "namespace"}, + createProducerFn: func(pulsar.ProducerOptions) (pulsar.Producer, error) { + return nil, errors.New("invalid topic") + }, + } + + producer, err := newProducer(client, WithProducerFullTopic("invalid-topic")) + + require.Nil(t, producer) + require.ErrorContains(t, err, "CreateProducer: failed to create producer: invalid topic") +} + +func TestProduceMessageForwardsOptions(t *testing.T) { + type contextKey struct{} + ctx := context.WithValue(context.Background(), contextKey{}, "value") + properties := map[string]string{"source": "unit-test"} + delay := 250 * time.Millisecond + var gotContext context.Context + var gotMessage *pulsar.ProducerMessage + producer := &mockProducer{ + sendFn: func(ctx context.Context, message *pulsar.ProducerMessage) (pulsar.MessageID, error) { + gotContext = ctx + gotMessage = message + return pulsar.NewMessageID(1, 2, 3, 4), nil + }, + } + + err := ProduceMessage( + producer, + WithContext(ctx), + WithMessageToSend("test message"), + WithMessageKey("message-key"), + WithProperties(properties), + WithDelay(delay), + ) + + require.NoError(t, err) + assert.Same(t, ctx, gotContext) + require.NotNil(t, gotMessage) + assert.JSONEq(t, `"test message"`, string(gotMessage.Payload)) + assert.Equal(t, properties, gotMessage.Properties) + assert.Equal(t, "message-key", gotMessage.Key) + assert.Equal(t, delay, gotMessage.DeliverAfter) +} diff --git a/pulsar/connector/scripts/pulsar.sh b/pulsar/connector/scripts/pulsar.sh deleted file mode 100755 index 769fa65..0000000 --- a/pulsar/connector/scripts/pulsar.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/bin/bash - -if command -v docker >/dev/null 2>&1; then - cr='docker' -elif command -v podman >/dev/null 2>&1; then - cr='podman' -fi - -app_port=${2:-6650} # Use this as default port if no argument is provided - -admin_port=${3:-8080} # Use this as default port if no argument is provided -echo "All arguments: $@" - -app_port=%d -admin_port=%d -container_name=%s - -echo "Starting pulsar using $cr on port $app_port and admin port $admin_port" - -$cr run --name=$container_name -d -p $app_port:6650 -p $admin_port:8080 docker.io/apachepulsar/pulsar:2.11.0 bin/pulsar standalone diff --git a/pulsar/connector/scripts/pulsar_stop.sh b/pulsar/connector/scripts/pulsar_stop.sh deleted file mode 100755 index 6e710e1..0000000 --- a/pulsar/connector/scripts/pulsar_stop.sh +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/bash - -if command -v docker >/dev/null 2>&1; then - cr='docker' -elif command -v podman >/dev/null 2>&1; then - cr='podman' -fi -container_name=%s -$cr rm -f $container_name || true - -# if [ $RET_VAL -eq '0' ]; then -# ps aux | grep rootle | awk '{print $2}' | xargs kill -9 || true -# fi \ No newline at end of file diff --git a/pulsar/connector/suite_test.go b/pulsar/connector/suite_test.go index 4b40e25..87c3c4d 100644 --- a/pulsar/connector/suite_test.go +++ b/pulsar/connector/suite_test.go @@ -1,14 +1,12 @@ +//go:build integration + package connector import ( "context" - _ "embed" "fmt" - "os/exec" "strconv" "sync" - - "net/http" "testing" "time" @@ -16,6 +14,7 @@ import ( "github.com/kubescape/messaging/pulsar/common/utils" "github.com/kubescape/messaging/pulsar/config" + "github.com/kubescape/messaging/pulsar/internal/pulsartest" "golang.org/x/sync/errgroup" "github.com/apache/pulsar-client-go/pulsar" @@ -23,18 +22,6 @@ import ( "github.com/stretchr/testify/suite" ) -const ( - pulsarClientPort = 6651 - pulsarAdminPort = 8081 -) - -var ( - //go:embed scripts/pulsar.sh - pulsarDockerCommand string - //go:embed scripts/pulsar_stop.sh - pulsarStopCommand string -) - func TestBasicConnection(t *testing.T) { suite.Run(t, new(MainTestSuite)) } @@ -44,13 +31,24 @@ type MainTestSuite struct { failOnUnconsummedMessages bool defaultTestConfig config.PulsarConfig pulsarClient Client - shutdownFunc func() + broker *pulsartest.Broker + cleanupOnce sync.Once + cleanupErr error } func (suite *MainTestSuite) SetupSuite() { + broker, err := pulsartest.Start(context.Background()) + if err != nil { + suite.FailNow("failed to start Pulsar", err.Error()) + } + suite.broker = broker + suite.T().Cleanup(func() { + suite.Require().NoError(suite.cleanup()) + }) + suite.defaultTestConfig = config.PulsarConfig{ - URL: fmt.Sprintf("pulsar://localhost:%d", pulsarClientPort), - AdminUrl: fmt.Sprintf("http://localhost:%d", pulsarAdminPort), + URL: broker.URL, + AdminUrl: broker.AdminURL, Tenant: "ca-messaging", Namespace: "test-namespace", Clusters: []string{"standalone"}, @@ -61,29 +59,30 @@ func (suite *MainTestSuite) SetupSuite() { x, _ := json.Marshal(suite.defaultTestConfig) fmt.Println(string(x)) - //start pulsar - suite.startPulsar() - - var err error - //ensure pulsar connection + // Ensure the wrapper can connect and initialize its namespaces. suite.pulsarClient, err = NewClient(WithConfig(&suite.defaultTestConfig)) if err != nil { suite.FailNow("failed to create pulsar client", err.Error()) } - suite.shutdownFunc = func() { - defer func() { - suite.pulsarClient.Close() - }() - } } func (suite *MainTestSuite) TearDownSuite() { suite.T().Log("tear down suite") - suite.shutdownFunc() - out, err := exec.Command("/bin/sh", "-c", fmt.Sprintf(pulsarStopCommand, "basic-suite")).CombinedOutput() - if err != nil { - suite.FailNow("failed to stop pulsar", err.Error(), string(out)) - } + suite.Require().NoError(suite.cleanup()) +} + +func (suite *MainTestSuite) cleanup() error { + suite.cleanupOnce.Do(func() { + if suite.pulsarClient != nil { + suite.pulsarClient.Close() + } + if suite.broker != nil { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + suite.cleanupErr = suite.broker.Terminate(ctx) + } + }) + return suite.cleanupErr } func (suite *MainTestSuite) SetupTest() { @@ -151,46 +150,6 @@ func (suite *MainTestSuite) TearDownTest() { suite.failOnUnconsummedMessages = false } -func (suite *MainTestSuite) TestCreateConsumer() { - //create consumer - chan1 := make(chan pulsar.ConsumerMessage) - - consumer, err := suite.pulsarClient.NewConsumer(WithMessageChannel(chan1), WithTopic("test-topic"), WithSubscriptionName("test-subscription")) - if err != nil { - suite.FailNow("failed to create consumer", err.Error()) - } - defer consumer.Close() -} - -func (suite *MainTestSuite) TestCreateConsumerWithFullTopics() { - chan1 := make(chan pulsar.ConsumerMessage) - - consumer, err := suite.pulsarClient.NewConsumer(WithMessageChannel(chan1), WithFullTopics([]TopicName{"persistent://ca-messaging/test-namespace/test-topic"}), WithSubscriptionName("test-subscription")) - if err != nil { - suite.FailNow("failed to create consumer", err.Error()) - } - defer consumer.Close() -} - -func (suite *MainTestSuite) TestCreateConsumerWithQueueSize() { - chan1 := make(chan pulsar.ConsumerMessage) - - consumer, err := suite.pulsarClient.NewConsumer(WithMessageChannel(chan1), WithTopic("test-topic"), WithSubscriptionName("test-subscription"), WithQueueSize(10)) - if err != nil { - suite.FailNow("failed to create consumer", err.Error()) - } - defer consumer.Close() -} - -func (suite *MainTestSuite) TestCreateProducer() { - //create producer - producer, err := suite.pulsarClient.NewProducer(WithProducerTopic("test-topic")) - if err != nil { - suite.FailNow("failed to create producer", err.Error()) - } - defer producer.Close() -} - func (suite *MainTestSuite) TestSetAndGetMaxUnackedMessagesOnConsumer() { topic := suite.defaultTestConfig.Tenant + "/" + suite.defaultTestConfig.Namespace + "/cloud-scanner-tasks-v2" fullTopic := "persistent://" + topic @@ -239,94 +198,14 @@ func (suite *MainTestSuite) TestSetAndGetMaxUnackedMessagesOnConsumer() { fmt.Printf("maxUnackedMessagesOnConsumer after remove: %d\n", maxUnacked) } -func (suite *MainTestSuite) TestCreateProducerFullTopic() { - //create producer - producer, err := suite.pulsarClient.NewProducer(WithProducerFullTopic("persistent://ca-messaging/test-namespace/test-topic")) - if err != nil { - suite.FailNow("failed to create producer", err.Error()) - } - defer producer.Close() -} - -func (suite *MainTestSuite) TestCreateProducerFullTopicNonPersistent() { - //create producer - BuildNonPersistentTopic("test-t", "test-ns", "test-topic") - producer, err := suite.pulsarClient.NewProducer(WithProducerFullTopic("non-persistent://ca-messaging/test-namespace/test-topic")) - if err != nil { - suite.FailNow("failed to create producer", err.Error()) - } - defer producer.Close() -} - func (suite *MainTestSuite) TestCreateProducerFullTopicInvalid() { - //create producer producer, err := suite.pulsarClient.NewProducer(WithProducerFullTopic("est-t/test-ns/test-topic")) - if err == nil { - suite.FailNow("created invalid topic producer") - } + suite.Require().Error(err) if producer != nil { - defer producer.Close() + producer.Close() } } -func (suite *MainTestSuite) TestProduceMessage() { - //create producer - producer, err := suite.pulsarClient.NewProducer(WithProducerTopic("test-topic")) - if err != nil { - suite.FailNow("failed to create producer", err.Error()) - } - defer producer.Close() - - //produce message - msg := "test message" - if err := ProduceMessage(producer, WithMessageToSend(msg), WithContext(context.Background())); err != nil { - suite.FailNow("failed to produce message", err.Error()) - } -} - -func (suite *MainTestSuite) startPulsar() { - suite.T().Log("stopping existing pulsar container") - exec.Command("/bin/sh", "-c", fmt.Sprintf(pulsarStopCommand, "basic-suite")).Run() - time.Sleep(2 * time.Second) - - suite.T().Log("starting pulsar") - out, err := exec.Command("/bin/sh", "-c", fmt.Sprintf(pulsarDockerCommand, pulsarClientPort, pulsarAdminPort, "basic-suite")).CombinedOutput() - suite.T().Logf("start script output:\n%s", string(out)) - if err != nil { - suite.FailNow("failed to start pulsar", err.Error(), string(out)) - } - req, err := http.NewRequest(http.MethodGet, suite.defaultTestConfig.AdminUrl+"/admin/v2/brokers/ready", nil) - if err != nil { - suite.FailNow("failed to create request", err.Error()) - } - suite.T().Log("waiting for pulsar to start") - client := http.Client{Timeout: 5 * time.Second} - var lastErr error - for i := 0; i < 120; i++ { - resp, err := client.Do(req) - if err == nil { - if resp.StatusCode == http.StatusOK { - suite.T().Log("pulsar started") - resp.Body.Close() - return - } - lastErr = fmt.Errorf("unexpected status code: %d", resp.StatusCode) - resp.Body.Close() - } else { - lastErr = err - } - if i%10 == 0 { - suite.T().Logf("waiting for pulsar... attempt %d/120, last error: %v", i, lastErr) - } - time.Sleep(2 * time.Second) - } - dockerPs, _ := exec.Command("/bin/sh", "-c", "docker ps -a 2>&1").CombinedOutput() - dockerLogs, _ := exec.Command("/bin/sh", "-c", "docker logs --tail 50 basic-suite 2>&1").CombinedOutput() - podmanPs, _ := exec.Command("/bin/sh", "-c", "podman ps -a 2>&1").CombinedOutput() - podmanLogs, _ := exec.Command("/bin/sh", "-c", "podman logs --tail 50 basic-suite 2>&1").CombinedOutput() - suite.FailNow("failed to start pulsar", fmt.Sprintf("last error: %v\ndocker ps:\n%s\ndocker logs:\n%s\npodman ps:\n%s\npodman logs:\n%s", lastErr, string(dockerPs), string(dockerLogs), string(podmanPs), string(podmanLogs))) -} - func loadJson[T any](jsonBytes []byte) T { var obj T if err := json.Unmarshal(jsonBytes, &obj); err != nil { @@ -807,101 +686,3 @@ func (suite *MainTestSuite) TestSafeReconsumeLaterWithDuration() { //fail on test teardown if there are unconsumed messages suite.failOnUnconsummedMessages = true } - -func (suite *MainTestSuite) TestReconsumeLaterPanicOnRetryDisabled() { - defer func() { - if r := recover(); r == nil { - suite.T().Errorf("The code did not panic") - } else { - suite.T().Logf("Recovered in TestReconsumeLaterPanics: %v", r) - } - }() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - producer, err := CreateTestProducer(ctx, suite.pulsarClient) - if err != nil { - suite.FailNow(err.Error(), "create producer") - } - if producer == nil { - suite.FailNow("producer is nil") - } - defer producer.Close() - - noRetrayEnabledConsumer, err := CreateTestConsumer(ctx, suite.pulsarClient) - if err != nil { - suite.FailNow(err.Error()) - } - defer noRetrayEnabledConsumer.Close() - //produce - if _, err := producer.Send(ctx, &pulsar.ProducerMessage{Payload: []byte(suite.T().Name())}); err != nil { - suite.FailNow(err.Error(), "send payload") - } - testMsg := func(msg pulsar.Message) { - if msg == nil { - suite.FailNow("msg is nil") - } - if string(msg.Payload()) != suite.T().Name() { - suite.FailNow("unexpected payload") - } - } - //consume - testConsumerCtx, consumerCancel := context.WithTimeout(ctx, time.Second*time.Duration(2)) - defer consumerCancel() - msg, err := noRetrayEnabledConsumer.Receive(testConsumerCtx) - if err != nil { - suite.FailNow(err.Error(), "receive payload") - } - testMsg(msg) - noRetrayEnabledConsumer.ReconsumeLater(msg, time.Millisecond) - suite.FailNow("should panic on call reconsume when retry option is not enabled") -} - -func (suite *MainTestSuite) TestReconsumeLaterPanicOnUnSafeReconsume() { - defer func() { - if r := recover(); r == nil { - suite.T().Errorf("The code did not panic") - } else { - suite.T().Logf("Recovered in TestReconsumeLaterPanicOnUnSafeReconsume: %v", r) - } - }() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - producer, err := CreateTestProducer(ctx, suite.pulsarClient) - if err != nil { - suite.FailNow(err.Error(), "create producer") - } - if producer == nil { - suite.FailNow("producer is nil") - } - defer producer.Close() - reconsumeLaterOptions := []CreateConsumerOption{ - WithRetryEnable(true, true, time.Second*2), - } - safeOnlyEnabledConsumer, err := CreateTestConsumer(ctx, suite.pulsarClient, reconsumeLaterOptions...) - if err != nil { - suite.FailNow(err.Error()) - } - defer safeOnlyEnabledConsumer.Close() - //produce - if _, err := producer.Send(ctx, &pulsar.ProducerMessage{Payload: []byte(suite.T().Name())}); err != nil { - suite.FailNow(err.Error(), "send payload") - } - testMsg := func(msg pulsar.Message) { - if msg == nil { - suite.FailNow("msg is nil") - } - if string(msg.Payload()) != suite.T().Name() { - suite.FailNow("unexpected payload") - } - } - //consume - testConsumerCtx, consumerCancel := context.WithTimeout(ctx, time.Second*time.Duration(2)) - defer consumerCancel() - msg, err := safeOnlyEnabledConsumer.Receive(testConsumerCtx) - if err != nil { - suite.FailNow(err.Error(), "receive payload") - } - testMsg(msg) - safeOnlyEnabledConsumer.ReconsumeLater(msg, time.Millisecond) - suite.FailNow("should panic on call ReconsumeLater when safe only option is set") -} diff --git a/pulsar/internal/pulsartest/container.go b/pulsar/internal/pulsartest/container.go new file mode 100644 index 0000000..2369d57 --- /dev/null +++ b/pulsar/internal/pulsartest/container.go @@ -0,0 +1,141 @@ +package pulsartest + +import ( + "context" + "fmt" + "io" + "net" + "net/url" + "strconv" + "strings" + "sync" + "time" + + "github.com/testcontainers/testcontainers-go" + testcontainerspulsar "github.com/testcontainers/testcontainers-go/modules/pulsar" + "github.com/testcontainers/testcontainers-go/wait" +) + +const ( + pulsarImage = "apachepulsar/pulsar:4.0.13" + pulsarPort = "6650/tcp" + pulsarAdminPort = "8080/tcp" + maxLogBytes = 64 * 1024 +) + +// Broker is a disposable Pulsar broker for integration tests. +type Broker struct { + container *testcontainerspulsar.Container + + URL string + AdminURL string + Port int + AdminPort int + + terminateOnce sync.Once + terminateErr error +} + +// Start creates a Pulsar standalone container and returns its mapped endpoints. +func Start(ctx context.Context) (*Broker, error) { + container, err := testcontainerspulsar.Run( + ctx, + pulsarImage, + testcontainers.WithEnv(map[string]string{ + "PULSAR_MEM": "-Xms512m -Xmx512m -XX:MaxDirectMemorySize=512m", + }), + // v0.40.0 waits for a Pulsar 2.x log line that 4.x no longer emits. + // Keep its HTTP readiness check and pair it with the broker port instead. + testcontainers.WithWaitStrategy( + wait.ForHTTP("/admin/v2/clusters"). + WithPort(pulsarAdminPort). + WithResponseMatcher(func(body io.Reader) bool { + contents, readErr := io.ReadAll(body) + return readErr == nil && strings.TrimSpace(string(contents)) == `["standalone"]` + }). + WithStartupTimeout(2*time.Minute), + wait.ForListeningPort(pulsarPort).WithStartupTimeout(2*time.Minute), + ), + ) + if err != nil { + return nil, containerError(ctx, container, err) + } + + brokerURL, err := container.BrokerURL(ctx) + if err != nil { + return nil, cleanupAfterStartError(container, fmt.Errorf("resolve Pulsar broker URL: %w", err)) + } + adminURL, err := container.HTTPServiceURL(ctx) + if err != nil { + return nil, cleanupAfterStartError(container, fmt.Errorf("resolve Pulsar admin URL: %w", err)) + } + port, err := portFromURL(brokerURL) + if err != nil { + return nil, cleanupAfterStartError(container, err) + } + adminPort, err := portFromURL(adminURL) + if err != nil { + return nil, cleanupAfterStartError(container, err) + } + + return &Broker{ + container: container, + URL: brokerURL, + AdminURL: adminURL, + Port: port, + AdminPort: adminPort, + }, nil +} + +// Terminate stops and removes the broker. Repeated calls return the first result. +func (b *Broker) Terminate(ctx context.Context) error { + if b == nil || b.container == nil { + return nil + } + b.terminateOnce.Do(func() { + b.terminateErr = b.container.Terminate(ctx) + }) + return b.terminateErr +} + +func portFromURL(rawURL string) (int, error) { + parsed, err := url.Parse(rawURL) + if err != nil { + return 0, fmt.Errorf("parse Pulsar URL %q: %w", rawURL, err) + } + _, portString, err := net.SplitHostPort(parsed.Host) + if err != nil { + return 0, fmt.Errorf("parse Pulsar address %q: %w", parsed.Host, err) + } + port, err := strconv.Atoi(portString) + if err != nil { + return 0, fmt.Errorf("parse Pulsar port %q: %w", portString, err) + } + return port, nil +} + +func containerError(ctx context.Context, container *testcontainerspulsar.Container, startErr error) error { + if container == nil { + return fmt.Errorf("start Pulsar container: %w", startErr) + } + + logs, err := container.Logs(ctx) + if err != nil { + return cleanupAfterStartError(container, fmt.Errorf("start Pulsar container: %w (read logs: %v)", startErr, err)) + } + defer logs.Close() + contents, readErr := io.ReadAll(io.LimitReader(logs, maxLogBytes)) + if readErr != nil { + return cleanupAfterStartError(container, fmt.Errorf("start Pulsar container: %w (read logs: %v)", startErr, readErr)) + } + return cleanupAfterStartError(container, fmt.Errorf("start Pulsar container: %w\ncontainer logs:\n%s", startErr, contents)) +} + +func cleanupAfterStartError(container *testcontainerspulsar.Container, startErr error) error { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := container.Terminate(cleanupCtx); err != nil { + return fmt.Errorf("%w; terminate container: %v", startErr, err) + } + return startErr +} diff --git a/pulsar/test/kill_port_process.go b/pulsar/test/kill_port_process.go deleted file mode 100644 index ecbf6ac..0000000 --- a/pulsar/test/kill_port_process.go +++ /dev/null @@ -1,69 +0,0 @@ -package test - -import ( - "bufio" - "fmt" - "os" - "os/exec" - "runtime" - "strconv" - "strings" - "syscall" -) - -const windows = "windows" - -func killPortProcess(targetPort int) error { - processes, err := getProcessesForPort(targetPort) - if err != nil { - return err - } - - for _, pid := range processes { - fmt.Printf("Killing process on port %d with PID %d\n", targetPort, pid) - - switch runtime.GOOS { - case windows: - killCmd := exec.Command("taskkill", "/F", "/PID", fmt.Sprint(pid)) - if err := killCmd.Run(); err != nil { - return err - } - default: - process, err := os.FindProcess(pid) - if err != nil { - return err - } - err = process.Signal(syscall.SIGTERM) - } - } - return nil -} - -func getProcessesForPort(targetPort int) ([]int, error) { - var cmd *exec.Cmd - processes := make([]int, 0) - switch runtime.GOOS { - case windows: - cmd = exec.Command("cmd", "/c", "netstat", "-ano", "|", "findstr", fmt.Sprintf(":%d", targetPort)) - default: - cmd = exec.Command("sh", "-c", fmt.Sprintf("lsof -iTCP:%d -n -P | awk '/.*LISTEN.*/ { print $2 }'", targetPort)) - } - output, err := cmd.Output() - if err != nil { - return nil, err - } - - scanner := bufio.NewScanner(strings.NewReader(string(output))) - for scanner.Scan() { - pidStr := scanner.Text() - pidStr = strings.TrimSpace(strings.Split(pidStr, " ")[0]) // Extracting PID from the output - - pid, err := strconv.Atoi(pidStr) - if err != nil { - continue // Not a valid number, skip - } - processes = append(processes, pid) - } - - return processes, scanner.Err() -} diff --git a/pulsar/test/scripts/pulsar.sh b/pulsar/test/scripts/pulsar.sh deleted file mode 100755 index 769fa65..0000000 --- a/pulsar/test/scripts/pulsar.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/bin/bash - -if command -v docker >/dev/null 2>&1; then - cr='docker' -elif command -v podman >/dev/null 2>&1; then - cr='podman' -fi - -app_port=${2:-6650} # Use this as default port if no argument is provided - -admin_port=${3:-8080} # Use this as default port if no argument is provided -echo "All arguments: $@" - -app_port=%d -admin_port=%d -container_name=%s - -echo "Starting pulsar using $cr on port $app_port and admin port $admin_port" - -$cr run --name=$container_name -d -p $app_port:6650 -p $admin_port:8080 docker.io/apachepulsar/pulsar:2.11.0 bin/pulsar standalone diff --git a/pulsar/test/scripts/pulsar_stop.sh b/pulsar/test/scripts/pulsar_stop.sh deleted file mode 100755 index 6e710e1..0000000 --- a/pulsar/test/scripts/pulsar_stop.sh +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/bash - -if command -v docker >/dev/null 2>&1; then - cr='docker' -elif command -v podman >/dev/null 2>&1; then - cr='podman' -fi -container_name=%s -$cr rm -f $container_name || true - -# if [ $RET_VAL -eq '0' ]; then -# ps aux | grep rootle | awk '{print $2}' | xargs kill -9 || true -# fi \ No newline at end of file diff --git a/pulsar/test/suite.go b/pulsar/test/suite.go index 9aac7b3..a675490 100644 --- a/pulsar/test/suite.go +++ b/pulsar/test/suite.go @@ -1,45 +1,50 @@ package test import ( - _ "embed" - "errors" + "context" + "encoding/json" "fmt" - "net" - "os/exec" - - "net/http" + "sync" "time" - "encoding/json" - "github.com/kubescape/messaging/pulsar/config" "github.com/kubescape/messaging/pulsar/connector" + "github.com/kubescape/messaging/pulsar/internal/pulsartest" "github.com/stretchr/testify/suite" ) -const ( - pulsarKAURL = "%s/admin/v2/brokers/ready" -) - -//go:embed scripts/pulsar.sh -var startPulsarScript string - -//go:embed scripts/pulsar_stop.sh -var pulsarStopCommand string - type PulsarTestSuite struct { suite.Suite DefaultTestConfig config.PulsarConfig Client connector.Client - AppPortStart int - AdminPortStart int - shutdownFunc func() + // AppPortStart is retained for source compatibility. SetupSuite replaces it + // with the dynamically mapped Pulsar broker port. + AppPortStart int + // AdminPortStart is retained for source compatibility. SetupSuite replaces it + // with the dynamically mapped Pulsar admin port. + AdminPortStart int + broker *pulsartest.Broker + cleanupOnce sync.Once + cleanupErr error } func (suite *PulsarTestSuite) SetupSuite() { suite.T().Log("setup suite") + broker, err := pulsartest.Start(context.Background()) + if err != nil { + suite.FailNow("failed to start Pulsar", err.Error()) + } + suite.broker = broker + suite.AppPortStart = broker.Port + suite.AdminPortStart = broker.AdminPort + suite.T().Cleanup(func() { + suite.Require().NoError(suite.cleanup()) + }) + suite.DefaultTestConfig = config.PulsarConfig{ + URL: broker.URL, + AdminUrl: broker.AdminURL, Tenant: "ca-messaging", Namespace: "test-namespace", Clusters: []string{"standalone"}, @@ -47,56 +52,32 @@ func (suite *PulsarTestSuite) SetupSuite() { RedeliveryDelaySeconds: 0, } - randomContainerName := fmt.Sprintf("pulsar-test-%d-%d", suite.AdminPortStart, time.Now().UnixNano()) - if suite.AppPortStart == 0 { - suite.AppPortStart = 6650 - } - if suite.AdminPortStart == 0 { - suite.AdminPortStart = 8080 - } - //start pulsar - suite.startPulsar(randomContainerName) - x, _ := json.Marshal(suite.DefaultTestConfig) fmt.Println(string(x)) - var err error - //ensure pulsar connection + // Ensure the wrapper can connect and initialize its namespaces. suite.Client, err = connector.NewClient(connector.WithConfig(&suite.DefaultTestConfig)) if err != nil { suite.FailNow("failed to create pulsar client", err.Error()) } - suite.shutdownFunc = func() { - defer func() { - suite.Client.Close() - formmatedScript := fmt.Sprintf(pulsarStopCommand, randomContainerName) - outbytes, err := exec.Command("/bin/sh", "-c", formmatedScript).CombinedOutput() - if err != nil { - suite.FailNow("failed to stop pulsar", err.Error(), string(outbytes)) - } - }() - } -} -func (suite *PulsarTestSuite) checkPulsarIsAlive() bool { - kaURL := fmt.Sprintf(pulsarKAURL, suite.DefaultTestConfig.AdminUrl) - fmt.Println("pulsar admin", kaURL) - req, err := http.NewRequest(http.MethodGet, kaURL, nil) - if err != nil { - suite.FailNow("failed to create request", err.Error()) - } - client := http.Client{Timeout: 5 * time.Second} - resp, err := client.Do(req) - if err == nil && resp.StatusCode == http.StatusOK { - suite.T().Log("pulsar started") - resp.Body.Close() - return true - } - return false } + func (suite *PulsarTestSuite) TearDownSuite() { suite.T().Log("tear down suite") - suite.shutdownFunc() - suite.Assert().NoError(killPortProcess(suite.AppPortStart)) - suite.Assert().NoError(killPortProcess(suite.AdminPortStart)) + suite.Require().NoError(suite.cleanup()) +} + +func (suite *PulsarTestSuite) cleanup() error { + suite.cleanupOnce.Do(func() { + if suite.Client != nil { + suite.Client.Close() + } + if suite.broker != nil { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + suite.cleanupErr = suite.broker.Terminate(ctx) + } + }) + return suite.cleanupErr } func (suite *PulsarTestSuite) SetupTest() { @@ -110,63 +91,3 @@ func (suite *PulsarTestSuite) TearDownTest() { suite.FailNow("failed to clear all messages", err.Error()) } } - -func findFreePort(rangeStart, rangeEnd int) (int, error) { - for port := rangeStart; port <= rangeEnd; port++ { - address := fmt.Sprintf("localhost:%d", port) - conn, err := net.DialTimeout("tcp", address, 1*time.Second) - if conn != nil { - conn.Close() - } - if err != nil { // port is available since we got no response - return port, nil - } - conn.Close() - } - return 0, errors.New("no free port found") -} - -func (suite *PulsarTestSuite) startPulsar(contName string) { - suite.T().Log("stopping existing pulsar container") - exec.Command("/bin/sh", "-c", fmt.Sprintf(pulsarStopCommand, contName)).Run() - suite.T().Log("starting pulsar") - - pulsarAppPort, err := findFreePort(suite.AppPortStart, suite.AppPortStart+100) - if err != nil { - suite.FailNow("failed to find free port", err.Error()) - } - suite.DefaultTestConfig.URL = fmt.Sprintf("pulsar://localhost:%d", pulsarAppPort) - suite.AppPortStart = pulsarAppPort - pulsarAdminPort, err := findFreePort(suite.AdminPortStart, suite.AdminPortStart+100) - if err != nil { - suite.FailNow("failed to find free port for pulsar admin", err.Error()) - } - suite.DefaultTestConfig.AdminUrl = fmt.Sprintf("http://localhost:%d", pulsarAdminPort) - suite.AdminPortStart = pulsarAdminPort - - formattedScript := fmt.Sprintf(startPulsarScript, pulsarAppPort, pulsarAdminPort, contName) - out, err := exec.Command("/bin/sh", "-c", formattedScript).CombinedOutput() - if err != nil { - if exitErr, ok := err.(*exec.ExitError); ok { - suite.FailNow("failed to start pulsar", err.Error(), string(exitErr.Stderr), string(out)) - } - suite.FailNow("failed to start pulsar", err.Error(), string(out)) - } - suite.T().Log("waiting for pulsar to start") - for i := 0; i < 120; i++ { - isAlive := suite.checkPulsarIsAlive() - if isAlive { - return - } - time.Sleep(2 * time.Second) - } - logs, _ := exec.Command("/bin/sh", "-c", fmt.Sprintf("docker logs --tail 50 %s || podman logs --tail 50 %s", contName, contName)).CombinedOutput() - formmatedScript := fmt.Sprintf(pulsarStopCommand, contName) - outbytes, err := exec.Command("/bin/sh", "-c", formmatedScript).CombinedOutput() - if err != nil { - fmt.Println(string(outbytes), err.Error()) - } - killPortProcess(suite.AppPortStart) - killPortProcess(suite.AdminPortStart) - suite.FailNow("failed to start pulsar", fmt.Sprintf("container logs:\n%s", string(logs))) -} diff --git a/pulsar/test/suite_integration_test.go b/pulsar/test/suite_integration_test.go new file mode 100644 index 0000000..679d7fd --- /dev/null +++ b/pulsar/test/suite_integration_test.go @@ -0,0 +1,40 @@ +//go:build integration + +package test_test + +import ( + "net/url" + "strconv" + "testing" + + messagingtest "github.com/kubescape/messaging/pulsar/test" + "github.com/stretchr/testify/suite" +) + +type publicPulsarSuite struct { + messagingtest.PulsarTestSuite +} + +func TestPublicPulsarTestSuite(t *testing.T) { + suite.Run(t, new(publicPulsarSuite)) +} + +func (suite *publicPulsarSuite) TestClientAndMappedEndpointsAreReady() { + suite.Require().NotNil(suite.Client) + suite.Require().Positive(suite.AppPortStart) + suite.Require().Positive(suite.AdminPortStart) + suite.Equal(strconv.Itoa(suite.AppPortStart), endpointPort(suite.DefaultTestConfig.URL)) + suite.Equal(strconv.Itoa(suite.AdminPortStart), endpointPort(suite.DefaultTestConfig.AdminUrl)) + + partitions, err := suite.Client.TopicPartitions("test-topic") + suite.Require().NoError(err) + suite.NotEmpty(partitions) +} + +func endpointPort(rawURL string) string { + parsed, err := url.Parse(rawURL) + if err != nil { + return "" + } + return parsed.Port() +}