Edits data gathering process to retry dataGatherer (#185)

* Edits datagatherer and posting data to retry fails

datagatherer uses backoff package to retry datagatherers
These are retried for 5 minutes

post data also retried for 5 minutes depending on the error given

Signed-off-by: Jamie Leppard <JammyL@users.noreply.github.com>

* Changes param1 to AlwaysFail in dummy datagatherer

param1 is no longer needed. To keep a way of testing the
exponential backoff this has been replaced with a boolean
value which allows the datagathering process to always fail

Signed-off-by: Jamie Leppard <JammyL@users.noreply.github.com>

Co-authored-by: Jamie Leppard <JammyL@users.noreply.github.com>

Author: JammyL

agent.yaml

@@ -8,4 +8,8 @@ data-gatherers:
 - kind: "dummy"
   name: "dummy"
   config:
-    param-1: "bar"
+    failed-attempts: 5
+- kind: "dummy"
+  name: "dummy-fail"
+  config:
+    always-fail: true

cmd/agent.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/jetstack/preflight/pkg/agent"
 	"github.com/spf13/cobra"
@@ -43,12 +44,12 @@ func init() {
 		"",
 		"Authorization token. If used, it will override the authorization token in the configuration file.",
 	)
-	agentCmd.PersistentFlags().UintVarP(
+	agentCmd.PersistentFlags().DurationVarP(
 		&agent.Period,
 		"period",
 		"p",
-		3600,
-		"Time between scans, in seconds.",
+		time.Hour,
+		"Time between scans (given as XhYmZs).",
 	)
 	agentCmd.PersistentFlags().StringVarP(
 		&agent.CredentialsPath,
@@ -78,5 +79,12 @@ func init() {
 		"",
 		"Input file path, if used, it will read data from a local file instead of gathering data from clusters",
 	)
+	agentCmd.PersistentFlags().DurationVarP(
+		&agent.BackoffMaxTime,
+		"backoff-max-time",
+		"",
+		5*time.Minute,
+		"Max time for retrying failed data gatherers (given as XhYmZs).",
+	)
 
 }

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/Azure/azure-storage-blob-go v0.8.0
 	github.com/aws/aws-sdk-go v1.25.30
 	github.com/blang/semver v3.5.1+incompatible
+	github.com/cenkalti/backoff v2.0.0+incompatible
 	github.com/d4l3k/messagediff v1.2.1 // indirect
 	github.com/gomarkdown/markdown v0.0.0-20191104174740-4d42851d4d5a
 	github.com/google/go-cmp v0.3.0

go.sum

@@ -97,6 +97,7 @@ github.com/bouk/httprouter v0.0.0-20160817010721-ee8b3818a7f5/go.mod h1:CDReaxg1
 github.com/c-bata/go-prompt v0.2.2/go.mod h1:VzqtzE2ksDBcdln8G7mk2RX9QyGjH+OVqOCSiVIqS34=
 github.com/caarlos0/ctrlc v1.0.0/go.mod h1:CdXpj4rmq0q/1Eb44M9zi2nKB0QraNKuRGYGrrHhcQw=
 github.com/campoy/unique v0.0.0-20180121183637-88950e537e7e/go.mod h1:9IOqJGCPMSc6E5ydlp5NIonxObaeu/Iub/X03EKPVYo=
+github.com/cenkalti/backoff v2.0.0+incompatible h1:5IIPUHhlnUZbcHQsQou5k1Tn58nJkeJL9U+ig5CHJbY=
 github.com/cenkalti/backoff v2.0.0+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=

pkg/agent/config_test.go

@@ -17,7 +17,7 @@ func TestValidConfigLoad(t *testing.T) {
       - name: d1
         kind: dummy
         config:
-          param-1: "bar"
+          always-fail: false
       input-path: "/home"
       output-path: "/nothome"
 `
@@ -35,7 +35,7 @@ func TestValidConfigLoad(t *testing.T) {
 				Name: "d1",
 				Kind: "dummy",
 				Config: &dummyConfig{
-					Param1: "bar",
+					AlwaysFail: false,
 				},
 			},
 		},
@@ -59,7 +59,7 @@ func TestValidConfigWithEndpointLoad(t *testing.T) {
       - name: d1
         kind: dummy
         config:
-          param-1: "bar"
+          always-fail: false
 `
 
 	loadedConfig, err := ParseConfig([]byte(configFileContents))
@@ -80,7 +80,7 @@ func TestValidConfigWithEndpointLoad(t *testing.T) {
 				Name: "d1",
 				Kind: "dummy",
 				Config: &dummyConfig{
-					Param1: "bar",
+					AlwaysFail: false,
 				},
 			},
 		},

pkg/agent/dummy_data_gatherer.go

@@ -8,7 +8,8 @@ import (
 )
 
 type dummyConfig struct {
-	Param1            string `yaml:"param-1"`
+	AlwaysFail        bool `yaml:"always-fail"`
+	FailedAttempts    int  `yaml:"failed-attempts"`
 	wantOnCreationErr bool
 }
 
@@ -17,14 +18,25 @@ func (c *dummyConfig) NewDataGatherer(ctx context.Context) (datagatherer.DataGat
 		return nil, fmt.Errorf("an error")
 	}
 	return &dummyDataGatherer{
-		Param1: c.Param1,
+		AlwaysFail:     c.AlwaysFail,
+		FailedAttempts: c.FailedAttempts,
 	}, nil
 }
 
 type dummyDataGatherer struct {
-	Param1 string
+	AlwaysFail     bool
+	attemptNumber  int
+	FailedAttempts int
 }
 
 func (c *dummyDataGatherer) Fetch() (interface{}, error) {
-	return nil, nil
+	var err error
+	if c.attemptNumber < c.FailedAttempts {
+		err = fmt.Errorf("First %d attempts will fail", c.FailedAttempts)
+	}
+	if c.AlwaysFail {
+		err = fmt.Errorf("This data gatherer will always fail")
+	}
+	c.attemptNumber++
+	return nil, err
 }

pkg/agent/run.go

@@ -12,6 +12,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/cenkalti/backoff"
+	"github.com/hashicorp/go-multierror"
 	"github.com/jetstack/preflight/api"
 	"github.com/jetstack/preflight/pkg/client"
 	"github.com/jetstack/preflight/pkg/datagatherer"
@@ -25,8 +27,8 @@ var ConfigFilePath string
 // AuthToken is the authorization token that will be used for API calls
 var AuthToken string
 
-// Period is the number of seconds between scans
-var Period uint
+// Period is the time waited between scans
+var Period time.Duration
 
 // OneShot flag causes agent to run once
 var OneShot bool
@@ -40,6 +42,9 @@ var OutputPath string
 // InputPath is where the agent will read data from instead of gathering from clusters if specified
 var InputPath string
 
+// BackoffMaxTime is the maximum time for which data gatherers will be retried
+var BackoffMaxTime time.Duration
+
 // Run starts the agent process
 func Run(cmd *cobra.Command, args []string) {
 	ctx := context.Background()
@@ -49,7 +54,7 @@ func Run(cmd *cobra.Command, args []string) {
 		if OneShot {
 			break
 		}
-		time.Sleep(time.Duration(Period) * time.Second)
+		time.Sleep(Period)
 	}
 }
 
@@ -170,7 +175,16 @@ func gatherAndOutputData(ctx context.Context, config Config, preflightClient *cl
 		}
 		log.Println("Data saved locally to", OutputPath)
 	} else {
-		postData(config, preflightClient, readings)
+		backOff := backoff.NewExponentialBackOff()
+		backOff.MaxElapsedTime = BackoffMaxTime
+		post := func() error {
+			return postData(config, preflightClient, readings)
+		}
+		err := backoff.RetryNotify(post, backOff, notify)
+		if err != nil {
+			log.Fatalf("%v", err)
+		}
+
 	}
 }
 
