Merge pull request #9081 from gyuho/network-fault-test-with-proxy

*: run network fault tests with proxy
This commit is contained in:
Gyuho Lee
2018-02-06 09:51:34 -08:00
committed by GitHub
34 changed files with 2607 additions and 278 deletions

5
.gitignore vendored
View File

@@ -7,7 +7,6 @@
/release
/machine*
/bin
.Dockerfile-test
.vagrant
*.etcd
*.log
@@ -15,9 +14,6 @@
*.swp
/hack/insta-discovery/.env
*.test
tools/functional-tester/docker/bin
hack/scripts-dev/docker-dns/.Dockerfile
hack/scripts-dev/docker-dns-srv/.Dockerfile
hack/tls-setup/certs
.idea
@@ -38,3 +34,4 @@ vendor/**/*
!vendor/**/LICENSE*
vendor/**/*_test.go
*.bak

View File

@@ -0,0 +1,53 @@
FROM ubuntu:17.10
RUN rm /bin/sh && ln -s /bin/bash /bin/sh
RUN echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
RUN apt-get -y update \
&& apt-get -y install \
build-essential \
gcc \
apt-utils \
pkg-config \
software-properties-common \
apt-transport-https \
libssl-dev \
sudo \
bash \
curl \
wget \
tar \
git \
&& apt-get -y update \
&& apt-get -y upgrade \
&& apt-get -y autoremove \
&& apt-get -y autoclean
ENV GOROOT /usr/local/go
ENV GOPATH /go
ENV PATH ${GOPATH}/bin:${GOROOT}/bin:${PATH}
ENV GO_VERSION REPLACE_ME_GO_VERSION
ENV GO_DOWNLOAD_URL https://storage.googleapis.com/golang
RUN rm -rf ${GOROOT} \
&& curl -s ${GO_DOWNLOAD_URL}/go${GO_VERSION}.linux-amd64.tar.gz | tar -v -C /usr/local/ -xz \
&& mkdir -p ${GOPATH}/src ${GOPATH}/bin \
&& go version
RUN mkdir -p ${GOPATH}/src/github.com/coreos/etcd
ADD . ${GOPATH}/src/github.com/coreos/etcd
RUN go get -v github.com/coreos/gofail \
&& pushd ${GOPATH}/src/github.com/coreos/etcd \
&& GO_BUILD_FLAGS="-v" ./build \
&& cp ./bin/etcd /etcd \
&& cp ./bin/etcdctl /etcdctl \
&& GO_BUILD_FLAGS="-v" FAILPOINTS=1 ./build \
&& cp ./bin/etcd /etcd-failpoints \
&& ./tools/functional-tester/build \
&& cp ./bin/etcd-agent /etcd-agent \
&& cp ./bin/etcd-tester /etcd-tester \
&& cp ./bin/etcd-runner /etcd-runner \
&& go build -v -o /benchmark ./cmd/tools/benchmark \
&& go build -v -o /etcd-test-proxy ./cmd/tools/etcd-test-proxy \
&& popd \
&& rm -rf ${GOPATH}/src/github.com/coreos/etcd

View File

