🐛 Chunk sending results by size

go.mod

@@ -31,7 +31,7 @@ require (
  github.com/spf13/pflag v1.0.6-0.20201009195203-85dd5c8bc61c
  github.com/spf13/viper v1.19.0
  github.com/stretchr/testify v1.9.0
- go.mondoo.com/cnquery/v11 v11.24.0
+ go.mondoo.com/cnquery/v11 v11.24.1-0.20241003101157-5788fb46e68e
  go.mondoo.com/mondoo-go v0.0.0-20240924071220-8972d78c3774
  go.mondoo.com/ranger-rpc v0.6.4
  go.opentelemetry.io/otel v1.30.0

go.sum

@@ -941,6 +941,8 @@ go.etcd.io/etcd/client/pkg/v3 v3.5.1/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3
 go.etcd.io/etcd/client/v2 v2.305.1/go.mod h1:pMEacxZW7o8pg4CrFE7pquyCJJzZvkvdD2RibOCCCGs=
 go.mondoo.com/cnquery/v11 v11.24.0 h1:Dahfra5OIlgHRmvCdo2yuWxRLKQR+WPgBYOKMtzEGyY=
 go.mondoo.com/cnquery/v11 v11.24.0/go.mod h1:RhCgz3xFT8Lp9+Zz7A/YprJw4BNvKP1IRJENNDQleUs=
+go.mondoo.com/cnquery/v11 v11.24.1-0.20241003101157-5788fb46e68e h1:X22nHyW6SN+x6p3/Q3AzaB42lwFCyW8Xdw4vS3+Cwvo=
+go.mondoo.com/cnquery/v11 v11.24.1-0.20241003101157-5788fb46e68e/go.mod h1:RhCgz3xFT8Lp9+Zz7A/YprJw4BNvKP1IRJENNDQleUs=
 go.mondoo.com/mondoo-go v0.0.0-20240924071220-8972d78c3774 h1:CKLSb7moOWfk6oqBjr1QUy+St0b/yb/0DayUgtUsoAQ=
 go.mondoo.com/mondoo-go v0.0.0-20240924071220-8972d78c3774/go.mod h1:dun0t/zVJCSB/u9dhSO4FXn13moiH7JSwS2m8Ucxx50=
 go.mondoo.com/ranger-rpc v0.6.4 h1:q01kjESvF2HSnbFO+TjpUQSiI2IM8JWGJLH3u0vNxZA=

policy/executor/internal/collector.go

@@ -11,6 +11,7 @@ import (
 
  "github.com/rs/zerolog/log"
  "go.mondoo.com/cnquery/v11/llx"
+ "go.mondoo.com/cnquery/v11/utils/iox"
  "go.mondoo.com/cnspec/v11/policy"
  "google.golang.org/protobuf/proto"
 )
@@ -317,23 +318,49 @@ func (c *PolicyServiceCollector) Sink(results []*llx.RawResult, scores []*policy
  return
  }
 
- resultsToSend := make(map[string]*llx.Result, len(results))
- for _, rr := range results {
- resultsToSend[rr.CodeID] = c.toResult(rr)
+ if len(results) > 0 {
+ llxResults := make([]*llx.Result, len(results))
+ for i, rr := range results {
+ llxResults[i] = c.toResult(rr)
+ }
+ err := iox.ChunkMessages(func(chunk []*llx.Result) error {
+ log.Debug().Msg("Sending datapoints")
+ resultsToSend := make(map[string]*llx.Result, len(chunk))
+ for _, rr := range chunk {
+ resultsToSend[rr.CodeId] = rr
+ }
+ _, err := c.resolver.StoreResults(context.Background(), &policy.StoreResultsReq{
+ AssetMrn: c.assetMrn,
+ Data: resultsToSend,
+ IsPreprocessed: true,
+ IsLastBatch: isDone,
+ })
+ if err != nil {
+ log.Error().Err(err).Msg("failed to send datapoints")
+ }
+ return nil
+ }, func(item *llx.Result, msgSize int) {
+ log.Warn().Msgf("Data %s %d exceeds maximum message size", item.CodeId, msgSize)
+ }, llxResults...)
+ if err != nil {
+ log.Error().Err(err).Msg("failed to send datapoints")
+ }
  }
 
- log.Debug().Msg("Sending datapoints and scores")
- _, err := c.resolver.StoreResults(context.Background(), &policy.StoreResultsReq{
- AssetMrn: c.assetMrn,
- Data: resultsToSend,
- Scores: scores,
- Risks: risks,
- IsPreprocessed: true,
- IsLastBatch: isDone,
- })
- if err != nil {
- log.Error().Err(err).Msg("failed to send datapoints and scores")
+ if len(scores) > 0 || len(risks) > 0 {
+ log.Debug().Msg("Sending scores")
+ _, err := c.resolver.StoreResults(context.Background(), &policy.StoreResultsReq{
+ AssetMrn: c.assetMrn,
+ Scores: scores,
+ Risks: risks,
+ IsPreprocessed: true,
+ IsLastBatch: isDone,
+ })
+ if err != nil {
+ log.Error().Err(err).Msg("failed to send datapoints and scores")
+ }
  }
+
 }
 
 type FuncCollector struct {