mirror of
https://github.com/pgsty/minio.git
synced 2026-07-21 13:10:22 +03:00
Convert more peer <--> peer REST calls (#19004)
* Convert more peer <--> peer REST calls * Clean up in general. * Add JSON wrapper. * Add slice wrapper. * Add option to make handler return nil error if no connection is given, `IgnoreNilConn`. Converts the following: ``` + HandlerGetMetrics + HandlerGetResourceMetrics + HandlerGetMemInfo + HandlerGetProcInfo + HandlerGetOSInfo + HandlerGetPartitions + HandlerGetNetInfo + HandlerGetCPUs + HandlerServerInfo + HandlerGetSysConfig + HandlerGetSysServices + HandlerGetSysErrors + HandlerGetAllBucketStats + HandlerGetBucketStats + HandlerGetSRMetrics + HandlerGetPeerMetrics + HandlerGetMetacacheListing + HandlerUpdateMetacacheListing + HandlerGetPeerBucketMetrics + HandlerStorageInfo + HandlerGetLocks + HandlerBackgroundHealStatus + HandlerGetLastDayTierStats + HandlerSignalService + HandlerGetBandwidth ```
This commit is contained in:
+113
-321
@@ -18,7 +18,6 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/gob"
|
||||
"encoding/hex"
|
||||
@@ -28,7 +27,6 @@ import (
|
||||
"io"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -38,9 +36,7 @@ import (
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
"github.com/minio/minio/internal/logger"
|
||||
"github.com/minio/minio/internal/rest"
|
||||
"github.com/minio/pkg/v2/logger/message/log"
|
||||
xnet "github.com/minio/pkg/v2/net"
|
||||
"github.com/tinylib/msgp/msgp"
|
||||
)
|
||||
|
||||
// client to talk to peer Nodes.
|
||||
@@ -154,111 +150,65 @@ func (client *peerRESTClient) Close() error {
|
||||
|
||||
// GetLocks - fetch older locks for a remote node.
|
||||
func (client *peerRESTClient) GetLocks() (lockMap map[string][]lockRequesterInfo, err error) {
|
||||
respBody, err := client.call(peerRESTMethodGetLocks, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
resp, err := getLocksRPC.Call(context.Background(), client.gridConn(), grid.NewMSS())
|
||||
if err != nil || resp == nil {
|
||||
return nil, err
|
||||
}
|
||||
lockMap = map[string][]lockRequesterInfo{}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&lockMap)
|
||||
return lockMap, err
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
// LocalStorageInfo - fetch server information for a remote node.
|
||||
func (client *peerRESTClient) LocalStorageInfo(metrics bool) (info StorageInfo, err error) {
|
||||
values := make(url.Values)
|
||||
values.Set(peerRESTMetrics, strconv.FormatBool(metrics))
|
||||
respBody, err := client.call(peerRESTMethodLocalStorageInfo, values, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := localStorageInfoRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTMetrics: strconv.FormatBool(metrics),
|
||||
}))
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// ServerInfo - fetch server information for a remote node.
|
||||
func (client *peerRESTClient) ServerInfo(metrics bool) (info madmin.ServerProperties, err error) {
|
||||
values := make(url.Values)
|
||||
values.Set(peerRESTMetrics, strconv.FormatBool(metrics))
|
||||
|
||||
respBody, err := client.call(peerRESTMethodServerInfo, values, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := serverInfoRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{peerRESTMetrics: strconv.FormatBool(metrics)}))
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetCPUs - fetch CPU information for a remote node.
|
||||
func (client *peerRESTClient) GetCPUs(ctx context.Context) (info madmin.CPUs, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodCPUInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getCPUsHandler.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetNetInfo - fetch network information for a remote node.
|
||||
func (client *peerRESTClient) GetNetInfo(ctx context.Context) (info madmin.NetInfo, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodNetHwInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getNetInfoRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetPartitions - fetch disk partition information for a remote node.
|
||||
func (client *peerRESTClient) GetPartitions(ctx context.Context) (info madmin.Partitions, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodDiskHwInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getPartitionsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetOSInfo - fetch OS information for a remote node.
|
||||
func (client *peerRESTClient) GetOSInfo(ctx context.Context) (info madmin.OSInfo, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodOsInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getOSInfoRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetSELinuxInfo - fetch SELinux information for a remote node.
|
||||
func (client *peerRESTClient) GetSELinuxInfo(ctx context.Context) (info madmin.SysServices, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodSysServices, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getSysServicesRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetSysConfig - fetch sys config for a remote node.
|
||||
func (client *peerRESTClient) GetSysConfig(ctx context.Context) (info madmin.SysConfig, err error) {
|
||||
sent := time.Now()
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodSysConfig, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
roundtrip := int32(time.Since(sent).Milliseconds())
|
||||
defer xhttp.DrainBody(respBody)
|
||||
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
resp, err := getSysConfigRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
info = resp.ValueOrZero()
|
||||
if ti, ok := info.Config["time-info"].(madmin.TimeInfo); ok {
|
||||
ti.RoundtripDuration = roundtrip
|
||||
rt := int32(time.Since(sent).Milliseconds())
|
||||
ti.RoundtripDuration = rt
|
||||
info.Config["time-info"] = ti
|
||||
}
|
||||
return info, err
|
||||
@@ -266,24 +216,14 @@ func (client *peerRESTClient) GetSysConfig(ctx context.Context) (info madmin.Sys
|
||||
|
||||
// GetSysErrors - fetch sys errors for a remote node.
|
||||
func (client *peerRESTClient) GetSysErrors(ctx context.Context) (info madmin.SysErrors, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodSysErrors, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getSysErrorsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetMemInfo - fetch memory information for a remote node.
|
||||
func (client *peerRESTClient) GetMemInfo(ctx context.Context) (info madmin.MemInfo, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodMemInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getMemInfoRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetMetrics - fetch metrics from a remote node.
|
||||
@@ -298,52 +238,34 @@ func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricTyp
|
||||
}
|
||||
values.Set(peerRESTJobID, opts.jobID)
|
||||
values.Set(peerRESTDepID, opts.depID)
|
||||
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodMetrics, values, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
v, err := getMetricsRPC.Call(ctx, client.gridConn(), grid.NewURLValuesWith(values))
|
||||
return v.ValueOrZero(), err
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) GetResourceMetrics(ctx context.Context) (<-chan Metric, error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodResourceMetrics, nil, nil, -1)
|
||||
st, err := getResourceMetricsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dec := gob.NewDecoder(respBody)
|
||||
ch := make(chan Metric)
|
||||
ch := make(chan Metric, 1)
|
||||
go func(ch chan<- Metric) {
|
||||
defer func() {
|
||||
xhttp.DrainBody(respBody)
|
||||
close(ch)
|
||||
}()
|
||||
for {
|
||||
var metric Metric
|
||||
if err := dec.Decode(&metric); err != nil {
|
||||
return
|
||||
}
|
||||
defer close(ch)
|
||||
st.Results(func(metric *Metric) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ch <- metric:
|
||||
return ctx.Err()
|
||||
case ch <- *metric:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
})
|
||||
}(ch)
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
// GetProcInfo - fetch MinIO process information for a remote node.
|
||||
func (client *peerRESTClient) GetProcInfo(ctx context.Context) (info madmin.ProcInfo, err error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodProcInfo, nil, nil, -1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
err = gob.NewDecoder(respBody).Decode(&info)
|
||||
return info, err
|
||||
resp, err := getProcInfoRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// StartProfiling - Issues profiling command on the peer node.
|
||||
@@ -371,51 +293,36 @@ func (client *peerRESTClient) DownloadProfileData() (data map[string][]byte, err
|
||||
|
||||
// GetBucketStats - load bucket statistics
|
||||
func (client *peerRESTClient) GetBucketStats(bucket string) (BucketStats, error) {
|
||||
values := make(url.Values)
|
||||
values.Set(peerRESTBucket, bucket)
|
||||
respBody, err := client.call(peerRESTMethodGetBucketStats, values, nil, -1)
|
||||
if err != nil {
|
||||
resp, err := getBucketStatsRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTBucket: bucket,
|
||||
}))
|
||||
if err != nil || resp == nil {
|
||||
return BucketStats{}, err
|
||||
}
|
||||
|
||||
var bs BucketStats
|
||||
defer xhttp.DrainBody(respBody)
|
||||
return bs, msgp.Decode(respBody, &bs)
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
// GetSRMetrics- loads site replication metrics, optionally for a specific bucket
|
||||
// GetSRMetrics loads site replication metrics, optionally for a specific bucket
|
||||
func (client *peerRESTClient) GetSRMetrics() (SRMetricsSummary, error) {
|
||||
values := make(url.Values)
|
||||
respBody, err := client.call(peerRESTMethodGetSRMetrics, values, nil, -1)
|
||||
if err != nil {
|
||||
resp, err := getSRMetricsRPC.Call(context.Background(), client.gridConn(), grid.NewMSS())
|
||||
if err != nil || resp == nil {
|
||||
return SRMetricsSummary{}, err
|
||||
}
|
||||
|
||||
var sm SRMetricsSummary
|
||||
defer xhttp.DrainBody(respBody)
|
||||
return sm, msgp.Decode(respBody, &sm)
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
// GetAllBucketStats - load replication stats for all buckets
|
||||
func (client *peerRESTClient) GetAllBucketStats() (BucketStatsMap, error) {
|
||||
values := make(url.Values)
|
||||
respBody, err := client.call(peerRESTMethodGetAllBucketStats, values, nil, -1)
|
||||
if err != nil {
|
||||
resp, err := getAllBucketStatsRPC.Call(context.Background(), client.gridConn(), grid.NewMSS())
|
||||
if err != nil || resp == nil {
|
||||
return BucketStatsMap{}, err
|
||||
}
|
||||
|
||||
bsMap := BucketStatsMap{}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
return bsMap, msgp.Decode(respBody, &bsMap)
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
// LoadBucketMetadata - load bucket metadata
|
||||
func (client *peerRESTClient) LoadBucketMetadata(bucket string) error {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := loadBucketMetadataHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err := loadBucketMetadataRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTBucket: bucket,
|
||||
}))
|
||||
return err
|
||||
@@ -423,11 +330,7 @@ func (client *peerRESTClient) LoadBucketMetadata(bucket string) error {
|
||||
|
||||
// DeleteBucketMetadata - Delete bucket metadata
|
||||
func (client *peerRESTClient) DeleteBucketMetadata(bucket string) error {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := deleteBucketMetadataHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err := deleteBucketMetadataRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTBucket: bucket,
|
||||
}))
|
||||
return err
|
||||
@@ -435,12 +338,7 @@ func (client *peerRESTClient) DeleteBucketMetadata(bucket string) error {
|
||||
|
||||
// DeletePolicy - delete a specific canned policy.
|
||||
func (client *peerRESTClient) DeletePolicy(policyName string) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = deletePolicyHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = deletePolicyRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTPolicy: policyName,
|
||||
}))
|
||||
return err
|
||||
@@ -448,12 +346,7 @@ func (client *peerRESTClient) DeletePolicy(policyName string) (err error) {
|
||||
|
||||
// LoadPolicy - reload a specific canned policy.
|
||||
func (client *peerRESTClient) LoadPolicy(policyName string) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = loadPolicyHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = loadPolicyRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTPolicy: policyName,
|
||||
}))
|
||||
return err
|
||||
@@ -461,12 +354,7 @@ func (client *peerRESTClient) LoadPolicy(policyName string) (err error) {
|
||||
|
||||
// LoadPolicyMapping - reload a specific policy mapping
|
||||
func (client *peerRESTClient) LoadPolicyMapping(userOrGroup string, userType IAMUserType, isGroup bool) error {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err := loadPolicyMappingHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err := loadPolicyMappingRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTUserOrGroup: userOrGroup,
|
||||
peerRESTUserType: strconv.Itoa(int(userType)),
|
||||
peerRESTIsGroup: strconv.FormatBool(isGroup),
|
||||
@@ -476,12 +364,7 @@ func (client *peerRESTClient) LoadPolicyMapping(userOrGroup string, userType IAM
|
||||
|
||||
// DeleteUser - delete a specific user.
|
||||
func (client *peerRESTClient) DeleteUser(accessKey string) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = deleteUserHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = deleteUserRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTUser: accessKey,
|
||||
}))
|
||||
return err
|
||||
@@ -489,12 +372,7 @@ func (client *peerRESTClient) DeleteUser(accessKey string) (err error) {
|
||||
|
||||
// DeleteServiceAccount - delete a specific service account.
|
||||
func (client *peerRESTClient) DeleteServiceAccount(accessKey string) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = deleteSvcActHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = deleteSvcActRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTUser: accessKey,
|
||||
}))
|
||||
return err
|
||||
@@ -502,12 +380,7 @@ func (client *peerRESTClient) DeleteServiceAccount(accessKey string) (err error)
|
||||
|
||||
// LoadUser - reload a specific user.
|
||||
func (client *peerRESTClient) LoadUser(accessKey string, temp bool) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = loadUserHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = loadUserRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTUser: accessKey,
|
||||
peerRESTUserTemp: strconv.FormatBool(temp),
|
||||
}))
|
||||
@@ -516,12 +389,7 @@ func (client *peerRESTClient) LoadUser(accessKey string, temp bool) (err error)
|
||||
|
||||
// LoadServiceAccount - reload a specific service account.
|
||||
func (client *peerRESTClient) LoadServiceAccount(accessKey string) (err error) {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err = loadSvcActHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err = loadSvcActRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTUser: accessKey,
|
||||
}))
|
||||
return err
|
||||
@@ -529,12 +397,7 @@ func (client *peerRESTClient) LoadServiceAccount(accessKey string) (err error) {
|
||||
|
||||
// LoadGroup - send load group command to peers.
|
||||
func (client *peerRESTClient) LoadGroup(group string) error {
|
||||
conn := client.gridConn()
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err := loadGroupHandler.Call(context.Background(), conn, grid.NewMSSWith(map[string]string{
|
||||
_, err := loadGroupRPC.Call(context.Background(), client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTGroup: group,
|
||||
}))
|
||||
return err
|
||||
@@ -546,7 +409,7 @@ func (client *peerRESTClient) ReloadSiteReplicationConfig(ctx context.Context) e
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err := reloadSiteReplicationConfigHandler.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
_, err := reloadSiteReplicationConfigRPC.Call(ctx, conn, grid.NewMSS())
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -577,28 +440,17 @@ func (client *peerRESTClient) CommitBinary(ctx context.Context) error {
|
||||
|
||||
// SignalService - sends signal to peer nodes.
|
||||
func (client *peerRESTClient) SignalService(sig serviceSignal, subSys string, dryRun bool) error {
|
||||
values := make(url.Values)
|
||||
values := grid.NewMSS()
|
||||
values.Set(peerRESTSignal, strconv.Itoa(int(sig)))
|
||||
values.Set(peerRESTDryRun, strconv.FormatBool(dryRun))
|
||||
values.Set(peerRESTSubSys, subSys)
|
||||
respBody, err := client.call(peerRESTMethodSignalService, values, nil, -1)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
return nil
|
||||
_, err := signalServiceRPC.Call(context.Background(), client.gridConn(), values)
|
||||
return err
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) BackgroundHealStatus() (madmin.BgHealState, error) {
|
||||
respBody, err := client.call(peerRESTMethodBackgroundHealStatus, nil, nil, -1)
|
||||
if err != nil {
|
||||
return madmin.BgHealState{}, err
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
|
||||
state := madmin.BgHealState{}
|
||||
err = gob.NewDecoder(respBody).Decode(&state)
|
||||
return state, err
|
||||
resp, err := getBackgroundHealStatusRPC.Call(context.Background(), client.gridConn(), grid.NewMSS())
|
||||
return resp.ValueOrZero(), err
|
||||
}
|
||||
|
||||
// GetMetacacheListing - get a new or existing metacache.
|
||||
@@ -607,19 +459,7 @@ func (client *peerRESTClient) GetMetacacheListing(ctx context.Context, o listPat
|
||||
resp := localMetacacheMgr.getBucket(ctx, o.Bucket).findCache(o)
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
var reader bytes.Buffer
|
||||
err := gob.NewEncoder(&reader).Encode(o)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodGetMetacacheListing, nil, &reader, int64(reader.Len()))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var resp metacache
|
||||
defer xhttp.DrainBody(respBody)
|
||||
return &resp, msgp.Decode(respBody, &resp)
|
||||
return getMetacacheListingRPC.Call(ctx, client.gridConn(), &o)
|
||||
}
|
||||
|
||||
// UpdateMetacacheListing - update an existing metacache it will unconditionally be updated to the new state.
|
||||
@@ -627,17 +467,11 @@ func (client *peerRESTClient) UpdateMetacacheListing(ctx context.Context, m meta
|
||||
if client == nil {
|
||||
return localMetacacheMgr.updateCacheEntry(m)
|
||||
}
|
||||
b, err := m.MarshalMsg(nil)
|
||||
if err != nil {
|
||||
return m, err
|
||||
resp, err := updateMetacacheListingRPC.Call(ctx, client.gridConn(), &m)
|
||||
if err != nil || resp == nil {
|
||||
return metacache{}, err
|
||||
}
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodUpdateMetacacheListing, nil, bytes.NewBuffer(b), int64(len(b)))
|
||||
if err != nil {
|
||||
return m, err
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
var resp metacache
|
||||
return resp, msgp.Decode(respBody, &resp)
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) ReloadPoolMeta(ctx context.Context) error {
|
||||
@@ -645,7 +479,7 @@ func (client *peerRESTClient) ReloadPoolMeta(ctx context.Context) error {
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := reloadPoolMetaHandler.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
_, err := reloadPoolMetaRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -654,7 +488,7 @@ func (client *peerRESTClient) StopRebalance(ctx context.Context) error {
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := stopRebalanceHandler.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
_, err := stopRebalanceRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -663,7 +497,7 @@ func (client *peerRESTClient) LoadRebalanceMeta(ctx context.Context, startRebala
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := loadRebalanceMetaHandler.Call(ctx, conn, grid.NewMSSWith(map[string]string{
|
||||
_, err := loadRebalanceMetaRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{
|
||||
peerRESTStartRebalance: strconv.FormatBool(startRebalance),
|
||||
}))
|
||||
return err
|
||||
@@ -674,7 +508,7 @@ func (client *peerRESTClient) LoadTransitionTierConfig(ctx context.Context) erro
|
||||
if conn == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := loadTransitionTierConfigHandler.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
_, err := loadTransitionTierConfigRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{}))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -711,7 +545,7 @@ func (client *peerRESTClient) doListen(ctx context.Context, listenCh chan<- []by
|
||||
if conn == nil {
|
||||
return
|
||||
}
|
||||
st, err := listenHandler.Call(ctx, conn, grid.NewURLValuesWith(v))
|
||||
st, err := listenRPC.Call(ctx, conn, grid.NewURLValuesWith(v))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -759,48 +593,31 @@ func (client *peerRESTClient) Trace(ctx context.Context, traceCh chan<- []byte,
|
||||
}()
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) doConsoleLog(logCh chan log.Info, doneCh <-chan struct{}) {
|
||||
// To cancel the REST request in case doneCh gets closed.
|
||||
ctx, cancel := context.WithCancel(GlobalContext)
|
||||
|
||||
cancelCh := make(chan struct{})
|
||||
defer close(cancelCh)
|
||||
go func() {
|
||||
select {
|
||||
case <-doneCh:
|
||||
case <-cancelCh:
|
||||
// There was an error in the REST request.
|
||||
}
|
||||
cancel()
|
||||
}()
|
||||
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodLog, nil, nil, -1)
|
||||
defer xhttp.DrainBody(respBody)
|
||||
func (client *peerRESTClient) doConsoleLog(ctx context.Context, kind madmin.LogMask, logCh chan<- []byte) {
|
||||
st, err := consoleLogRPC.Call(ctx, client.gridConn(), grid.NewMSSWith(map[string]string{
|
||||
peerRESTLogMask: strconv.Itoa(int(kind)),
|
||||
}))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
dec := gob.NewDecoder(respBody)
|
||||
for {
|
||||
var lg log.Info
|
||||
if err = dec.Decode(&lg); err != nil {
|
||||
break
|
||||
}
|
||||
st.Results(func(b *grid.Bytes) error {
|
||||
select {
|
||||
case logCh <- lg:
|
||||
case logCh <- *b:
|
||||
default:
|
||||
consoleLogRPC.PutResponse(b)
|
||||
// Do not block on slow receivers.
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// ConsoleLog - sends request to peer nodes to get console logs
|
||||
func (client *peerRESTClient) ConsoleLog(logCh chan log.Info, doneCh <-chan struct{}) {
|
||||
func (client *peerRESTClient) ConsoleLog(ctx context.Context, kind madmin.LogMask, logCh chan<- []byte) {
|
||||
go func() {
|
||||
for {
|
||||
client.doConsoleLog(logCh, doneCh)
|
||||
client.doConsoleLog(ctx, kind, logCh)
|
||||
select {
|
||||
case <-doneCh:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
// There was error in the REST request, retry after sometime as probably the peer is down.
|
||||
@@ -838,71 +655,53 @@ func newPeerRestClients(endpoints EndpointServerPools) (remote, all []*peerRESTC
|
||||
|
||||
// MonitorBandwidth - send http trace request to peer nodes
|
||||
func (client *peerRESTClient) MonitorBandwidth(ctx context.Context, buckets []string) (*bandwidth.BucketBandwidthReport, error) {
|
||||
values := make(url.Values)
|
||||
values.Set(peerRESTBuckets, strings.Join(buckets, ","))
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodGetBandwidth, values, nil, -1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
|
||||
dec := gob.NewDecoder(respBody)
|
||||
var bandwidthReport bandwidth.BucketBandwidthReport
|
||||
err = dec.Decode(&bandwidthReport)
|
||||
return &bandwidthReport, err
|
||||
values := grid.NewURLValuesWith(map[string][]string{
|
||||
peerRESTBuckets: buckets,
|
||||
})
|
||||
return getBandwidthRPC.Call(ctx, client.gridConn(), values)
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) GetPeerMetrics(ctx context.Context) (<-chan Metric, error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodGetPeerMetrics, nil, nil, -1)
|
||||
resp, err := getPeerMetricsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dec := gob.NewDecoder(respBody)
|
||||
ch := make(chan Metric)
|
||||
go func(ch chan<- Metric) {
|
||||
defer func() {
|
||||
xhttp.DrainBody(respBody)
|
||||
close(ch)
|
||||
}()
|
||||
for {
|
||||
var metric Metric
|
||||
if err := dec.Decode(&metric); err != nil {
|
||||
return
|
||||
go func() {
|
||||
defer close(ch)
|
||||
for _, m := range resp.Value() {
|
||||
if m == nil {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ch <- metric:
|
||||
case ch <- *m:
|
||||
}
|
||||
}
|
||||
}(ch)
|
||||
}()
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) GetPeerBucketMetrics(ctx context.Context) (<-chan Metric, error) {
|
||||
respBody, err := client.callWithContext(ctx, peerRESTMethodGetPeerBucketMetrics, nil, nil, -1)
|
||||
resp, err := getPeerBucketMetricsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dec := gob.NewDecoder(respBody)
|
||||
ch := make(chan Metric)
|
||||
go func(ch chan<- Metric) {
|
||||
defer func() {
|
||||
xhttp.DrainBody(respBody)
|
||||
close(ch)
|
||||
}()
|
||||
for {
|
||||
var metric Metric
|
||||
if err := dec.Decode(&metric); err != nil {
|
||||
return
|
||||
go func() {
|
||||
defer close(ch)
|
||||
for _, m := range resp.Value() {
|
||||
if m == nil {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ch <- metric:
|
||||
case ch <- *m:
|
||||
}
|
||||
}
|
||||
}(ch)
|
||||
}()
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
@@ -966,18 +765,11 @@ func (client *peerRESTClient) DriveSpeedTest(ctx context.Context, opts madmin.Dr
|
||||
}
|
||||
|
||||
func (client *peerRESTClient) GetLastDayTierStats(ctx context.Context) (DailyAllTierStats, error) {
|
||||
var result map[string]lastDayTierStats
|
||||
respBody, err := client.callWithContext(context.Background(), peerRESTMethodGetLastDayTierStats, nil, nil, -1)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
defer xhttp.DrainBody(respBody)
|
||||
|
||||
err = gob.NewDecoder(respBody).Decode(&result)
|
||||
if err != nil {
|
||||
resp, err := getLastDayTierStatsRPC.Call(ctx, client.gridConn(), grid.NewMSS())
|
||||
if err != nil || resp == nil {
|
||||
return DailyAllTierStats{}, err
|
||||
}
|
||||
return DailyAllTierStats(result), nil
|
||||
return *resp, nil
|
||||
}
|
||||
|
||||
// DevNull - Used by netperf to pump data to peer
|
||||
|
||||
Reference in New Issue
Block a user