all: new exportable indexer and datastore packages (#587)

In order to leverage claircore as a library dependency some
packages (and associated interfaces) need to be publicly consumable.
This change also unifies the postgres implementations.

To implement a new backend or a noop backend it's
be good to get all the postgres specific implementation
details in one place.

Signed-off-by: crozzy <[email protected]>

Author: crozzy

alpine/distributionscanner.go

@@ -13,7 +13,7 @@ import (
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/indexer"
+	"github.com/quay/claircore/indexer"
 	"github.com/quay/claircore/osrelease"
 	"github.com/quay/claircore/pkg/tarfs"
 )

alpine/ecosystem.go

@@ -3,8 +3,8 @@ package alpine
 import (
 	"context"
 
-	"github.com/quay/claircore/internal/indexer"
-	"github.com/quay/claircore/internal/indexer/linux"
+	"github.com/quay/claircore/indexer"
+	"github.com/quay/claircore/indexer/linux"
 )
 
 // NewEcosystem provides the set of scanners and coalescers for the alpine ecosystem

alpine/packagescanner.go

@@ -10,7 +10,7 @@ import (
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/indexer"
+	"github.com/quay/claircore/indexer"
 	"github.com/quay/claircore/pkg/tarfs"
 )
 

aws/distributionscanner.go

@@ -9,7 +9,7 @@ import (
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/indexer"
+	"github.com/quay/claircore/indexer"
 )
 
 // AWS Linux keeps a consistent os-release file between

cmd/cctool/unpack.go

@@ -61,10 +61,9 @@ func Unpack(cmd context.Context, cfg *commonConfig, args []string) error {
 	defer done()
 
 	log.Printf("fetching layers")
-	a := &libindex.FetchArena{}
-	a.Init(http.DefaultClient, os.TempDir())
-	f := a.Fetcher()
-	err = f.Fetch(ctx, m.Layers)
+	a := libindex.NewRemoteFetchArena(http.DefaultClient, os.TempDir())
+	r := a.Realizer(ctx)
+	err = r.Realize(ctx, m.Layers)
 	if err != nil {
 		return err
 	}
@@ -115,7 +114,7 @@ func Unpack(cmd context.Context, cfg *commonConfig, args []string) error {
 			log.Printf("failed to recursively remove %v: %v", td, err)
 		}
 		log.Printf("deleting downloaded layers in tmp dir")
-		err = f.Close()
+		err = r.Close()
 		if err != nil {
 			log.Printf("failed to clean layer files in tmp directory: %v", err)
 		}

cmd/libindexhttp/main.go

@@ -11,7 +11,9 @@ import (
 	"github.com/quay/zlog"
 	"github.com/rs/zerolog"
 
+	"github.com/quay/claircore/datastore/postgres"
 	"github.com/quay/claircore/libindex"
+	"github.com/quay/claircore/pkg/ctxlock"
 )
 
 // Config this struct is using the goconfig library for simple flag and env var
@@ -41,9 +43,19 @@ func main() {
 	log = log.Level(logLevel(conf))
 	zlog.Set(&log)
 
-	opts := &libindex.Opts{
-		ConnString: conf.ConnString,
-		Migrations: true,
+	pool, err := postgres.Connect(ctx, conf.ConnString, "libindex")
+	if err != nil {
+		log.Fatal().Msgf("failed to create db pool: %v", err)
+	}
+	store, err := postgres.InitPostgresIndexerStore(ctx, pool, true)
+	if err != nil {
+		log.Fatal().Msgf("failed to create store: %v", err)
+	}
+
+	opts := &libindex.Options{
+		Store:      store,
+		Locker:     &ctxlock.Locker{},
+		FetchArena: libindex.NewRemoteFetchArena(http.DefaultClient, os.TempDir()),
 	}
 
 	// create libindex

internal/vulnstore/enrichment.go datastore/enrichment.go

@@ -1,4 +1,4 @@
-package vulnstore
+package datastore
 
 import (
 	"context"

datastore/matcher_store.go

@@ -0,0 +1,8 @@
+package datastore
+
+// MatcherStore aggregates all interface types
+type MatcherStore interface {
+	Updater
+	Vulnerability
+	Enrichment
+}

internal/indexer/postgres/affected_manifests_e2e_test.go datastore/postgres/affected_manifests_e2e_test.go

@@ -12,8 +12,10 @@ import (
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/indexer"
+	"github.com/quay/claircore/indexer"
+	"github.com/quay/claircore/pkg/omnimatcher"
 	"github.com/quay/claircore/test/integration"
+	pgtest "github.com/quay/claircore/test/postgres"
 )
 
 type affectedE2E struct {
@@ -27,8 +29,8 @@ type affectedE2E struct {
 func TestAffectedE2E(t *testing.T) {
 	integration.NeedDB(t)
 	ctx := zlog.Test(context.Background(), t)
-	pool := TestDatabase(ctx, t)
-	store := NewStore(pool)
+	pool := pgtest.TestIndexerDB(ctx, t)
+	store := NewIndexerStore(pool)
 
 	table := []struct {
 		// name of the defined affectedE2E test
@@ -268,8 +270,9 @@ func (e *affectedE2E) IndexManifest(t *testing.T) {
 // manifest is affected.
 func (e *affectedE2E) AffectedManifests(t *testing.T) {
 	ctx := zlog.Test(e.ctx, t)
+	om := omnimatcher.New(nil)
 	for _, vuln := range e.vr.Vulnerabilities {
-		hashes, err := e.store.AffectedManifests(ctx, *vuln)
+		hashes, err := e.store.AffectedManifests(ctx, *vuln, om.Vulnerable)
 		if err != nil {
 			t.Fatalf("failed to retrieve affected manifest for vuln %s: %v", vuln.ID, err)
 		}

internal/indexer/postgres/affectedmanifest.go datastore/postgres/affectedmanifest.go

@@ -15,7 +15,6 @@ import (
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/pkg/omnimatcher"
 )
 
 var (
@@ -68,7 +67,7 @@ var (
 //
 // The manifest index is then queried to resolve a list of manifest hashes containing the affected
 // artifacts.
-func (s *store) AffectedManifests(ctx context.Context, v claircore.Vulnerability) ([]claircore.Digest, error) {
+func (s *IndexerStore) AffectedManifests(ctx context.Context, v claircore.Vulnerability, vulnFunc claircore.CheckVulnernableFunc) ([]claircore.Digest, error) {
 	const (
 		selectPackages = `
 SELECT
@@ -108,7 +107,7 @@ WHERE
 		);
 `
 	)
-	ctx = zlog.ContextWithValues(ctx, "component", "internal/indexer/postgres/affectedManifests")
+	ctx = zlog.ContextWithValues(ctx, "component", "datastore/postgres/affectedManifests")
 
 	// confirm the incoming vuln can be
 	// resolved into a prototype index record
@@ -179,10 +178,9 @@ WHERE
 	// for each package discovered create an index record
 	// and determine if any in-tree matcher finds the record vulnerable
 	var filteredRecords []claircore.IndexRecord
-	om := omnimatcher.New(nil)
 	for _, pkg := range pkgsToFilter {
 		pr.Package = &pkg
-		match, err := om.Vulnerable(ctx, &pr, &v)
+		match, err := vulnFunc(ctx, &pr, &v)
 		if err != nil {
 			return nil, err
 		}
@@ -276,7 +274,7 @@ func protoRecord(ctx context.Context, pool *pgxpool.Pool, v claircore.Vulnerabil
 		`
 		timeout = 5 * time.Second
 	)
-	ctx = zlog.ContextWithValues(ctx, "component", "internal/indexer/postgres/protoRecord")
+	ctx = zlog.ContextWithValues(ctx, "component", "datastore/postgres/protoRecord")
 
 	protoRecord := claircore.IndexRecord{}
 	// fill dist into prototype index record if exists

datastore/postgres/connect.go

@@ -0,0 +1,38 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/quay/claircore/pkg/poolstats"
+	"github.com/quay/zlog"
+)
+
+// Connect initialize a postgres pgxpool.Pool based on the connection string
+func Connect(ctx context.Context, connString string, applicationName string) (*pgxpool.Pool, error) {
+	// we are going to use pgx for more control over connection pool and
+	// and a cleaner api around bulk inserts
+	cfg, err := pgxpool.ParseConfig(connString)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse ConnString: %v", err)
+	}
+	cfg.MaxConns = 30
+	const appnameKey = `application_name`
+	params := cfg.ConnConfig.RuntimeParams
+	if _, ok := params[appnameKey]; !ok {
+		params[appnameKey] = applicationName
+	}
+
+	pool, err := pgxpool.ConnectConfig(ctx, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create ConnPool: %v", err)
+	}
+
+	if err := prometheus.Register(poolstats.NewCollector(pool, applicationName)); err != nil {
+		zlog.Info(ctx).Msg("pool metrics already registered")
+	}
+
+	return pool, nil
+}

internal/indexer/postgres/deletemanifests.go datastore/postgres/deletemanifests.go

@@ -31,16 +31,16 @@ var (
 	)
 )
 
-func (s *store) DeleteManifests(ctx context.Context, d ...claircore.Digest) ([]claircore.Digest, error) {
-	ctx = zlog.ContextWithValues(ctx, "component", "internal/indexer/postgres/DeleteManifests")
+func (s *IndexerStore) DeleteManifests(ctx context.Context, d ...claircore.Digest) ([]claircore.Digest, error) {
+	ctx = zlog.ContextWithValues(ctx, "component", "datastore/postgres/DeleteManifests")
 	rm, err := s.deleteManifests(ctx, d)
 	if err != nil {
 		return nil, err
 	}
 	return rm, s.layerCleanup(ctx)
 }
 
-func (s *store) deleteManifests(ctx context.Context, d []claircore.Digest) ([]claircore.Digest, error) {
+func (s *IndexerStore) deleteManifests(ctx context.Context, d []claircore.Digest) ([]claircore.Digest, error) {
 	const deleteManifest = `DELETE FROM manifest WHERE hash = ANY($1::TEXT[]) RETURNING manifest.hash;`
 	var err error
 	defer promTimer(deleteManifestsDuration, "deleteManifest", &err)()
@@ -72,7 +72,7 @@ func (s *store) deleteManifests(ctx context.Context, d []claircore.Digest) ([]cl
 	return rm, nil
 }
 
-func (s *store) layerCleanup(ctx context.Context) (err error) {
+func (s *IndexerStore) layerCleanup(ctx context.Context) (err error) {
 	const layerCleanup = `DELETE FROM layer WHERE NOT EXISTS (SELECT FROM manifest_layer WHERE manifest_layer.layer_id = layer.id);`
 	defer promTimer(deleteManifestsDuration, "layerCleanup", &err)()
 	tag, err := s.pool.Exec(ctx, layerCleanup)

internal/indexer/postgres/deletemanifests_test.go datastore/postgres/deletemanifests_test.go

@@ -10,13 +10,14 @@ import (
 	"github.com/quay/claircore"
 	"github.com/quay/claircore/test"
 	"github.com/quay/claircore/test/integration"
+	pgtest "github.com/quay/claircore/test/postgres"
 )
 
 func TestDeleteManifests(t *testing.T) {
 	integration.NeedDB(t)
 	ctx := zlog.Test(context.Background(), t)
-	pool := TestDatabase(ctx, t)
-	store := NewStore(pool)
+	pool := pgtest.TestIndexerDB(ctx, t)
+	store := NewIndexerStore(pool)
 	defer store.Close(ctx)
 
 	t.Run("Nonexistent", func(t *testing.T) {

internal/indexer/postgres/digest.go datastore/postgres/digest.go

@@ -12,7 +12,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/indexer"
+	"github.com/quay/claircore/indexer"
 )
 
 var (
@@ -37,7 +37,7 @@ var (
 	)
 )
 
-func (s *store) DistributionsByLayer(ctx context.Context, hash claircore.Digest, scnrs indexer.VersionedScanners) ([]*claircore.Distribution, error) {
+func (s *IndexerStore) DistributionsByLayer(ctx context.Context, hash claircore.Digest, scnrs indexer.VersionedScanners) ([]*claircore.Distribution, error) {
 	const (
 		selectScanner = `
 		SELECT id

internal/indexer/postgres/digest_test.go datastore/postgres/digest_test.go

@@ -63,7 +63,7 @@ var (
 // UpdateEnrichments creates a new UpdateOperation, inserts the provided
 // EnrichmentRecord(s), and ensures enrichments from previous updates are not
 // queried by clients.
-func (s *Store) UpdateEnrichments(ctx context.Context, name string, fp driver.Fingerprint, es []driver.EnrichmentRecord) (uuid.UUID, error) {
+func (s *MatcherStore) UpdateEnrichments(ctx context.Context, name string, fp driver.Fingerprint, es []driver.EnrichmentRecord) (uuid.UUID, error) {
 	const (
 		create = `
 INSERT
@@ -107,7 +107,7 @@ ON CONFLICT
 DO
 	NOTHING;`
 	)
-	ctx = zlog.ContextWithValues(ctx, "component", "internal/vulnstore/postgres/UpdateEnrichments")
+	ctx = zlog.ContextWithValues(ctx, "component", "datastore/postgres/UpdateEnrichments")
 
 	var id uint64
 	var ref uuid.UUID
@@ -172,7 +172,7 @@ func hashEnrichment(r *driver.EnrichmentRecord) (k string, d []byte) {
 	return "md5", h.Sum(nil)
 }
 
-func (s *Store) GetEnrichment(ctx context.Context, name string, tags []string) (res []driver.EnrichmentRecord, err error) {
+func (s *MatcherStore) GetEnrichment(ctx context.Context, name string, tags []string) (res []driver.EnrichmentRecord, err error) {
 	const query = `
 WITH
 	latest
@@ -195,7 +195,7 @@ WHERE
 	AND uo.enrich = e.id
 	AND e.tags && $2::text[];`
 
-	ctx = zlog.ContextWithValues(ctx, "component", "internal/vulnstore/postgres/GetEnrichment")
+	ctx = zlog.ContextWithValues(ctx, "component", "datastore/postgres/GetEnrichment")
 	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
 		getEnrichmentsDuration.WithLabelValues("query", strconv.FormatBool(errors.Is(err, nil)))
 	}))

internal/indexer/postgres/distributionsbylayer.go datastore/postgres/distributionsbylayer.go

@@ -53,7 +53,7 @@ const (
 // The GC is throttled to not overload the database with cascade deletes.
 // If a full GC is required run this method until the returned int64 value
 // is 0.
-func (s *Store) GC(ctx context.Context, keep int) (int64, error) {
+func (s *MatcherStore) GC(ctx context.Context, keep int) (int64, error) {
 	// obtain update operations which need deletin'
 	ops, totalOps, err := eligibleUpdateOpts(ctx, s.pool, keep)
 	if err != nil {

internal/indexer/postgres/doc.go datastore/postgres/doc.go

@@ -16,6 +16,7 @@ import (
 	"github.com/quay/claircore/pkg/ctxlock"
 	"github.com/quay/claircore/test"
 	"github.com/quay/claircore/test/integration"
+	pgtest "github.com/quay/claircore/test/postgres"
 )
 
 type updaterMock struct {
@@ -102,16 +103,16 @@ func TestGC(t *testing.T) {
 	for _, tt := range table {
 		t.Run(tt.name, func(t *testing.T) {
 			ctx := zlog.Test(context.Background(), t)
-			pool := TestDB(ctx, t)
-			store := NewVulnStore(pool)
+			pool := pgtest.TestMatcherDB(ctx, t)
+			store := NewMatcherStore(pool)
 			locks, err := ctxlock.New(ctx, pool)
 			if err != nil {
 				t.Error(err)
 			}
 			defer locks.Close(ctx)
 			mgr, err := updates.NewManager(
 				ctx,
-				NewVulnStore(pool),
+				NewMatcherStore(pool),
 				locks,
 				http.DefaultClient,
 				updates.WithEnabled([]string{}),

internal/vulnstore/postgres/enrichment.go datastore/postgres/enrichment.go

@@ -9,10 +9,10 @@ import (
 	"github.com/jackc/pgx/v4"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/quay/claircore/datastore"
 	"github.com/quay/zlog"
 
 	"github.com/quay/claircore"
-	"github.com/quay/claircore/internal/vulnstore"
 )
 
 var (
@@ -37,7 +37,7 @@ var (
 )
 
 // Get implements vulnstore.Vulnerability.
-func (s *Store) Get(ctx context.Context, records []*claircore.IndexRecord, opts vulnstore.GetOpts) (map[string][]*claircore.Vulnerability, error) {
+func (s *MatcherStore) Get(ctx context.Context, records []*claircore.IndexRecord, opts datastore.GetOpts) (map[string][]*claircore.Vulnerability, error) {
 	ctx = zlog.ContextWithValues(ctx, "component", "internal/vulnstore/postgres/Get")
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {

internal/vulnstore/postgres/gc.go datastore/postgres/gc.go

@@ -55,7 +55,7 @@ var (
 	)
 )
 
-func (s *Store) GetUpdateDiff(ctx context.Context, prev, cur uuid.UUID) (*driver.UpdateDiff, error) {
+func (s *MatcherStore) GetUpdateDiff(ctx context.Context, prev, cur uuid.UUID) (*driver.UpdateDiff, error) {
 	// confirmRefs will return a row only if both refs are kind = 'vulnerability'
 	// therefore, if a pgx.ErrNoRows is returned from this query, at least one
 	// of the incoming refs is not of kind = 'vulnerability'.