refactor

Author: jesusfcr

pkg/store/findings.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -417,6 +418,54 @@ func findingsStates(ctx context.Context, tx *sqlx.Tx, log *log.Logger, s SourceF
 	return states, nil
 }
 
+func findingsStatesV2(ctx context.Context, tx *sqlx.Tx, log *log.Logger, source string, components []string) ([]findingState, error) {
+	findingsQ := `
+		WITH relevant_sources AS(
+			SELECT *
+			FROM sources
+			WHERE target_id = $1 AND component = ANY($2::TEXT[])
+		)
+		SELECT fe.finding_id as finding_id, fe.score as score, fe.fingerprint as fingerprint,
+		fe.affected_resource_string as affected_resource_string, fe.details as details,
+		fe.resources as resources, s.id as source_id, s.time as source_time,
+		f.status as status
+		FROM relevant_sources s
+		LEFT JOIN finding_events fe ON fe.source_id=s.id
+		LEFT JOIN findings f ON f.id=fe.finding_id
+		ORDER BY fe.finding_id, s.time;
+	`
+
+	// The query returns all the sources of any family that are capable of detecting any issue
+	// ever found for the target by a source belonging to the family of the source we are processing.
+	// Along with the source data, it retrieves the finding events, if any, sorted by finding_id and time.
+	//
+	// The 'relevant_sources' query gets all the sources that have been executed against the specified
+	// target which are in the list of all the sources that have ever detected an issue from the whole
+	// list of issues ever detected by the source we are processing.
+	//
+	// Example of data returned:
+	// 		           finding_id              | score |              source_id               |     source_time
+	// 	--------------------------------------+-------+--------------------------------------+---------------------
+	// 	 509b7663-7467-4bf4-95df-495aadcf70ac |   3.9 | ee51f3eb-cb59-474e-b6bd-449bf08990c8 | 2019-06-08 09:32:08
+	// 	 509b7663-7467-4bf4-95df-495aadcf70ac |   3.9 | 8b912b11-b66b-4b79-9acd-d1a47305162b | 2019-10-08 08:22:30
+	// 	 85ccdcc8-edbb-4691-be7b-32d9dd24b696 |   3.9 | 8b912b11-b66b-4b79-9acd-d1a47305162b | 2019-10-08 08:22:30
+	// 	 9ce52ab3-35ff-4c54-ad1c-e8b2b9d73e71 |   3.9 | ee51f3eb-cb59-474e-b6bd-449bf08990c8 | 2019-06-08 09:32:08
+	// 	 9ce52ab3-35ff-4c54-ad1c-e8b2b9d73e71 |   3.9 | 8b912b11-b66b-4b79-9acd-d1a47305162b | 2019-10-08 08:22:30
+	// 	                                      |       | 8b912b11-b66b-4b79-9acd-d1a47305162b | 2019-10-08 08:22:30
+	// 	                                      |       | ee51f3eb-cb59-474e-b6bd-449bf08990c8 | 2019-06-08 09:32:08
+
+	args := []interface{}{source, pq.Array(components)}
+	logQuery(log, "FindingStates", findingsQ, args...)
+	sourceF := []sourceFindings{}
+	err := tx.Select(&sourceF, findingsQ, args...)
+	if err != nil && !IsNotFoundErr(err) {
+		return nil, err
+	}
+
+	states := buildFindingStates(sourceF)
+	return states, nil
+}
+
 func buildFindingStates(sourceF []sourceFindings) []findingState {
 	if len(sourceF) == 0 {
 		return nil

pkg/store/sources.go

@@ -126,11 +126,12 @@ func (db *psqlxStore) ProcessSourceExecution(s Source, sourceFindings []SourceFi
 	// conditions from other possible concurrent sources being processed.
 	// To do that we have to lock based on all related sources that can
 	// have an effect on the current source being processed and its findings.
-	relatedFamilies, err := db.relatedFamilies(s.SourceFamily)
+	relatedComponents, err := db.relatedComponents(s.SourceFamily)
 	if err != nil {
 		return Source{}, err
 	}
-	err = db.lockTxBySources(tx, relatedFamilies)
+	// err = db.lockTxBySources(tx, relatedFamilies)
+	err = db.lockTxByComponent(tx, s.Target, relatedComponents)
 	if err != nil {
 		return Source{}, err
 	}
@@ -166,7 +167,7 @@ func (db *psqlxStore) ProcessSourceExecution(s Source, sourceFindings []SourceFi
 	}
 
 	tFindings := time.Now()
-	findingStates, err := findingsStates(ctx, tx, db.logger, screated.SourceFamily)
+	findingStates, err := findingsStatesV2(ctx, tx, db.logger, s.Target, relatedComponents)
 	if err != nil {
 		tx.Rollback()
 		return Source{}, err
@@ -257,6 +258,31 @@ func (db *psqlxStore) relatedFamilies(sf SourceFamily) (SourceFamilies, error) {
 	return families, nil
 }
 
+// RelatedFamilies returns the related families for the input SourceFamily.
+// That is the source families that are capable of detecting at least one
+// of the issues that the input source has ever detected for the specified target.
+func (db *psqlxStore) relatedComponents(sf SourceFamily) ([]string, error) {
+	q := `
+		SELECT UNNEST(comps)
+		FROM (
+			SELECT DISTINCT ARRAY_AGG(DISTINCT s.component) comps
+			FROM sources s INNER JOIN source_issues si ON si.source_id = s.id
+			WHERE s.target_id = $3 AND s.name=$1
+			GROUP BY issue_id
+			HAVING $2 = ANY(ARRAY_AGG(DISTINCT s.component))
+		) C
+	`
+
+	args := []interface{}{sf.Name, sf.Component, sf.Target}
+	logQuery(db.logger, "relatedFamilies", q, args...)
+	components := []string{}
+	err := db.DB.Select(&components, q, args...)
+	if err != nil && !IsNotFoundErr(err) {
+		return nil, err
+	}
+	return components, nil
+}
+
 // lockTxBySources performs an exclusive advisory lock on the given
 // transaction genereting a lock for each specified source family.
 func (db *psqlxStore) lockTxBySources(tx *sqlx.Tx, sff SourceFamilies) error {
@@ -285,6 +311,33 @@ func (db *psqlxStore) lockTxBySources(tx *sqlx.Tx, sff SourceFamilies) error {
 	return nil
 }
 
+// lockTxBySources performs an exclusive advisory lock on the given
+// transaction genereting a lock for each specified source family.
+func (db *psqlxStore) lockTxByComponent(tx *sqlx.Tx, target string, components []string) error {
+	// We use advisory locks and transactions to avoid race conditions
+	// and inconsistent DB state, for instance, with a source without its
+	// finding events, that would lead to wrong findind exposures calculation.
+
+	for _, sf := range components {
+		// A lock is obtained for every source family in scope.
+		// The key for the lock is built by concatenating the
+		// Name, Component and Target for each source family.
+		fkey := fmt.Sprintf("%s%s", sf, target)
+
+		h := fnv.New32()
+		h.Write([]byte(fkey))
+		k := h.Sum32()
+
+		_, err := tx.Exec("SELECT pg_advisory_xact_lock($1)", k)
+		if err != nil {
+			tx.Rollback()
+			return err
+		}
+	}
+
+	return nil
+}
+
 func createSource(ctx context.Context, tx *sqlx.Tx, s Source) (*Source, error) {
 	r := tx.QueryRowxContext(ctx, `INSERT INTO sources (name, component, instance, options, time, target_id)
 	  VALUES ($1, $2, $3, $4, $5, $6) RETURNING *`, s.Name, s.Component, s.Instance, s.Options, s.Time, s.Target)