Expose certifier and deps.dev batch size and add optional latency (defaults to none)

cmd/guaccollect/cmd/deps_dev.go

@@ -51,6 +51,8 @@ type depsDevOptions struct {
 	prometheusPort int
 	// enable/disable message publish to queue
 	publishToQueue bool
+	// sets artificial latency on the deps.dev collector (default to nil)
+	addedLatency *time.Duration
 }
 
 var depsDevCmd = &cobra.Command{
@@ -87,6 +89,7 @@ you have access to read and write to the respective blob store.`,
 			viper.GetBool("enable-prometheus"),
 			viper.GetInt("prometheus-port"),
 			viper.GetBool("publish-to-queue"),
+			viper.GetString("deps-dev-latency"),
 			args,
 		)
 		if err != nil {
@@ -95,7 +98,7 @@ you have access to read and write to the respective blob store.`,
 			os.Exit(1)
 		}
 		// Register collector
-		depsDevCollector, err := deps_dev.NewDepsCollector(ctx, opts.dataSource, opts.poll, opts.retrieveDependencies, 30*time.Second)
+		depsDevCollector, err := deps_dev.NewDepsCollector(ctx, opts.dataSource, opts.poll, opts.retrieveDependencies, 30*time.Second, opts.addedLatency)
 		if err != nil {
 			logger.Fatalf("unable to register oci collector: %v", err)
 		}
