diff --git a/Makefile b/Makefile index 052b1fef..7be13758 100644 --- a/Makefile +++ b/Makefile @@ -98,9 +98,10 @@ RUNNER_EXE=tools/runner/runner MANIFEST_TOOL_DIR=vendor/github.com/estesp/manifest-tool MANIFEST_TOOL_EXE=$(MANIFEST_TOOL_DIR)/manifest-tool TEST_TLS_EXE=test/tls/tls +NETWORKTESTER_EXE=test/images/network-tester/webserver # All binaries together in a list -EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(RUNNER_EXE) $(TEST_TLS_EXE) $(MANIFEST_TOOL_EXE) +EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE) $(WEAVEWAIT_EXE) $(WEAVEWAIT_NOOP_EXE) $(WEAVEWAIT_NOMCAST_EXE) $(WEAVEUTIL_EXE) $(RUNNER_EXE) $(TEST_TLS_EXE) $(MANIFEST_TOOL_EXE) $(NETWORKTESTER_EXE) # These stamp files are used to mark the current state of the build; whether an image has been built or not BUILD_UPTODATE=.build.uptodate @@ -110,8 +111,9 @@ PLUGIN_UPTODATE=.net-plugin$(ARCH_EXT).uptodate WEAVEKUBE_UPTODATE=.weavekube$(ARCH_EXT).uptodate WEAVENPC_UPTODATE=.weavenpc$(ARCH_EXT).uptodate WEAVEDB_UPTODATE=.weavedb.uptodate +NETWORKTESTER_UPTODATE=.networktester$(ARCH_EXT).uptodate -IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVENPC_UPTODATE) +IMAGES_UPTODATE=$(WEAVER_UPTODATE) $(WEAVEEXEC_UPTODATE) $(WEAVEKUBE_UPTODATE) $(WEAVENPC_UPTODATE) $(NETWORKTESTER_UPTODATE) # The names of the images. Note that the images for other architectures than amd64 have a suffix in the image name. WEAVER_IMAGE=$(DOCKERHUB_USER)/weave$(ARCH_EXT) @@ -121,8 +123,9 @@ WEAVENPC_IMAGE=$(DOCKERHUB_USER)/weave-npc$(ARCH_EXT) BUILD_IMAGE=weaveworks/weavebuild WEAVEDB_IMAGE=$(DOCKERHUB_USER)/weavedb PLUGIN_IMAGE=$(DOCKERHUB_USER)/net-plugin +NETWORKTESTER_IMAGE=$(DOCKERHUB_USER)/network-tester -IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVENPC_IMAGE) $(WEAVEDB_IMAGE) +IMAGES=$(WEAVER_IMAGE) $(WEAVEEXEC_IMAGE) $(WEAVEKUBE_IMAGE) $(WEAVENPC_IMAGE) $(WEAVEDB_IMAGE) $(NETWORKTESTER_IMAGE) PLUGIN_WORK_DIR="prog/net-plugin/rootfs" PLUGIN_BUILD_IMG="plugin-builder" @@ -167,6 +170,7 @@ $(MANIFEST_TOOL_EXE): $(MANIFEST_TOOL_DIR)/*.go $(WEAVEWAIT_NOOP_EXE): prog/weavewait/*.go $(WEAVEWAIT_EXE): prog/weavewait/*.go net/*.go $(WEAVEWAIT_NOMCAST_EXE): prog/weavewait/*.go net/*.go +$(NETWORKTESTER_EXE): test/images/network-tester/*.go tests: tools/.git lint: tools/.git @@ -199,7 +203,7 @@ else endif $(NETGO_CHECK) -$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE): +$(WEAVEUTIL_EXE) $(KUBEPEERS_EXE) $(WEAVENPC_EXE) $(NETWORKTESTER_EXE): go build $(BUILD_FLAGS) -o $@ ./$(@D) $(NETGO_CHECK) @@ -296,6 +300,10 @@ $(WEAVEDB_UPTODATE): prog/weavedb/Dockerfile $(SUDO) docker build -t $(WEAVEDB_IMAGE) prog/weavedb touch $@ +$(NETWORKTESTER_UPTODATE): test/images/network-tester/Dockerfile $(NETWORKTESTER_EXE) + $(SUDO) docker build -t $(NETWORKTESTER_IMAGE) test/images/network-tester + touch $@ + $(WEAVE_EXPORT): $(IMAGES_UPTODATE) $(WEAVEDB_UPTODATE) $(SUDO) DOCKER_HOST=$(DOCKER_HOST) docker save $(addsuffix :latest,$(IMAGES)) | gzip > $@ diff --git a/test/101_cross_hosts_2_test.sh b/test/101_cross_hosts_2_test.sh new file mode 100755 index 00000000..70039b2b --- /dev/null +++ b/test/101_cross_hosts_2_test.sh @@ -0,0 +1,28 @@ +#! /bin/bash + +. "$(dirname "$0")/config.sh" + +C1=10.32.1.4 +C2=10.32.1.7 +IMAGE=weaveworks/network-tester + +network_tester_status() { + # we only need to contact one container as it checks contact in both directions + status=$($SSH $HOST1 curl -sS http://127.0.0.1:8080/status) + [ -n "$status" -a "$status" != "running" ] +} + +start_suite "Network test over cross-host weave network" + +weave_on $HOST1 launch +weave_on $HOST2 launch $HOST1 + +docker_on $HOST1 run --name=c1 -dt -p 8080:8080 $IMAGE -peers=2 --iface=ethwe $C1 $C2 +weave_on $HOST1 attach $C1/24 c1 +docker_on $HOST2 run --name=c2 -dt -p 8080:8080 $IMAGE -peers=2 --iface=ethwe $C1 $C2 +weave_on $HOST2 attach $C2/24 c2 + +wait_for_x network_tester_status "network tester status" +assert "echo $status" "pass" + +end_suite diff --git a/test/840_weave_kube_3_test.sh b/test/840_weave_kube_3_test.sh index ac1a9c00..4e4fef80 100755 --- a/test/840_weave_kube_3_test.sh +++ b/test/840_weave_kube_3_test.sh @@ -18,6 +18,7 @@ NUM_HOSTS=$(howmany $HOSTS) SUCCESS="$(( $NUM_HOSTS * ($NUM_HOSTS-1) )) established" KUBECTL="sudo kubectl --kubeconfig /etc/kubernetes/admin.conf" KUBE_PORT=6443 +IMAGE=weaveworks/network-tester:latest tear_down_kubeadm @@ -44,19 +45,11 @@ sed -e "s%imagePullPolicy: Always%imagePullPolicy: Never$COVERAGE_ARGS%" "$(dirn sleep 5 -wait_for_connections() { - for i in $(seq 1 45); do - if run_on $HOST1 "curl -sS http://127.0.0.1:6784/status | grep \"$SUCCESS\"" ; then - return - fi - echo "Waiting for connections" - sleep 1 - done - echo "Timed out waiting for connections to establish" >&2 - exit 1 +check_connections() { + run_on $HOST1 "curl -sS http://127.0.0.1:6784/status | grep \"$SUCCESS\"" } -assert_raises wait_for_connections +assert_raises 'wait_for_x check_connections "connections to establish"' # Check we can ping between the Weave bridg IPs on each host HOST1EXPIP=$($SSH $HOST1 "weave expose" || true) @@ -71,22 +64,42 @@ assert "run_on $HOST1 ps aux | grep -c '[d]efunct'" "0" assert "run_on $HOST2 ps aux | grep -c '[d]efunct'" "0" assert "run_on $HOST3 ps aux | grep -c '[d]efunct'" "0" -# See if we can get some pods running that connect to the network -run_on $HOST1 "$KUBECTL run hello --image=weaveworks/hello-world --replicas=3" - -wait_for_pods() { - for i in $(seq 1 45); do - if run_on $HOST1 "$KUBECTL get pods | grep 'hello.*Running'" ; then - return - fi - echo "Waiting for pods" - sleep 1 - done - echo "Timed out waiting for pods" >&2 - exit 1 +check_ready() { + run_on $HOST1 "$KUBECTL get nodes | grep -c -w Ready | grep $NUM_HOSTS" } -assert_raises wait_for_pods +assert_raises 'wait_for_x check_ready "hosts to be ready"' + +# See if we can get some pods running that connect to the network +run_on $HOST1 "$KUBECTL run --image-pull-policy=Never nettest --image=$IMAGE --replicas=3 -- -peers=3 -dns-name=nettest.default.svc.cluster.local." +# Create a headless service so they can be found in Kubernetes DNS +run_on $HOST1 "$KUBECTL create -f -" </dev/null") + if [ -n podIP ] ; then + status=$($SSH $HOST1 "curl -s -S http://$podIP:8080/status") + if [ $status = "pass" ] ; then + return 0 + fi + fi + return 1 +} + +assert_raises 'wait_for_x check_all_pods_communicate pods' tear_down_kubeadm diff --git a/test/config.sh b/test/config.sh index d25c9152..2174934e 100644 --- a/test/config.sh +++ b/test/config.sh @@ -218,18 +218,28 @@ proxy_start_container_with_dns() { proxy docker_on $host run "$@" -dt $DNS_IMAGE /bin/sh } -wait_for_proxy() { - for i in $(seq 1 120); do - echo "Waiting for proxy to start" - if proxy docker_on $1 info > /dev/null 2>&1 ; then +# Execute command $1 repeatedly until it returns true; description in $2, optional repeat limit in $3 +wait_for_x() { + reps=${3:-45} + for i in $(seq 1 $reps); do + echo "Waiting for $2" + if $1 ; then return fi sleep 1 done - echo "Timed out waiting for proxy to start" >&2 + echo "Timed out waiting for $2" >&2 exit 1 } +check_proxy() { + proxy docker_on $1 info >/dev/null 2>&1 +} + +wait_for_proxy() { + wait_for_x check_proxy "proxy to start" 120 +} + rm_containers() { host=$1 shift @@ -244,18 +254,12 @@ container_pid() { docker_on $1 inspect -f '{{.State.Pid}}' $2 } +check_attached() { + exec_on $1 $2 $CHECK_ETHWE_UP >/dev/null 2>&1 +} + wait_for_attached() { - host=$1 - container=$2 - for i in $(seq 1 10); do - echo "Waiting for $container on $host to be attached" - if exec_on $host $container $CHECK_ETHWE_UP > /dev/null 2>&1 ; then - return - fi - sleep 1 - done - echo "Timed out waiting for $container on $host to be attached" >&2 - exit 1 + wait_for_x "check_attached $1 $2" "$2 on $1 to be attached" } # assert_dns_record [ ...] diff --git a/test/images/network-tester/Dockerfile b/test/images/network-tester/Dockerfile new file mode 100644 index 00000000..da21fd32 --- /dev/null +++ b/test/images/network-tester/Dockerfile @@ -0,0 +1,18 @@ +# Copyright 2016 The Kubernetes 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. + +FROM bash +ADD webserver webserver +EXPOSE 8080 +ENTRYPOINT ["/webserver"] diff --git a/test/images/network-tester/webserver.go b/test/images/network-tester/webserver.go new file mode 100644 index 00000000..e27fd0e6 --- /dev/null +++ b/test/images/network-tester/webserver.go @@ -0,0 +1,300 @@ +// Adapted from k8s.io/kubernetes/test/images/network-tester/webserver.go + +// A tiny web server for checking networking connectivity. +// +// Will dial out to, and expect to hear from, every other server +// Options to look for other servers: +// 1. via command-line args, following options +// +// Will serve a webserver on given -port. +// +// Visit /read to see the current state, or /quit to shut down. +// +// Visit /status to see pass/running/fail determination. (literally, it will +// return one of those words.) +// +// /write is used by other network test servers to register connectivity. +package main + +import ( + "bytes" + "encoding/json" + "flag" + "fmt" + "io/ioutil" + "log" + "net" + "net/http" + "os" + "os/signal" + "sync" + "syscall" + "time" +) + +var ( + port = flag.Int("port", 8080, "Port number to serve at.") + peerCount = flag.Int("peers", 8, "Must find at least this many peers for the test to pass.") + dnsName = flag.String("dns-name", "", "DNS name to find other network test pods via.") + delayShutdown = flag.Int("delay-shutdown", 0, "Number of seconds to delay shutdown when receiving SIGTERM.") + waitIface = flag.String("iface", "", "name of interface to wait for") +) + +// State tracks the internal state of our little http server. +// It's returned verbatim over the /read endpoint. +type State struct { + // Hostname is set once and never changed-- it's always safe to read. + Hostname string + + // The below fields require that lock is held before reading or writing. + Sent map[string]int + Received map[string]int + StillContactingPeers bool + + lock sync.Mutex +} + +func (s *State) doneContactingPeers() { + s.lock.Lock() + defer s.lock.Unlock() + s.StillContactingPeers = false +} + +// serveStatus returns "pass", "running", or "fail". +func (s *State) serveStatus(w http.ResponseWriter, r *http.Request) { + s.Logf("Checking status for %q with %d sent and %d received and %d peers", *dnsName, len(s.Sent), len(s.Received), *peerCount) + s.lock.Lock() + defer s.lock.Unlock() + if len(s.Sent) >= *peerCount && len(s.Received) >= *peerCount { + fmt.Fprintf(w, "pass") + return + } + if s.StillContactingPeers { + fmt.Fprintf(w, "running") + return + } + // Logf can't be called while holding the lock, so defer using a goroutine + go s.Logf("Declaring failure for %q with %d sent and %d received and %d peers", *dnsName, len(s.Sent), len(s.Received), *peerCount) + fmt.Fprintf(w, "fail") +} + +// serveRead writes our json encoded state +func (s *State) serveRead(w http.ResponseWriter, r *http.Request) { + s.lock.Lock() + defer s.lock.Unlock() + w.WriteHeader(http.StatusOK) + b, err := json.MarshalIndent(s, "", "\t") + s.appendErr(err) + _, err = w.Write(b) + s.appendErr(err) +} + +// WritePost is the format that (json encoded) requests to the /write handler should take. +type WritePost struct { + Source string + Dest string +} + +// WriteResp is returned by /write +type WriteResp struct { + Hostname string +} + +// serveWrite records the contact in our state. +func (s *State) serveWrite(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + s.lock.Lock() + defer s.lock.Unlock() + w.WriteHeader(http.StatusOK) + var wp WritePost + s.appendErr(json.NewDecoder(r.Body).Decode(&wp)) + if wp.Source == "" { + s.appendErr(fmt.Errorf("%v: Got request with no source", s.Hostname)) + } else { + if s.Received == nil { + s.Received = map[string]int{} + } + s.Received[wp.Source]++ + } + s.appendErr(json.NewEncoder(w).Encode(&WriteResp{Hostname: s.Hostname})) +} + +// appendErr logs an error, if err is not nil. +func (s *State) appendErr(err error) { + if err != nil { + log.Printf("error: %s", err) + } +} + +// Logf writes to the log. +func (s *State) Logf(format string, args ...interface{}) { + log.Printf(format, args...) +} + +// s must not be locked +func (s *State) appendSuccessfulSend(toHostname string) { + s.lock.Lock() + defer s.lock.Unlock() + if s.Sent == nil { + s.Sent = map[string]int{} + } + s.Sent[toHostname]++ +} + +var ( + // Our one and only state object + state State +) + +func main() { + flag.Parse() + + hostname, err := os.Hostname() + if err != nil { + log.Fatalf("Error getting hostname: %v", err) + } + + if *waitIface != "" { + _, err = EnsureInterface(*waitIface, 30) + if err != nil { + log.Fatalf("Error getting interface: %v", err) + } + } + + if *delayShutdown > 0 { + termCh := make(chan os.Signal) + signal.Notify(termCh, syscall.SIGTERM) + go func() { + <-termCh + log.Printf("Sleeping %d seconds before exit ...", *delayShutdown) + time.Sleep(time.Duration(*delayShutdown) * time.Second) + os.Exit(0) + }() + } + + state := State{ + Hostname: hostname, + StillContactingPeers: true, + } + + go contactOthers(&state) + + http.HandleFunc("/quit", func(w http.ResponseWriter, r *http.Request) { + os.Exit(0) + }) + + http.HandleFunc("/read", state.serveRead) + http.HandleFunc("/write", state.serveWrite) + http.HandleFunc("/status", state.serveStatus) + + go log.Fatal(http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", *port), nil)) + + select {} +} + +// Find all siblings and post to their /write handler. +func contactOthers(state *State) { + sleepTime := 1 * time.Second + // In large cluster getting all endpoints is pretty expensive. + // Thus, we will limit ourselves to send on average at most 10 such + // requests per second + if sleepTime < time.Duration(*peerCount/10)*time.Second { + sleepTime = time.Duration(*peerCount/10) * time.Second + } + timeout := 5 * time.Minute + // Similarly we need to bump timeout so that it is reasonable in large + // clusters. + if timeout < time.Duration(*peerCount)*time.Second { + timeout = time.Duration(*peerCount) * time.Second + } + defer state.doneContactingPeers() + + for start := time.Now(); time.Since(start) < timeout; time.Sleep(sleepTime) { + eps := getWebserverEndpoints() + if len(eps) >= *peerCount { + break + } + state.Logf("%q has %v endpoints (%v), which is less than %v as expected. Waiting for all endpoints to come up.", *dnsName, len(eps), eps, *peerCount) + } + + // Do this repeatedly, in case there's some propagation delay with getting + // newly started siblings into the endpoints list. + for i := 0; i < 15; i++ { + eps := getWebserverEndpoints() + for ep := range eps { + state.Logf("Attempting to contact %s", ep) + contactSingle(ep, state) + } + time.Sleep(sleepTime) + } +} + +//getWebserverEndpoints returns the webserver endpoints as a set of String, each in the format like "http://{ip}:{port}" +func getWebserverEndpoints() map[string]struct{} { + eps := map[string]struct{}{} + if *dnsName != "" { + addrs, err := net.LookupHost(*dnsName) + if err != nil { + state.Logf("Error from DNS lookup of %q: %s", *dnsName, err) + return nil + } + for _, a := range addrs { + eps[fmt.Sprintf("http://%s:%d", a, *port)] = struct{}{} + } + } + for _, a := range flag.Args() { + eps[fmt.Sprintf("http://%s:%d", a, *port)] = struct{}{} + } + return eps +} + +// contactSingle dials the address 'e' and tries to POST to its /write address. +func contactSingle(e string, state *State) { + body, err := json.Marshal(&WritePost{ + Dest: e, + Source: state.Hostname, + }) + if err != nil { + log.Fatalf("json marshal error: %v", err) + } + resp, err := http.Post(e+"/write", "application/json", bytes.NewReader(body)) + if err != nil { + state.Logf("Warning: unable to contact the endpoint %q: %v", e, err) + return + } + defer resp.Body.Close() + + body, err = ioutil.ReadAll(resp.Body) + if err != nil { + state.Logf("Warning: unable to read response from '%v': '%v'", e, err) + return + } + var wr WriteResp + err = json.Unmarshal(body, &wr) + if err != nil { + state.Logf("Warning: unable to unmarshal response (%v) from '%v': '%v'", string(body), e, err) + return + } + state.appendSuccessfulSend(wr.Hostname) +} + +func EnsureInterface(ifaceName string, wait int) (iface *net.Interface, err error) { + if iface, err = findInterface(ifaceName); err == nil || wait == 0 { + return + } + for ; err != nil && wait > 0; wait-- { + time.Sleep(1 * time.Second) + iface, err = findInterface(ifaceName) + } + return +} + +func findInterface(ifaceName string) (iface *net.Interface, err error) { + if iface, err = net.InterfaceByName(ifaceName); err != nil { + return iface, fmt.Errorf("Unable to find interface %s", ifaceName) + } + if 0 == (net.FlagUp & iface.Flags) { + return iface, fmt.Errorf("Interface %s is not up", ifaceName) + } + return +}