Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Our profiles from production show that a lot of CPU and memory in receivers is used for unmarshaling protobuf messages. Although it is not possible to change the remote-write format, we have the freedom to change the protocol used for replicating timeseries data. This commit introduces a new feature in receivers where replication can be done using Cap'n Proto instead of gRPC + Protobuf. The advantage of the former protocol is that deserialization is far cheaper and fields can be accessed directly from the received message (byte slice) without allocating intermediate objects. There is an additional cost for serialization because we have to convert from Protobuf to the Cap'n proto format, but in our setup this still results in a net reduction in resource usage. Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com>
- Loading branch information
1 parent
6ff5e1b
commit ee70d10
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| // Copyright (c) The Thanos Authors. | ||
| // Licensed under the Apache License 2.0. | ||
|
|
||
| package receive | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
|
|
||
| "capnproto.org/go/capnp/v3" | ||
| "capnproto.org/go/capnp/v3/rpc" | ||
| "github.com/go-kit/log" | ||
| "github.com/go-kit/log/level" | ||
| "github.com/pkg/errors" | ||
|
|
||
| "github.com/thanos-io/thanos/pkg/receive/writecapnp" | ||
| ) | ||
|
|
||
| type CapNProtoServer struct { | ||
| listener net.Listener | ||
| server writecapnp.Writer | ||
| } | ||
|
|
||
| func NewCapNProtoServer(listener net.Listener, handler *CapNProtoHandler) *CapNProtoServer { | ||
| return &CapNProtoServer{ | ||
| listener: listener, | ||
| server: writecapnp.Writer_ServerToClient(handler), | ||
| } | ||
| } | ||
|
|
||
| func (c *CapNProtoServer) ListenAndServe() error { | ||
| for { | ||
| conn, err := c.listener.Accept() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| go func() { | ||
| defer conn.Close() | ||
| rpcConn := rpc.NewConn(rpc.NewPackedStreamTransport(conn), &rpc.Options{ | ||
| // The BootstrapClient is the RPC interface that will be made available | ||
| // to the remote endpoint by default. | ||
| BootstrapClient: capnp.Client(c.server).AddRef(), | ||
| }) | ||
| <-rpcConn.Done() | ||
| }() | ||
| } | ||
| } | ||
|
|
||
| func (c *CapNProtoServer) Shutdown() { | ||
| c.server.Release() | ||
| } | ||
|
|
||
| type CapNProtoHandler struct { | ||
| writer *CapNProtoWriter | ||
| logger log.Logger | ||
| } | ||
|
|
||
| func NewCapNProtoHandler(logger log.Logger, writer *CapNProtoWriter) *CapNProtoHandler { | ||
| return &CapNProtoHandler{logger: logger, writer: writer} | ||
| } | ||
|
|
||
| func (c CapNProtoHandler) Write(ctx context.Context, call writecapnp.Writer_write) error { | ||
| call.Go() | ||
| wr, err := call.Args().Wr() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| t, err := wr.Tenant() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| req, err := writecapnp.NewRequest(wr) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer req.Close() | ||
|
|
||
| var errs writeErrors | ||
| errs.Add(c.writer.Write(ctx, t, req)) | ||
| if err := errs.ErrOrNil(); err != nil { | ||
| level.Debug(c.logger).Log("msg", "failed to handle request", "err", err) | ||
| result, allocErr := call.AllocResults() | ||
| if allocErr != nil { | ||
| return allocErr | ||
| } | ||
|
|
||
| switch errors.Cause(err) { | ||
| case nil: | ||
| return nil | ||
| case errNotReady: | ||
| result.SetError(writecapnp.WriteError_unavailable) | ||
| case errUnavailable: | ||
| result.SetError(writecapnp.WriteError_unavailable) | ||
| case errConflict: | ||
| result.SetError(writecapnp.WriteError_alreadyExists) | ||
| case errBadReplica: | ||
| result.SetError(writecapnp.WriteError_invalidArgument) | ||
| default: | ||
| result.SetError(writecapnp.WriteError_internal) | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| type BufferedListener struct { | ||
| ctx context.Context | ||
| cancel context.CancelFunc | ||
|
|
||
| conns chan net.Conn | ||
| } | ||
|
|
||
| func (b BufferedListener) Accept() (net.Conn, error) { | ||
| select { | ||
| case <-b.ctx.Done(): | ||
| return nil, b.ctx.Err() | ||
| case c := <-b.conns: | ||
| return c, nil | ||
| } | ||
| } | ||
|
|
||
| func (b BufferedListener) Close() error { | ||
| b.cancel() | ||
| return nil | ||
| } | ||
|
|
||
| func (b BufferedListener) Addr() net.Addr { | ||
| return addr{} | ||
| } | ||
|
|
||
| type addr struct{} | ||
|
|
||
| func (addr) Network() string { return "bufconn" } | ||
| func (addr) String() string { return "bufconn" } |
Oops, something went wrong.