Merge pull request #313 from sgotti/improve_errors_handling

*: Improve error handling

internal/datamanager/data.go

@@ -28,7 +28,6 @@ import (
 
 	"agola.io/agola/internal/objectstorage"
 	"agola.io/agola/internal/sequence"
-	"agola.io/agola/internal/util"
 
 	"github.com/gofrs/uuid"
 	errors "golang.org/x/xerrors"
@@ -205,7 +204,7 @@ func (d *DataManager) writeDataSnapshot(ctx context.Context, wals []*WalData) er
 		return err
 	}
 	if err := d.ost.WriteObject(d.dataStatusPath(dataSequence), bytes.NewReader(dataStatusj), int64(len(dataStatusj)), true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 
 	return nil
@@ -217,15 +216,15 @@ func (d *DataManager) writeDataFile(ctx context.Context, buf *bytes.Buffer, size
 	}
 
 	if err := d.ost.WriteObject(d.DataFilePath(dataType, dataFileID), buf, size, true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 
 	dataFileIndexj, err := json.Marshal(dataFileIndex)
 	if err != nil {
 		return err
 	}
 	if err := d.ost.WriteObject(d.DataFileIndexPath(dataType, dataFileID), bytes.NewReader(dataFileIndexj), int64(len(dataFileIndexj)), true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 
 	return nil
@@ -341,7 +340,7 @@ func (d *DataManager) writeDataType(ctx context.Context, wi walIndex, dataType s
 			// TODO(sgotti) instead of reading all entries in memory decode it's contents one by one when needed
 			oldDataf, err := d.ost.ReadObject(d.DataFilePath(dataType, actionGroup.DataStatusFile.ID))
 			if err != nil && !objectstorage.IsNotExist(err) {
-				return nil, err
+				return nil, fromOSTError(err)
 			}
 			if !objectstorage.IsNotExist(err) {
 				dec := json.NewDecoder(oldDataf)
@@ -500,7 +499,7 @@ func (d *DataManager) Read(dataType, id string) (io.Reader, error) {
 	var matchingDataFileID string
 	// get the matching data file for the action entry ID
 	if len(curFiles[dataType]) == 0 {
-		return nil, util.NewErrNotExist(errors.Errorf("datatype %q doesn't exists", dataType))
+		return nil, newErrNotExist(errors.Errorf("datatype %q doesn't exists", dataType))
 	}
 
 	matchingDataFileID = curFiles[dataType][0].ID
@@ -513,7 +512,7 @@ func (d *DataManager) Read(dataType, id string) (io.Reader, error) {
 
 	dataFileIndexf, err := d.ost.ReadObject(d.DataFileIndexPath(dataType, matchingDataFileID))
 	if err != nil {
-		return nil, err
+		return nil, fromOSTError(err)
 	}
 	var dataFileIndex *DataFileIndex
 	dec := json.NewDecoder(dataFileIndexf)
@@ -526,12 +525,12 @@ func (d *DataManager) Read(dataType, id string) (io.Reader, error) {
 
 	pos, ok := dataFileIndex.Index[id]
 	if !ok {
-		return nil, util.NewErrNotExist(errors.Errorf("datatype %q, id %q doesn't exists", dataType, id))
+		return nil, newErrNotExist(errors.Errorf("datatype %q, id %q doesn't exists", dataType, id))
 	}
 
 	dataf, err := d.ost.ReadObject(d.DataFilePath(dataType, matchingDataFileID))
 	if err != nil {
-		return nil, err
+		return nil, fromOSTError(err)
 	}
 	if _, err := dataf.Seek(int64(pos), io.SeekStart); err != nil {
 		dataf.Close()
@@ -560,7 +559,7 @@ func (d *DataManager) GetFirstDataStatusSequences(n int) ([]*sequence.Sequence,
 	defer close(doneCh)
 	for object := range d.ost.List(d.storageDataDir()+"/", "", false, doneCh) {
 		if object.Err != nil {
-			return nil, object.Err
+			return nil, fromOSTError(object.Err)
 		}
 		if m := DataStatusFileRegexp.FindStringSubmatch(path.Base(object.Path)); m != nil {
 			seq, err := sequence.Parse(m[1])
@@ -597,7 +596,7 @@ func (d *DataManager) GetLastDataStatusSequences(n int) ([]*sequence.Sequence, e
 
 	for object := range d.ost.List(d.storageDataDir()+"/", "", false, doneCh) {
 		if object.Err != nil {
-			return nil, object.Err
+			return nil, fromOSTError(object.Err)
 		}
 		if m := DataStatusFileRegexp.FindStringSubmatch(path.Base(object.Path)); m != nil {
 			seq, err := sequence.Parse(m[1])
@@ -629,7 +628,7 @@ func (d *DataManager) GetLastDataStatusSequences(n int) ([]*sequence.Sequence, e
 func (d *DataManager) GetDataStatus(dataSequence *sequence.Sequence) (*DataStatus, error) {
 	dataStatusf, err := d.ost.ReadObject(d.dataStatusPath(dataSequence))
 	if err != nil {
-		return nil, err
+		return nil, fromOSTError(err)
 	}
 	defer dataStatusf.Close()
 	var dataStatus *DataStatus
@@ -692,7 +691,7 @@ func (d *DataManager) Export(ctx context.Context, w io.Writer) error {
 		for _, dsf := range curDataStatusFiles {
 			dataf, err := d.ost.ReadObject(d.DataFilePath(dataType, dsf.ID))
 			if err != nil {
-				return err
+				return fromOSTError(err)
 			}
 			if _, err := io.Copy(w, dataf); err != nil {
 				dataf.Close()
@@ -825,7 +824,7 @@ func (d *DataManager) Import(ctx context.Context, r io.Reader) error {
 		return err
 	}
 	if err := d.ost.WriteObject(d.dataStatusPath(dataSequence), bytes.NewReader(dataStatusj), int64(len(dataStatusj)), true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 
 	// initialize etcd providing the specific datastatus
@@ -863,7 +862,7 @@ func (d *DataManager) cleanOldCheckpoints(ctx context.Context, dataStatusSequenc
 		defer close(doneCh)
 		for object := range d.ost.List(d.storageDataDir()+"/", "", false, doneCh) {
 			if object.Err != nil {
-				return object.Err
+				return fromOSTError(object.Err)
 			}
 
 			skip := false
@@ -882,7 +881,7 @@ func (d *DataManager) cleanOldCheckpoints(ctx context.Context, dataStatusSequenc
 				d.log.Infof("removing %q", object.Path)
 				if err := d.ost.DeleteObject(object.Path); err != nil {
 					if !objectstorage.IsNotExist(err) {
-						return err
+						return fromOSTError(err)
 					}
 				}
 			}
@@ -910,7 +909,7 @@ func (d *DataManager) cleanOldCheckpoints(ctx context.Context, dataStatusSequenc
 
 	for object := range d.ost.List(d.storageDataDir()+"/", "", true, doneCh) {
 		if object.Err != nil {
-			return object.Err
+			return fromOSTError(object.Err)
 		}
 
 		p := object.Path
@@ -950,7 +949,7 @@ func (d *DataManager) cleanOldCheckpoints(ctx context.Context, dataStatusSequenc
 			d.log.Infof("removing %q", object.Path)
 			if err := d.ost.DeleteObject(object.Path); err != nil {
 				if !objectstorage.IsNotExist(err) {
-					return err
+					return fromOSTError(err)
 				}
 			}
 		}

internal/datamanager/datamanager.go

@@ -50,6 +50,35 @@ var (
 	ErrConcurrency = errors.New("wal concurrency error: change groups already updated")
 )
 
+type ErrNotExist struct {
+	err error
+}
+
+func newErrNotExist(err error) error {
+	return &ErrNotExist{err: err}
+}
+
+func (e *ErrNotExist) Error() string {
+	return e.err.Error()
+}
+
+func (e *ErrNotExist) Unwrap() error {
+	return e.err
+}
+
+func IsNotExist(err error) bool {
+	var e *ErrNotExist
+	return errors.As(err, &e)
+}
+
+func fromOSTError(err error) error {
+	if objectstorage.IsNotExist(err) {
+		return newErrNotExist(err)
+	}
+
+	return err
+}
+
 var (
 	// Storage paths. Always use path (not filepath) to use the "/" separator
 	storageDataDir = "data"

internal/datamanager/datamanager_test.go

@@ -31,7 +31,6 @@ import (
 
 	"agola.io/agola/internal/objectstorage"
 	"agola.io/agola/internal/testutil"
-	"agola.io/agola/internal/util"
 
 	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
@@ -559,8 +558,8 @@ func TestReadObject(t *testing.T) {
 
 	// should not exists
 	_, _, err = dm.ReadObject("datatype01", "object1", nil)
-	if !util.IsNotExist(err) {
-		t.Fatalf("expected err %v, got: %v", &util.ErrNotExist{}, err)
+	if !IsNotExist(err) {
+		t.Fatalf("expected not exist error, got: %v", err)
 	}
 	// should exist
 	_, _, err = dm.ReadObject("datatype01", "object19", nil)
@@ -583,8 +582,8 @@ func TestReadObject(t *testing.T) {
 
 	// should not exists
 	_, _, err = dm.ReadObject("datatype01", "object1", nil)
-	if !util.IsNotExist(err) {
-		t.Fatalf("expected err %v, got: %v", &util.ErrNotExist{}, err)
+	if !IsNotExist(err) {
+		t.Fatalf("expected not exist error, got: %v", err)
 	}
 	// should exist
 	_, _, err = dm.ReadObject("datatype01", "object19", nil)

internal/datamanager/wal.go

@@ -28,7 +28,6 @@ import (
 	"agola.io/agola/internal/etcd"
 	"agola.io/agola/internal/objectstorage"
 	"agola.io/agola/internal/sequence"
-	"agola.io/agola/internal/util"
 
 	"github.com/gofrs/uuid"
 	etcdclientv3 "go.etcd.io/etcd/clientv3"
@@ -124,7 +123,7 @@ func (d *DataManager) ReadObject(dataType, id string, cgNames []string) (io.Read
 				}
 			}
 		}
-		return nil, nil, util.NewErrNotExist(errors.Errorf("no datatype %q, id %q in wal %s", dataType, id, walseq))
+		return nil, nil, newErrNotExist(errors.Errorf("no datatype %q, id %q in wal %s", dataType, id, walseq))
 	}
 
 	f, err := d.Read(dataType, id)
@@ -137,15 +136,15 @@ func (d *DataManager) HasOSTWal(walseq string) (bool, error) {
 		return false, nil
 	}
 	if err != nil {
-		return false, err
+		return false, fromOSTError(err)
 	}
 	return true, nil
 }
 
 func (d *DataManager) ReadWal(walseq string) (*WalHeader, error) {
 	walFilef, err := d.ost.ReadObject(d.storageWalStatusFile(walseq) + ".committed")
 	if err != nil {
-		return nil, err
+		return nil, fromOSTError(err)
 	}
 	defer walFilef.Close()
 	dec := json.NewDecoder(walFilef)
@@ -158,7 +157,8 @@ func (d *DataManager) ReadWal(walseq string) (*WalHeader, error) {
 }
 
 func (d *DataManager) ReadWalData(walFileID string) (io.ReadCloser, error) {
-	return d.ost.ReadObject(d.storageWalDataFile(walFileID))
+	r, err := d.ost.ReadObject(d.storageWalDataFile(walFileID))
+	return r, fromOSTError(err)
 }
 
 type WalFile struct {
@@ -183,7 +183,7 @@ func (d *DataManager) ListOSTWals(start string) <-chan *WalFile {
 		for object := range d.ost.List(d.storageWalStatusDir()+"/", startPath, true, doneCh) {
 			if object.Err != nil {
 				walCh <- &WalFile{
-					Err: object.Err,
+					Err: fromOSTError(object.Err),
 				}
 				return
 			}
@@ -453,7 +453,7 @@ func (d *DataManager) WriteWalAdditionalOps(ctx context.Context, actions []*Acti
 		}
 	}
 	if err := d.ost.WriteObject(walDataFilePath, bytes.NewReader(buf.Bytes()), int64(buf.Len()), true); err != nil {
-		return nil, err
+		return nil, fromOSTError(err)
 	}
 	d.log.Debugf("wrote wal file: %s", walDataFilePath)
 
@@ -599,7 +599,7 @@ func (d *DataManager) sync(ctx context.Context) error {
 
 			walFileCommittedPath := walFilePath + ".committed"
 			if err := d.ost.WriteObject(walFileCommittedPath, bytes.NewReader(headerj), int64(len(headerj)), true); err != nil {
-				return err
+				return fromOSTError(err)
 			}
 
 			d.log.Debugf("updating wal to state %q", WalStatusCommittedStorage)
@@ -888,7 +888,7 @@ func (d *DataManager) storageWalCleaner(ctx context.Context) error {
 
 	for object := range d.ost.List(d.storageWalStatusDir()+"/", "", true, doneCh) {
 		if object.Err != nil {
-			return err
+			return fromOSTError(err)
 		}
 		name := path.Base(object.Path)
 		ext := path.Ext(name)
@@ -910,15 +910,15 @@ func (d *DataManager) storageWalCleaner(ctx context.Context) error {
 			d.log.Infof("removing %q", walStatusFilePath)
 			if err := d.ost.DeleteObject(walStatusFilePath); err != nil {
 				if !objectstorage.IsNotExist(err) {
-					return err
+					return fromOSTError(err)
 				}
 			}
 
 			// then remove wal status files
 			d.log.Infof("removing %q", object.Path)
 			if err := d.ost.DeleteObject(object.Path); err != nil {
 				if !objectstorage.IsNotExist(err) {
-					return err
+					return fromOSTError(err)
 				}
 			}
 		}
@@ -929,7 +929,7 @@ func (d *DataManager) storageWalCleaner(ctx context.Context) error {
 			d.log.Infof("removing %q", object.Path)
 			if err := d.ost.DeleteObject(object.Path); err != nil {
 				if !objectstorage.IsNotExist(err) {
-					return err
+					return fromOSTError(err)
 				}
 			}
 		}
@@ -1049,7 +1049,7 @@ func (d *DataManager) InitEtcd(ctx context.Context, dataStatus *DataStatus) erro
 	writeWal := func(wal *WalFile, prevWalSequence string) error {
 		walFile, err := d.ost.ReadObject(d.storageWalStatusFile(wal.WalSequence) + ".committed")
 		if err != nil {
-			return err
+			return fromOSTError(err)
 		}
 		dec := json.NewDecoder(walFile)
 		var header *WalHeader
@@ -1200,7 +1200,7 @@ func (d *DataManager) InitEtcd(ctx context.Context, dataStatus *DataStatus) erro
 	walKey := etcdWalKey(walSequence.String())
 
 	if err := d.ost.WriteObject(walDataFilePath, bytes.NewReader([]byte{}), 0, true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 	d.log.Debugf("wrote wal file: %s", walDataFilePath)
 
@@ -1216,7 +1216,7 @@ func (d *DataManager) InitEtcd(ctx context.Context, dataStatus *DataStatus) erro
 	}
 	walFileCommittedPath := walFilePath + ".committed"
 	if err := d.ost.WriteObject(walFileCommittedPath, bytes.NewReader(headerj), int64(len(headerj)), true); err != nil {
-		return err
+		return fromOSTError(err)
 	}
 
 	walData := &WalData{

internal/gitsources/agolagit/agolagit.go

@@ -28,7 +28,7 @@ import (
 
 	gitsource "agola.io/agola/internal/gitsources"
 	"agola.io/agola/internal/services/types"
-	errors "golang.org/x/xerrors"
+	"agola.io/agola/internal/util"
 )
 
 var (
@@ -96,20 +96,8 @@ func (c *Client) getResponse(method, path string, query url.Values, header http.
 		return nil, err
 	}
 
-	if resp.StatusCode/100 != 2 {
-		defer resp.Body.Close()
-		data, err := ioutil.ReadAll(resp.Body)
-		if err != nil {
-			return nil, err
-		}
-
-		if len(data) <= 1 {
-			return resp, errors.New(resp.Status)
-		}
-
-		// TODO(sgotti) use a json error response
-
-		return resp, errors.New(string(data))
+	if err := util.ErrFromRemote(resp); err != nil {
+		return resp, err
 	}
 
 	return resp, nil