From 0bb526b76fdddbc034255d1bf64c9b2419f2ebdc Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 14:58:13 +0800 Subject: [PATCH 1/6] create cdc gh action Signed-off-by: pingyu --- .github/workflows/ci-cdc.yml | 61 +++++++++++++++++++++++++++++++++ cdc/Makefile | 6 +++- cdc/tools/check/check-errdoc.sh | 8 +++-- 3 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/ci-cdc.yml diff --git a/.github/workflows/ci-cdc.yml b/.github/workflows/ci-cdc.yml new file mode 100644 index 00000000..5b5541d6 --- /dev/null +++ b/.github/workflows/ci-cdc.yml @@ -0,0 +1,61 @@ +name: CI +on: + push: + branches: main + paths: cdc/** + pull_request: + branches: main + paths: cdc/** + +permissions: + contents: read + +jobs: + cdc-lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-go@v2 + with: + go-version: '1.16.1' + - name: make check-static + shell: bash + run: | + cd cdc/ + make check-static + cdc-check-gosec: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-go@v2 + with: + go-version: '1.16.1' + - name: make check-gosec + shell: bash + run: | + cd cdc + make check-gosec + cdc-build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-go@v2 + with: + go-version: '1.16.1' + - name: make cdc + shell: bash + run: | + cd cdc + make cdc + cdc-unit-test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-go@v2 + with: + go-version: '1.16.1' + - name: make unit_test + shell: bash + run: | + cd br + make unit_test diff --git a/cdc/Makefile b/cdc/Makefile index 6b81bea1..5d93dfc8 100644 --- a/cdc/Makefile +++ b/cdc/Makefile @@ -12,7 +12,7 @@ export PATH := $(CURDIR)/bin:$(path_to_add):$(PATH) SHELL := /usr/bin/env bash -TEST_DIR := /tmp/tidb_cdc_test +TEST_DIR ?= /tmp/tikv_cdc_test GO := GO111MODULE=on go ifeq (${CDC_ENABLE_VENDOR}, 1) @@ -155,6 +155,10 @@ check-merge-conflicts: @echo "check-merge-conflicts" @./scripts/check-merge-conflicts.sh +check-gosec: + $(GO) install github.com/securego/gosec/v2/cmd/gosec@v2.9.1 + gosec -fmt=junit-xml -out=results.xml -stdout -verbose=text -exclude=G103,G104,G204,G304,G307,G401,G404,G501,G505,G601 ./... + check: check-copyright fmt check-static tidy errdoc check-leaktest-added check-merge-conflicts unit_test_coverage: diff --git a/cdc/tools/check/check-errdoc.sh b/cdc/tools/check/check-errdoc.sh index e3eaabd6..e0817f7b 100755 --- a/cdc/tools/check/check-errdoc.sh +++ b/cdc/tools/check/check-errdoc.sh @@ -16,6 +16,8 @@ set -euo pipefail cd -P . -cp errors.toml /tmp/errors.toml.before -./tools/bin/errdoc-gen --source . --module github.com/migration/cdc/cdc --output errors.toml --ignore proto -diff -q errors.toml /tmp/errors.toml.before +TMP=$(mktemp -t errors.toml.before.XXXXXXXXXX) + +cp errors.toml ${TMP} +./tools/bin/errdoc-gen --source . --module github.com/tikv/migration/cdc --output errors.toml --ignore proto +diff -q errors.toml ${TMP} From 45428c76cf651c87b11df760ba32cab955483d50 Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 15:11:32 +0800 Subject: [PATCH 2/6] ci fix Signed-off-by: pingyu --- .github/workflows/ci-cdc.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci-cdc.yml b/.github/workflows/ci-cdc.yml index 5b5541d6..114ba218 100644 --- a/.github/workflows/ci-cdc.yml +++ b/.github/workflows/ci-cdc.yml @@ -33,7 +33,7 @@ jobs: - name: make check-gosec shell: bash run: | - cd cdc + cd cdc/ make check-gosec cdc-build: runs-on: ubuntu-latest @@ -45,7 +45,7 @@ jobs: - name: make cdc shell: bash run: | - cd cdc + cd cdc/ make cdc cdc-unit-test: runs-on: ubuntu-latest @@ -57,5 +57,5 @@ jobs: - name: make unit_test shell: bash run: | - cd br + cd cdc/ make unit_test From 9203b6f4ad3133884642ef6a28807906f810d37a Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 17:39:30 +0800 Subject: [PATCH 3/6] fix CI Signed-off-by: pingyu --- cdc/go.mod | 3 + cdc/go.sum | 27 ++ .../move_table/conf/diff_config.toml | 29 -- .../move_table/conf/workload | 13 - .../integration_tests/move_table/main.go | 326 ------------------ cdc/tests/integration_tests/move_table/run.sh | 65 ---- cdc/tests/utils/cdc_state_checker/reactor.go | 6 +- .../utils/many_sorters_test/many_sorters.go | 169 --------- .../many_sorters_test/many_sorters_test.go | 20 -- .../utils/many_sorters_test/sorter/ticdc_lock | 0 .../utils/sorter_stress_test/sorter_stress.go | 171 --------- 11 files changed, 33 insertions(+), 796 deletions(-) delete mode 100644 cdc/tests/integration_tests/move_table/conf/diff_config.toml delete mode 100644 cdc/tests/integration_tests/move_table/conf/workload delete mode 100644 cdc/tests/integration_tests/move_table/main.go delete mode 100644 cdc/tests/integration_tests/move_table/run.sh delete mode 100644 cdc/tests/utils/many_sorters_test/many_sorters.go delete mode 100644 cdc/tests/utils/many_sorters_test/many_sorters_test.go delete mode 100644 cdc/tests/utils/many_sorters_test/sorter/ticdc_lock delete mode 100644 cdc/tests/utils/sorter_stress_test/sorter_stress.go diff --git a/cdc/go.mod b/cdc/go.mod index 5095ee27..e9f45a16 100644 --- a/cdc/go.mod +++ b/cdc/go.mod @@ -8,6 +8,7 @@ require ( github.com/Shopify/sarama v1.27.2 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 github.com/benbjohnson/clock v1.1.0 + github.com/bradleyjkemp/grpc-tools v0.2.7 github.com/cenkalti/backoff v2.2.1+incompatible github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e github.com/cockroachdb/pebble v0.0.0-20210719141320-8c3bd06debb5 @@ -30,6 +31,7 @@ require ( github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/integralist/go-findroot v0.0.0-20160518114804-ac90681525dc github.com/jarcoal/httpmock v1.0.5 + github.com/jmoiron/sqlx v1.3.5 github.com/json-iterator/go v1.1.12 // indirect github.com/linkedin/goavro/v2 v2.9.8 github.com/mattn/go-colorable v0.1.11 // indirect @@ -77,6 +79,7 @@ require ( golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba google.golang.org/grpc v1.40.0 sigs.k8s.io/yaml v1.2.0 // indirect + upper.io/db.v3 v3.8.0+incompatible ) replace ( diff --git a/cdc/go.sum b/cdc/go.sum index 65b6e53f..ecace1f1 100644 --- a/cdc/go.sum +++ b/cdc/go.sum @@ -123,6 +123,9 @@ github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kB github.com/bketelsen/crypt v0.0.4/go.mod h1:aI6NrJ0pMGgvZKL1iVgXLnfIFJtfV+bKCoqOes/6LfM= github.com/blacktear23/go-proxyprotocol v0.0.0-20180807104634-af7a81e8dd0d h1:rQlvB2AYWme2bIB18r/SipGiMEVJYE9U0z+MGoU/LtQ= github.com/blacktear23/go-proxyprotocol v0.0.0-20180807104634-af7a81e8dd0d/go.mod h1:VKt7CNAQxpFpSDz3sXyj9hY/GbVsQCr0sB3w59nE7lU= +github.com/bradleyjkemp/cupaloy/v2 v2.6.0/go.mod h1:bm7JXdkRd4BHJk9HpwqAI8BoAY1lps46Enkdqw6aRX0= +github.com/bradleyjkemp/grpc-tools v0.2.7 h1:oeEqBpWpvbtlpqgcy3dhE3JHmY08IZuuAHJthZjdkoA= +github.com/bradleyjkemp/grpc-tools v0.2.7/go.mod h1:TAZ3YBe/P438YK2SoPp8MVwYCKz2wUePIOxpZoEq9fg= github.com/cakturk/go-netstat v0.0.0-20200220111822-e5b49efee7a5 h1:BjkPE3785EwPhhyuFkbINB+2a1xATwk8SNDWnJiD41g= github.com/cakturk/go-netstat v0.0.0-20200220111822-e5b49efee7a5/go.mod h1:jtAfVaU/2cu1+wdSRPWE2c1N2qeAA3K4RH9pYgqwets= github.com/carlmjohnson/flagext v0.21.0 h1:/c4uK3ie786Z7caXLcIMvePNSSiH3bQVGDvmGLMme60= @@ -206,6 +209,8 @@ github.com/danjacques/gofslock v0.0.0-20191023191349-0a45f885bc37/go.mod h1:DC3J github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f h1:U5y3Y5UE0w7amNe7Z5G/twsBW0KEalRQXZzf8ufSh9I= +github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f/go.mod h1:xH/i4TFMt8koVQZ6WFms69WAsDWr2XsYL3Hkl7jkoLE= github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4= github.com/dgraph-io/ristretto v0.0.1 h1:cJwdnj42uV8Jg4+KLrYovLiCgIfz9wtWm6E6KA+1tLs= github.com/dgraph-io/ristretto v0.0.1/go.mod h1:T40EBc7CJke8TkpiYfGGKAeFjSaxuFXhuXRyumBd6RE= @@ -445,6 +450,7 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+ github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM= github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= +github.com/gordonklaus/ineffassign v0.0.0-20200309095847-7953dde2c7bf/go.mod h1:cuNKsD1zp2v6XfE/orVX2QE1LC+i254ceGcVeDT3pTU= github.com/gorilla/handlers v1.4.2/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ= github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH4= github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q= @@ -454,6 +460,7 @@ github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI= github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So= github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= +github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= @@ -499,6 +506,8 @@ github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334/go.mod h1:SK73t github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= github.com/imkira/go-interpol v1.1.0/go.mod h1:z0h2/2T3XF8kyEPpRgJ3kmNv+C43p+I/CoI+jC3w2iA= +github.com/improbable-eng/grpc-web v0.13.0 h1:7XqtaBWaOCH0cVGKHyvhtcuo6fgW32Y10yRKrDHFHOc= +github.com/improbable-eng/grpc-web v0.13.0/go.mod h1:6hRR09jOEG81ADP5wCQju1z71g6OL4eEvELdran/3cs= github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/influxdata/tdigest v0.0.1/go.mod h1:Z0kXnxzbTC2qrx4NaIzYkE1k66+6oEDQTvL95hQFh5Y= @@ -515,6 +524,7 @@ github.com/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem github.com/jcmturner/gofork v1.0.0/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o= github.com/jedib0t/go-pretty/v6 v6.2.2/go.mod h1:+nE9fyyHGil+PuISTCrp7avEdo6bqoMwqZnuiK2r2a0= github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI= +github.com/jhump/protoreflect v1.7.0/go.mod h1:RZkzh7Hi9J7qT/sPlWnJ/UwZqCJvciFxKDA0UCeltSM= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/jinzhu/now v1.1.2/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= @@ -523,6 +533,8 @@ github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9Y github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U= +github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g= +github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ= github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg= github.com/joho/sqltocsv v0.0.0-20210428211105-a6d6801d59df h1:Zrb0IbuLOGHL7nrO2WrcuNWgDTlzFv3zY69QMx4ggQE= github.com/joho/sqltocsv v0.0.0-20210428211105-a6d6801d59df/go.mod h1:mAVCUAYtW9NG31eB30umMSLKcDt6mCUWSjoSn5qBh0k= @@ -585,6 +597,8 @@ github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdA github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII= github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w= github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY= +github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0= +github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo= github.com/linkedin/goavro/v2 v2.9.8 h1:jN50elxBsGBDGVDEKqUlDuU1cFwJ11K/yrJCBMe/7Wg= github.com/linkedin/goavro/v2 v2.9.8/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA= github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= @@ -618,6 +632,8 @@ github.com/mattn/go-runewidth v0.0.12/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRC github.com/mattn/go-shellwords v1.0.12 h1:M2zGm7EW6UQJvDeQxo4T51eKPurbeFbe8WtebGE2xrk= github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y= github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI= +github.com/mattn/go-sqlite3 v1.14.6 h1:dNPt6NO46WmLVt2DLNpwczCmdV5boIZ6g/tlDrlRUbg= +github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw= github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= @@ -649,6 +665,8 @@ github.com/montanaflynn/stats v0.5.0 h1:2EkzeTSqBB4V4bJwWrt5gIIrZmpJBcoIRGS2kWLg github.com/montanaflynn/stats v0.5.0/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= github.com/moul/http2curl v1.0.0/go.mod h1:8UbvGypXm98wA/IqH45anm5Y2Z6ep6O31QGOAZ3H0fQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f h1:KUppIJq7/+SVif2QVs3tOP0zanoHgBEVAwHxUSIzRqU= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/nats-io/nats.go v1.8.1/go.mod h1:BrFz9vVn0fU3AcH9Vn4Kd7W0NpJ651tD5omQ3M8LwxM= github.com/nats-io/nkeys v0.0.2/go.mod h1:dab7URMsZm6Z/jp9Z5UGa87Uutgc2mVpXLC4B7TDb/4= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= @@ -661,6 +679,7 @@ github.com/ngaut/sync2 v0.0.0-20141008032647-7a24ed77b2ef h1:K0Fn+DoFqNqktdZtdV3 github.com/ngaut/sync2 v0.0.0-20141008032647-7a24ed77b2ef/go.mod h1:7WjlapSfwQyo6LNmIvEWzsW1hbBQfpUO4JWnuQRmva8= github.com/nicksnyder/go-i18n v1.10.0/go.mod h1:HrK7VCrbOvQoUAQ7Vpy7i87N7JZZZ7R2xBGjv0j365Q= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= +github.com/nishanths/predeclared v0.0.0-20200524104333-86fad755b4d3/go.mod h1:nt3d53pc1VYcphSCIaYAJtnPYnr3Zyn8fMq2wvPGPso= github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78= github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= @@ -812,6 +831,7 @@ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8= github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE= +github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik= github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU= github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -837,6 +857,7 @@ github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd/go.mod h1:TrYk7fJV github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88= +github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE= github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= @@ -1341,6 +1362,7 @@ golang.org/x/tools v0.0.0-20200331025713-a30bf2db82d4/go.mod h1:Sl4aGygMT6LrqrWc golang.org/x/tools v0.0.0-20200501065659-ab2804fb9c9d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20200522201501-cb1345f3a375/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= @@ -1409,6 +1431,7 @@ google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCID google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc= google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c= google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc= +google.golang.org/genproto v0.0.0-20170818010345-ee236bd376b0/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20180518175338-11a468237815/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20181004005441-af9cb2a35e7f/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= @@ -1465,6 +1488,7 @@ google.golang.org/genproto v0.0.0-20210813162853-db860fec028c/go.mod h1:cFeNkxwy google.golang.org/genproto v0.0.0-20210825212027-de86158e7fda h1:iT5uhT54PtbqUsWddv/nnEWdE5e/MTr+Nv3vjxlBP1A= google.golang.org/genproto v0.0.0-20210825212027-de86158e7fda/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY= google.golang.org/grpc v0.0.0-20180607172857-7a6a684ca69e/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= +google.golang.org/grpc v1.8.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.12.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38= @@ -1537,6 +1561,7 @@ gopkg.in/jcmturner/gokrb5.v7 v7.5.0 h1:a9tsXlIDD9SKxotJMK3niV7rPZAJeX2aD/0yg3qlI gopkg.in/jcmturner/gokrb5.v7 v7.5.0/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= +gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce h1:xcEWjVhvbDy+nHP67nPDDpbYrY+ILlfndk4bRioVHaU= gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA= gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8= gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k= @@ -1597,3 +1622,5 @@ sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0 h1:ucqkfp sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod h1:hI742Nqp5OhwiqlzhgfbWU4mW4yO10fP+LoT9WOswdU= sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 h1:e1sMhtVq9AfcEy8AXNb8eSg6gbzfdpYhoNqnPJa+GzI= sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:L5q+DGLGOQFpo1snNEkLOJT2d1YTW66rWNzatr3He1k= +upper.io/db.v3 v3.8.0+incompatible h1:XNeEO2vQRVqq70M98ghzq6M30F5Bzo+99ess5v+eVYw= +upper.io/db.v3 v3.8.0+incompatible/go.mod h1:FgTdD24eBjJAbPKsQSiHUNgXjOR4Lub3u1UMHSIh82Y= diff --git a/cdc/tests/integration_tests/move_table/conf/diff_config.toml b/cdc/tests/integration_tests/move_table/conf/diff_config.toml deleted file mode 100644 index c6aeddb7..00000000 --- a/cdc/tests/integration_tests/move_table/conf/diff_config.toml +++ /dev/null @@ -1,29 +0,0 @@ -# diff Configuration. - -check-thread-count = 4 - -export-fix-sql = true - -check-struct-only = false - -[task] - output-dir = "/tmp/tidb_cdc_test/move_table/sync_diff/output" - - source-instances = ["mysql1"] - - target-instance = "tidb0" - - target-check-tables = ["move_table.usertable*"] - -[data-sources] -[data-sources.mysql1] - host = "127.0.0.1" - port = 4000 - user = "root" - password = "" - -[data-sources.tidb0] - host = "127.0.0.1" - port = 3306 - user = "root" - password = "" diff --git a/cdc/tests/integration_tests/move_table/conf/workload b/cdc/tests/integration_tests/move_table/conf/workload deleted file mode 100644 index 128a6404..00000000 --- a/cdc/tests/integration_tests/move_table/conf/workload +++ /dev/null @@ -1,13 +0,0 @@ -threadcount=10 -recordcount=6000 -operationcount=0 -workload=core - -readallfields=true - -readproportion=0 -updateproportion=0 -scanproportion=0 -insertproportion=0 - -requestdistribution=uniform diff --git a/cdc/tests/integration_tests/move_table/main.go b/cdc/tests/integration_tests/move_table/main.go deleted file mode 100644 index d9f2ac47..00000000 --- a/cdc/tests/integration_tests/move_table/main.go +++ /dev/null @@ -1,326 +0,0 @@ -// Copyright 2020 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -// This is a program that drives the CDC cluster to move a table -package main - -import ( - "bytes" - "context" - "encoding/json" - "flag" - "fmt" - "io" - "io/ioutil" - "net/http" - "os" - "strings" - "time" - - "github.com/pingcap/errors" - "github.com/pingcap/log" - "github.com/tikv/migration/cdc/cdc/model" - cerrors "github.com/tikv/migration/cdc/pkg/errors" - "github.com/tikv/migration/cdc/pkg/etcd" - "github.com/tikv/migration/cdc/pkg/httputil" - "github.com/tikv/migration/cdc/pkg/retry" - "github.com/tikv/migration/cdc/pkg/security" - "go.etcd.io/etcd/clientv3" - "go.etcd.io/etcd/pkg/logutil" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - "google.golang.org/grpc" - "google.golang.org/grpc/backoff" -) - -var ( - pd = flag.String("pd", "http://127.0.0.1:2379", "PD address and port") - logLevel = flag.String("log-level", "debug", "Set log level of the logger") -) - -const ( - maxCheckSourceEmptyRetries = 30 -) - -// This program moves all tables replicated by a certain capture to other captures, -// and makes sure that the original capture becomes empty. -func main() { - flag.Parse() - if strings.ToLower(*logLevel) == "debug" { - log.SetLevel(zapcore.DebugLevel) - } - - log.Info("table mover started") - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute) - defer cancel() - - cluster, err := newCluster(ctx, *pd) - if err != nil { - log.Fatal("failed to create cluster info", zap.Error(err)) - } - err = retry.Do(ctx, func() error { - err := cluster.refreshInfo(ctx) - if err != nil { - log.Warn("error refreshing cluster info", zap.Error(err)) - } - - log.Info("task status", zap.Reflect("status", cluster.captures)) - - if len(cluster.captures) <= 1 { - return errors.New("too few captures") - } - return nil - }, retry.WithBackoffBaseDelay(100), retry.WithMaxTries(20), retry.WithIsRetryableErr(cerrors.IsRetryableError)) - - if err != nil { - log.Fatal("Fail to get captures", zap.Error(err)) - } - - var sourceCapture string - - for capture, tables := range cluster.captures { - if len(tables) == 0 { - continue - } - sourceCapture = capture - break - } - - var targetCapture string - - for candidateCapture := range cluster.captures { - if candidateCapture != sourceCapture { - targetCapture = candidateCapture - } - } - - if targetCapture == "" { - log.Fatal("no target, unexpected") - } - - // move all tables to another capture - for _, table := range cluster.captures[sourceCapture] { - err = moveTable(ctx, cluster.ownerAddr, table.Changefeed, targetCapture, table.ID) - if err != nil { - log.Warn("failed to move table", zap.Error(err)) - continue - } - - log.Info("moved table successful", zap.Int64("tableID", table.ID)) - } - - log.Info("all tables are moved", zap.String("sourceCapture", sourceCapture), zap.String("targetCapture", targetCapture)) - - for counter := 0; counter < maxCheckSourceEmptyRetries; counter++ { - err := retry.Do(ctx, func() error { - return cluster.refreshInfo(ctx) - }, retry.WithBackoffBaseDelay(100), retry.WithMaxTries(5+1), retry.WithIsRetryableErr(cerrors.IsRetryableError)) - if err != nil { - log.Warn("error refreshing cluster info", zap.Error(err)) - } - - tables, ok := cluster.captures[sourceCapture] - if !ok { - log.Warn("source capture is gone", zap.String("sourceCapture", sourceCapture)) - break - } - - if len(tables) == 0 { - log.Info("source capture is now empty", zap.String("sourceCapture", sourceCapture)) - break - } - - if counter != maxCheckSourceEmptyRetries { - log.Debug("source capture is not empty, will try again", zap.String("sourceCapture", sourceCapture)) - time.Sleep(time.Second * 10) - } else { - // non-zero error code indicates failed test. - os.Exit(1) - } - } -} - -type tableInfo struct { - ID int64 - Changefeed string -} - -type cluster struct { - ownerAddr string - captures map[string][]*tableInfo - cdcEtcdCli etcd.CDCEtcdClient -} - -func newCluster(ctx context.Context, pd string) (*cluster, error) { - logConfig := logutil.DefaultZapLoggerConfig - logConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel) - - etcdCli, err := clientv3.New(clientv3.Config{ - Endpoints: []string{pd}, - TLS: nil, - Context: ctx, - LogConfig: &logConfig, - DialTimeout: 5 * time.Second, - DialOptions: []grpc.DialOption{ - grpc.WithInsecure(), - grpc.WithBlock(), - grpc.WithConnectParams(grpc.ConnectParams{ - Backoff: backoff.Config{ - BaseDelay: time.Second, - Multiplier: 1.1, - Jitter: 0.1, - MaxDelay: 3 * time.Second, - }, - MinConnectTimeout: 3 * time.Second, - }), - }, - }) - if err != nil { - return nil, errors.Trace(err) - } - - ret := &cluster{ - ownerAddr: "", - captures: nil, - cdcEtcdCli: etcd.NewCDCEtcdClient(ctx, etcdCli), - } - - log.Info("new cluster initialized") - - return ret, nil -} - -func (c *cluster) refreshInfo(ctx context.Context) error { - ownerID, err := c.cdcEtcdCli.GetOwnerID(ctx, etcd.CaptureOwnerKey) - if err != nil { - return errors.Trace(err) - } - - log.Debug("retrieved owner ID", zap.String("ownerID", ownerID)) - - captureInfo, err := c.cdcEtcdCli.GetCaptureInfo(ctx, ownerID) - if err != nil { - return errors.Trace(err) - } - - log.Debug("retrieved owner addr", zap.String("ownerAddr", captureInfo.AdvertiseAddr)) - c.ownerAddr = captureInfo.AdvertiseAddr - - _, changefeeds, err := c.cdcEtcdCli.GetChangeFeeds(ctx) - if err != nil { - return errors.Trace(err) - } - if len(changefeeds) == 0 { - return errors.New("No changefeed") - } - - log.Debug("retrieved changefeeds", zap.Reflect("changefeeds", changefeeds)) - var changefeed string - for k := range changefeeds { - changefeed = k - break - } - - c.captures = make(map[string][]*tableInfo) - _, captures, err := c.cdcEtcdCli.GetCaptures(ctx) - if err != nil { - return errors.Trace(err) - } - for _, capture := range captures { - c.captures[capture.ID] = make([]*tableInfo, 0) - processorDetails, err := queryProcessor(c.ownerAddr, changefeed, capture.ID) - if err != nil { - return errors.Trace(err) - } - - log.Debug("retrieved processor details", - zap.String("changefeed-id", changefeed), - zap.String("capture-id", capture.ID), - zap.Any("processor-detail", processorDetails)) - for _, tableID := range processorDetails.Tables { - c.captures[capture.ID] = append(c.captures[capture.ID], &tableInfo{ - ID: tableID, - Changefeed: changefeed, - }) - } - } - return nil -} - -// queryProcessor invokes the following API to get the mapping from -// captureIDs to tableIDs: -// GET /api/v1/processors/{changefeed_id}/{capture_id} -func queryProcessor( - apiEndpoint string, - changefeed string, - captureID string, -) (*model.ProcessorDetail, error) { - httpClient, err := httputil.NewClient(&security.Credential{ /* no TLS */ }) - if err != nil { - return nil, errors.Trace(err) - } - - requestURL := fmt.Sprintf("http://%s/api/v1/processors/%s/%s", apiEndpoint, changefeed, captureID) - httpClient.Timeout = 3 * time.Second - resp, err := httpClient.Get(requestURL) - if err != nil { - return nil, errors.Trace(err) - } - defer func() { - _ = resp.Body.Close() - }() - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, errors.Trace( - errors.Errorf("HTTP API returned error status: %d, url: %s", resp.StatusCode, requestURL)) - } - - bodyBytes, err := ioutil.ReadAll(resp.Body) - if err != nil { - return nil, errors.Trace(err) - } - - var ret model.ProcessorDetail - err = json.Unmarshal(bodyBytes, &ret) - if err != nil { - return nil, errors.Trace(err) - } - - return &ret, nil -} - -func moveTable(ctx context.Context, ownerAddr string, changefeed string, target string, tableID int64) error { - formStr := fmt.Sprintf("cf-id=%s&target-cp-id=%s&table-id=%d", changefeed, target, tableID) - log.Debug("preparing HTTP API call to owner", zap.String("formStr", formStr)) - rd := bytes.NewReader([]byte(formStr)) - req, err := http.NewRequestWithContext(ctx, "POST", "http://"+ownerAddr+"/capture/owner/move_table", rd) - if err != nil { - return errors.Trace(err) - } - - req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err := http.DefaultClient.Do(req) - if err != nil { - return errors.Trace(err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - body, err := io.ReadAll(resp.Body) - if err != nil { - return errors.Trace(err) - } - log.Warn("http error", zap.ByteString("body", body)) - return errors.New(resp.Status) - } - - return nil -} diff --git a/cdc/tests/integration_tests/move_table/run.sh b/cdc/tests/integration_tests/move_table/run.sh deleted file mode 100644 index edc9e1da..00000000 --- a/cdc/tests/integration_tests/move_table/run.sh +++ /dev/null @@ -1,65 +0,0 @@ -#!/bin/bash - -set -eu - -CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -source $CUR/../_utils/test_prepare -WORK_DIR=$OUT_DIR/$TEST_NAME -CDC_BINARY=cdc.test -SINK_TYPE=$1 - -function run() { - # TODO: remove after kafka-consumer/main.go is fixed - if [ "$SINK_TYPE" == "kafka" ]; then - return - fi - - rm -rf $WORK_DIR && mkdir -p $WORK_DIR - - start_tidb_cluster --workdir $WORK_DIR - - cd $WORK_DIR - - start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) - run_sql "CREATE DATABASE move_table;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - go-ycsb load mysql -P $CUR/conf/workload -p mysql.host=${UP_TIDB_HOST} -p mysql.port=${UP_TIDB_PORT} -p mysql.user=root -p mysql.db=move_table - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --loglevel "debug" --logsuffix "1" --addr 127.0.0.1:8300 - - TOPIC_NAME="ticdc-move-table-test-$RANDOM" - case $SINK_TYPE in - kafka) SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&kafka-version=${KAFKA_VERSION}&max-message-bytes=10485760" ;; - *) SINK_URI="mysql://normal:123456@127.0.0.1:3306/?max-txn-row=1" ;; - esac - - run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" - if [ "$SINK_TYPE" == "kafka" ]; then - run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" - fi - - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --loglevel "debug" --logsuffix "2" --addr 127.0.0.1:8301 - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --loglevel "debug" --logsuffix "3" --addr 127.0.0.1:8302 - - # Add a check table to reduce check time, or if we check data with sync diff - # directly, there maybe a lot of diff data at first because of the incremental scan - run_sql "CREATE table move_table.check1(id int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - check_table_exists "move_table.USERTABLE" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} - - cd $CUR - GO111MODULE=on go run main.go 2>&1 | tee $WORK_DIR/tester.log & - cd $WORK_DIR - - check_table_exists "move_table.check1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 90 - check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - run_sql "truncate table move_table.USERTABLE" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - run_sql "CREATE table move_table.check2(id int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - check_table_exists "move_table.check2" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 90 - check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - - cleanup_process $CDC_BINARY -} - -trap stop_tidb_cluster EXIT -run $* -check_logs $WORK_DIR -echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>" diff --git a/cdc/tests/utils/cdc_state_checker/reactor.go b/cdc/tests/utils/cdc_state_checker/reactor.go index 778ce47c..d76e55ad 100644 --- a/cdc/tests/utils/cdc_state_checker/reactor.go +++ b/cdc/tests/utils/cdc_state_checker/reactor.go @@ -82,12 +82,12 @@ func (r *cdcMonitReactor) verifyStartTs() error { actualCheckpointTs := cStatus.CheckpointTs for captureID, status := range statuses { - for tableID, operation := range status.Operation { + for keyspanID, operation := range status.Operation { if operation.Status != model.OperFinished && !operation.Delete { - startTs := status.Tables[tableID].StartTs + startTs := status.KeySpans[keyspanID].StartTs if startTs < actualCheckpointTs { return errors.Errorf("startTs too small, globalCkpt = %d, startTs = %d, table = %d, capture = %s, cfid = %s", - actualCheckpointTs, startTs, tableID, captureID, changfeedID) + actualCheckpointTs, startTs, keyspanID, captureID, changfeedID) } } } diff --git a/cdc/tests/utils/many_sorters_test/many_sorters.go b/cdc/tests/utils/many_sorters_test/many_sorters.go deleted file mode 100644 index d2ff106d..00000000 --- a/cdc/tests/utils/many_sorters_test/many_sorters.go +++ /dev/null @@ -1,169 +0,0 @@ -// Copyright 2020 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "context" - "flag" - "fmt" - "math/rand" - "net/http" - "os" - "strings" - "sync/atomic" - - "github.com/pingcap/errors" - "github.com/pingcap/failpoint" - "github.com/pingcap/log" - "github.com/tikv/migration/cdc/cdc/model" - "github.com/tikv/migration/cdc/cdc/sorter" - "github.com/tikv/migration/cdc/cdc/sorter/unified" - "github.com/tikv/migration/cdc/pkg/config" - cerrors "github.com/tikv/migration/cdc/pkg/errors" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - "golang.org/x/sync/errgroup" -) - -var ( - sorterDir = flag.String("dir", "./sorter", "temporary directory used for sorting") - numSorters = flag.Int("num-sorters", 256, "number of instances of sorters") - numEvents = flag.Int("num-events-per-sorter", 10000, "number of events sent to a sorter") - percentageResolves = flag.Int("percentage-resolve-events", 70, "percentage of resolved events") -) - -func main() { - flag.Parse() - err := failpoint.Enable("github.com/tikv/migration/cdc/cdc/sorter/unified/sorterDebug", "return(true)") - if err != nil { - log.Fatal("Could not enable failpoint", zap.Error(err)) - } - log.SetLevel(zapcore.DebugLevel) - - conf := config.GetDefaultServerConfig() - conf.Sorter = &config.SorterConfig{ - NumConcurrentWorker: 8, - ChunkSizeLimit: 1 * 1024 * 1024 * 1024, - MaxMemoryPressure: 60, - MaxMemoryConsumption: 16 * 1024 * 1024 * 1024, - NumWorkerPoolGoroutine: 16, - } - config.StoreGlobalServerConfig(conf) - - go func() { - _ = http.ListenAndServe("localhost:6060", nil) - }() - - err = os.MkdirAll(*sorterDir, 0o700) - if err != nil { - log.Error("sorter_stress_test:", zap.Error(err)) - } - - sorters := make([]sorter.EventSorter, *numSorters) - ctx0, cancel := context.WithCancel(context.Background()) - errg, ctx := errgroup.WithContext(ctx0) - - errg.Go(func() error { - return unified.RunWorkerPool(ctx) - }) - - var finishCount int32 - for i := 0; i < *numSorters; i++ { - sorters[i], err = unified.NewUnifiedSorter(*sorterDir, - "test-cf", - fmt.Sprintf("test-%d", i), - model.TableID(i), - "0.0.0.0:0") - if err != nil { - log.Panic("many_sorters", zap.Error(err)) - } - finalI := i - - // run sorter - errg.Go(func() error { - return printError(sorters[finalI].Run(ctx)) - }) - - // run producer - errg.Go(func() error { - for j := 0; j < *numEvents; j++ { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - - ev := generateEvent(uint64(finalI), uint64(j<<5)) - sorters[finalI].AddEntry(ctx, ev) - } - sorters[finalI].AddEntry(ctx, model.NewResolvedPolymorphicEvent(uint64(finalI), uint64(((*numEvents)<<5)+1))) - return nil - }) - - // run consumer - errg.Go(func() error { - for { - var ev *model.PolymorphicEvent - select { - case <-ctx.Done(): - return ctx.Err() - case ev = <-sorters[finalI].Output(): - } - - if ev.CRTs == uint64(((*numEvents)<<5)+1) { - log.Info("Sorter finished", zap.Int("sorter-id", finalI)) - if atomic.AddInt32(&finishCount, 1) == int32(*numSorters) { - log.Info("Many Sorters test finished, cancelling all goroutines") - cancel() - } - return nil - } - } - }) - } - - _ = printError(errg.Wait()) - if atomic.LoadInt32(&finishCount) == int32(*numSorters) { - log.Info("Test was successful!") - } -} - -func generateEvent(region uint64, ts uint64) *model.PolymorphicEvent { - r := rand.Int() % 100 - if r < *percentageResolves { - return model.NewResolvedPolymorphicEvent(region, ts) - } - return model.NewPolymorphicEvent(&model.RawKVEntry{ - OpType: model.OpTypePut, - Key: []byte("keykeykey"), - Value: []byte("valuevaluevalue"), - OldValue: nil, - StartTs: ts - 5, - CRTs: ts, - RegionID: region, - }) -} - -// printError is a helper for tracing errors on function returns -func printError(err error) error { - if err != nil && errors.Cause(err) != context.Canceled && - errors.Cause(err) != context.DeadlineExceeded && - !strings.Contains(err.Error(), "context canceled") && - !strings.Contains(err.Error(), "context deadline exceeded") && - cerrors.ErrWorkerPoolHandleCancelled.NotEqual(errors.Cause(err)) { - - log.Warn("Unified Sorter: Error detected", zap.Error(err)) - } - return err -} diff --git a/cdc/tests/utils/many_sorters_test/many_sorters_test.go b/cdc/tests/utils/many_sorters_test/many_sorters_test.go deleted file mode 100644 index 7ffeb2f9..00000000 --- a/cdc/tests/utils/many_sorters_test/many_sorters_test.go +++ /dev/null @@ -1,20 +0,0 @@ -// Copyright 2020 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import "testing" - -func TestFoo(_ *testing.T) { - main() -} diff --git a/cdc/tests/utils/many_sorters_test/sorter/ticdc_lock b/cdc/tests/utils/many_sorters_test/sorter/ticdc_lock deleted file mode 100644 index e69de29b..00000000 diff --git a/cdc/tests/utils/sorter_stress_test/sorter_stress.go b/cdc/tests/utils/sorter_stress_test/sorter_stress.go deleted file mode 100644 index 8f3a9e49..00000000 --- a/cdc/tests/utils/sorter_stress_test/sorter_stress.go +++ /dev/null @@ -1,171 +0,0 @@ -// Copyright 2020 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "context" - "flag" - "math/rand" - "net/http" - _ "net/http/pprof" // #nosec G108 - "os" - "strings" - - "github.com/pingcap/failpoint" - - "github.com/pingcap/log" - "github.com/tikv/migration/cdc/cdc/model" - "github.com/tikv/migration/cdc/cdc/sorter" - "github.com/tikv/migration/cdc/cdc/sorter/unified" - "github.com/tikv/migration/cdc/pkg/config" - "go.uber.org/zap" - "golang.org/x/sync/errgroup" -) - -var ( - sorterDir = flag.String("dir", "./sorter", "temporary directory used for sorting") - numBatches = flag.Int("num-batches", 256, "number of batches of ordered events") - msgsPerBatch = flag.Int("num-messages-per-batch", 1024, "number of events in a batch") - bytesPerMsg = flag.Int("bytes-per-message", 1024, "number of bytes in an event") -) - -func main() { - flag.Parse() - log.SetLevel(zap.DebugLevel) - err := failpoint.Enable("github.com/tikv/migration/cdc/cdc/sorter/unified/sorterDebug", "return(true)") - if err != nil { - log.Fatal("Could not enable failpoint", zap.Error(err)) - } - - conf := config.GetDefaultServerConfig() - conf.Sorter = &config.SorterConfig{ - NumConcurrentWorker: 8, - ChunkSizeLimit: 1 * 1024 * 1024 * 1024, - MaxMemoryPressure: 60, - MaxMemoryConsumption: 16 * 1024 * 1024 * 1024, - } - config.StoreGlobalServerConfig(conf) - - go func() { - _ = http.ListenAndServe("localhost:6060", nil) - }() - - err = os.MkdirAll(*sorterDir, 0o700) - if err != nil { - log.Error("sorter_stress_test:", zap.Error(err)) - } - - sorter, err := unified.NewUnifiedSorter(*sorterDir, "test-cf", "test", 0, "0.0.0.0:0") - if err != nil { - log.Panic("sorter_stress_test:", zap.Error(err)) - } - - ctx1, cancel := context.WithCancel(context.Background()) - - eg, ctx := errgroup.WithContext(ctx1) - - eg.Go(func() error { - return unified.RunWorkerPool(ctx) - }) - - eg.Go(func() error { - return sorter.Run(ctx) - }) - - // launch the consumer - eg.Go(func() error { - counter := 0 - lastTs := uint64(0) - for { - select { - case <-ctx.Done(): - return ctx.Err() - case event := <-sorter.Output(): - if event.RawKV.OpType != model.OpTypeResolved { - if event.CRTs < lastTs { - panic("regressed") - } - lastTs = event.CRTs - counter += 1 - if counter%10000 == 0 { - log.Debug("Messages received", zap.Int("counter", counter)) - } - if counter >= *numBatches**msgsPerBatch { - log.Debug("Unified Sorter test successful") - cancel() - return nil - } - } - } - } - }) - - eg1 := errgroup.Group{} - for i := 0; i < *numBatches; i++ { - eg1.Go(func() error { - generateGroup(ctx, sorter) - return nil - }) - } - - err = eg1.Wait() - if err != nil { - log.Error("sorter_stress_test:", zap.Error(err)) - } - - sorter.AddEntry(ctx, model.NewResolvedPolymorphicEvent(0, uint64((*msgsPerBatch<<5)+256))) - - err = eg.Wait() - if err != nil { - if strings.Contains(err.Error(), "context canceled") { - return - } - log.Error("sorter_stress_test:", zap.Error(err)) - } -} - -func generateGroup(ctx context.Context, sorter sorter.EventSorter) { - for i := 0; i < *msgsPerBatch; i++ { - ts := (i << 5) + rand.Intn(256) - event := model.NewPolymorphicEvent(newMockRawKV(uint64(ts))) - sorter.AddEntry(ctx, event) - } -} - -var ( - key = []byte(randSeq(10)) - value = []byte(randSeq(*bytesPerMsg)) -) - -func newMockRawKV(ts uint64) *model.RawKVEntry { - return &model.RawKVEntry{ - OpType: model.OpTypePut, - Key: key, - Value: value, - OldValue: nil, - StartTs: ts - 5, - CRTs: ts, - RegionID: 0, - } -} - -var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") - -func randSeq(n int) string { - b := make([]rune, n) - for i := range b { - b[i] = letters[rand.Intn(len(letters))] - } - return string(b) -} From 5722ddd97410fc293170bd7043f1c6cd49cc4157 Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 19:33:30 +0800 Subject: [PATCH 4/6] fix CI Signed-off-by: pingyu --- cdc/cdc/kv/client_test.go | 2 +- cdc/errors.toml | 30 +++++++++++++++--------------- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/cdc/cdc/kv/client_test.go b/cdc/cdc/kv/client_test.go index 51a7d035..dc2c7e38 100644 --- a/cdc/cdc/kv/client_test.go +++ b/cdc/cdc/kv/client_test.go @@ -3404,7 +3404,7 @@ func (s *clientSuite) TestRegionWorkerExitWhenIsIdle(c *check.C) { select { case <-server1Stopped: - case <-time.After(time.Second): + case <-time.After(10 * time.Second): c.Error("stream is not terminated by cdc kv client") } cancel() diff --git a/cdc/errors.toml b/cdc/errors.toml index 77e74567..a1878116 100755 --- a/cdc/errors.toml +++ b/cdc/errors.toml @@ -456,6 +456,16 @@ error = ''' kafka send message failed ''' +["CDC:ErrKeySpanListenReplicated"] +error = ''' +A keyspan(%d) is being replicated by at least two processors(%s, %s), please report a bug +''' + +["CDC:ErrKeySpanProcessorStoppedSafely"] +error = ''' +keyspan processor stopped safely +''' + ["CDC:ErrLeaseExpired"] error = ''' owner lease expired @@ -743,7 +753,7 @@ prewrite not match, key: %s, start-ts: %d, commit-ts: %d, type: %s, optype: %s ["CDC:ErrProcessorDuplicateOperations"] error = ''' -table processor duplicate operation, table-id: %d +keyspan processor duplicate operation, keyspan-id: %d ''' ["CDC:ErrProcessorEtcdWatch"] @@ -751,14 +761,14 @@ error = ''' etcd watch returns error ''' -["CDC:ErrProcessorSortDir"] +["CDC:ErrProcessorKeySpanNotFound"] error = ''' -sort dir error +keyspan not found in processor cache ''' -["CDC:ErrProcessorTableNotFound"] +["CDC:ErrProcessorSortDir"] error = ''' -table not found in processor cache +sort dir error ''' ["CDC:ErrProcessorUnknown"] @@ -961,16 +971,6 @@ error = ''' some tables are not eligible to replicate(%v), if you want to ignore these tables, please set ignore_ineligible_table to true ''' -["CDC:ErrTableListenReplicated"] -error = ''' -A table(%d) is being replicated by at least two processors(%s, %s), please report a bug -''' - -["CDC:ErrTableProcessorStoppedSafely"] -error = ''' -table processor stopped safely -''' - ["CDC:ErrTargetTsBeforeStartTs"] error = ''' fail to create changefeed because target-ts %d is earlier than start-ts %d From 92dc6b304f65a2619e699d291fb70bd8d8c1b082 Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 19:59:29 +0800 Subject: [PATCH 5/6] fix CI Signed-off-by: pingyu --- cdc/Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc/Makefile b/cdc/Makefile index 07a5ff2f..7d8ba9ba 100644 --- a/cdc/Makefile +++ b/cdc/Makefile @@ -100,7 +100,7 @@ unit_test: check_failpoint_ctl mkdir -p "$(TEST_DIR)" $(FAILPOINT_ENABLE) @export log_level=error;\ - $(GOTEST) -cover -covermode=atomic -coverprofile="$(TEST_DIR)/cov.unit.out" $(PACKAGES) \ + $(GOTEST) -cover -covermode=atomic -coverprofile="$(TEST_DIR)/cov.unit.out" -timeout 30m $(PACKAGES) \ || { $(FAILPOINT_DISABLE); exit 1; } $(FAILPOINT_DISABLE) From a06472f443a0260b5e14054948cabe45e1868611 Mon Sep 17 00:00:00 2001 From: pingyu Date: Mon, 18 Apr 2022 20:42:56 +0800 Subject: [PATCH 6/6] fix CI Signed-off-by: pingyu --- .github/workflows/ci-cdc.yml | 4 ++-- cdc/Makefile | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci-cdc.yml b/.github/workflows/ci-cdc.yml index 114ba218..cad8811c 100644 --- a/.github/workflows/ci-cdc.yml +++ b/.github/workflows/ci-cdc.yml @@ -54,8 +54,8 @@ jobs: - uses: actions/setup-go@v2 with: go-version: '1.16.1' - - name: make unit_test + - name: make unit_test_in_verify_ci shell: bash run: | cd cdc/ - make unit_test + make unit_test_in_verify_ci diff --git a/cdc/Makefile b/cdc/Makefile index 7d8ba9ba..9fe881d4 100644 --- a/cdc/Makefile +++ b/cdc/Makefile @@ -100,7 +100,7 @@ unit_test: check_failpoint_ctl mkdir -p "$(TEST_DIR)" $(FAILPOINT_ENABLE) @export log_level=error;\ - $(GOTEST) -cover -covermode=atomic -coverprofile="$(TEST_DIR)/cov.unit.out" -timeout 30m $(PACKAGES) \ + $(GOTEST) -cover -covermode=atomic -coverprofile="$(TEST_DIR)/cov.unit.out" $(PACKAGES) \ || { $(FAILPOINT_DISABLE); exit 1; } $(FAILPOINT_DISABLE) @@ -108,7 +108,7 @@ unit_test_in_verify_ci: check_failpoint_ctl tools/bin/gotestsum tools/bin/gocov mkdir -p "$(TEST_DIR)" $(FAILPOINT_ENABLE) @export log_level=error;\ - CGO_ENABLED=1 tools/bin/gotestsum --junitfile cdc-junit-report.xml -- -v -timeout 5m -p $(P) --race \ + CGO_ENABLED=1 tools/bin/gotestsum --junitfile cdc-junit-report.xml -- -v -timeout 30m -p $(P) --race \ -covermode=atomic -coverprofile="$(TEST_DIR)/cov.unit.out" $(PACKAGES) \ || { $(FAILPOINT_DISABLE); exit 1; } tools/bin/gocov convert "$(TEST_DIR)/cov.unit.out" | tools/bin/gocov-xml > cdc-coverage.xml