Merge pull request #2918 from weaveworks/network-tester

Add a network test that isn't ping

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 > $@
 

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

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 -" <<EOF
+apiVersion: v1
+kind: Service
+metadata:
+  name: nettest
+spec:
+  clusterIP: None
+  ports:
+  - port: 80
+    targetPort: 8080
+    protocol: TCP
+  selector:
+    run: nettest
+EOF
+
+check_all_pods_communicate() {
+    podIP=$($SSH $HOST1 "$KUBECTL get pods -l run=nettest -o go-template='{{(index .items 0).status.podIP}}' 2>/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
 

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 <host> <container> <name> [<ip> ...]

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"]

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
+}