1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
# Creating custom toxics
Creating a toxic is done by implementing the `Toxic` interface:
```go
type Toxic interface {
Pipe(*toxics.ToxicStub)
}
```
The `Pipe()` function defines how data flows through the toxic, and is passed a
`ToxicStub` to operate on. A `ToxicStub` stores the input and output channels for
the toxic, as well as an interrupt channel that is used to pause operation of the
toxic.
The input and output channels in a `ToxicStub` send and receive `StreamChunk` structs,
which are similar to network packets. A `StreamChunk` contains a `byte[]` of stream
data, and a timestamp of when Toxiproxy received the data from the client or server.
This is used instead of just a plain `byte[]` so that toxics like latency can find out
how long a chunk of data has been waiting in the proxy.
Toxics are registered in an `init()` function so that they can be used by the server:
```go
func init() {
toxics.Register("toxic_name", new(ExampleToxic))
}
```
In order to use your own toxics, you will need to compile your own binary. This can be
done by copying [toxiproxy.go](https://github.com/Shopify/toxiproxy/blob/master/cmd/toxiproxy.go)
into a new project and registering your toxic with the server. This will allow you to add toxics
without having to make a full fork of the project. If you think your toxics will be useful
to others, contribute them back with a Pull Request.
An example project for building a separate binary can be found here:
https://github.com/xthexder/toxic-example
## A basic toxic
The most basic implementation of a toxic is the [noop toxic](https://github.com/Shopify/toxiproxy/blob/master/toxics/noop.go),
which just passes data through without any modifications.
```go
type NoopToxic struct{}
func (t *NoopToxic) Pipe(stub *toxics.ToxicStub) {
for {
select {
case <-stub.Interrupt:
return
case c := <-stub.Input:
if c == nil {
stub.Close()
return
}
stub.Output <- c
}
}
}
```
The above code reads from `stub.Input` in a loop, and passes the `StreamChunk` along to
`stub.Output`. Since reading from `stub.Input` will block until a chunk is available,
we need to check for interrupts as the same time.
Toxics will be interrupted whenever they are being updated, or possibly removed. This can
happen at any point within the `Pipe()` function, so all blocking operations (including sleep),
should be interruptible. When an interrupt is received, the toxic should return from the `Pipe()`
function after it has written any "in-flight" data back to `stub.Output`. It is important that
all data read from `stub.Input` is passed along to `stub.Output`, otherwise the stream will be
missing bytes and become corrupted.
When an `end of stream` is reached, `stub.Input` will return a `nil` chunk. Whenever a
nil chunk is returned, the toxic should call `Close()` on the stub, and return from `Pipe()`.
## Toxic configuration
Toxic configuration information can be stored in the toxic struct. The toxic will be json
encoded and decoded by the api, so all public fields will be api accessible.
An example of a toxic that uses configuration values is the [latency toxic](https://github.com/Shopify/toxiproxy/blob/master/toxics/latency.go)
```go
type LatencyToxic struct {
Latency int64 `json:"latency"`
Jitter int64 `json:"jitter"`
}
```
These fields can be used inside the `Pipe()` function, but generally should not be written
to from the toxic. A separate instance of the toxic exists for each connection through the
proxy, and may be replaced when updated by the api. If state is required in your toxic, it
is better to use a local variable at the top of `Pipe()`, since struct fields are not
guaranteed to be persisted across interrupts.
## Toxic buffering
By default, toxics are not buffered. This means that writes to `stub.Output` will block until
either the endpoint or another toxic reads it. Since toxics are chained together, this means
not reading from `stub.Input` will block other toxics (and endpoint writes) from operating.
If this is not behavior you want your toxic to have, you can specify a buffer size for your
toxic's input. The [latency toxic](https://github.com/Shopify/toxiproxy/blob/master/toxics/latency.go)
uses this in order to prevent added latency from limiting the proxy bandwidth.
Specifying a buffer size is done by implementing the `BufferedToxic` interface, which adds the
`GetBufferSize()` function:
```go
func (t *LatencyToxic) GetBufferSize() int {
return 1024
}
```
The unit used by `GetBufferSize()` is `StreamChunk`s. Chunks are generally anywhere from
1 byte, up to 32KB, so keep this in mind when thinking about how much buffering you need,
and how much memory you are comfortable with using.
## Using `io.Reader` and `io.Writer`
If your toxic involves modifying the data going through a proxy, you can use the `ChanReader`
and `ChanWriter` interfaces in the [stream package](https://github.com/Shopify/toxiproxy/tree/master/stream).
These allow reading and writing from the input and output channels as you would a normal data
stream such as a TCP socket.
An implementation of the noop toxic above using the stream package would look something like this:
```go
func (t *NoopToxic) Pipe(stub *toxics.ToxicStub) {
buf := make([]byte, 32*1024)
writer := stream.NewChanWriter(stub.Output)
reader := stream.NewChanReader(stub.Input)
reader.SetInterrupt(stub.Interrupt)
for {
n, err := reader.Read(buf)
if err == stream.ErrInterrupted {
writer.Write(buf[:n])
return
} else if err == io.EOF {
stub.Close()
return
}
writer.Write(buf[:n])
}
}
```
See https://github.com/xthexder/toxic-example/blob/master/http.go for a full example of using
the stream package with Go's http package.
|
