Merge pull request #1 from Shopify/initial

Initial

.gitignore

@@ -0,0 +1,5 @@
+toxiproxy
+toxiproxy.test
+cpu.out
+cover*.out
+coverage.html

README.md

@@ -1,2 +1,42 @@
-toxiproxy
-=========
+# Toxiproxy
+
+Toxiproxy is a framework for simulating network conditions. It's made to work in
+testing/CI environments as well as development. It consists of two parts: a Go
+proxy that all network connections go through, as well as a client library that
+can apply a condition to the link. The client library controls Toxiproxy through
+an HTTP interface:
+
+```bash
+$ curl -i -d '{"Name": "redis", "Upstream": "localhost:6379"}'
+localhost:8474/proxies
+HTTP/1.1 201 Created
+Content-Type: application/json
+Date: Sun, 07 Sep 2014 23:38:53 GMT
+Content-Length: 71
+
+{"Name":"redis","Listen":"localhost:40736","Upstream":"localhost:6379"}
+
+$ redis-cli -p 40736
+127.0.0.1:53646> SET omg pandas
+OK
+127.0.0.1:53646> GET omg
+"pandas"
+
+$ curl -i localhost:8474/proxies
+HTTP/1.1 200 OK
+Content-Type: application/json
+Date: Sun, 07 Sep 2014 23:39:16 GMT
+Content-Length: 81
+
+{"redis":{"Name":"redis","Listen":"localhost:40736","Upstream":"localhost:6379"}}
+
+$ curl -i -X DELETE localhost:8474/proxies/redis
+HTTP/1.1 204 No Content
+Date: Sun, 07 Sep 2014 23:40:00 GMT
+
+$ telnet localhost 53646
+Trying 127.0.0.1...
+telnet: Unable to connect to remote host: Connection refused
+
+```
+>>>>>>> more

api.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/gorilla/mux"
+)
+
+type server struct {
+	collection *ProxyCollection
+}
+
+func NewServer() *server {
+	return &server{
+		collection: NewProxyCollection(),
+	}
+}
+
+func (server *server) Listen() {
+	r := mux.NewRouter()
+	r.HandleFunc("/proxies", server.ProxyIndex).Methods("GET")
+	r.HandleFunc("/proxies", server.ProxyCreate).Methods("POST")
+	r.HandleFunc("/proxies/{name}", server.ProxyDelete).Methods("DELETE")
+	http.Handle("/", r)
+
+	err := http.ListenAndServe(":8474", nil)
+	if err != nil {
+		log.Fatal("ListenAndServe: ", err)
+	}
+}
+
+func (server *server) ProxyIndex(response http.ResponseWriter, request *http.Request) {
+	data, err := json.Marshal(server.collection.Proxies())
+	if err != nil {
+		http.Error(response, fmt.Sprint(err), http.StatusInternalServerError)
+		return
+	}
+
+	response.Header().Set("Content-Type", "application/json")
+	response.WriteHeader(http.StatusOK)
+	_, err = response.Write(data)
+	if err != nil {
+		logrus.Warn("ProxyIndex: Failed to write response to client", err)
+	}
+}
+
+func (server *server) ProxyCreate(response http.ResponseWriter, request *http.Request) {
+	proxy := NewProxy()
+	err := json.NewDecoder(request.Body).Decode(&proxy)
+	if err != nil {
+		http.Error(response, server.apiError(err, http.StatusBadRequest), http.StatusBadRequest)
+		return
+	}
+
+	err = server.collection.Add(proxy)
+	if err != nil {
+		http.Error(response, server.apiError(err, http.StatusConflict), http.StatusConflict)
+		return
+	}
+
+	proxy.Start()
+	<-proxy.started
+
+	data, err := json.Marshal(&proxy)
+	if err != nil {
+		http.Error(response, server.apiError(err, http.StatusInternalServerError), http.StatusInternalServerError)
+		return
+	}
+
+	response.Header().Set("Content-Type", "application/json")
+	response.WriteHeader(http.StatusCreated)
+	_, err = response.Write(data)
+	if err != nil {
+		logrus.Warn("ProxyIndex: Failed to write response to client", err)
+	}
+}
+
+func (server *server) ProxyDelete(response http.ResponseWriter, request *http.Request) {
+	vars := mux.Vars(request)
+
+	err := server.collection.Remove(vars["name"])
+	if err != nil {
+		http.Error(response, server.apiError(err, http.StatusNotFound), http.StatusNotFound)
+		return
+	}
+
+	response.WriteHeader(http.StatusNoContent)
+	_, err = response.Write(nil)
+	if err != nil {
+		logrus.Warn("ProxyIndex: Failed to write headers to client", err)
+	}
+}
+
+func (server *server) apiError(err error, code int) string {
+	return fmt.Sprintf(`
+{
+	"title": "%s",
+	"status": %d
+}
+	`, err.Error(), code)
+}

api_test.go

