-
Notifications
You must be signed in to change notification settings - Fork 4.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
server: wire up in-process Resource Service (#16978)
- Loading branch information
Showing
5 changed files
with
230 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| // Copyright (c) HashiCorp, Inc. | ||
| // SPDX-License-Identifier: MPL-2.0 | ||
|
|
||
| package resolver | ||
|
|
||
| import "github.com/hashicorp/consul/acl" | ||
|
|
||
| // DANGER_NO_AUTH implements an ACL resolver short-circuit authorization in | ||
| // cases where it is handled somewhere else or expressly not required. | ||
| type DANGER_NO_AUTH struct{} | ||
|
|
||
| // ResolveTokenAndDefaultMeta returns an authorizer with unfettered permissions. | ||
| func (DANGER_NO_AUTH) ResolveTokenAndDefaultMeta(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) (Result, error) { | ||
| return Result{Authorizer: acl.ManageAll()}, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| // Copyright (c) HashiCorp, Inc. | ||
| // SPDX-License-Identifier: MPL-2.0 | ||
|
|
||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "net" | ||
| "sync/atomic" | ||
| ) | ||
|
|
||
| // ErrPipeClosed is returned when calling Accept or DialContext on a closed | ||
| // PipeListener. | ||
| var ErrPipeClosed = errors.New("pipe listener has been closed") | ||
|
|
||
| // PipeListener implements the net.Listener interface using a net.Pipe so that | ||
| // you can interact with a gRPC service in the same process without going over | ||
| // the network. | ||
| type PipeListener struct { | ||
| conns chan net.Conn | ||
| closed atomic.Bool | ||
| done chan struct{} | ||
| } | ||
|
|
||
| var _ net.Listener = (*PipeListener)(nil) | ||
|
|
||
| // NewPipeListener creates a new PipeListener. | ||
| func NewPipeListener() *PipeListener { | ||
| return &PipeListener{ | ||
| conns: make(chan net.Conn), | ||
| done: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| // Accept a connection. | ||
| func (p *PipeListener) Accept() (net.Conn, error) { | ||
| select { | ||
| case conn := <-p.conns: | ||
| return conn, nil | ||
| case <-p.done: | ||
| return nil, ErrPipeClosed | ||
| } | ||
| } | ||
|
|
||
| // Close the listener. | ||
| func (p *PipeListener) Close() error { | ||
| if p.closed.CompareAndSwap(false, true) { | ||
| close(p.done) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // DialContext dials the server over an in-process pipe. | ||
| func (p *PipeListener) DialContext(ctx context.Context, _ string) (net.Conn, error) { | ||
| if p.closed.Load() { | ||
| return nil, ErrPipeClosed | ||
| } | ||
|
|
||
| serverConn, clientConn := net.Pipe() | ||
|
|
||
| select { | ||
| // Send the server connection to whatever is accepting connections from the | ||
| // PipeListener. This will block until something has accepted the conn. | ||
| case p.conns <- serverConn: | ||
| return clientConn, nil | ||
| case <-ctx.Done(): | ||
| serverConn.Close() | ||
| clientConn.Close() | ||
| return nil, ctx.Err() | ||
| case <-p.done: | ||
| serverConn.Close() | ||
| clientConn.Close() | ||
| return nil, ErrPipeClosed | ||
| } | ||
| } | ||
|
|
||
| // Add returns the listener's address. | ||
| func (*PipeListener) Addr() net.Addr { return pipeAddr{} } | ||
|
|
||
| type pipeAddr struct{} | ||
|
|
||
| func (pipeAddr) Network() string { return "pipe" } | ||
| func (pipeAddr) String() string { return "pipe" } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // Copyright (c) HashiCorp, Inc. | ||
| // SPDX-License-Identifier: MPL-2.0 | ||
|
|
||
| package internal | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "context" | ||
| "net" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestPipeListener_RoundTrip(t *testing.T) { | ||
| lis := NewPipeListener() | ||
| t.Cleanup(func() { _ = lis.Close() }) | ||
|
|
||
| go echoServer(lis) | ||
|
|
||
| conn, err := lis.DialContext(context.Background(), "") | ||
| require.NoError(t, err) | ||
| t.Cleanup(func() { _ = conn.Close() }) | ||
|
|
||
| input := []byte("Hello World\n") | ||
| _, err = conn.Write(input) | ||
| require.NoError(t, err) | ||
|
|
||
| output := make([]byte, len(input)) | ||
| _, err = conn.Read(output) | ||
| require.NoError(t, err) | ||
|
|
||
| require.Equal(t, string(input), string(output)) | ||
| } | ||
|
|
||
| func TestPipeListener_Closed(t *testing.T) { | ||
| lis := NewPipeListener() | ||
| require.NoError(t, lis.Close()) | ||
|
|
||
| _, err := lis.Accept() | ||
| require.ErrorIs(t, err, ErrPipeClosed) | ||
|
|
||
| _, err = lis.DialContext(context.Background(), "") | ||
| require.ErrorIs(t, err, ErrPipeClosed) | ||
| } | ||
|
|
||
| func echoServer(lis net.Listener) { | ||
| handleConn := func(conn net.Conn) { | ||
| defer conn.Close() | ||
|
|
||
| reader := bufio.NewReader(conn) | ||
| for { | ||
| msg, err := reader.ReadBytes('\n') | ||
| if err != nil { | ||
| return | ||
| } | ||
| if _, err := conn.Write(msg); err != nil { | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| for { | ||
| conn, err := lis.Accept() | ||
| if err != nil { | ||
| return | ||
| } | ||
| go handleConn(conn) | ||
| } | ||
| } |