@@ -193,39 +207,67 @@ func gatherData(ctx context.Context, config Config) []*api.DataReading {
 		dataGatherers[dgConfig.Name] = dg
 	}
 
-	// Fetch from all datagatherers
-	now := time.Now()
+	//TODO Change backoff parameters to those desired
+	backOff := backoff.NewExponentialBackOff()
+	backOff.MaxElapsedTime = BackoffMaxTime
 	readings := []*api.DataReading{}
-	failedDataGatherers := []string{}
-	for k, dg := range dataGatherers {
-		i, err := dg.Fetch()
-		if err != nil {
-			log.Printf("Error fetching with DataGatherer %q: %s", k, err)
-			failedDataGatherers = append(failedDataGatherers, k)
-			continue
-		}
+	completedDataGatherers := make(map[string]bool, len(dataGatherers))
 
-		log.Printf("Gathered data for %q:\n", k)
-
-		readings = append(readings, &api.DataReading{
-			ClusterID:    config.ClusterID,
-			DataGatherer: k,
-			Timestamp:    api.Time{Time: now},
-			Data:         i,
-		})
+	// Fetch from all datagatherers
+	getReadings := func() error {
+		var dgError *multierror.Error
+		for k, dg := range dataGatherers {
+			if completedDataGatherers[k] {
+				continue
+			}
+			dgData, err := dg.Fetch()
+			if err != nil {
+				err = fmt.Errorf("%s: %v", k, err)
+				dgError = multierror.Append(dgError, err)
+				continue
+			} else {
+				completedDataGatherers[k] = true
+
+				log.Printf("Successfully gathered data for %q", k)
+				now := time.Now()
+
+				readings = append(readings, &api.DataReading{
+					ClusterID:    config.ClusterID,
+					DataGatherer: k,
+					Timestamp:    api.Time{Time: now},
+					Data:         dgData,
+				})
+			}
+		}
+		dgError.ErrorFormat = func(es []error) string {
+			points := make([]string, len(es))
+			for i, err := range es {
+				points[i] = fmt.Sprintf("* %s", err)
+			}
+			return fmt.Sprintf(
+				"The following %d data gatherer(s) have failed:\n\t%s",
+				len(es), strings.Join(points, "\n\t"))
+		}
+		return dgError
 	}
 
-	if len(failedDataGatherers) > 0 {
-		log.Printf(
-			"Warning, the following DataGatherers failed, %s. Their data is not being sent.",
-			strings.Join(failedDataGatherers, ", "),
-		)
+	err := backoff.RetryNotify(getReadings, backOff, notify)
+	if err != nil {
+		log.Println(err)
+		log.Printf("This will not be retried")
+	} else {
+		log.Printf("All data gatherers successfull")
 	}
 	return readings
 }
 
-func postData(config Config, preflightClient *client.PreflightClient, readings []*api.DataReading) {
+func notify(err error, t time.Duration) {
+	log.Println(err, "\nRetrying...")
+}
+
+func postData(config Config, preflightClient *client.PreflightClient, readings []*api.DataReading) error {
 	baseURL := config.Server
+	var err error
 
 	log.Println("Running Agent...")
 	log.Println("Posting data to ", baseURL)
@@ -241,7 +283,7 @@ func postData(config Config, preflightClient *client.PreflightClient, readings [
 		res, err := preflightClient.Post(path, bytes.NewBuffer(data))
 
 		if err != nil {
-			log.Fatalf("Failed to post data: %+v", err)
+			return fmt.Errorf("Failed to post data: %+v", err)
 		}
 		if code := res.StatusCode; code < 200 || code >= 300 {
 			errorContent := ""
@@ -251,15 +293,13 @@ func postData(config Config, preflightClient *client.PreflightClient, readings [
 			}
 			defer res.Body.Close()
 
-			log.Fatalf("Received response with status code %d. Body: %s", code, errorContent)
+			return fmt.Errorf("Received response with status code %d. Body: %s", code, errorContent)
 		}
 	} else {
 		err := preflightClient.PostDataReadings(config.OrganizationID, readings)
-		// TODO: handle errors gracefully: e.g. handle retries when it is possible
 		if err != nil {
-			log.Fatalf("Post to server failed: %+v", err)
+			return fmt.Errorf("Post to server failed: %+v", err)
 		}
 	}
-
-	log.Println("Data sent successfully.")
+	return err
 }