[Triage Metadata] Clean orphaned pending metadata in searchcache polling (#2493)

* Add cleanOrphanedPendingMetadata to searchcache polling

* Add logging and use logger.Infof in poll.go

* Address comments

Co-authored-by: Kyle Ju <[email protected]>

Author: KyleJu

api/query/cache/poll/poll.go

@@ -5,9 +5,12 @@
 package poll
 
 import (
+	"context"
 	"net/http"
+	"strconv"
 	"time"
 
+	"github.com/google/go-github/v33/github"
 	"github.com/web-platform-tests/wpt.fyi/api/query"
 	"github.com/web-platform-tests/wpt.fyi/api/query/cache/index"
 	"github.com/web-platform-tests/wpt.fyi/shared"
@@ -84,20 +87,76 @@ func wait(start time.Time, total time.Duration) {
 	}
 }
 
-// KeepMetadataUpdated implements updates to metadataMapCached via simple polling every
+// StartMetadataPollingService performs metadata-related services via simple polling every
 // interval duration.
-func KeepMetadataUpdated(client *http.Client, logger shared.Logger, interval time.Duration) {
-	logger.Infof("Metadata update via polling started")
+func StartMetadataPollingService(ctx context.Context, logger shared.Logger, interval time.Duration) {
+	logger.Infof("Starting Metadata polling service.")
+	netClient := &http.Client{Timeout: time.Second * 5}
+	cacheSet := shared.NewRedisSet()
+	gitHubClient, err := shared.NewAppEngineAPI(ctx).GetGitHubClient()
+	if err != nil {
+		logger.Infof("Unable to get GitHub client: %v", err)
+	}
+
 	for {
-		metadataCache, err := shared.GetWPTMetadataArchive(client, nil)
+		keepMetadataUpdated(netClient, logger)
+		if gitHubClient != nil {
+			cleanOrphanedPendingMetadata(ctx, gitHubClient, cacheSet, logger)
+		} else {
+			logger.Infof("GitHub client is not initialized, skipping cleanOrphanedPendingMetadata.")
+		}
+		time.Sleep(interval)
+	}
+}
+
+// keepMetadataUpdated fetches a new copy of the wpt-metadata repo and updates metadataMapCached.
+func keepMetadataUpdated(client *http.Client, logger shared.Logger) {
+	logger.Infof("Running keepMetadataUpdated...")
+	metadataCache, err := shared.GetWPTMetadataArchive(client, nil)
+	if err != nil {
+		logger.Infof("Error fetching Metadata for update: %v", err)
+		return
+	}
+
+	if metadataCache != nil {
+		query.MetadataMapCached = metadataCache
+	}
+}
+
+// cleanOrphanedPendingMetadata cleans and removes orphaned pending metadata in Redis.
+func cleanOrphanedPendingMetadata(ctx context.Context, ghClient *github.Client, cacheSet shared.RedisSet, logger shared.Logger) {
+	logger.Infof("Running cleanOrphanedPendingMetadata...")
+	prs, err := cacheSet.GetAll(shared.PendingMetadataCacheKey)
+	if err != nil {
+		logger.Infof("Error fetching pending PRs from cacheSet: %v", err)
+		return
+	}
+	logger.Infof("Pending PR numbers in cacheSet are: %v", prs)
+
+	for _, pr := range prs {
+		// Parse PR string into integer
+		prInt, err := strconv.Atoi(pr)
 		if err != nil {
-			logger.Errorf("Error fetching Metadata for update: %v", err)
+			logger.Infof("Error parsing %s into integer in cleanOrphanedPendingMetadata", pr)
+			// Not an integer; remove it from the cache set.
+			cacheSet.Remove(shared.PendingMetadataCacheKey, pr)
+			shared.DeleteCache(shared.PendingMetadataCachePrefix + pr)
+			continue
 		}
 
-		if err == nil && metadataCache != nil {
-			query.MetadataMapCached = metadataCache
+		res, _, err := ghClient.PullRequests.Get(ctx, shared.SourceOwner, shared.SourceRepo, prInt)
+		if err != nil {
+			logger.Infof("Error getting information for PR %s: %v", pr, err)
+			continue
 		}
 
-		time.Sleep(interval)
+		if res.State == nil || *res.State != "closed" {
+			continue
+		}
+
+		logger.Infof("Removing PR %s and its pending metadata from Redis", pr)
+		// pr is closed; remove pr from the cache set and its pending metadata from Redis.
+		cacheSet.Remove(shared.PendingMetadataCacheKey, pr)
+		shared.DeleteCache(shared.PendingMetadataCachePrefix + pr)
 	}
 }

api/query/cache/service/app.staging.yaml

@@ -17,3 +17,10 @@ liveness_check:
 
 readiness_check:
   path: "/_ah/readiness_check"
+
+network:
+  name: default
+
+env_variables:
+  REDISHOST: "10.171.142.203"
+  REDISPORT: "6379"

api/query/cache/service/app.yaml

@@ -21,3 +21,10 @@ liveness_check:
 
 readiness_check:
   path: "/_ah/readiness_check"
+
+network:
+  name: default
+
+env_variables:
+  REDISHOST: "10.171.142.203"
+  REDISPORT: "6379"

api/query/cache/service/main.go

@@ -141,18 +141,14 @@ func main() {
 	// after backfilling was started.
 	go poll.KeepRunsUpdated(store, logger, *updateInterval, *updateMaxRuns, idx)
 
-	var netClient = &http.Client{
-		Timeout: time.Second * 5,
-	}
-
 	// Initializes clients.
 	if err = shared.Clients.Init(context.Background()); err != nil {
 		logrus.Fatalf("Failed to initialize Google Cloud clients: %v", err)
 	}
 	defer shared.Clients.Close()
 
 	// Polls Metadata update every 10 minutes.
-	go poll.KeepMetadataUpdated(netClient, logger, time.Minute*10)
+	go poll.StartMetadataPollingService(context.Background(), logger, time.Minute*10)
 
 	http.HandleFunc("/_ah/liveness_check", livenessCheckHandler)
 	http.HandleFunc("/_ah/readiness_check", readinessCheckHandler)

shared/cache.go

@@ -434,6 +434,19 @@ func FlushCache() error {
 	return err
 }
 
+// DeleteCache deletes the object stored at key in Redis.
+// A key is ignored if it does not exist.
+func DeleteCache(key string) error {
+	if Clients.redisPool == nil {
+		return errNoRedis
+	}
+	conn := Clients.redisPool.Get()
+	defer conn.Close()
+	// https://redis.io/commands/del
+	_, err := conn.Do("DEL", key)
+	return err
+}
+
 // RedisSet is an interface for an redisSetReadWritable,
 // which performs Add/Remove/GetAll operations via the App Engine Redis API.
 type RedisSet interface {

shared/metadata_util.go

@@ -27,8 +27,11 @@ const PendingMetadataCacheKey = "WPT-PENDING-METADATA"
 // stored in Redis.
 const PendingMetadataCachePrefix = "PENDING-PR-"
 
-const sourceOwner string = "web-platform-tests"
-const sourceRepo string = "wpt-metadata"
+// SourceOwner is the owner name of the wpt-metadata repo.
+const SourceOwner string = "web-platform-tests"
+
+// SourceRepo is the wpt-metadata repo.
+const SourceRepo string = "wpt-metadata"
 const baseBranch string = "master"
 
 // MetadataFetcher is an abstract interface that encapsulates the Fetch() method. Fetch() fetches metadata
@@ -39,7 +42,7 @@ type MetadataFetcher interface {
 
 // GetWPTMetadataMasterSHA returns the SHA of the master branch of the wpt-metadata repo.
 func GetWPTMetadataMasterSHA(ctx context.Context, gitHubClient *github.Client) (*string, error) {
-	baseRef, _, err := gitHubClient.Git.GetRef(ctx, sourceOwner, sourceRepo, "refs/heads/"+baseBranch)
+	baseRef, _, err := gitHubClient.Git.GetRef(ctx, SourceOwner, SourceRepo, "refs/heads/"+baseBranch)
 	if err != nil {
 		return nil, err
 	}

shared/triage_metadata.go

@@ -68,16 +68,16 @@ func getNewCommitBranchName(ctx context.Context, client *github.Client, sourceOw
 }
 
 func getWptmetadataGitHubInfo(ctx context.Context, client *github.Client) wptmetadataGitHubInfo {
-	commitBranch := getNewCommitBranchName(ctx, client, sourceOwner, sourceRepo)
+	commitBranch := getNewCommitBranchName(ctx, client, SourceOwner, SourceRepo)
 
 	return wptmetadataGitHubInfo{
-		sourceOwner:   sourceOwner,
-		sourceRepo:    sourceRepo,
+		sourceOwner:   SourceOwner,
+		sourceRepo:    SourceRepo,
 		commitMessage: "Commit New Metadata",
 		commitBranch:  commitBranch,
 		baseBranch:    baseBranch,
-		prRepoOwner:   sourceOwner,
-		prRepo:        sourceRepo,
+		prRepoOwner:   SourceOwner,
+		prRepo:        SourceRepo,
 		prBranch:      baseBranch,
 		prSubject:     "Automatically Triage New Metadata",
 		prDescription: "This metadata PR was generated via the wpt.fyi `/api/metadata/triage` endpoint. See [the documentation](https://github.com/web-platform-tests/wpt.fyi/tree/master/api#apimetadatatriage) for more information about how to use this service."}