⭐️ plugin cache + verify github conn once

.vscode/launch.json

@@ -198,6 +198,18 @@
         "shell", "ssh", "[email protected]",
       ],
     },
+    {
+      "name": "scan github org",
+      "type": "go",
+      "request": "launch",
+      "program": "${workspaceRoot}/apps/cnquery/cnquery.go",
+      "args": [
+        "scan",
+        "github",
+        "org", "hit-training",
+        "--log-level", "trace"
+      ]
+    },
     {
       "name": "Configure Built-in Providers",
       "type": "go",

Makefile

@@ -700,6 +700,10 @@ test: test/go test/lint
 benchmark/go:
 	go test -bench=. -benchmem go.mondoo.com/cnquery/v11/explorer/scan/benchmark
 
+race/go:
+	go test -race go.mondoo.com/cnquery/v11/internal/workerpool
+	go test -race go.mondoo.com/cnquery/v11/explorer/scan
+
 test/generate: prep/tools/mockgen
 	go generate ./providers
 

explorer/scan/discovery.go

@@ -6,11 +6,13 @@ package scan
 import (
 	"context"
 	"errors"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog/log"
 	"go.mondoo.com/cnquery/v11/cli/config"
 	"go.mondoo.com/cnquery/v11/cli/execruntime"
+	"go.mondoo.com/cnquery/v11/internal/workerpool"
 	"go.mondoo.com/cnquery/v11/llx"
 	"go.mondoo.com/cnquery/v11/logger"
 	"go.mondoo.com/cnquery/v11/providers"
@@ -20,6 +22,9 @@ import (
 	"go.mondoo.com/cnquery/v11/providers-sdk/v1/upstream"
 )
 
+// number of parallel goroutines discovering assets
+const workers = 10
+
 type AssetWithRuntime struct {
 	Asset   *inventory.Asset
 	Runtime *providers.Runtime
@@ -34,28 +39,30 @@ type DiscoveredAssets struct {
 	platformIds map[string]struct{}
 	Assets      []*AssetWithRuntime
 	Errors      []*AssetWithError
+	assetsLock  sync.Mutex
 }
 
 // Add adds an asset and its runtime to the discovered assets list. It returns true if the
 // asset has been added, false if it is a duplicate
 func (d *DiscoveredAssets) Add(asset *inventory.Asset, runtime *providers.Runtime) bool {
-	isDuplicate := false
+	d.assetsLock.Lock()
+	defer d.assetsLock.Unlock()
+
 	for _, platformId := range asset.PlatformIds {
 		if _, ok := d.platformIds[platformId]; ok {
-			isDuplicate = true
-			break
+			// duplicate
+			return false
 		}
 		d.platformIds[platformId] = struct{}{}
 	}
-	if isDuplicate {
-		return false
-	}
 
 	d.Assets = append(d.Assets, &AssetWithRuntime{Asset: asset, Runtime: runtime})
 	return true
 }
 
 func (d *DiscoveredAssets) AddError(asset *inventory.Asset, err error) {
+	d.assetsLock.Lock()
+	defer d.assetsLock.Unlock()
 	d.Errors = append(d.Errors, &AssetWithError{Asset: asset, Err: err})
 }
 
@@ -161,17 +168,28 @@ func discoverAssets(rootAssetWithRuntime *AssetWithRuntime, resolvedRootAsset *i
 		return
 	}
 
+	pool := workerpool.New[*AssetWithRuntime](workers)
+	pool.Start()
+	defer pool.Close()
+
 	// for all discovered assets, we apply mondoo-specific labels and annotations that come from the root asset
-	for _, a := range rootAssetWithRuntime.Runtime.Provider.Connection.Inventory.Spec.Assets {
-		// create runtime for root asset
-		assetWithRuntime, err := createRuntimeForAsset(a, upstream, recording)
-		if err != nil {
-			log.Error().Err(err).Str("asset", a.Name).Msg("unable to create runtime for asset")
-			discoveredAssets.AddError(a, err)
-			continue
-		}
+	for _, asset := range rootAssetWithRuntime.Runtime.Provider.Connection.Inventory.Spec.Assets {
+		pool.Submit(func() (*AssetWithRuntime, error) {
+			assetWithRuntime, err := createRuntimeForAsset(asset, upstream, recording)
+			if err != nil {
+				log.Error().Err(err).Str("asset", asset.GetName()).Msg("unable to create runtime for asset")
+				discoveredAssets.AddError(asset, err)
+			}
+			return assetWithRuntime, nil
+		})
+	}
+
+	// Wait for the workers to finish processing
+	pool.Wait()
 
-		// If no asset was returned and no error, then we observed a duplicate asset with a
+	// Get all assets with runtimes from the pool
+	for _, assetWithRuntime := range pool.GetResults() {
+		// If asset is nil, then we observed a duplicate asset with a
 		// runtime that already exists.
 		if assetWithRuntime == nil {
 			continue

internal/datalakes/inmemory/inmemory.go

@@ -8,20 +8,21 @@ import (
 
 	"github.com/google/uuid"
 	"go.mondoo.com/cnquery/v11/explorer"
+	"go.mondoo.com/cnquery/v11/internal/datalakes/inmemory/store"
 	"go.mondoo.com/cnquery/v11/llx"
 )
 
 // Db is the database backend, it allows the interaction with the underlying data.
 type Db struct {
-	cache       kvStore
+	cache       store.KeyValue
 	services    *explorer.LocalServices // bidirectional connection between db + services
 	uuid        string                  // used for all object identifiers to prevent clashes (eg in-memory pubsub)
 	nowProvider func() time.Time
 }
 
 // NewServices creates a new set of backend services
 func NewServices(runtime llx.Runtime) (*Db, *explorer.LocalServices, error) {
-	var cache kvStore = newKissDb()
+	var cache store.KeyValue = store.NewKissDb()
 
 	db := &Db{
 		cache:       cache,

internal/datalakes/inmemory/kiss.go → internal/datalakes/inmemory/store/kiss.go

@@ -1,12 +1,12 @@
 // Copyright (c) Mondoo, Inc.
 // SPDX-License-Identifier: BUSL-1.1
 
-package inmemory
+package store
 
 import "sync"
 
-// kvStore is an general-purpose abstraction for key-value stores
-type kvStore interface {
+// KeyValue is an general-purpose abstraction for key-value stores
+type KeyValue interface {
 	Get(key interface{}) (interface{}, bool)
 	Set(key interface{}, value interface{}, cost int64) bool
 	Del(key interface{})
@@ -18,7 +18,7 @@ type kissDb struct {
 	data map[string]interface{}
 }
 
-func newKissDb() *kissDb {
+func NewKissDb() *kissDb {
 	return &kissDb{
 		data: map[string]interface{}{},
 	}

internal/workerpool/pool.go

@@ -92,7 +92,7 @@ func (p *Pool[R]) Close() {
 
 // Wait waits until all tasks have been processed.
 func (p *Pool[R]) Wait() {
-	ticker := time.NewTicker(100 * time.Millisecond)
+	ticker := time.NewTicker(10 * time.Millisecond)
 	for {
 		if !p.Processing() {
 			return