mirror of https://github.com/docker/cli.git
add containerd/ttrpc f02858b1457c5ca3aaec3a0803eb0d59f96e41d6
Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
This commit is contained in:
Sebastiaan van Stijn
parent
51de9a883a
commit
41fe464139
|
|
@ -7,6 +7,7 @@ github.com/containerd/console 0650fd9eeb50bab4fc99dceb9f2e
|
||||||
github.com/containerd/containerd 3a3f0aac8819165839a41fee77a4f4ac8b103097
|
github.com/containerd/containerd 3a3f0aac8819165839a41fee77a4f4ac8b103097
|
||||||
github.com/containerd/continuity aaeac12a7ffcd198ae25440a9dff125c2e2703a7
|
github.com/containerd/continuity aaeac12a7ffcd198ae25440a9dff125c2e2703a7
|
||||||
github.com/containerd/fifo a9fb20d87448d386e6d50b1f2e1fa70dcf0de43c
|
github.com/containerd/fifo a9fb20d87448d386e6d50b1f2e1fa70dcf0de43c
|
||||||
|
github.com/containerd/ttrpc f02858b1457c5ca3aaec3a0803eb0d59f96e41d6
|
||||||
github.com/containerd/typeurl 2a93cfde8c20b23de8eb84a5adbc234ddf7a9e8d
|
github.com/containerd/typeurl 2a93cfde8c20b23de8eb84a5adbc234ddf7a9e8d
|
||||||
github.com/coreos/etcd d57e8b8d97adfc4a6c224fe116714bf1a1f3beb9 # v3.3.12
|
github.com/coreos/etcd d57e8b8d97adfc4a6c224fe116714bf1a1f3beb9 # v3.3.12
|
||||||
github.com/cpuguy83/go-md2man 20f5889cbdc3c73dbd2862796665e7c465ade7d1 # v1.0.8
|
github.com/cpuguy83/go-md2man 20f5889cbdc3c73dbd2862796665e7c465ade7d1 # v1.0.8
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,201 @@
|
||||||
|
Apache License
|
||||||
|
Version 2.0, January 2004
|
||||||
|
http://www.apache.org/licenses/
|
||||||
|
|
||||||
|
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
|
||||||
|
|
||||||
|
1. Definitions.
|
||||||
|
|
||||||
|
"License" shall mean the terms and conditions for use, reproduction,
|
||||||
|
and distribution as defined by Sections 1 through 9 of this document.
|
||||||
|
|
||||||
|
"Licensor" shall mean the copyright owner or entity authorized by
|
||||||
|
the copyright owner that is granting the License.
|
||||||
|
|
||||||
|
"Legal Entity" shall mean the union of the acting entity and all
|
||||||
|
other entities that control, are controlled by, or are under common
|
||||||
|
control with that entity. For the purposes of this definition,
|
||||||
|
"control" means (i) the power, direct or indirect, to cause the
|
||||||
|
direction or management of such entity, whether by contract or
|
||||||
|
otherwise, or (ii) ownership of fifty percent (50%) or more of the
|
||||||
|
outstanding shares, or (iii) beneficial ownership of such entity.
|
||||||
|
|
||||||
|
"You" (or "Your") shall mean an individual or Legal Entity
|
||||||
|
exercising permissions granted by this License.
|
||||||
|
|
||||||
|
"Source" form shall mean the preferred form for making modifications,
|
||||||
|
including but not limited to software source code, documentation
|
||||||
|
source, and configuration files.
|
||||||
|
|
||||||
|
"Object" form shall mean any form resulting from mechanical
|
||||||
|
transformation or translation of a Source form, including but
|
||||||
|
not limited to compiled object code, generated documentation,
|
||||||
|
and conversions to other media types.
|
||||||
|
|
||||||
|
"Work" shall mean the work of authorship, whether in Source or
|
||||||
|
Object form, made available under the License, as indicated by a
|
||||||
|
copyright notice that is included in or attached to the work
|
||||||
|
(an example is provided in the Appendix below).
|
||||||
|
|
||||||
|
"Derivative Works" shall mean any work, whether in Source or Object
|
||||||
|
form, that is based on (or derived from) the Work and for which the
|
||||||
|
editorial revisions, annotations, elaborations, or other modifications
|
||||||
|
represent, as a whole, an original work of authorship. For the purposes
|
||||||
|
of this License, Derivative Works shall not include works that remain
|
||||||
|
separable from, or merely link (or bind by name) to the interfaces of,
|
||||||
|
the Work and Derivative Works thereof.
|
||||||
|
|
||||||
|
"Contribution" shall mean any work of authorship, including
|
||||||
|
the original version of the Work and any modifications or additions
|
||||||
|
to that Work or Derivative Works thereof, that is intentionally
|
||||||
|
submitted to Licensor for inclusion in the Work by the copyright owner
|
||||||
|
or by an individual or Legal Entity authorized to submit on behalf of
|
||||||
|
the copyright owner. For the purposes of this definition, "submitted"
|
||||||
|
means any form of electronic, verbal, or written communication sent
|
||||||
|
to the Licensor or its representatives, including but not limited to
|
||||||
|
communication on electronic mailing lists, source code control systems,
|
||||||
|
and issue tracking systems that are managed by, or on behalf of, the
|
||||||
|
Licensor for the purpose of discussing and improving the Work, but
|
||||||
|
excluding communication that is conspicuously marked or otherwise
|
||||||
|
designated in writing by the copyright owner as "Not a Contribution."
|
||||||
|
|
||||||
|
"Contributor" shall mean Licensor and any individual or Legal Entity
|
||||||
|
on behalf of whom a Contribution has been received by Licensor and
|
||||||
|
subsequently incorporated within the Work.
|
||||||
|
|
||||||
|
2. Grant of Copyright License. Subject to the terms and conditions of
|
||||||
|
this License, each Contributor hereby grants to You a perpetual,
|
||||||
|
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
|
||||||
|
copyright license to reproduce, prepare Derivative Works of,
|
||||||
|
publicly display, publicly perform, sublicense, and distribute the
|
||||||
|
Work and such Derivative Works in Source or Object form.
|
||||||
|
|
||||||
|
3. Grant of Patent License. Subject to the terms and conditions of
|
||||||
|
this License, each Contributor hereby grants to You a perpetual,
|
||||||
|
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
|
||||||
|
(except as stated in this section) patent license to make, have made,
|
||||||
|
use, offer to sell, sell, import, and otherwise transfer the Work,
|
||||||
|
where such license applies only to those patent claims licensable
|
||||||
|
by such Contributor that are necessarily infringed by their
|
||||||
|
Contribution(s) alone or by combination of their Contribution(s)
|
||||||
|
with the Work to which such Contribution(s) was submitted. If You
|
||||||
|
institute patent litigation against any entity (including a
|
||||||
|
cross-claim or counterclaim in a lawsuit) alleging that the Work
|
||||||
|
or a Contribution incorporated within the Work constitutes direct
|
||||||
|
or contributory patent infringement, then any patent licenses
|
||||||
|
granted to You under this License for that Work shall terminate
|
||||||
|
as of the date such litigation is filed.
|
||||||
|
|
||||||
|
4. Redistribution. You may reproduce and distribute copies of the
|
||||||
|
Work or Derivative Works thereof in any medium, with or without
|
||||||
|
modifications, and in Source or Object form, provided that You
|
||||||
|
meet the following conditions:
|
||||||
|
|
||||||
|
(a) You must give any other recipients of the Work or
|
||||||
|
Derivative Works a copy of this License; and
|
||||||
|
|
||||||
|
(b) You must cause any modified files to carry prominent notices
|
||||||
|
stating that You changed the files; and
|
||||||
|
|
||||||
|
(c) You must retain, in the Source form of any Derivative Works
|
||||||
|
that You distribute, all copyright, patent, trademark, and
|
||||||
|
attribution notices from the Source form of the Work,
|
||||||
|
excluding those notices that do not pertain to any part of
|
||||||
|
the Derivative Works; and
|
||||||
|
|
||||||
|
(d) If the Work includes a "NOTICE" text file as part of its
|
||||||
|
distribution, then any Derivative Works that You distribute must
|
||||||
|
include a readable copy of the attribution notices contained
|
||||||
|
within such NOTICE file, excluding those notices that do not
|
||||||
|
pertain to any part of the Derivative Works, in at least one
|
||||||
|
of the following places: within a NOTICE text file distributed
|
||||||
|
as part of the Derivative Works; within the Source form or
|
||||||
|
documentation, if provided along with the Derivative Works; or,
|
||||||
|
within a display generated by the Derivative Works, if and
|
||||||
|
wherever such third-party notices normally appear. The contents
|
||||||
|
of the NOTICE file are for informational purposes only and
|
||||||
|
do not modify the License. You may add Your own attribution
|
||||||
|
notices within Derivative Works that You distribute, alongside
|
||||||
|
or as an addendum to the NOTICE text from the Work, provided
|
||||||
|
that such additional attribution notices cannot be construed
|
||||||
|
as modifying the License.
|
||||||
|
|
||||||
|
You may add Your own copyright statement to Your modifications and
|
||||||
|
may provide additional or different license terms and conditions
|
||||||
|
for use, reproduction, or distribution of Your modifications, or
|
||||||
|
for any such Derivative Works as a whole, provided Your use,
|
||||||
|
reproduction, and distribution of the Work otherwise complies with
|
||||||
|
the conditions stated in this License.
|
||||||
|
|
||||||
|
5. Submission of Contributions. Unless You explicitly state otherwise,
|
||||||
|
any Contribution intentionally submitted for inclusion in the Work
|
||||||
|
by You to the Licensor shall be under the terms and conditions of
|
||||||
|
this License, without any additional terms or conditions.
|
||||||
|
Notwithstanding the above, nothing herein shall supersede or modify
|
||||||
|
the terms of any separate license agreement you may have executed
|
||||||
|
with Licensor regarding such Contributions.
|
||||||
|
|
||||||
|
6. Trademarks. This License does not grant permission to use the trade
|
||||||
|
names, trademarks, service marks, or product names of the Licensor,
|
||||||
|
except as required for reasonable and customary use in describing the
|
||||||
|
origin of the Work and reproducing the content of the NOTICE file.
|
||||||
|
|
||||||
|
7. Disclaimer of Warranty. Unless required by applicable law or
|
||||||
|
agreed to in writing, Licensor provides the Work (and each
|
||||||
|
Contributor provides its Contributions) on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||||
|
implied, including, without limitation, any warranties or conditions
|
||||||
|
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
|
||||||
|
PARTICULAR PURPOSE. You are solely responsible for determining the
|
||||||
|
appropriateness of using or redistributing the Work and assume any
|
||||||
|
risks associated with Your exercise of permissions under this License.
|
||||||
|
|
||||||
|
8. Limitation of Liability. In no event and under no legal theory,
|
||||||
|
whether in tort (including negligence), contract, or otherwise,
|
||||||
|
unless required by applicable law (such as deliberate and grossly
|
||||||
|
negligent acts) or agreed to in writing, shall any Contributor be
|
||||||
|
liable to You for damages, including any direct, indirect, special,
|
||||||
|
incidental, or consequential damages of any character arising as a
|
||||||
|
result of this License or out of the use or inability to use the
|
||||||
|
Work (including but not limited to damages for loss of goodwill,
|
||||||
|
work stoppage, computer failure or malfunction, or any and all
|
||||||
|
other commercial damages or losses), even if such Contributor
|
||||||
|
has been advised of the possibility of such damages.
|
||||||
|
|
||||||
|
9. Accepting Warranty or Additional Liability. While redistributing
|
||||||
|
the Work or Derivative Works thereof, You may choose to offer,
|
||||||
|
and charge a fee for, acceptance of support, warranty, indemnity,
|
||||||
|
or other liability obligations and/or rights consistent with this
|
||||||
|
License. However, in accepting such obligations, You may act only
|
||||||
|
on Your own behalf and on Your sole responsibility, not on behalf
|
||||||
|
of any other Contributor, and only if You agree to indemnify,
|
||||||
|
defend, and hold each Contributor harmless for any liability
|
||||||
|
incurred by, or claims asserted against, such Contributor by reason
|
||||||
|
of your accepting any such warranty or additional liability.
|
||||||
|
|
||||||
|
END OF TERMS AND CONDITIONS
|
||||||
|
|
||||||
|
APPENDIX: How to apply the Apache License to your work.
|
||||||
|
|
||||||
|
To apply the Apache License to your work, attach the following
|
||||||
|
boilerplate notice, with the fields enclosed by brackets "[]"
|
||||||
|
replaced with your own identifying information. (Don't include
|
||||||
|
the brackets!) The text should be enclosed in the appropriate
|
||||||
|
comment syntax for the file format. We also recommend that a
|
||||||
|
file or class name and description of purpose be included on the
|
||||||
|
same "printed page" as the copyright notice for easier
|
||||||
|
identification within third-party archives.
|
||||||
|
|
||||||
|
Copyright [yyyy] [name of copyright owner]
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
|
@ -0,0 +1,62 @@
|
||||||
|
# ttrpc
|
||||||
|
|
||||||
|
[](https://travis-ci.org/containerd/ttrpc)
|
||||||
|
|
||||||
|
GRPC for low-memory environments.
|
||||||
|
|
||||||
|
The existing grpc-go project requires a lot of memory overhead for importing
|
||||||
|
packages and at runtime. While this is great for many services with low density
|
||||||
|
requirements, this can be a problem when running a large number of services on
|
||||||
|
a single machine or on a machine with a small amount of memory.
|
||||||
|
|
||||||
|
Using the same GRPC definitions, this project reduces the binary size and
|
||||||
|
protocol overhead required. We do this by eliding the `net/http`, `net/http2`
|
||||||
|
and `grpc` package used by grpc replacing it with a lightweight framing
|
||||||
|
protocol. The result are smaller binaries that use less resident memory with
|
||||||
|
the same ease of use as GRPC.
|
||||||
|
|
||||||
|
Please note that while this project supports generating either end of the
|
||||||
|
protocol, the generated service definitions will be incompatible with regular
|
||||||
|
GRPC services, as they do not speak the same protocol.
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
|
||||||
|
Create a gogo vanity binary (see
|
||||||
|
[`cmd/protoc-gen-gogottrpc/main.go`](cmd/protoc-gen-gogottrpc/main.go) for an
|
||||||
|
example with the ttrpc plugin enabled.
|
||||||
|
|
||||||
|
It's recommended to use [`protobuild`](https://github.com//stevvooe/protobuild)
|
||||||
|
to build the protobufs for this project, but this will work with protoc
|
||||||
|
directly, if required.
|
||||||
|
|
||||||
|
# Differences from GRPC
|
||||||
|
|
||||||
|
- The protocol stack has been replaced with a lighter protocol that doesn't
|
||||||
|
require http, http2 and tls.
|
||||||
|
- The client and server interface are identical whereas in GRPC there is a
|
||||||
|
client and server interface that are different.
|
||||||
|
- The Go stdlib context package is used instead.
|
||||||
|
- No support for streams yet.
|
||||||
|
|
||||||
|
# Status
|
||||||
|
|
||||||
|
Very new. YMMV.
|
||||||
|
|
||||||
|
TODO:
|
||||||
|
|
||||||
|
- [X] Plumb error codes and GRPC status
|
||||||
|
- [X] Remove use of any type and dependency on typeurl package
|
||||||
|
- [X] Ensure that protocol can support streaming in the future
|
||||||
|
- [ ] Document protocol layout
|
||||||
|
- [ ] Add testing under concurrent load to ensure
|
||||||
|
- [ ] Verify connection error handling
|
||||||
|
|
||||||
|
# Project details
|
||||||
|
|
||||||
|
ttrpc is a containerd sub-project, licensed under the [Apache 2.0 license](./LICENSE).
|
||||||
|
As a containerd sub-project, you will find the:
|
||||||
|
* [Project governance](https://github.com/containerd/project/blob/master/GOVERNANCE.md),
|
||||||
|
* [Maintainers](https://github.com/containerd/project/blob/master/MAINTAINERS),
|
||||||
|
* and [Contributing guidelines](https://github.com/containerd/project/blob/master/CONTRIBUTING.md)
|
||||||
|
|
||||||
|
information in our [`containerd/project`](https://github.com/containerd/project) repository.
|
||||||
|
|
@ -0,0 +1,154 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"context"
|
||||||
|
"encoding/binary"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"google.golang.org/grpc/codes"
|
||||||
|
"google.golang.org/grpc/status"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
messageHeaderLength = 10
|
||||||
|
messageLengthMax = 4 << 20
|
||||||
|
)
|
||||||
|
|
||||||
|
type messageType uint8
|
||||||
|
|
||||||
|
const (
|
||||||
|
messageTypeRequest messageType = 0x1
|
||||||
|
messageTypeResponse messageType = 0x2
|
||||||
|
)
|
||||||
|
|
||||||
|
// messageHeader represents the fixed-length message header of 10 bytes sent
|
||||||
|
// with every request.
|
||||||
|
type messageHeader struct {
|
||||||
|
Length uint32 // length excluding this header. b[:4]
|
||||||
|
StreamID uint32 // identifies which request stream message is a part of. b[4:8]
|
||||||
|
Type messageType // message type b[8]
|
||||||
|
Flags uint8 // reserved b[9]
|
||||||
|
}
|
||||||
|
|
||||||
|
func readMessageHeader(p []byte, r io.Reader) (messageHeader, error) {
|
||||||
|
_, err := io.ReadFull(r, p[:messageHeaderLength])
|
||||||
|
if err != nil {
|
||||||
|
return messageHeader{}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return messageHeader{
|
||||||
|
Length: binary.BigEndian.Uint32(p[:4]),
|
||||||
|
StreamID: binary.BigEndian.Uint32(p[4:8]),
|
||||||
|
Type: messageType(p[8]),
|
||||||
|
Flags: p[9],
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func writeMessageHeader(w io.Writer, p []byte, mh messageHeader) error {
|
||||||
|
binary.BigEndian.PutUint32(p[:4], mh.Length)
|
||||||
|
binary.BigEndian.PutUint32(p[4:8], mh.StreamID)
|
||||||
|
p[8] = byte(mh.Type)
|
||||||
|
p[9] = mh.Flags
|
||||||
|
|
||||||
|
_, err := w.Write(p[:])
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
var buffers sync.Pool
|
||||||
|
|
||||||
|
type channel struct {
|
||||||
|
conn net.Conn
|
||||||
|
bw *bufio.Writer
|
||||||
|
br *bufio.Reader
|
||||||
|
hrbuf [messageHeaderLength]byte // avoid alloc when reading header
|
||||||
|
hwbuf [messageHeaderLength]byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func newChannel(conn net.Conn) *channel {
|
||||||
|
return &channel{
|
||||||
|
conn: conn,
|
||||||
|
bw: bufio.NewWriter(conn),
|
||||||
|
br: bufio.NewReader(conn),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// recv a message from the channel. The returned buffer contains the message.
|
||||||
|
//
|
||||||
|
// If a valid grpc status is returned, the message header
|
||||||
|
// returned will be valid and caller should send that along to
|
||||||
|
// the correct consumer. The bytes on the underlying channel
|
||||||
|
// will be discarded.
|
||||||
|
func (ch *channel) recv(ctx context.Context) (messageHeader, []byte, error) {
|
||||||
|
mh, err := readMessageHeader(ch.hrbuf[:], ch.br)
|
||||||
|
if err != nil {
|
||||||
|
return messageHeader{}, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
if mh.Length > uint32(messageLengthMax) {
|
||||||
|
if _, err := ch.br.Discard(int(mh.Length)); err != nil {
|
||||||
|
return mh, nil, errors.Wrapf(err, "failed to discard after receiving oversized message")
|
||||||
|
}
|
||||||
|
|
||||||
|
return mh, nil, status.Errorf(codes.ResourceExhausted, "message length %v exceed maximum message size of %v", mh.Length, messageLengthMax)
|
||||||
|
}
|
||||||
|
|
||||||
|
p := ch.getmbuf(int(mh.Length))
|
||||||
|
if _, err := io.ReadFull(ch.br, p); err != nil {
|
||||||
|
return messageHeader{}, nil, errors.Wrapf(err, "failed reading message")
|
||||||
|
}
|
||||||
|
|
||||||
|
return mh, p, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (ch *channel) send(ctx context.Context, streamID uint32, t messageType, p []byte) error {
|
||||||
|
if err := writeMessageHeader(ch.bw, ch.hwbuf[:], messageHeader{Length: uint32(len(p)), StreamID: streamID, Type: t}); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := ch.bw.Write(p)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return ch.bw.Flush()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (ch *channel) getmbuf(size int) []byte {
|
||||||
|
// we can't use the standard New method on pool because we want to allocate
|
||||||
|
// based on size.
|
||||||
|
b, ok := buffers.Get().(*[]byte)
|
||||||
|
if !ok || cap(*b) < size {
|
||||||
|
// TODO(stevvooe): It may be better to allocate these in fixed length
|
||||||
|
// buckets to reduce fragmentation but its not clear that would help
|
||||||
|
// with performance. An ilogb approach or similar would work well.
|
||||||
|
bb := make([]byte, size)
|
||||||
|
b = &bb
|
||||||
|
} else {
|
||||||
|
*b = (*b)[:size]
|
||||||
|
}
|
||||||
|
return *b
|
||||||
|
}
|
||||||
|
|
||||||
|
func (ch *channel) putmbuf(p []byte) {
|
||||||
|
buffers.Put(&p)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,290 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/gogo/protobuf/proto"
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
"google.golang.org/grpc/status"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrClosed is returned by client methods when the underlying connection is
|
||||||
|
// closed.
|
||||||
|
var ErrClosed = errors.New("ttrpc: closed")
|
||||||
|
|
||||||
|
type Client struct {
|
||||||
|
codec codec
|
||||||
|
conn net.Conn
|
||||||
|
channel *channel
|
||||||
|
calls chan *callRequest
|
||||||
|
|
||||||
|
closed chan struct{}
|
||||||
|
closeOnce sync.Once
|
||||||
|
closeFunc func()
|
||||||
|
done chan struct{}
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewClient(conn net.Conn) *Client {
|
||||||
|
c := &Client{
|
||||||
|
codec: codec{},
|
||||||
|
conn: conn,
|
||||||
|
channel: newChannel(conn),
|
||||||
|
calls: make(chan *callRequest),
|
||||||
|
closed: make(chan struct{}),
|
||||||
|
done: make(chan struct{}),
|
||||||
|
closeFunc: func() {},
|
||||||
|
}
|
||||||
|
|
||||||
|
go c.run()
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
|
||||||
|
type callRequest struct {
|
||||||
|
ctx context.Context
|
||||||
|
req *Request
|
||||||
|
resp *Response // response will be written back here
|
||||||
|
errs chan error // error written here on completion
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) Call(ctx context.Context, service, method string, req, resp interface{}) error {
|
||||||
|
payload, err := c.codec.Marshal(req)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
creq = &Request{
|
||||||
|
Service: service,
|
||||||
|
Method: method,
|
||||||
|
Payload: payload,
|
||||||
|
}
|
||||||
|
|
||||||
|
cresp = &Response{}
|
||||||
|
)
|
||||||
|
|
||||||
|
if dl, ok := ctx.Deadline(); ok {
|
||||||
|
creq.TimeoutNano = dl.Sub(time.Now()).Nanoseconds()
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.dispatch(ctx, creq, cresp); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.codec.Unmarshal(cresp.Payload, resp); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if cresp.Status == nil {
|
||||||
|
return errors.New("no status provided on response")
|
||||||
|
}
|
||||||
|
|
||||||
|
return status.ErrorProto(cresp.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) dispatch(ctx context.Context, req *Request, resp *Response) error {
|
||||||
|
errs := make(chan error, 1)
|
||||||
|
call := &callRequest{
|
||||||
|
ctx: ctx,
|
||||||
|
req: req,
|
||||||
|
resp: resp,
|
||||||
|
errs: errs,
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
case c.calls <- call:
|
||||||
|
case <-c.done:
|
||||||
|
return c.err
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
case err := <-errs:
|
||||||
|
return filterCloseErr(err)
|
||||||
|
case <-c.done:
|
||||||
|
return c.err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) Close() error {
|
||||||
|
c.closeOnce.Do(func() {
|
||||||
|
close(c.closed)
|
||||||
|
})
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// OnClose allows a close func to be called when the server is closed
|
||||||
|
func (c *Client) OnClose(closer func()) {
|
||||||
|
c.closeFunc = closer
|
||||||
|
}
|
||||||
|
|
||||||
|
type message struct {
|
||||||
|
messageHeader
|
||||||
|
p []byte
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) run() {
|
||||||
|
var (
|
||||||
|
streamID uint32 = 1
|
||||||
|
waiters = make(map[uint32]*callRequest)
|
||||||
|
calls = c.calls
|
||||||
|
incoming = make(chan *message)
|
||||||
|
shutdown = make(chan struct{})
|
||||||
|
shutdownErr error
|
||||||
|
)
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer close(shutdown)
|
||||||
|
|
||||||
|
// start one more goroutine to recv messages without blocking.
|
||||||
|
for {
|
||||||
|
mh, p, err := c.channel.recv(context.TODO())
|
||||||
|
if err != nil {
|
||||||
|
_, ok := status.FromError(err)
|
||||||
|
if !ok {
|
||||||
|
// treat all errors that are not an rpc status as terminal.
|
||||||
|
// all others poison the connection.
|
||||||
|
shutdownErr = err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case incoming <- &message{
|
||||||
|
messageHeader: mh,
|
||||||
|
p: p[:mh.Length],
|
||||||
|
err: err,
|
||||||
|
}:
|
||||||
|
case <-c.done:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
defer c.conn.Close()
|
||||||
|
defer close(c.done)
|
||||||
|
defer c.closeFunc()
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case call := <-calls:
|
||||||
|
if err := c.send(call.ctx, streamID, messageTypeRequest, call.req); err != nil {
|
||||||
|
call.errs <- err
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
waiters[streamID] = call
|
||||||
|
streamID += 2 // enforce odd client initiated request ids
|
||||||
|
case msg := <-incoming:
|
||||||
|
call, ok := waiters[msg.StreamID]
|
||||||
|
if !ok {
|
||||||
|
logrus.Errorf("ttrpc: received message for unknown channel %v", msg.StreamID)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
call.errs <- c.recv(call.resp, msg)
|
||||||
|
delete(waiters, msg.StreamID)
|
||||||
|
case <-shutdown:
|
||||||
|
if shutdownErr != nil {
|
||||||
|
shutdownErr = filterCloseErr(shutdownErr)
|
||||||
|
} else {
|
||||||
|
shutdownErr = ErrClosed
|
||||||
|
}
|
||||||
|
|
||||||
|
shutdownErr = errors.Wrapf(shutdownErr, "ttrpc: client shutting down")
|
||||||
|
|
||||||
|
c.err = shutdownErr
|
||||||
|
for _, waiter := range waiters {
|
||||||
|
waiter.errs <- shutdownErr
|
||||||
|
}
|
||||||
|
c.Close()
|
||||||
|
return
|
||||||
|
case <-c.closed:
|
||||||
|
if c.err == nil {
|
||||||
|
c.err = ErrClosed
|
||||||
|
}
|
||||||
|
// broadcast the shutdown error to the remaining waiters.
|
||||||
|
for _, waiter := range waiters {
|
||||||
|
waiter.errs <- c.err
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) send(ctx context.Context, streamID uint32, mtype messageType, msg interface{}) error {
|
||||||
|
p, err := c.codec.Marshal(msg)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return c.channel.send(ctx, streamID, mtype, p)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) recv(resp *Response, msg *message) error {
|
||||||
|
if msg.err != nil {
|
||||||
|
return msg.err
|
||||||
|
}
|
||||||
|
|
||||||
|
if msg.Type != messageTypeResponse {
|
||||||
|
return errors.New("unkown message type received")
|
||||||
|
}
|
||||||
|
|
||||||
|
defer c.channel.putmbuf(msg.p)
|
||||||
|
return proto.Unmarshal(msg.p, resp)
|
||||||
|
}
|
||||||
|
|
||||||
|
// filterCloseErr rewrites EOF and EPIPE errors to ErrClosed. Use when
|
||||||
|
// returning from call or handling errors from main read loop.
|
||||||
|
//
|
||||||
|
// This purposely ignores errors with a wrapped cause.
|
||||||
|
func filterCloseErr(err error) error {
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if err == io.EOF {
|
||||||
|
return ErrClosed
|
||||||
|
}
|
||||||
|
|
||||||
|
if strings.Contains(err.Error(), "use of closed network connection") {
|
||||||
|
return ErrClosed
|
||||||
|
}
|
||||||
|
|
||||||
|
// if we have an epipe on a write, we cast to errclosed
|
||||||
|
if oerr, ok := err.(*net.OpError); ok && oerr.Op == "write" {
|
||||||
|
if serr, ok := oerr.Err.(*os.SyscallError); ok && serr.Err == syscall.EPIPE {
|
||||||
|
return ErrClosed
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,42 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/gogo/protobuf/proto"
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
)
|
||||||
|
|
||||||
|
type codec struct{}
|
||||||
|
|
||||||
|
func (c codec) Marshal(msg interface{}) ([]byte, error) {
|
||||||
|
switch v := msg.(type) {
|
||||||
|
case proto.Message:
|
||||||
|
return proto.Marshal(v)
|
||||||
|
default:
|
||||||
|
return nil, errors.Errorf("ttrpc: cannot marshal unknown type: %T", msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c codec) Unmarshal(p []byte, msg interface{}) error {
|
||||||
|
switch v := msg.(type) {
|
||||||
|
case proto.Message:
|
||||||
|
return proto.Unmarshal(p, v)
|
||||||
|
default:
|
||||||
|
return errors.Errorf("ttrpc: cannot unmarshal into unknown type: %T", msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,39 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import "github.com/pkg/errors"
|
||||||
|
|
||||||
|
type serverConfig struct {
|
||||||
|
handshaker Handshaker
|
||||||
|
}
|
||||||
|
|
||||||
|
type ServerOpt func(*serverConfig) error
|
||||||
|
|
||||||
|
// WithServerHandshaker can be passed to NewServer to ensure that the
|
||||||
|
// handshaker is called before every connection attempt.
|
||||||
|
//
|
||||||
|
// Only one handshaker is allowed per server.
|
||||||
|
func WithServerHandshaker(handshaker Handshaker) ServerOpt {
|
||||||
|
return func(c *serverConfig) error {
|
||||||
|
if c.handshaker != nil {
|
||||||
|
return errors.New("only one handshaker allowed per server")
|
||||||
|
}
|
||||||
|
c.handshaker = handshaker
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,50 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Handshaker defines the interface for connection handshakes performed on the
|
||||||
|
// server or client when first connecting.
|
||||||
|
type Handshaker interface {
|
||||||
|
// Handshake should confirm or decorate a connection that may be incoming
|
||||||
|
// to a server or outgoing from a client.
|
||||||
|
//
|
||||||
|
// If this returns without an error, the caller should use the connection
|
||||||
|
// in place of the original connection.
|
||||||
|
//
|
||||||
|
// The second return value can contain credential specific data, such as
|
||||||
|
// unix socket credentials or TLS information.
|
||||||
|
//
|
||||||
|
// While we currently only have implementations on the server-side, this
|
||||||
|
// interface should be sufficient to implement similar handshakes on the
|
||||||
|
// client-side.
|
||||||
|
Handshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
type handshakerFunc func(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error)
|
||||||
|
|
||||||
|
func (fn handshakerFunc) Handshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error) {
|
||||||
|
return fn(ctx, conn)
|
||||||
|
}
|
||||||
|
|
||||||
|
func noopHandshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error) {
|
||||||
|
return conn, nil, nil
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,471 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"math/rand"
|
||||||
|
"net"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
"google.golang.org/grpc/codes"
|
||||||
|
"google.golang.org/grpc/status"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
ErrServerClosed = errors.New("ttrpc: server closed")
|
||||||
|
)
|
||||||
|
|
||||||
|
type Server struct {
|
||||||
|
config *serverConfig
|
||||||
|
services *serviceSet
|
||||||
|
codec codec
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
listeners map[net.Listener]struct{}
|
||||||
|
connections map[*serverConn]struct{} // all connections to current state
|
||||||
|
done chan struct{} // marks point at which we stop serving requests
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewServer(opts ...ServerOpt) (*Server, error) {
|
||||||
|
config := &serverConfig{}
|
||||||
|
for _, opt := range opts {
|
||||||
|
if err := opt(config); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return &Server{
|
||||||
|
config: config,
|
||||||
|
services: newServiceSet(),
|
||||||
|
done: make(chan struct{}),
|
||||||
|
listeners: make(map[net.Listener]struct{}),
|
||||||
|
connections: make(map[*serverConn]struct{}),
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) Register(name string, methods map[string]Method) {
|
||||||
|
s.services.register(name, methods)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) Serve(ctx context.Context, l net.Listener) error {
|
||||||
|
s.addListener(l)
|
||||||
|
defer s.closeListener(l)
|
||||||
|
|
||||||
|
var (
|
||||||
|
backoff time.Duration
|
||||||
|
handshaker = s.config.handshaker
|
||||||
|
)
|
||||||
|
|
||||||
|
if handshaker == nil {
|
||||||
|
handshaker = handshakerFunc(noopHandshake)
|
||||||
|
}
|
||||||
|
|
||||||
|
for {
|
||||||
|
conn, err := l.Accept()
|
||||||
|
if err != nil {
|
||||||
|
select {
|
||||||
|
case <-s.done:
|
||||||
|
return ErrServerClosed
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
if terr, ok := err.(interface {
|
||||||
|
Temporary() bool
|
||||||
|
}); ok && terr.Temporary() {
|
||||||
|
if backoff == 0 {
|
||||||
|
backoff = time.Millisecond
|
||||||
|
} else {
|
||||||
|
backoff *= 2
|
||||||
|
}
|
||||||
|
|
||||||
|
if max := time.Second; backoff > max {
|
||||||
|
backoff = max
|
||||||
|
}
|
||||||
|
|
||||||
|
sleep := time.Duration(rand.Int63n(int64(backoff)))
|
||||||
|
logrus.WithError(err).Errorf("ttrpc: failed accept; backoff %v", sleep)
|
||||||
|
time.Sleep(sleep)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
backoff = 0
|
||||||
|
|
||||||
|
approved, handshake, err := handshaker.Handshake(ctx, conn)
|
||||||
|
if err != nil {
|
||||||
|
logrus.WithError(err).Errorf("ttrpc: refusing connection after handshake")
|
||||||
|
conn.Close()
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
sc := s.newConn(approved, handshake)
|
||||||
|
go sc.run(ctx)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) Shutdown(ctx context.Context) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
select {
|
||||||
|
case <-s.done:
|
||||||
|
default:
|
||||||
|
// protected by mutex
|
||||||
|
close(s.done)
|
||||||
|
}
|
||||||
|
lnerr := s.closeListeners()
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
ticker := time.NewTicker(200 * time.Millisecond)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
if s.closeIdleConns() {
|
||||||
|
return lnerr
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
case <-ticker.C:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close the server without waiting for active connections.
|
||||||
|
func (s *Server) Close() error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-s.done:
|
||||||
|
default:
|
||||||
|
// protected by mutex
|
||||||
|
close(s.done)
|
||||||
|
}
|
||||||
|
|
||||||
|
err := s.closeListeners()
|
||||||
|
for c := range s.connections {
|
||||||
|
c.close()
|
||||||
|
delete(s.connections, c)
|
||||||
|
}
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) addListener(l net.Listener) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
s.listeners[l] = struct{}{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) closeListener(l net.Listener) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
return s.closeListenerLocked(l)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) closeListenerLocked(l net.Listener) error {
|
||||||
|
defer delete(s.listeners, l)
|
||||||
|
return l.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) closeListeners() error {
|
||||||
|
var err error
|
||||||
|
for l := range s.listeners {
|
||||||
|
if cerr := s.closeListenerLocked(l); cerr != nil && err == nil {
|
||||||
|
err = cerr
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) addConnection(c *serverConn) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
s.connections[c] = struct{}{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) closeIdleConns() bool {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
quiescent := true
|
||||||
|
for c := range s.connections {
|
||||||
|
st, ok := c.getState()
|
||||||
|
if !ok || st != connStateIdle {
|
||||||
|
quiescent = false
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
c.close()
|
||||||
|
delete(s.connections, c)
|
||||||
|
}
|
||||||
|
return quiescent
|
||||||
|
}
|
||||||
|
|
||||||
|
type connState int
|
||||||
|
|
||||||
|
const (
|
||||||
|
connStateActive = iota + 1 // outstanding requests
|
||||||
|
connStateIdle // no requests
|
||||||
|
connStateClosed // closed connection
|
||||||
|
)
|
||||||
|
|
||||||
|
func (cs connState) String() string {
|
||||||
|
switch cs {
|
||||||
|
case connStateActive:
|
||||||
|
return "active"
|
||||||
|
case connStateIdle:
|
||||||
|
return "idle"
|
||||||
|
case connStateClosed:
|
||||||
|
return "closed"
|
||||||
|
default:
|
||||||
|
return "unknown"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) newConn(conn net.Conn, handshake interface{}) *serverConn {
|
||||||
|
c := &serverConn{
|
||||||
|
server: s,
|
||||||
|
conn: conn,
|
||||||
|
handshake: handshake,
|
||||||
|
shutdown: make(chan struct{}),
|
||||||
|
}
|
||||||
|
c.setState(connStateIdle)
|
||||||
|
s.addConnection(c)
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
|
||||||
|
type serverConn struct {
|
||||||
|
server *Server
|
||||||
|
conn net.Conn
|
||||||
|
handshake interface{} // data from handshake, not used for now
|
||||||
|
state atomic.Value
|
||||||
|
|
||||||
|
shutdownOnce sync.Once
|
||||||
|
shutdown chan struct{} // forced shutdown, used by close
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *serverConn) getState() (connState, bool) {
|
||||||
|
cs, ok := c.state.Load().(connState)
|
||||||
|
return cs, ok
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *serverConn) setState(newstate connState) {
|
||||||
|
c.state.Store(newstate)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *serverConn) close() error {
|
||||||
|
c.shutdownOnce.Do(func() {
|
||||||
|
close(c.shutdown)
|
||||||
|
})
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *serverConn) run(sctx context.Context) {
|
||||||
|
type (
|
||||||
|
request struct {
|
||||||
|
id uint32
|
||||||
|
req *Request
|
||||||
|
}
|
||||||
|
|
||||||
|
response struct {
|
||||||
|
id uint32
|
||||||
|
resp *Response
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
ch = newChannel(c.conn)
|
||||||
|
ctx, cancel = context.WithCancel(sctx)
|
||||||
|
active int
|
||||||
|
state connState = connStateIdle
|
||||||
|
responses = make(chan response)
|
||||||
|
requests = make(chan request)
|
||||||
|
recvErr = make(chan error, 1)
|
||||||
|
shutdown = c.shutdown
|
||||||
|
done = make(chan struct{})
|
||||||
|
)
|
||||||
|
|
||||||
|
defer c.conn.Close()
|
||||||
|
defer cancel()
|
||||||
|
defer close(done)
|
||||||
|
|
||||||
|
go func(recvErr chan error) {
|
||||||
|
defer close(recvErr)
|
||||||
|
sendImmediate := func(id uint32, st *status.Status) bool {
|
||||||
|
select {
|
||||||
|
case responses <- response{
|
||||||
|
// even though we've had an invalid stream id, we send it
|
||||||
|
// back on the same stream id so the client knows which
|
||||||
|
// stream id was bad.
|
||||||
|
id: id,
|
||||||
|
resp: &Response{
|
||||||
|
Status: st.Proto(),
|
||||||
|
},
|
||||||
|
}:
|
||||||
|
return true
|
||||||
|
case <-c.shutdown:
|
||||||
|
return false
|
||||||
|
case <-done:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-c.shutdown:
|
||||||
|
return
|
||||||
|
case <-done:
|
||||||
|
return
|
||||||
|
default: // proceed
|
||||||
|
}
|
||||||
|
|
||||||
|
mh, p, err := ch.recv(ctx)
|
||||||
|
if err != nil {
|
||||||
|
status, ok := status.FromError(err)
|
||||||
|
if !ok {
|
||||||
|
recvErr <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// in this case, we send an error for that particular message
|
||||||
|
// when the status is defined.
|
||||||
|
if !sendImmediate(mh.StreamID, status) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if mh.Type != messageTypeRequest {
|
||||||
|
// we must ignore this for future compat.
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
var req Request
|
||||||
|
if err := c.server.codec.Unmarshal(p, &req); err != nil {
|
||||||
|
ch.putmbuf(p)
|
||||||
|
if !sendImmediate(mh.StreamID, status.Newf(codes.InvalidArgument, "unmarshal request error: %v", err)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
ch.putmbuf(p)
|
||||||
|
|
||||||
|
if mh.StreamID%2 != 1 {
|
||||||
|
// enforce odd client initiated identifiers.
|
||||||
|
if !sendImmediate(mh.StreamID, status.Newf(codes.InvalidArgument, "StreamID must be odd for client initiated streams")) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Forward the request to the main loop. We don't wait on s.done
|
||||||
|
// because we have already accepted the client request.
|
||||||
|
select {
|
||||||
|
case requests <- request{
|
||||||
|
id: mh.StreamID,
|
||||||
|
req: &req,
|
||||||
|
}:
|
||||||
|
case <-done:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}(recvErr)
|
||||||
|
|
||||||
|
for {
|
||||||
|
newstate := state
|
||||||
|
switch {
|
||||||
|
case active > 0:
|
||||||
|
newstate = connStateActive
|
||||||
|
shutdown = nil
|
||||||
|
case active == 0:
|
||||||
|
newstate = connStateIdle
|
||||||
|
shutdown = c.shutdown // only enable this branch in idle mode
|
||||||
|
}
|
||||||
|
|
||||||
|
if newstate != state {
|
||||||
|
c.setState(newstate)
|
||||||
|
state = newstate
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case request := <-requests:
|
||||||
|
active++
|
||||||
|
go func(id uint32) {
|
||||||
|
ctx, cancel := getRequestContext(ctx, request.req)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
p, status := c.server.services.call(ctx, request.req.Service, request.req.Method, request.req.Payload)
|
||||||
|
resp := &Response{
|
||||||
|
Status: status.Proto(),
|
||||||
|
Payload: p,
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case responses <- response{
|
||||||
|
id: id,
|
||||||
|
resp: resp,
|
||||||
|
}:
|
||||||
|
case <-done:
|
||||||
|
}
|
||||||
|
}(request.id)
|
||||||
|
case response := <-responses:
|
||||||
|
p, err := c.server.codec.Marshal(response.resp)
|
||||||
|
if err != nil {
|
||||||
|
logrus.WithError(err).Error("failed marshaling response")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := ch.send(ctx, response.id, messageTypeResponse, p); err != nil {
|
||||||
|
logrus.WithError(err).Error("failed sending message on channel")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
active--
|
||||||
|
case err := <-recvErr:
|
||||||
|
// TODO(stevvooe): Not wildly clear what we should do in this
|
||||||
|
// branch. Basically, it means that we are no longer receiving
|
||||||
|
// requests due to a terminal error.
|
||||||
|
recvErr = nil // connection is now "closing"
|
||||||
|
if err != nil && err != io.EOF {
|
||||||
|
logrus.WithError(err).Error("error receiving message")
|
||||||
|
}
|
||||||
|
case <-shutdown:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var noopFunc = func() {}
|
||||||
|
|
||||||
|
func getRequestContext(ctx context.Context, req *Request) (retCtx context.Context, cancel func()) {
|
||||||
|
cancel = noopFunc
|
||||||
|
if req.TimeoutNano == 0 {
|
||||||
|
return ctx, cancel
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx, cancel = context.WithTimeout(ctx, time.Duration(req.TimeoutNano))
|
||||||
|
return ctx, cancel
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,150 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"os"
|
||||||
|
"path"
|
||||||
|
|
||||||
|
"github.com/gogo/protobuf/proto"
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"google.golang.org/grpc/codes"
|
||||||
|
"google.golang.org/grpc/status"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Method func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error)
|
||||||
|
|
||||||
|
type ServiceDesc struct {
|
||||||
|
Methods map[string]Method
|
||||||
|
|
||||||
|
// TODO(stevvooe): Add stream support.
|
||||||
|
}
|
||||||
|
|
||||||
|
type serviceSet struct {
|
||||||
|
services map[string]ServiceDesc
|
||||||
|
}
|
||||||
|
|
||||||
|
func newServiceSet() *serviceSet {
|
||||||
|
return &serviceSet{
|
||||||
|
services: make(map[string]ServiceDesc),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *serviceSet) register(name string, methods map[string]Method) {
|
||||||
|
if _, ok := s.services[name]; ok {
|
||||||
|
panic(errors.Errorf("duplicate service %v registered", name))
|
||||||
|
}
|
||||||
|
|
||||||
|
s.services[name] = ServiceDesc{
|
||||||
|
Methods: methods,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *serviceSet) call(ctx context.Context, serviceName, methodName string, p []byte) ([]byte, *status.Status) {
|
||||||
|
p, err := s.dispatch(ctx, serviceName, methodName, p)
|
||||||
|
st, ok := status.FromError(err)
|
||||||
|
if !ok {
|
||||||
|
st = status.New(convertCode(err), err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
return p, st
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *serviceSet) dispatch(ctx context.Context, serviceName, methodName string, p []byte) ([]byte, error) {
|
||||||
|
method, err := s.resolve(serviceName, methodName)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
unmarshal := func(obj interface{}) error {
|
||||||
|
switch v := obj.(type) {
|
||||||
|
case proto.Message:
|
||||||
|
if err := proto.Unmarshal(p, v); err != nil {
|
||||||
|
return status.Errorf(codes.Internal, "ttrpc: error unmarshaling payload: %v", err.Error())
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return status.Errorf(codes.Internal, "ttrpc: error unsupported request type: %T", v)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := method(ctx, unmarshal)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
switch v := resp.(type) {
|
||||||
|
case proto.Message:
|
||||||
|
r, err := proto.Marshal(v)
|
||||||
|
if err != nil {
|
||||||
|
return nil, status.Errorf(codes.Internal, "ttrpc: error marshaling payload: %v", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
return r, nil
|
||||||
|
default:
|
||||||
|
return nil, status.Errorf(codes.Internal, "ttrpc: error unsupported response type: %T", v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *serviceSet) resolve(service, method string) (Method, error) {
|
||||||
|
srv, ok := s.services[service]
|
||||||
|
if !ok {
|
||||||
|
return nil, status.Errorf(codes.NotFound, "service %v", service)
|
||||||
|
}
|
||||||
|
|
||||||
|
mthd, ok := srv.Methods[method]
|
||||||
|
if !ok {
|
||||||
|
return nil, status.Errorf(codes.NotFound, "method %v", method)
|
||||||
|
}
|
||||||
|
|
||||||
|
return mthd, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// convertCode maps stdlib go errors into grpc space.
|
||||||
|
//
|
||||||
|
// This is ripped from the grpc-go code base.
|
||||||
|
func convertCode(err error) codes.Code {
|
||||||
|
switch err {
|
||||||
|
case nil:
|
||||||
|
return codes.OK
|
||||||
|
case io.EOF:
|
||||||
|
return codes.OutOfRange
|
||||||
|
case io.ErrClosedPipe, io.ErrNoProgress, io.ErrShortBuffer, io.ErrShortWrite, io.ErrUnexpectedEOF:
|
||||||
|
return codes.FailedPrecondition
|
||||||
|
case os.ErrInvalid:
|
||||||
|
return codes.InvalidArgument
|
||||||
|
case context.Canceled:
|
||||||
|
return codes.Canceled
|
||||||
|
case context.DeadlineExceeded:
|
||||||
|
return codes.DeadlineExceeded
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case os.IsExist(err):
|
||||||
|
return codes.AlreadyExists
|
||||||
|
case os.IsNotExist(err):
|
||||||
|
return codes.NotFound
|
||||||
|
case os.IsPermission(err):
|
||||||
|
return codes.PermissionDenied
|
||||||
|
}
|
||||||
|
return codes.Unknown
|
||||||
|
}
|
||||||
|
|
||||||
|
func fullPath(service, method string) string {
|
||||||
|
return "/" + path.Join("/", service, method)
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
spb "google.golang.org/genproto/googleapis/rpc/status"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Request struct {
|
||||||
|
Service string `protobuf:"bytes,1,opt,name=service,proto3"`
|
||||||
|
Method string `protobuf:"bytes,2,opt,name=method,proto3"`
|
||||||
|
Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3"`
|
||||||
|
TimeoutNano int64 `protobuf:"varint,4,opt,name=timeout_nano,proto3"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *Request) Reset() { *r = Request{} }
|
||||||
|
func (r *Request) String() string { return fmt.Sprintf("%+#v", r) }
|
||||||
|
func (r *Request) ProtoMessage() {}
|
||||||
|
|
||||||
|
type Response struct {
|
||||||
|
Status *spb.Status `protobuf:"bytes,1,opt,name=status,proto3"`
|
||||||
|
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *Response) Reset() { *r = Response{} }
|
||||||
|
func (r *Response) String() string { return fmt.Sprintf("%+#v", r) }
|
||||||
|
func (r *Response) ProtoMessage() {}
|
||||||
|
|
@ -0,0 +1,108 @@
|
||||||
|
/*
|
||||||
|
Copyright The containerd Authors.
|
||||||
|
|
||||||
|
Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
you may not use this file except in compliance with the License.
|
||||||
|
You may obtain a copy of the License at
|
||||||
|
|
||||||
|
http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
|
||||||
|
Unless required by applicable law or agreed to in writing, software
|
||||||
|
distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
See the License for the specific language governing permissions and
|
||||||
|
limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package ttrpc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net"
|
||||||
|
"os"
|
||||||
|
"syscall"
|
||||||
|
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"golang.org/x/sys/unix"
|
||||||
|
)
|
||||||
|
|
||||||
|
type UnixCredentialsFunc func(*unix.Ucred) error
|
||||||
|
|
||||||
|
func (fn UnixCredentialsFunc) Handshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error) {
|
||||||
|
uc, err := requireUnixSocket(conn)
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, errors.Wrap(err, "ttrpc.UnixCredentialsFunc: require unix socket")
|
||||||
|
}
|
||||||
|
|
||||||
|
rs, err := uc.SyscallConn()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, errors.Wrap(err, "ttrpc.UnixCredentialsFunc: (net.UnixConn).SyscallConn failed")
|
||||||
|
}
|
||||||
|
var (
|
||||||
|
ucred *unix.Ucred
|
||||||
|
ucredErr error
|
||||||
|
)
|
||||||
|
if err := rs.Control(func(fd uintptr) {
|
||||||
|
ucred, ucredErr = unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED)
|
||||||
|
}); err != nil {
|
||||||
|
return nil, nil, errors.Wrapf(err, "ttrpc.UnixCredentialsFunc: (*syscall.RawConn).Control failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
if ucredErr != nil {
|
||||||
|
return nil, nil, errors.Wrapf(err, "ttrpc.UnixCredentialsFunc: failed to retrieve socket peer credentials")
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := fn(ucred); err != nil {
|
||||||
|
return nil, nil, errors.Wrapf(err, "ttrpc.UnixCredentialsFunc: credential check failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
return uc, ucred, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// UnixSocketRequireUidGid requires specific *effective* UID/GID, rather than the real UID/GID.
|
||||||
|
//
|
||||||
|
// For example, if a daemon binary is owned by the root (UID 0) with SUID bit but running as an
|
||||||
|
// unprivileged user (UID 1001), the effective UID becomes 0, and the real UID becomes 1001.
|
||||||
|
// So calling this function with uid=0 allows a connection from effective UID 0 but rejects
|
||||||
|
// a connection from effective UID 1001.
|
||||||
|
//
|
||||||
|
// See socket(7), SO_PEERCRED: "The returned credentials are those that were in effect at the time of the call to connect(2) or socketpair(2)."
|
||||||
|
func UnixSocketRequireUidGid(uid, gid int) UnixCredentialsFunc {
|
||||||
|
return func(ucred *unix.Ucred) error {
|
||||||
|
return requireUidGid(ucred, uid, gid)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func UnixSocketRequireRoot() UnixCredentialsFunc {
|
||||||
|
return UnixSocketRequireUidGid(0, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
// UnixSocketRequireSameUser resolves the current effective unix user and returns a
|
||||||
|
// UnixCredentialsFunc that will validate incoming unix connections against the
|
||||||
|
// current credentials.
|
||||||
|
//
|
||||||
|
// This is useful when using abstract sockets that are accessible by all users.
|
||||||
|
func UnixSocketRequireSameUser() UnixCredentialsFunc {
|
||||||
|
euid, egid := os.Geteuid(), os.Getegid()
|
||||||
|
return UnixSocketRequireUidGid(euid, egid)
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireRoot(ucred *unix.Ucred) error {
|
||||||
|
return requireUidGid(ucred, 0, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireUidGid(ucred *unix.Ucred, uid, gid int) error {
|
||||||
|
if (uid != -1 && uint32(uid) != ucred.Uid) || (gid != -1 && uint32(gid) != ucred.Gid) {
|
||||||
|
return errors.Wrap(syscall.EPERM, "ttrpc: invalid credentials")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireUnixSocket(conn net.Conn) (*net.UnixConn, error) {
|
||||||
|
uc, ok := conn.(*net.UnixConn)
|
||||||
|
if !ok {
|
||||||
|
return nil, errors.New("a unix socket connection is required")
|
||||||
|
}
|
||||||
|
|
||||||
|
return uc, nil
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue