Reverts the following:
Revert "TUN-8621: Fix cloudflared version in change notes." Revert "PPIP-2310: Update quick tunnel disclaimer" Revert "TUN-8621: Prevent QUIC connection from closing before grace period after unregistering" Revert "TUN-8484: Print response when QuickTunnel can't be unmarshalled" Revert "TUN-8592: Use metadata from the edge to determine if request body is empty for QUIC transport"
This commit is contained in:
parent
ec07269122
commit
2437675c04
|
@ -1,7 +1,3 @@
|
||||||
## 2024.9.0
|
|
||||||
### Bug Fixes
|
|
||||||
- We fixed a bug related to `--grace-period`. Tunnels that use QUIC as transport weren't abiding by this waiting period before forcefully closing the connections to the edge. From now on, both QUIC and HTTP2 tunnels will wait for either the grace period to end (defaults to 30 seconds) or until the last in-flight request is handled. Users that wish to maintain the previous behavior should set `--grace-period` to 0 if `--protocol` is set to `quic`. This will force `cloudflared` to shutdown as soon as either SIGTERM or SIGINT is received.
|
|
||||||
|
|
||||||
## 2024.2.1
|
## 2024.2.1
|
||||||
### Notices
|
### Notices
|
||||||
- Starting from this version, tunnel diagnostics will be enabled by default. This will allow the engineering team to remotely get diagnostics from cloudflared during debug activities. Users still have the capability to opt-out of this feature by defining `--management-diagnostics=false` (or env `TUNNEL_MANAGEMENT_DIAGNOSTICS`).
|
- Starting from this version, tunnel diagnostics will be enabled by default. This will allow the engineering team to remotely get diagnostics from cloudflared during debug activities. Users still have the capability to opt-out of this feature by defining `--management-diagnostics=false` (or env `TUNNEL_MANAGEMENT_DIAGNOSTICS`).
|
||||||
|
|
|
@ -3,7 +3,6 @@ package tunnel
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
@ -16,7 +15,10 @@ import (
|
||||||
|
|
||||||
const httpTimeout = 15 * time.Second
|
const httpTimeout = 15 * time.Second
|
||||||
|
|
||||||
const disclaimer = "Thank you for trying Cloudflare Tunnel. Doing so, without a Cloudflare account, is a quick way to experiment and try it out. However, be aware that these account-less Tunnels have no uptime guarantee, are subject to the Cloudflare Online Services Terms of Use (https://www.cloudflare.com/website-terms/), and Cloudflare reserves the right to investigate your use of Tunnels for violations of such terms. If you intend to use Tunnels in production you should use a pre-created named tunnel by following: https://developers.cloudflare.com/cloudflare-one/connections/connect-apps"
|
const disclaimer = "Thank you for trying Cloudflare Tunnel. Doing so, without a Cloudflare account, is a quick way to" +
|
||||||
|
" experiment and try it out. However, be aware that these account-less Tunnels have no uptime guarantee. If you " +
|
||||||
|
"intend to use Tunnels in production you should use a pre-created named tunnel by following: " +
|
||||||
|
"https://developers.cloudflare.com/cloudflare-one/connections/connect-apps"
|
||||||
|
|
||||||
// RunQuickTunnel requests a tunnel from the specified service.
|
// RunQuickTunnel requests a tunnel from the specified service.
|
||||||
// We use this to power quick tunnels on trycloudflare.com, but the
|
// We use this to power quick tunnels on trycloudflare.com, but the
|
||||||
|
@ -45,17 +47,8 @@ func RunQuickTunnel(sc *subcommandContext) error {
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
|
|
||||||
// This will read the entire response into memory so we can print it in case of error
|
|
||||||
rsp_body, err := io.ReadAll(resp.Body)
|
|
||||||
if err != nil {
|
|
||||||
return errors.Wrap(err, "failed to read quick-tunnel response")
|
|
||||||
}
|
|
||||||
|
|
||||||
var data QuickTunnelResponse
|
var data QuickTunnelResponse
|
||||||
if err := json.Unmarshal(rsp_body, &data); err != nil {
|
if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
|
||||||
rsp_string := string(rsp_body)
|
|
||||||
fields := map[string]interface{}{"status_code": resp.Status}
|
|
||||||
sc.log.Err(err).Fields(fields).Msgf("Error unmarshaling QuickTunnel response: %s", rsp_string)
|
|
||||||
return errors.Wrap(err, "failed to unmarshal quick Tunnel")
|
return errors.Wrap(err, "failed to unmarshal quick Tunnel")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -45,7 +45,6 @@ class TestTermination:
|
||||||
with connected:
|
with connected:
|
||||||
connected.wait(self.timeout)
|
connected.wait(self.timeout)
|
||||||
# Send signal after the SSE connection is established
|
# Send signal after the SSE connection is established
|
||||||
with self.within_grace_period():
|
|
||||||
self.terminate_by_signal(cloudflared, signal)
|
self.terminate_by_signal(cloudflared, signal)
|
||||||
self.wait_eyeball_thread(
|
self.wait_eyeball_thread(
|
||||||
in_flight_req, self.grace_period + self.timeout)
|
in_flight_req, self.grace_period + self.timeout)
|
||||||
|
@ -67,7 +66,7 @@ class TestTermination:
|
||||||
|
|
||||||
with connected:
|
with connected:
|
||||||
connected.wait(self.timeout)
|
connected.wait(self.timeout)
|
||||||
with self.within_grace_period(has_connection=False):
|
with self.within_grace_period():
|
||||||
# Send signal after the SSE connection is established
|
# Send signal after the SSE connection is established
|
||||||
self.terminate_by_signal(cloudflared, signal)
|
self.terminate_by_signal(cloudflared, signal)
|
||||||
self.wait_eyeball_thread(in_flight_req, self.grace_period)
|
self.wait_eyeball_thread(in_flight_req, self.grace_period)
|
||||||
|
@ -79,7 +78,7 @@ class TestTermination:
|
||||||
with start_cloudflared(
|
with start_cloudflared(
|
||||||
tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1"], new_process=True, capture_output=False) as cloudflared:
|
tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1"], new_process=True, capture_output=False) as cloudflared:
|
||||||
wait_tunnel_ready(tunnel_url=config.get_url())
|
wait_tunnel_ready(tunnel_url=config.get_url())
|
||||||
with self.within_grace_period(has_connection=False):
|
with self.within_grace_period():
|
||||||
self.terminate_by_signal(cloudflared, signal)
|
self.terminate_by_signal(cloudflared, signal)
|
||||||
|
|
||||||
def terminate_by_signal(self, cloudflared, sig):
|
def terminate_by_signal(self, cloudflared, sig):
|
||||||
|
@ -93,21 +92,13 @@ class TestTermination:
|
||||||
|
|
||||||
# Using this context asserts logic within the context is executed within grace period
|
# Using this context asserts logic within the context is executed within grace period
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def within_grace_period(self, has_connection=True):
|
def within_grace_period(self):
|
||||||
try:
|
try:
|
||||||
start = time.time()
|
start = time.time()
|
||||||
yield
|
yield
|
||||||
finally:
|
finally:
|
||||||
|
|
||||||
# If the request takes longer than the grace period then we need to wait at most the grace period.
|
|
||||||
# If the request fell within the grace period cloudflared can close earlier, but to ensure that it doesn't
|
|
||||||
# close immediately we add a minimum boundary. If cloudflared shutdown in less than 1s it's likely that
|
|
||||||
# it shutdown as soon as it received SIGINT. The only way cloudflared can close immediately is if it has no
|
|
||||||
# in-flight requests
|
|
||||||
minimum = 1 if has_connection else 0
|
|
||||||
duration = time.time() - start
|
duration = time.time() - start
|
||||||
# Here we truncate to ensure that we don't fail on minute differences like 10.1 instead of 10
|
assert duration < self.grace_period
|
||||||
assert minimum <= int(duration) <= self.grace_period
|
|
||||||
|
|
||||||
def stream_request(self, config, connected, early_terminate):
|
def stream_request(self, config, connected, early_terminate):
|
||||||
expected_terminate_message = "502 Bad Gateway"
|
expected_terminate_message = "502 Bad Gateway"
|
||||||
|
|
|
@ -6,8 +6,6 @@ import (
|
||||||
"net"
|
"net"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/pkg/errors"
|
|
||||||
|
|
||||||
"github.com/cloudflare/cloudflared/management"
|
"github.com/cloudflare/cloudflared/management"
|
||||||
"github.com/cloudflare/cloudflared/tunnelrpc"
|
"github.com/cloudflare/cloudflared/tunnelrpc"
|
||||||
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
|
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
|
||||||
|
@ -118,32 +116,27 @@ func (c *controlStream) ServeControlStream(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.waitForUnregister(ctx, registrationClient)
|
c.waitForUnregister(ctx, registrationClient)
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *controlStream) waitForUnregister(ctx context.Context, registrationClient tunnelrpc.RegistrationClient) error {
|
func (c *controlStream) waitForUnregister(ctx context.Context, registrationClient tunnelrpc.RegistrationClient) {
|
||||||
// wait for connection termination or start of graceful shutdown
|
// wait for connection termination or start of graceful shutdown
|
||||||
defer registrationClient.Close()
|
defer registrationClient.Close()
|
||||||
var shutdownError error
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
shutdownError = ctx.Err()
|
|
||||||
break
|
break
|
||||||
case <-c.gracefulShutdownC:
|
case <-c.gracefulShutdownC:
|
||||||
c.stoppedGracefully = true
|
c.stoppedGracefully = true
|
||||||
}
|
}
|
||||||
|
|
||||||
c.observer.sendUnregisteringEvent(c.connIndex)
|
c.observer.sendUnregisteringEvent(c.connIndex)
|
||||||
err := registrationClient.GracefulShutdown(ctx, c.gracePeriod)
|
registrationClient.GracefulShutdown(ctx, c.gracePeriod)
|
||||||
if err != nil {
|
|
||||||
return errors.Wrap(err, "Error shutting down control stream")
|
|
||||||
}
|
|
||||||
c.observer.log.Info().
|
c.observer.log.Info().
|
||||||
Int(management.EventTypeKey, int(management.Cloudflared)).
|
Int(management.EventTypeKey, int(management.Cloudflared)).
|
||||||
Uint8(LogFieldConnIndex, c.connIndex).
|
Uint8(LogFieldConnIndex, c.connIndex).
|
||||||
IPAddr(LogFieldIPAddress, c.edgeAddress).
|
IPAddr(LogFieldIPAddress, c.edgeAddress).
|
||||||
Msg("Unregistered tunnel connection")
|
Msg("Unregistered tunnel connection")
|
||||||
return shutdownError
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *controlStream) IsStopped() bool {
|
func (c *controlStream) IsStopped() bool {
|
||||||
|
|
|
@ -192,9 +192,8 @@ func (mc mockNamedTunnelRPCClient) RegisterConnection(
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) error {
|
func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
|
||||||
close(mc.unregistered)
|
close(mc.unregistered)
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mockNamedTunnelRPCClient) Close() {}
|
func (mockNamedTunnelRPCClient) Close() {}
|
||||||
|
|
|
@ -42,26 +42,12 @@ const (
|
||||||
HTTPMethodKey = "HttpMethod"
|
HTTPMethodKey = "HttpMethod"
|
||||||
// HTTPHostKey is used to get or set http Method in QUIC ALPN if the underlying proxy connection type is HTTP.
|
// HTTPHostKey is used to get or set http Method in QUIC ALPN if the underlying proxy connection type is HTTP.
|
||||||
HTTPHostKey = "HttpHost"
|
HTTPHostKey = "HttpHost"
|
||||||
// HTTPRequestBodyHintKey is used in ConnectRequest metadata to indicate if the request has body
|
|
||||||
HTTPRequestBodyHintKey = "HttpReqBodyHint"
|
|
||||||
|
|
||||||
QUICMetadataFlowID = "FlowID"
|
QUICMetadataFlowID = "FlowID"
|
||||||
// emperically this capacity has been working well
|
// emperically this capacity has been working well
|
||||||
demuxChanCapacity = 16
|
demuxChanCapacity = 16
|
||||||
)
|
)
|
||||||
|
|
||||||
type RequestBodyHint uint64
|
|
||||||
|
|
||||||
const (
|
|
||||||
RequestBodyHintMissing RequestBodyHint = iota
|
|
||||||
RequestBodyHintEmpty
|
|
||||||
RequestBodyHintHasData
|
|
||||||
)
|
|
||||||
|
|
||||||
func (rbh RequestBodyHint) String() string {
|
|
||||||
return [...]string{"missing", "empty", "data"}[rbh]
|
|
||||||
}
|
|
||||||
|
|
||||||
var (
|
var (
|
||||||
portForConnIndex = make(map[uint8]int, 0)
|
portForConnIndex = make(map[uint8]int, 0)
|
||||||
portMapMutex sync.Mutex
|
portMapMutex sync.Mutex
|
||||||
|
@ -83,7 +69,6 @@ type QUICConnection struct {
|
||||||
|
|
||||||
rpcTimeout time.Duration
|
rpcTimeout time.Duration
|
||||||
streamWriteTimeout time.Duration
|
streamWriteTimeout time.Duration
|
||||||
gracePeriod time.Duration
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewQUICConnection returns a new instance of QUICConnection.
|
// NewQUICConnection returns a new instance of QUICConnection.
|
||||||
|
@ -101,7 +86,6 @@ func NewQUICConnection(
|
||||||
packetRouterConfig *ingress.GlobalRouterConfig,
|
packetRouterConfig *ingress.GlobalRouterConfig,
|
||||||
rpcTimeout time.Duration,
|
rpcTimeout time.Duration,
|
||||||
streamWriteTimeout time.Duration,
|
streamWriteTimeout time.Duration,
|
||||||
gracePeriod time.Duration,
|
|
||||||
) (*QUICConnection, error) {
|
) (*QUICConnection, error) {
|
||||||
udpConn, err := createUDPConnForConnIndex(connIndex, localAddr, logger)
|
udpConn, err := createUDPConnForConnIndex(connIndex, localAddr, logger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -138,7 +122,6 @@ func NewQUICConnection(
|
||||||
connIndex: connIndex,
|
connIndex: connIndex,
|
||||||
rpcTimeout: rpcTimeout,
|
rpcTimeout: rpcTimeout,
|
||||||
streamWriteTimeout: streamWriteTimeout,
|
streamWriteTimeout: streamWriteTimeout,
|
||||||
gracePeriod: gracePeriod,
|
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -161,17 +144,8 @@ func (q *QUICConnection) Serve(ctx context.Context) error {
|
||||||
// In the future, if cloudflared can autonomously push traffic to the edge, we have to make sure the control
|
// In the future, if cloudflared can autonomously push traffic to the edge, we have to make sure the control
|
||||||
// stream is already fully registered before the other goroutines can proceed.
|
// stream is already fully registered before the other goroutines can proceed.
|
||||||
errGroup.Go(func() error {
|
errGroup.Go(func() error {
|
||||||
// err is equal to nil if we exit due to unregistration. If that happens we want to wait the full
|
defer cancel()
|
||||||
// amount of the grace period, allowing requests to finish before we cancel the context, which will
|
return q.serveControlStream(ctx, controlStream)
|
||||||
// make cloudflared exit.
|
|
||||||
if err := q.serveControlStream(ctx, controlStream); err == nil {
|
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
|
||||||
case <-time.Tick(q.gracePeriod):
|
|
||||||
}
|
|
||||||
}
|
|
||||||
cancel()
|
|
||||||
return err
|
|
||||||
})
|
})
|
||||||
errGroup.Go(func() error {
|
errGroup.Go(func() error {
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
@ -500,6 +474,7 @@ func buildHTTPRequest(
|
||||||
dest := connectRequest.Dest
|
dest := connectRequest.Dest
|
||||||
method := metadata[HTTPMethodKey]
|
method := metadata[HTTPMethodKey]
|
||||||
host := metadata[HTTPHostKey]
|
host := metadata[HTTPHostKey]
|
||||||
|
isWebsocket := connectRequest.Type == pogs.ConnectionTypeWebsocket
|
||||||
|
|
||||||
req, err := http.NewRequestWithContext(ctx, method, dest, body)
|
req, err := http.NewRequestWithContext(ctx, method, dest, body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -524,8 +499,13 @@ func buildHTTPRequest(
|
||||||
return nil, fmt.Errorf("Error setting content-length: %w", err)
|
return nil, fmt.Errorf("Error setting content-length: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if shouldSetRequestBodyToEmpty(connectRequest, metadata, req) {
|
// Go's client defaults to chunked encoding after a 200ms delay if the following cases are true:
|
||||||
log.Debug().Str("host", req.Host).Str("method", req.Method).Msg("Set request to have no body")
|
// * the request body blocks
|
||||||
|
// * the content length is not set (or set to -1)
|
||||||
|
// * the method doesn't usually have a body (GET, HEAD, DELETE, ...)
|
||||||
|
// * there is no transfer-encoding=chunked already set.
|
||||||
|
// So, if transfer cannot be chunked and content length is 0, we dont set a request body.
|
||||||
|
if !isWebsocket && !isTransferEncodingChunked(req) && req.ContentLength == 0 {
|
||||||
req.Body = http.NoBody
|
req.Body = http.NoBody
|
||||||
}
|
}
|
||||||
stripWebsocketUpgradeHeader(req)
|
stripWebsocketUpgradeHeader(req)
|
||||||
|
@ -550,35 +530,6 @@ func isTransferEncodingChunked(req *http.Request) bool {
|
||||||
return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
|
return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Borrowed from https://github.com/golang/go/blob/go1.22.6/src/net/http/request.go#L1541
|
|
||||||
func requestMethodUsuallyLacksBody(req *http.Request) bool {
|
|
||||||
switch strings.ToUpper(req.Method) {
|
|
||||||
case "GET", "HEAD", "DELETE", "OPTIONS", "PROPFIND", "SEARCH":
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
func shouldSetRequestBodyToEmpty(connectRequest *pogs.ConnectRequest, metadata map[string]string, req *http.Request) bool {
|
|
||||||
switch metadata[HTTPRequestBodyHintKey] {
|
|
||||||
case RequestBodyHintEmpty.String():
|
|
||||||
return true
|
|
||||||
case RequestBodyHintHasData.String():
|
|
||||||
return false
|
|
||||||
default:
|
|
||||||
}
|
|
||||||
|
|
||||||
isWebsocket := connectRequest.Type == pogs.ConnectionTypeWebsocket
|
|
||||||
// Go's client defaults to chunked encoding after a 200ms delay if the following cases are true:
|
|
||||||
// * the request body blocks
|
|
||||||
// * the content length is not set (or set to -1)
|
|
||||||
// * the method doesn't usually have a body (GET, HEAD, DELETE, ...)
|
|
||||||
// * there is no transfer-encoding=chunked already set.
|
|
||||||
// So, if transfer cannot be chunked and content length is 0, we dont set a request body.
|
|
||||||
// Reference: https://github.com/golang/go/blob/go1.22.2/src/net/http/transfer.go#L192-L206
|
|
||||||
return !isWebsocket && requestMethodUsuallyLacksBody(req) && !isTransferEncodingChunked(req) && req.ContentLength == 0
|
|
||||||
}
|
|
||||||
|
|
||||||
// A helper struct that guarantees a call to close only affects read side, but not write side.
|
// A helper struct that guarantees a call to close only affects read side, but not write side.
|
||||||
type nopCloserReadWriter struct {
|
type nopCloserReadWriter struct {
|
||||||
io.ReadWriteCloser
|
io.ReadWriteCloser
|
||||||
|
|
|
@ -484,125 +484,6 @@ func TestBuildHTTPRequest(t *testing.T) {
|
||||||
},
|
},
|
||||||
body: io.NopCloser(&bytes.Buffer{}),
|
body: io.NopCloser(&bytes.Buffer{}),
|
||||||
},
|
},
|
||||||
{
|
|
||||||
name: "if edge sends the body is empty hint, set body to empty",
|
|
||||||
connectRequest: &pogs.ConnectRequest{
|
|
||||||
Dest: "http://test.com",
|
|
||||||
Metadata: []pogs.Metadata{
|
|
||||||
{
|
|
||||||
Key: "HttpHeader:Another-Header",
|
|
||||||
Val: "Misc",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpHost",
|
|
||||||
Val: "cf.host",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpMethod",
|
|
||||||
Val: "put",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: HTTPRequestBodyHintKey,
|
|
||||||
Val: RequestBodyHintEmpty.String(),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
req: &http.Request{
|
|
||||||
Method: "put",
|
|
||||||
URL: &url.URL{
|
|
||||||
Scheme: "http",
|
|
||||||
Host: "test.com",
|
|
||||||
},
|
|
||||||
Proto: "HTTP/1.1",
|
|
||||||
ProtoMajor: 1,
|
|
||||||
ProtoMinor: 1,
|
|
||||||
Header: http.Header{
|
|
||||||
"Another-Header": []string{"Misc"},
|
|
||||||
},
|
|
||||||
ContentLength: 0,
|
|
||||||
Host: "cf.host",
|
|
||||||
Body: http.NoBody,
|
|
||||||
},
|
|
||||||
body: io.NopCloser(&bytes.Buffer{}),
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "if edge sends the body has data hint, don't set body to empty",
|
|
||||||
connectRequest: &pogs.ConnectRequest{
|
|
||||||
Dest: "http://test.com",
|
|
||||||
Metadata: []pogs.Metadata{
|
|
||||||
{
|
|
||||||
Key: "HttpHeader:Another-Header",
|
|
||||||
Val: "Misc",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpHost",
|
|
||||||
Val: "cf.host",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpMethod",
|
|
||||||
Val: "put",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: HTTPRequestBodyHintKey,
|
|
||||||
Val: RequestBodyHintHasData.String(),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
req: &http.Request{
|
|
||||||
Method: "put",
|
|
||||||
URL: &url.URL{
|
|
||||||
Scheme: "http",
|
|
||||||
Host: "test.com",
|
|
||||||
},
|
|
||||||
Proto: "HTTP/1.1",
|
|
||||||
ProtoMajor: 1,
|
|
||||||
ProtoMinor: 1,
|
|
||||||
Header: http.Header{
|
|
||||||
"Another-Header": []string{"Misc"},
|
|
||||||
},
|
|
||||||
ContentLength: 0,
|
|
||||||
Host: "cf.host",
|
|
||||||
Body: io.NopCloser(&bytes.Buffer{}),
|
|
||||||
},
|
|
||||||
body: io.NopCloser(&bytes.Buffer{}),
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "if the http method usually has body, don't set body to empty",
|
|
||||||
connectRequest: &pogs.ConnectRequest{
|
|
||||||
Dest: "http://test.com",
|
|
||||||
Metadata: []pogs.Metadata{
|
|
||||||
{
|
|
||||||
Key: "HttpHeader:Another-Header",
|
|
||||||
Val: "Misc",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpHost",
|
|
||||||
Val: "cf.host",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Key: "HttpMethod",
|
|
||||||
Val: "post",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
req: &http.Request{
|
|
||||||
Method: "post",
|
|
||||||
URL: &url.URL{
|
|
||||||
Scheme: "http",
|
|
||||||
Host: "test.com",
|
|
||||||
},
|
|
||||||
Proto: "HTTP/1.1",
|
|
||||||
ProtoMajor: 1,
|
|
||||||
ProtoMinor: 1,
|
|
||||||
Header: http.Header{
|
|
||||||
"Another-Header": []string{"Misc"},
|
|
||||||
},
|
|
||||||
ContentLength: 0,
|
|
||||||
Host: "cf.host",
|
|
||||||
Body: io.NopCloser(&bytes.Buffer{}),
|
|
||||||
},
|
|
||||||
body: io.NopCloser(&bytes.Buffer{}),
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log := zerolog.Nop()
|
log := zerolog.Nop()
|
||||||
|
@ -855,7 +736,6 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T, index uint8) *QU
|
||||||
nil,
|
nil,
|
||||||
15*time.Second,
|
15*time.Second,
|
||||||
0*time.Second,
|
0*time.Second,
|
||||||
0*time.Second,
|
|
||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
return qc
|
return qc
|
||||||
|
|
|
@ -604,7 +604,6 @@ func (e *EdgeTunnelServer) serveQUIC(
|
||||||
e.config.PacketConfig,
|
e.config.PacketConfig,
|
||||||
e.config.RPCTimeout,
|
e.config.RPCTimeout,
|
||||||
e.config.WriteStreamTimeout,
|
e.config.WriteStreamTimeout,
|
||||||
e.config.GracePeriod,
|
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
connLogger.ConnAwareLogger().Err(err).Msgf("Failed to create new quic connection")
|
connLogger.ConnAwareLogger().Err(err).Msgf("Failed to create new quic connection")
|
||||||
|
|
|
@ -23,7 +23,7 @@ type RegistrationClient interface {
|
||||||
edgeAddress net.IP,
|
edgeAddress net.IP,
|
||||||
) (*pogs.ConnectionDetails, error)
|
) (*pogs.ConnectionDetails, error)
|
||||||
SendLocalConfiguration(ctx context.Context, config []byte) error
|
SendLocalConfiguration(ctx context.Context, config []byte) error
|
||||||
GracefulShutdown(ctx context.Context, gracePeriod time.Duration) error
|
GracefulShutdown(ctx context.Context, gracePeriod time.Duration)
|
||||||
Close()
|
Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -79,7 +79,7 @@ func (r *registrationClient) SendLocalConfiguration(ctx context.Context, config
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *registrationClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) error {
|
func (r *registrationClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
|
||||||
ctx, cancel := context.WithTimeout(ctx, gracePeriod)
|
ctx, cancel := context.WithTimeout(ctx, gracePeriod)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
defer metrics.CapnpMetrics.ClientOperations.WithLabelValues(metrics.Registration, metrics.OperationUnregisterConnection).Inc()
|
defer metrics.CapnpMetrics.ClientOperations.WithLabelValues(metrics.Registration, metrics.OperationUnregisterConnection).Inc()
|
||||||
|
@ -88,9 +88,7 @@ func (r *registrationClient) GracefulShutdown(ctx context.Context, gracePeriod t
|
||||||
err := r.client.UnregisterConnection(ctx)
|
err := r.client.UnregisterConnection(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
metrics.CapnpMetrics.ClientFailures.WithLabelValues(metrics.Registration, metrics.OperationUnregisterConnection).Inc()
|
metrics.CapnpMetrics.ClientFailures.WithLabelValues(metrics.Registration, metrics.OperationUnregisterConnection).Inc()
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *registrationClient) Close() {
|
func (r *registrationClient) Close() {
|
||||||
|
|
Loading…
Reference in New Issue