Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Christian Schwarz
|
0eb7032735 | ||
|
Christian Schwarz
|
3ff1966cab | ||
|
Christian Schwarz
|
a3842155c5 | ||
|
Christian Schwarz
|
02b3b4f80c | ||
|
Christian Schwarz
|
0882290595 |
@@ -1,8 +1,9 @@
|
||||
[](https://github.com/zrepl/zrepl/blob/master/LICENSE)
|
||||
[](https://golang.org/)
|
||||
[](https://zrepl.github.io)
|
||||
[](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
|
||||
[](https://www.patreon.com/zrepl)
|
||||
[](https://liberapay.com/zrepl/donate)
|
||||
[](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
|
||||
[](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)
|
||||
|
||||
# zrepl
|
||||
@@ -23,6 +24,7 @@ zrepl is a one-stop ZFS backup & replication solution.
|
||||
If so, think of an expressive configuration example.
|
||||
2. Think of at least one use case that generalizes from your concrete application.
|
||||
3. Open an issue on GitHub with example conf & use case attached.
|
||||
4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.
|
||||
|
||||
The above does not apply if you already implemented everything.
|
||||
Check out the *Coding Workflow* section below for details.
|
||||
@@ -46,11 +48,8 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
|
||||
The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
|
||||
|
||||
To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
|
||||
|
||||
### Overall Architecture
|
||||
|
||||
The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
|
||||
Make sure to develop an understanding how zrepl is typically used by studying the user docs first.
|
||||
`lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
|
||||
If you just want to install the Go dependencies, run `./lazy.sh godep`.
|
||||
|
||||
### Project Structure
|
||||
|
||||
@@ -62,14 +61,15 @@ Make sure to develop an understanding how zrepl is typically used by studying th
|
||||
│ └── samples
|
||||
├── daemon # the implementation of `zrepl daemon` subcommand
|
||||
│ ├── filters
|
||||
│ ├── hooks # snapshot hooks
|
||||
│ ├── job # job implementations
|
||||
│ ├── logging # logging outlets + formatters
|
||||
│ ├── nethelpers
|
||||
│ ├── prometheus
|
||||
│ ├── pruner # pruner implementation
|
||||
│ ├── snapper # snapshotter implementation
|
||||
├── docs # sphinx-based documentation
|
||||
├── dist # supplemental material for users & package maintainers
|
||||
├── docs # sphinx-based documentation
|
||||
│ ├── **/*.rst # documentation in reStructuredText
|
||||
│ ├── sphinxconf
|
||||
│ │ └── conf.py # sphinx config (see commit 445a280 why its not in docs/)
|
||||
@@ -78,6 +78,7 @@ Make sure to develop an understanding how zrepl is typically used by studying th
|
||||
│ └── public_git # checkout of zrepl.github.io managed by above shell script
|
||||
├── endpoint # implementation of replication endpoints (=> package replication)
|
||||
├── logger # our own logger package
|
||||
├── platformtest # test suite for our zfs abstractions (error classification, etc)
|
||||
├── pruning # pruning rules (the logic, not the actual execution)
|
||||
│ └── retentiongrid
|
||||
├── replication
|
||||
@@ -92,14 +93,13 @@ Make sure to develop an understanding how zrepl is typically used by studying th
|
||||
│ ├── transportmux # TCP connecter and listener used to split control & data traffic
|
||||
│ └── versionhandshake # replication protocol version handshake perfomed on newly established connections
|
||||
├── tlsconf # abstraction for Go TLS server + client config
|
||||
├── transport # transports implementation
|
||||
├── transport # transport implementations
|
||||
│ ├── fromconfig
|
||||
│ ├── local
|
||||
│ ├── ssh
|
||||
│ ├── tcp
|
||||
│ └── tls
|
||||
├── util
|
||||
├── vendor # managed by dep
|
||||
├── version # abstraction for versions (filled during build by Makefile)
|
||||
└── zfs # zfs(8) wrappers
|
||||
```
|
||||
@@ -134,3 +134,15 @@ There will not be a big refactoring (an attempt was made, but it's destroying to
|
||||
|
||||
However, new contributions & patches should fix naming without further notice in the commit message.
|
||||
|
||||
### RPC debugging
|
||||
|
||||
Optionally, there are various RPC-related environment varibles, that if set to something != `""` will produce additional debug output on stderr:
|
||||
|
||||
https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
|
||||
|
||||
https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
|
||||
|
||||
https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
|
||||
|
||||
https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
|
||||
|
||||
|
||||
@@ -60,6 +60,8 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
|
||||
confFilter = j.Filesystems
|
||||
case *config.PushJob:
|
||||
confFilter = j.Filesystems
|
||||
case *config.SnapJob:
|
||||
confFilter = j.Filesystems
|
||||
default:
|
||||
return fmt.Errorf("job type %T does not have filesystems filter", j)
|
||||
}
|
||||
|
||||
+2
-2
@@ -147,8 +147,8 @@ func (j *controlJob) Run(ctx context.Context) {
|
||||
server := http.Server{
|
||||
Handler: mux,
|
||||
// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
|
||||
WriteTimeout: 1 * time.Second,
|
||||
ReadTimeout: 1 * time.Second,
|
||||
WriteTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_WRITE_TIMEOUT", 1*time.Second),
|
||||
ReadTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_READ_TIMEOUT", 1*time.Second),
|
||||
}
|
||||
|
||||
outer:
|
||||
|
||||
+2
-31
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/golang/protobuf/proto"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
@@ -181,39 +180,11 @@ func (j *PassiveSide) Run(ctx context.Context) {
|
||||
}
|
||||
|
||||
ctxInterceptor := func(handlerCtx context.Context) context.Context {
|
||||
reqLog := log.WithField(logging.ReqIDField, rpc.MustGetRequestID(handlerCtx))
|
||||
return logging.WithSubsystemLoggers(handlerCtx, reqLog)
|
||||
}
|
||||
|
||||
// ctxInterceptor already injected request loggers (with request id)
|
||||
pre := func(ctx context.Context, method string, req interface{}) {
|
||||
l := endpoint.GetLogger(ctx).WithField("method", method).WithField("reqT", fmt.Sprintf("%T", req))
|
||||
if p, ok := req.(proto.Message); ok {
|
||||
l = l.WithField("req", proto.CompactTextString(p))
|
||||
}
|
||||
ci, ok := ctx.Value(endpoint.ClientIdentityKey).(string)
|
||||
if !ok {
|
||||
// like endpoint, we must assume that rpc.Server sets it
|
||||
panic(method)
|
||||
}
|
||||
l = l.WithField("clientIdentity", ci)
|
||||
l.Debug("incoming")
|
||||
}
|
||||
post := func(ctx context.Context, res interface{}, err error) {
|
||||
l := endpoint.GetLogger(ctx).
|
||||
WithField("responseT", fmt.Sprintf("%T", res)).
|
||||
WithField("errT", fmt.Sprintf("%T", err))
|
||||
if s, ok := res.(fmt.Stringer); ok {
|
||||
l = l.WithField("response", s.String())
|
||||
}
|
||||
if err != nil {
|
||||
l = l.WithError(err)
|
||||
}
|
||||
l.Debug("handler done")
|
||||
return logging.WithSubsystemLoggers(handlerCtx, log)
|
||||
}
|
||||
|
||||
rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above
|
||||
server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor, pre, post)
|
||||
server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor)
|
||||
|
||||
listener, err := j.listen()
|
||||
if err != nil {
|
||||
|
||||
@@ -96,11 +96,6 @@ func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Contex
|
||||
Control: log.WithField(SubsysField, SubsysRPCControl),
|
||||
Data: log.WithField(SubsysField, SubsysRPCData),
|
||||
},
|
||||
// rpc.Loggers{
|
||||
// General: logger.NewNullLogger(),
|
||||
// Control: logger.NewNullLogger(),
|
||||
// Data: logger.NewNullLogger(),
|
||||
// },
|
||||
)
|
||||
return ctx
|
||||
}
|
||||
|
||||
@@ -22,7 +22,6 @@ const (
|
||||
const (
|
||||
JobField string = "job"
|
||||
SubsysField string = "subsystem"
|
||||
ReqIDField string = "reqid"
|
||||
)
|
||||
|
||||
type MetadataFlags int64
|
||||
@@ -86,7 +85,7 @@ func (f *HumanFormatter) Format(e *logger.Entry) (out []byte, err error) {
|
||||
fmt.Fprintf(&line, "[%s]", col.Sprint(e.Level.Short()))
|
||||
}
|
||||
|
||||
prefixFields := []string{JobField, SubsysField, ReqIDField}
|
||||
prefixFields := []string{JobField, SubsysField}
|
||||
prefixed := make(map[string]bool, len(prefixFields)+2)
|
||||
for _, field := range prefixFields {
|
||||
val, ok := e.Fields[field]
|
||||
|
||||
@@ -101,9 +101,9 @@ and serves as a reference for build dependencies and procedure:
|
||||
|
||||
::
|
||||
|
||||
git clone https://github.com/zrepl/zrepl.git
|
||||
cd zrepl
|
||||
sudo docker build -t zrepl_build -f build.Dockerfile .
|
||||
git clone https://github.com/zrepl/zrepl.git && \
|
||||
cd zrepl && \
|
||||
sudo docker build -t zrepl_build -f build.Dockerfile . && \
|
||||
sudo docker run -it --rm \
|
||||
-v "${PWD}:/src" \
|
||||
--user "$(id -u):$(id -g)" \
|
||||
|
||||
@@ -19,10 +19,6 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
|
||||
return context.WithValue(ctx, contextKeyLogger, log)
|
||||
}
|
||||
|
||||
func GetLogger(ctx context.Context) Logger {
|
||||
return getLogger(ctx)
|
||||
}
|
||||
|
||||
func getLogger(ctx context.Context) Logger {
|
||||
if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
|
||||
return l
|
||||
|
||||
@@ -159,7 +159,6 @@ func sendArgsFromPDUAndValidateExists(ctx context.Context, fs string, fsv *pdu.F
|
||||
}
|
||||
|
||||
func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
|
||||
getLogger(ctx).WithField("req", r.String()).Debug("incoming Send request")
|
||||
|
||||
_, err := s.filterCheckFS(r.Filesystem)
|
||||
if err != nil {
|
||||
@@ -260,7 +259,6 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
}
|
||||
|
||||
func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
|
||||
getLogger(ctx).WithField("req", r.String()).Debug("incoming SendCompleted request")
|
||||
orig := r.GetOriginalReq() // may be nil, always use proto getters
|
||||
fs := orig.GetFilesystem()
|
||||
|
||||
@@ -512,7 +510,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
|
||||
// present filesystem without the root_fs prefix
|
||||
fss := make([]*pdu.Filesystem, 0, len(filtered))
|
||||
for _, a := range filtered {
|
||||
l := getLogger(ctx).WithField("fs", a.ToString())
|
||||
l := getLogger(ctx).WithField("fs", a)
|
||||
ph, err := zfs.ZFSGetFilesystemPlaceholderState(a)
|
||||
if err != nil {
|
||||
l.WithError(err).Error("error getting placeholder state")
|
||||
@@ -599,7 +597,7 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zf
|
||||
var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
|
||||
|
||||
func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
|
||||
getLogger(ctx).WithField("rr", req.String()).Debug("incoming Receive")
|
||||
getLogger(ctx).Debug("incoming Receive")
|
||||
defer receive.Close()
|
||||
|
||||
root := s.clientRootFromCtx(ctx)
|
||||
|
||||
@@ -7,7 +7,6 @@ require (
|
||||
github.com/gdamore/tcell v1.2.0
|
||||
github.com/go-logfmt/logfmt v0.4.0
|
||||
github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
|
||||
github.com/gogo/protobuf v1.1.1
|
||||
github.com/golang/protobuf v1.3.2
|
||||
github.com/google/uuid v1.1.1
|
||||
github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
|
||||
|
||||
@@ -68,7 +68,6 @@ github.com/go-toolsmith/strparse v1.0.0/go.mod h1:YI2nUKP9YGZnL/L1/DLFBfixrcjslW
|
||||
github.com/go-toolsmith/typep v0.0.0-20181030061450-d63dc7650676/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
|
||||
github.com/go-toolsmith/typep v1.0.0/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
|
||||
github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
|
||||
github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
|
||||
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
|
||||
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
|
||||
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
|
||||
|
||||
@@ -5,7 +5,6 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -15,7 +14,6 @@ import (
|
||||
"google.golang.org/grpc/status"
|
||||
|
||||
"github.com/zrepl/zrepl/replication/report"
|
||||
"github.com/zrepl/zrepl/tracing"
|
||||
"github.com/zrepl/zrepl/util/chainlock"
|
||||
"github.com/zrepl/zrepl/util/envconst"
|
||||
)
|
||||
@@ -208,7 +206,6 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
|
||||
}
|
||||
run.attempts = append(run.attempts, cur)
|
||||
run.l.DropWhile(func() {
|
||||
ctx := tracing.Child(ctx, fmt.Sprintf("attempt#%d", ano)) // shadow
|
||||
cur.do(ctx, prev)
|
||||
})
|
||||
prev = cur
|
||||
@@ -284,7 +281,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
|
||||
}
|
||||
|
||||
func (a *attempt) do(ctx context.Context, prev *attempt) {
|
||||
pfss, err := a.planner.Plan(tracing.Child(ctx, "plan"))
|
||||
pfss, err := a.planner.Plan(ctx)
|
||||
errTime := time.Now()
|
||||
defer a.l.Lock().Unlock()
|
||||
if err != nil {
|
||||
@@ -364,7 +361,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
|
||||
fssesDone.Add(1)
|
||||
go func(f *fs) {
|
||||
defer fssesDone.Done()
|
||||
f.do(tracing.Child(ctx, f.fs.ReportInfo().Name), stepQueue, prevs[f])
|
||||
f.do(ctx, stepQueue, prevs[f])
|
||||
}(f)
|
||||
}
|
||||
a.l.DropWhile(func() {
|
||||
@@ -375,8 +372,6 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
|
||||
|
||||
func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
|
||||
|
||||
fmt.Fprintf(os.Stderr, "CHILD STACK: %v\n", tracing.GetStack(ctx))
|
||||
|
||||
defer fs.l.Lock().Unlock()
|
||||
|
||||
// get planned steps from replication logic
|
||||
|
||||
@@ -262,7 +262,6 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
|
||||
return nil, err
|
||||
}
|
||||
sfss := slfssres.GetFilesystems()
|
||||
// no progress here since we could run in a live-lock on connectivity issues
|
||||
|
||||
rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
|
||||
if err != nil {
|
||||
@@ -606,7 +605,7 @@ func (s *Step) doReplication(ctx context.Context) error {
|
||||
log := getLogger(ctx)
|
||||
sr := s.buildSendRequest(false)
|
||||
|
||||
log.WithField("sr", sr.String()).Debug("initiate send request")
|
||||
log.Debug("initiate send request")
|
||||
sres, sstreamCopier, err := s.sender.Send(ctx, sr)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("send request failed")
|
||||
@@ -633,7 +632,7 @@ func (s *Step) doReplication(ctx context.Context) error {
|
||||
To: sr.GetTo(),
|
||||
ClearResumeToken: !sres.UsedResumeToken,
|
||||
}
|
||||
log.WithField("rr", rr.String()).Debug("initiate receive request")
|
||||
log.Debug("initiate receive request")
|
||||
_, err = s.receiver.Receive(ctx, rr, byteCountingStream)
|
||||
if err != nil {
|
||||
log.
|
||||
@@ -649,11 +648,10 @@ func (s *Step) doReplication(ctx context.Context) error {
|
||||
}
|
||||
log.Debug("receive finished")
|
||||
|
||||
scr := &pdu.SendCompletedReq{
|
||||
log.Debug("tell sender replication completed")
|
||||
_, err = s.sender.SendCompleted(ctx, &pdu.SendCompletedReq{
|
||||
OriginalReq: sr,
|
||||
}
|
||||
log.WithField("scr", scr.String()).Debug("tell sender replication completed")
|
||||
_, err = s.sender.SendCompleted(ctx, scr)
|
||||
})
|
||||
if err != nil {
|
||||
log.WithError(err).Error("error telling sender that replication completed successfully")
|
||||
return err
|
||||
|
||||
@@ -75,7 +75,7 @@ func (c *Client) recv(ctx context.Context, conn *stream.Conn, res proto.Message)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
header := string(headerBuf) // FIXME
|
||||
header := string(headerBuf)
|
||||
if strings.HasPrefix(header, responseHeaderHandlerErrorPrefix) {
|
||||
// FIXME distinguishable error type
|
||||
return &RemoteHandlerError{strings.TrimPrefix(header, responseHeaderHandlerErrorPrefix)}
|
||||
|
||||
@@ -14,8 +14,8 @@ import (
|
||||
"github.com/zrepl/zrepl/zfs"
|
||||
)
|
||||
|
||||
// ReqInterceptor has a chance to exchange the context and connection on each request.
|
||||
type ReqInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
|
||||
// WireInterceptor has a chance to exchange the context and connection on each client connection.
|
||||
type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
|
||||
|
||||
// Handler implements the functionality that is exposed by Server to the Client.
|
||||
type Handler interface {
|
||||
@@ -30,27 +30,19 @@ type Handler interface {
|
||||
PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
|
||||
}
|
||||
|
||||
type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
|
||||
|
||||
type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
|
||||
|
||||
type Logger = logger.Logger
|
||||
|
||||
type Server struct {
|
||||
h Handler
|
||||
ri ReqInterceptor
|
||||
log Logger
|
||||
pre PreHandlerInspector
|
||||
post PostHandlerInspector
|
||||
h Handler
|
||||
wi WireInterceptor
|
||||
log Logger
|
||||
}
|
||||
|
||||
func NewServer(ri ReqInterceptor, logger Logger, pre PreHandlerInspector, handler Handler, post PostHandlerInspector) *Server {
|
||||
func NewServer(wi WireInterceptor, logger Logger, handler Handler) *Server {
|
||||
return &Server{
|
||||
h: handler,
|
||||
ri: ri,
|
||||
wi: wi,
|
||||
log: logger,
|
||||
pre: pre,
|
||||
post: post,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,8 +83,8 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
defer s.log.Debug("serveConn done")
|
||||
|
||||
ctx := context.Background()
|
||||
if s.ri != nil {
|
||||
ctx, nc = s.ri(ctx, nc)
|
||||
if s.wi != nil {
|
||||
ctx, nc = s.wi(ctx, nc)
|
||||
}
|
||||
|
||||
c := stream.Wrap(nc, HeartbeatInterval, HeartbeatPeerTimeout)
|
||||
@@ -108,7 +100,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
s.log.WithError(err).Error("error reading structured part")
|
||||
return
|
||||
}
|
||||
endpoint := string(header) // FIXME
|
||||
endpoint := string(header)
|
||||
|
||||
reqStructured, err := c.ReadStreamedMessage(ctx, RequestStructuredMaxSize, ReqStructured)
|
||||
if err != nil {
|
||||
@@ -128,7 +120,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
s.log.WithError(err).Error("cannot unmarshal send request")
|
||||
return
|
||||
}
|
||||
s.pre(ctx, endpoint, &req)
|
||||
res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
|
||||
case EndpointRecv:
|
||||
var req pdu.ReceiveReq
|
||||
@@ -136,7 +127,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
s.log.WithError(err).Error("cannot unmarshal receive request")
|
||||
return
|
||||
}
|
||||
s.pre(ctx, endpoint, &req)
|
||||
res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
|
||||
case EndpointPing:
|
||||
var req pdu.PingReq
|
||||
@@ -144,7 +134,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
s.log.WithError(err).Error("cannot unmarshal ping request")
|
||||
return
|
||||
}
|
||||
s.pre(ctx, endpoint, &req)
|
||||
res, handlerErr = s.h.PingDataconn(ctx, &req) // SHADOWING
|
||||
default:
|
||||
s.log.WithField("endpoint", endpoint).Error("unknown endpoint")
|
||||
@@ -153,8 +142,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
|
||||
|
||||
s.log.WithField("endpoint", endpoint).WithField("errType", fmt.Sprintf("%T", handlerErr)).Debug("handler returned")
|
||||
|
||||
s.post(ctx, res, handlerErr)
|
||||
|
||||
// prepare protobuf now to return the protobuf error in the header
|
||||
// if marshaling fails. We consider failed marshaling a handler error
|
||||
var protobuf *bytes.Buffer
|
||||
|
||||
@@ -49,7 +49,7 @@ type Wire interface {
|
||||
// No data that could otherwise be Read is lost as a consequence of this call.
|
||||
// The use case for this API is abortive connection shutdown.
|
||||
// To provide any value over draining Wire using io.Read, an implementation
|
||||
// will likely use out-of-bounds messaging mechanisms.
|
||||
// will likely use out-of-band messaging mechanisms.
|
||||
// TODO WaitForPeerClose() (supported bool, err error)
|
||||
}
|
||||
|
||||
|
||||
@@ -101,11 +101,7 @@ func (*transportCredentials) OverrideServerName(string) error {
|
||||
|
||||
type ContextInterceptor = func(ctx context.Context) context.Context
|
||||
|
||||
type PreHandlerInspector func(ctx context.Context, endpoint string, req interface{})
|
||||
|
||||
type PostHandlerInspector func(ctx context.Context, response interface{}, err error)
|
||||
|
||||
func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor, pre PreHandlerInspector, post PostHandlerInspector) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
|
||||
func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
|
||||
unary = func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
|
||||
logger.WithField("fullMethod", info.FullMethod).Debug("request")
|
||||
p, ok := peer.FromContext(ctx)
|
||||
@@ -122,10 +118,7 @@ func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxIntercepto
|
||||
if ctxInterceptor != nil {
|
||||
ctx = ctxInterceptor(ctx)
|
||||
}
|
||||
pre(ctx, info.FullMethod, req)
|
||||
res, err := handler(ctx, req)
|
||||
post(ctx, res, err)
|
||||
return res, err
|
||||
return handler(ctx, req)
|
||||
}
|
||||
stream = func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
||||
panic("unimplemented")
|
||||
|
||||
@@ -4,18 +4,14 @@ package grpchelper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/keepalive"
|
||||
"google.golang.org/grpc/metadata"
|
||||
|
||||
"github.com/zrepl/zrepl/logger"
|
||||
"github.com/zrepl/zrepl/rpc/grpcclientidentity"
|
||||
"github.com/zrepl/zrepl/rpc/netadaptor"
|
||||
"github.com/zrepl/zrepl/tracing"
|
||||
"github.com/zrepl/zrepl/transport"
|
||||
)
|
||||
|
||||
@@ -32,40 +28,15 @@ type Logger = logger.Logger
|
||||
|
||||
// ClientConn is an easy-to-use wrapper around the Dialer and TransportCredentials interface
|
||||
// to produce a grpc.ClientConn
|
||||
func ClientConn(cn transport.Connecter, log Logger, genReqID func() string) *grpc.ClientConn {
|
||||
func ClientConn(cn transport.Connecter, log Logger) *grpc.ClientConn {
|
||||
ka := grpc.WithKeepaliveParams(keepalive.ClientParameters{
|
||||
Time: StartKeepalivesAfterInactivityDuration,
|
||||
Timeout: KeepalivePeerTimeout,
|
||||
PermitWithoutStream: true,
|
||||
})
|
||||
unaryIntcpt := grpc.WithUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
|
||||
|
||||
traceStack := tracing.GetStack(ctx)
|
||||
if len(traceStack) == 0 {
|
||||
panic("implementation error: expecting trace stack")
|
||||
}
|
||||
reqId := genReqID() + " " + strings.Join(traceStack, " <> ")
|
||||
|
||||
l := log
|
||||
l = l.WithField("reqID", reqId)
|
||||
l = l.WithField("reqT", fmt.Sprintf("%T", req))
|
||||
l = l.WithField("repT", fmt.Sprintf("%T", reply))
|
||||
l = l.WithField("ctxT", fmt.Sprintf("%T", ctx))
|
||||
l.Debug("unary intercepted")
|
||||
|
||||
ctx = metadata.AppendToOutgoingContext(ctx, "zrepl-req-id", reqId)
|
||||
|
||||
err := invoker(ctx, method, req, reply, cc, opts...)
|
||||
|
||||
l = log.WithField("repT", fmt.Sprintf("%T", reply))
|
||||
l = log.WithField("errT", fmt.Sprintf("%T", err))
|
||||
l.Debug("reply intercepted")
|
||||
|
||||
return err
|
||||
})
|
||||
dialerOption := grpc.WithDialer(grpcclientidentity.NewDialer(log, cn))
|
||||
cred := grpc.WithTransportCredentials(grpcclientidentity.NewTransportCredentials(log))
|
||||
cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka, unaryIntcpt)
|
||||
cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("cannot create gRPC client conn (non-blocking)")
|
||||
// It's ok to panic here: the we call grpc.DialContext without the
|
||||
@@ -79,7 +50,7 @@ func ClientConn(cn transport.Connecter, log Logger, genReqID func() string) *grp
|
||||
}
|
||||
|
||||
// NewServer is a convenience interface around the TransportCredentials and Interceptors interface.
|
||||
func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, interceptor grpcclientidentity.ContextInterceptor, pre grpcclientidentity.PreHandlerInspector, post grpcclientidentity.PostHandlerInspector) (srv *grpc.Server, serve func() error) {
|
||||
func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, ctxInterceptor grpcclientidentity.ContextInterceptor) (srv *grpc.Server, serve func() error) {
|
||||
ka := grpc.KeepaliveParams(keepalive.ServerParameters{
|
||||
Time: StartKeepalivesAfterInactivityDuration,
|
||||
Timeout: KeepalivePeerTimeout,
|
||||
@@ -89,17 +60,8 @@ func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey i
|
||||
PermitWithoutStream: true,
|
||||
})
|
||||
tcs := grpcclientidentity.NewTransportCredentials(logger)
|
||||
unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, interceptor, pre, post)
|
||||
unary2 := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
|
||||
md, ok := metadata.FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
logger.Error("no request id in incoming request")
|
||||
} else {
|
||||
logger.Info(fmt.Sprintf("req-id = %v", md.Get("zrepl-req-id")))
|
||||
}
|
||||
return unary(ctx, req, info, handler)
|
||||
}
|
||||
srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary2), grpc.StreamInterceptor(stream), ka, ep)
|
||||
unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, ctxInterceptor)
|
||||
srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary), grpc.StreamInterceptor(stream), ka, ep)
|
||||
|
||||
serve = func() error {
|
||||
if err := srv.Serve(netadaptor.New(authListener, logger)); err != nil {
|
||||
|
||||
+3
-7
@@ -26,7 +26,6 @@ import (
|
||||
// Client implements the active side of a replication setup.
|
||||
// It satisfies the Endpoint, Sender and Receiver interface defined by package replication.
|
||||
type Client struct {
|
||||
reqIDGen *requestIDGenerator
|
||||
dataClient *dataconn.Client
|
||||
controlClient pdu.ReplicationClient // this the grpc client instance, see constructor
|
||||
controlConn *grpc.ClientConn
|
||||
@@ -48,13 +47,10 @@ func NewClient(cn transport.Connecter, loggers Loggers) *Client {
|
||||
muxedConnecter := mux(cn)
|
||||
|
||||
c := &Client{
|
||||
loggers: loggers,
|
||||
closed: make(chan struct{}),
|
||||
reqIDGen: newRequestIDGenerator(),
|
||||
loggers: loggers,
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control, func() string {
|
||||
return c.reqIDGen.newID().String()
|
||||
})
|
||||
grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control)
|
||||
|
||||
go func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
@@ -1,58 +0,0 @@
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type requestIDGenerator struct{}
|
||||
|
||||
func newRequestIDGenerator() *requestIDGenerator {
|
||||
return &requestIDGenerator{}
|
||||
}
|
||||
|
||||
type RequestID struct{ s string }
|
||||
|
||||
func (r RequestID) String() string { return r.s }
|
||||
|
||||
type requestIDContextKey int
|
||||
|
||||
const (
|
||||
requestIDContextKeyRequestID requestIDContextKey = 1 + iota
|
||||
)
|
||||
|
||||
func (g *requestIDGenerator) newID() RequestID {
|
||||
id := uuid.New()
|
||||
var buf strings.Builder
|
||||
enc := base64.NewEncoder(base64.RawStdEncoding, &buf)
|
||||
n, err := enc.Write(id[:])
|
||||
if err != nil {
|
||||
panic(err)
|
||||
} else if n != len(id) {
|
||||
panic(n)
|
||||
}
|
||||
if err := enc.Close(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return RequestID{buf.String()}
|
||||
}
|
||||
|
||||
func (g *requestIDGenerator) inject(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, requestIDContextKeyRequestID, g.newID())
|
||||
}
|
||||
|
||||
func GetRequestID(ctx context.Context) (id RequestID, ok bool) {
|
||||
id, ok = ctx.Value(requestIDContextKeyRequestID).(RequestID)
|
||||
return id, ok
|
||||
}
|
||||
|
||||
func MustGetRequestID(ctx context.Context) (id RequestID) {
|
||||
id, ok := GetRequestID(ctx)
|
||||
if !ok {
|
||||
panic("calling context expectes request id to bet set")
|
||||
}
|
||||
return id
|
||||
}
|
||||
+4
-25
@@ -7,7 +7,6 @@ import (
|
||||
"github.com/zrepl/zrepl/endpoint"
|
||||
"github.com/zrepl/zrepl/replication/logic/pdu"
|
||||
"github.com/zrepl/zrepl/rpc/dataconn"
|
||||
"github.com/zrepl/zrepl/rpc/grpcclientidentity"
|
||||
"github.com/zrepl/zrepl/rpc/grpcclientidentity/grpchelper"
|
||||
"github.com/zrepl/zrepl/rpc/versionhandshake"
|
||||
"github.com/zrepl/zrepl/transport"
|
||||
@@ -31,35 +30,15 @@ type Server struct {
|
||||
dataServerServe serveFunc
|
||||
}
|
||||
|
||||
type CtxInterceptor = grpcclientidentity.ContextInterceptor
|
||||
|
||||
func chainInterceptors(interceptors []CtxInterceptor) CtxInterceptor {
|
||||
return func(ctx context.Context) (_ context.Context) {
|
||||
for _, i := range interceptors {
|
||||
ctx = i(ctx)
|
||||
}
|
||||
return ctx
|
||||
}
|
||||
}
|
||||
|
||||
type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
|
||||
|
||||
type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
|
||||
type HandlerContextInterceptor func(ctx context.Context) context.Context
|
||||
|
||||
// config must be valid (use its Validate function).
|
||||
func NewServer(handler Handler, loggers Loggers, ctxInterceptor CtxInterceptor, pre PreHandlerInspector, post PostHandlerInspector) *Server {
|
||||
|
||||
reqIdGen := newRequestIDGenerator()
|
||||
|
||||
ctxInterceptor = chainInterceptors([]CtxInterceptor{
|
||||
reqIdGen.inject,
|
||||
ctxInterceptor,
|
||||
})
|
||||
func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextInterceptor) *Server {
|
||||
|
||||
// setup control server
|
||||
controlServerServe := func(ctx context.Context, controlListener transport.AuthenticatedListener, errOut chan<- error) {
|
||||
|
||||
controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor, pre, post)
|
||||
controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor)
|
||||
pdu.RegisterReplicationServer(controlServer, handler)
|
||||
|
||||
// give time for graceful stop until deadline expires, then hard stop
|
||||
@@ -84,7 +63,7 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor CtxInterceptor,
|
||||
}
|
||||
return ctx, wire
|
||||
}
|
||||
dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, pre, handler, post)
|
||||
dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, handler)
|
||||
dataServerServe := func(ctx context.Context, dataListener transport.AuthenticatedListener, errOut chan<- error) {
|
||||
dataServer.Serve(ctx, dataListener)
|
||||
errOut <- nil // TODO bad design of dataServer?
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
package tracing
|
||||
|
||||
import "context"
|
||||
|
||||
type tracingContextKey int
|
||||
|
||||
const (
|
||||
CallerContext tracingContextKey = 1 + iota
|
||||
)
|
||||
|
||||
type jobSubtree struct {
|
||||
jobid string
|
||||
}
|
||||
|
||||
type ctx struct {
|
||||
parent *ctx
|
||||
job *jobSubtree
|
||||
ident string
|
||||
}
|
||||
|
||||
var root = &ctx{nil, nil, ""}
|
||||
|
||||
func getParentOrRoot(c context.Context) *ctx {
|
||||
parent, ok := c.Value(CallerContext).(*ctx)
|
||||
if !ok {
|
||||
parent = root
|
||||
}
|
||||
return parent
|
||||
}
|
||||
|
||||
func makeChild(c context.Context, child *ctx) context.Context {
|
||||
if child.parent == nil {
|
||||
panic(child)
|
||||
}
|
||||
return context.WithValue(c, CallerContext, child)
|
||||
}
|
||||
|
||||
func Child(c context.Context, ident string) context.Context {
|
||||
parent := getParentOrRoot(c)
|
||||
return makeChild(c, &ctx{parent: parent, ident: ident})
|
||||
}
|
||||
|
||||
func GetStack(c context.Context) (idents []string) {
|
||||
ct, ok := c.Value(CallerContext).(*ctx)
|
||||
if !ok {
|
||||
return idents
|
||||
}
|
||||
for ct.parent != nil {
|
||||
idents = append(idents, ct.ident)
|
||||
ct = ct.parent
|
||||
}
|
||||
return idents
|
||||
}
|
||||
@@ -1,21 +0,0 @@
|
||||
package tracing
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestIt(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
ctx = Child(ctx, "a")
|
||||
ctx = Child(ctx, "b")
|
||||
ctx = Child(ctx, "c")
|
||||
ctx = Child(ctx, "d")
|
||||
|
||||
assert.Equal(t, "dcba", strings.Join(GetStack(ctx), ""))
|
||||
|
||||
}
|
||||
@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp
|
||||
|
||||
serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "cannot parse cer/key pair")
|
||||
return nil, errors.Wrap(err, "cannot parse cert/key pair")
|
||||
}
|
||||
|
||||
clientCNs := make(map[string]struct{}, len(in.ClientCNs))
|
||||
|
||||
Reference in New Issue
Block a user