@@ -27,26 +27,15 @@ clean:
_GO_VERSION = 1.9.3
ifdef GO_VERSION
_GO_VERSION = $(GO_VERSION)
endif
_ETCD_VERSION ?= $(shell git rev-parse --short HEAD || echo "GitNotFound")
ifdef ETCD_VERSION
_ETCD_VERSION = $(ETCD_VERSION)
endif
GO_VERSION ?= 1.9.3
ETCD_VERSION ?= $(shell git rev-parse --short HEAD || echo "GitNotFound")
TEST_SUFFIX = $(shell date +%s | base64 | head -c 15)
TEST_OPTS ?= PASSES='unit'
_TEST_OPTS = PASSES='unit'
ifdef TEST_OPTS
_TEST_OPTS = $(TEST_OPTS)
endif
_TMP_DIR_MOUNT_FLAG = --mount type=tmpfs,destination=/tmp
TMP_DIR_MOUNT_FLAG = --mount type=tmpfs,destination=/tmp
ifdef HOST_TMP_DIR
_TMP_DIR_MOUNT_FLAG = --mount type=bind,source=$(HOST_TMP_DIR),destination=/tmp
TMP_DIR_MOUNT_FLAG = --mount type=bind,source=$(HOST_TMP_DIR),destination=/tmp
endif
@@ -62,20 +51,20 @@ endif
# make pull-docker-test -f ./hack/scripts-dev/Makefile
build-docker-test:
$(info GO_VERSION: $(_GO_VERSION))
@cat ./Dockerfile-test | sed s/REPLACE_ME_GO_VERSION/$(_GO_VERSION)/ \
> ./.Dockerfile-test
$(info GO_VERSION: $(GO_VERSION))
@sed -i.bak 's|REPLACE_ME_GO_VERSION|$(GO_VERSION)|g' ./Dockerfile-test
docker build \
--tag gcr.io/etcd-development/etcd-test:go$(_GO_VERSION) \
--file ./.Dockerfile-test .
--tag gcr.io/etcd-development/etcd-test:go$(GO_VERSION) \
--file ./Dockerfile-test .
@mv ./Dockerfile-test.bak ./Dockerfile-test
push-docker-test:
$(info GO_VERSION: $(_GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-test:go$(GO_VERSION)
pull-docker-test:
$(info GO_VERSION: $(_GO_VERSION))
docker pull gcr.io/etcd-development/etcd-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
docker pull gcr.io/etcd-development/etcd-test:go$(GO_VERSION)
@@ -84,11 +73,11 @@ pull-docker-test:
# make compile-with-docker-test -f ./hack/scripts-dev/Makefile
compile-with-docker-test:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
docker run \
--rm \
--mount type=bind,source=`pwd`,destination=/etcd \
gcr.io/etcd-development/etcd-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && GO_BUILD_FLAGS=-v ./build && ./bin/etcd --version"
@@ -111,44 +100,46 @@ compile-with-docker-test:
#
# Semaphore CI (test with docker):
# TEST_OPTS="RELEASE_TEST=y INTEGRATION=y PASSES='build unit release integration_e2e functional'" make docker-test -f ./hack/scripts-dev/Makefile
# HOST_TMP_DI=/tmp TEST_OPTS="RELEASE_TEST=y INTEGRATION=y PASSES='build unit release integration_e2e functional'" make docker-test -f ./hack/scripts-dev/Makefile
# HOST_TMP_DIR=/tmp TEST_OPTS="RELEASE_TEST=y INTEGRATION=y PASSES='build unit release integration_e2e functional'" make docker-test -f ./hack/scripts-dev/Makefile
# TEST_OPTS="GOARCH=386 PASSES='build unit integration_e2e'" make docker-test -f ./hack/scripts-dev/Makefile
#
# grpc-proxy tests (test with docker):
# TEST_OPTS="PASSES='build grpcproxy'" make docker-test -f ./hack/scripts-dev/Makefile
# HOST_TMP_DI=/tmp TEST_OPTS="PASSES='build grpcproxy'" make docker-test -f ./hack/scripts-dev/Makefile
# HOST_TMP_DIR=/tmp TEST_OPTS="PASSES='build grpcproxy'" make docker-test -f ./hack/scripts-dev/Makefile
.PHONY: test
test:
$(info TEST_OPTS: $(_TEST_OPTS))
$(info TEST_OPTS: $(TEST_OPTS))
$(info log-file: test-$(TEST_SUFFIX).log)
$(_TEST_OPTS) ./test 2>&1 | tee test-$(TEST_SUFFIX).log
$(TEST_OPTS) ./test 2>&1 | tee test-$(TEST_SUFFIX).log
! egrep "(--- FAIL:|panic: test timed out|appears to have leaked)" -B50 -A10 test-$(TEST_SUFFIX).log
docker-test:
$(info GO_VERSION: $(_GO_VERSION))
$(info TEST_OPTS: $(_TEST_OPTS))
$(info GO_VERSION: $(GO_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
$(info TEST_OPTS: $(TEST_OPTS))
$(info log-file: test-$(TEST_SUFFIX).log)
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`,destination=/go/src/github.com/coreos/etcd \
gcr.io/etcd-development/etcd-test:go$(_GO_VERSION) \
/bin/bash -c "$(_TEST_OPTS) ./test 2>&1 | tee test-$(TEST_SUFFIX).log"
gcr.io/etcd-development/etcd-test:go$(GO_VERSION) \
/bin/bash -c "$(TEST_OPTS) ./test 2>&1 | tee test-$(TEST_SUFFIX).log"
! egrep "(--- FAIL:|panic: test timed out|appears to have leaked)" -B50 -A10 test-$(TEST_SUFFIX).log
docker-test-coverage:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
$(info log-file: docker-test-coverage-$(TEST_SUFFIX).log)
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`,destination=/go/src/github.com/coreos/etcd \
gcr.io/etcd-development/etcd-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-test:go$(GO_VERSION) \
/bin/bash -c "COVERDIR=covdir PASSES='build build_cov cov' ./test 2>&1 | tee docker-test-coverage-$(TEST_SUFFIX).log && /codecov -t 6040de41-c073-4d6f-bbf8-d89256ef31e1"
! egrep "(--- FAIL:|panic: test timed out|appears to have leaked)" -B50 -A10 docker-test-coverage-$(TEST_SUFFIX).log
@@ -160,22 +151,22 @@ docker-test-coverage:
# gsutil -m acl ch -u allUsers:R -r gs://artifacts.etcd-development.appspot.com
build-docker-release-master: compile-with-docker-test
$(info ETCD_VERSION: $(_ETCD_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
cp ./Dockerfile-release ./bin/Dockerfile-release
docker build \
--tag gcr.io/etcd-development/etcd:$(_ETCD_VERSION) \
--tag gcr.io/etcd-development/etcd:$(ETCD_VERSION) \
--file ./bin/Dockerfile-release \
./bin
rm -f ./bin/Dockerfile-release
docker run \
--rm \
gcr.io/etcd-development/etcd:$(_ETCD_VERSION) \
gcr.io/etcd-development/etcd:$(ETCD_VERSION) \
/bin/sh -c "/usr/local/bin/etcd --version && ETCDCTL_API=3 /usr/local/bin/etcdctl version"
push-docker-release-master:
$(info ETCD_VERSION: $(_ETCD_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd:$(_ETCD_VERSION)
$(info ETCD_VERSION: $(ETCD_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd:$(ETCD_VERSION)
@@ -191,47 +182,46 @@ push-docker-release-master:
# make docker-static-ip-test-certs-metrics-proxy-run -f ./hack/scripts-dev/Makefile
build-docker-static-ip-test:
$(info GO_VERSION: $(_GO_VERSION))
@cat ./hack/scripts-dev/docker-static-ip/Dockerfile | sed s/REPLACE_ME_GO_VERSION/$(_GO_VERSION)/ \
> ./hack/scripts-dev/docker-static-ip/.Dockerfile
$(info GO_VERSION: $(GO_VERSION))
@sed -i.bak 's|REPLACE_ME_GO_VERSION|$(GO_VERSION)|g' ./hack/scripts-dev/docker-static-ip/Dockerfile
docker build \
--tag gcr.io/etcd-development/etcd-static-ip-test:go$(_GO_VERSION) \
--file ./hack/scripts-dev/docker-static-ip/.Dockerfile \
--tag gcr.io/etcd-development/etcd-static-ip-test:go$(GO_VERSION) \
--file ./hack/scripts-dev/docker-static-ip/Dockerfile \
./hack/scripts-dev/docker-static-ip
@mv ./hack/scripts-dev/docker-static-ip/Dockerfile.bak ./hack/scripts-dev/docker-static-ip/Dockerfile
push-docker-static-ip-test:
$(info GO_VERSION: $(_GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-static-ip-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-static-ip-test:go$(GO_VERSION)
pull-docker-static-ip-test:
$(info GO_VERSION: $(_GO_VERSION))
docker pull gcr.io/etcd-development/etcd-static-ip-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
docker pull gcr.io/etcd-development/etcd-static-ip-test:go$(GO_VERSION)
docker-static-ip-test-certs-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-static-ip/certs,destination=/certs \
gcr.io/etcd-development/etcd-static-ip-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-static-ip-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs/run.sh && rm -rf m*.etcd"
docker-static-ip-test-certs-metrics-proxy-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-static-ip/certs-metrics-proxy,destination=/certs-metrics-proxy \
gcr.io/etcd-development/etcd-static-ip-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-static-ip-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-metrics-proxy/run.sh && rm -rf m*.etcd"
@@ -251,97 +241,96 @@ docker-static-ip-test-certs-metrics-proxy-run:
# make docker-dns-test-certs-common-name-multi-run -f ./hack/scripts-dev/Makefile
build-docker-dns-test:
$(info GO_VERSION: $(_GO_VERSION))
@cat ./hack/scripts-dev/docker-dns/Dockerfile | sed s/REPLACE_ME_GO_VERSION/$(_GO_VERSION)/ \
> ./hack/scripts-dev/docker-dns/.Dockerfile
$(info GO_VERSION: $(GO_VERSION))
@sed -i.bak 's|REPLACE_ME_GO_VERSION|$(GO_VERSION)|g' ./hack/scripts-dev/docker-dns/Dockerfile
docker build \
--tag gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
--file ./hack/scripts-dev/docker-dns/.Dockerfile \
--tag gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
--file ./hack/scripts-dev/docker-dns/Dockerfile \
./hack/scripts-dev/docker-dns
@mv ./hack/scripts-dev/docker-dns/Dockerfile.bak ./hack/scripts-dev/docker-dns/Dockerfile
docker run \
--rm \
--dns 127.0.0.1 \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "/etc/init.d/bind9 start && cat /dev/null >/etc/hosts && dig etcd.local"
push-docker-dns-test:
$(info GO_VERSION: $(_GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION)
pull-docker-dns-test:
$(info GO_VERSION: $(_GO_VERSION))
docker pull gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
docker pull gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION)
docker-dns-test-certs-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns/certs,destination=/certs \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs/run.sh && rm -rf m*.etcd"
docker-dns-test-certs-gateway-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns/certs-gateway,destination=/certs-gateway \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-gateway/run.sh && rm -rf m*.etcd"
docker-dns-test-certs-wildcard-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns/certs-wildcard,destination=/certs-wildcard \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-wildcard/run.sh && rm -rf m*.etcd"
docker-dns-test-certs-common-name-auth-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns/certs-common-name-auth,destination=/certs-common-name-auth \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-common-name-auth/run.sh && rm -rf m*.etcd"
docker-dns-test-certs-common-name-multi-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns/certs-common-name-multi,destination=/certs-common-name-multi \
gcr.io/etcd-development/etcd-dns-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-common-name-multi/run.sh && rm -rf m*.etcd"
@@ -359,67 +348,113 @@ docker-dns-test-certs-common-name-multi-run:
# make docker-dns-srv-test-certs-wildcard-run -f ./hack/scripts-dev/Makefile
build-docker-dns-srv-test:
$(info GO_VERSION: $(_GO_VERSION))
@cat ./hack/scripts-dev/docker-dns-srv/Dockerfile | sed s/REPLACE_ME_GO_VERSION/$(_GO_VERSION)/ \
> ./hack/scripts-dev/docker-dns-srv/.Dockerfile
$(info GO_VERSION: $(GO_VERSION))
@sed -i.bak 's|REPLACE_ME_GO_VERSION|$(GO_VERSION)|g' ./hack/scripts-dev/docker-dns-srv/Dockerfile
docker build \
--tag gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION) \
--file ./hack/scripts-dev/docker-dns-srv/.Dockerfile \
--tag gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION) \
--file ./hack/scripts-dev/docker-dns-srv/Dockerfile \
./hack/scripts-dev/docker-dns-srv
@mv ./hack/scripts-dev/docker-dns-srv/Dockerfile.bak ./hack/scripts-dev/docker-dns-srv/Dockerfile
docker run \
--rm \
--dns 127.0.0.1 \
gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION) \
/bin/bash -c "/etc/init.d/bind9 start && cat /dev/null >/etc/hosts && dig +noall +answer SRV _etcd-client-ssl._tcp.etcd.local && dig +noall +answer SRV _etcd-server-ssl._tcp.etcd.local && dig +noall +answer m1.etcd.local m2.etcd.local m3.etcd.local"
push-docker-dns-srv-test:
$(info GO_VERSION: $(_GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION)
pull-docker-dns-srv-test:
$(info GO_VERSION: $(_GO_VERSION))
docker pull gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION)
$(info GO_VERSION: $(GO_VERSION))
docker pull gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION)
docker-dns-srv-test-certs-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns-srv/certs,destination=/certs \
gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs/run.sh && rm -rf m*.etcd"
docker-dns-srv-test-certs-gateway-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns-srv/certs-gateway,destination=/certs-gateway \
gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-gateway/run.sh && rm -rf m*.etcd"
docker-dns-srv-test-certs-wildcard-run:
$(info GO_VERSION: $(_GO_VERSION))
$(info GO_VERSION: $(GO_VERSION))
$(info HOST_TMP_DIR: $(HOST_TMP_DIR))
$(info TMP_DIR_MOUNT_FLAG: $(_TMP_DIR_MOUNT_FLAG))
$(info TMP_DIR_MOUNT_FLAG: $(TMP_DIR_MOUNT_FLAG))
docker run \
--rm \
--tty \
--dns 127.0.0.1 \
$(_TMP_DIR_MOUNT_FLAG) \
$(TMP_DIR_MOUNT_FLAG) \
--mount type=bind,source=`pwd`/bin,destination=/etcd \
--mount type=bind,source=`pwd`/hack/scripts-dev/docker-dns-srv/certs-wildcard,destination=/certs-wildcard \
gcr.io/etcd-development/etcd-dns-srv-test:go$(_GO_VERSION) \
gcr.io/etcd-development/etcd-dns-srv-test:go$(GO_VERSION) \
/bin/bash -c "cd /etcd && /certs-wildcard/run.sh && rm -rf m*.etcd"
# Example:
# make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
build-etcd-test-proxy:
go build -v -o ./bin/etcd-test-proxy ./cmd/tools/etcd-test-proxy
# Example:
# make build-docker-functional-tester -f ./hack/scripts-dev/Makefile
# make push-docker-functional-tester -f ./hack/scripts-dev/Makefile
# make pull-docker-functional-tester -f ./hack/scripts-dev/Makefile
build-docker-functional-tester:
$(info GO_VERSION: $(GO_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
@sed -i.bak 's|REPLACE_ME_GO_VERSION|$(GO_VERSION)|g' ./Dockerfile-functional-tester
docker build \
--tag gcr.io/etcd-development/etcd-functional-tester:go$(GO_VERSION) \
--file ./Dockerfile-functional-tester \
.
@mv ./Dockerfile-functional-tester.bak ./Dockerfile-functional-tester
docker run \
--rm \
gcr.io/etcd-development/etcd-functional-tester:go$(GO_VERSION) \
/bin/bash -c "/etcd --version && \
/etcd-failpoints --version && \
ETCDCTL_API=3 /etcdctl version && \
/etcd-agent -help || true && \
/etcd-tester -help || true && \
/etcd-runner --help || true && \
/benchmark --help || true && \
/etcd-test-proxy -help || true"
push-docker-functional-tester:
$(info GO_VERSION: $(GO_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
gcloud docker -- push gcr.io/etcd-development/etcd-functional-tester:go$(GO_VERSION)
pull-docker-functional-tester:
$(info GO_VERSION: $(GO_VERSION))
$(info ETCD_VERSION: $(ETCD_VERSION))
docker pull gcr.io/etcd-development/etcd-functional-tester:go$(GO_VERSION)

View File

@@ -0,0 +1,19 @@
{
"key": {
"algo": "rsa",
"size": 2048
},
"names": [
{
"O": "etcd",
"OU": "etcd Security",
"L": "San Francisco",
"ST": "California",
"C": "USA"
}
],
"CN": "ca",
"ca": {
"expiry": "87600h"
}
}

View File

@@ -0,0 +1,22 @@
-----BEGIN CERTIFICATE-----
MIIDsTCCApmgAwIBAgIUZzOo4zcHY/nEXY1PD8A7povXlWUwDQYJKoZIhvcNAQEL
BQAwbzEMMAoGA1UEBhMDVVNBMRMwEQYDVQQIEwpDYWxpZm9ybmlhMRYwFAYDVQQH
Ew1TYW4gRnJhbmNpc2NvMQ0wCwYDVQQKEwRldGNkMRYwFAYDVQQLEw1ldGNkIFNl
Y3VyaXR5MQswCQYDVQQDEwJjYTAeFw0xODAxMDIxNjQxMDBaFw0yNzEyMzExNjQx
MDBaMG8xDDAKBgNVBAYTA1VTQTETMBEGA1UECBMKQ2FsaWZvcm5pYTEWMBQGA1UE
BxMNU2FuIEZyYW5jaXNjbzENMAsGA1UEChMEZXRjZDEWMBQGA1UECxMNZXRjZCBT
ZWN1cml0eTELMAkGA1UEAxMCY2EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
AoIBAQDD4Ys48LDWGyojj3Rcr6fnESY+UycaaGoTXADWLPmm+sQR3KcsJxF4054S
d2G+NBfJHZvTHhVqOeqZxNtoqgje4paY2A5TbWBdV+xoGfbakwwngiX1yeF1I54k
KH19zb8rBKAm7xixO60hE2CIYzMuw9lDkwoHpI6/PJdy7jwtytbo2Oac512JiO9Y
dHp9dr3mrCzoKEBRtL1asRKfzp6gBC5rIw5T4jrq37feerV4pDEJX7fvexxVocVm
tT4bmMq3Ap6OFFAzmE/ITI8pXvFaOd9lyebNXQmrreKJLUfEIZa6JulLCYxfkJ8z
+CcNLyn6ZXNMaIZ8G9Hm6VRdRi8/AgMBAAGjRTBDMA4GA1UdDwEB/wQEAwIBBjAS
BgNVHRMBAf8ECDAGAQH/AgECMB0GA1UdDgQWBBRDLNYEX8XI7nM53k1rUR+mpTjQ
NTANBgkqhkiG9w0BAQsFAAOCAQEACDe3Fa1KE/rvVtyCLW/IBfKV01NShFTsb6x8
GrPEQ6NJLZQ2MzdyJgAF2a/nZ9KVgrhGXoyoZBCKP9Dd/JDzSSZcBztfNK8dRv2A
XHBBF6tZ19I+XY9c7/CfhJ2CEYJpeN9r3GKSqV+njkmg8n/On2BTlFsij88plK8H
ORyemc1nQI+ARPSu2r3rJbYa4yI2U6w4L4BTCVImg3bX50GImmXGlwvnJMFik1FX
+0hdfetRxxMZ1pm2Uy6099KkULnSKabZGwRiBUHQJYh0EeuAOQ4a6MG5DRkURWNs
dInjPOLY9/7S5DQKwz/NtqXA8EEymZosHxpiRp+zzKB4XaV9Ig==
-----END CERTIFICATE-----

View File

@@ -0,0 +1,13 @@
{
"signing": {
"default": {
"usages": [
"signing",
"key encipherment",
"server auth",
"client auth"
],
"expiry": "87600h"
}
}
}

View File

@@ -0,0 +1,26 @@
#!/bin/bash
if ! [[ "$0" =~ "./gencerts.sh" ]]; then
echo "must be run from 'fixtures'"
exit 255
fi
if ! which cfssl; then
echo "cfssl is not installed"
exit 255
fi
cfssl gencert --initca=true ./ca-csr.json | cfssljson --bare ./ca
mv ca.pem ca.crt
openssl x509 -in ca.crt -noout -text
# generate DNS: localhost, IP: 127.0.0.1, CN: example.com certificates
cfssl gencert \
--ca ./ca.crt \
--ca-key ./ca-key.pem \
--config ./gencert.json \
./server-ca-csr.json | cfssljson --bare ./server
mv server.pem server.crt
mv server-key.pem server.key.insecure
rm -f *.csr *.pem *.stderr *.txt

View File

@@ -0,0 +1,20 @@
{
"key": {
"algo": "rsa",
"size": 2048
},
"names": [
{
"O": "etcd",
"OU": "etcd Security",
"L": "San Francisco",
"ST": "California",
"C": "USA"
}
],
"CN": "example.com",
"hosts": [
"127.0.0.1",
"localhost"
]
}

View File

@@ -0,0 +1,24 @@
-----BEGIN CERTIFICATE-----
MIIEEjCCAvqgAwIBAgIUIYc+vmysep1pDc2ua/VQEeMFQVAwDQYJKoZIhvcNAQEL
BQAwbzEMMAoGA1UEBhMDVVNBMRMwEQYDVQQIEwpDYWxpZm9ybmlhMRYwFAYDVQQH
Ew1TYW4gRnJhbmNpc2NvMQ0wCwYDVQQKEwRldGNkMRYwFAYDVQQLEw1ldGNkIFNl
Y3VyaXR5MQswCQYDVQQDEwJjYTAeFw0xODAxMDIxNjQxMDBaFw0yNzEyMzExNjQx
MDBaMHgxDDAKBgNVBAYTA1VTQTETMBEGA1UECBMKQ2FsaWZvcm5pYTEWMBQGA1UE
BxMNU2FuIEZyYW5jaXNjbzENMAsGA1UEChMEZXRjZDEWMBQGA1UECxMNZXRjZCBT
ZWN1cml0eTEUMBIGA1UEAxMLZXhhbXBsZS5jb20wggEiMA0GCSqGSIb3DQEBAQUA
A4IBDwAwggEKAoIBAQDEq7aT2BQZfmJ2xpUm8xWJlN0c3cOLVZRH9mIrEutIHmip
BYq3ZIq3q52w+T3sMcaJNMGjCteE8Lu+G9YSmtfZMAWnkaM02KOjVMkkQcK7Z4vM
lOUjlO+dsvhfmw3CPghqSs6M1K2CTqhuEiXdOBofuEMmwKNRgkV/jT92PUs0h8kq
loc/I3/H+hx/ZJ1i0S0xkZKpaImc0oZ9ZDo07biMrsUIzjwbN69mEs+CtVkah4sy
k6UyRoU2k21lyRTK0LxNjWc9ylzDNUuf6DwduU7lPZsqTaJrFNAAPpOlI4k2EcjL
3zD8amKkJGDm+PQz97PbTA381ec4ZAtB8volxCebAgMBAAGjgZwwgZkwDgYDVR0P
AQH/BAQDAgWgMB0GA1UdJQQWMBQGCCsGAQUFBwMBBggrBgEFBQcDAjAMBgNVHRMB
Af8EAjAAMB0GA1UdDgQWBBTTZQnMn5tuUgVE+8c9W0hmbghGoDAfBgNVHSMEGDAW
gBRDLNYEX8XI7nM53k1rUR+mpTjQNTAaBgNVHREEEzARgglsb2NhbGhvc3SHBH8A
AAEwDQYJKoZIhvcNAQELBQADggEBAKUQVj0YDuxg4tinlOZhp4ge7tCA+gL7vV+Q
iDrkWfOlGjDgwYqWMYDXMHWKIW9ea8LzyI/bVEcaHlnBmNOYuS7g47EWNiU7WUA5
iTkm3CKA5zHFFPcXHW0GQeCQrX9y3SepKS3cP8TAyZFfC/FvV24Kn1oQhJbEe0ZV
In/vPHssW7jlVe0FGVUn7FutRQgiA1pTAtS6AP4LeZ9O41DTWkPqV4nBgcxlvkgD
KjEoXXSb5C0LoR5zwAo9zB3RtmqnmvkHAOv3G92YctdS2VbCmd8CNLj9H7gMmQiH
ThsStVOhb2uo6Ni4PgzUIYKGTd4ZjUXCYxFKck//ajDyCHlL8v4=
-----END CERTIFICATE-----

View File

@@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEogIBAAKCAQEAxKu2k9gUGX5idsaVJvMViZTdHN3Di1WUR/ZiKxLrSB5oqQWK
t2SKt6udsPk97DHGiTTBowrXhPC7vhvWEprX2TAFp5GjNNijo1TJJEHCu2eLzJTl
I5TvnbL4X5sNwj4IakrOjNStgk6obhIl3TgaH7hDJsCjUYJFf40/dj1LNIfJKpaH
PyN/x/ocf2SdYtEtMZGSqWiJnNKGfWQ6NO24jK7FCM48GzevZhLPgrVZGoeLMpOl
MkaFNpNtZckUytC8TY1nPcpcwzVLn+g8HblO5T2bKk2iaxTQAD6TpSOJNhHIy98w
/GpipCRg5vj0M/ez20wN/NXnOGQLQfL6JcQnmwIDAQABAoIBAGTx1eaQk9B6BEP+
rXOudTGGzO8SDFop9M/y8HQ3Y7hCk2mdxJNY8bJQTcIWS+g9rC+kencbC3/aqCJt
2zT1cTCy61QU9nYbc/JThGIttqvF/AVnryzSNyL0R3Oa/Dbk7CDSgK3cQ6qMgPru
Ka0gLJh3VVBAtBMUEGPltdsUntM4sHTh5FAabP0ioBJ1QLG6Aak7LOQikjBEFJoc
Tea4uRsE7IreP5Mn7UW92nkt1ey5UGzBtNNtpHbVaHmfQojwlwkLtnV35sumbvK6
6KTMNREZv6xSIMwkYxm1zRE3Cus/1jGIc8MZF0BxgcCR+G37l+BKwL8CSymHPxhH
dvGxoPECgYEA3STp52CbI/KyVfvjxK2OIex/NV1jKh85wQsLtkaRv3/a/EEg7MV7
54dEvo5KKOZXfeOd9r9G9h1RffjSD9MhxfPhyGwuOcqa8IE1zNwlY/v7KL7HtDIf
2mrXWF5Klafh8aXYcaRH0ZSLnl/nXUXYht4/0NRGiXnttUgqs6hvY70CgYEA46tO
J5QkgF3YVY0gx10wRCAnnKLkAaHdtxtteXOJh79xsGXQ4LLngc+mz1hLt+TNJza+
BZhoWwY/ZgyiTH0pebGr/U0QUMoUHlGgjgj3Aa/XFpOhtyLU+IU/PYl0BUz9dqsN
TDtv6p/HQhfd98vUNsbACQda+YAo+oRdO5kLQjcCgYB3OAZNcXxRte5EgoY5KqN8
UGYH2++w7qKRGqZWvtamGYRyB557Zr+0gu0hmc4LHJrASGyJcHcOCaI8Ol7snxMP
B7qJ9SA6kapTzCS361rQ+zBct/UrhPY9JuovPq4Q3i/luVXldf4t01otqGAvnY7s
rnZS242nYa8v0tcKgdyDNQKBgB3Z60BzQyn1pBTrkT2ysU5tbOQz03OHVrvYg80l
4gWDi5OWdgHQU1yI7pVHPX5aKLAYlGfFaQFuW0e1Jl6jFpoXOrbWsOn25RZom4Wk
FUcKWEhkiRKrJYOEbRtTd3vucVlq6i5xqKX51zWKTZddCXE5NBq69Sm7rSPT0Sms
UnaXAoGAXYAE5slvjcylJpMV4lxTBmNtA9+pw1T7I379mIyqZ0OS25nmpskHU7FR
SQDSRHw7hHuyjEHyhMoHEGLfUMIltQoi+pcrieVQelJdSuX7VInzHPAR5RppUVFl
jOZZKlIiqs+UfCoOgsIblXuw7a/ATnAnXakutSFgHU1lN1gN02U=
-----END RSA PRIVATE KEY-----

801
pkg/transport/proxy.go Normal file
View File

@@ -0,0 +1,801 @@
// Copyright 2018 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package transport
import (
"fmt"
"io"
mrand "math/rand"
"net"
"net/http"
"net/url"
"os"
"strings"
"sync"
"time"
humanize "github.com/dustin/go-humanize"
"google.golang.org/grpc/grpclog"
)
// Proxy defines proxy layer that simulates common network faults,
// such as latency spikes, packet drop/corruption, etc..
type Proxy interface {
// From returns proxy source address in "scheme://host:port" format.
From() string
// To returns proxy destination address in "scheme://host:port" format.
To() string
// Ready returns when proxy is ready to serve.
Ready() <-chan struct{}
// Done returns when proxy has been closed.
Done() <-chan struct{}
// Error sends errors while serving proxy.
Error() <-chan error
// Close closes listener and transport.
Close() error
// DelayAccept adds latency ± random variable to accepting new incoming connections.
DelayAccept(latency, rv time.Duration)
// UndelayAccept removes sending latencies.
UndelayAccept()
// LatencyAccept returns current latency on accepting new incoming connections.
LatencyAccept() time.Duration
// DelayTx adds latency ± random variable to "sending" layer.
DelayTx(latency, rv time.Duration)
// UndelayTx removes sending latencies.
UndelayTx()
// LatencyTx returns current send latency.
LatencyTx() time.Duration
// DelayRx adds latency ± random variable to "receiving" layer.
DelayRx(latency, rv time.Duration)
// UndelayRx removes "receiving" latencies.
UndelayRx()
// LatencyRx returns current receive latency.
LatencyRx() time.Duration
// PauseAccept stops accepting new connections.
PauseAccept()
// UnpauseAccept removes pause operation on accepting new connections.
UnpauseAccept()
// PauseTx stops "forwarding" packets.
PauseTx()
// UnpauseTx removes "forwarding" pause operation.
UnpauseTx()
// PauseRx stops "receiving" packets to client.
PauseRx()
// UnpauseRx removes "receiving" pause operation.
UnpauseRx()
// BlackholeTx drops all incoming packets before "forwarding".
BlackholeTx()
// UnblackholeTx removes blackhole operation on "sending".
UnblackholeTx()
// BlackholeRx drops all incoming packets to client.
BlackholeRx()
// UnblackholeRx removes blackhole operation on "receiving".
UnblackholeRx()
// CorruptTx corrupts incoming packets from the listener.
CorruptTx(f func(data []byte) []byte)
// UncorruptTx removes corrupt operation on "forwarding".
UncorruptTx()
// CorruptRx corrupts incoming packets to client.
CorruptRx(f func(data []byte) []byte)
// UncorruptRx removes corrupt operation on "receiving".
UncorruptRx()
// ResetListener closes and restarts listener.
ResetListener() error
}
type proxy struct {
from, to url.URL
tlsInfo TLSInfo
dialTimeout time.Duration
bufferSize int
retryInterval time.Duration
logger grpclog.LoggerV2
readyc chan struct{}
donec chan struct{}
errc chan error
closeOnce sync.Once
closeWg sync.WaitGroup
listenerMu sync.RWMutex
listener net.Listener
latencyAcceptMu sync.RWMutex
latencyAccept time.Duration
latencyTxMu sync.RWMutex
latencyTx time.Duration
latencyRxMu sync.RWMutex
latencyRx time.Duration
corruptTxMu sync.RWMutex
corruptTx func(data []byte) []byte
corruptRxMu sync.RWMutex
corruptRx func(data []byte) []byte
acceptMu sync.Mutex
pauseAcceptc chan struct{}
txMu sync.Mutex
pauseTxc chan struct{}
blackholeTxc chan struct{}
rxMu sync.Mutex
pauseRxc chan struct{}
blackholeRxc chan struct{}
}
// ProxyConfig defines proxy configuration.
type ProxyConfig struct {
From url.URL
To url.URL
TLSInfo TLSInfo
DialTimeout time.Duration
BufferSize int
RetryInterval time.Duration
Logger grpclog.LoggerV2
}
var (
defaultDialTimeout = 3 * time.Second
defaultBufferSize = 48 * 1024
defaultRetryInterval = 10 * time.Millisecond
defaultLogger = grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 0)
)
// NewProxy returns a proxy implementation with no iptables/tc dependencies.
// The proxy layer overhead is <1ms.
func NewProxy(cfg ProxyConfig) Proxy {
p := &proxy{
from: cfg.From,
to: cfg.To,
tlsInfo: cfg.TLSInfo,
dialTimeout: cfg.DialTimeout,
bufferSize: cfg.BufferSize,
retryInterval: cfg.RetryInterval,
logger: cfg.Logger,
readyc: make(chan struct{}),
donec: make(chan struct{}),
errc: make(chan error, 16),
pauseAcceptc: make(chan struct{}),
pauseTxc: make(chan struct{}),
blackholeTxc: make(chan struct{}),
pauseRxc: make(chan struct{}),
blackholeRxc: make(chan struct{}),
}
if p.dialTimeout == 0 {
p.dialTimeout = defaultDialTimeout
}
if p.bufferSize == 0 {
p.bufferSize = defaultBufferSize
}
if p.retryInterval == 0 {
p.retryInterval = defaultRetryInterval
}
if p.logger == nil {
p.logger = defaultLogger
}
close(p.pauseAcceptc)
close(p.pauseTxc)
close(p.pauseRxc)
if strings.HasPrefix(p.from.Scheme, "http") {
p.from.Scheme = "tcp"
}
if strings.HasPrefix(p.to.Scheme, "http") {
p.to.Scheme = "tcp"
}
var ln net.Listener
var err error
if !p.tlsInfo.Empty() {
ln, err = NewListener(p.from.Host, p.from.Scheme, &p.tlsInfo)
} else {
ln, err = net.Listen(p.from.Scheme, p.from.Host)
}
if err != nil {
p.errc <- err
p.Close()
return p
}
p.listener = ln
p.closeWg.Add(1)
go p.listenAndServe()
p.logger.Infof("started proxying [%s -> %s]", p.From(), p.To())
return p
}
func (p *proxy) From() string {
return fmt.Sprintf("%s://%s", p.from.Scheme, p.from.Host)
}
func (p *proxy) To() string {
return fmt.Sprintf("%s://%s", p.to.Scheme, p.to.Host)
}
// TODO: implement packet reordering from multiple TCP connections
// buffer packets per connection for awhile, reorder before transmit
// - https://github.com/coreos/etcd/issues/5614
// - https://github.com/coreos/etcd/pull/6918#issuecomment-264093034
func (p *proxy) listenAndServe() {
defer p.closeWg.Done()
p.logger.Infof("listen %q", p.From())
close(p.readyc)
for {
p.acceptMu.Lock()
pausec := p.pauseAcceptc
p.acceptMu.Unlock()
select {
case <-pausec:
case <-p.donec:
return
}
p.latencyAcceptMu.RLock()
lat := p.latencyAccept
p.latencyAcceptMu.RUnlock()
if lat > 0 {
select {
case <-time.After(lat):
case <-p.donec:
return
}
}
p.listenerMu.RLock()
ln := p.listener
p.listenerMu.RUnlock()
in, err := ln.Accept()
if err != nil {
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
if p.logger.V(5) {
p.logger.Errorf("listener accept error %q", err.Error())
}
if strings.HasSuffix(err.Error(), "use of closed network connection") {
select {
case <-time.After(p.retryInterval):
case <-p.donec:
return
}
if p.logger.V(5) {
p.logger.Errorf("listener is closed; retry listen %q", p.From())
}
if err = p.ResetListener(); err != nil {
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
p.logger.Errorf("failed to reset listener %q", err.Error())
}
}
continue
}
var out net.Conn
if !p.tlsInfo.Empty() {
var tp *http.Transport
tp, err = NewTransport(p.tlsInfo, p.dialTimeout)
if err != nil {
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
continue
}
out, err = tp.Dial(p.to.Scheme, p.to.Host)
} else {
out, err = net.Dial(p.to.Scheme, p.to.Host)
}
if err != nil {
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
if p.logger.V(5) {
p.logger.Errorf("dial error %q", err.Error())
}
continue
}
go func() {
// read incoming bytes from listener, dispatch to outgoing connection
p.transmit(out, in)
out.Close()
in.Close()
}()
go func() {
// read response from outgoing connection, write back to listener
p.receive(in, out)
in.Close()
out.Close()
}()
}
}
func (p *proxy) transmit(dst io.Writer, src io.Reader) { p.ioCopy(dst, src, true) }
func (p *proxy) receive(dst io.Writer, src io.Reader) { p.ioCopy(dst, src, false) }
func (p *proxy) ioCopy(dst io.Writer, src io.Reader, proxySend bool) {
buf := make([]byte, p.bufferSize)
for {
nr, err := src.Read(buf)
if err != nil {
if err == io.EOF {
return
}
// connection already closed
if strings.HasSuffix(err.Error(), "read: connection reset by peer") {
return
}
if strings.HasSuffix(err.Error(), "use of closed network connection") {
return
}
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
if p.logger.V(5) {
p.logger.Errorf("read error %q", err.Error())
}
return
}
if nr == 0 {
return
}
data := buf[:nr]
var pausec chan struct{}
var blackholec chan struct{}
if proxySend {
p.txMu.Lock()
pausec = p.pauseTxc
blackholec = p.blackholeTxc
p.txMu.Unlock()
} else {
p.rxMu.Lock()
pausec = p.pauseRxc
blackholec = p.blackholeRxc
p.rxMu.Unlock()
}
select {
case <-pausec:
case <-p.donec:
return
}
blackholed := false
select {
case <-blackholec:
blackholed = true
case <-p.donec:
return
default:
}
if blackholed {
if p.logger.V(5) {
if proxySend {
p.logger.Infof("dropped %s [%s -> %s]", humanize.Bytes(uint64(nr)), p.From(), p.To())
} else {
p.logger.Infof("dropped %s [%s <- %s]", humanize.Bytes(uint64(nr)), p.From(), p.To())
}
}
continue
}
var lat time.Duration
if proxySend {
p.latencyTxMu.RLock()
lat = p.latencyTx
p.latencyTxMu.RUnlock()
} else {
p.latencyRxMu.RLock()
lat = p.latencyRx
p.latencyRxMu.RUnlock()
}
if lat > 0 {
select {
case <-time.After(lat):
case <-p.donec:
return
}
}
if proxySend {
p.corruptTxMu.RLock()
if p.corruptTx != nil {
data = p.corruptTx(data)
}
p.corruptTxMu.RUnlock()
} else {
p.corruptRxMu.RLock()
if p.corruptRx != nil {
data = p.corruptRx(data)
}
p.corruptRxMu.RUnlock()
}
var nw int
nw, err = dst.Write(data)
if err != nil {
if err == io.EOF {
return
}
select {
case p.errc <- err:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
if p.logger.V(5) {
if proxySend {
p.logger.Errorf("write error while sending (%q)", err.Error())
} else {
p.logger.Errorf("write error while receiving (%q)", err.Error())
}
}
return
}
if nr != nw {
select {
case p.errc <- io.ErrShortWrite:
select {
case <-p.donec:
return
default:
}
case <-p.donec:
return
}
if proxySend {
p.logger.Errorf("write error while sending (%q); read %d bytes != wrote %d bytes", io.ErrShortWrite.Error(), nr, nw)
} else {
p.logger.Errorf("write error while receiving (%q); read %d bytes != wrote %d bytes", io.ErrShortWrite.Error(), nr, nw)
}
return
}
if p.logger.V(5) {
if proxySend {
p.logger.Infof("transmitted %s [%s -> %s]", humanize.Bytes(uint64(nr)), p.From(), p.To())
} else {
p.logger.Infof("received %s [%s <- %s]", humanize.Bytes(uint64(nr)), p.From(), p.To())
}
}
}
}
func (p *proxy) Ready() <-chan struct{} { return p.readyc }
func (p *proxy) Done() <-chan struct{} { return p.donec }
func (p *proxy) Error() <-chan error { return p.errc }
func (p *proxy) Close() (err error) {
p.closeOnce.Do(func() {
close(p.donec)
p.listenerMu.Lock()
if p.listener != nil {
err = p.listener.Close()
p.logger.Infof("closed proxy listener on %q", p.From())
}
p.listenerMu.Unlock()
})
p.closeWg.Wait()
return err
}
func (p *proxy) DelayAccept(latency, rv time.Duration) {
if latency <= 0 {
return
}
d := computeLatency(latency, rv)
p.latencyAcceptMu.Lock()
p.latencyAccept = d
p.latencyAcceptMu.Unlock()
p.logger.Infof("set accept latency %v(%v±%v) [%s -> %s]", d, latency, rv, p.From(), p.To())
}
func (p *proxy) UndelayAccept() {
p.latencyAcceptMu.Lock()
d := p.latencyAccept
p.latencyAccept = 0
p.latencyAcceptMu.Unlock()
p.logger.Infof("removed accept latency %v [%s -> %s]", d, p.From(), p.To())
}
func (p *proxy) LatencyAccept() time.Duration {
p.latencyAcceptMu.RLock()
d := p.latencyAccept
p.latencyAcceptMu.RUnlock()
return d
}
func (p *proxy) DelayTx(latency, rv time.Duration) {
if latency <= 0 {
return
}
d := computeLatency(latency, rv)
p.latencyTxMu.Lock()
p.latencyTx = d
p.latencyTxMu.Unlock()
p.logger.Infof("set transmit latency %v(%v±%v) [%s -> %s]", d, latency, rv, p.From(), p.To())
}
func (p *proxy) UndelayTx() {
p.latencyTxMu.Lock()
d := p.latencyTx
p.latencyTx = 0
p.latencyTxMu.Unlock()
p.logger.Infof("removed transmit latency %v [%s -> %s]", d, p.From(), p.To())
}
func (p *proxy) LatencyTx() time.Duration {
p.latencyTxMu.RLock()
d := p.latencyTx
p.latencyTxMu.RUnlock()
return d
}
func (p *proxy) DelayRx(latency, rv time.Duration) {
if latency <= 0 {
return
}
d := computeLatency(latency, rv)
p.latencyRxMu.Lock()
p.latencyRx = d
p.latencyRxMu.Unlock()
p.logger.Infof("set receive latency %v(%v±%v) [%s <- %s]", d, latency, rv, p.From(), p.To())
}
func (p *proxy) UndelayRx() {
p.latencyRxMu.Lock()
d := p.latencyRx
p.latencyRx = 0
p.latencyRxMu.Unlock()
p.logger.Infof("removed receive latency %v [%s <- %s]", d, p.From(), p.To())
}
func (p *proxy) LatencyRx() time.Duration {
p.latencyRxMu.RLock()
d := p.latencyRx
p.latencyRxMu.RUnlock()
return d
}
func computeLatency(lat, rv time.Duration) time.Duration {
if rv == 0 {
return lat
}
if rv < 0 {
rv *= -1
}
if rv > lat {
rv = lat / 10
}
now := time.Now()
mrand.Seed(int64(now.Nanosecond()))
sign := 1
if now.Second()%2 == 0 {
sign = -1
}
return lat + time.Duration(int64(sign)*mrand.Int63n(rv.Nanoseconds()))
}
func (p *proxy) PauseAccept() {
p.acceptMu.Lock()
p.pauseAcceptc = make(chan struct{})
p.acceptMu.Unlock()
p.logger.Infof("paused accepting new connections [%s -> %s]", p.From(), p.To())
}
func (p *proxy) UnpauseAccept() {
p.acceptMu.Lock()
select {
case <-p.pauseAcceptc: // already unpaused
case <-p.donec:
p.acceptMu.Unlock()
return
default:
close(p.pauseAcceptc)
}
p.acceptMu.Unlock()
p.logger.Infof("unpaused accepting new connections [%s -> %s]", p.From(), p.To())
}
func (p *proxy) PauseTx() {
p.txMu.Lock()
p.pauseTxc = make(chan struct{})
p.txMu.Unlock()
p.logger.Infof("paused transmit listen [%s -> %s]", p.From(), p.To())
}
func (p *proxy) UnpauseTx() {
p.txMu.Lock()
select {
case <-p.pauseTxc: // already unpaused
case <-p.donec:
p.txMu.Unlock()
return
default:
close(p.pauseTxc)
}
p.txMu.Unlock()
p.logger.Infof("unpaused transmit listen [%s -> %s]", p.From(), p.To())
}
func (p *proxy) PauseRx() {
p.rxMu.Lock()
p.pauseRxc = make(chan struct{})
p.rxMu.Unlock()
p.logger.Infof("paused receive listen [%s <- %s]", p.From(), p.To())
}
func (p *proxy) UnpauseRx() {
p.rxMu.Lock()
select {
case <-p.pauseRxc: // already unpaused
case <-p.donec:
p.rxMu.Unlock()
return
default:
close(p.pauseRxc)
}
p.rxMu.Unlock()
p.logger.Infof("unpaused receive listen [%s <- %s]", p.From(), p.To())
}
func (p *proxy) BlackholeTx() {
p.txMu.Lock()
select {
case <-p.blackholeTxc: // already blackholed
case <-p.donec:
p.txMu.Unlock()
return
default:
close(p.blackholeTxc)
}
p.txMu.Unlock()
p.logger.Infof("blackholed transmit [%s -> %s]", p.From(), p.To())
}
func (p *proxy) UnblackholeTx() {
p.txMu.Lock()
p.blackholeTxc = make(chan struct{})
p.txMu.Unlock()
p.logger.Infof("unblackholed transmit [%s -> %s]", p.From(), p.To())
}
func (p *proxy) BlackholeRx() {
p.rxMu.Lock()
select {
case <-p.blackholeRxc: // already blackholed
case <-p.donec:
p.rxMu.Unlock()
return
default:
close(p.blackholeRxc)
}
p.rxMu.Unlock()
p.logger.Infof("blackholed receive [%s <- %s]", p.From(), p.To())
}
func (p *proxy) UnblackholeRx() {
p.rxMu.Lock()
p.blackholeRxc = make(chan struct{})
p.rxMu.Unlock()
p.logger.Infof("unblackholed receive [%s <- %s]", p.From(), p.To())
}
func (p *proxy) CorruptTx(f func([]byte) []byte) {
p.corruptTxMu.Lock()
p.corruptTx = f
p.corruptTxMu.Unlock()
p.logger.Infof("corrupting transmit [%s -> %s]", p.From(), p.To())
}
func (p *proxy) UncorruptTx() {
p.corruptTxMu.Lock()
p.corruptTx = nil
p.corruptTxMu.Unlock()
p.logger.Infof("stopped corrupting transmit [%s -> %s]", p.From(), p.To())
}
func (p *proxy) CorruptRx(f func([]byte) []byte) {
p.corruptRxMu.Lock()
p.corruptRx = f
p.corruptRxMu.Unlock()
p.logger.Infof("corrupting receive [%s <- %s]", p.From(), p.To())
}
func (p *proxy) UncorruptRx() {
p.corruptRxMu.Lock()
p.corruptRx = nil
p.corruptRxMu.Unlock()
p.logger.Infof("stopped corrupting receive [%s <- %s]", p.From(), p.To())
}
func (p *proxy) ResetListener() error {
p.listenerMu.Lock()
defer p.listenerMu.Unlock()
if err := p.listener.Close(); err != nil {
// already closed
if !strings.HasSuffix(err.Error(), "use of closed network connection") {
return err
}
}
var ln net.Listener
var err error
if !p.tlsInfo.Empty() {
ln, err = NewListener(p.from.Host, p.from.Scheme, &p.tlsInfo)
} else {
ln, err = net.Listen(p.from.Scheme, p.from.Host)
}
if err != nil {
return err
}
p.listener = ln
p.logger.Infof("reset listener %q", p.From())
return nil
}

606
pkg/transport/proxy_test.go Normal file
View File

@@ -0,0 +1,606 @@
// Copyright 2018 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package transport
import (
"bytes"
"crypto/tls"
"fmt"
"io/ioutil"
"math/rand"
"net"
"net/http"
"net/url"
"os"
"strings"
"testing"
"time"
"google.golang.org/grpc/grpclog"
)
var testTLSInfo = TLSInfo{
KeyFile: "./fixtures/server.key.insecure",
CertFile: "./fixtures/server.crt",
TrustedCAFile: "./fixtures/ca.crt",
ClientCertAuth: true,
}
func TestProxy_Unix_Insecure(t *testing.T) { testProxy(t, "unix", false, false) }
func TestProxy_TCP_Insecure(t *testing.T) { testProxy(t, "tcp", false, false) }
func TestProxy_Unix_Secure(t *testing.T) { testProxy(t, "unix", true, false) }
func TestProxy_TCP_Secure(t *testing.T) { testProxy(t, "tcp", true, false) }
func TestProxy_Unix_Insecure_DelayTx(t *testing.T) { testProxy(t, "unix", false, true) }
func TestProxy_TCP_Insecure_DelayTx(t *testing.T) { testProxy(t, "tcp", false, true) }
func TestProxy_Unix_Secure_DelayTx(t *testing.T) { testProxy(t, "unix", true, true) }
func TestProxy_TCP_Secure_DelayTx(t *testing.T) { testProxy(t, "tcp", true, true) }
func testProxy(t *testing.T, scheme string, secure bool, delayTx bool) {
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
if scheme == "tcp" {
ln1, ln2 := listen(t, "tcp", "localhost:0", TLSInfo{}), listen(t, "tcp", "localhost:0", TLSInfo{})
srcAddr, dstAddr = ln1.Addr().String(), ln2.Addr().String()
ln1.Close()
ln2.Close()
} else {
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
}
tlsInfo := testTLSInfo
if !secure {
tlsInfo = TLSInfo{}
}
ln := listen(t, scheme, dstAddr, tlsInfo)
defer ln.Close()
cfg := ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
}
if secure {
cfg.TLSInfo = testTLSInfo
}
p := NewProxy(cfg)
<-p.Ready()
defer p.Close()
data1 := []byte("Hello World!")
donec, writec := make(chan struct{}), make(chan []byte)
go func() {
defer close(donec)
for data := range writec {
send(t, data, scheme, srcAddr, tlsInfo)
}
}()
recvc := make(chan []byte)
go func() {
for i := 0; i < 2; i++ {
recvc <- receive(t, ln)
}
}()
writec <- data1
now := time.Now()
if d := <-recvc; !bytes.Equal(data1, d) {
t.Fatalf("expected %q, got %q", string(data1), string(d))
}
took1 := time.Since(now)
t.Logf("took %v with no latency", took1)
lat, rv := 50*time.Millisecond, 5*time.Millisecond
if delayTx {
p.DelayTx(lat, rv)
}
data2 := []byte("new data")
writec <- data2
now = time.Now()
if d := <-recvc; !bytes.Equal(data2, d) {
t.Fatalf("expected %q, got %q", string(data2), string(d))
}
took2 := time.Since(now)
if delayTx {
t.Logf("took %v with latency %v±%v", took2, lat, rv)
} else {
t.Logf("took %v with no latency", took2)
}
if delayTx {
p.UndelayTx()
if took1 >= took2 {
t.Fatalf("expected took1 %v < took2 %v (with latency)", took1, took2)
}
}
close(writec)
select {
case <-donec:
case <-time.After(3 * time.Second):
t.Fatal("took too long to write")
}
select {
case <-p.Done():
t.Fatal("unexpected done")
case err := <-p.Error():
t.Fatal(err)
default:
}
if err := p.Close(); err != nil {
t.Fatal(err)
}
select {
case <-p.Done():
case err := <-p.Error():
if !strings.HasPrefix(err.Error(), "accept ") &&
!strings.HasSuffix(err.Error(), "use of closed network connection") {
t.Fatal(err)
}
case <-time.After(3 * time.Second):
t.Fatal("took too long to close")
}
}
func TestProxy_Unix_Insecure_DelayAccept(t *testing.T) { testProxyDelayAccept(t, false) }
func TestProxy_Unix_Secure_DelayAccept(t *testing.T) { testProxyDelayAccept(t, true) }
func testProxyDelayAccept(t *testing.T, secure bool) {
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
tlsInfo := testTLSInfo
if !secure {
tlsInfo = TLSInfo{}
}
scheme := "unix"
ln := listen(t, scheme, dstAddr, tlsInfo)
defer ln.Close()
cfg := ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
}
if secure {
cfg.TLSInfo = testTLSInfo
}
p := NewProxy(cfg)
<-p.Ready()
defer p.Close()
data := []byte("Hello World!")
now := time.Now()
send(t, data, scheme, srcAddr, tlsInfo)
if d := receive(t, ln); !bytes.Equal(data, d) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
took1 := time.Since(now)
t.Logf("took %v with no latency", took1)
lat, rv := 700*time.Millisecond, 10*time.Millisecond
p.DelayAccept(lat, rv)
defer p.UndelayAccept()
if err := p.ResetListener(); err != nil {
t.Fatal(err)
}
time.Sleep(200 * time.Millisecond)
now = time.Now()
send(t, data, scheme, srcAddr, tlsInfo)
if d := receive(t, ln); !bytes.Equal(data, d) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
took2 := time.Since(now)
t.Logf("took %v with latency %v±%v", took2, lat, rv)
if took1 >= took2 {
t.Fatalf("expected took1 %v < took2 %v", took1, took2)
}
}
func TestProxy_PauseTx(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
p := NewProxy(ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
})
<-p.Ready()
defer p.Close()
p.PauseTx()
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, TLSInfo{})
recvc := make(chan []byte)
go func() {
recvc <- receive(t, ln)
}()
select {
case d := <-recvc:
t.Fatalf("received unexpected data %q during pause", string(d))
case <-time.After(200 * time.Millisecond):
}
p.UnpauseTx()
select {
case d := <-recvc:
if !bytes.Equal(data, d) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
case <-time.After(2 * time.Second):
t.Fatal("took too long to receive after unpause")
}
}
func TestProxy_BlackholeTx(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
p := NewProxy(ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
})
<-p.Ready()
defer p.Close()
p.BlackholeTx()
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, TLSInfo{})
recvc := make(chan []byte)
go func() {
recvc <- receive(t, ln)
}()
select {
case d := <-recvc:
t.Fatalf("unexpected data receive %q during blackhole", string(d))
case <-time.After(200 * time.Millisecond):
}
p.UnblackholeTx()
// expect different data, old data dropped
data[0]++
send(t, data, scheme, srcAddr, TLSInfo{})
select {
case d := <-recvc:
if !bytes.Equal(data, d) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
case <-time.After(2 * time.Second):
t.Fatal("took too long to receive after unblackhole")
}
}
func TestProxy_CorruptTx(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
p := NewProxy(ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
})
<-p.Ready()
defer p.Close()
p.CorruptTx(func(d []byte) []byte {
d[len(d)/2]++
return d
})
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, TLSInfo{})
if d := receive(t, ln); bytes.Equal(d, data) {
t.Fatalf("expected corrupted data, got %q", string(d))
}
p.UncorruptTx()
send(t, data, scheme, srcAddr, TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected uncorrupted data, got %q", string(d))
}
}
func TestProxy_Shutdown(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
p := NewProxy(ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
})
<-p.Ready()
defer p.Close()
px, _ := p.(*proxy)
px.listener.Close()
time.Sleep(200 * time.Millisecond)
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
}
func TestProxy_ShutdownListener(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
p := NewProxy(ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
})
<-p.Ready()
defer p.Close()
// shut down destination
ln.Close()
time.Sleep(200 * time.Millisecond)
ln = listen(t, scheme, dstAddr, TLSInfo{})
defer ln.Close()
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected %q, got %q", string(data), string(d))
}
}
func TestProxyHTTP_Insecure_DelayTx(t *testing.T) { testProxyHTTP(t, false, true) }
func TestProxyHTTP_Secure_DelayTx(t *testing.T) { testProxyHTTP(t, true, true) }
func TestProxyHTTP_Insecure_DelayRx(t *testing.T) { testProxyHTTP(t, false, false) }
func TestProxyHTTP_Secure_DelayRx(t *testing.T) { testProxyHTTP(t, true, false) }
func testProxyHTTP(t *testing.T, secure, delayTx bool) {
scheme := "tcp"
ln1, ln2 := listen(t, scheme, "localhost:0", TLSInfo{}), listen(t, scheme, "localhost:0", TLSInfo{})
srcAddr, dstAddr := ln1.Addr().String(), ln2.Addr().String()
ln1.Close()
ln2.Close()
mux := http.NewServeMux()
mux.HandleFunc("/hello", func(w http.ResponseWriter, req *http.Request) {
d, err := ioutil.ReadAll(req.Body)
if err != nil {
t.Fatal(err)
}
if _, err = w.Write([]byte(fmt.Sprintf("%q(confirmed)", string(d)))); err != nil {
t.Fatal(err)
}
})
var tlsConfig *tls.Config
var err error
if secure {
tlsConfig, err = testTLSInfo.ServerConfig()
if err != nil {
t.Fatal(err)
}
}
srv := &http.Server{
Addr: dstAddr,
Handler: mux,
TLSConfig: tlsConfig,
}
donec := make(chan struct{})
defer func() {
srv.Close()
<-donec
}()
go func() {
defer close(donec)
if !secure {
srv.ListenAndServe()
} else {
srv.ListenAndServeTLS(testTLSInfo.CertFile, testTLSInfo.KeyFile)
}
}()
time.Sleep(200 * time.Millisecond)
cfg := ProxyConfig{
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
Logger: grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5),
}
if secure {
cfg.TLSInfo = testTLSInfo
}
p := NewProxy(cfg)
<-p.Ready()
defer p.Close()
data := "Hello World!"
now := time.Now()
var resp *http.Response
if secure {
tp, terr := NewTransport(testTLSInfo, 3*time.Second)
if terr != nil {
t.Fatal(terr)
}
cli := &http.Client{Transport: tp}
resp, err = cli.Post("https://"+srcAddr+"/hello", "", strings.NewReader(data))
} else {
resp, err = http.Post("http://"+srcAddr+"/hello", "", strings.NewReader(data))
}
if err != nil {
t.Fatal(err)
}
d, err := ioutil.ReadAll(resp.Body)
if err != nil {
t.Fatal(err)
}
took1 := time.Since(now)
t.Logf("took %v with no latency", took1)
rs1 := string(d)
exp := fmt.Sprintf("%q(confirmed)", data)
if rs1 != exp {
t.Fatalf("got %q, expected %q", rs1, exp)
}
lat, rv := 100*time.Millisecond, 10*time.Millisecond
if delayTx {
p.DelayTx(lat, rv)
defer p.UndelayTx()
} else {
p.DelayRx(lat, rv)
defer p.UndelayRx()
}
now = time.Now()
if secure {
tp, terr := NewTransport(testTLSInfo, 3*time.Second)
if terr != nil {
t.Fatal(terr)
}
cli := &http.Client{Transport: tp}
resp, err = cli.Post("https://"+srcAddr+"/hello", "", strings.NewReader(data))
} else {
resp, err = http.Post("http://"+srcAddr+"/hello", "", strings.NewReader(data))
}
if err != nil {
t.Fatal(err)
}
d, err = ioutil.ReadAll(resp.Body)
if err != nil {
t.Fatal(err)
}
took2 := time.Since(now)
t.Logf("took %v with latency %v±%v", took2, lat, rv)
rs2 := string(d)
if rs2 != exp {
t.Fatalf("got %q, expected %q", rs2, exp)
}
if took1 > took2 {
t.Fatalf("expected took1 %v < took2 %v", took1, took2)
}
}
func newUnixAddr() string {
now := time.Now().UnixNano()
rand.Seed(now)
addr := fmt.Sprintf("%X%X.unix-conn", now, rand.Intn(35000))
os.RemoveAll(addr)
return addr
}
func listen(t *testing.T, scheme, addr string, tlsInfo TLSInfo) (ln net.Listener) {
var err error
if !tlsInfo.Empty() {
ln, err = NewListener(addr, scheme, &tlsInfo)
} else {
ln, err = net.Listen(scheme, addr)
}
if err != nil {
t.Fatal(err)
}
return ln
}
func send(t *testing.T, data []byte, scheme, addr string, tlsInfo TLSInfo) {
var out net.Conn
var err error
if !tlsInfo.Empty() {
tp, terr := NewTransport(tlsInfo, 3*time.Second)
if terr != nil {
t.Fatal(terr)
}
out, err = tp.Dial(scheme, addr)
} else {
out, err = net.Dial(scheme, addr)
}
if err != nil {
t.Fatal(err)
}
if _, err = out.Write(data); err != nil {
t.Fatal(err)
}
if err = out.Close(); err != nil {
t.Fatal(err)
}
}
func receive(t *testing.T, ln net.Listener) (data []byte) {
buf := bytes.NewBuffer(make([]byte, 0, 1024))
for {
in, err := ln.Accept()
if err != nil {
t.Fatal(err)
}
var n int64
n, err = buf.ReadFrom(in)
if err != nil {
t.Fatal(err)
}
if n > 0 {
break
}
}
return buf.Bytes()
}

