# 启动server
» ./toxiproxy-server

# 新建proxy
» ./toxiproxy-cli create -l localhost:26379 -u localhost:6379 redis                                                             130 ↵
Created new proxy redis

# 注入延迟异常（toxic）
» ./toxiproxy-cli toxic add -t latency -a latency=1000 redis
Added downstream latency toxic 'latency_downstream' on proxy 'redis'

# 客户端测试：从耗时上看已经生效了
» redis-cli -p 26379
127.0.0.1:26379> ping
PONG
(1.00s)

type ApiServer struct {
	Collection *ProxyCollection // proxy collection
	Metrics    *metricsContainer // 指标收集
	Logger     *zerolog.Logger
	http       *http.Server // http API
}

// 启动所有proxy。并且移除所有toxics
r.HandleFunc("/reset", server.ResetState).Methods("POST").Name("ResetState")
// list 所有proxy
r.HandleFunc("/proxies", server.ProxyIndex).Methods("GET").Name("ProxyIndex")
// 创建proxy
r.HandleFunc("/proxies", server.ProxyCreate).Methods("POST").Name("ProxyCreate")
// 批量添加或者更新（比如名称、状态）proxy
r.HandleFunc("/populate", server.Populate).Methods("POST").Name("Populate")
// 查看特定proxy
r.HandleFunc("/proxies/{proxy}", server.ProxyShow).Methods("GET").Name("ProxyShow")
// 更新proxy
r.HandleFunc("/proxies/{proxy}", server.ProxyUpdate).Methods("POST", "PATCH").Name("ProxyUpdate")
// 删除proxy
r.HandleFunc("/proxies/{proxy}", server.ProxyDelete).Methods("DELETE").Name("ProxyDelete")
// List active toxics
r.HandleFunc("/proxies/{proxy}/toxics", server.ToxicIndex).Methods("GET").Name("ToxicIndex")
// 创建toxic
r.HandleFunc("/proxies/{proxy}/toxics", server.ToxicCreate).Methods("POST").Name("ToxicCreate")
// 查看、更新、删除proxy下的toxic
r.HandleFunc("/proxies/{proxy}/toxics/{toxic}", server.ToxicShow).Methods("GET").Name("ToxicShow")
r.HandleFunc("/proxies/{proxy}/toxics/{toxic}", server.ToxicUpdate).Methods("POST", "PATCH").Name("ToxicUpdate")
r.HandleFunc("/proxies/{proxy}/toxics/{toxic}", server.ToxicDelete).Methods("DELETE").Name("ToxicDelete")

type ProxyCollection struct {
	sync.RWMutex
	proxies map[string]*Proxy
}

[
  {
    "name": "redis",
    "listen": "[::]:16379",
    "upstream": "127.0.0.1:6379",
    "enabled": true
  },
  {
    "name": "mysql",
    "listen": "[::]:14355",
    "upstream": "127.0.0.1:4355",
    "enabled": true
  }
]

type Proxy struct {
	sync.Mutex

	Name     string `json:"name"`
	Listen   string `json:"listen"`
	Upstream string `json:"upstream"`
	Enabled  bool   `json:"enabled"`

	listener net.Listener // 本地tcp监听listener
	started  chan error

	tomb        tomb.Tomb // 用于优雅退出时关闭proxy下的connection list
	connections ConnectionList   // 客户端连接+upstream连接列表。每一个客户端链接存在一个对应的upstream链接
	Toxics      *ToxicCollection `json:"-"` // 指向ToxicCollection对象
	apiServer   *ApiServer       /// 指向server对象
	Logger      *zerolog.Logger
}
// 客户端&upstream连接
type ConnectionList struct {
	list map[string]net.Conn
	lock sync.Mutex
}

var (
	// 存储所有注册的toxic指针
	ToxicRegistry map[string]Toxic
	registryMutex sync.RWMutex
)

// in toxics/latency.go
func init() {
	Register("latency", new(LatencyToxic))
}