@@ -129,6 +132,7 @@ func validateDepsDevFlags(
 	enablePrometheus bool,
 	prometheusPort int,
 	pubToQueue bool,
+	addedLatencyStr string,
 	args []string,
 ) (depsDevOptions, error) {
 	var opts depsDevOptions
@@ -139,6 +143,17 @@ func validateDepsDevFlags(
 	opts.enablePrometheus = enablePrometheus
 	opts.prometheusPort = prometheusPort
 	opts.publishToQueue = pubToQueue
+
+	if addedLatencyStr != "" {
+		addedLatency, err := time.ParseDuration(addedLatencyStr)
+		if err != nil {
+			return opts, fmt.Errorf("failed to parser duration with error: %w", err)
+		}
+		opts.addedLatency = &addedLatency
+	} else {
+		opts.addedLatency = nil
+	}
+
 	if useCsub {
 		csubOpts, err := csubclient.ValidateCsubClientFlags(csubAddr, csubTls, csubTlsSkipVerify)
 		if err != nil {
@@ -174,7 +189,7 @@ func validateDepsDevFlags(
 }
 
 func init() {
-	set, err := cli.BuildFlags([]string{"retrieve-dependencies", "prometheus-port"})
+	set, err := cli.BuildFlags([]string{"retrieve-dependencies", "prometheus-port", "deps-dev-latency"})
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err)
 		os.Exit(1)

cmd/guaccollect/cmd/osv.go

@@ -57,6 +57,10 @@ type osvOptions struct {
 	// days since the last vulnerability scan was run.
 	// 0 means only run once
 	daysSinceLastScan int
+	// sets artificial latency on the certifier (default to nil)
+	addedLatency *time.Duration
+	// sets the batch size for pagination query for the certifier
+	batchSize int
 }
 
 var osvCmd = &cobra.Command{
@@ -87,6 +91,8 @@ you have access to read and write to the respective blob store.`,
 			viper.GetBool("service-poll"),
 			viper.GetBool("publish-to-queue"),
 			viper.GetInt("last-scan"),
+			viper.GetString("certifier-latency"),
+			viper.GetInt("certifier-batch-size"),
 		)
 		if err != nil {
 			fmt.Printf("unable to validate flags: %v\n", err)
@@ -105,7 +111,7 @@ you have access to read and write to the respective blob store.`,
 		httpClient := http.Client{Transport: transport}
 		gqlclient := graphql.NewClient(opts.graphqlEndpoint, &httpClient)
 
-		packageQueryFunc, err := getPackageQuery(gqlclient, opts.daysSinceLastScan)
+		packageQueryFunc, err := getPackageQuery(gqlclient, opts.daysSinceLastScan, opts.batchSize, opts.addedLatency)
 		if err != nil {
 			logger.Errorf("error: %v", err)
 			os.Exit(1)
@@ -122,7 +128,10 @@ func validateOSVFlags(
 	blobAddr,
 	interval string,
 	poll bool,
-	pubToQueue bool, daysSince int) (osvOptions, error) {
+	pubToQueue bool,
+	daysSince int,
+	certifierLatencyStr string,
+	batchSize int) (osvOptions, error) {
 
 	var opts osvOptions
 
@@ -139,6 +148,19 @@ func validateOSVFlags(
 	}
 	opts.interval = i
 	opts.daysSinceLastScan = daysSince
+
+	if certifierLatencyStr != "" {
+		addedLatency, err := time.ParseDuration(certifierLatencyStr)
+		if err != nil {
+			return opts, fmt.Errorf("failed to parser duration with error: %w", err)
+		}
+		opts.addedLatency = &addedLatency
+	} else {
+		opts.addedLatency = nil
+	}
+
+	opts.batchSize = batchSize
+
 	return opts, nil
 }
 
@@ -148,9 +170,9 @@ func getCertifierPublish(ctx context.Context, blobStore *blob.BlobStore, pubsub
 	}, nil
 }
 
-func getPackageQuery(client graphql.Client, daysSinceLastScan int) (func() certifier.QueryComponents, error) {
+func getPackageQuery(client graphql.Client, daysSinceLastScan int, batchSize int, addedLatency *time.Duration) (func() certifier.QueryComponents, error) {
 	return func() certifier.QueryComponents {
-		packageQuery := root_package.NewPackageQuery(client, daysSinceLastScan)
+		packageQuery := root_package.NewPackageQuery(client, daysSinceLastScan, batchSize, addedLatency)
 		return packageQuery
 	}, nil
 }
@@ -233,7 +255,8 @@ func initializeNATsandCertifier(ctx context.Context, blobAddr, pubsubAddr string
 
 func init() {
 	set, err := cli.BuildFlags([]string{"interval",
-		"last-scan", "header-file"})
+		"last-scan", "header-file", "certifier-latency",
+		"certifier-batch-size"})
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err)
 		os.Exit(1)

cmd/guaccollect/cmd/scorecard.go

@@ -48,6 +48,10 @@ type scorecardOptions struct {
 	publishToQueue bool
 	// setting "daysSinceLastScan" to 0 does not check the timestamp on the scorecard that exist
 	daysSinceLastScan int
+	// sets artificial latency on the certifier (default to nil)
+	addedLatency *time.Duration
+	// sets the batch size for pagination query for the certifier
+	batchSize int
 }
 
 var scorecardCmd = &cobra.Command{
@@ -78,6 +82,8 @@ you have access to read and write to the respective blob store.`,
 			viper.GetBool("service-poll"),
 			viper.GetBool("publish-to-queue"),
 			viper.GetInt("last-scan"),
+			viper.GetString("certifier-latency"),
+			viper.GetInt("certifier-batch-size"),
 		)
 		if err != nil {
 			fmt.Printf("unable to validate flags: %v\n", err)
@@ -114,7 +120,7 @@ you have access to read and write to the respective blob store.`,
 		httpClient := http.Client{Transport: transport}
 		gqlclient := graphql.NewClient(opts.graphqlEndpoint, &httpClient)
 
-		query, err := sc.NewCertifier(gqlclient, opts.daysSinceLastScan)
+		query, err := sc.NewCertifier(gqlclient, opts.daysSinceLastScan, opts.batchSize, opts.addedLatency)
 		if err != nil {
 			logger.Errorf("unable to create source query: %v\n", err)
 			os.Exit(1)
@@ -131,7 +137,10 @@ func validateScorecardFlags(
 	blobAddr,
 	interval string,
 	poll bool,
-	pubToQueue bool, daysSince int) (scorecardOptions, error) {
+	pubToQueue bool,
+	daysSince int,
+	certifierLatencyStr string,
+	batchSize int) (scorecardOptions, error) {
 
 	var opts scorecardOptions
 
@@ -149,12 +158,25 @@ func validateScorecardFlags(
 	opts.interval = i
 	opts.daysSinceLastScan = daysSince
 
+	if certifierLatencyStr != "" {
+		addedLatency, err := time.ParseDuration(certifierLatencyStr)
+		if err != nil {
+			return opts, fmt.Errorf("failed to parser duration with error: %w", err)
+		}
+		opts.addedLatency = &addedLatency
+	} else {
+		opts.addedLatency = nil
+	}
+
+	opts.batchSize = batchSize
+
 	return opts, nil
 }
 
 func init() {
 	set, err := cli.BuildFlags([]string{"interval",
-		"last-scan", "header-file"})
+		"last-scan", "header-file", "certifier-latency",
+		"certifier-batch-size"})
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err)
 		os.Exit(1)

cmd/guacone/cmd/deps_dev.go

@@ -51,6 +51,8 @@ type depsDevOptions struct {
 	graphqlEndpoint      string
 	headerFile           string
 	queryVulnOnIngestion bool
+	// sets artificial latency on the deps.dev collector (default to nil)
+	addedLatency *time.Duration
 }
 
 var depsDevCmd = &cobra.Command{
@@ -69,7 +71,7 @@ var depsDevCmd = &cobra.Command{
 		transport := cli.HTTPHeaderTransport(ctx, opts.headerFile, http.DefaultTransport)
 
 		// Register collector
-		depsDevCollector, err := deps_dev.NewDepsCollector(ctx, opts.dataSource, opts.poll, opts.retrieveDependencies, 30*time.Second)
+		depsDevCollector, err := deps_dev.NewDepsCollector(ctx, opts.dataSource, opts.poll, opts.retrieveDependencies, 30*time.Second, opts.addedLatency)
 		if err != nil {
 			logger.Fatalf("unable to register depsdev collector: %v", err)
 		}
@@ -142,6 +144,18 @@ func validateDepsDevFlags(args []string) (*depsDevOptions, client.Client, error)
 		headerFile:           viper.GetString("header-file"),
 		queryVulnOnIngestion: viper.GetBool("add-vuln-on-ingest"),
 	}
+
+	addedLatencyStr := viper.GetString("deps-dev-latency")
+	if addedLatencyStr != "" {
+		addedLatency, err := time.ParseDuration(addedLatencyStr)
+		if err != nil {
+			return opts, nil, fmt.Errorf("failed to parser duration with error: %w", err)
+		}
+		opts.addedLatency = &addedLatency
+	} else {
+		opts.addedLatency = nil
+	}
+
 	useCsub := viper.GetBool("use-csub")
 	if useCsub {
 		csubAddr := viper.GetString("csub-addr")
@@ -184,7 +198,7 @@ func validateDepsDevFlags(args []string) (*depsDevOptions, client.Client, error)
 }
 
 func init() {
-	set, err := cli.BuildFlags([]string{"poll", "retrieve-dependencies", "use-csub"})
+	set, err := cli.BuildFlags([]string{"poll", "retrieve-dependencies", "use-csub", "deps-dev-latency"})
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err)
 		os.Exit(1)

cmd/guacone/cmd/osv.go

@@ -47,6 +47,10 @@ type osvOptions struct {
 	csubClientOptions    csub_client.CsubClientOptions
 	interval             time.Duration
 	queryVulnOnIngestion bool
+	// sets artificial latency on the certifier (default to nil)
+	addedLatency *time.Duration
+	// sets the batch size for pagination query for the certifier
+	batchSize int
 }
 
 var osvCmd = &cobra.Command{
@@ -62,6 +66,8 @@ var osvCmd = &cobra.Command{
 			viper.GetBool("csub-tls"),
 			viper.GetBool("csub-tls-skip-verify"),
 			viper.GetBool("add-vuln-on-ingest"),
+			viper.GetString("certifier-latency"),
+			viper.GetInt("certifier-batch-size"),
 		)
 		if err != nil {
 			fmt.Printf("unable to validate flags: %v\n", err)
@@ -88,7 +94,7 @@ var osvCmd = &cobra.Command{
 
 		httpClient := http.Client{Transport: transport}
 		gqlclient := graphql.NewClient(opts.graphqlEndpoint, &httpClient)
-		packageQuery := root_package.NewPackageQuery(gqlclient, 0)
+		packageQuery := root_package.NewPackageQuery(gqlclient, 0, opts.batchSize, opts.addedLatency)
 
 		totalNum := 0
 		docChan := make(chan *processor.Document)
@@ -216,6 +222,8 @@ func validateOSVFlags(
 	csubTls,
 	csubTlsSkipVerify bool,
 	queryVulnIngestion bool,
+	certifierLatencyStr string,
+	batchSize int,
 ) (osvOptions, error) {
 	var opts osvOptions
 	opts.graphqlEndpoint = graphqlEndpoint
@@ -227,6 +235,18 @@ func validateOSVFlags(
 	}
 	opts.interval = i
 
+	if certifierLatencyStr != "" {
+		addedLatency, err := time.ParseDuration(certifierLatencyStr)
+		if err != nil {
+			return opts, fmt.Errorf("failed to parser duration with error: %w", err)
+		}
+		opts.addedLatency = &addedLatency
+	} else {
+		opts.addedLatency = nil
+	}
+
+	opts.batchSize = batchSize
+
 	csubOpts, err := csub_client.ValidateCsubClientFlags(csubAddr, csubTls, csubTlsSkipVerify)
 	if err != nil {
 		return opts, fmt.Errorf("unable to validate csub client flags: %w", err)
@@ -238,5 +258,16 @@ func validateOSVFlags(
 }
 
 func init() {
+	set, err := cli.BuildFlags([]string{"certifier-latency",
+		"certifier-batch-size"})
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err)
+		os.Exit(1)
+	}
+	osvCmd.PersistentFlags().AddFlagSet(set)
+	if err := viper.BindPFlags(osvCmd.PersistentFlags()); err != nil {
+		fmt.Fprintf(os.Stderr, "failed to bind flags: %v", err)
+		os.Exit(1)
+	}
 	certifierCmd.AddCommand(osvCmd)
 }