fix: Always pay attention to return val of api.SendRecords

pkg/api/api.go

@@ -166,7 +166,7 @@ type Source interface {
 	Records() []*Record
 }
 
-func SendRecords(ctx context.Context, ch chan<- *Record, name string, s Source) bool {
+func SendRecords(ctx context.Context, ch chan<- *Record, name string, s Source) error {
 	ts := time.Now().UnixNano()
 
 	for _, r := range s.Records() {
@@ -175,10 +175,11 @@ func SendRecords(ctx context.Context, ch chan<- *Record, name string, s Source)
 		select {
 		case ch <- r:
 		case <-ctx.Done():
-			return false
+			return ctx.Err()
 		}
 	}
-	return true
+
+	return nil
 }
 
 func FirstError(errors ...error) error {

pkg/service/apigateway/deployments.go

@@ -70,7 +70,9 @@ func (fn *GetDeployments) New(name string, config interface{}) ([]api.Request, e
 					restApiName:          restApi.Name,
 				}
 
-				if !api.SendRecords(ctx, ch, name, source) {
+				if err := api.SendRecords(ctx, ch, name, source); err != nil {
+					innerErr = err
+
 					return false
 				}
 			}

pkg/service/apigateway/restapis.go

@@ -36,9 +36,18 @@ func (fn *GetRestApis) New(name string, config interface{}) ([]api.Request, erro
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.GetRestApisPagesWithContext(ctx, &input, func(output *apigateway.GetRestApisOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &GetRestApisOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.GetRestApisPagesWithContext(ctx, &input, func(output *apigateway.GetRestApisOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &GetRestApisOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/apigateway/stages.go

@@ -70,7 +70,9 @@ func (fn *GetStages) New(name string, config interface{}) ([]api.Request, error)
 					restApiName:     restApi.Name,
 				}
 
-				if !api.SendRecords(ctx, ch, name, source) {
+				if err := api.SendRecords(ctx, ch, name, source); err != nil {
+					innerErr = err
+
 					return false
 				}
 			}

pkg/service/autoscaling/autoscalinggroups.go

@@ -37,9 +37,18 @@ func (fn *DescribeAutoScalingGroups) New(name string, config interface{}) ([]api
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeAutoScalingGroupsPagesWithContext(ctx, &input, func(output *autoscaling.DescribeAutoScalingGroupsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeAutoScalingGroupsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeAutoScalingGroupsPagesWithContext(ctx, &input, func(output *autoscaling.DescribeAutoScalingGroupsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeAutoScalingGroupsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/autoscaling/launchconfigurations.go

@@ -37,9 +37,19 @@ func (fn *DescribeLaunchConfigurations) New(name string, config interface{}) ([]
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeLaunchConfigurationsPagesWithContext(ctx, &input, func(output *autoscaling.DescribeLaunchConfigurationsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeLaunchConfigurationsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeLaunchConfigurationsPagesWithContext(ctx, &input, func(output *autoscaling.DescribeLaunchConfigurationsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeLaunchConfigurationsOutput{output}); err != nil {
+				innerErr = err
+
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/cloudformation/stacks.go

@@ -38,13 +38,22 @@ func (fn *DescribeStacks) New(name string, config interface{}) ([]api.Request, e
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
+		var outerErr, innerErr error
+
 		err := fn.DescribeStacksPagesWithContext(ctx, &input, func(output *cloudformation.DescribeStacksOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeStacksOutput{output})
+			if err := api.SendRecords(ctx, ch, name, &DescribeStacksOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
-		if aerr, ok := err.(awserr.Error); ok && aerr.Code() == "AccessDenied" {
-			return nil
+
+		if aerr, ok := err.(awserr.Error); !ok || aerr.Code() != "AccessDenied" {
+			outerErr = err
 		}
-		return err
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/cloudfront/distributions.go

@@ -37,9 +37,18 @@ func (fn *ListDistributions) New(name string, config interface{}) ([]api.Request
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.ListDistributionsPagesWithContext(ctx, &input, func(output *cloudfront.ListDistributionsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &ListDistributionsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.ListDistributionsPagesWithContext(ctx, &input, func(output *cloudfront.ListDistributionsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &ListDistributionsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/cloudwatchlogs/cloudwatchlogs.go

@@ -49,9 +49,18 @@ func (fn *DescribeLogGroups) New(name string, config interface{}) ([]api.Request
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeLogGroupsPagesWithContext(ctx, &input, func(output *cloudwatchlogs.DescribeLogGroupsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeLogGroupsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeLogGroupsPagesWithContext(ctx, &input, func(output *cloudwatchlogs.DescribeLogGroupsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeLogGroupsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/dynamodb/scan.go

@@ -110,13 +110,22 @@ func (fn *Scan) New(name string, config interface{}) ([]api.Request, error) {
 			tableKeys = append(tableKeys, element.AttributeName)
 		}
 
-		return fn.ScanPagesWithContext(ctx, &input, func(output *dynamodb.ScanOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &ScanOutput{
+		var outerErr, innerErr error
+
+		outerErr = fn.ScanPagesWithContext(ctx, &input, func(output *dynamodb.ScanOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &ScanOutput{
 				ScanOutput: output,
 				TableArn:   describeTableOutput.Table.TableArn,
 				TableKeys:  tableKeys,
-			})
+			}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/dynamodb/table.go

@@ -50,7 +50,9 @@ func (fn *DescribeTable) New(name string, config interface{}) ([]api.Request, er
 					return false
 				}
 
-				if !api.SendRecords(ctx, ch, name, &DescribeTableOutput{describeTableOutput}) {
+				if err := api.SendRecords(ctx, ch, name, &DescribeTableOutput{describeTableOutput}); err != nil {
+					innerErr = err
+
 					// failed to send records, stop handling tables
 					return false
 				}

pkg/service/ec2/addresses.go

@@ -40,8 +40,8 @@ func (fn *DescribeAddresses) New(name string, config interface{}) ([]api.Request
 		if err != nil {
 			return err
 		}
-		api.SendRecords(ctx, ch, name, &DescribeAddressesOutput{output})
-		return nil
+
+		return api.SendRecords(ctx, ch, name, &DescribeAddressesOutput{output})
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/flowlogs.go

@@ -36,9 +36,18 @@ func (fn *DescribeFlowLogs) New(name string, config interface{}) ([]api.Request,
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeFlowLogsPagesWithContext(ctx, &input, func(output *ec2.DescribeFlowLogsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeFlowLogsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeFlowLogsPagesWithContext(ctx, &input, func(output *ec2.DescribeFlowLogsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeFlowLogsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/instance.go

@@ -62,9 +62,18 @@ func (fn *DescribeInstances) New(name string, config interface{}) ([]api.Request
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeInstancesPagesWithContext(ctx, &input, func(output *ec2.DescribeInstancesOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeInstancesOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeInstancesPagesWithContext(ctx, &input, func(output *ec2.DescribeInstancesOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeInstancesOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/internetgateway.go

@@ -36,9 +36,18 @@ func (fn *DescribeInternetGateways) New(name string, config interface{}) ([]api.
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeInternetGatewaysPagesWithContext(ctx, &input, func(output *ec2.DescribeInternetGatewaysOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeInternetGatewaysOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeInternetGatewaysPagesWithContext(ctx, &input, func(output *ec2.DescribeInternetGatewaysOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeInternetGatewaysOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/natgateway.go

@@ -36,9 +36,18 @@ func (fn *DescribeNatGateways) New(name string, config interface{}) ([]api.Reque
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeNatGatewaysPagesWithContext(ctx, &input, func(output *ec2.DescribeNatGatewaysOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeNatGatewaysOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeNatGatewaysPagesWithContext(ctx, &input, func(output *ec2.DescribeNatGatewaysOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeNatGatewaysOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/networkacl.go

@@ -36,9 +36,17 @@ func (fn *DescribeNetworkAcls) New(name string, config interface{}) ([]api.Reque
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeNetworkAclsPagesWithContext(ctx, &input, func(output *ec2.DescribeNetworkAclsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeNetworkAclsOutput{output})
+		var outerErr, innerErr error
+		outerErr = fn.DescribeNetworkAclsPagesWithContext(ctx, &input, func(output *ec2.DescribeNetworkAclsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeNetworkAclsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/networkinterfaces.go

@@ -36,9 +36,18 @@ func (fn *DescribeNetworkInterfaces) New(name string, config interface{}) ([]api
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeNetworkInterfacesPagesWithContext(ctx, &input, func(output *ec2.DescribeNetworkInterfacesOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeNetworkInterfacesOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeNetworkInterfacesPagesWithContext(ctx, &input, func(output *ec2.DescribeNetworkInterfacesOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeNetworkInterfacesOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/routetable.go

@@ -36,9 +36,18 @@ func (fn *DescribeRouteTables) New(name string, config interface{}) ([]api.Reque
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeRouteTablesPagesWithContext(ctx, &input, func(output *ec2.DescribeRouteTablesOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeRouteTablesOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeRouteTablesPagesWithContext(ctx, &input, func(output *ec2.DescribeRouteTablesOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeRouteTablesOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil

pkg/service/ec2/securitygroup.go

@@ -36,9 +36,18 @@ func (fn *DescribeSecurityGroups) New(name string, config interface{}) ([]api.Re
 	}
 
 	call := func(ctx context.Context, ch chan<- *api.Record) error {
-		return fn.DescribeSecurityGroupsPagesWithContext(ctx, &input, func(output *ec2.DescribeSecurityGroupsOutput, last bool) bool {
-			return api.SendRecords(ctx, ch, name, &DescribeSecurityGroupsOutput{output})
+		var outerErr, innerErr error
+
+		outerErr = fn.DescribeSecurityGroupsPagesWithContext(ctx, &input, func(output *ec2.DescribeSecurityGroupsOutput, last bool) bool {
+			if err := api.SendRecords(ctx, ch, name, &DescribeSecurityGroupsOutput{output}); err != nil {
+				innerErr = err
+				return false
+			}
+
+			return true
 		})
+
+		return api.FirstError(outerErr, innerErr)
 	}
 
 	return []api.Request{call}, nil