// The NoopToxic passes all data through without any toxic effects.
type NoopToxic struct{}

func (t *NoopToxic) Pipe(stub *ToxicStub) {
	for {
		select {
		case <-stub.Interrupt:
			return
		case c := <-stub.Input:
			if c == nil {
				stub.Close()
				return
			}
			stub.Output <- c
		}
	}
}

func (t *LatencyToxic) Pipe(stub *ToxicStub) {
	for {
		select {
		case <-stub.Interrupt:
			return
		case c := <-stub.Input:
			if c == nil {
				stub.Close()
				return
			}
			sleep := t.delay() - time.Since(c.Timestamp)
			select {
			case <-time.After(sleep):
				c.Timestamp = c.Timestamp.Add(sleep)
				stub.Output <- c
			case <-stub.Interrupt:
				// Exit fast without applying latency.
				stub.Output <- c // Don't drop any data on the floor
				return
			}
		}
	}
}

type ToxicCollection struct {
	noop  *toxics.ToxicWrapper     // 初始化创建noop
	proxy *Proxy                   // 指向归属proxy
	chain [][]*toxics.ToxicWrapper // upstream + downstream toxics的数组。第0项为noop，每增加一个toxic，都会在数组中append一个新的toxic
	links map[string]*ToxicLink    // 每条连接的建立会有两条link。分别作用downstream、upstream的数据流处理。
}

type ToxicLink struct {
	stubs     []*toxics.ToxicStub
	proxy     *Proxy
	toxics    *ToxicCollection
	input     *stream.ChanWriter
	output    *stream.ChanReader
	direction stream.Direction
	Logger    *zerolog.Logger
}

// proxy监听协程，处理客户端的proxy连接【去掉部分非核心代码】
func (proxy *Proxy) server() {
	err := proxy.listen()

	for {
		// 会阻塞，等待客户端连接。后者listener被关闭
		client, err := proxy.listener.Accept()
		
		// 每accept新的链接，创建一个到upstream的对应的链接
		upstream, err := net.Dial("tcp", proxy.Upstream)

		name := client.RemoteAddr().String()
		proxy.connections.list[name+"upstream"] = upstream
		proxy.connections.list[name+"downstream"] = client
		
		// 启动UP和DOWN链接的toxics link创建以及调度
		proxy.Toxics.StartLink(proxy.apiServer, name+"upstream", client, upstream, stream.Upstream)
		proxy.Toxics.StartLink(proxy.apiServer, name+"downstream", upstream, client, stream.Downstream)
	}

func (link *ToxicLink) read(
	metricLabels []string,
	server *ApiServer,
	source io.Reader,
) {
	// 会一直阻塞，直到source结束。过程中没有metrics数据？
	bytes, err := io.Copy(link.input, source)

	if server.Metrics.proxyMetricsEnabled() {
		server.Metrics.ProxyMetrics.ReceivedBytesTotal.
			WithLabelValues(metricLabels...).Add(float64(bytes))
	}
	fmt.Println("read from source end", bytes)
	link.input.Close()
}

for i, toxic := range link.toxics.chain[link.direction] {
	// 链接+toxic 相关的状态数据（通常toxic时proxy下的所有链接共享的，或者不需要私有数据）
	if stateful, ok := toxic.Toxic.(toxics.StatefulToxic); ok {
		link.stubs[i].State = stateful.NewState()
	}

	go link.stubs[i].Run(toxic)
}

// 调度toxic。每个toxic都有一个goroutine。根据Toxicity大小，toxic会随机执行。
func (s *ToxicStub) Run(toxic *ToxicWrapper) {
	s.running = make(chan struct{})
	defer close(s.running)
	randomToxicity := rand.Float32() // #nosec G404 -- was ignored before too
	if randomToxicity < toxic.Toxicity {
		toxic.Pipe(s)
	} else {
		new(NoopToxic).Pipe(s)
	}
}