10
test
View File

@@ -111,7 +111,7 @@ function functional_pass {
for a in 1 2 3; do
mkdir -p ./agent-$a
./bin/etcd-agent -etcd-path ./bin/etcd -etcd-log-dir "./agent-$a" -port ":${a}9027" -use-root=false &
./bin/etcd-agent -etcd-path ./bin/etcd -etcd-log-dir "./agent-$a" -port ":${a}9027" &
pid="$!"
agent_pids="${agent_pids} $pid"
done
@@ -126,10 +126,12 @@ function functional_pass {
echo "Starting 'etcd-tester'"
./bin/etcd-tester \
-agent-endpoints "127.0.0.1:19027,127.0.0.1:29027,127.0.0.1:39027" \
-client-ports 12379,22379,32379 \
-peer-ports 12380,22380,32380 \
-client-ports 1379,2379,3379 \
-advertise-client-ports 13790,23790,33790 \
-peer-ports 1380,2380,3380 \
-advertise-peer-ports 13800,23800,33800 \
-limit 1 \
-schedule-cases "0 1 2 3 4 5" \
-schedule-cases "0 1 2 3 4 5 6 7 8 9" \
-stress-qps 1000 \
-stress-key-txn-count 100 \
-stress-key-txn-ops 10 \

View File

@@ -0,0 +1,14 @@
s1: bin/etcd --name s1 --data-dir /tmp/etcd-test-proxy-data.s1 --listen-client-urls http://127.0.0.1:1379 --advertise-client-urls http://127.0.0.1:13790 --listen-peer-urls http://127.0.0.1:1380 --initial-advertise-peer-urls http://127.0.0.1:13800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
s1-client-proxy: bin/etcd-test-proxy --from localhost:13790 --to localhost:1379 --http-port 1378
s1-peer-proxy: bin/etcd-test-proxy --from localhost:13800 --to localhost:1380 --http-port 1381
s2: bin/etcd --name s2 --data-dir /tmp/etcd-test-proxy-data.s2 --listen-client-urls http://127.0.0.1:2379 --advertise-client-urls http://127.0.0.1:23790 --listen-peer-urls http://127.0.0.1:2380 --initial-advertise-peer-urls http://127.0.0.1:23800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
s2-client-proxy: bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378
s2-peer-proxy: bin/etcd-test-proxy --from localhost:23800 --to localhost:2380 --http-port 2381
s3: bin/etcd --name s3 --data-dir /tmp/etcd-test-proxy-data.s3 --listen-client-urls http://127.0.0.1:3379 --advertise-client-urls http://127.0.0.1:33790 --listen-peer-urls http://127.0.0.1:3380 --initial-advertise-peer-urls http://127.0.0.1:33800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
s3-client-proxy: bin/etcd-test-proxy --from localhost:33790 --to localhost:3379 --http-port 3378
s3-client-proxy: bin/etcd-test-proxy --from localhost:33800 --to localhost:3380 --http-port 3381

View File

@@ -0,0 +1,184 @@
#### etcd-test-proxy
Proxy layer that simulates various network conditions.
Test locally
```bash
$ ./build
$ ./bin/etcd
$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
$ ./bin/etcd-test-proxy --help
$ ./bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378 --verbose
$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:2379 put foo bar
$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
```
Proxy overhead per request is under 500μs
```bash
$ go build -v -o ./bin/benchmark ./cmd/tools/benchmark
$ ./bin/benchmark \
--endpoints localhost:2379 \
--conns 5 \
--clients 15 \
put \
--key-size 48 \
--val-size 50000 \
--total 10000
<<COMMENT
Summary:
Total: 8.4611 secs.
Slowest: 0.1324 secs.
Fastest: 0.0011 secs.
Average: 0.0121 secs.
Stddev: 0.0125 secs.
Requests/sec: 1181.8758
Response time histogram:
0.0011 [1] |
0.0142 [7899] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
0.0273 [1339] |∎∎∎∎∎∎
0.0405 [543] |∎∎
0.0536 [67] |
0.0667 [49] |
0.0798 [9] |
0.0930 [15] |
0.1061 [42] |
0.1192 [21] |
0.1324 [15] |
Latency distribution:
10% in 0.0049 secs.
25% in 0.0064 secs.
50% in 0.0085 secs.
75% in 0.0126 secs.
90% in 0.0243 secs.
95% in 0.0307 secs.
99% in 0.0686 secs.
99.9% in 0.1294 secs.
COMMENT
$ ./bin/benchmark \
--endpoints localhost:23790 \
--conns 5 \
--clients 15 \
put \
--key-size 48 \
--val-size 50000 \
--total 10000
<<COMMENT
Summary:
Total: 9.1128 secs.
Slowest: 0.1363 secs.
Fastest: 0.0015 secs.
Average: 0.0131 secs.
Stddev: 0.0113 secs.
Requests/sec: 1097.3613
Response time histogram:
0.0015 [1] |
0.0150 [7407] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
0.0285 [2017] |∎∎∎∎∎∎∎∎∎∎
0.0419 [440] |∎∎
0.0554 [30] |
0.0689 [13] |
0.0824 [12] |
0.0959 [48] |
0.1093 [2] |
0.1228 [16] |
0.1363 [14] |
Latency distribution:
10% in 0.0054 secs.
25% in 0.0071 secs.
50% in 0.0100 secs.
75% in 0.0153 secs.
90% in 0.0241 secs.
95% in 0.0297 secs.
99% in 0.0584 secs.
99.9% in 0.1312 secs.
COMMENT
```
Delay client transmit
```bash
$ curl -L http://localhost:2378/delay-tx -X PUT \
-d "latency=5s&random-variable=100ms"
# added send latency 5s±100ms (current latency 4.92143955s)
$ curl -L http://localhost:2378/delay-tx
# current send latency 4.92143955s
$ ETCDCTL_API=3 ./bin/etcdctl \
--endpoints localhost:23790 \
--command-timeout=3s \
put foo bar
# Error: context deadline exceeded
$ curl -L http://localhost:2378/delay-tx -X DELETE
# removed latency 4.92143955s
$ curl -L http://localhost:2378/delay-tx
# current send latency 0s
$ ETCDCTL_API=3 ./bin/etcdctl \
--endpoints localhost:23790 \
--command-timeout=3s \
put foo bar
# OK
```
Pause client transmit
```bash
$ curl -L http://localhost:2378/pause-tx -X PUT
# paused forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
$ ETCDCTL_API=3 ./bin/etcdctl \
--endpoints localhost:23790 \
put foo bar
# Error: context deadline exceeded
$ curl -L http://localhost:2378/pause-tx -X DELETE
# unpaused forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
```
Drop client packets
```bash
$ curl -L http://localhost:2378/blackhole-tx -X PUT
# blackholed; dropping packets [tcp://localhost:23790 -> tcp://localhost:2379]
$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
# Error: context deadline exceeded
$ curl -L http://localhost:2378/blackhole-tx -X DELETE
# unblackholed; restart forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
```
Trigger leader election
```bash
$ ./build
$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
$ rm -rf /tmp/etcd-test-proxy-data.s*
$ goreman -f ./tools/etcd-test-proxy/Procfile start
$ ETCDCTL_API=3 ./bin/etcdctl \
--endpoints localhost:13790,localhost:23790,localhost:33790 \
member list
# isolate s1 when s1 is the current leader
$ curl -L http://localhost:1381/blackhole-tx -X PUT
$ curl -L http://localhost:1381/blackhole-rx -X PUT
# s1 becomes follower after election timeout
```

View File

@@ -0,0 +1,216 @@
// Copyright 2018 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"context"
"flag"
"fmt"
"net/http"
"net/url"
"os"
"os/signal"
"syscall"
"time"
"github.com/coreos/etcd/pkg/transport"
"google.golang.org/grpc/grpclog"
)
var from string
var to string
var httpPort int
var verbose bool
func main() {
// TODO: support TLS
flag.StringVar(&from, "from", "localhost:23790", "Address URL to proxy from.")
flag.StringVar(&to, "to", "localhost:2379", "Address URL to forward.")
flag.IntVar(&httpPort, "http-port", 2378, "Port to serve etcd-test-proxy API.")
flag.BoolVar(&verbose, "verbose", false, "'true' to run proxy in verbose mode.")
flag.Usage = func() {
fmt.Fprintf(os.Stderr, "Usage of %q:\n", os.Args[0])
fmt.Fprintln(os.Stderr, `
etcd-test-proxy simulates various network conditions for etcd testing purposes.
See README.md for more examples.
Example:
# build etcd
$ ./build
$ ./bin/etcd
# build etcd-test-proxy
$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
# to test etcd with proxy layer
$ ./bin/etcd-test-proxy --help
$ ./bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378 --verbose
$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:2379 put foo bar
$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
`)
flag.PrintDefaults()
}
flag.Parse()
cfg := transport.ProxyConfig{
From: url.URL{Scheme: "tcp", Host: from},
To: url.URL{Scheme: "tcp", Host: to},
}
if verbose {
cfg.Logger = grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5)
}
p := transport.NewProxy(cfg)
<-p.Ready()
defer p.Close()
mux := http.NewServeMux()
mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
w.Write([]byte(fmt.Sprintf("proxying [%s -> %s]\n", p.From(), p.To())))
})
mux.HandleFunc("/delay-tx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodGet:
w.Write([]byte(fmt.Sprintf("current send latency %v\n", p.LatencyTx())))
case http.MethodPut, http.MethodPost:
if err := req.ParseForm(); err != nil {
w.Write([]byte(fmt.Sprintf("wrong form %q\n", err.Error())))
return
}
lat, err := time.ParseDuration(req.PostForm.Get("latency"))
if err != nil {
w.Write([]byte(fmt.Sprintf("wrong latency form %q\n", err.Error())))
return
}
rv, err := time.ParseDuration(req.PostForm.Get("random-variable"))
if err != nil {
w.Write([]byte(fmt.Sprintf("wrong random-variable form %q\n", err.Error())))
return
}
p.DelayTx(lat, rv)
w.Write([]byte(fmt.Sprintf("added send latency %v±%v (current latency %v)\n", lat, rv, p.LatencyTx())))
case http.MethodDelete:
lat := p.LatencyTx()
p.UndelayTx()
w.Write([]byte(fmt.Sprintf("removed latency %v\n", lat)))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
mux.HandleFunc("/delay-rx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodGet:
w.Write([]byte(fmt.Sprintf("current receive latency %v\n", p.LatencyRx())))
case http.MethodPut, http.MethodPost:
if err := req.ParseForm(); err != nil {
w.Write([]byte(fmt.Sprintf("wrong form %q\n", err.Error())))
return
}
lat, err := time.ParseDuration(req.PostForm.Get("latency"))
if err != nil {
w.Write([]byte(fmt.Sprintf("wrong latency form %q\n", err.Error())))
return
}
rv, err := time.ParseDuration(req.PostForm.Get("random-variable"))
if err != nil {
w.Write([]byte(fmt.Sprintf("wrong random-variable form %q\n", err.Error())))
return
}
p.DelayRx(lat, rv)
w.Write([]byte(fmt.Sprintf("added receive latency %v±%v (current latency %v)\n", lat, rv, p.LatencyRx())))
case http.MethodDelete:
lat := p.LatencyRx()
p.UndelayRx()
w.Write([]byte(fmt.Sprintf("removed latency %v\n", lat)))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
mux.HandleFunc("/pause-tx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodPut, http.MethodPost:
p.PauseTx()
w.Write([]byte(fmt.Sprintf("paused forwarding [%s -> %s]\n", p.From(), p.To())))
case http.MethodDelete:
p.UnpauseTx()
w.Write([]byte(fmt.Sprintf("unpaused forwarding [%s -> %s]\n", p.From(), p.To())))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
mux.HandleFunc("/pause-rx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodPut, http.MethodPost:
p.PauseRx()
w.Write([]byte(fmt.Sprintf("paused forwarding [%s <- %s]\n", p.From(), p.To())))
case http.MethodDelete:
p.UnpauseRx()
w.Write([]byte(fmt.Sprintf("unpaused forwarding [%s <- %s]\n", p.From(), p.To())))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
mux.HandleFunc("/blackhole-tx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodPut, http.MethodPost:
p.BlackholeTx()
w.Write([]byte(fmt.Sprintf("blackholed; dropping packets [%s -> %s]\n", p.From(), p.To())))
case http.MethodDelete:
p.UnblackholeTx()
w.Write([]byte(fmt.Sprintf("unblackholed; restart forwarding [%s -> %s]\n", p.From(), p.To())))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
mux.HandleFunc("/blackhole-rx", func(w http.ResponseWriter, req *http.Request) {
switch req.Method {
case http.MethodPut, http.MethodPost:
p.BlackholeRx()
w.Write([]byte(fmt.Sprintf("blackholed; dropping packets [%s <- %s]\n", p.From(), p.To())))
case http.MethodDelete:
p.UnblackholeRx()
w.Write([]byte(fmt.Sprintf("unblackholed; restart forwarding [%s <- %s]\n", p.From(), p.To())))
default:
w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
}
})
srv := &http.Server{
Addr: fmt.Sprintf(":%d", httpPort),
Handler: mux,
}
defer srv.Close()
sig := make(chan os.Signal, 1)
signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
defer signal.Stop(sig)
go func() {
s := <-sig
fmt.Printf("\n\nreceived signal %q, shutting down HTTP server\n\n", s)
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
err := srv.Shutdown(ctx)
cancel()
fmt.Printf("gracefully stopped HTTP server with %v\n\n", err)
os.Exit(0)
}()
fmt.Printf("\nserving HTTP server http://localhost:%d\n\n", httpPort)
err := srv.ListenAndServe()
fmt.Printf("HTTP server exit with error %v\n", err)
}

View File

@@ -1,4 +0,0 @@
agent-1: mkdir -p agent-1 && cd agent-1 && ../bin/etcd-agent -etcd-path ../bin/etcd -port 127.0.0.1:19027 -use-root=false
agent-2: mkdir -p agent-2 && cd agent-2 && ../bin/etcd-agent -etcd-path ../bin/etcd -port 127.0.0.1:29027 -use-root=false
agent-3: mkdir -p agent-3 && cd agent-3 && ../bin/etcd-agent -etcd-path ../bin/etcd -port 127.0.0.1:39027 -use-root=false
stresser: sleep 1s && bin/etcd-tester -agent-endpoints "127.0.0.1:19027,127.0.0.1:29027,127.0.0.1:39027" -client-ports 12379,22379,32379 -peer-ports 12380,22380,32380

View File

@@ -10,42 +10,38 @@ The environment of the cluster must be stable enough, so etcd test suite can ass
## etcd agent
etcd agent is a daemon on each machines. It can start, stop, restart, isolate and terminate an etcd process. The agent exposes these functionality via HTTP RPC.
etcd agent is a daemon on each machines. It can start, stop, restart, isolate and terminate an etcd process. The agent exposes these functionality via HTTP RPC.
## etcd tester
etcd functional tester control the progress of the functional tests. It calls the RPC of the etcd agent to simulate various test cases. For example, it can start a three members cluster by sending three start RPC calls to three different etcd agents. It can make one of the member failed by sending stop RPC call to one etcd agent.
## with Docker (optionally)
### Run locally
To run the functional tests using Docker, the provided script can be used to set up an environment using Docker Compose.
Script (on linux):
```sh
./tools/functional-tester/test
```
$ PASSES=functional ./test
```
Running the script requires:
### Run with Docker
- Docker 1.9+ (with networking support) - to create isolated network
- docker-compose - to create etcd cluster and tester
- A multi-arch Go toolchain (OSX)
To run locally, first build tester image:
Notes:
- Docker image is based on Alpine Linux OS running in privileged mode to allow iptables manipulation.
- To specify testing parameters (etcd-tester arguments) modify tools/functional-tester/docker/docker-compose.yml or start etcd-tester manually
- (OSX) make sure that etcd binary is built for linux/amd64 (eg. `rm bin/etcd;GOOS=linux GOARCH=amd64 ./tools/functional-tester/test`) otherwise it will return `exec format error`
```bash
pushd ../..
GO_VERSION=1.9.3 \
make build-docker-functional-tester \
-f ./hack/scripts-dev/Makefile
## with Goreman
To run the functional tests on a single machine using Goreman, build with the provided build script and run with the provided Procfile:
```sh
./tools/functional-tester/build
goreman -f tools/functional-tester/Procfile start
popd
```
Notes:
- The etcd-agent will not run with root privileges; iptables manipulation is disabled.
- To specify testing parameters (etcd-tester arguments) modify tools/functional-tester/Procfile or start etcd-tester manually
And run [example scripts](./scripts).
```bash
./scripts/agent-1.sh
./scripts/agent-2.sh
./scripts/agent-3.sh
./scripts/tester-limit.sh
```

View File

@@ -1,8 +0,0 @@
FROM alpine
RUN apk update
RUN apk add -v iptables sudo
ADD bin/etcd-agent /
ADD bin/etcd /
ADD bin/etcd-tester /
RUN mkdir /failure_archive
CMD ["./etcd-agent", "-etcd-path", "./etcd"]

View File

@@ -1,28 +0,0 @@
# build according provided Dockerfile
a1:
build: .
privileged: true
net: etcd-functional
a2:
build: .
privileged: true
net: etcd-functional
a3:
build: .
privileged: true
net: etcd-functional
tester:
build: .
privileged: true
net: etcd-functional
command:
- /etcd-tester
- -agent-endpoints
- "172.20.0.2:9027,172.20.0.3:9027,172.20.0.4:9027"
- -limit
- "1"
- -stress-key-count
- "1"
- -stress-key-size
- "1"

View File

@@ -15,14 +15,19 @@
package main
import (
"fmt"
"net"
"net/url"
"os"
"os/exec"
"path/filepath"
"strconv"
"sync"
"syscall"
"time"
"github.com/coreos/etcd/pkg/fileutil"
"github.com/coreos/etcd/pkg/netutil"
"github.com/coreos/etcd/pkg/transport"
"github.com/coreos/etcd/tools/functional-tester/etcd-agent/client"
)
@@ -40,13 +45,15 @@ type Agent struct {
logfile *os.File
cfg AgentConfig
pmu sync.Mutex
advertisePortToProxy map[int]transport.Proxy
}
type AgentConfig struct {
EtcdPath string
LogDir string
FailpointAddr string
UseRoot bool
}
func newAgent(cfg AgentConfig) (*Agent, error) {
@@ -69,7 +76,13 @@ func newAgent(cfg AgentConfig) (*Agent, error) {
return nil, err
}
return &Agent{state: stateUninitialized, cmd: c, logfile: f, cfg: cfg}, nil
return &Agent{
state: stateUninitialized,
cmd: c,
logfile: f,
cfg: cfg,
advertisePortToProxy: make(map[int]transport.Proxy),
}, nil
}
// start starts a new etcd process with the given args.
@@ -85,6 +98,85 @@ func (a *Agent) start(args ...string) error {
}
a.state = stateStarted
a.pmu.Lock()
defer a.pmu.Unlock()
if len(a.advertisePortToProxy) == 0 {
// enough time for etcd start before setting up proxy
time.Sleep(time.Second)
var (
err error
s string
listenClientURL *url.URL
advertiseClientURL *url.URL
advertiseClientURLPort int
listenPeerURL *url.URL
advertisePeerURL *url.URL
advertisePeerURLPort int
)
for i := range args {
switch args[i] {
case "--listen-client-urls":
listenClientURL, err = url.Parse(args[i+1])
if err != nil {
return err
}
case "--advertise-client-urls":
advertiseClientURL, err = url.Parse(args[i+1])
if err != nil {
return err
}
_, s, err = net.SplitHostPort(advertiseClientURL.Host)
if err != nil {
return err
}
advertiseClientURLPort, err = strconv.Atoi(s)
if err != nil {
return err
}
case "--listen-peer-urls":
listenPeerURL, err = url.Parse(args[i+1])
if err != nil {
return err
}
case "--initial-advertise-peer-urls":
advertisePeerURL, err = url.Parse(args[i+1])
if err != nil {
return err
}
_, s, err = net.SplitHostPort(advertisePeerURL.Host)
if err != nil {
return err
}
advertisePeerURLPort, err = strconv.Atoi(s)
if err != nil {
return err
}
}
}
clientProxy := transport.NewProxy(transport.ProxyConfig{
From: *advertiseClientURL,
To: *listenClientURL,
})
select {
case err = <-clientProxy.Error():
return err
case <-time.After(time.Second):
}
a.advertisePortToProxy[advertiseClientURLPort] = clientProxy
peerProxy := transport.NewProxy(transport.ProxyConfig{
From: *advertisePeerURL,
To: *listenPeerURL,
})
select {
case err = <-peerProxy.Error():
return err
case <-time.After(time.Second):
}
a.advertisePortToProxy[advertisePeerURLPort] = peerProxy
}
return nil
}
@@ -94,6 +186,24 @@ func (a *Agent) stopWithSig(sig os.Signal) error {
return nil
}
a.pmu.Lock()
if len(a.advertisePortToProxy) > 0 {
for _, p := range a.advertisePortToProxy {
if err := p.Close(); err != nil {
a.pmu.Unlock()
return err
}
select {
case <-p.Done():
// enough time to release port
time.Sleep(time.Second)
case <-time.After(time.Second):
}
}
a.advertisePortToProxy = make(map[int]transport.Proxy)
}
a.pmu.Unlock()
err := stopWithSig(a.cmd, sig)
if err != nil {
return err
@@ -178,27 +288,46 @@ func (a *Agent) terminate() error {
}
func (a *Agent) dropPort(port int) error {
if !a.cfg.UseRoot {
return nil
a.pmu.Lock()
defer a.pmu.Unlock()
p, ok := a.advertisePortToProxy[port]
if !ok {
return fmt.Errorf("%d does not have proxy", port)
}
return netutil.DropPort(port)
p.BlackholeTx()
p.BlackholeRx()
return nil
}
func (a *Agent) recoverPort(port int) error {
if !a.cfg.UseRoot {
return nil
a.pmu.Lock()
defer a.pmu.Unlock()
p, ok := a.advertisePortToProxy[port]
if !ok {
return fmt.Errorf("%d does not have proxy", port)
}
return netutil.RecoverPort(port)
p.UnblackholeTx()
p.UnblackholeRx()
return nil
}
func (a *Agent) setLatency(ms, rv int) error {
if !a.cfg.UseRoot {
return nil
}
a.pmu.Lock()
defer a.pmu.Unlock()
if ms == 0 {
return netutil.RemoveLatency()
for _, p := range a.advertisePortToProxy {
p.UndelayTx()
p.UndelayRx()
}
}
return netutil.SetLatency(ms, rv)
for _, p := range a.advertisePortToProxy {
p.DelayTx(time.Duration(ms)*time.Millisecond, time.Duration(rv)*time.Millisecond)
p.DelayRx(time.Duration(ms)*time.Millisecond, time.Duration(rv)*time.Millisecond)
}
return nil
}
func (a *Agent) status() client.Status {

View File

@@ -16,7 +16,6 @@ package main
import (
"flag"
"fmt"
"os"
"path/filepath"
@@ -29,7 +28,6 @@ func main() {
etcdPath := flag.String("etcd-path", filepath.Join(os.Getenv("GOPATH"), "bin/etcd"), "the path to etcd binary")
etcdLogDir := flag.String("etcd-log-dir", "etcd-log", "directory to store etcd logs, data directories, failure archive")
port := flag.String("port", ":9027", "port to serve agent server")
useRoot := flag.Bool("use-root", true, "use root permissions")
failpointAddr := flag.String("failpoint-addr", ":2381", "interface for gofail's HTTP server")
flag.Parse()
@@ -37,17 +35,7 @@ func main() {
EtcdPath: *etcdPath,
LogDir: *etcdLogDir,
FailpointAddr: *failpointAddr,
UseRoot: *useRoot,
}
if *useRoot && os.Getuid() != 0 {
fmt.Println("got --use-root=true but not root user")
os.Exit(1)
}
if !*useRoot {
fmt.Println("root permissions disabled, agent will not modify network")
}
a, err := newAgent(cfg)
if err != nil {
plog.Fatal(err)

View File

@@ -30,10 +30,12 @@ import (
// agentConfig holds information needed to interact/configure an agent and its etcd process
type agentConfig struct {
endpoint string
clientPort int
peerPort int
failpointPort int
endpoint string
clientPort int
advertiseClientPort int
peerPort int
advertisePeerPort int
failpointPort int
}
type cluster struct {
@@ -61,12 +63,14 @@ func (c *cluster) bootstrap() error {
return err
}
members[i] = &member{
Agent: agent,
Endpoint: a.endpoint,
Name: fmt.Sprintf("etcd-%d", i),
ClientURL: fmt.Sprintf("http://%s:%d", host, a.clientPort),
PeerURL: fmt.Sprintf("http://%s:%d", host, a.peerPort),
FailpointURL: fmt.Sprintf("http://%s:%d", host, a.failpointPort),
Agent: agent,
Endpoint: a.endpoint,
Name: fmt.Sprintf("etcd-%d", i),
ClientURL: fmt.Sprintf("http://%s:%d", host, a.clientPort),
AdvertiseClientURL: fmt.Sprintf("http://%s:%d", host, a.advertiseClientPort),
PeerURL: fmt.Sprintf("http://%s:%d", host, a.peerPort),
AdvertisePeerURL: fmt.Sprintf("http://%s:%d", host, a.advertisePeerPort),
FailpointURL: fmt.Sprintf("http://%s:%d", host, a.failpointPort),
}
memberNameURLs[i] = members[i].ClusterEntry()
}

View File

@@ -128,7 +128,10 @@ func (f *failureDelay) Inject(c *cluster, round int) error {
if err := f.failure.Inject(c, round); err != nil {
return err
}
time.Sleep(f.delayDuration)
if f.delayDuration > 0 {
plog.Infof("sleeping delay duration %v for %q", f.delayDuration, f.failure.Desc())
time.Sleep(f.delayDuration)
}
return nil
}

View File

@@ -24,6 +24,9 @@ const (
slowNetworkLatency = 500 // 500 millisecond
randomVariation = 50
// delay duration to trigger leader election (default election timeout 1s)
triggerElectionDur = 5 * time.Second
// Wait more when it recovers from slow network, because network layer
// needs extra time to propagate traffic control (tc command) change.
// Otherwise, we get different hash values from the previous revision.
@@ -82,19 +85,27 @@ func injectDropPort(m *member) error { return m.Agent.DropPort(m.peerPort()) }
func recoverDropPort(m *member) error { return m.Agent.RecoverPort(m.peerPort()) }
func newFailureIsolate() failure {
return &failureOne{
f := &failureOne{
description: "isolate one member",
injectMember: injectDropPort,
recoverMember: recoverDropPort,
}
return &failureDelay{
failure: f,
delayDuration: triggerElectionDur,
}
}
func newFailureIsolateAll() failure {
return &failureAll{
f := &failureAll{
description: "isolate all members",
injectMember: injectDropPort,
recoverMember: recoverDropPort,
}
return &failureDelay{
failure: f,
delayDuration: triggerElectionDur,
}
}
func injectLatency(m *member) error {
@@ -115,11 +126,15 @@ func recoverLatency(m *member) error {
func newFailureSlowNetworkOneMember() failure {
desc := fmt.Sprintf("slow down one member's network by adding %d ms latency", slowNetworkLatency)
return &failureOne{
f := &failureOne{
description: description(desc),
injectMember: injectLatency,
recoverMember: recoverLatency,
}
return &failureDelay{
failure: f,
delayDuration: triggerElectionDur,
}
}
func newFailureSlowNetworkLeader() failure {
@@ -129,15 +144,23 @@ func newFailureSlowNetworkLeader() failure {
injectMember: injectLatency,
recoverMember: recoverLatency,
}
return &failureLeader{ff, 0}
f := &failureLeader{ff, 0}
return &failureDelay{
failure: f,
delayDuration: triggerElectionDur,
}
}
func newFailureSlowNetworkAll() failure {
return &failureAll{
f := &failureAll{
description: "slow down all members' network",
injectMember: injectLatency,
recoverMember: recoverLatency,
}
return &failureDelay{
failure: f,
delayDuration: triggerElectionDur,
}
}
func newFailureNop() failure {

View File

@@ -41,7 +41,9 @@ const (
func main() {
endpointStr := flag.String("agent-endpoints", "localhost:9027", "HTTP RPC endpoints of agents. Do not specify the schema.")
clientPorts := flag.String("client-ports", "", "etcd client port for each agent endpoint")
advertiseClientPorts := flag.String("advertise-client-ports", "", "etcd advertise client port for each agent endpoint")
peerPorts := flag.String("peer-ports", "", "etcd peer port for each agent endpoint")
advertisePeerPorts := flag.String("advertise-peer-ports", "", "etcd advertise peer port for each agent endpoint")
failpointPorts := flag.String("failpoint-ports", "", "etcd failpoint port for each agent endpoint")
stressKeyLargeSize := flag.Uint("stress-key-large-size", 32*1024+1, "the size of each large key written into etcd.")
@@ -67,14 +69,18 @@ func main() {
eps := strings.Split(*endpointStr, ",")
cports := portsFromArg(*clientPorts, len(eps), defaultClientPort)
acports := portsFromArg(*advertiseClientPorts, len(eps), defaultClientPort)
pports := portsFromArg(*peerPorts, len(eps), defaultPeerPort)
apports := portsFromArg(*advertisePeerPorts, len(eps), defaultPeerPort)
fports := portsFromArg(*failpointPorts, len(eps), defaultFailpointPort)
agents := make([]agentConfig, len(eps))
for i := range eps {
agents[i].endpoint = eps[i]
agents[i].clientPort = cports[i]
agents[i].advertiseClientPort = acports[i]
agents[i].peerPort = pports[i]
agents[i].advertisePeerPort = apports[i]
agents[i].failpointPort = fports[i]
}

View File

@@ -29,23 +29,25 @@ import (
)
type member struct {
Agent client.Agent
Endpoint string
Name string
ClientURL string
PeerURL string
FailpointURL string
Agent client.Agent
Endpoint string
Name string
ClientURL string
AdvertiseClientURL string
PeerURL string
AdvertisePeerURL string
FailpointURL string
}
func (m *member) ClusterEntry() string { return m.Name + "=" + m.PeerURL }
func (m *member) ClusterEntry() string { return m.Name + "=" + m.AdvertisePeerURL }
func (m *member) Flags() []string {
return []string{
"--name", m.Name,
"--listen-client-urls", m.ClientURL,
"--advertise-client-urls", m.ClientURL,
"--advertise-client-urls", m.AdvertiseClientURL,
"--listen-peer-urls", m.PeerURL,
"--initial-advertise-peer-urls", m.PeerURL,
"--initial-advertise-peer-urls", m.AdvertisePeerURL,
"--initial-cluster-state", "new",
"--experimental-initial-corrupt-check",
}
@@ -54,7 +56,7 @@ func (m *member) Flags() []string {
func (m *member) CheckCompact(rev int64) error {
cli, err := m.newClientV3()
if err != nil {
return fmt.Errorf("%v (endpoint %s)", err, m.ClientURL)
return fmt.Errorf("%v (endpoint %s)", err, m.AdvertiseClientURL)
}
defer cli.Close()
@@ -64,29 +66,29 @@ func (m *member) CheckCompact(rev int64) error {
cancel()
if !ok {
return fmt.Errorf("watch channel terminated (endpoint %s)", m.ClientURL)
return fmt.Errorf("watch channel terminated (endpoint %s)", m.AdvertiseClientURL)
}
if wr.CompactRevision != rev {
return fmt.Errorf("got compact revision %v, wanted %v (endpoint %s)", wr.CompactRevision, rev, m.ClientURL)
return fmt.Errorf("got compact revision %v, wanted %v (endpoint %s)", wr.CompactRevision, rev, m.AdvertiseClientURL)
}
return nil
}
func (m *member) Defrag() error {
plog.Printf("defragmenting %s\n", m.ClientURL)
plog.Printf("defragmenting %s\n", m.AdvertiseClientURL)
cli, err := m.newClientV3()
if err != nil {
return err
}
defer cli.Close()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
_, err = cli.Defragment(ctx, m.ClientURL)
_, err = cli.Defragment(ctx, m.AdvertiseClientURL)
cancel()
if err != nil {
return err
}
plog.Printf("defragmented %s\n", m.ClientURL)
plog.Printf("defragmented %s\n", m.AdvertiseClientURL)
return nil
}
@@ -114,7 +116,7 @@ func (m *member) Rev(ctx context.Context) (int64, error) {
return 0, err
}
defer cli.Close()
resp, err := cli.Status(ctx, m.ClientURL)
resp, err := cli.Status(ctx, m.AdvertiseClientURL)
if err != nil {
return 0, err
}
@@ -127,7 +129,7 @@ func (m *member) IsLeader() (bool, error) {
return false, err
}
defer cli.Close()
resp, err := cli.Status(context.Background(), m.ClientURL)
resp, err := cli.Status(context.Background(), m.AdvertiseClientURL)
if err != nil {
return false, err
}
@@ -137,7 +139,7 @@ func (m *member) IsLeader() (bool, error) {
func (m *member) SetHealthKeyV3() error {
cli, err := m.newClientV3()
if err != nil {
return fmt.Errorf("%v (%s)", err, m.ClientURL)
return fmt.Errorf("%v (%s)", err, m.AdvertiseClientURL)
}
defer cli.Close()
// give enough time-out in case expensive requests (range/delete) are pending
@@ -145,14 +147,14 @@ func (m *member) SetHealthKeyV3() error {
_, err = cli.Put(ctx, "health", "good")
cancel()
if err != nil {
return fmt.Errorf("%v (%s)", err, m.ClientURL)
return fmt.Errorf("%v (%s)", err, m.AdvertiseClientURL)
}
return nil
}
func (m *member) newClientV3() (*clientv3.Client, error) {
return clientv3.New(clientv3.Config{
Endpoints: []string{m.ClientURL},
Endpoints: []string{m.AdvertiseClientURL},
DialTimeout: 5 * time.Second,
})
}
@@ -163,7 +165,7 @@ func (m *member) dialGRPC() (*grpc.ClientConn, error) {
// grpcAddr gets the host from clientURL so it works with grpc.Dial()
func (m *member) grpcAddr() string {
u, err := url.Parse(m.ClientURL)
u, err := url.Parse(m.AdvertiseClientURL)
if err != nil {
panic(err)
}
@@ -171,7 +173,7 @@ func (m *member) grpcAddr() string {
}
func (m *member) peerPort() (port int) {
u, err := url.Parse(m.PeerURL)
u, err := url.Parse(m.AdvertisePeerURL)
if err != nil {
panic(err)
}

View File

@@ -0,0 +1,37 @@
#!/usr/bin/env bash
<<COMMENT
# to run agent
./scripts/agent-1.sh
# to run with failpoints
ETCD_EXEC_PATH=/etcd-failpoints ./scripts/agent-1.sh
COMMENT
if ! [[ "$0" =~ "scripts/agent-1.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
if [ -z "${ETCD_EXEC_PATH}" ]; then
ETCD_EXEC_PATH=/etcd
echo "Running agent without failpoints:" ${ETCD_EXEC_PATH}
elif [[ "${ETCD_EXEC_PATH}" == "/etcd-failpoints" ]]; then
echo "Running agent with failpoints:" ${ETCD_EXEC_PATH}
else
echo "Cannot find executable:" ${ETCD_EXEC_PATH}
exit 255
fi
rm -rf `pwd`/agent-1 && mkdir -p `pwd`/agent-1
docker run \
--rm \
--net=host \
--name agent-1 \
--mount type=bind,source=`pwd`/agent-1,destination=/agent-1 \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-agent \
--etcd-path ${ETCD_EXEC_PATH} \
--etcd-log-dir /agent-1 \
--port :19027 \
--failpoint-addr :7381"

View File

@@ -0,0 +1,37 @@
#!/usr/bin/env bash
<<COMMENT
# to run agent
./scripts/agent-2.sh
# to run with failpoints
ETCD_EXEC_PATH=/etcd-failpoints ./scripts/agent-2.sh
COMMENT
if ! [[ "$0" =~ "scripts/agent-2.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
if [ -z "${ETCD_EXEC_PATH}" ]; then
ETCD_EXEC_PATH=/etcd
echo "Running agent without failpoints:" ${ETCD_EXEC_PATH}
elif [[ "${ETCD_EXEC_PATH}" == "/etcd-failpoints" ]]; then
echo "Running agent with failpoints:" ${ETCD_EXEC_PATH}
else
echo "Cannot find executable:" ${ETCD_EXEC_PATH}
exit 255
fi
rm -rf `pwd`/agent-2 && mkdir -p `pwd`/agent-2
docker run \
--rm \
--net=host \
--name agent-2 \
--mount type=bind,source=`pwd`/agent-2,destination=/agent-2 \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-agent \
--etcd-path ${ETCD_EXEC_PATH} \
--etcd-log-dir /agent-2 \
--port :29027 \
--failpoint-addr :7382"

View File

@@ -0,0 +1,37 @@
#!/usr/bin/env bash
<<COMMENT
# to run agent
./scripts/agent-3.sh
# to run with failpoints
ETCD_EXEC_PATH=/etcd-failpoints ./scripts/agent-3.sh
COMMENT
if ! [[ "$0" =~ "scripts/agent-3.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
if [ -z "${ETCD_EXEC_PATH}" ]; then
ETCD_EXEC_PATH=/etcd
echo "Running agent without failpoints:" ${ETCD_EXEC_PATH}
elif [[ "${ETCD_EXEC_PATH}" == "/etcd-failpoints" ]]; then
echo "Running agent with failpoints:" ${ETCD_EXEC_PATH}
else
echo "Cannot find executable:" ${ETCD_EXEC_PATH}
exit 255
fi
rm -rf `pwd`/agent-3 && mkdir -p `pwd`/agent-3
docker run \
--rm \
--net=host \
--name agent-3 \
--mount type=bind,source=`pwd`/agent-3,destination=/agent-3 \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-agent \
--etcd-path ${ETCD_EXEC_PATH} \
--etcd-log-dir /agent-3 \
--port :39027 \
--failpoint-addr :7383"

View File

@@ -0,0 +1,24 @@
#!/usr/bin/env bash
if ! [[ "$0" =~ "scripts/tester-limit.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
# to run only 1 test round
docker run \
--rm \
--net=host \
--name tester \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-tester \
--agent-endpoints '127.0.0.1:19027,127.0.0.1:29027,127.0.0.1:39027' \
--client-ports 1379,2379,3379 \
--advertise-client-ports 13790,23790,33790 \
--peer-ports 1380,2380,3380 \
--advertise-peer-ports 13800,23800,33800 \
--limit 1 \
--stress-qps=2500 \
--stress-key-txn-count 100 \
--stress-key-txn-ops 10 \
--exit-on-failure"

View File

@@ -0,0 +1,25 @@
#!/usr/bin/env bash
if ! [[ "$0" =~ "scripts/tester-runner.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
# to run with etcd-runner
docker run \
--rm \
--net=host \
--name tester \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-tester \
--agent-endpoints '127.0.0.1:19027,127.0.0.1:29027,127.0.0.1:39027' \
--client-ports 1379,2379,3379 \
--advertise-client-ports 13790,23790,33790 \
--peer-ports 1380,2380,3380 \
--advertise-peer-ports 13800,23800,33800 \
--stress-qps=2500 \
--stress-key-txn-count 100 \
--stress-key-txn-ops 10 \
--etcd-runner /etcd-runner \
--stresser=keys,lease,election-runner,watch-runner,lock-racer-runner,lease-runner \
--exit-on-failure"

View File

@@ -0,0 +1,22 @@
#!/usr/bin/env bash
if ! [[ "$0" =~ "scripts/tester.sh" ]]; then
echo "must be run from tools/functional-tester"
exit 255
fi
docker run \
--rm \
--net=host \
--name tester \
gcr.io/etcd-development/etcd-functional-tester:go1.9.3 \
/bin/bash -c "/etcd-tester \
--agent-endpoints '127.0.0.1:19027,127.0.0.1:29027,127.0.0.1:39027' \
--client-ports 1379,2379,3379 \
--advertise-client-ports 13790,23790,33790 \
--peer-ports 1380,2380,3380 \
--advertise-peer-ports 13800,23800,33800 \
--stress-qps=2500 \
--stress-key-txn-count 100 \
--stress-key-txn-ops 10 \
--exit-on-failure"

View File

@@ -1,23 +0,0 @@
#!/bin/sh -e
set -x
set -e
# 1. build etcd binaries
[ -f bin/etcd ] || ./build
# 2. build agent & tester
[ -f bin/etcd-agent -a -f bin/etcd-tester ] || ./tools/functional-tester/build
# 3. build docker image (alpine based)
mkdir -p ./tools/functional-tester/docker/bin
cp -v bin/etcd-agent bin/etcd-tester bin/etcd ./tools/functional-tester/docker/bin
docker-compose -f tools/functional-tester/docker/docker-compose.yml build
# 4. create network (assumption - no overlaps)
docker network ls | grep etcd-functional || docker network create --subnet 172.20.0.0/16 etcd-functional
# 5. run cluster and tester (assumption - agents'll get first ip addresses)
docker-compose -f tools/functional-tester/docker/docker-compose.yml up -d a1 a2 a3
# 6. run tester
docker-compose -f tools/functional-tester/docker/docker-compose.yml run tester