Package gfit: Retry failing API calls.

To this end, wrap all API calls in utility functions (this was already
done for DataSource.Create and Dataset.Patch; Dataset.Get was added).
Use the "retry" package inside these functions to retry temporary
(HTTP 5xx) errors.

diff --git a/gfit/gfit.go b/gfit/gfit.go
index 78d28d7..df7bd2c 100644 (file)
--- a/gfit/gfit.go
+++ b/gfit/gfit.go
@@ -9,6 +9,7 @@ import (
 
        "github.com/octo/kraftakt/app"
        "github.com/octo/kraftakt/fitbit"
+       "github.com/octo/retry"
        "golang.org/x/oauth2"
        oauth2google "golang.org/x/oauth2/google"
        fitness "google.golang.org/api/fitness/v1"
@@ -113,8 +114,28 @@ func DataStreamID(dataSource *fitness.DataSource) string {
        return strings.Join(fields, ":")
 }
 
+func wrapGoogleError(err error) error {
+       if err == nil {
+               return nil
+       }
+
+       if gerr, ok := err.(*googleapi.Error); ok && gerr.Code >= 400 && gerr.Code < 500 {
+               return retry.Abort(err)
+       }
+
+       return err
+}
+
 func (c *Client) DataSourceCreate(ctx context.Context, dataSource *fitness.DataSource) (string, error) {
-       res, err := c.Service.Users.DataSources.Create(userID, dataSource).Context(ctx).Do()
+       var dataStreamID string
+       err := retry.Do(ctx, func(ctx context.Context) error {
+               res, err := c.Service.Users.DataSources.Create(userID, dataSource).Context(ctx).Do()
+               if err != nil {
+                       return wrapGoogleError(err)
+               }
+               dataStreamID = res.DataStreamId
+               return nil
+       })
        if err != nil {
                if gerr, ok := err.(*googleapi.Error); ok && gerr.Code == http.StatusConflict {
                        if dataSource.DataStreamId != "" {
@@ -122,13 +143,32 @@ func (c *Client) DataSourceCreate(ctx context.Context, dataSource *fitness.DataS
                        }
                        return DataStreamID(dataSource), nil
                }
-               log.Errorf(ctx, "c.Service.Users.DataSources.Create(%q) = (%+v, %v)", dataSource, res, err)
+               log.Errorf(ctx, "c.Service.Users.DataSources.Create(%q) = %v", DataStreamID(dataSource), err)
                return "", err
        }
-       return res.DataStreamId, nil
+       return dataStreamID, nil
 }
 
-func (c *Client) DataSetPatch(ctx context.Context, dataSourceID string, points []*fitness.DataPoint) error {
+func (c *Client) DatasetGet(ctx context.Context, dataStreamID string, startTime, endTime time.Time) (*fitness.Dataset, error) {
+       datasetID := fmt.Sprintf("%d-%d", startTime.UnixNano(), endTime.UnixNano())
+
+       var dataset *fitness.Dataset
+       err := retry.Do(ctx, func(ctx context.Context) error {
+               res, err := c.Service.Users.DataSources.Datasets.Get(userID, dataStreamID, datasetID).Context(ctx).Do()
+               if err != nil {
+                       return wrapGoogleError(err)
+               }
+               dataset = res
+               return nil
+       })
+       if err != nil {
+               log.Errorf(ctx, "c.Service.Users.DataSources.Datasets.Get(%q, %q) = %v", dataStreamID, datasetID, err)
+               return nil, err
+       }
+       return dataset, nil
+}
+
+func (c *Client) DatasetPatch(ctx context.Context, dataSourceID string, points []*fitness.DataPoint) error {
        startTimeNanos, endTimeNanos := int64(-1), int64(-1)
        for _, p := range points {
                if startTimeNanos == -1 || startTimeNanos > p.StartTimeNanos {
@@ -147,7 +187,10 @@ func (c *Client) DataSetPatch(ctx context.Context, dataSourceID string, points [
                Point:          points,
        }
 
-       _, err := c.Service.Users.DataSources.Datasets.Patch(userID, dataSourceID, datasetID, dataset).Context(ctx).Do()
+       err := retry.Do(ctx, func(ctx context.Context) error {
+               _, err := c.Service.Users.DataSources.Datasets.Patch(userID, dataSourceID, datasetID, dataset).Context(ctx).Do()
+               return wrapGoogleError(err)
+       })
        if err != nil {
                log.Errorf(ctx, "c.Service.Users.DataSources.Datasets.Patch() = %v", err)
                return err
@@ -255,10 +298,8 @@ func (c *Client) SetActivities(ctx context.Context, activities []Activity, start
 
        endOfDay := startOfDay.Add(24 * time.Hour).Add(-1 * time.Nanosecond)
 
-       datasetID := fmt.Sprintf("%d-%d", startOfDay.UnixNano(), endOfDay.UnixNano())
-       res, err := c.Service.Users.DataSources.Datasets.Get(userID, dataStreamID, datasetID).Context(ctx).Do()
+       dataset, err := c.DatasetGet(ctx, dataStreamID, startOfDay, endOfDay)
        if err != nil {
-               log.Errorf(ctx, "c.Service.Users.DataSources.Datasets.Get(%q, %q) = %v", dataStreamID, datasetID, err)
                return err
        }
 
@@ -268,7 +309,7 @@ Next:
                startTimeNanos := a.Start.UnixNano()
                endTimeNanos := a.End.UnixNano()
 
-               for _, p := range res.Point {
+               for _, p := range dataset.Point {
                        if p.StartTimeNanos == startTimeNanos && p.EndTimeNanos == endTimeNanos && p.Value[0].IntVal == a.Type {
                                log.Debugf(ctx, "activity %s already stored in Google Fit", a)
                                continue Next
@@ -290,7 +331,7 @@ Next:
                return nil
        }
 
-       return c.DataSetPatch(ctx, dataStreamID, dataPoints)
+       return c.DatasetPatch(ctx, dataStreamID, dataPoints)
 }
 
 func (c *Client) updateCumulative(ctx context.Context, dataSource *fitness.DataSource, rawValue *fitness.Value, startOfDay time.Time) error {
@@ -346,7 +387,7 @@ func (c *Client) updateCumulative(ctx context.Context, dataSource *fitness.DataS
        }
        log.Debugf(ctx, "add  cumulative data %s until %v: %+v", dataSource.DataStreamId, endTime, diffValue)
 
-       return c.DataSetPatch(ctx, dataSource.DataStreamId, []*fitness.DataPoint{
+       return c.DatasetPatch(ctx, dataSource.DataStreamId, []*fitness.DataPoint{
                &fitness.DataPoint{
                        DataTypeName:   dataSource.DataType.Name,
                        StartTimeNanos: startTime.UnixNano(),
@@ -357,22 +398,19 @@ func (c *Client) updateCumulative(ctx context.Context, dataSource *fitness.DataS
 }
 
 func (c *Client) readCumulative(ctx context.Context, dataSource *fitness.DataSource, startTime, endTime time.Time) (*fitness.Value, time.Time, error) {
-       datasetID := fmt.Sprintf("%d-%d", startTime.UnixNano(), endTime.UnixNano())
-
-       res, err := c.Service.Users.DataSources.Datasets.Get(userID, dataSource.DataStreamId, datasetID).Context(ctx).Do()
+       dataset, err := c.DatasetGet(ctx, dataSource.DataStreamId, startTime, endTime)
        if err != nil {
-               log.Errorf(ctx, "c.Service.Users.DataSources.Datasets.Get(%q, %q) = %v", dataSource.DataStreamId, datasetID, err)
                return nil, time.Time{}, err
        }
 
-       if len(res.Point) == 0 {
+       if len(dataset.Point) == 0 {
                log.Debugf(ctx, "read cumulative data %s until %v: []", dataSource.DataStreamId, endTime)
                return &fitness.Value{}, startTime, nil
        }
 
        var sum fitness.Value
        maxEndTime := startTime
-       for _, p := range res.Point {
+       for _, p := range dataset.Point {
                switch f := dataSource.DataType.Field[0].Format; f {
                case "integer":
                        sum.IntVal += p.Value[0].IntVal
@@ -412,21 +450,18 @@ func (res heartRateDurations) find(min, max int) (*heartRateDuration, bool) {
 }
 
 func (c *Client) heartRate(ctx context.Context, dataSource *fitness.DataSource, startTime, endTime time.Time) (heartRateDurations, time.Time, error) {
-       datasetID := fmt.Sprintf("%d-%d", startTime.UnixNano(), endTime.UnixNano())
-
-       res, err := c.Service.Users.DataSources.Datasets.Get(userID, dataSource.DataStreamId, datasetID).Context(ctx).Do()
+       dataset, err := c.DatasetGet(ctx, dataSource.DataStreamId, startTime, endTime)
        if err != nil {
-               log.Errorf(ctx, "c.Service.Users.DataSources.Datasets.Get(%q, %q) = %v", dataSource.DataStreamId, datasetID, err)
                return nil, time.Time{}, err
        }
 
-       if len(res.Point) == 0 {
+       if len(dataset.Point) == 0 {
                return nil, startTime, nil
        }
 
        var results heartRateDurations
        maxEndTime := startTime
-       for _, p := range res.Point {
+       for _, p := range dataset.Point {
                max := int(p.Value[1].FpVal)
                min := int(p.Value[2].FpVal)
                duration := time.Unix(0, p.EndTimeNanos).Sub(time.Unix(0, p.StartTimeNanos))
@@ -551,5 +586,5 @@ func (c *Client) SetHeartRate(ctx context.Context, totalDurations []fitbit.Heart
        if len(dataPoints) == 0 {
                return nil
        }
-       return c.DataSetPatch(ctx, dataSource.DataStreamId, dataPoints)
+       return c.DatasetPatch(ctx, dataSource.DataStreamId, dataPoints)
 }