feat(performance): concurrent engine scans by platform (#3085) (#3061)

Signed-off-by: João Reigota <[email protected]>
Co-authored-by: Rogério Peixoto <[email protected]>

e2e/cli_test.go

@@ -389,7 +389,7 @@ var tests = []struct {
 			},
 		},
 		validation: func(outputText string) bool {
-			match, _ := regexp.MatchString(`Total CPU usage for inspect: \d+`, outputText)
+			match, _ := regexp.MatchString(`Total CPU usage for start_scan: \d+`, outputText)
 			return match
 		},
 		wantStatus: []int{50},
@@ -405,7 +405,7 @@ var tests = []struct {
 			},
 		},
 		validation: func(outputText string) bool {
-			match, _ := regexp.MatchString(`Total MEM usage for inspect: \d+`, outputText)
+			match, _ := regexp.MatchString(`Total MEM usage for start_scan: \d+`, outputText)
 			return match
 		},
 		wantStatus: []int{50},

internal/console/helpers/helpers.go

@@ -31,11 +31,12 @@ var reportGenerators = map[string]func(path, filename string, body interface{})
 // ProgressBar represents a Progress
 // Writer is the writer output for progress bar
 type ProgressBar struct {
-	Writer   io.Writer
-	label    string
-	space    int
-	total    float64
-	progress chan float64
+	Writer          io.Writer
+	label           string
+	space           int
+	total           float64
+	currentProgress float64
+	progress        chan float64
 }
 
 // Printer wil print console output with colors
