Merge pull request #12609 from steven-zou/fix/p2p_preheat_event

fix(preheat):fix issues of event-based preheat

src/controller/event/handler/p2p/preheat.go

@@ -16,6 +16,9 @@ package p2p
 
 import (
 	"context"
+
+	"github.com/goharbor/harbor/src/controller/tag"
+
 	"github.com/goharbor/harbor/src/controller/artifact"
 	"github.com/goharbor/harbor/src/controller/artifact/processor/image"
 	"github.com/goharbor/harbor/src/controller/event"
@@ -55,7 +58,13 @@ func (p *Handler) handlePushArtifact(event *event.PushArtifactEvent) error {
 	if event.Artifact.Type != image.ArtifactTypeImage {
 		return nil
 	}
-	log.Debugf("preheat artifact event %s:%s", event.Artifact.RepositoryName, event.Artifact.Digest)
+
+	// NOTES: So far, we only support artifact with tags
+	if len(event.Tags) == 0 {
+		return nil
+	}
+
+	log.Debugf("preheat: artifact pushed %s:%s@%s", event.Artifact.RepositoryName, event.Tags, event.Artifact.Digest)
 
 	art, err := artifact.Ctl.Get(p.Context(), event.Artifact.ID, &artifact.Option{
 		WithTag:   true,
@@ -64,12 +73,26 @@ func (p *Handler) handlePushArtifact(event *event.PushArtifactEvent) error {
 	if err != nil {
 		return err
 	}
+
+	// Only with the pushed tags, ignore other tags
+	pt := make([]*tag.Tag, 0)
+	for _, tg := range art.Tags {
+		if tg.Name == event.Tags[0] {
+			pt = append(pt, tg)
+			break
+		}
+	}
+	art.Tags = pt
+
 	_, err = preheat.Enf.PreheatArtifact(p.Context(), art)
 	return err
 }
 
 func (p *Handler) handleImageScanned(event *event.ScanImageEvent) error {
-	log.Debugf("preheat image scanned %s:%s", event.Artifact.Repository, event.Artifact.Tag)
+	// TODO: If the scan is targeting an manifest list, here the artifacts we get are all the children
+	//  artifacts of the manifest list. The children artifacts are high probably untagged ones that
+	//  will be definitely ignored by the tag filter. We need to find a way to resolve this issue.
+	log.Debugf("preheat: image scanned %s:%s", event.Artifact.Repository, event.Artifact.Tag)
 	art, err := artifact.Ctl.GetByReference(p.Context(), event.Artifact.Repository, event.Artifact.Digest,
 		&artifact.Option{
 			WithTag:   true,
@@ -87,10 +110,16 @@ func (p *Handler) handleArtifactLabeled(event *event.ArtifactLabeledEvent) error
 		WithTag:   true,
 		WithLabel: true,
 	})
+
+	if err != nil {
+		return err
+	}
+
+	// Only care image at this moment
 	if art.Type != image.ArtifactTypeImage {
 		return nil
 	}
-	log.Debugf("preheat artifact labeled %s:%s", art.Artifact.RepositoryName, art.Artifact.Digest)
+	log.Debugf("preheat: artifact labeled %s:%s", art.Artifact.RepositoryName, art.Artifact.Digest)
 
 	_, err = preheat.Enf.PreheatArtifact(p.Context(), art)
 	return err

src/controller/p2p/preheat/enforcer.go

@@ -103,6 +103,12 @@ type extURLGetter func(c *selector.Candidate) (string, error)
 // The purpose of defining such a func template is decoupling code
 type accessCredMaker func(c *selector.Candidate) (string, error)
 
+// matchedPolicy is a temporary intermediary struct for passing parameters
+type matchedPolicy struct {
+	policy   *pol.Schema
+	filtered []*selector.Candidate
+}
+
 var (
 	// Enf default enforcer
 	Enf = NewEnforcer()
@@ -254,8 +260,7 @@ func (de *defaultEnforcer) PreheatArtifact(ctx context.Context, art *artifact.Ar
 		return nil, enforceErrorExt(err, art)
 	}
 
-	matched := 0
+	matched := make([]*matchedPolicy, 0)
-	ids := make([]int64, 0)
 	for _, pl := range l {
 		// Skip disabled policies
 		if !pl.Enabled {
@@ -269,19 +274,6 @@ func (de *defaultEnforcer) PreheatArtifact(ctx context.Context, art *artifact.Ar
 			continue
 		}
 
-		// Get and check if the provider instance bound with the policy is healthy
-		inst, err := de.instMgr.Get(ctx, pl.ProviderID)
-		if err != nil {
-			log.Errorf("Failed to get the preheat provider instance bound with the policy %d:%s with error: %s", pl.ID, pl.Name, err.Error())
-			continue
-		}
-
-		// Skip unhealthy instance
-		if err := checkProviderHealthy(inst); err != nil {
-			log.Errorf("The preheat provider instance bound with the policy %d:%s is not healthy: %s", pl.ID, pl.Name, err.Error())
-			continue
-		}
-
 		// Override security settings if necessary
 		ov := overrideSecuritySettings(pl, p)
 		for _, ss := range ov {
@@ -295,25 +287,44 @@ func (de *defaultEnforcer) PreheatArtifact(ctx context.Context, art *artifact.Ar
 			continue
 		}
 
-		matched++
+		// The artifact candidate is matched with the policy
-
 		if len(filtered) > 0 {
-			// Matched
+			matched = append(matched, &matchedPolicy{pl, filtered})
-			eid, err := de.launchExecutions(ctx, filtered, pl, inst)
-			if err != nil {
-				// Log error and continue
-				log.Errorf("Failed to launch execution for policy %d:%s with error: %s", pl.ID, pl.Name, err.Error())
-			} else {
-				// Success and then append the execution id to list
-				ids = append(ids, eid)
-			}
 		}
 	}
 
-	if matched != len(ids) {
+	ids := make([]int64, 0)
+	// Launch preheat executions for all the matched policies.
+	// Check the health of the instance bound with the policy at this moment.
+	for _, mp := range matched {
+		// Get and check if the provider instance bound with the policy is healthy
+		inst, err := de.instMgr.Get(ctx, mp.policy.ProviderID)
+		if err != nil {
+			log.Errorf("Failed to get the preheat provider instance bound with the policy %d:%s with error: %s", mp.policy.ID, mp.policy.Name, err.Error())
+			continue
+		}
+
+		// Skip unhealthy instance
+		if err := checkProviderHealthy(inst); err != nil {
+			log.Errorf("The preheat provider instance bound with the policy %d:%s is not healthy: %s", mp.policy.ID, mp.policy.Name, err.Error())
+			continue
+		}
+
+		// Launch executions now
+		eid, err := de.launchExecutions(ctx, mp.filtered, mp.policy, inst)
+		if err != nil {
+			// Log error and continue
+			log.Errorf("Failed to launch execution for policy %d:%s with error: %s", mp.policy.ID, mp.policy.Name, err.Error())
+		} else {
+			// Success and then append the execution id to list
+			ids = append(ids, eid)
+		}
+	}
+
+	if len(matched) != len(ids) {
 		// Some policy enforcement are failed
 		// Treat it as an error case
-		return ids, enforceErrorExt(errors.Errorf("%d policies matched but only %d successfully enforced", matched, len(ids)), art)
+		return ids, enforceErrorExt(errors.Errorf("%d policies matched but only %d successfully enforced", len(matched), len(ids)), art)
 	}
 
 	return ids, nil
@@ -495,6 +506,7 @@ func (de *defaultEnforcer) toCandidates(ctx context.Context, p *models.Project,
 		}
 
 		// If artifact has more than one tag, then split them into separate candidate for easy filtering.
+		// TODO: Do we need to support untagged artifacts here?
 		for _, t := range a.Tags {
 			candidates = append(candidates, &selector.Candidate{
 				NamespaceID: p.ProjectID,