@@ -0,0 +1,147 @@
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+)
+
+var testServer *server
+
+func WithServer(t *testing.T, f func(string)) {
+	// Make sure only one server is running at a time. Apparently there's no clean
+	// way to shut it down between each test run.
+	if testServer == nil {
+		testServer = NewServer()
+		go testServer.Listen()
+
+		// Allow server to start. There's no clean way to know when it listens.
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	f("http://localhost:8474")
+
+	err := testServer.collection.Clear()
+	if err != nil {
+		t.Error("Failed to clear collection", err)
+	}
+}
+
+func CreateProxy(t *testing.T, addr string) *http.Response {
+	body := `
+	{
+		"Name": "mysql_master",
+		"Listen": "localhost:3310",
+		"Upstream": "localhost:20001"
+	}`
+
+	resp, err := http.Post(addr+"/proxies", "application/json", strings.NewReader(body))
+	if err != nil {
+		t.Fatal("Failed to get index", err)
+	}
+
+	return resp
+}
+
+func ListProxies(t *testing.T, addr string) map[string]Proxy {
+	resp, err := http.Get(addr + "/proxies")
+	if err != nil {
+		t.Fatal("Failed to get index", err)
+	}
+
+	proxies := make(map[string]Proxy)
+	err = json.NewDecoder(resp.Body).Decode(&proxies)
+	if err != nil {
+		t.Fatal("Failed to parse JSON response from index")
+	}
+
+	return proxies
+}
+
+func DeleteProxy(t *testing.T, addr string, name string) *http.Response {
+	client := &http.Client{}
+	req, err := http.NewRequest("DELETE", addr+"/proxies/"+name, nil)
+	if err != nil {
+		t.Fatal("Failed to create request", err)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal("Failed to issue request", err)
+	}
+
+	return resp
+}
+
+func TestIndexWithNoProxies(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if len(ListProxies(t, addr)) > 0 {
+			t.Error("Expected no proxies in list")
+		}
+	})
+}
+
+func TestCreateProxy(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if resp := CreateProxy(t, addr); resp.StatusCode != http.StatusCreated {
+			t.Fatal("Unable to create proxy")
+		}
+	})
+}
+
+func TestIndexWithProxies(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if resp := CreateProxy(t, addr); resp.StatusCode != http.StatusCreated {
+			t.Fatal("Unable to create proxy")
+		}
+
+		proxies := ListProxies(t, addr)
+		if len(proxies) == 0 {
+			t.Error("Expected new proxy in list")
+		}
+	})
+}
+
+func TestDeleteProxy(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if resp := CreateProxy(t, addr); resp.StatusCode != http.StatusCreated {
+			t.Fatal("Unable to create proxy")
+		}
+
+		proxies := ListProxies(t, addr)
+		if len(proxies) == 0 {
+			t.Fatal("Expected new proxy in list")
+		}
+
+		if resp := DeleteProxy(t, addr, "mysql_master"); resp.StatusCode != http.StatusNoContent {
+			t.Fatal("Unable to delete proxy")
+		}
+
+		proxies = ListProxies(t, addr)
+		if len(proxies) > 0 {
+			t.Error("Expected proxy to be deleted from list")
+		}
+	})
+}
+
+func TestCreateProxyTwice(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if resp := CreateProxy(t, addr); resp.StatusCode != http.StatusCreated {
+			t.Fatal("Unable to create proxy")
+		}
+
+		if resp := CreateProxy(t, addr); resp.StatusCode != http.StatusConflict {
+			t.Fatal("Expected http.StatusConflict Conflict back from API")
+		}
+	})
+}
+
+func TestDeleteNonExistantProxy(t *testing.T) {
+	WithServer(t, func(addr string) {
+		if resp := DeleteProxy(t, addr, "non_existant"); resp.StatusCode != http.StatusNotFound {
+			t.Fatal("Expected http.StatusNotFound Not found when deleting non existant proxy")
+		}
+	})
+}

link.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"io"
+	"net"
+	"sync"
+
+	"github.com/Sirupsen/logrus"
+)
+
+// Link is the TCP link between a client and an upstream.
+//
+// Client <-> toxiproxy <-> Upstream
+//
+// Its responsibility is to shove from each side to the other. Clients don't
+// need to know they are talking to the upsream through toxiproxy.
+type link struct {
+	sync.Mutex
+	proxy *Proxy
+
+	client   net.Conn
+	upstream net.Conn
+}
+
+func NewLink(proxy *Proxy, client net.Conn) *link {
+	return &link{
+		proxy:    proxy,
+		client:   client,
+		upstream: &net.TCPConn{},
+	}
+}
+
+func (link *link) Open() (err error) {
+	link.Lock()
+	defer link.Unlock()
+
+	link.upstream, err = net.Dial("tcp", link.proxy.Upstream)
+	if err != nil {
+		return err
+	}
+
+	go link.pipe(link.client, link.upstream)
+	go link.pipe(link.upstream, link.client)
+
+	return nil
+}
+
+func (link *link) pipe(src, dst net.Conn) {
+	bytes, err := io.Copy(dst, src)
+	if err != nil {
+		logrus.WithFields(logrus.Fields{
+			"name":     link.proxy.Name,
+			"upstream": link.proxy.Upstream,
+			"bytes":    bytes,
+			"err":      err,
+		}).Warn("Client or source terminated")
+	}
+
+	link.Close()
+}
+
+func (link *link) Close() {
+	link.Lock()
+	defer link.Unlock()
+
+	link.client.Close()
+	link.upstream.Close()
+}

omg.rb

@@ -0,0 +1,11 @@
+require 'socket'
+
+server = TCPServer.new(8000)
+
+loop do
+  puts "Waiting for client.."
+  server.accept
+  puts "Got client.."
+
+  p server.read
+end