clair: Implement worker detector support

The worker is changed to accommodate the new database model and API.
Worker is refactored to move the database query helper functions to pkg.

cmd/clair/main.go

@@ -102,27 +102,13 @@ func stopCPUProfiling(f *os.File) {
 }
 
 func configClairVersion(config *Config) {
-	listers := featurefmt.ListListers()
-	detectors := featurens.ListDetectors()
-	updaters := vulnsrc.ListUpdaters()
+	clair.EnabledDetectors = append(featurefmt.ListListers(), featurens.ListDetectors()...)
+	clair.EnabledUpdaters = strutil.Intersect(config.Updater.EnabledUpdaters, vulnsrc.ListUpdaters())
 
 	log.WithFields(log.Fields{
-		"Listers":   strings.Join(listers, ","),
-		"Detectors": strings.Join(detectors, ","),
-		"Updaters":  strings.Join(updaters, ","),
-	}).Info("Clair registered components")
-
-	unregUpdaters := strutil.CompareStringLists(config.Updater.EnabledUpdaters, updaters)
-	if len(unregUpdaters) != 0 {
-		log.WithFields(log.Fields{
-			"Unknown Updaters":   strings.Join(unregUpdaters, ","),
-			"Available Updaters": strings.Join(vulnsrc.ListUpdaters(), ","),
-		}).Fatal("Unknown or unregistered components are configured")
-	}
-
-	// All listers and detectors are enabled.
-	clair.Processors = database.Processors{Detectors: detectors, Listers: listers}
-	clair.EnabledUpdaters = strutil.CompareStringListsInBoth(config.Updater.EnabledUpdaters, updaters)
+		"Detectors": database.SerializeDetectors(clair.EnabledDetectors),
+		"Updaters":  clair.EnabledUpdaters,
+	}).Info("enabled Clair extensions")
 }
 
 // Boot starts Clair instance with the provided config.
