internal,lib,uplink: add monkit task to missing places (#2118)

* internal,lib,uplink: add monkit task to missing places

Change-Id: I490053eee4ed517502f9fe00c6394f0095bd13d0

* Include Monkit

* Add missing context

* Another missing ctx

* More ctx missing

* Linting

* go imports

Change-Id: Ibf0ed072eba339f027727ed8039f7bce1f223fa7

* fix semantic merge conflict

Change-Id: I67fb1f4e7b6cd5e89d69987ed7b3966b7d30ee37

internal/post/oauth2/auth.go

@@ -4,6 +4,7 @@
 package oauth2
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -14,6 +15,11 @@ import (
 	"time"
 
 	"github.com/zeebo/errs"
+	monkit "gopkg.in/spacemonkeygo/monkit.v2"
+)
+
+var (
+	mon = monkit.Package()
 )
 
 // Auth is XOAUTH2 implementation of smtp.Auth interface
@@ -25,11 +31,13 @@ type Auth struct {
 
 // Start returns proto and auth credentials for first auth msg
 func (auth *Auth) Start(server *smtp.ServerInfo) (proto string, toServer []byte, err error) {
+	ctx := context.TODO()
+	defer mon.Task()(&ctx)(&err)
 	if !server.TLS {
 		return "", nil, errs.New("unencrypted connection")
 	}
 
-	token, err := auth.Storage.Token()
+	token, err := auth.Storage.Token(ctx)
 	if err != nil {
 		return "", nil, err
 	}
@@ -78,14 +86,15 @@ func NewTokenStore(creds Credentials, token Token) *TokenStore {
 }
 
 // Token retrieves token in a thread safe way and refreshes it if needed
-func (s *TokenStore) Token() (*Token, error) {
+func (s *TokenStore) Token(ctx context.Context) (_ *Token, err error) {
+	defer mon.Task()(&ctx)(&err)
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	token := new(Token)
 	if s.token.Expiry.Before(time.Now()) {
 		var err error
-		token, err = RefreshToken(s.creds, s.token.RefreshToken)
+		token, err = RefreshToken(ctx, s.creds, s.token.RefreshToken)
 		if err != nil {
 			return nil, err
 		}
@@ -97,7 +106,9 @@ func (s *TokenStore) Token() (*Token, error) {
 }
 
 // RefreshToken is a helper method that refreshes token with given credentials and OUATH2 refresh token
-func RefreshToken(creds Credentials, refreshToken string) (*Token, error) {
+func RefreshToken(ctx context.Context, creds Credentials, refreshToken string) (_ *Token, err error) {
+	defer mon.Task()(&ctx)(&err)
+
 	values := url.Values{
 		"grant_type":    {"refresh_token"},
 		"refresh_token": {refreshToken},

internal/sync2/copy.go

@@ -6,14 +6,19 @@ package sync2
 import (
 	"context"
 	"io"
+
+	monkit "gopkg.in/spacemonkeygo/monkit.v2"
 )
 
+var mon = monkit.Package()
+
 type readerFunc func(p []byte) (n int, err error)
 
 func (rf readerFunc) Read(p []byte) (n int, err error) { return rf(p) }
 
 // Copy implements copying with cancellation
 func Copy(ctx context.Context, dst io.Writer, src io.Reader) (written int64, err error) {
+	defer mon.Task()(&ctx)(&err)
 	written, err = io.Copy(dst, readerFunc(func(p []byte) (int, error) {
 		select {
 		case <-ctx.Done():

internal/version/service.go

@@ -49,7 +49,8 @@ func NewService(config Config, info Info, service string) (client *Service) {
 }
 
 // CheckVersion checks to make sure the version is still okay, returning an error if not
-func (srv *Service) CheckVersion(ctx context.Context) error {
+func (srv *Service) CheckVersion(ctx context.Context) (err error) {
+	defer mon.Task()(&ctx)(&err)
 	if !srv.checkVersion(ctx) {
 		return fmt.Errorf("outdated software version (%v), please update", srv.info.Version.String())
 	}
@@ -58,12 +59,14 @@ func (srv *Service) CheckVersion(ctx context.Context) error {
 
 // CheckProcessVersion is not meant to be used for peers but is meant to be
 // used for other utilities
-func CheckProcessVersion(ctx context.Context, config Config, info Info, service string) error {
+func CheckProcessVersion(ctx context.Context, config Config, info Info, service string) (err error) {
+	defer mon.Task()(&ctx)(&err)
 	return NewService(config, info, service).CheckVersion(ctx)
 }
 
 // Run logs the current version information
-func (srv *Service) Run(ctx context.Context) error {
+func (srv *Service) Run(ctx context.Context) (err error) {
+	defer mon.Task()(&ctx)(&err)
 	if !srv.checked.Released() {
 		err := srv.CheckVersion(ctx)
 		if err != nil {

lib/uplink/object.go

@@ -80,7 +80,9 @@ type Object struct {
 
 // DownloadRange returns an Object's data. A length of -1 will mean
 // (Object.Size - offset).
-func (o *Object) DownloadRange(ctx context.Context, offset, length int64) (io.ReadCloser, error) {
+func (o *Object) DownloadRange(ctx context.Context, offset, length int64) (_ io.ReadCloser, err error) {
+	defer mon.Task()(&ctx)(&err)
+
 	readOnlyStream, err := o.metainfoDB.GetObjectStream(ctx, o.Meta.Bucket, o.Meta.Path)
 	if err != nil {
 		return nil, err

lib/uplink/uplink.go

@@ -95,7 +95,9 @@ type Uplink struct {
 
 // NewUplink creates a new Uplink. This is the first step to create an uplink
 // session with a user specified config or with default config, if nil config
-func NewUplink(ctx context.Context, cfg *Config) (*Uplink, error) {
+func NewUplink(ctx context.Context, cfg *Config) (_ *Uplink, err error) {
+	defer mon.Task()(&ctx)(&err)
+
 	ident, err := identity.NewFullIdentity(ctx, identity.NewCAOptions{
 		Difficulty:  0,
 		Concurrency: 1,

satellite/peer.go

@@ -508,7 +508,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config *Config, ve
 				ClientSecret: mailConfig.ClientSecret,
 				TokenURI:     mailConfig.TokenURI,
 			}
-			token, err := oauth2.RefreshToken(creds, mailConfig.RefreshToken)
+			token, err := oauth2.RefreshToken(context.TODO(), creds, mailConfig.RefreshToken)
 			if err != nil {
 				return nil, errs.Combine(err, peer.Close())
 			}

uplink/piecestore/client.go

@@ -58,8 +58,9 @@ func NewClient(log *zap.Logger, signer signing.Signer, conn *grpc.ClientConn, co
 }
 
 // Delete uses delete order limit to delete a piece on piece store.
-func (client *Client) Delete(ctx context.Context, limit *pb.OrderLimit2) error {
-	_, err := client.client.Delete(ctx, &pb.PieceDeleteRequest{
+func (client *Client) Delete(ctx context.Context, limit *pb.OrderLimit2) (err error) {
+	defer mon.Task()(&ctx)(&err)
+	_, err = client.client.Delete(ctx, &pb.PieceDeleteRequest{
 		Limit: limit,
 	})
 	return Error.Wrap(err)

uplink/piecestore/download.go

@@ -41,7 +41,9 @@ type Download struct {
 }
 
 // Download starts a new download using the specified order limit at the specified offset and size.
-func (client *Client) Download(ctx context.Context, limit *pb.OrderLimit2, offset, size int64) (Downloader, error) {
+func (client *Client) Download(ctx context.Context, limit *pb.OrderLimit2, offset, size int64) (_ Downloader, err error) {
+	defer mon.Task()(&ctx)(&err)
+
 	stream, err := client.client.Download(ctx)
 	if err != nil {
 		return nil, err

uplink/piecestore/upload.go

@@ -47,7 +47,8 @@ type Upload struct {
 }
 
 // Upload initiates an upload to the storage node.
-func (client *Client) Upload(ctx context.Context, limit *pb.OrderLimit2) (Uploader, error) {
+func (client *Client) Upload(ctx context.Context, limit *pb.OrderLimit2) (_ Uploader, err error) {
+	defer mon.Task()(&ctx)(&err)
 	stream, err := client.client.Upload(ctx)
 	if err != nil {
 		return nil, err

uplink/piecestore/verification.go

@@ -24,7 +24,8 @@ var (
 )
 
 // VerifyPieceHash verifies piece hash which is sent by peer.
-func (client *Client) VerifyPieceHash(ctx context.Context, peer *identity.PeerIdentity, limit *pb.OrderLimit2, hash *pb.PieceHash, expectedHash []byte) error {
+func (client *Client) VerifyPieceHash(ctx context.Context, peer *identity.PeerIdentity, limit *pb.OrderLimit2, hash *pb.PieceHash, expectedHash []byte) (err error) {
+	defer mon.Task()(&ctx)(&err)
 	if peer == nil || limit == nil || hash == nil || len(expectedHash) == 0 {
 		return ErrProtocol.New("invalid arguments")
 	}