Skip to content

Commit

Permalink
grpc-proxy: cleanup intermediate client logging
Browse files Browse the repository at this point in the history 
Longhorn-4008

Signed-off-by: Chin-Ya Huang <chin-ya.huang@suse.com>
  • Loading branch information
@c3y1huang @innobead
c3y1huang authored and innobead committed May 23, 2022
1 parent e2aa440 commit b0df772
Show file tree
Hide file tree
Showing 5 changed files with 301 additions and 159 deletions.
35 changes: 27 additions & 8 deletions pkg/client/proxy.go
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package client

import (
"fmt"
"time"

"github.com/pkg/errors"
Expand All @@ -17,9 +18,18 @@ import (
)

var (
ErrParameter = errors.Errorf("missing required parameter")
ErrParameterFmt = "missing required %v parameter"
)

func validateProxyMethodParameters(input map[string]string) error {
for k, v := range input {
if v == "" {
return errors.Errorf(ErrParameterFmt, k)
}
}
return nil
}

type ServiceContext struct {
cc *grpc.ClientConn

Expand All @@ -36,7 +46,7 @@ func (s ServiceContext) GetConnectionState() connectivity.State {
func (c *ProxyClient) Close() error {
c.quit()
if err := c.cc.Close(); err != nil {
return errors.Wrapf(err, "error closing proxy gRPC connection")
return errors.Wrap(err, "failed to close proxy gRPC connection")
}
return nil
}
Expand Down Expand Up @@ -80,19 +90,28 @@ const (
GRPCServiceTimeout = 3 * time.Minute
)

func (c *ProxyClient) getProxyErrorPrefix(destination string) string {
return fmt.Sprintf("proxyServer=%v destination=%v:", c.ServiceURL, destination)
}

func (c *ProxyClient) ServerVersionGet(serviceAddress string) (version *emeta.VersionOutput, err error) {
if serviceAddress == "" {
return version, errors.Wrapf(ErrParameter, "failed to get server version")
input := map[string]string{
"serviceAddress": serviceAddress,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get server version")
}
log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debug("Getting server version via proxy")

defer func() {
err = errors.Wrapf(err, "%v failed to get server version", c.getProxyErrorPrefix(serviceAddress))
}()

req := &rpc.ProxyEngineRequest{
Address: serviceAddress,
}
resp, err := c.service.ServerVersionGet(c.ctx, req)
if err != nil {
return version, errors.Wrapf(err, "failed to get server version via proxy %v to %v", c.ServiceURL, serviceAddress)
return nil, err
}

serverVersion := resp.Version
Expand All @@ -111,6 +130,6 @@ func (c *ProxyClient) ServerVersionGet(serviceAddress string) (version *emeta.Ve
}

func (c *ProxyClient) ClientVersionGet() (version emeta.VersionOutput) {
logrus.Debug("Getting client version on proxy")
logrus.Debug("Getting client version")
return emeta.GetVersion()
}
155 changes: 99 additions & 56 deletions pkg/client/proxy_backup.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (

"github.com/golang/protobuf/ptypes"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"

"github.com/longhorn/backupstore"

Expand All @@ -16,12 +15,16 @@ func (c *ProxyClient) SnapshotBackup(serviceAddress,
backupName, snapshotName, backupTarget,
backingImageName, backingImageChecksum string,
labels map[string]string, envs []string) (backupID, replicaAddress string, err error) {
if serviceAddress == "" {
return "", "", errors.Wrapf(ErrParameter, "failed to backup snapshot")
input := map[string]string{
"serviceAddress": serviceAddress,
}
if err := validateProxyMethodParameters(input); err != nil {
return "", "", errors.Wrap(err, "failed to backup snapshot")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Backing up snapshot %v to %v via proxy", snapshotName, backupName)
defer func() {
err = errors.Wrapf(err, "%v failed to backup snapshot %v to %v", c.getProxyErrorPrefix(serviceAddress), snapshotName, backupName)
}()

req := &rpc.EngineSnapshotBackupRequest{
ProxyEngineRequest: &rpc.ProxyEngineRequest{
Expand All @@ -37,19 +40,24 @@ func (c *ProxyClient) SnapshotBackup(serviceAddress,
}
recv, err := c.service.SnapshotBackup(c.ctx, req)
if err != nil {
return "", "", errors.Wrapf(err, "failed to backup snapshot %v to %v via proxy %v to %v", snapshotName, backupName, c.ServiceURL, serviceAddress)
return "", "", err
}

return recv.BackupId, recv.Replica, nil
}

func (c *ProxyClient) SnapshotBackupStatus(serviceAddress, backupName, replicaAddress string) (status *SnapshotBackupStatus, err error) {
if serviceAddress == "" || backupName == "" {
return nil, errors.Wrapf(ErrParameter, "failed to get backup status")
input := map[string]string{
"serviceAddress": serviceAddress,
"backupName": backupName,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get backup status")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Getting %v backup status via proxy", backupName)
defer func() {
err = errors.Wrapf(err, "%v failed to get %v backup status", c.getProxyErrorPrefix(serviceAddress), backupName)
}()

req := &rpc.EngineSnapshotBackupStatusRequest{
ProxyEngineRequest: &rpc.ProxyEngineRequest{
Expand All @@ -60,7 +68,7 @@ func (c *ProxyClient) SnapshotBackupStatus(serviceAddress, backupName, replicaAd
}
recv, err := c.service.SnapshotBackupStatus(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to get %v backup status via proxy %v to %v", backupName, c.ServiceURL, serviceAddress)
return nil, err
}

status = &SnapshotBackupStatus{
Expand All @@ -74,13 +82,20 @@ func (c *ProxyClient) SnapshotBackupStatus(serviceAddress, backupName, replicaAd
return status, nil
}

func (c *ProxyClient) BackupRestore(serviceAddress, url, target, volumeName string, envs []string) error {
if serviceAddress == "" || url == "" || target == "" || volumeName == "" {
return errors.Wrapf(ErrParameter, "failed to restore backup to volume")
func (c *ProxyClient) BackupRestore(serviceAddress, url, target, volumeName string, envs []string) (err error) {
input := map[string]string{
"serviceAddress": serviceAddress,
"url": url,
"target": target,
"volumeName": volumeName,
}
if err := validateProxyMethodParameters(input); err != nil {
return errors.Wrap(err, "failed to restore backup to volume")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Restoring %v backup to %v via proxy", url, volumeName)
defer func() {
err = errors.Wrapf(err, "%v failed to restore backup %v to volume %v", c.getProxyErrorPrefix(serviceAddress), url, volumeName)
}()

req := &rpc.EngineBackupRestoreRequest{
ProxyEngineRequest: &rpc.ProxyEngineRequest{
Expand All @@ -93,35 +108,41 @@ func (c *ProxyClient) BackupRestore(serviceAddress, url, target, volumeName stri
}
recv, err := c.service.BackupRestore(c.ctx, req)
if err != nil {
return errors.Wrapf(err, "failed to restore backup %v to %v via proxy %v to %v", url, volumeName, c.ServiceURL, serviceAddress)
return err
}

if recv.TaskError != nil {
var taskErr TaskError
if jsonErr := json.Unmarshal(recv.TaskError, &taskErr); jsonErr != nil {
return errors.Wrapf(jsonErr, "Cannot unmarshal the restore error, maybe it's not caused by the replica restore failure")
err = errors.Wrap(jsonErr, "Cannot unmarshal the restore error, maybe it's not caused by the replica restore failure")
return err
}

return taskErr
err = taskErr
return err
}

return nil
}

func (c *ProxyClient) BackupRestoreStatus(serviceAddress string) (status map[string]*BackupRestoreStatus, err error) {
if serviceAddress == "" {
return nil, errors.Wrapf(ErrParameter, "failed to get backup restore status")
input := map[string]string{
"serviceAddress": serviceAddress,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get backup restore status")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debug("Getting backup restore status via proxy")
defer func() {
err = errors.Wrapf(err, "%v failed to get backup restore status", c.getProxyErrorPrefix(serviceAddress))
}()

req := &rpc.ProxyEngineRequest{
Address: serviceAddress,
}
recv, err := c.service.BackupRestoreStatus(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to get backup restore status via proxy %v to %v", c.ServiceURL, serviceAddress)
return nil, err
}

status = map[string]*BackupRestoreStatus{}
Expand All @@ -141,40 +162,48 @@ func (c *ProxyClient) BackupRestoreStatus(serviceAddress string) (status map[str
}

func (c *ProxyClient) BackupGet(destURL string, envs []string) (info *EngineBackupInfo, err error) {
if destURL == "" {
return nil, errors.Wrapf(ErrParameter, "failed to get backup")
input := map[string]string{
"destURL": destURL,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get backup")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Getting %v backup via proxy", destURL)
defer func() {
err = errors.Wrapf(err, "%v failed to get backup", c.getProxyErrorPrefix(destURL))
}()

req := &rpc.EngineBackupGetRequest{
Envs: envs,
DestUrl: destURL,
}
recv, err := c.service.BackupGet(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to get %v backup via proxy %v", destURL, c.ServiceURL)
return nil, err
}

return parseBackup(recv.Backup), nil
}

func (c *ProxyClient) BackupVolumeGet(destURL string, envs []string) (info *EngineBackupVolumeInfo, err error) {
if destURL == "" {
return nil, errors.Wrapf(ErrParameter, "failed to get backup volume")
input := map[string]string{
"destURL": destURL,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get backup volume")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Getting %v backup volume via proxy", destURL)
defer func() {
err = errors.Wrapf(err, "%v failed to get backup volume", c.getProxyErrorPrefix(destURL))
}()

req := &rpc.EngineBackupVolumeGetRequest{
Envs: envs,
DestUrl: destURL,
}
recv, err := c.service.BackupVolumeGet(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to get %v backup volume via proxy %v", destURL, c.ServiceURL)
return nil, err
}

info = &EngineBackupVolumeInfo{
Expand All @@ -194,15 +223,20 @@ func (c *ProxyClient) BackupVolumeGet(destURL string, envs []string) (info *Engi
}

func (c *ProxyClient) BackupVolumeList(destURL, volumeName string, volumeOnly bool, envs []string) (info map[string]*EngineBackupVolumeInfo, err error) {
if destURL == "" {
return nil, errors.Wrapf(ErrParameter, "failed to list backup volumes")
input := map[string]string{
"destURL": destURL,
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
if volumeName != "" {
log = log.WithField("volume", volumeName)
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to list backup for volumes")
}
log.Debugf("Listing %v backup volumes via proxy", destURL)

defer func() {
if volumeName == "" {
err = errors.Wrapf(err, "%v failed to list backup for volumes", c.getProxyErrorPrefix(destURL))
} else {
err = errors.Wrapf(err, "%v failed to list backup for volume %v", c.getProxyErrorPrefix(destURL), volumeName)
}
}()

req := &rpc.EngineBackupVolumeListRequest{
Envs: envs,
Expand All @@ -212,7 +246,7 @@ func (c *ProxyClient) BackupVolumeList(destURL, volumeName string, volumeOnly bo
}
recv, err := c.service.BackupVolumeList(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to list %v backup volumes via proxy %v", destURL, c.ServiceURL)
return nil, err
}

info = map[string]*EngineBackupVolumeInfo{}
Expand Down Expand Up @@ -261,25 +295,29 @@ func parseBackup(in *rpc.EngineBackupInfo) (out *EngineBackupInfo) {
}

func (c *ProxyClient) BackupConfigMetaGet(destURL string, envs []string) (meta *backupstore.ConfigMetadata, err error) {
if destURL == "" {
return nil, errors.Wrapf(ErrParameter, "failed to get backup config metadata")
input := map[string]string{
"destURL": destURL,
}
if err := validateProxyMethodParameters(input); err != nil {
return nil, errors.Wrap(err, "failed to get backup config metadata")
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
log.Debugf("Getting %v backup config metadata via proxy", destURL)
defer func() {
err = errors.Wrapf(err, "%v failed to get backup config metadata", c.getProxyErrorPrefix(destURL))
}()

req := &rpc.EngineBackupConfigMetaGetRequest{
Envs: envs,
DestUrl: destURL,
}
recv, err := c.service.BackupConfigMetaGet(c.ctx, req)
if err != nil {
return nil, errors.Wrapf(err, "failed to get %v backup config metadata via proxy %v", destURL, c.ServiceURL)
return nil, err
}

ts, err := ptypes.Timestamp(recv.ModificationTime)
if err != nil {
return nil, errors.Wrapf(err, "failed convert protobuf timestamp %v", recv.ModificationTime)
return nil, err
}

return &backupstore.ConfigMetadata{
Expand All @@ -288,15 +326,20 @@ func (c *ProxyClient) BackupConfigMetaGet(destURL string, envs []string) (meta *
}

func (c *ProxyClient) BackupRemove(destURL, volumeName string, envs []string) (err error) {
if destURL == "" {
return errors.Wrapf(ErrParameter, "failed to remove backup")
input := map[string]string{
"destURL": destURL,
}

log := logrus.WithFields(logrus.Fields{"serviceURL": c.ServiceURL})
if volumeName != "" {
log = log.WithField("volume", volumeName)
if err := validateProxyMethodParameters(input); err != nil {
return errors.Wrap(err, "failed to remove backup")
}
log.Debugf("Removing %v backup via proxy", destURL)

defer func() {
if volumeName == "" {
err = errors.Wrapf(err, "%v failed to remove backup", c.getProxyErrorPrefix(destURL))
} else {
err = errors.Wrapf(err, "%v failed to remove backup for volume %v", c.getProxyErrorPrefix(destURL), volumeName)
}
}()

req := &rpc.EngineBackupRemoveRequest{
Envs: envs,
Expand All @@ -305,7 +348,7 @@ func (c *ProxyClient) BackupRemove(destURL, volumeName string, envs []string) (e
}
_, err = c.service.BackupRemove(c.ctx, req)
if err != nil {
return errors.Wrapf(err, "failed to remove %v backup via proxy %v", destURL, c.ServiceURL)
return err
}

return nil
Expand Down
Loading

0 comments on commit b0df772

Please sign in to comment.