Log the number of items from each dg

This returns the number of items collected by each datagatherer so that
the logs tell us a bit more about what was found in the cluster, and
can help find where any items have been missed.

Author: James Westby

pkg/agent/dummy_data_gatherer.go

@@ -44,7 +44,7 @@ func (g *dummyDataGatherer) Delete() error {
 	return nil
 }
 
-func (c *dummyDataGatherer) Fetch() (interface{}, error) {
+func (c *dummyDataGatherer) Fetch() (interface{}, int, error) {
 	var err error
 	if c.attemptNumber < c.FailedAttempts {
 		err = fmt.Errorf("First %d attempts will fail", c.FailedAttempts)
@@ -53,5 +53,5 @@ func (c *dummyDataGatherer) Fetch() (interface{}, error) {
 		err = fmt.Errorf("This data gatherer will always fail")
 	}
 	c.attemptNumber++
-	return nil, err
+	return nil, -1, err
 }

pkg/agent/run.go

@@ -362,14 +362,18 @@ func gatherData(config Config, dataGatherers map[string]datagatherer.DataGathere
 
 	var dgError *multierror.Error
 	for k, dg := range dataGatherers {
-		dgData, err := dg.Fetch()
+		dgData, count, err := dg.Fetch()
 		if err != nil {
 			dgError = multierror.Append(dgError, fmt.Errorf("error in datagatherer %s: %w", k, err))
 
 			continue
 		}
 
-		log.Printf("successfully gathered data from %q datagatherer", k)
+		if count >= 0 {
+			log.Printf("successfully gathered %d items from %q datagatherer", count, k)
+		} else {
+			log.Printf("successfully gathered data from %q datagatherer", k)
+		}
 		readings = append(readings, &api.DataReading{
 			ClusterID:     config.ClusterID,
 			DataGatherer:  k,

pkg/datagatherer/datagatherer.go

@@ -12,7 +12,9 @@ type Config interface {
 // DataGatherer is the interface for Data Gatherers. Data Gatherers are in charge of fetching data from a certain cloud provider API or Kubernetes component.
 type DataGatherer interface {
 	// Fetch retrieves data.
-	Fetch() (interface{}, error)
+	// count is the number of items that were discovered. A negative count means the number
+	// of items was indeterminate.
+	Fetch() (data interface{}, count int, err error)
 	// Run starts the data gatherer's informers for resource collection.
 	// Returns error if the data gatherer informer wasn't initialized
 	Run(stopCh <-chan struct{}) error

pkg/datagatherer/k8s/discovery.go

@@ -62,16 +62,16 @@ func (g *DataGathererDiscovery) Delete() error {
 }
 
 // Fetch will fetch discovery data from the apiserver, or return an error
-func (g *DataGathererDiscovery) Fetch() (interface{}, error) {
+func (g *DataGathererDiscovery) Fetch() (interface{}, int, error) {
 	data, err := g.cl.ServerVersion()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get server version: %v", err)
+		return nil, -1, fmt.Errorf("failed to get server version: %v", err)
 	}
 
 	response := map[string]interface{}{
 		// data has type Info: https://godoc.org/k8s.io/apimachinery/pkg/version#Info
 		"server_version": data,
 	}
 
-	return response, nil
+	return response, len(response), nil
 }

pkg/datagatherer/k8s/dynamic.go

@@ -307,9 +307,9 @@ func (g *DataGathererDynamic) Delete() error {
 
 // Fetch will fetch the requested data from the apiserver, or return an error
 // if fetching the data fails.
-func (g *DataGathererDynamic) Fetch() (interface{}, error) {
+func (g *DataGathererDynamic) Fetch() (interface{}, int, error) {
 	if g.groupVersionResource.String() == "" {
-		return nil, fmt.Errorf("resource type must be specified")
+		return nil, -1, fmt.Errorf("resource type must be specified")
 	}
 
 	var list = map[string]interface{}{}
@@ -333,19 +333,19 @@ func (g *DataGathererDynamic) Fetch() (interface{}, error) {
 			}
 			continue
 		}
-		return nil, fmt.Errorf("failed to parse cached resource")
+		return nil, -1, fmt.Errorf("failed to parse cached resource")
 	}
 
 	// Redact Secret data
 	err := redactList(items)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return nil, -1, errors.WithStack(err)
 	}
 
 	// add gathered resources to items
 	list["items"] = items
 
-	return list, nil
+	return list, len(items), nil
 }
 
 func redactList(list []*api.GatheredResource) error {

pkg/datagatherer/k8s/dynamic_test.go

@@ -633,7 +633,7 @@ func TestDynamicGatherer_Fetch(t *testing.T) {
 			if waitTimeout(&wg, 30*time.Second) {
 				t.Fatalf("unexpected timeout")
 			}
-			res, err := dynamiDg.Fetch()
+			res, count, err := dynamiDg.Fetch()
 			if err != nil && !tc.err {
 				t.Errorf("expected no error but got: %v", err)
 			}
@@ -662,6 +662,10 @@ func TestDynamicGatherer_Fetch(t *testing.T) {
 					gotJSON, _ := json.MarshalIndent(list, "", "  ")
 					t.Fatalf("unexpected JSON: \ngot \n%s\nwant\n%s", string(gotJSON), expectedJSON)
 				}
+
+				if len(list) != count {
+					t.Errorf("wrong count of resources reported: got %d, want %d", count, len(list))
+				}
 			}
 		})
 	}
@@ -922,7 +926,7 @@ func TestDynamicGathererNativeResources_Fetch(t *testing.T) {
 			if waitTimeout(&wg, 5*time.Second) {
 				t.Fatalf("unexpected timeout")
 			}
-			res, err := dynamiDg.Fetch()
+			res, count, err := dynamiDg.Fetch()
 			if err != nil && !tc.err {
 				t.Errorf("expected no error but got: %v", err)
 			}
@@ -951,6 +955,10 @@ func TestDynamicGathererNativeResources_Fetch(t *testing.T) {
 					gotJSON, _ := json.MarshalIndent(list, "", "  ")
 					t.Fatalf("unexpected JSON: \ngot \n%s\nwant\n%s", string(gotJSON), expectedJSON)
 				}
+
+				if len(list) != count {
+					t.Errorf("wrong count of resources reported: got %d, want %d", count, len(list))
+				}
 			}
 		})
 	}

pkg/datagatherer/local/local.go

@@ -54,10 +54,10 @@ func (g *DataGatherer) WaitForCacheSync(stopCh <-chan struct{}) error {
 }
 
 // Fetch loads and returns the data from the LocalDatagatherer's dataPath
-func (g *DataGatherer) Fetch() (interface{}, error) {
+func (g *DataGatherer) Fetch() (interface{}, int, error) {
 	dataBytes, err := ioutil.ReadFile(g.dataPath)
 	if err != nil {
-		return nil, err
+		return nil, -1, err
 	}
-	return dataBytes, nil
+	return dataBytes, -1, nil
 }