@@ -147,6 +133,7 @@ func Boot(config *Config) {
 
 	defer db.Close()
 
+	clair.InitWorker(db)
 	// Start notifier
 	st.Begin()
 	go clair.RunNotifier(config.Notifier, db, st)
@@ -167,6 +154,18 @@ func Boot(config *Config) {
 	st.Stop()
 }
 
+// Initialize logging system
+func configureLogger(flagLogLevel *string) {
+	logLevel, err := log.ParseLevel(strings.ToUpper(*flagLogLevel))
+	if err != nil {
+		log.WithError(err).Error("failed to set logger parser level")
+	}
+
+	log.SetLevel(logLevel)
+	log.SetOutput(os.Stdout)
+	log.SetFormatter(&formatter.JSONExtendedFormatter{ShowLn: true})
+}
+
 func main() {
 	// Parse command-line arguments
 	flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError)
@@ -176,6 +175,7 @@ func main() {
 	flagInsecureTLS := flag.Bool("insecure-tls", false, "Disable TLS server's certificate chain and hostname verification when pulling layers.")
 	flag.Parse()
 
+	configureLogger(flagLogLevel)
 	// Check for dependencies.
 	for _, bin := range BinaryDependencies {
 		_, err := exec.LookPath(bin)
@@ -184,12 +184,6 @@ func main() {
 		}
 	}
 
-	// Initialize logging system
-	logLevel, err := log.ParseLevel(strings.ToUpper(*flagLogLevel))
-	log.SetLevel(logLevel)
-	log.SetOutput(os.Stdout)
-	log.SetFormatter(&formatter.JSONExtendedFormatter{ShowLn: true})
-
 	config, err := LoadConfig(*flagConfigPath)
 	if err != nil {
 		log.WithError(err).Fatal("failed to load configuration")

pkg/dbutil/dbutil.go

@@ -0,0 +1,288 @@
+// Copyright 2018 clair authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dbutil
+
+import (
+	"github.com/deckarep/golang-set"
+
+	"github.com/coreos/clair/database"
+)
+
+// DeduplicateNamespaces deduplicates a list of namespaces.
+func DeduplicateNamespaces(namespaces ...database.Namespace) []database.Namespace {
+	nsSet := mapset.NewSet()
+	for _, ns := range namespaces {
+		nsSet.Add(ns)
+	}
+
+	result := make([]database.Namespace, 0, nsSet.Cardinality())
+	for ns := range nsSet.Iter() {
+		result = append(result, ns.(database.Namespace))
+	}
+
+	return result
+}
+
+// DeduplicateFeatures deduplicates a list of list of features.
+func DeduplicateFeatures(features ...database.Feature) []database.Feature {
+	fSet := mapset.NewSet()
+	for _, f := range features {
+		fSet.Add(f)
+	}
+
+	result := make([]database.Feature, 0, fSet.Cardinality())
+	for f := range fSet.Iter() {
+		result = append(result, f.(database.Feature))
+	}
+
+	return result
+}
+
+// PersistPartialLayer wraps session PersistLayer function with begin and
+// commit.
+func PersistPartialLayer(datastore database.Datastore, layer *database.Layer) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if err := tx.PersistLayer(layer.Hash, layer.Features, layer.Namespaces, layer.By); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+// PersistFeatures wraps session PersistFeatures function with begin and commit.
+func PersistFeatures(datastore database.Datastore, features []database.Feature) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if err := tx.PersistFeatures(features); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+// PersistNamespaces wraps session PersistNamespaces function with begin and
+// commit.
+func PersistNamespaces(datastore database.Datastore, namespaces []database.Namespace) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if err := tx.PersistNamespaces(namespaces); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+// FindAncestry wraps session FindAncestry function with begin and rollback.
+func FindAncestry(datastore database.Datastore, name string) (database.Ancestry, bool, error) {
+	tx, err := datastore.Begin()
+	defer tx.Rollback()
+
+	if err != nil {
+		return database.Ancestry{}, false, err
+	}
+
+	return tx.FindAncestry(name)
+}
+
+// FindLayer wraps session FindLayer function with begin and rollback.
+func FindLayer(datastore database.Datastore, hash string) (layer database.Layer, ok bool, err error) {
+	var tx database.Session
+	if tx, err = datastore.Begin(); err != nil {
+		return
+	}
+
+	defer tx.Rollback()
+	layer, ok, err = tx.FindLayer(hash)
+	return
+}
+
+// DeduplicateNamespacedFeatures returns a copy of all unique features in the
+// input.
+func DeduplicateNamespacedFeatures(features []database.NamespacedFeature) []database.NamespacedFeature {
+	nsSet := mapset.NewSet()
+	for _, ns := range features {
+		nsSet.Add(ns)
+	}
+
+	result := make([]database.NamespacedFeature, 0, nsSet.Cardinality())
+	for ns := range nsSet.Iter() {
+		result = append(result, ns.(database.NamespacedFeature))
+	}
+
+	return result
+}
+
+// GetAncestryFeatures returns a list of unique namespaced features in the
+// ancestry.
+func GetAncestryFeatures(ancestry database.Ancestry) []database.NamespacedFeature {
+	features := []database.NamespacedFeature{}
+	for _, layer := range ancestry.Layers {
+		features = append(features, layer.GetFeatures()...)
+	}
+
+	return DeduplicateNamespacedFeatures(features)
+}
+
+// UpsertAncestry wraps session UpsertAncestry function with begin and commit.
+func UpsertAncestry(datastore database.Datastore, ancestry database.Ancestry) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+
+	if err = tx.UpsertAncestry(ancestry); err != nil {
+		tx.Rollback()
+		return err
+	}
+
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// PersistNamespacedFeatures wraps session PersistNamespacedFeatures function
+// with begin and commit.
+func PersistNamespacedFeatures(datastore database.Datastore, features []database.NamespacedFeature) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+
+	if err := tx.PersistNamespacedFeatures(features); err != nil {
+		tx.Rollback()
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// CacheRelatedVulnerability wraps session CacheAffectedNamespacedFeatures
+// function with begin and commit.
+func CacheRelatedVulnerability(datastore database.Datastore, features []database.NamespacedFeature) error {
+	tx, err := datastore.Begin()
+	if err != nil {
+		return err
+	}
+
+	if err := tx.CacheAffectedNamespacedFeatures(features); err != nil {
+		tx.Rollback()
+		return err
+	}
+
+	return tx.Commit()
+}
+
+// IntersectDetectors returns the detectors in both d1 and d2.
+func IntersectDetectors(d1 []database.Detector, d2 []database.Detector) []database.Detector {
+	d1Set := mapset.NewSet()
+	for _, d := range d1 {
+		d1Set.Add(d)
+	}
+
+	d2Set := mapset.NewSet()
+	for _, d := range d2 {
+		d2Set.Add(d)
+	}
+
+	inter := d1Set.Intersect(d2Set)
+	result := make([]database.Detector, 0, inter.Cardinality())
+	for d := range inter.Iter() {
+		result = append(result, d.(database.Detector))
+	}
+
+	return result
+}
+
+// DiffDetectors returns the detectors belongs to d1 but not d2
+func DiffDetectors(d1 []database.Detector, d2 []database.Detector) []database.Detector {
+	d1Set := mapset.NewSet()
+	for _, d := range d1 {
+		d1Set.Add(d)
+	}
+
+	d2Set := mapset.NewSet()
+	for _, d := range d2 {
+		d2Set.Add(d)
+	}
+
+	diff := d1Set.Difference(d2Set)
+	result := make([]database.Detector, 0, diff.Cardinality())
+	for d := range diff.Iter() {
+		result = append(result, d.(database.Detector))
+	}
+
+	return result
+}
+
+// MergeLayers merges all content in new layer to l, where the content is
+// updated.
+func MergeLayers(l *database.Layer, new *database.Layer) *database.Layer {
+	featureSet := mapset.NewSet()
+	namespaceSet := mapset.NewSet()
+	bySet := mapset.NewSet()
+
+	for _, f := range l.Features {
+		featureSet.Add(f)
+	}
+
+	for _, ns := range l.Namespaces {
+		namespaceSet.Add(ns)
+	}
+
+	for _, d := range l.By {
+		bySet.Add(d)
+	}
+
+	for _, feature := range new.Features {
+		if !featureSet.Contains(feature) {
+			l.Features = append(l.Features, feature)
+			featureSet.Add(feature)
+		}
+	}
+
+	for _, namespace := range new.Namespaces {
+		if !namespaceSet.Contains(namespace) {
+			l.Namespaces = append(l.Namespaces, namespace)
+			namespaceSet.Add(namespace)
+		}
+	}
+
+	for _, detector := range new.By {
+		if !bySet.Contains(detector) {
+			l.By = append(l.By, detector)
+			bySet.Add(detector)
+		}
+	}
+
+	return l
+}

pkg/strutil/strutil.go

@@ -14,42 +14,46 @@
 
 package strutil
 
-// CompareStringLists returns the strings that are present in X but not in Y.
-func CompareStringLists(X, Y []string) []string {
-	m := make(map[string]bool)
+import (
+	"regexp"
 
-	for _, y := range Y {
-		m[y] = true
-	}
+	set "github.com/deckarep/golang-set"
+)
+
+var urlParametersRegexp = regexp.MustCompile(`(\?|\&)([^=]+)\=([^ &]+)`)
 
-	diff := []string{}
+func convertToSet(X []string) set.Set {
+	s := set.NewSet()
 	for _, x := range X {
-		if m[x] {
-			continue
-		}
+		s.Add(x)
+	}
+	return s
+}
 
-		diff = append(diff, x)
-		m[x] = true
+func setToStringSlice(s set.Set) []string {
+	strs := make([]string, 0, s.Cardinality())
+	for _, str := range s.ToSlice() {
+		strs = append(strs, str.(string))
 	}
 
-	return diff
+	return strs
 }
 
-// CompareStringListsInBoth returns the strings that are present in both X and Y.
-func CompareStringListsInBoth(X, Y []string) []string {
-	m := make(map[string]struct{})
-
-	for _, y := range Y {
-		m[y] = struct{}{}
-	}
+// Difference returns the strings that are present in X but not in Y.
+func Difference(X, Y []string) []string {
+	x := convertToSet(X)
+	y := convertToSet(Y)
+	return setToStringSlice(x.Difference(y))
+}
 
-	diff := []string{}
-	for _, x := range X {
-		if _, e := m[x]; e {
-			diff = append(diff, x)
-			delete(m, x)
-		}
-	}
+// Intersect returns the strings that are present in both X and Y.
+func Intersect(X, Y []string) []string {
+	x := convertToSet(X)
+	y := convertToSet(Y)
+	return setToStringSlice(x.Intersect(y))
+}
 
-	return diff
+// CleanURL removes all parameters from an URL.
+func CleanURL(str string) string {
+	return urlParametersRegexp.ReplaceAllString(str, "")
 }

pkg/strutil/strutil_test.go

@@ -21,12 +21,12 @@ import (
 )
 
 func TestStringComparison(t *testing.T) {
-	cmp := CompareStringLists([]string{"a", "b", "b", "a"}, []string{"a", "c"})
+	cmp := Difference([]string{"a", "b", "b", "a"}, []string{"a", "c"})
 	assert.Len(t, cmp, 1)
 	assert.NotContains(t, cmp, "a")
 	assert.Contains(t, cmp, "b")
 
-	cmp = CompareStringListsInBoth([]string{"a", "a", "b", "c"}, []string{"a", "c", "c"})
+	cmp = Intersect([]string{"a", "a", "b", "c"}, []string{"a", "c", "c"})
 	assert.Len(t, cmp, 2)
 	assert.NotContains(t, cmp, "b")
 	assert.Contains(t, cmp, "a")

updater.go

@@ -21,6 +21,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/coreos/clair/pkg/dbutil"
+
 	"github.com/pborman/uuid"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
@@ -208,7 +210,7 @@ func update(datastore database.Datastore, firstUpdate bool) {
 		namespaces = append(namespaces, ns)
 	}
 
-	if err := persistNamespaces(datastore, namespaces); err != nil {
+	if err := dbutil.PersistNamespaces(datastore, namespaces); err != nil {
 		log.WithError(err).Error("Unable to insert namespaces")
 		return
 	}

worker.go

@@ -16,9 +16,9 @@ package clair
 
 import (
 	"errors"
-	"regexp"
 	"sync"
 
+	"github.com/deckarep/golang-set"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/coreos/clair/database"
@@ -26,11 +26,9 @@ import (
 	"github.com/coreos/clair/ext/featurens"
 	"github.com/coreos/clair/ext/imagefmt"
 	"github.com/coreos/clair/pkg/commonerr"
+	"github.com/coreos/clair/pkg/dbutil"
 	"github.com/coreos/clair/pkg/strutil"
-)
-
-const (
-	logLayerName = "layer"
+	"github.com/coreos/clair/pkg/tarutil"
 )
 
 var (
@@ -38,18 +36,8 @@ var (
 	// manager is not supported.
 	ErrUnsupported = commonerr.NewBadRequestError("worker: OS and/or package manager are not supported")
 
-	// ErrParentUnknown is the error that should be raised when a parent layer
-	// has yet to be processed for the current layer.
-	ErrParentUnknown = commonerr.NewBadRequestError("worker: parent layer is unknown, it must be processed first")
-
-	urlParametersRegexp = regexp.MustCompile(`(\?|\&)([^=]+)\=([^ &]+)`)
-
-	// Processors contain the names of namespace detectors and feature listers
-	// enabled in this instance of Clair.
-	//
-	// Processors are initialized during booting and configured in the
-	// configuration file.
-	Processors database.Processors
+	// EnabledDetectors are detectors to be used to scan the layers.
+	EnabledDetectors []database.Detector
 )
 
 // LayerRequest represents all information necessary to download and process a
@@ -60,296 +48,176 @@ type LayerRequest struct {
 	Headers map[string]string
 }
 
-// partialLayer stores layer's content detected by `processedBy` processors.
-type partialLayer struct {
-	hash        string
-	processedBy database.Processors
-	namespaces  []database.Namespace
-	features    []database.Feature
-
-	err error
+type processResult struct {
+	existingLayer   *database.Layer
+	newLayerContent *database.Layer
+	err             error
 }
 
-// processRequest stores parameters used for processing layers.
+// processRequest stores parameters used for processing a layer.
 type processRequest struct {
-	request LayerRequest
-	// notProcessedBy represents a set of processors used to process the
-	// request.
-	notProcessedBy database.Processors
+	LayerRequest
+
+	existingLayer *database.Layer
+	detectors     []database.Detector
 }
 
-// cleanURL removes all parameters from an URL.
-func cleanURL(str string) string {
-	return urlParametersRegexp.ReplaceAllString(str, "")
+type introducedFeature struct {
+	feature    database.AncestryFeature
+	layerIndex int
 }
 
-// processLayers in parallel processes a set of requests for unique set of layers
+// processRequests in parallel processes a set of requests for unique set of layers
 // and returns sets of unique namespaces, features and layers to be inserted
 // into the database.
-func processRequests(imageFormat string, toDetect []processRequest) ([]database.Namespace, []database.Feature, map[string]partialLayer, error) {
+func processRequests(imageFormat string, toDetect map[string]*processRequest) (map[string]*processResult, error) {
 	wg := &sync.WaitGroup{}
 	wg.Add(len(toDetect))
-	results := make([]partialLayer, len(toDetect))
+
+	results := map[string]*processResult{}
+	for i := range toDetect {
+		results[i] = nil
+	}
+
 	for i := range toDetect {
-		go func(req *processRequest, res *partialLayer) {
-			res.hash = req.request.Hash
-			res.processedBy = req.notProcessedBy
-			res.namespaces, res.features, res.err = detectContent(imageFormat, req.request.Hash, req.request.Path, req.request.Headers, req.notProcessedBy)
+		result := processResult{}
+		results[i] = &result
+		go func(req *processRequest, res *processResult) {
+			*res = *detectContent(imageFormat, req)
 			wg.Done()
-		}(&toDetect[i], &results[i])
+		}(toDetect[i], &result)
 	}
-	wg.Wait()
-	distinctNS := map[database.Namespace]struct{}{}
-	distinctF := map[database.Feature]struct{}{}
 
+	wg.Wait()
 	errs := []error{}
 	for _, r := range results {
 		errs = append(errs, r.err)
 	}
 
 	if err := commonerr.CombineErrors(errs...); err != nil {
-		return nil, nil, nil, err
-	}
-
-	updates := map[string]partialLayer{}
-	for _, r := range results {
-		for _, ns := range r.namespaces {
-			distinctNS[ns] = struct{}{}
-		}
-
-		for _, f := range r.features {
-			distinctF[f] = struct{}{}
-		}
-
-		if _, ok := updates[r.hash]; !ok {
-			updates[r.hash] = r
-		} else {
-			return nil, nil, nil, errors.New("Duplicated updates is not allowed")
-		}
-	}
-
-	namespaces := make([]database.Namespace, 0, len(distinctNS))
-	features := make([]database.Feature, 0, len(distinctF))
-
-	for ns := range distinctNS {
-		namespaces = append(namespaces, ns)
+		return nil, err
 	}
 
-	for f := range distinctF {
-		features = append(features, f)
-	}
-	return namespaces, features, updates, nil
+	return results, nil
 }
 
-func getLayer(datastore database.Datastore, req LayerRequest) (layer database.Layer, preq *processRequest, err error) {
-	var (
-		tx database.Session
-		ok bool
-	)
-
-	if tx, err = datastore.Begin(); err != nil {
-		return
-	}
-
-	defer tx.Rollback()
-
-	if layer, ok, err = tx.FindLayer(req.Hash); err != nil {
+func getProcessRequest(datastore database.Datastore, req LayerRequest) (preq *processRequest, err error) {
+	layer, ok, err := dbutil.FindLayer(datastore, req.Hash)
+	if err != nil {
 		return
 	}
 
 	if !ok {
-		layer = database.Layer{
-			LayerMetadata: database.LayerMetadata{
-				Hash: req.Hash,
-			},
-		}
-
+		log.WithField("layer", req.Hash).Debug("found no existing layer in database")
 		preq = &processRequest{
-			request:        req,
-			notProcessedBy: Processors,
+			LayerRequest:  req,
+			existingLayer: &database.Layer{Hash: req.Hash},
+			detectors:     EnabledDetectors,
 		}
 	} else {
-		notProcessed := getNotProcessedBy(layer.ProcessedBy)
-		if !(len(notProcessed.Detectors) == 0 && len(notProcessed.Listers) == 0 && ok) {
-			preq = &processRequest{
-				request:        req,
-				notProcessedBy: notProcessed,
-			}
+		log.WithFields(log.Fields{
+			"layer":           layer.Hash,
+			"detectors":       layer.By,
+			"feature count":   len(layer.Features),
+			"namespace count": len(layer.Namespaces),
+		}).Debug("found existing layer in database")
+
+		preq = &processRequest{
+			LayerRequest:  req,
+			existingLayer: &layer,
+			detectors:     dbutil.DiffDetectors(EnabledDetectors, layer.By),
 		}
 	}
 
 	return
 }
 
-// processLayers processes a set of post layer requests, stores layers and
-// returns an ordered list of processed layers with detected features and
-// namespaces.
-func processLayers(datastore database.Datastore, imageFormat string, requests []LayerRequest) ([]database.Layer, error) {
-	toDetect := []processRequest{}
-	layers := map[string]database.Layer{}
-	for _, req := range requests {
-		if _, ok := layers[req.Hash]; ok {
-			continue
-		}
-		layer, preq, err := getLayer(datastore, req)
-		if err != nil {
-			return nil, err
-		}
-		layers[req.Hash] = layer
-		if preq != nil {
-			toDetect = append(toDetect, *preq)
-		}
-	}
-
-	namespaces, features, partialLayers, err := processRequests(imageFormat, toDetect)
-	if err != nil {
-		return nil, err
+func persistProcessResult(datastore database.Datastore, results map[string]*processResult) error {
+	features := []database.Feature{}
+	namespaces := []database.Namespace{}
+	for _, r := range results {
+		features = append(features, r.newLayerContent.GetFeatures()...)
+		namespaces = append(namespaces, r.newLayerContent.GetNamespaces()...)
 	}
 
-	// Store partial results.
-	if err := persistNamespaces(datastore, namespaces); err != nil {
-		return nil, err
+	features = dbutil.DeduplicateFeatures(features...)
+	namespaces = dbutil.DeduplicateNamespaces(namespaces...)
+	if err := dbutil.PersistNamespaces(datastore, namespaces); err != nil {
+		return err
 	}
 
-	if err := persistFeatures(datastore, features); err != nil {
-		return nil, err
+	if err := dbutil.PersistFeatures(datastore, features); err != nil {
+		return err
 	}
 
-	for _, layer := range partialLayers {
-		if err := persistPartialLayer(datastore, layer); err != nil {
-			return nil, err
-		}
-
-		log.WithFields(log.Fields{
-			"Hash":                layer.hash,
-			"namespace count":     len(layer.namespaces),
-			"feature count":       len(layer.features),
-			"namespace detectors": layer.processedBy.Detectors,
-			"feature listers":     layer.processedBy.Listers,
-		}).Debug("saved layer")
-	}
-
-	// NOTE(Sida): The full layers are computed using partially
-	// processed layers in current database session. If any other instances of
-	// Clair are changing some layers in this set of layers, it might generate
-	// different results especially when the other Clair is with different
-	// processors.
-	completeLayers := []database.Layer{}
-	for _, req := range requests {
-		if partialLayer, ok := partialLayers[req.Hash]; ok {
-			completeLayers = append(completeLayers, combineLayers(layers[req.Hash], partialLayer))
-		} else {
-			completeLayers = append(completeLayers, layers[req.Hash])
+	for _, layer := range results {
+		if err := dbutil.PersistPartialLayer(datastore, layer.newLayerContent); err != nil {
+			return err
 		}
 	}
 
-	return completeLayers, nil
+	return nil
 }
 
-func persistPartialLayer(datastore database.Datastore, layer partialLayer) error {
-	tx, err := datastore.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
+// processLayers processes a set of post layer requests, stores layers and
+// returns an ordered list of processed layers with detected features and
+// namespaces.
+func processLayers(datastore database.Datastore, imageFormat string, requests []LayerRequest) ([]database.Layer, error) {
+	var (
+		reqMap = make(map[string]*processRequest)
+		err    error
+	)
 
-	if err := tx.PersistLayer(layer.hash, layer.namespaces, layer.features, layer.processedBy); err != nil {
-		return err
+	for _, r := range requests {
+		reqMap[r.Hash], err = getProcessRequest(datastore, r)
+		if err != nil {
+			return nil, err
+		}
 	}
 
-	return tx.Commit()
-}
-
-func persistFeatures(datastore database.Datastore, features []database.Feature) error {
-	tx, err := datastore.Begin()
+	results, err := processRequests(imageFormat, reqMap)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	defer tx.Rollback()
 
-	if err := tx.PersistFeatures(features); err != nil {
-		return err
-	}
-	return tx.Commit()
-}
-
-func persistNamespaces(datastore database.Datastore, namespaces []database.Namespace) error {
-	tx, err := datastore.Begin()
-	if err != nil {
-		return err
+	if err := persistProcessResult(datastore, results); err != nil {
+		return nil, err
 	}
-	defer tx.Rollback()
 
-	if err := tx.PersistNamespaces(namespaces); err != nil {
-		return err
+	completeLayers := getProcessResultLayers(results)
+	layers := make([]database.Layer, 0, len(requests))
+	for _, r := range requests {
+		layers = append(layers, completeLayers[r.Hash])
 	}
 
-	return tx.Commit()
+	return layers, nil
 }
 
-// combineLayers merges `layer` and `partial` without duplicated content.
-func combineLayers(layer database.Layer, partial partialLayer) database.Layer {
-	mapF := map[database.Feature]struct{}{}
-	mapNS := map[database.Namespace]struct{}{}
-	for _, f := range layer.Features {
-		mapF[f] = struct{}{}
-	}
-	for _, ns := range layer.Namespaces {
-		mapNS[ns] = struct{}{}
-	}
-	for _, f := range partial.features {
-		mapF[f] = struct{}{}
-	}
-	for _, ns := range partial.namespaces {
-		mapNS[ns] = struct{}{}
-	}
-	features := make([]database.Feature, 0, len(mapF))
-	namespaces := make([]database.Namespace, 0, len(mapNS))
-	for f := range mapF {
-		features = append(features, f)
-	}
-	for ns := range mapNS {
-		namespaces = append(namespaces, ns)
+func getProcessResultLayers(results map[string]*processResult) map[string]database.Layer {
+	layers := map[string]database.Layer{}
+	for name, r := range results {
+		layers[name] = *dbutil.MergeLayers(r.existingLayer, r.newLayerContent)
 	}
 
-	layer.ProcessedBy.Detectors = append(layer.ProcessedBy.Detectors, strutil.CompareStringLists(partial.processedBy.Detectors, layer.ProcessedBy.Detectors)...)
-	layer.ProcessedBy.Listers = append(layer.ProcessedBy.Listers, strutil.CompareStringLists(partial.processedBy.Listers, layer.ProcessedBy.Listers)...)
-	return database.Layer{
-		LayerMetadata: database.LayerMetadata{
-			Hash:        layer.Hash,
-			ProcessedBy: layer.ProcessedBy,
-		},
-		Features:   features,
-		Namespaces: namespaces,
-	}
+	return layers
 }
 
 func isAncestryProcessed(datastore database.Datastore, name string) (bool, error) {
-	tx, err := datastore.Begin()
-	if err != nil {
-		return false, err
-	}
-	defer tx.Rollback()
-	ancestry, ok, err := tx.FindAncestry(name)
-	if err != nil {
-		return false, err
-	}
-	if !ok {
-		return false, nil
+	ancestry, ok, err := dbutil.FindAncestry(datastore, name)
+	if err != nil || !ok {
+		return ok, err
 	}
 
-	notProcessed := getNotProcessedBy(ancestry.ProcessedBy)
-	return len(notProcessed.Detectors) == 0 && len(notProcessed.Listers) == 0, nil
+	return len(dbutil.DiffDetectors(EnabledDetectors, ancestry.By)) == 0, nil
 }
 
 // ProcessAncestry downloads and scans an ancestry if it's not scanned by all
 // enabled processors in this instance of Clair.
 func ProcessAncestry(datastore database.Datastore, imageFormat, name string, layerRequest []LayerRequest) error {
 	var (
-		err              error
-		ok               bool
-		layers           []database.Layer
-		commonProcessors database.Processors
+		err    error
+		ok     bool
+		layers []database.Layer
 	)
 
 	if name == "" {
@@ -360,10 +228,12 @@ func ProcessAncestry(datastore database.Datastore, imageFormat, name string, lay
 		return commonerr.NewBadRequestError("could not process a layer which does not have a format")
 	}
 
+	log.WithField("ancestry", name).Debug("start processing ancestry...")
 	if ok, err = isAncestryProcessed(datastore, name); err != nil {
+		log.WithError(err).Error("could not determine if ancestry is processed")
 		return err
 	} else if ok {
-		log.WithField("name", name).Debug("ancestry is already processed")
+		log.WithField("ancestry", name).Debug("ancestry is already processed")
 		return nil
 	}
 
@@ -371,155 +241,100 @@ func ProcessAncestry(datastore database.Datastore, imageFormat, name string, lay
 		return err
 	}
 
-	if commonProcessors, err = getProcessors(layers); err != nil {
-		return err
-	}
-
-	return processAncestry(datastore, name, layers, commonProcessors)
-}
-
-// getNamespacedFeatures extracts the namespaced features introduced in each
-// layer into one array.
-func getNamespacedFeatures(layers []database.AncestryLayer) []database.NamespacedFeature {
-	features := []database.NamespacedFeature{}
-	for _, layer := range layers {
-		features = append(features, layer.DetectedFeatures...)
-	}
-	return features
+	return processAncestry(datastore, name, layers)
 }
 
-func processAncestry(datastore database.Datastore, name string, layers []database.Layer, commonProcessors database.Processors) error {
+func processAncestry(datastore database.Datastore, name string, layers []database.Layer) error {
 	var (
-		ancestry database.Ancestry
+		ancestry = database.Ancestry{Name: name}
 		err      error
 	)
 
-	ancestry.Name = name
-	ancestry.ProcessedBy = commonProcessors
-	ancestry.Layers, err = computeAncestryLayers(layers, commonProcessors)
+	ancestry.Layers, ancestry.By, err = computeAncestryLayers(layers)
 	if err != nil {
 		return err
 	}
 
-	ancestryFeatures := getNamespacedFeatures(ancestry.Layers)
+	ancestryFeatures := dbutil.GetAncestryFeatures(ancestry)
 	log.WithFields(log.Fields{
-		"ancestry":           name,
-		"number of features": len(ancestryFeatures),
-		"processed by":       Processors,
-		"number of layers":   len(ancestry.Layers),
+		"ancestry":       name,
+		"processed by":   EnabledDetectors,
+		"features count": len(ancestryFeatures),
+		"layer count":    len(ancestry.Layers),
 	}).Debug("compute ancestry features")
 
-	if err := persistNamespacedFeatures(datastore, ancestryFeatures); err != nil {
+	if err := dbutil.PersistNamespacedFeatures(datastore, ancestryFeatures); err != nil {
+		log.WithField("ancestry", name).WithError(err).Error("could not persist namespaced features for ancestry")
 		return err
 	}
 
-	tx, err := datastore.Begin()
-	if err != nil {
+	if err := dbutil.CacheRelatedVulnerability(datastore, ancestryFeatures); err != nil {
+		log.WithField("ancestry", name).WithError(err).Error("failed to cache feature related vulnerability")
 		return err
 	}
 
-	err = tx.UpsertAncestry(ancestry)
-	if err != nil {
-		tx.Rollback()
+	if err := dbutil.UpsertAncestry(datastore, ancestry); err != nil {
+		log.WithField("ancestry", name).WithError(err).Error("could not upsert ancestry")
 		return err
 	}
 
-	err = tx.Commit()
-	if err != nil {
-		return err
-	}
 	return nil
 }
 
-func persistNamespacedFeatures(datastore database.Datastore, features []database.NamespacedFeature) error {
-	tx, err := datastore.Begin()
-	if err != nil {
-		return err
-	}
-
-	if err := tx.PersistNamespacedFeatures(features); err != nil {
-		tx.Rollback()
-		return err
-	}
-
-	if err := tx.Commit(); err != nil {
-		return err
-	}
-
-	tx, err = datastore.Begin()
-	if err != nil {
-		return err
+func getCommonDetectors(layers []database.Layer) mapset.Set {
+	// find the common detector for all layers and filter the namespaces and
+	// features based on that.
+	commonDetectors := mapset.NewSet()
+	for _, d := range layers[0].By {
+		commonDetectors.Add(d)
 	}
 
-	if err := tx.CacheAffectedNamespacedFeatures(features); err != nil {
-		tx.Rollback()
-		return err
-	}
-
-	return tx.Commit()
-}
-
-// getProcessors retrieves common subset of the processors of each layer.
-func getProcessors(layers []database.Layer) (database.Processors, error) {
-	if len(layers) == 0 {
-		return database.Processors{}, nil
-	}
-
-	detectors := layers[0].ProcessedBy.Detectors
-	listers := layers[0].ProcessedBy.Listers
-
-	detectorsLen := len(detectors)
-	listersLen := len(listers)
-
-	for _, l := range layers[1:] {
-		detectors := strutil.CompareStringListsInBoth(detectors, l.ProcessedBy.Detectors)
-		listers := strutil.CompareStringListsInBoth(listers, l.ProcessedBy.Listers)
-
-		if len(detectors) != detectorsLen || len(listers) != listersLen {
-			// This error might be triggered because of multiple workers are
-			// processing the same instance with different processors.
-			// TODO(sidchen): Once the features can be associated with
-			// Detectors/Listers, we can support dynamically generating ancestry's
-			// detector/lister based on the layers.
-			return database.Processors{}, errors.New("processing layers with different Clair instances is currently unsupported")
+	for _, l := range layers {
+		detectors := mapset.NewSet()
+		for _, d := range l.By {
+			detectors.Add(d)
 		}
+
+		commonDetectors = commonDetectors.Intersect(detectors)
 	}
-	return database.Processors{
-		Detectors: detectors,
-		Listers:   listers,
-	}, nil
-}
 
-type introducedFeature struct {
-	feature    database.NamespacedFeature
-	layerIndex int
+	return commonDetectors
 }
 
 // computeAncestryLayers computes ancestry's layers along with what features are
 // introduced.
-func computeAncestryLayers(layers []database.Layer, commonProcessors database.Processors) ([]database.AncestryLayer, error) {
-	// TODO(sidchen): Once the features are linked to specific processor, we
-	// will use commonProcessors to filter out the features for this ancestry.
+func computeAncestryLayers(layers []database.Layer) ([]database.AncestryLayer, []database.Detector, error) {
+	if len(layers) == 0 {
+		return nil, nil, nil
+	}
 
+	commonDetectors := getCommonDetectors(layers)
 	// version format -> namespace
-	namespaces := map[string]database.Namespace{}
+	namespaces := map[string]database.LayerNamespace{}
 	// version format -> feature ID -> feature
 	features := map[string]map[string]introducedFeature{}
 	ancestryLayers := []database.AncestryLayer{}
 	for index, layer := range layers {
-		// Initialize the ancestry Layer
-		initializedLayer := database.AncestryLayer{LayerMetadata: layer.LayerMetadata, DetectedFeatures: []database.NamespacedFeature{}}
+		initializedLayer := database.AncestryLayer{Hash: layer.Hash}
 		ancestryLayers = append(ancestryLayers, initializedLayer)
 
 		// Precondition: namespaces and features contain the result from union
 		// of all parents.
 		for _, ns := range layer.Namespaces {
+			if !commonDetectors.Contains(ns.By) {
+				continue
+			}
+
 			namespaces[ns.VersionFormat] = ns
 		}
 
 		// version format -> feature ID -> feature
 		currentFeatures := map[string]map[string]introducedFeature{}
 		for _, f := range layer.Features {
+			if !commonDetectors.Contains(f.By) {
+				continue
+			}
+
 			if ns, ok := namespaces[f.VersionFormat]; ok {
 				var currentMap map[string]introducedFeature
 				if currentMap, ok = currentFeatures[f.VersionFormat]; !ok {
@@ -537,16 +352,20 @@ func computeAncestryLayers(layers []database.Layer, commonProcessors database.Pr
 
 				if !inherited {
 					currentMap[f.Name+":"+f.Version] = introducedFeature{
-						feature: database.NamespacedFeature{
-							Feature:   f,
-							Namespace: ns,
+						feature: database.AncestryFeature{
+							NamespacedFeature: database.NamespacedFeature{
+								Feature:   f.Feature,
+								Namespace: ns.Namespace,
+							},
+							NamespaceBy: ns.By,
+							FeatureBy:   f.By,
 						},
 						layerIndex: index,
 					}
 				}
 
 			} else {
-				return nil, errors.New("No corresponding version format")
+				return nil, nil, errors.New("No corresponding version format")
 			}
 		}
 
@@ -564,57 +383,97 @@ func computeAncestryLayers(layers []database.Layer, commonProcessors database.Pr
 
 	for _, featureMap := range features {
 		for _, feature := range featureMap {
-			ancestryLayers[feature.layerIndex].DetectedFeatures = append(
-				ancestryLayers[feature.layerIndex].DetectedFeatures,
+			ancestryLayers[feature.layerIndex].Features = append(
+				ancestryLayers[feature.layerIndex].Features,
 				feature.feature,
 			)
 		}
 	}
 
-	return ancestryLayers, nil
+	detectors := make([]database.Detector, 0, commonDetectors.Cardinality())
+	for d := range commonDetectors.Iter() {
+		detectors = append(detectors, d.(database.Detector))
+	}
+
+	return ancestryLayers, detectors, nil
 }
 
-// getNotProcessedBy returns a processors, which contains the detectors and
-// listers not in `processedBy` but implemented in the current clair instance.
-func getNotProcessedBy(processedBy database.Processors) database.Processors {
-	notProcessedLister := strutil.CompareStringLists(Processors.Listers, processedBy.Listers)
-	notProcessedDetector := strutil.CompareStringLists(Processors.Detectors, processedBy.Detectors)
-	return database.Processors{
-		Listers:   notProcessedLister,
-		Detectors: notProcessedDetector,
+func extractRequiredFiles(imageFormat string, req *processRequest) (tarutil.FilesMap, error) {
+	requiredFiles := append(featurefmt.RequiredFilenames(req.detectors), featurens.RequiredFilenames(req.detectors)...)
+	if len(requiredFiles) == 0 {
+		log.WithFields(log.Fields{
+			"layer":     req.Hash,
+			"detectors": req.detectors,
+		}).Info("layer requires no file to extract")
+		return make(tarutil.FilesMap), nil
 	}
-}
 
-// detectContent downloads a layer and detects all features and namespaces.
-func detectContent(imageFormat, name, path string, headers map[string]string, toProcess database.Processors) (namespaces []database.Namespace, featureVersions []database.Feature, err error) {
-	log.WithFields(log.Fields{"Hash": name}).Debug("Process Layer")
-	totalRequiredFiles := append(featurefmt.RequiredFilenames(toProcess.Listers), featurens.RequiredFilenames(toProcess.Detectors)...)
-	files, err := imagefmt.Extract(imageFormat, path, headers, totalRequiredFiles)
+	files, err := imagefmt.Extract(imageFormat, req.Path, req.Headers, requiredFiles)
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
-			logLayerName: name,
-			"path":       cleanURL(path),
+			"layer": req.Hash,
+			"path":  strutil.CleanURL(req.Path),
 		}).Error("failed to extract data from path")
+		return nil, err
+	}
+
+	return files, err
+}
+
+// detectContent downloads a layer and detects all features and namespaces.
+func detectContent(imageFormat string, req *processRequest) (res *processResult) {
+	var (
+		files tarutil.FilesMap
+		layer = database.Layer{Hash: req.Hash, By: req.detectors}
+	)
+
+	res = &processResult{req.existingLayer, &layer, nil}
+	log.WithFields(log.Fields{
+		"layer":     req.Hash,
+		"detectors": req.detectors,
+	}).Info("detecting layer content...")
+
+	files, res.err = extractRequiredFiles(imageFormat, req)
+	if res.err != nil {
 		return
 	}
 
-	namespaces, err = featurens.Detect(files, toProcess.Detectors)
-	if err != nil {
+	if layer.Namespaces, res.err = featurens.Detect(files, req.detectors); res.err != nil {
 		return
 	}
 
-	if len(featureVersions) > 0 {
-		log.WithFields(log.Fields{logLayerName: name, "count": len(namespaces)}).Debug("detected layer namespaces")
+	if layer.Features, res.err = featurefmt.ListFeatures(files, req.detectors); res.err != nil {
+		return
 	}
 
-	featureVersions, err = featurefmt.ListFeatures(files, toProcess.Listers)
-	if err != nil {
+	log.WithFields(log.Fields{
+		"layer":           req.Hash,
+		"detectors":       req.detectors,
+		"namespace count": len(layer.Namespaces),
+		"feature count":   len(layer.Features),
+	}).Info("processed layer")
+
+	return
+}
+
+// InitWorker initializes the worker.
+func InitWorker(datastore database.Datastore) {
+	if len(EnabledDetectors) == 0 {
+		log.Warn("no enabled detector, and therefore, no ancestry will be processed.")
 		return
 	}
 
-	if len(featureVersions) > 0 {
-		log.WithFields(log.Fields{logLayerName: name, "count": len(featureVersions)}).Debug("detected layer features")
+	tx, err := datastore.Begin()
+	if err != nil {
+		log.WithError(err).Fatal("cannot connect to database to initialize worker")
 	}
 
-	return
+	defer tx.Rollback()
+	if err := tx.PersistDetectors(EnabledDetectors); err != nil {
+		log.WithError(err).Fatal("cannot insert detectors to initialize worker")
+	}
+
+	if err := tx.Commit(); err != nil {
+		log.WithError(err).Fatal("cannot commit detector changes to initialize worker")
+	}
 }

worker_test.go

@@ -22,12 +22,14 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/coreos/clair/database"
 	"github.com/coreos/clair/ext/featurefmt"
 	"github.com/coreos/clair/ext/featurens"
 	"github.com/coreos/clair/ext/versionfmt/dpkg"
-	"github.com/coreos/clair/pkg/strutil"
+	"github.com/coreos/clair/pkg/dbutil"
+	"github.com/coreos/clair/pkg/testutil"
 
 	// Register the required detectors.
 	_ "github.com/coreos/clair/ext/featurefmt/dpkg"
@@ -58,55 +60,27 @@ type mockSession struct {
 func copyDatastore(md *mockDatastore) mockDatastore {
 	layers := map[string]database.Layer{}
 	for k, l := range md.layers {
-		features := append([]database.Feature(nil), l.Features...)
-		namespaces := append([]database.Namespace(nil), l.Namespaces...)
-		listers := append([]string(nil), l.ProcessedBy.Listers...)
-		detectors := append([]string(nil), l.ProcessedBy.Detectors...)
 		layers[k] = database.Layer{
-			LayerMetadata: database.LayerMetadata{
-				Hash: l.Hash,
-				ProcessedBy: database.Processors{
-					Listers:   listers,
-					Detectors: detectors,
-				},
-			},
-			Features:   features,
-			Namespaces: namespaces,
+			Hash:       l.Hash,
+			By:         append([]database.Detector{}, l.By...),
+			Features:   append([]database.LayerFeature{}, l.Features...),
+			Namespaces: append([]database.LayerNamespace{}, l.Namespaces...),
 		}
 	}
 
 	ancestry := map[string]database.Ancestry{}
 	for k, a := range md.ancestry {
 		ancestryLayers := []database.AncestryLayer{}
-		layers := []database.LayerMetadata{}
-
 		for _, layer := range a.Layers {
-			layers = append(layers, database.LayerMetadata{
-				Hash: layer.Hash,
-				ProcessedBy: database.Processors{
-					Detectors: append([]string(nil), layer.LayerMetadata.ProcessedBy.Detectors...),
-					Listers:   append([]string(nil), layer.LayerMetadata.ProcessedBy.Listers...),
-				},
-			})
-
 			ancestryLayers = append(ancestryLayers, database.AncestryLayer{
-				LayerMetadata: database.LayerMetadata{
-					Hash: layer.Hash,
-					ProcessedBy: database.Processors{
-						Detectors: append([]string(nil), layer.LayerMetadata.ProcessedBy.Detectors...),
-						Listers:   append([]string(nil), layer.LayerMetadata.ProcessedBy.Listers...),
-					},
-				},
-				DetectedFeatures: append([]database.NamespacedFeature(nil), layer.DetectedFeatures...),
+				Hash:     layer.Hash,
+				Features: append([]database.AncestryFeature{}, layer.Features...),
 			})
 		}
 
 		ancestry[k] = database.Ancestry{
-			Name: a.Name,
-			ProcessedBy: database.Processors{
-				Detectors: append([]string(nil), a.ProcessedBy.Detectors...),
-				Listers:   append([]string(nil), a.ProcessedBy.Listers...),
-			},
+			Name:   a.Name,
+			By:     append([]database.Detector{}, a.By...),
 			Layers: ancestryLayers,
 		}
 	}
@@ -125,6 +99,7 @@ func copyDatastore(md *mockDatastore) mockDatastore {
 	for k, f := range md.namespacedFeatures {
 		namespacedFeatures[k] = f
 	}
+
 	return mockDatastore{
 		layers:             layers,
 		ancestry:           ancestry,
@@ -194,10 +169,7 @@ func newMockDatastore() *mockDatastore {
 				return errSessionDone
 			}
 			for _, n := range ns {
-				_, ok := session.copy.namespaces[n.Name]
-				if !ok {
-					session.copy.namespaces[n.Name] = n
-				}
+				session.copy.namespaces[NamespaceKey(&n)] = n
 			}
 			return nil
 		}
@@ -207,63 +179,36 @@ func newMockDatastore() *mockDatastore {
 				return errSessionDone
 			}
 			for _, f := range fs {
-				key := FeatureKey(&f)
-				_, ok := session.copy.features[key]
-				if !ok {
-					session.copy.features[key] = f
-				}
+				session.copy.features[FeatureKey(&f)] = f
 			}
+
 			return nil
 		}
 
-		session.FctPersistLayer = func(hash string, namespaces []database.Namespace, features []database.Feature, processedBy database.Processors) error {
+		session.FctPersistLayer = func(hash string, features []database.LayerFeature, namespaces []database.LayerNamespace, by []database.Detector) error {
 			if session.terminated {
 				return errSessionDone
 			}
 
-			// update the layer
-			_, ok := session.copy.layers[hash]
-			if !ok {
-				session.copy.layers[hash] = database.Layer{}
-			}
-
-			layer, ok := session.copy.layers[hash]
-			if !ok {
-				return errors.New("Failed to insert layer")
-			}
-
-			layerFeatures := map[string]database.Feature{}
-			layerNamespaces := map[string]database.Namespace{}
-			for _, f := range layer.Features {
-				layerFeatures[FeatureKey(&f)] = f
-			}
-			for _, n := range layer.Namespaces {
-				layerNamespaces[n.Name] = n
-			}
-
-			// ensure that all the namespaces, features are in the database
 			for _, ns := range namespaces {
-				if _, ok := session.copy.namespaces[ns.Name]; !ok {
-					return errors.New("Namespaces should be in the database")
-				}
-				if _, ok := layerNamespaces[ns.Name]; !ok {
-					layer.Namespaces = append(layer.Namespaces, ns)
-					layerNamespaces[ns.Name] = ns
+				if _, ok := session.copy.namespaces[NamespaceKey(&ns.Namespace)]; !ok {
+					panic("")
 				}
 			}
 
 			for _, f := range features {
-				if _, ok := session.copy.features[FeatureKey(&f)]; !ok {
-					return errors.New("Namespaces should be in the database")
-				}
-				if _, ok := layerFeatures[FeatureKey(&f)]; !ok {
-					layer.Features = append(layer.Features, f)
-					layerFeatures[FeatureKey(&f)] = f
+				if _, ok := session.copy.features[FeatureKey(&f.Feature)]; !ok {
+					panic("")
 				}
 			}
 
-			layer.ProcessedBy.Detectors = append(layer.ProcessedBy.Detectors, strutil.CompareStringLists(processedBy.Detectors, layer.ProcessedBy.Detectors)...)
-			layer.ProcessedBy.Listers = append(layer.ProcessedBy.Listers, strutil.CompareStringLists(processedBy.Listers, layer.ProcessedBy.Listers)...)
+			layer, _ := session.copy.layers[hash]
+			dbutil.MergeLayers(&layer, &database.Layer{
+				Hash:       hash,
+				By:         by,
+				Namespaces: namespaces,
+				Features:   features,
+			})
 
 			session.copy.layers[hash] = layer
 			return nil
@@ -274,11 +219,12 @@ func newMockDatastore() *mockDatastore {
 				return errSessionDone
 			}
 
-			features := getNamespacedFeatures(ancestry.Layers)
-			// ensure features are in the database
-			for _, f := range features {
-				if _, ok := session.copy.namespacedFeatures[NamespacedFeatureKey(&f)]; !ok {
-					return errors.New("namespaced feature not in db")
+			// ensure the namespaces features are in the code base
+			for _, l := range ancestry.Layers {
+				for _, f := range l.GetFeatures() {
+					if _, ok := session.copy.namespacedFeatures[NamespacedFeatureKey(&f)]; !ok {
+						panic("")
+					}
 				}
 			}
 
@@ -288,6 +234,14 @@ func newMockDatastore() *mockDatastore {
 
 		session.FctPersistNamespacedFeatures = func(namespacedFeatures []database.NamespacedFeature) error {
 			for i, f := range namespacedFeatures {
+				if _, ok := session.copy.features[FeatureKey(&f.Feature)]; !ok {
+					panic("")
+				}
+
+				if _, ok := session.copy.namespaces[NamespaceKey(&f.Namespace)]; !ok {
+					panic("")
+				}
+
 				session.copy.namespacedFeatures[NamespacedFeatureKey(&f)] = namespacedFeatures[i]
 			}
 			return nil
@@ -304,10 +258,7 @@ func newMockDatastore() *mockDatastore {
 }
 
 func TestMain(m *testing.M) {
-	Processors = database.Processors{
-		Listers:   featurefmt.ListListers(),
-		Detectors: featurens.ListDetectors(),
-	}
+	EnabledDetectors = append(featurefmt.ListListers(), featurens.ListDetectors()...)
 	m.Run()
 }
 
@@ -315,11 +266,16 @@ func FeatureKey(f *database.Feature) string {
 	return strings.Join([]string{f.Name, f.VersionFormat, f.Version}, "__")
 }
 
+func NamespaceKey(ns *database.Namespace) string {
+	return strings.Join([]string{ns.Name, ns.VersionFormat}, "__")
+}
+
 func NamespacedFeatureKey(f *database.NamespacedFeature) string {
 	return strings.Join([]string{f.Name, f.Namespace.Name}, "__")
 }
 
 func TestProcessAncestryWithDistUpgrade(t *testing.T) {
+	// TODO(sidac): Change to use table driven tests.
 	// Create the list of Features that should not been upgraded from one layer to another.
 	nonUpgradedFeatures := []database.Feature{
 		{Name: "libtext-wrapi18n-perl", Version: "0.06-7"},
@@ -358,7 +314,12 @@ func TestProcessAncestryWithDistUpgrade(t *testing.T) {
 	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock", layers))
 
 	// check the ancestry features
-	features := getNamespacedFeatures(datastore.ancestry["Mock"].Layers)
+	features := []database.AncestryFeature{}
+	for i, l := range datastore.ancestry["Mock"].Layers {
+		assert.Equal(t, layers[i].Hash, l.Hash)
+		features = append(features, l.Features...)
+	}
+
 	assert.Len(t, features, 74)
 	for _, f := range features {
 		if _, ok := nonUpgradedMap[f.Feature]; ok {
@@ -367,12 +328,6 @@ func TestProcessAncestryWithDistUpgrade(t *testing.T) {
 			assert.Equal(t, "debian:8", f.Namespace.Name)
 		}
 	}
-
-	assert.Equal(t, []database.LayerMetadata{
-		{Hash: "blank"},
-		{Hash: "wheezy"},
-		{Hash: "jessie"},
-	}, datastore.ancestry["Mock"].Layers)
 }
 
 func TestProcessLayers(t *testing.T) {
@@ -404,8 +359,7 @@ func TestProcessLayers(t *testing.T) {
 
 	// Ensure each layer has expected namespaces and features detected
 	if blank, ok := datastore.layers["blank"]; ok {
-		assert.Equal(t, blank.ProcessedBy.Detectors, Processors.Detectors)
-		assert.Equal(t, blank.ProcessedBy.Listers, Processors.Listers)
+		testutil.AssertDetectorsEqual(t, EnabledDetectors, blank.By)
 		assert.Len(t, blank.Namespaces, 0)
 		assert.Len(t, blank.Features, 0)
 	} else {
@@ -414,9 +368,11 @@ func TestProcessLayers(t *testing.T) {
 	}
 
 	if wheezy, ok := datastore.layers["wheezy"]; ok {
-		assert.Equal(t, wheezy.ProcessedBy.Detectors, Processors.Detectors)
-		assert.Equal(t, wheezy.ProcessedBy.Listers, Processors.Listers)
-		assert.Equal(t, wheezy.Namespaces, []database.Namespace{{Name: "debian:7", VersionFormat: dpkg.ParserName}})
+		testutil.AssertDetectorsEqual(t, EnabledDetectors, wheezy.By)
+		assert.Equal(t, []database.LayerNamespace{
+			{database.Namespace{"debian:7", dpkg.ParserName}, database.NewNamespaceDetector("os-release", "1.0")},
+		}, wheezy.Namespaces)
+
 		assert.Len(t, wheezy.Features, 52)
 	} else {
 		assert.Fail(t, "wheezy is not stored")
@@ -424,9 +380,10 @@ func TestProcessLayers(t *testing.T) {
 	}
 
 	if jessie, ok := datastore.layers["jessie"]; ok {
-		assert.Equal(t, jessie.ProcessedBy.Detectors, Processors.Detectors)
-		assert.Equal(t, jessie.ProcessedBy.Listers, Processors.Listers)
-		assert.Equal(t, jessie.Namespaces, []database.Namespace{{Name: "debian:8", VersionFormat: dpkg.ParserName}})
+		testutil.AssertDetectorsEqual(t, EnabledDetectors, jessie.By)
+		assert.Equal(t, []database.LayerNamespace{
+			{database.Namespace{"debian:8", dpkg.ParserName}, database.NewNamespaceDetector("os-release", "1.0")},
+		}, jessie.Namespaces)
 		assert.Len(t, jessie.Features, 74)
 	} else {
 		assert.Fail(t, "jessie is not stored")
@@ -434,157 +391,124 @@ func TestProcessLayers(t *testing.T) {
 	}
 }
 
-// TestUpgradeClair checks if a clair is upgraded and certain ancestry's
-// features should not change. We assume that Clair should only upgrade
-func TestClairUpgrade(t *testing.T) {
-	_, f, _, _ := runtime.Caller(0)
-	testDataPath := filepath.Join(filepath.Dir(f)) + "/testdata/DistUpgrade/"
-
-	datastore := newMockDatastore()
-
-	// suppose there are two ancestries.
-	layers := []LayerRequest{
-		{Hash: "blank", Path: testDataPath + "blank.tar.gz"},
-		{Hash: "wheezy", Path: testDataPath + "wheezy.tar.gz"},
-		{Hash: "jessie", Path: testDataPath + "jessie.tar.gz"},
+func getFeatures(a database.Ancestry) []database.AncestryFeature {
+	features := []database.AncestryFeature{}
+	for _, l := range a.Layers {
+		features = append(features, l.Features...)
 	}
 
-	layers2 := []LayerRequest{
-		{Hash: "blank", Path: testDataPath + "blank.tar.gz"},
-		{Hash: "wheezy", Path: testDataPath + "wheezy.tar.gz"},
-	}
-
-	// Suppose user scan an ancestry with an old instance of Clair.
-	Processors = database.Processors{
-		Detectors: []string{"os-release"},
-		Listers:   []string{"rpm"},
-	}
-
-	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock", layers))
-	assert.Len(t, getNamespacedFeatures(datastore.ancestry["Mock"].Layers), 0)
-
-	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock2", layers2))
-	assert.Len(t, getNamespacedFeatures(datastore.ancestry["Mock2"].Layers), 0)
-
-	// Clair is upgraded to use a new namespace detector. The expected
-	// behavior is that all layers will be rescanned with "apt-sources" and
-	// the ancestry's features are recalculated.
-	Processors = database.Processors{
-		Detectors: []string{"os-release", "apt-sources"},
-		Listers:   []string{"rpm"},
-	}
-
-	// Even though Clair processors are upgraded, the ancestry's features should
-	// not be upgraded without posting the ancestry to Clair again.
-	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock", layers))
-	assert.Len(t, getNamespacedFeatures(datastore.ancestry["Mock"].Layers), 0)
-
-	// Clair is upgraded to use a new feature lister. The expected behavior is
-	// that all layers will be rescanned with "dpkg" and the ancestry's features
-	// are invalidated and recalculated.
-	Processors = database.Processors{
-		Detectors: []string{"os-release", "apt-sources"},
-		Listers:   []string{"rpm", "dpkg"},
-	}
-
-	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock", layers))
-	assert.Len(t, getNamespacedFeatures(datastore.ancestry["Mock"].Layers), 74)
-	assert.Nil(t, ProcessAncestry(datastore, "Docker", "Mock2", layers2))
-	assert.Len(t, getNamespacedFeatures(datastore.ancestry["Mock2"].Layers), 52)
-
-	// check the namespaces are correct
-	for _, f := range getNamespacedFeatures(datastore.ancestry["Mock"].Layers) {
-		if !assert.NotEqual(t, database.Namespace{}, f.Namespace) {
-			assert.Fail(t, "Every feature should have a namespace attached")
-		}
-	}
-
-	for _, f := range getNamespacedFeatures(datastore.ancestry["Mock2"].Layers) {
-		if !assert.NotEqual(t, database.Namespace{}, f.Namespace) {
-			assert.Fail(t, "Every feature should have a namespace attached")
-		}
-	}
+	return features
 }
 
-// TestMultipleNamespaces tests computing ancestry features
 func TestComputeAncestryFeatures(t *testing.T) {
 	vf1 := "format 1"
 	vf2 := "format 2"
 
-	ns1a := database.Namespace{
-		Name:          "namespace 1:a",
-		VersionFormat: vf1,
-	}
-
-	ns1b := database.Namespace{
-		Name:          "namespace 1:b",
-		VersionFormat: vf1,
-	}
-
-	ns2a := database.Namespace{
-		Name:          "namespace 2:a",
-		VersionFormat: vf2,
-	}
-
-	ns2b := database.Namespace{
-		Name:          "namespace 2:b",
-		VersionFormat: vf2,
-	}
-
-	f1 := database.Feature{
-		Name:          "feature 1",
-		Version:       "0.1",
-		VersionFormat: vf1,
-	}
-
-	f2 := database.Feature{
+	nd1 := database.NewNamespaceDetector("apk", "1.0")
+	fd1 := database.NewFeatureDetector("fd1", "1.0")
+	// this detector only scans one layer with one extra feature, this one
+	// should be omitted.
+	fd2 := database.NewFeatureDetector("fd2", "1.0")
+
+	ns1a := database.LayerNamespace{
+		database.Namespace{
+			Name:          "namespace 1:a",
+			VersionFormat: vf1,
+		}, nd1,
+	}
+
+	ns1b := database.LayerNamespace{
+		database.Namespace{
+			Name:          "namespace 1:b",
+			VersionFormat: vf1,
+		}, nd1}
+
+	ns2a := database.LayerNamespace{
+		database.Namespace{
+			Name:          "namespace 2:a",
+			VersionFormat: vf2,
+		}, nd1}
+
+	ns2b := database.LayerNamespace{
+		database.Namespace{
+			Name:          "namespace 2:b",
+			VersionFormat: vf2,
+		}, nd1}
+
+	f1 := database.LayerFeature{
+		database.Feature{
+			Name:          "feature 1",
+			Version:       "0.1",
+			VersionFormat: vf1,
+		}, fd1}
+
+	f2 := database.LayerFeature{database.Feature{
 		Name:          "feature 2",
 		Version:       "0.2",
 		VersionFormat: vf1,
-	}
-
-	f3 := database.Feature{
-		Name:          "feature 1",
-		Version:       "0.3",
-		VersionFormat: vf2,
-	}
-
-	f4 := database.Feature{
-		Name:          "feature 2",
-		Version:       "0.3",
-		VersionFormat: vf2,
+	}, fd2}
+
+	f3 := database.LayerFeature{
+		database.Feature{
+			Name:          "feature 1",
+			Version:       "0.3",
+			VersionFormat: vf2,
+		}, fd1}
+
+	f4 := database.LayerFeature{
+		database.Feature{
+			Name:          "feature 2",
+			Version:       "0.3",
+			VersionFormat: vf2,
+		}, fd1}
+
+	f5 := database.LayerFeature{
+		database.Feature{
+			Name:          "feature 3",
+			Version:       "0.3",
+			VersionFormat: vf2,
+		},
+		fd2,
 	}
 
 	// Suppose Clair is watching two files for namespaces one containing ns1
 	// changes e.g. os-release and the other one containing ns2 changes e.g.
 	// node.
-	blank := database.Layer{LayerMetadata: database.LayerMetadata{Hash: "blank"}}
+	blank := database.Layer{
+		Hash: "blank",
+		By:   []database.Detector{nd1, fd1, fd1},
+	}
 	initNS1a := database.Layer{
-		LayerMetadata: database.LayerMetadata{Hash: "init ns1a"},
-		Namespaces:    []database.Namespace{ns1a},
-		Features:      []database.Feature{f1, f2},
+		Hash:       "initNS1a",
+		By:         []database.Detector{nd1, fd1, fd1},
+		Namespaces: []database.LayerNamespace{ns1a},
+		Features:   []database.LayerFeature{f1, f2},
 	}
 
 	upgradeNS2b := database.Layer{
-		LayerMetadata: database.LayerMetadata{Hash: "upgrade ns2b"},
-		Namespaces:    []database.Namespace{ns2b},
+		Hash:       "upgradeNS2b",
+		By:         []database.Detector{nd1, fd1, fd1},
+		Namespaces: []database.LayerNamespace{ns2b},
 	}
 
 	upgradeNS1b := database.Layer{
-		LayerMetadata: database.LayerMetadata{Hash: "upgrade ns1b"},
-		Namespaces:    []database.Namespace{ns1b},
-		Features:      []database.Feature{f1, f2},
+		Hash:       "upgradeNS1b",
+		By:         []database.Detector{nd1, fd1, fd1, fd2},
+		Namespaces: []database.LayerNamespace{ns1b},
+		Features:   []database.LayerFeature{f1, f2, f5},
 	}
 
 	initNS2a := database.Layer{
-		LayerMetadata: database.LayerMetadata{Hash: "init ns2a"},
-		Namespaces:    []database.Namespace{ns2a},
-		Features:      []database.Feature{f3, f4},
+		Hash:       "initNS2a",
+		By:         []database.Detector{nd1, fd1, fd1},
+		Namespaces: []database.LayerNamespace{ns2a},
+		Features:   []database.LayerFeature{f3, f4},
 	}
 
 	removeF2 := database.Layer{
-		LayerMetadata: database.LayerMetadata{Hash: "remove f2"},
-		Features:      []database.Feature{f1},
+		Hash:     "removeF2",
+		By:       []database.Detector{nd1, fd1, fd1},
+		Features: []database.LayerFeature{f1},
 	}
 
 	// blank -> ns1:a, f1 f2 (init)
@@ -597,44 +521,65 @@ func TestComputeAncestryFeatures(t *testing.T) {
 	// -> blank (empty)
 
 	layers := []database.Layer{
-		blank,
-		initNS1a,
-		removeF2,
-		initNS2a,
-		upgradeNS2b,
-		blank,
-		upgradeNS1b,
-		removeF2,
+		blank,       // empty
+		initNS1a,    // namespace: NS1a, features: f1, f2
+		removeF2,    // namespace:     , features: f1
+		initNS2a,    // namespace: NS2a, features: f3, f4 ( under NS2a )
+		upgradeNS2b, // namespace: NS2b, ( f3, f4 are now under NS2b )
+		blank,       // empty
+		upgradeNS1b, // namespace: NS1b, ( f1, f2 are now under NS1b, and they are introduced in this layer. )
+		removeF2,    // namespace:     , features: f1
 		blank,
 	}
 
-	expected := map[database.NamespacedFeature]bool{
+	expected := []database.AncestryLayer{
+		{
+			"blank",
+			[]database.AncestryFeature{},
+		},
 		{
-			Feature:   f1,
-			Namespace: ns1a,
-		}: false,
+			"initNS1a",
+			[]database.AncestryFeature{{database.NamespacedFeature{f1.Feature, ns1a.Namespace}, f1.By, ns1a.By}},
+		},
 		{
-			Feature:   f3,
-			Namespace: ns2a,
-		}: false,
+			"removeF2",
+			[]database.AncestryFeature{},
+		},
 		{
-			Feature:   f4,
-			Namespace: ns2a,
-		}: false,
+			"initNS2a",
+			[]database.AncestryFeature{
+				{database.NamespacedFeature{f3.Feature, ns2a.Namespace}, f3.By, ns2a.By},
+				{database.NamespacedFeature{f4.Feature, ns2a.Namespace}, f4.By, ns2a.By},
+			},
+		},
+		{
+			"upgradeNS2b",
+			[]database.AncestryFeature{},
+		},
+		{
+			"blank",
+			[]database.AncestryFeature{},
+		},
+		{
+			"upgradeNS1b",
+			[]database.AncestryFeature{},
+		},
+		{
+			"removeF2",
+			[]database.AncestryFeature{},
+		},
+		{
+			"blank",
+			[]database.AncestryFeature{},
+		},
 	}
 
-	ancestryLayers, err := computeAncestryLayers(layers, database.Processors{})
-	assert.Nil(t, err)
-	features := getNamespacedFeatures(ancestryLayers)
-	for _, f := range features {
-		if assert.Contains(t, expected, f) {
-			if assert.False(t, expected[f]) {
-				expected[f] = true
-			}
-		}
-	}
+	expectedDetectors := []database.Detector{nd1, fd1}
+	ancestryLayers, detectors, err := computeAncestryLayers(layers)
+	require.Nil(t, err)
 
-	for f, visited := range expected {
-		assert.True(t, visited, "expected feature is missing : "+f.Namespace.Name+":"+f.Name)
+	testutil.AssertDetectorsEqual(t, expectedDetectors, detectors)
+	for i := range expected {
+		testutil.AssertAncestryLayerEqual(t, &expected[i], &ancestryLayers[i])
 	}
 }