2021-04-18 15:41:13 -04:00
|
|
|
// Copyright (c) 2015-2021 MinIO, Inc.
|
|
|
|
//
|
|
|
|
// This file is part of MinIO Object Storage stack
|
|
|
|
//
|
|
|
|
// This program is free software: you can redistribute it and/or modify
|
|
|
|
// it under the terms of the GNU Affero General Public License as published by
|
|
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
|
|
// (at your option) any later version.
|
|
|
|
//
|
|
|
|
// This program is distributed in the hope that it will be useful
|
|
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
|
// GNU Affero General Public License for more details.
|
|
|
|
//
|
|
|
|
// You should have received a copy of the GNU Affero General Public License
|
|
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
2018-10-04 20:44:06 -04:00
|
|
|
|
|
|
|
package cmd
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
2019-11-13 15:17:45 -05:00
|
|
|
"context"
|
2019-09-26 02:08:24 -04:00
|
|
|
"encoding/gob"
|
|
|
|
"encoding/hex"
|
2020-06-17 17:49:26 -04:00
|
|
|
"errors"
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
"fmt"
|
2018-10-04 20:44:06 -04:00
|
|
|
"io"
|
2022-10-18 16:50:46 -04:00
|
|
|
"net/http"
|
2018-10-04 20:44:06 -04:00
|
|
|
"net/url"
|
|
|
|
"path"
|
|
|
|
"strconv"
|
2019-12-23 19:31:03 -05:00
|
|
|
"strings"
|
2020-12-10 16:03:22 -05:00
|
|
|
"sync"
|
2024-04-15 04:25:46 -04:00
|
|
|
"sync/atomic"
|
2021-03-04 17:36:23 -05:00
|
|
|
"time"
|
2019-01-30 13:53:57 -05:00
|
|
|
|
2023-06-19 20:53:08 -04:00
|
|
|
"github.com/minio/madmin-go/v3"
|
2024-02-23 16:28:14 -05:00
|
|
|
"github.com/minio/minio/internal/cachevalue"
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
"github.com/minio/minio/internal/grid"
|
2021-06-01 17:59:40 -04:00
|
|
|
xhttp "github.com/minio/minio/internal/http"
|
2024-01-28 13:04:17 -05:00
|
|
|
xioutil "github.com/minio/minio/internal/ioutil"
|
2021-06-01 17:59:40 -04:00
|
|
|
"github.com/minio/minio/internal/rest"
|
2023-09-04 15:57:37 -04:00
|
|
|
xnet "github.com/minio/pkg/v2/net"
|
2021-01-07 22:27:31 -05:00
|
|
|
xbufio "github.com/philhofer/fwd"
|
2020-11-02 20:07:52 -05:00
|
|
|
"github.com/tinylib/msgp/msgp"
|
2018-10-04 20:44:06 -04:00
|
|
|
)
|
|
|
|
|
2019-02-13 18:29:46 -05:00
|
|
|
func isNetworkError(err error) bool {
|
2018-10-04 20:44:06 -04:00
|
|
|
if err == nil {
|
|
|
|
return false
|
|
|
|
}
|
2023-06-01 18:40:28 -04:00
|
|
|
|
2019-07-29 17:48:18 -04:00
|
|
|
if nerr, ok := err.(*rest.NetworkError); ok {
|
2023-06-01 18:40:28 -04:00
|
|
|
if down := xnet.IsNetworkOrHostDown(nerr.Err, false); down {
|
|
|
|
return true
|
|
|
|
}
|
2024-04-28 13:53:50 -04:00
|
|
|
if errors.Is(nerr.Err, rest.ErrClientClosed) {
|
|
|
|
return true
|
|
|
|
}
|
2019-05-02 10:09:57 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
if errors.Is(err, grid.ErrDisconnected) {
|
|
|
|
return true
|
|
|
|
}
|
2023-06-01 18:40:28 -04:00
|
|
|
// More corner cases suitable for storage REST API
|
2023-02-06 13:42:08 -05:00
|
|
|
switch {
|
2022-10-18 16:50:46 -04:00
|
|
|
// A peer node can be in shut down phase and proactively
|
2024-04-28 13:53:50 -04:00
|
|
|
// return 503 server closed error, consider it as an offline node
|
2023-02-06 13:42:08 -05:00
|
|
|
case strings.Contains(err.Error(), http.ErrServerClosed.Error()):
|
|
|
|
return true
|
|
|
|
// Corner case, the server closed the connection with a keep-alive timeout
|
|
|
|
// some requests are not retried internally, such as POST request with written body
|
|
|
|
case strings.Contains(err.Error(), "server closed idle connection"):
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
return false
|
2019-05-02 10:09:57 -04:00
|
|
|
}
|
|
|
|
|
2020-01-14 21:45:17 -05:00
|
|
|
// Converts network error to storageErr. This function is
|
|
|
|
// written so that the storageAPI errors are consistent
|
|
|
|
// across network disks.
|
2018-10-04 20:44:06 -04:00
|
|
|
func toStorageErr(err error) error {
|
|
|
|
if err == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-02-13 18:29:46 -05:00
|
|
|
if isNetworkError(err) {
|
2019-05-29 13:21:47 -04:00
|
|
|
return errDiskNotFound
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
switch err.Error() {
|
2019-10-01 16:12:15 -04:00
|
|
|
case errFaultyDisk.Error():
|
|
|
|
return errFaultyDisk
|
2024-01-27 19:13:41 -05:00
|
|
|
case errFaultyRemoteDisk.Error():
|
|
|
|
return errFaultyRemoteDisk
|
2019-10-01 16:12:15 -04:00
|
|
|
case errFileCorrupt.Error():
|
|
|
|
return errFileCorrupt
|
2018-10-04 20:44:06 -04:00
|
|
|
case errUnexpected.Error():
|
|
|
|
return errUnexpected
|
|
|
|
case errDiskFull.Error():
|
|
|
|
return errDiskFull
|
|
|
|
case errVolumeNotFound.Error():
|
|
|
|
return errVolumeNotFound
|
|
|
|
case errVolumeExists.Error():
|
|
|
|
return errVolumeExists
|
|
|
|
case errFileNotFound.Error():
|
|
|
|
return errFileNotFound
|
2020-09-21 18:16:16 -04:00
|
|
|
case errFileVersionNotFound.Error():
|
|
|
|
return errFileVersionNotFound
|
2018-10-04 20:44:06 -04:00
|
|
|
case errFileNameTooLong.Error():
|
|
|
|
return errFileNameTooLong
|
|
|
|
case errFileAccessDenied.Error():
|
|
|
|
return errFileAccessDenied
|
2021-01-02 15:01:29 -05:00
|
|
|
case errPathNotFound.Error():
|
|
|
|
return errPathNotFound
|
2018-10-04 20:44:06 -04:00
|
|
|
case errIsNotRegular.Error():
|
|
|
|
return errIsNotRegular
|
|
|
|
case errVolumeNotEmpty.Error():
|
|
|
|
return errVolumeNotEmpty
|
|
|
|
case errVolumeAccessDenied.Error():
|
|
|
|
return errVolumeAccessDenied
|
|
|
|
case errCorruptedFormat.Error():
|
|
|
|
return errCorruptedFormat
|
2024-01-12 17:48:44 -05:00
|
|
|
case errCorruptedBackend.Error():
|
|
|
|
return errCorruptedBackend
|
2018-10-04 20:44:06 -04:00
|
|
|
case errUnformattedDisk.Error():
|
|
|
|
return errUnformattedDisk
|
|
|
|
case errInvalidAccessKeyID.Error():
|
|
|
|
return errInvalidAccessKeyID
|
|
|
|
case errAuthentication.Error():
|
|
|
|
return errAuthentication
|
|
|
|
case errRPCAPIVersionUnsupported.Error():
|
|
|
|
return errRPCAPIVersionUnsupported
|
|
|
|
case errServerTimeMismatch.Error():
|
|
|
|
return errServerTimeMismatch
|
2019-10-01 16:12:15 -04:00
|
|
|
case io.EOF.Error():
|
|
|
|
return io.EOF
|
|
|
|
case io.ErrUnexpectedEOF.Error():
|
|
|
|
return io.ErrUnexpectedEOF
|
2019-10-25 13:37:53 -04:00
|
|
|
case errDiskStale.Error():
|
|
|
|
return errDiskNotFound
|
2020-07-21 16:54:06 -04:00
|
|
|
case errDiskNotFound.Error():
|
|
|
|
return errDiskNotFound
|
2024-01-27 19:13:41 -05:00
|
|
|
case errMaxVersionsExceeded.Error():
|
|
|
|
return errMaxVersionsExceeded
|
|
|
|
case errInconsistentDisk.Error():
|
|
|
|
return errInconsistentDisk
|
|
|
|
case errDriveIsRoot.Error():
|
|
|
|
return errDriveIsRoot
|
|
|
|
case errDiskOngoingReq.Error():
|
|
|
|
return errDiskOngoingReq
|
|
|
|
case grid.ErrUnknownHandler.Error():
|
|
|
|
return errInconsistentDisk
|
2024-02-03 04:04:33 -05:00
|
|
|
case grid.ErrDisconnected.Error():
|
|
|
|
return errDiskNotFound
|
2019-01-30 13:53:57 -05:00
|
|
|
}
|
2018-10-04 20:44:06 -04:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Abstracts a remote disk.
|
|
|
|
type storageRESTClient struct {
|
2024-04-15 04:25:46 -04:00
|
|
|
endpoint Endpoint
|
|
|
|
restClient *rest.Client
|
|
|
|
gridConn *grid.Subroute
|
|
|
|
diskID atomic.Pointer[string]
|
2021-03-04 17:36:23 -05:00
|
|
|
|
2024-02-23 16:28:14 -05:00
|
|
|
diskInfoCache *cachevalue.Cache[DiskInfo]
|
2021-03-04 17:36:23 -05:00
|
|
|
}
|
|
|
|
|
|
|
|
// Retrieve location indexes.
|
|
|
|
func (client *storageRESTClient) GetDiskLoc() (poolIdx, setIdx, diskIdx int) {
|
2024-04-11 13:45:28 -04:00
|
|
|
return client.endpoint.PoolIdx, client.endpoint.SetIdx, client.endpoint.DiskIdx
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2024-04-15 04:25:46 -04:00
|
|
|
// Wrapper to restClient.Call to handle network errors, in case of network error the connection is disconnected
|
|
|
|
// and a healthcheck routine gets invoked that would reconnect.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) call(ctx context.Context, method string, values url.Values, body io.Reader, length int64) (io.ReadCloser, error) {
|
2019-02-13 18:29:46 -05:00
|
|
|
if values == nil {
|
|
|
|
values = make(url.Values)
|
|
|
|
}
|
2024-04-15 04:25:46 -04:00
|
|
|
values.Set(storageRESTDiskID, *client.diskID.Load())
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.restClient.Call(ctx, method, values, body, length)
|
2024-03-26 02:24:59 -04:00
|
|
|
if err != nil {
|
|
|
|
return nil, toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
2024-03-26 02:24:59 -04:00
|
|
|
return respBody, nil
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// Stringer provides a canonicalized representation of network device.
|
|
|
|
func (client *storageRESTClient) String() string {
|
|
|
|
return client.endpoint.String()
|
|
|
|
}
|
|
|
|
|
|
|
|
// IsOnline - returns whether RPC client failed to connect or not.
|
|
|
|
func (client *storageRESTClient) IsOnline() bool {
|
2024-02-28 12:54:52 -05:00
|
|
|
return client.gridConn.State() == grid.StateConnected
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2021-05-11 12:19:15 -04:00
|
|
|
// LastConn - returns when the disk is seen to be connected the last time
|
|
|
|
func (client *storageRESTClient) LastConn() time.Time {
|
|
|
|
return client.restClient.LastConn()
|
|
|
|
}
|
|
|
|
|
2020-05-19 17:27:20 -04:00
|
|
|
func (client *storageRESTClient) IsLocal() bool {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2020-01-13 16:09:10 -05:00
|
|
|
func (client *storageRESTClient) Hostname() string {
|
|
|
|
return client.endpoint.Host
|
|
|
|
}
|
|
|
|
|
2020-09-28 22:39:32 -04:00
|
|
|
func (client *storageRESTClient) Endpoint() Endpoint {
|
|
|
|
return client.endpoint
|
|
|
|
}
|
|
|
|
|
2021-03-04 17:36:23 -05:00
|
|
|
func (client *storageRESTClient) Healing() *healingTracker {
|
2022-06-25 11:50:16 -04:00
|
|
|
// This call is not implemented for remote client on purpose.
|
|
|
|
// healing tracker is always for local disks.
|
|
|
|
return nil
|
2020-09-28 22:39:32 -04:00
|
|
|
}
|
|
|
|
|
2024-01-02 16:51:24 -05:00
|
|
|
func (client *storageRESTClient) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
|
2024-01-28 13:04:17 -05:00
|
|
|
defer xioutil.SafeClose(updates)
|
2020-12-10 16:03:22 -05:00
|
|
|
|
2024-02-19 17:54:46 -05:00
|
|
|
st, err := storageNSScannerRPC.Call(ctx, client.gridConn, &nsScannerOptions{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
ScanMode: int(scanMode),
|
|
|
|
Cache: &cache,
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return cache, toStorageErr(err)
|
|
|
|
}
|
|
|
|
var final *dataUsageCache
|
|
|
|
err = st.Results(func(resp *nsScannerResp) error {
|
|
|
|
if resp.Update != nil {
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case updates <- *resp.Update:
|
|
|
|
}
|
2021-05-19 17:38:30 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
if resp.Final != nil {
|
|
|
|
final = resp.Final
|
2022-07-19 11:35:29 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
// We can't reuse the response since it is sent upstream.
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return cache, toStorageErr(err)
|
2021-05-19 17:38:30 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
if final == nil {
|
|
|
|
return cache, errors.New("no final cache")
|
2021-05-19 17:38:30 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
return *final, nil
|
2019-12-12 09:02:37 -05:00
|
|
|
}
|
|
|
|
|
2020-03-27 17:48:30 -04:00
|
|
|
func (client *storageRESTClient) GetDiskID() (string, error) {
|
2024-01-23 17:11:46 -05:00
|
|
|
if !client.IsOnline() {
|
|
|
|
// make sure to check if the disk is offline, since the underlying
|
|
|
|
// value is cached we should attempt to invalidate it if such calls
|
|
|
|
// were attempted. This can lead to false success under certain conditions
|
|
|
|
// - this change attempts to avoid stale information if the underlying
|
|
|
|
// transport is already down.
|
|
|
|
return "", errDiskNotFound
|
|
|
|
}
|
|
|
|
|
2020-05-28 16:03:04 -04:00
|
|
|
// This call should never be over the network, this is always
|
|
|
|
// a cached value - caller should make sure to use this
|
|
|
|
// function on a fresh disk or make sure to look at the error
|
|
|
|
// from a different networked call to validate the GetDiskID()
|
2024-04-15 04:25:46 -04:00
|
|
|
return *client.diskID.Load(), nil
|
2020-03-27 17:48:30 -04:00
|
|
|
}
|
|
|
|
|
2019-10-25 13:37:53 -04:00
|
|
|
func (client *storageRESTClient) SetDiskID(id string) {
|
2024-04-15 04:25:46 -04:00
|
|
|
client.diskID.Store(&id)
|
2019-10-25 13:37:53 -04:00
|
|
|
}
|
|
|
|
|
2024-01-25 15:45:46 -05:00
|
|
|
func (client *storageRESTClient) DiskInfo(ctx context.Context, opts DiskInfoOptions) (info DiskInfo, err error) {
|
2024-02-28 12:54:52 -05:00
|
|
|
if !client.IsOnline() {
|
2021-04-29 12:54:16 -04:00
|
|
|
// make sure to check if the disk is offline, since the underlying
|
|
|
|
// value is cached we should attempt to invalidate it if such calls
|
|
|
|
// were attempted. This can lead to false success under certain conditions
|
|
|
|
// - this change attempts to avoid stale information if the underlying
|
|
|
|
// transport is already down.
|
2023-11-24 12:07:14 -05:00
|
|
|
return info, errDiskNotFound
|
2021-04-29 12:54:16 -04:00
|
|
|
}
|
2023-12-29 18:52:41 -05:00
|
|
|
|
2024-02-26 13:49:19 -05:00
|
|
|
// if 'NoOp' we do not cache the value.
|
|
|
|
if opts.NoOp {
|
2024-02-23 12:21:38 -05:00
|
|
|
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
|
|
defer cancel()
|
2024-01-25 15:45:46 -05:00
|
|
|
|
2024-04-15 04:25:46 -04:00
|
|
|
opts.DiskID = *client.diskID.Load()
|
2024-02-23 12:21:38 -05:00
|
|
|
|
|
|
|
infop, err := storageDiskInfoRPC.Call(ctx, client.gridConn, &opts)
|
|
|
|
if err != nil {
|
|
|
|
return info, toStorageErr(err)
|
|
|
|
}
|
|
|
|
info = *infop
|
|
|
|
if info.Error != "" {
|
|
|
|
return info, toStorageErr(errors.New(info.Error))
|
|
|
|
}
|
|
|
|
return info, nil
|
|
|
|
} // In all other cases cache the value upto 1sec.
|
|
|
|
|
2024-02-28 12:09:09 -05:00
|
|
|
client.diskInfoCache.InitOnce(time.Second,
|
|
|
|
cachevalue.Opts{CacheError: true},
|
|
|
|
func() (info DiskInfo, err error) {
|
2024-02-23 12:21:38 -05:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
|
|
defer cancel()
|
|
|
|
|
2024-04-15 04:25:46 -04:00
|
|
|
nopts := DiskInfoOptions{DiskID: *client.diskID.Load(), Metrics: true}
|
2024-02-23 12:21:38 -05:00
|
|
|
infop, err := storageDiskInfoRPC.Call(ctx, client.gridConn, &nopts)
|
|
|
|
if err != nil {
|
|
|
|
return info, toStorageErr(err)
|
|
|
|
}
|
|
|
|
info = *infop
|
|
|
|
if info.Error != "" {
|
|
|
|
return info, toStorageErr(errors.New(info.Error))
|
|
|
|
}
|
|
|
|
return info, nil
|
2024-02-28 12:09:09 -05:00
|
|
|
},
|
|
|
|
)
|
2024-02-23 12:21:38 -05:00
|
|
|
|
2024-02-26 13:49:19 -05:00
|
|
|
return client.diskInfoCache.Get()
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2019-12-23 19:31:03 -05:00
|
|
|
// MakeVolBulk - create multiple volumes in a bulk operation.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) MakeVolBulk(ctx context.Context, volumes ...string) (err error) {
|
2024-01-17 23:41:23 -05:00
|
|
|
return errInvalidArgument
|
2019-12-23 19:31:03 -05:00
|
|
|
}
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// MakeVol - create a volume on a remote disk.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) MakeVol(ctx context.Context, volume string) (err error) {
|
2024-01-17 23:41:23 -05:00
|
|
|
return errInvalidArgument
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// ListVols - List all volumes on a remote disk.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) ListVols(ctx context.Context) (vols []VolInfo, err error) {
|
2024-01-17 23:41:23 -05:00
|
|
|
return nil, errInvalidArgument
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// StatVol - get volume info over the network.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) StatVol(ctx context.Context, volume string) (vol VolInfo, err error) {
|
2024-02-19 17:54:46 -05:00
|
|
|
v, err := storageStatVolRPC.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{
|
2024-04-15 04:25:46 -04:00
|
|
|
storageRESTDiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
storageRESTVolume: volume,
|
|
|
|
}))
|
2018-10-04 20:44:06 -04:00
|
|
|
if err != nil {
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
return vol, toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
vol = *v
|
2024-02-01 15:41:20 -05:00
|
|
|
// Performs shallow copy, so we can reuse.
|
2024-02-19 17:54:46 -05:00
|
|
|
storageStatVolRPC.PutResponse(v)
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
return vol, nil
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// DeleteVol - Deletes a volume over the network.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) DeleteVol(ctx context.Context, volume string, forceDelete bool) (err error) {
|
2024-01-17 23:41:23 -05:00
|
|
|
return errInvalidArgument
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2019-01-17 07:58:18 -05:00
|
|
|
// AppendFile - append to a file.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) AppendFile(ctx context.Context, volume string, path string, buf []byte) error {
|
2018-10-04 20:44:06 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
2020-09-04 12:45:06 -04:00
|
|
|
reader := bytes.NewReader(buf)
|
|
|
|
respBody, err := client.call(ctx, storageRESTMethodAppendFile, values, reader, -1)
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2018-10-04 20:44:06 -04:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2024-01-30 15:43:25 -05:00
|
|
|
func (client *storageRESTClient) CreateFile(ctx context.Context, origvolume, volume, path string, size int64, reader io.Reader) error {
|
2018-10-04 20:44:06 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
2020-09-04 12:45:06 -04:00
|
|
|
values.Set(storageRESTLength, strconv.Itoa(int(size)))
|
2024-01-30 15:43:25 -05:00
|
|
|
values.Set(storageRESTOrigVolume, origvolume)
|
|
|
|
|
2022-09-19 14:05:16 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodCreateFile, values, io.NopCloser(reader), size)
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2021-08-27 12:16:36 -04:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
_, err = waitForHTTPResponse(respBody)
|
2024-03-26 02:24:59 -04:00
|
|
|
return toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2024-01-30 15:43:25 -05:00
|
|
|
func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
2024-01-30 15:43:25 -05:00
|
|
|
OrigVolume: origvolume,
|
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
FI: fi,
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
|
2023-08-25 10:58:11 -04:00
|
|
|
func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
UpdateOpts: opts,
|
|
|
|
FI: fi,
|
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2021-04-04 16:32:31 -04:00
|
|
|
}
|
|
|
|
|
2023-12-29 18:52:41 -05:00
|
|
|
func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) (err error) {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
ForceDelMarker: forceDelMarker,
|
|
|
|
FI: fi,
|
2023-12-29 18:52:41 -05:00
|
|
|
Opts: opts,
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
|
2018-11-14 09:18:35 -05:00
|
|
|
// WriteAll - write all data to a file.
|
2020-11-02 19:14:31 -05:00
|
|
|
func (client *storageRESTClient) WriteAll(ctx context.Context, volume string, path string, b []byte) error {
|
2024-02-28 12:54:52 -05:00
|
|
|
_, err := storageWriteAllRPC.Call(ctx, client.gridConn, &WriteAllHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
2024-02-28 12:54:52 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
Buf: b,
|
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2018-11-14 09:18:35 -05:00
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
// CheckParts - stat all file parts.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) CheckParts(ctx context.Context, volume string, path string, fi FileInfo) error {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err := storageCheckPartsRPC.Call(ctx, client.gridConn, &CheckPartsHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
FI: fi,
|
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// RenameData - rename source path to destination path atomically, metadata and data file.
|
2024-04-23 13:15:52 -04:00
|
|
|
func (client *storageRESTClient) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo,
|
|
|
|
dstVolume, dstPath string, opts RenameOptions,
|
|
|
|
) (res RenameDataResp, err error) {
|
2024-04-01 19:42:09 -04:00
|
|
|
params := RenameDataHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
SrcVolume: srcVolume,
|
|
|
|
SrcPath: srcPath,
|
|
|
|
DstPath: dstPath,
|
|
|
|
DstVolume: dstVolume,
|
|
|
|
FI: fi,
|
2023-12-29 18:52:41 -05:00
|
|
|
Opts: opts,
|
2024-04-01 19:42:09 -04:00
|
|
|
}
|
|
|
|
var resp *RenameDataResp
|
|
|
|
if fi.Data == nil {
|
|
|
|
resp, err = storageRenameDataRPC.Call(ctx, client.gridConn, ¶ms)
|
|
|
|
} else {
|
|
|
|
resp, err = storageRenameDataInlineRPC.Call(ctx, client.gridConn, &RenameDataInlineHandlerParams{params})
|
|
|
|
}
|
2022-09-05 19:51:37 -04:00
|
|
|
if err != nil {
|
2024-04-23 13:15:52 -04:00
|
|
|
return res, toStorageErr(err)
|
2022-09-05 19:51:37 -04:00
|
|
|
}
|
2024-04-01 19:42:09 -04:00
|
|
|
|
2024-02-19 17:54:46 -05:00
|
|
|
defer storageRenameDataRPC.PutResponse(resp)
|
2024-04-23 13:15:52 -04:00
|
|
|
return *resp, nil
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
|
2021-01-07 22:27:31 -05:00
|
|
|
// where we keep old *Readers
|
|
|
|
var readMsgpReaderPool = sync.Pool{New: func() interface{} { return &msgp.Reader{} }}
|
|
|
|
|
|
|
|
// mspNewReader returns a *Reader that reads from the provided reader.
|
|
|
|
// The reader will be buffered.
|
2023-07-06 19:02:08 -04:00
|
|
|
// Return with readMsgpReaderPoolPut when done.
|
2021-01-07 22:27:31 -05:00
|
|
|
func msgpNewReader(r io.Reader) *msgp.Reader {
|
|
|
|
p := readMsgpReaderPool.Get().(*msgp.Reader)
|
|
|
|
if p.R == nil {
|
2023-07-06 19:02:08 -04:00
|
|
|
p.R = xbufio.NewReaderSize(r, 4<<10)
|
2021-01-07 22:27:31 -05:00
|
|
|
} else {
|
|
|
|
p.R.Reset(r)
|
|
|
|
}
|
|
|
|
return p
|
|
|
|
}
|
|
|
|
|
2023-07-06 19:02:08 -04:00
|
|
|
// readMsgpReaderPoolPut can be used to reuse a *msgp.Reader.
|
|
|
|
func readMsgpReaderPoolPut(r *msgp.Reader) {
|
|
|
|
if r != nil {
|
|
|
|
readMsgpReaderPool.Put(r)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-30 15:43:25 -05:00
|
|
|
func (client *storageRESTClient) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) {
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
// Use websocket when not reading data.
|
2023-11-21 00:33:47 -05:00
|
|
|
if !opts.ReadData {
|
2024-02-19 17:54:46 -05:00
|
|
|
resp, err := storageReadVersionRPC.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{
|
2024-04-15 04:25:46 -04:00
|
|
|
storageRESTDiskID: *client.diskID.Load(),
|
2024-01-30 15:43:25 -05:00
|
|
|
storageRESTOrigVolume: origvolume,
|
|
|
|
storageRESTVolume: volume,
|
|
|
|
storageRESTFilePath: path,
|
|
|
|
storageRESTVersionID: versionID,
|
|
|
|
storageRESTReadData: strconv.FormatBool(opts.ReadData),
|
|
|
|
storageRESTHealing: strconv.FormatBool(opts.Healing),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
}))
|
|
|
|
if err != nil {
|
|
|
|
return fi, toStorageErr(err)
|
|
|
|
}
|
|
|
|
return *resp, nil
|
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
values := make(url.Values)
|
2024-01-30 15:43:25 -05:00
|
|
|
values.Set(storageRESTOrigVolume, origvolume)
|
2020-06-12 23:04:01 -04:00
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
|
|
|
values.Set(storageRESTVersionID, versionID)
|
2023-11-21 00:33:47 -05:00
|
|
|
values.Set(storageRESTReadData, strconv.FormatBool(opts.ReadData))
|
|
|
|
values.Set(storageRESTHealing, strconv.FormatBool(opts.Healing))
|
2020-06-12 23:04:01 -04:00
|
|
|
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodReadVersion, values, nil, -1)
|
2018-10-04 20:44:06 -04:00
|
|
|
if err != nil {
|
2020-06-12 23:04:01 -04:00
|
|
|
return fi, err
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2021-01-07 22:27:31 -05:00
|
|
|
|
|
|
|
dec := msgpNewReader(respBody)
|
2023-07-06 19:02:08 -04:00
|
|
|
defer readMsgpReaderPoolPut(dec)
|
2021-01-07 22:27:31 -05:00
|
|
|
|
|
|
|
err = fi.DecodeMsg(dec)
|
2020-06-12 23:04:01 -04:00
|
|
|
return fi, err
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2022-04-20 15:49:05 -04:00
|
|
|
// ReadXL - reads all contents of xl.meta of a file.
|
|
|
|
func (client *storageRESTClient) ReadXL(ctx context.Context, volume string, path string, readData bool) (rf RawFileInfo, err error) {
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
// Use websocket when not reading data.
|
|
|
|
if !readData {
|
2024-02-19 17:54:46 -05:00
|
|
|
resp, err := storageReadXLRPC.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{
|
2024-04-15 04:25:46 -04:00
|
|
|
storageRESTDiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
storageRESTVolume: volume,
|
|
|
|
storageRESTFilePath: path,
|
|
|
|
storageRESTReadData: "false",
|
|
|
|
}))
|
|
|
|
if err != nil {
|
|
|
|
return rf, toStorageErr(err)
|
|
|
|
}
|
|
|
|
return *resp, nil
|
|
|
|
}
|
|
|
|
|
2022-04-20 15:49:05 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
|
|
|
values.Set(storageRESTReadData, strconv.FormatBool(readData))
|
|
|
|
respBody, err := client.call(ctx, storageRESTMethodReadXL, values, nil, -1)
|
|
|
|
if err != nil {
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
return rf, toStorageErr(err)
|
2022-04-20 15:49:05 -04:00
|
|
|
}
|
|
|
|
defer xhttp.DrainBody(respBody)
|
|
|
|
|
|
|
|
dec := msgpNewReader(respBody)
|
2023-07-06 19:02:08 -04:00
|
|
|
defer readMsgpReaderPoolPut(dec)
|
2022-04-20 15:49:05 -04:00
|
|
|
|
|
|
|
err = rf.DecodeMsg(dec)
|
|
|
|
return rf, err
|
|
|
|
}
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// ReadAll - reads all contents of a file.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) ReadAll(ctx context.Context, volume string, path string) ([]byte, error) {
|
2024-02-19 17:54:46 -05:00
|
|
|
gridBytes, err := storageReadAllRPC.Call(ctx, client.gridConn, &ReadAllHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
2024-01-25 15:45:46 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
})
|
2018-10-04 20:44:06 -04:00
|
|
|
if err != nil {
|
2024-01-25 15:45:46 -05:00
|
|
|
return nil, toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
2024-01-25 15:45:46 -05:00
|
|
|
|
|
|
|
return *gridBytes, nil
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2019-01-17 07:58:18 -05:00
|
|
|
// ReadFileStream - returns a reader for the requested file.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) ReadFileStream(ctx context.Context, volume, path string, offset, length int64) (io.ReadCloser, error) {
|
2019-01-17 07:58:18 -05:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
|
|
|
values.Set(storageRESTOffset, strconv.Itoa(int(offset)))
|
|
|
|
values.Set(storageRESTLength, strconv.Itoa(int(length)))
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodReadFileStream, values, nil, -1)
|
2019-01-17 07:58:18 -05:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return respBody, nil
|
|
|
|
}
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// ReadFile - reads section of a file.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) ReadFile(ctx context.Context, volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (int64, error) {
|
2018-10-04 20:44:06 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
|
|
|
values.Set(storageRESTOffset, strconv.Itoa(int(offset)))
|
2020-09-04 12:45:06 -04:00
|
|
|
values.Set(storageRESTLength, strconv.Itoa(len(buf)))
|
2018-10-04 20:44:06 -04:00
|
|
|
if verifier != nil {
|
|
|
|
values.Set(storageRESTBitrotAlgo, verifier.algorithm.String())
|
|
|
|
values.Set(storageRESTBitrotHash, hex.EncodeToString(verifier.sum))
|
|
|
|
} else {
|
|
|
|
values.Set(storageRESTBitrotAlgo, "")
|
|
|
|
values.Set(storageRESTBitrotHash, "")
|
|
|
|
}
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodReadFile, values, nil, -1)
|
2018-10-04 20:44:06 -04:00
|
|
|
if err != nil {
|
|
|
|
return 0, err
|
|
|
|
}
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2020-09-04 12:45:06 -04:00
|
|
|
n, err := io.ReadFull(respBody, buf)
|
2024-03-26 02:24:59 -04:00
|
|
|
return int64(n), toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// ListDir - lists a directory.
|
2024-01-30 15:43:25 -05:00
|
|
|
func (client *storageRESTClient) ListDir(ctx context.Context, origvolume, volume, dirPath string, count int) (entries []string, err error) {
|
2024-02-19 17:54:46 -05:00
|
|
|
values := grid.NewMSS()
|
2018-10-04 20:44:06 -04:00
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTDirPath, dirPath)
|
|
|
|
values.Set(storageRESTCount, strconv.Itoa(count))
|
2024-01-30 15:43:25 -05:00
|
|
|
values.Set(storageRESTOrigVolume, origvolume)
|
2024-04-15 04:25:46 -04:00
|
|
|
values.Set(storageRESTDiskID, *client.diskID.Load())
|
2024-01-30 15:43:25 -05:00
|
|
|
|
2024-02-19 17:54:46 -05:00
|
|
|
st, err := storageListDirRPC.Call(ctx, client.gridConn, values)
|
2018-10-04 20:44:06 -04:00
|
|
|
if err != nil {
|
2024-02-19 17:54:46 -05:00
|
|
|
return nil, toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
2024-02-19 17:54:46 -05:00
|
|
|
err = st.Results(func(resp *ListDirResult) error {
|
|
|
|
entries = resp.Entries
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
return entries, toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
// DeleteFile - deletes a file.
|
2022-07-11 12:15:54 -04:00
|
|
|
func (client *storageRESTClient) Delete(ctx context.Context, volume string, path string, deleteOpts DeleteOptions) error {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err := storageDeleteFileRPC.Call(ctx, client.gridConn, &DeleteFileHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
Volume: volume,
|
|
|
|
FilePath: path,
|
|
|
|
Opts: deleteOpts,
|
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
// DeleteVersions - deletes list of specified versions if present
|
2023-12-29 18:52:41 -05:00
|
|
|
func (client *storageRESTClient) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) (errs []error) {
|
2020-06-12 23:04:01 -04:00
|
|
|
if len(versions) == 0 {
|
|
|
|
return errs
|
2019-05-13 15:25:49 -04:00
|
|
|
}
|
|
|
|
|
2020-03-11 11:56:36 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
2020-06-12 23:04:01 -04:00
|
|
|
values.Set(storageRESTTotalVersions, strconv.Itoa(len(versions)))
|
2020-03-11 11:56:36 -04:00
|
|
|
|
|
|
|
var buffer bytes.Buffer
|
2020-11-02 20:07:52 -05:00
|
|
|
encoder := msgp.NewWriter(&buffer)
|
2020-06-12 23:04:01 -04:00
|
|
|
for _, version := range versions {
|
2020-11-02 20:07:52 -05:00
|
|
|
version.EncodeMsg(encoder)
|
2020-03-11 11:56:36 -04:00
|
|
|
}
|
2024-04-04 08:04:40 -04:00
|
|
|
storageLogIf(ctx, encoder.Flush())
|
2020-03-11 11:56:36 -04:00
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
errs = make([]error, len(versions))
|
|
|
|
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodDeleteVersions, values, &buffer, -1)
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2020-03-11 11:56:36 -04:00
|
|
|
if err != nil {
|
2022-01-06 13:47:49 -05:00
|
|
|
if contextCanceled(ctx) {
|
|
|
|
err = ctx.Err()
|
|
|
|
}
|
2020-06-12 23:04:01 -04:00
|
|
|
for i := range errs {
|
|
|
|
errs[i] = err
|
|
|
|
}
|
|
|
|
return errs
|
2020-03-11 11:56:36 -04:00
|
|
|
}
|
|
|
|
|
2020-03-18 19:19:29 -04:00
|
|
|
reader, err := waitForHTTPResponse(respBody)
|
2020-03-11 11:56:36 -04:00
|
|
|
if err != nil {
|
2020-06-12 23:04:01 -04:00
|
|
|
for i := range errs {
|
2024-03-26 02:24:59 -04:00
|
|
|
errs[i] = toStorageErr(err)
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
return errs
|
2020-03-11 11:56:36 -04:00
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
dErrResp := &DeleteVersionsErrsResp{}
|
2020-03-11 11:56:36 -04:00
|
|
|
if err = gob.NewDecoder(reader).Decode(dErrResp); err != nil {
|
2020-06-12 23:04:01 -04:00
|
|
|
for i := range errs {
|
2024-03-26 02:24:59 -04:00
|
|
|
errs[i] = toStorageErr(err)
|
2020-06-12 23:04:01 -04:00
|
|
|
}
|
|
|
|
return errs
|
2019-05-13 15:25:49 -04:00
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
for i, dErr := range dErrResp.Errs {
|
|
|
|
errs[i] = toStorageErr(dErr)
|
2019-05-13 15:25:49 -04:00
|
|
|
}
|
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
return errs
|
2019-05-13 15:25:49 -04:00
|
|
|
}
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// RenameFile - renames a file.
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string) (err error) {
|
2024-02-19 17:54:46 -05:00
|
|
|
_, err = storageRenameFileRPC.Call(ctx, client.gridConn, &RenameFileHandlerParams{
|
2024-04-15 04:25:46 -04:00
|
|
|
DiskID: *client.diskID.Load(),
|
2024-01-25 15:45:46 -05:00
|
|
|
SrcVolume: srcVolume,
|
|
|
|
SrcFilePath: srcPath,
|
|
|
|
DstVolume: dstVolume,
|
|
|
|
DstFilePath: dstPath,
|
|
|
|
})
|
|
|
|
return toStorageErr(err)
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2020-09-04 12:45:06 -04:00
|
|
|
func (client *storageRESTClient) VerifyFile(ctx context.Context, volume, path string, fi FileInfo) error {
|
2019-07-08 16:51:18 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
2019-09-12 16:08:02 -04:00
|
|
|
|
2020-06-12 23:04:01 -04:00
|
|
|
var reader bytes.Buffer
|
2020-11-02 20:07:52 -05:00
|
|
|
if err := msgp.Encode(&reader, &fi); err != nil {
|
2020-06-12 23:04:01 -04:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2020-09-04 12:45:06 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodVerifyFile, values, &reader, -1)
|
2021-04-26 11:59:54 -04:00
|
|
|
defer xhttp.DrainBody(respBody)
|
2019-07-08 16:51:18 -04:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2020-06-12 23:04:01 -04:00
|
|
|
|
|
|
|
respReader, err := waitForHTTPResponse(respBody)
|
2020-03-11 11:56:36 -04:00
|
|
|
if err != nil {
|
2024-03-26 02:24:59 -04:00
|
|
|
return toStorageErr(err)
|
2019-07-08 16:51:18 -04:00
|
|
|
}
|
2020-06-12 23:04:01 -04:00
|
|
|
|
2019-07-08 16:51:18 -04:00
|
|
|
verifyResp := &VerifyFileResp{}
|
2020-06-12 23:04:01 -04:00
|
|
|
if err = gob.NewDecoder(respReader).Decode(verifyResp); err != nil {
|
2024-03-26 02:24:59 -04:00
|
|
|
return toStorageErr(err)
|
2019-07-08 16:51:18 -04:00
|
|
|
}
|
2020-06-12 23:04:01 -04:00
|
|
|
|
2019-07-08 16:51:18 -04:00
|
|
|
return toStorageErr(verifyResp.Err)
|
|
|
|
}
|
|
|
|
|
2021-10-01 14:50:00 -04:00
|
|
|
func (client *storageRESTClient) StatInfoFile(ctx context.Context, volume, path string, glob bool) (stat []StatInfo, err error) {
|
2021-07-09 14:29:16 -04:00
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
2023-07-11 10:46:58 -04:00
|
|
|
values.Set(storageRESTGlob, strconv.FormatBool(glob))
|
2021-07-09 14:29:16 -04:00
|
|
|
respBody, err := client.call(ctx, storageRESTMethodStatInfoFile, values, nil, -1)
|
|
|
|
if err != nil {
|
|
|
|
return stat, err
|
|
|
|
}
|
|
|
|
defer xhttp.DrainBody(respBody)
|
|
|
|
respReader, err := waitForHTTPResponse(respBody)
|
|
|
|
if err != nil {
|
2024-03-26 02:24:59 -04:00
|
|
|
return stat, toStorageErr(err)
|
2021-07-09 14:29:16 -04:00
|
|
|
}
|
2021-10-01 14:50:00 -04:00
|
|
|
rd := msgpNewReader(respReader)
|
2023-07-06 19:02:08 -04:00
|
|
|
defer readMsgpReaderPoolPut(rd)
|
2021-10-01 14:50:00 -04:00
|
|
|
for {
|
|
|
|
var st StatInfo
|
|
|
|
err = st.DecodeMsg(rd)
|
|
|
|
if err != nil {
|
|
|
|
if errors.Is(err, io.EOF) {
|
|
|
|
err = nil
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
|
|
|
stat = append(stat, st)
|
|
|
|
}
|
|
|
|
|
2024-03-26 02:24:59 -04:00
|
|
|
return stat, toStorageErr(err)
|
2021-07-09 14:29:16 -04:00
|
|
|
}
|
|
|
|
|
2022-07-19 11:35:29 -04:00
|
|
|
// ReadMultiple will read multiple files and send each back as response.
|
|
|
|
// Files are read and returned in the given order.
|
|
|
|
// The resp channel is closed before the call returns.
|
|
|
|
// Only a canceled context or network errors returns an error.
|
|
|
|
func (client *storageRESTClient) ReadMultiple(ctx context.Context, req ReadMultipleReq, resp chan<- ReadMultipleResp) error {
|
2024-01-28 13:04:17 -05:00
|
|
|
defer xioutil.SafeClose(resp)
|
2022-07-19 11:35:29 -04:00
|
|
|
body, err := req.MarshalMsg(nil)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
respBody, err := client.call(ctx, storageRESTMethodReadMultiple, nil, bytes.NewReader(body), int64(len(body)))
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer xhttp.DrainBody(respBody)
|
2024-03-26 02:24:59 -04:00
|
|
|
|
2022-07-19 11:35:29 -04:00
|
|
|
pr, pw := io.Pipe()
|
|
|
|
go func() {
|
|
|
|
pw.CloseWithError(waitForHTTPStream(respBody, pw))
|
|
|
|
}()
|
|
|
|
mr := msgp.NewReader(pr)
|
2023-07-06 19:02:08 -04:00
|
|
|
defer readMsgpReaderPoolPut(mr)
|
2022-07-19 11:35:29 -04:00
|
|
|
for {
|
|
|
|
var file ReadMultipleResp
|
|
|
|
if err := file.DecodeMsg(mr); err != nil {
|
|
|
|
if errors.Is(err, io.EOF) {
|
|
|
|
err = nil
|
|
|
|
}
|
|
|
|
pr.CloseWithError(err)
|
2024-03-26 02:24:59 -04:00
|
|
|
return toStorageErr(err)
|
2022-07-19 11:35:29 -04:00
|
|
|
}
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return ctx.Err()
|
|
|
|
case resp <- file:
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-11-28 13:20:55 -05:00
|
|
|
// CleanAbandonedData will read metadata of the object on disk
|
|
|
|
// and delete any data directories and inline data that isn't referenced in metadata.
|
|
|
|
func (client *storageRESTClient) CleanAbandonedData(ctx context.Context, volume string, path string) error {
|
|
|
|
values := make(url.Values)
|
|
|
|
values.Set(storageRESTVolume, volume)
|
|
|
|
values.Set(storageRESTFilePath, path)
|
|
|
|
respBody, err := client.call(ctx, storageRESTMethodCleanAbandoned, values, nil, -1)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer xhttp.DrainBody(respBody)
|
2023-11-06 17:26:08 -05:00
|
|
|
_, err = waitForHTTPResponse(respBody)
|
2024-03-26 02:24:59 -04:00
|
|
|
return toStorageErr(err)
|
2022-11-28 13:20:55 -05:00
|
|
|
}
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// Close - marks the client as closed.
|
|
|
|
func (client *storageRESTClient) Close() error {
|
2018-11-20 14:07:19 -05:00
|
|
|
client.restClient.Close()
|
2018-10-04 20:44:06 -04:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2024-04-15 04:25:46 -04:00
|
|
|
var emptyDiskID = ""
|
|
|
|
|
2018-10-04 20:44:06 -04:00
|
|
|
// Returns a storage rest client.
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
func newStorageRESTClient(endpoint Endpoint, healthCheck bool, gm *grid.Manager) (*storageRESTClient, error) {
|
2018-10-04 20:44:06 -04:00
|
|
|
serverURL := &url.URL{
|
2019-11-13 15:17:45 -05:00
|
|
|
Scheme: endpoint.Scheme,
|
2018-10-04 20:44:06 -04:00
|
|
|
Host: endpoint.Host,
|
2019-11-04 12:30:59 -05:00
|
|
|
Path: path.Join(storageRESTPrefix, endpoint.Path, storageRESTVersion),
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|
|
|
|
|
2021-11-23 12:51:53 -05:00
|
|
|
restClient := rest.NewClient(serverURL, globalInternodeTransport, newCachedAuthToken())
|
2020-11-10 12:28:23 -05:00
|
|
|
|
2023-08-01 13:54:26 -04:00
|
|
|
if healthCheck {
|
2020-11-10 12:28:23 -05:00
|
|
|
// Use a separate client to avoid recursive calls.
|
2021-11-23 12:51:53 -05:00
|
|
|
healthClient := rest.NewClient(serverURL, globalInternodeTransport, newCachedAuthToken())
|
2021-06-08 17:09:26 -04:00
|
|
|
healthClient.NoMetrics = true
|
2020-10-26 13:29:29 -04:00
|
|
|
restClient.HealthCheckFn = func() bool {
|
2021-01-28 16:38:12 -05:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), restClient.HealthCheckTimeout)
|
2020-12-13 14:57:08 -05:00
|
|
|
defer cancel()
|
2020-11-10 12:28:23 -05:00
|
|
|
respBody, err := healthClient.Call(ctx, storageRESTMethodHealth, nil, nil, -1)
|
2020-10-26 13:29:29 -04:00
|
|
|
xhttp.DrainBody(respBody)
|
2020-11-19 16:53:49 -05:00
|
|
|
return toStorageErr(err) != errDiskNotFound
|
2020-10-26 13:29:29 -04:00
|
|
|
}
|
2020-06-17 17:49:26 -04:00
|
|
|
}
|
perf: websocket grid connectivity for all internode communication (#18461)
This PR adds a WebSocket grid feature that allows servers to communicate via
a single two-way connection.
There are two request types:
* Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small
roundtrips with small payloads.
* Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`,
which allows for different combinations of full two-way streams with an initial payload.
Only a single stream is created between two machines - and there is, as such, no
server/client relation since both sides can initiate and handle requests. Which server
initiates the request is decided deterministically on the server names.
Requests are made through a mux client and server, which handles message
passing, congestion, cancelation, timeouts, etc.
If a connection is lost, all requests are canceled, and the calling server will try
to reconnect. Registered handlers can operate directly on byte
slices or use a higher-level generics abstraction.
There is no versioning of handlers/clients, and incompatible changes should
be handled by adding new handlers.
The request path can be changed to a new one for any protocol changes.
First, all servers create a "Manager." The manager must know its address
as well as all remote addresses. This will manage all connections.
To get a connection to any remote, ask the manager to provide it given
the remote address using.
```
func (m *Manager) Connection(host string) *Connection
```
All serverside handlers must also be registered on the manager. This will
make sure that all incoming requests are served. The number of in-flight
requests and responses must also be given for streaming requests.
The "Connection" returned manages the mux-clients. Requests issued
to the connection will be sent to the remote.
* `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)`
performs a single request and returns the result. Any deadline provided on the request is
forwarded to the server, and canceling the context will make the function return at once.
* `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)`
will initiate a remote call and send the initial payload.
```Go
// A Stream is a two-way stream.
// All responses *must* be read by the caller.
// If the call is canceled through the context,
//The appropriate error will be returned.
type Stream struct {
// Responses from the remote server.
// Channel will be closed after an error or when the remote closes.
// All responses *must* be read by the caller until either an error is returned or the channel is closed.
// Canceling the context will cause the context cancellation error to be returned.
Responses <-chan Response
// Requests sent to the server.
// If the handler is defined with 0 incoming capacity this will be nil.
// Channel *must* be closed to signal the end of the stream.
// If the request context is canceled, the stream will no longer process requests.
Requests chan<- []byte
}
type Response struct {
Msg []byte
Err error
}
```
There are generic versions of the server/client handlers that allow the use of type
safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
|
|
|
conn := gm.Connection(endpoint.GridHost()).Subroute(endpoint.Path)
|
|
|
|
if conn == nil {
|
|
|
|
return nil, fmt.Errorf("unable to find connection for %s in targets: %v", endpoint.GridHost(), gm.Targets())
|
|
|
|
}
|
2024-04-15 04:25:46 -04:00
|
|
|
client := &storageRESTClient{
|
2024-02-28 12:54:52 -05:00
|
|
|
endpoint: endpoint,
|
|
|
|
restClient: restClient,
|
2024-02-23 12:21:38 -05:00
|
|
|
gridConn: conn,
|
2024-02-23 16:28:14 -05:00
|
|
|
diskInfoCache: cachevalue.New[DiskInfo](),
|
2024-04-15 04:25:46 -04:00
|
|
|
}
|
|
|
|
client.SetDiskID(emptyDiskID)
|
|
|
|
return client, nil
|
2018-10-04 20:44:06 -04:00
|
|
|
}
|