@@ -80,13 +81,14 @@ func (p *ProgressBar) Start(wg *sync.WaitGroup) {
 		const hundredPercent = 100
 		formmatingString := "\r" + p.label + "[%s %4.1f%% %s]"
 		for {
-			currentProgress, ok := <-p.progress
-			if !ok || currentProgress >= p.total {
+			newProgress, ok := <-p.progress
+			p.currentProgress += newProgress
+			if !ok || p.currentProgress >= p.total {
 				fmt.Fprintf(p.Writer, formmatingString, strings.Repeat("=", p.space), 100.0, strings.Repeat("=", p.space))
 				break
 			}
 
-			percentage := currentProgress / p.total * hundredPercent
+			percentage := p.currentProgress / p.total * hundredPercent
 			convertedPercentage := int(math.Round(float64(p.space+p.space) / hundredPercent * math.Round(percentage)))
 			if percentage >= hundredPercent/2 {
 				firstHalfPercentage = strings.Repeat("=", p.space)

internal/console/helpers/helpers_test.go

@@ -127,10 +127,10 @@ func TestProgressBar(t *testing.T) {
 			}
 			go progressBar.Start(&wg)
 			if tt.shouldCheckOutput {
-				for i := 0; i < 101; i++ {
-					progress <- float64(i)
+				for i := 0; i < 100; i++ {
+					progress <- float64(1)
 				}
-				progress <- float64(100)
+				progress <- float64(1)
 			}
 			wg.Wait()
 			splittedOut := strings.Split(out.String(), "\r")

internal/console/scan.go

@@ -29,6 +29,7 @@ import (
 	yamlParser "github.com/Checkmarx/kics/pkg/parser/yaml"
 	"github.com/Checkmarx/kics/pkg/resolver"
 	"github.com/Checkmarx/kics/pkg/resolver/helm"
+	"github.com/Checkmarx/kics/pkg/scanner"
 	"github.com/getsentry/sentry-go"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
@@ -397,7 +398,7 @@ func analyzePaths(paths, types, exclude []string) (typesRes, excludeRes []string
 func createService(inspector *engine.Inspector,
 	t kics.Tracker,
 	store kics.Storage,
-	querySource source.FilesystemSource) (*kics.Service, error) {
+	querySource source.FilesystemSource) ([]*kics.Service, error) {
 	filesSource, err := getFileSystemSourceProvider()
 	if err != nil {
 		return nil, err
@@ -421,14 +422,19 @@ func createService(inspector *engine.Inspector,
 		return nil, err
 	}
 
-	return &kics.Service{
-		SourceProvider: filesSource,
-		Storage:        store,
-		Parser:         combinedParser,
-		Inspector:      inspector,
-		Tracker:        t,
-		Resolver:       combinedResolver,
-	}, nil
+	services := make([]*kics.Service, 0, len(combinedParser))
+
+	for _, parser := range combinedParser {
+		services = append(services, &kics.Service{
+			SourceProvider: filesSource,
+			Storage:        store,
+			Parser:         parser,
+			Inspector:      inspector,
+			Tracker:        t,
+			Resolver:       combinedResolver,
+		})
+	}
+	return services, nil
 }
 
 func scan(changedDefaultQueryPath bool) error {
@@ -475,15 +481,15 @@ func scan(changedDefaultQueryPath bool) error {
 		return err
 	}
 
-	service, err := createService(inspector, t, store, *querySource)
+	services, err := createService(inspector, t, store, *querySource)
 	if err != nil {
 		log.Err(err)
 		return err
 	}
 
-	if scanErr := service.StartScan(ctx, scanID, noProgress); scanErr != nil {
-		log.Err(scanErr)
-		return scanErr
+	if err = scanner.StartScan(ctx, scanID, noProgress, services); err != nil {
+		log.Err(err)
+		return err
 	}
 
 	results, err := store.GetVulnerabilities(ctx, scanID)
@@ -523,7 +529,7 @@ func getSummary(t *tracker.CITracker, results []model.Vulnerability) model.Summa
 		ScannedFiles:           t.FoundFiles,
 		ParsedFiles:            t.ParsedFiles,
 		TotalQueries:           t.LoadedQueries,
-		FailedToExecuteQueries: t.LoadedQueries - t.ExecutedQueries,
+		FailedToExecuteQueries: t.ExecutingQueries - t.ExecutedQueries,
 		FailedSimilarityID:     t.FailedSimilarityID,
 	}
 
@@ -581,10 +587,12 @@ func printOutput(outputPath, filename string, body interface{}, formats []string
 func gracefulShutdown() {
 	c := make(chan os.Signal)
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
-	go func() {
+	showErrors := consoleHelpers.ShowError("errors")
+	interruptCode := constants.SignalInterruptCode
+	go func(showErrors bool, interruptCode int) {
 		<-c
-		if consoleHelpers.ShowError("errors") {
-			os.Exit(constants.SignalInterruptCode)
+		if showErrors {
+			os.Exit(interruptCode)
 		}
-	}()
+	}(showErrors, interruptCode)
 }

internal/tracker/ci.go

@@ -10,6 +10,7 @@ import (
 // and how many files were found and executed
 type CITracker struct {
 	LoadedQueries      int
+	ExecutingQueries   int
 	ExecutedQueries    int
 	FoundFiles         int
 	ParsedFiles        int
@@ -39,6 +40,11 @@ func (c *CITracker) TrackQueryLoad(queryAggregation int) {
 	c.LoadedQueries += queryAggregation
 }
 
+// TrackQueryExecuting adds a executing queries
+func (c *CITracker) TrackQueryExecuting(queryAggregation int) {
+	c.ExecutingQueries += queryAggregation
+}
+
 // TrackQueryExecution adds a query executed
 func (c *CITracker) TrackQueryExecution(queryAggregation int) {
 	c.ExecutedQueries += queryAggregation

pkg/engine/inspector.go

@@ -3,12 +3,9 @@ package engine
 import (
 	"context"
 	"encoding/json"
-	"fmt"
-	"io"
-	"sync"
+	"strings"
 	"time"
 
-	consoleHelpers "github.com/Checkmarx/kics/internal/console/helpers"
 	"github.com/Checkmarx/kics/internal/metrics"
 	"github.com/Checkmarx/kics/pkg/detector"
 	"github.com/Checkmarx/kics/pkg/detector/docker"
@@ -53,6 +50,7 @@ type VulnerabilityBuilder func(ctx *QueryContext, tracker Tracker, v interface{}
 // GetOutputLines returns the number of lines to be displayed in results outputs
 type Tracker interface {
 	TrackQueryLoad(queryAggregation int)
+	TrackQueryExecuting(queryAggregation int)
 	TrackQueryExecution(queryAggregation int)
 	FailedDetectLine()
 	FailedComputeSimilarityID()
@@ -158,6 +156,7 @@ func NewInspector(
 			})
 		}
 	}
+
 	failedQueries := make(map[string]error)
 
 	queriesNumber := sumAllAggregatedQueries(opaQueries)
@@ -193,22 +192,15 @@ func sumAllAggregatedQueries(opaQueries []*preparedQuery) int {
 	return sum
 }
 
-func startProgressBar(hideProgress bool, total int, wg *sync.WaitGroup, progressChannel chan float64) {
-	wg.Add(1)
-	progressBar := consoleHelpers.NewProgressBar("Executing queries: ", 10, float64(total), progressChannel)
-	if hideProgress {
-		progressBar.Writer = io.Discard
-	}
-	go progressBar.Start(wg)
-}
-
 // Inspect scan files and return the a list of vulnerabilities found on the process
 func (c *Inspector) Inspect(
 	ctx context.Context,
 	scanID string,
 	files model.FileMetadatas,
 	hideProgress bool,
-	baseScanPaths []string) ([]model.Vulnerability, error) {
+	baseScanPaths []string,
+	platforms []string,
+	currentQuery chan<- float64) ([]model.Vulnerability, error) {
 	log.Debug().Msg("engine.Inspect()")
 	combinedFiles := files.Combine()
 
@@ -219,12 +211,9 @@ func (c *Inspector) Inspect(
 
 	var vulnerabilities []model.Vulnerability
 	vulnerabilities = make([]model.Vulnerability, 0)
-	currentQuery := make(chan float64, 1)
-	var wg sync.WaitGroup
-	startProgressBar(hideProgress, len(c.queries), &wg, currentQuery)
-	for idx, query := range c.queries {
+	for _, query := range c.getQueriesByPlat(platforms) {
 		if !hideProgress {
-			currentQuery <- float64(idx)
+			currentQuery <- float64(1)
 		}
 
 		vuls, err := c.doRun(&QueryContext{
@@ -250,12 +239,32 @@ func (c *Inspector) Inspect(
 
 		c.tracker.TrackQueryExecution(query.metadata.Aggregation)
 	}
-	close(currentQuery)
-	wg.Wait()
-	fmt.Println("\r")
+
 	return vulnerabilities, nil
 }
 
+// LenQueriesByPlat returns the number of queries by platforms
+func (c *Inspector) LenQueriesByPlat(platforms []string) int {
+	count := 0
+	for _, query := range c.queries {
+		if contains(platforms, query.metadata.Platform) {
+			c.tracker.TrackQueryExecuting(query.metadata.Aggregation)
+			count++
+		}
+	}
+	return count
+}
+
+func (c *Inspector) getQueriesByPlat(platforms []string) []*preparedQuery {
+	queries := make([]*preparedQuery, 0)
+	for _, query := range c.queries {
+		if contains(platforms, query.metadata.Platform) {
+			queries = append(queries, query)
+		}
+	}
+	return queries
+}
+
 // EnableCoverageReport enables the flag to create a coverage report
 func (c *Inspector) EnableCoverageReport() {
 	c.enableCoverageReport = true
@@ -360,3 +369,20 @@ func (c *Inspector) decodeQueryResults(ctx *QueryContext, results rego.ResultSet
 
 	return vulnerabilities, nil
 }
+
+// contains is a simple method to check if a slice
+// contains an entry
+func contains(s []string, e string) bool {
+	if e == "common" {
+		return true
+	}
+	if e == "k8s" {
+		e = "kubernetes"
+	}
+	for _, a := range s {
+		if strings.EqualFold(a, e) {
+			return true
+		}
+	}
+	return false
+}

pkg/engine/inspector_test.go

@@ -289,6 +289,7 @@ func TestInspect(t *testing.T) { //nolint
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			currentQuery := make(chan float64)
 			c := &Inspector{
 				queries:              tt.fields.queries,
 				vb:                   tt.fields.vb,
@@ -299,7 +300,8 @@ func TestInspect(t *testing.T) { //nolint
 				detector:             inspDetector,
 				queryExecTimeout:     time.Duration(60) * time.Second,
 			}
-			got, err := c.Inspect(tt.args.ctx, tt.args.scanID, tt.args.files, true, []string{filepath.FromSlash("assets/queries/")})
+			got, err := c.Inspect(tt.args.ctx, tt.args.scanID, tt.args.files,
+				true, []string{filepath.FromSlash("assets/queries/")}, []string{""}, currentQuery)
 			if tt.wantErr {
 				if err == nil {
 					t.Errorf("Inspector.Inspect() = %v,\nwant %v", err, tt.want)

pkg/engine/provider/filesystem.go

@@ -94,6 +94,9 @@ func (s *FileSystemSourceProvider) GetSources(ctx context.Context,
 		if !fileInfo.IsDir() {
 			c, openFileErr := openScanFile(scanPath, extensions)
 			if openFileErr != nil {
+				if openFileErr == ErrNotSupportedFile {
+					continue
+				}
 				return openFileErr
 			}
 			if sinkErr := sink(ctx, scanPath, c); sinkErr != nil {
@@ -185,7 +188,7 @@ func (s *FileSystemSourceProvider) checkConditions(info os.FileInfo, extensions
 	}
 
 	if f, ok := s.excludes[info.Name()]; ok && containsFile(f, info) {
-		log.Info().Msgf("File ignored: %s", path)
+		log.Trace().Msgf("File ignored: %s", path)
 		return true, nil
 	}
 	if !extensions.Include(filepath.Ext(path)) && !extensions.Include(filepath.Base(path)) {