To implement a reverse proxy for caching data, the first idea is to start from response The data is read from the body, and then returned to the requesting client and the local Cache module respectively. The pseudo code is as follows:
func HandleResponse(rw http.ResponseWriter, resp *http.Response) {
bodyBytes, _ := ioutil.ReadAll(resp.Body)
go func() {
// cache response on local disk
cacher.Write(bodyBytes)
}
// client reads data from response
rw.Write(bodyBytes)
}
After deep thinking, in Kubernetes system, the above implementation will lead to the following problems:
-
Question 1: how to process streaming data (such as the watch request in K8s), which means ioutil Readall() cannot return all data in one call. That is, how to return stream data and cache stream data at the same time.
-
Problem 2: at the same time, it may be necessary to clean the incoming byte slice data before caching the data locally. This means that you need to modify byte slice, or back up byte slice before processing. This will cause a lot of memory consumption. At the same time, it is difficult to deal with the size of slice applied for streaming data.
3. Discussion on elegant implementation
For the above problems, we abstract them one by one to find more elegant implementation methods.
- Question 1: how to read and write streaming data at the same time
For reading and writing streaming data (caching while returning), as shown in the figure below, all you need is to put the response Convert body (io.Reader) into an IO Reader and an IO Writer. Or an IO Reader and Io Writer composes an IO Reader. It's easy to think of the Tee command in Linux.
In Golang, the Tee command is implemented as io Teereader, the pseudo code of question 1 is as follows:
func HandleResponse(rw http.ResponseWriter, resp *http.Response) {
// create TeeReader with response.Body and cacher
newRespBody := io.TeeReader(resp.Body, cacher)
// client reads data from response
io.Copy(rw, newRespBody)
}
Response through TeeReader The integration of the body and the Cache, when the client side requests from the response When reading data from the body, the return data will be written to the Cache at the same time, which gracefully solves the processing of streaming data.
- Question 2: how to clean the stream data before caching
As shown in the figure below, the stream data is cleaned before caching, and the requester and filter need to read the response at the same time Body (problem of reading twice). That is, you need to convert response.Body(io.Reader) into two IO Reader.
It also means that problem 2 is transformed into: io on the cache side in problem 1 Convert writer to io of Data Filter Reader. In fact, similar commands can be found in Linux commands, namely pipes. Therefore, the pseudo code of question 2 is as follows:
func HandleResponse(rw http.ResponseWriter, resp *http.Response) {
pr, pw := io.Pipe()
// create TeeReader with response.Body and Pipe writer
newRespBody := io.TeeReader(resp.Body, pw)
go func() {
// filter reads data from response
io.Copy(dataFilter, pr)
}
// client reads data from response
io.Copy(rw, newRespBody)
}
Via io Teereader and Io Pipe, when the client side requests from the Response When reading data from the body, the Filter will read data from the Response at the same time, which gracefully solves the problem of twice reading streaming data.
===============================================================================
Finally, let's take a look at the related implementation in YurtHub, because response The body is Io Readcloser, so dualReadCloser is implemented. At the same time, YurtHub may also face the challenge of HTTP Request, so the isRespBody parameter is added to determine whether to be responsible for closing the response Body.
// https://github.com/openyurtio/openyurt/blob/master/pkg/yurthub/util/util.go#L156
// NewDualReadCloser create an dualReadCloser object
func NewDualReadCloser(rc io.ReadCloser, isRespBody bool) (io.ReadCloser, io.ReadCloser) {
pr, pw := io.Pipe()
dr := &dualReadCloser{
rc: rc,
pw: pw,
isRespBody: isRespBody,
}
return dr, pr
}
type dualReadCloser struct {
rc io.ReadCloser
pw *io.PipeWriter
// isRespBody shows rc(is.ReadCloser) is a response.Body
// or not(maybe a request.Body). if it is true(it's a response.Body),
// we should close the response body in Close func, else not,
// it(request body) will be closed by http request caller
isRespBody bool
}
// Read read data into p and write into pipe
func (dr *dualReadCloser) Read(p []byte) (n int, err error) {
n, err = dr.rc.Read(p)
if n > 0 {
if n, err := dr.pw.Write(p[:n]); err != nil {
klog.Errorf("dualReader: failed to write %v", err)
return n, err
}
}
return
}
// Close close two readers
func (dr *dualReadCloser) Close() error {
errs := make([]error, 0)
if dr.isRespBody {
if err := dr.rc.Close(); err != nil {
errs = append(errs, err)
}
}
if err := dr.pw.Close(); err != nil {
errs = append(errs, err)
# Write at the end
**[CodeChina Open source project: [first tier big factory] Java Analysis of interview questions+Core summary learning notes+Latest explanation Video]](https://codechina.csdn.net/m0_60958482/java-p7)**
There's another one JAVA Sorting of core knowledge points( PDF): **JVM，JAVA Gather, JAVA Multithreading concurrency, JAVA Basics, Spring Principles, microservices, Netty And RPC**，Network, log, Zookeeper，Kafka，RabbitMQ，Hbase，**MongoDB，Cassandra，Design pattern, load balancing, database, consistency hash, JAVA Algorithm, data structure, encryption algorithm, distributed cache**，Hadoop，Spark，Storm，YARN，Machine learning, cloud computing...
tps://codechina.csdn.net/m0_60958482/java-p7)**
There's another one JAVA Sorting of core knowledge points( PDF): **JVM，JAVA Gather, JAVA Multithreading concurrency, JAVA Basics, Spring Principles, microservices, Netty And RPC**，Network, log, Zookeeper，Kafka，RabbitMQ，Hbase，**MongoDB，Cassandra，Design pattern, load balancing, database, consistency hash, JAVA Algorithm, data structure, encryption algorithm, distributed cache**，Hadoop，Spark，Storm，YARN，Machine learning, cloud computing...
