Merge pull request #617 from puerco/watch-jobs

Implement job-level observers

Author: k8s-ci-robot

go.mod

@@ -1,13 +1,14 @@
 module sigs.k8s.io/tejolote
 
-go 1.26.1
+go 1.26.2
 
 require (
 	cloud.google.com/go/pubsub/v2 v2.5.1
 	cloud.google.com/go/storage v1.62.0
 	github.com/carabiner-dev/vcslocator v0.4.2
 	github.com/go-git/go-git/v5 v5.17.2
 	github.com/google/go-containerregistry v0.21.5
+	github.com/google/go-github/v84 v84.0.0
 	github.com/in-toto/attestation v1.2.0
 	github.com/magefile/mage v1.17.1
 	github.com/protobom/protobom v0.5.4
@@ -158,7 +159,6 @@ require (
 	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/go-github/v73 v73.0.0 // indirect
-	github.com/google/go-github/v84 v84.0.0 // indirect
 	github.com/google/go-querystring v1.2.0 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/google/uuid v1.6.0 // indirect

internal/cmd/attest.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os"
 	"slices"
+	"strings"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -39,8 +40,16 @@ type attestOptions struct {
 	encodedSnapshots string
 	slsaVersion      string
 	artifacts        []string
+	watchJobs        []string
 }
 
+const (
+	githubEnvVarJob     = "GITHUB_JOB"
+	githubEnvVarActions = "GITHUB_ACTIONS"
+	githubEnvVarRepo    = "GITHUB_REPOSITORY"
+	githubEnvVarRunID   = "GITHUB_RUN_ID"
+)
+
 var slsaVersions = []string{"1", "1.0", "0.2"}
 
 func (o *attestOptions) Verify() error {
@@ -136,6 +145,22 @@ build data and generates the provenance attestation.
 
 			w.Options.WaitForBuild = attestOpts.waitForBuild
 			w.Options.SLSAVersion = attestOpts.slsaVersion
+			w.Options.WatchJobs = attestOpts.watchJobs
+
+			// Auto detect if tejolote is running inside a GitHub Actions
+			// workflow and the spec URL points to the same run. The we automatically
+			// enable job-level watching to avoid deadlock where the run never
+			// completes because the attester job is part of it.
+			if isSameActionsRun(args[0]) {
+				// Get our own job name:
+				currentJob := os.Getenv(githubEnvVarJob)
+				if len(w.Options.WatchJobs) == 0 {
+					logrus.Infof("Same-run detected (job: %q), will watch all other jobs", currentJob)
+				} else {
+					logrus.Infof("Same-run detected (job: %q), watching specified jobs: %v", currentJob, w.Options.WatchJobs)
+				}
+				w.Options.ExcludeJob = currentJob
+			}
 
 			if !attestOpts.waitForBuild {
 				logrus.Warn("watcher will not wait for build, data may be incomplete")
@@ -154,7 +179,7 @@ build data and generates the provenance attestation.
 				return fmt.Errorf("fetching run: %w", err)
 			}
 
-			// Watch the run run :)
+			// Watch the run, run :)
 			if err := w.Watch(r); err != nil {
 				return fmt.Errorf("waiting for the run to finish: %w", err)
 			}
@@ -304,5 +329,45 @@ build data and generates the provenance attestation.
 	_ = attestCmd.PersistentFlags().MarkHidden("encoded-attestation")
 	_ = attestCmd.PersistentFlags().MarkHidden("encoded-snapshots")
 
+	attestCmd.PersistentFlags().StringSliceVar(
+		&attestOpts.watchJobs,
+		"watch-jobs",
+		[]string{},
+		"watch specific jobs (by name) instead of the entire run",
+	)
+
 	parentCmd.AddCommand(attestCmd)
 }
+
+// isSameActionsRun checks if tejolote is running inside the same GitHub Actions
+// workflow run that it is being asked to observe.
+//
+// It compares the spec URL against the GITHUB_REPOSITORY and GITHUB_RUN_ID
+// environment variables set by the runner.
+func isSameActionsRun(specURL string) bool {
+	if os.Getenv(githubEnvVarActions) != "true" {
+		return false
+	}
+
+	ghRepo := os.Getenv(githubEnvVarRepo)
+	ghRunID := os.Getenv(githubEnvVarRunID)
+	if ghRepo == "" || ghRunID == "" {
+		return false
+	}
+
+	// Parse org/repo and run ID from the spec URL (e.g. github://org/repo/12345)
+	parts := strings.SplitN(specURL, "://", 2)
+	if len(parts) != 2 || parts[0] != "github" {
+		return false
+	}
+
+	pathParts := strings.SplitN(parts[1], "/", 3)
+	if len(pathParts) != 3 {
+		return false
+	}
+
+	specRepo := pathParts[0] + "/" + pathParts[1]
+	specRunID := strings.TrimSuffix(pathParts[2], "/")
+
+	return specRepo == ghRepo && specRunID == ghRunID
+}

pkg/builder/builder.go

@@ -54,6 +54,12 @@ func (b *Builder) Snap() error {
 	return nil
 }
 
+// Driver returns the underlying build system driver. This allows callers
+// to type-assert for optional interfaces like driver.JobWatcher.
+func (b *Builder) Driver() driver.BuildSystem {
+	return b.driver
+}
+
 func (b *Builder) GetRun(identifier string) (*run.Run, error) {
 	return b.driver.GetRun(identifier)
 }

pkg/builder/driver/driver.go

@@ -38,6 +38,14 @@ type BuildSystem interface {
 	ArtifactStores() []store.Store
 }
 
+// JobWatcher is an optional interface that ba uild system can implement
+// to support watching individual jobs rather than the entire run.
+type JobWatcher interface {
+	// AreJobsCompleted checks if the specified jobs are done. If jobNames
+	// is empty, it checks all jobs except excludeJob.
+	AreJobsCompleted(jobNames []string, excludeJob string) (completed bool, err error)
+}
+
 func NewFromSpecURL(specURL string) (BuildSystem, error) {
 	u, err := url.Parse(specURL)
 	if err != nil {

pkg/builder/driver/github.go

@@ -27,6 +27,7 @@ import (
 	"strings"
 	"time"
 
+	gogithub "github.com/google/go-github/v84/github"
 	intoto "github.com/in-toto/attestation/go/v1"
 	"github.com/sirupsen/logrus"
 	"sigs.k8s.io/tejolote/pkg/attestation"
@@ -41,6 +42,10 @@ type GitHubWorkflow struct {
 	Organization string
 	Repository   string
 	RunID        int
+
+	// workflow caches the parsed workflow YAML data to avoid
+	// repeated fetches when building predicates or discovering jobs.
+	workflow *github.WorkflowData
 }
 
 func parseGitHubURL(specURL string) (org, repo string, runID int64, err error) {
@@ -139,6 +144,37 @@ func (ghw *GitHubWorkflow) RefreshRun(r *run.Run) error {
 	return nil
 }
 
+// GetWorkflow returns the parsed workflow YAML data, fetching and caching
+// it on first call. Requires that RefreshRun has been called at least once
+// so that Organization, Repository and the run's SystemData are populated.
+func (ghw *GitHubWorkflow) GetWorkflow(r *run.Run) (*github.WorkflowData, error) {
+	if ghw.workflow != nil {
+		return ghw.workflow, nil
+	}
+
+	ghrun, ok := r.SystemData.(*github.Run)
+	if !ok {
+		return nil, fmt.Errorf("run system data is not a GitHub run")
+	}
+
+	wf, err := github.FetchWorkflow(
+		ghw.Organization, ghw.Repository, ghrun.Path, ghrun.HeadSHA,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("fetching workflow: %w", err)
+	}
+
+	ghw.workflow = wf
+	return wf, nil
+}
+
+// GetRunJobs fetches the jobs for this workflow run from the GitHub API.
+func (ghw *GitHubWorkflow) GetRunJobs() ([]*gogithub.WorkflowJob, error) {
+	return github.GetRunJobs(
+		ghw.Organization, ghw.Repository, int64(ghw.RunID),
+	)
+}
+
 // BuildPredicate builds a predicate from the run data
 func (ghw *GitHubWorkflow) BuildPredicate(
 	r *run.Run, draft attestation.Predicate,
@@ -189,12 +225,13 @@ func (ghw *GitHubWorkflow) BuildPredicate(
 			},
 		)
 
-		// Fetch the workflow YAML and compute effective inputs
-		definedInputs, err := github.FetchWorkflowInputs(org, repo, ghrun.Path, ghrun.HeadSHA)
+		// Fetch the workflow YAML (cached) and compute effective inputs
+		wf, err := ghw.GetWorkflow(r)
 		if err != nil {
-			return nil, fmt.Errorf("fetching workflow inputs: %w", err)
+			return nil, fmt.Errorf("fetching workflow: %w", err)
 		}
 
+		definedInputs := wf.Inputs()
 		if len(definedInputs) > 0 {
 			effective := github.EffectiveInputs(definedInputs, ghrun.Inputs)
 			for k, v := range effective {
@@ -234,6 +271,60 @@ func (ghw *GitHubWorkflow) BuildPredicate(
 	return predicate, nil
 }
 
+// AreJobsCompleted checks whether the specified jobs (by name) have all
+// completed. If jobNames is empty, all jobs in the run are checked except
+// the one matching excludeJob (useful for excluding the attester's own job).
+// Job name matching is prefix-based to handle reusable workflow jobs whose
+// API names are formatted as "caller_job / inner_job".
+func (ghw *GitHubWorkflow) AreJobsCompleted(jobNames []string, excludeJob string) (bool, error) {
+	jobs, err := ghw.GetRunJobs()
+	if err != nil {
+		return false, fmt.Errorf("fetching run jobs: %w", err)
+	}
+
+	for _, job := range jobs {
+		name := job.GetName()
+		status := job.GetStatus()
+
+		// Skip the excluded job (our own attester job)
+		if excludeJob != "" && matchJobName(name, excludeJob) {
+			logrus.Debugf("Skipping excluded job %q", name)
+			continue
+		}
+
+		// If specific jobs were requested, only check those
+		if len(jobNames) > 0 && !matchesAnyJobName(name, jobNames) {
+			continue
+		}
+
+		if status != "completed" {
+			logrus.Infof("Job %q status: %s — still running", name, status)
+			return false, nil
+		}
+
+		logrus.Debugf("Job %q completed with conclusion: %s", name, job.GetConclusion())
+	}
+
+	return true, nil
+}
+
+// matchJobName checks if an API job name matches a YAML job key.
+// GitHub Actions formats reusable workflow job names as "caller_key / inner_job",
+// so we match if the API name equals the key or starts with "key / ".
+func matchJobName(apiName, yamlKey string) bool {
+	return apiName == yamlKey || strings.HasPrefix(apiName, yamlKey+" / ")
+}
+
+// matchesAnyJobName checks if an API job name matches any of the provided names.
+func matchesAnyJobName(apiName string, names []string) bool {
+	for _, n := range names {
+		if matchJobName(apiName, n) {
+			return true
+		}
+	}
+	return false
+}
+
 // ArtifactStores returns the native artifact store of github actions
 func (ghw *GitHubWorkflow) ArtifactStores() []store.Store {
 	d, err := store.New(

pkg/github/github.go

@@ -18,6 +18,7 @@ package github
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -26,6 +27,7 @@ import (
 	"strings"
 	"time"
 
+	gogithub "github.com/google/go-github/v84/github"
 	"github.com/sirupsen/logrus"
 	khttp "sigs.k8s.io/release-utils/http"
 )
@@ -83,6 +85,31 @@ func APIGetRequest(url string) (*http.Response, error) {
 	return res, nil
 }
 
+// GetRunJobs fetches the jobs for a given workflow run from the GitHub API.
+func GetRunJobs(org, repo string, runID int64) ([]*gogithub.WorkflowJob, error) {
+	u := fmt.Sprintf(
+		"https://api.github.com/repos/%s/%s/actions/runs/%d/jobs",
+		org, repo, runID,
+	)
+	res, err := APIGetRequest(u)
+	if err != nil {
+		return nil, fmt.Errorf("querying jobs API: %w", err)
+	}
+	defer res.Body.Close()
+
+	rawData, err := io.ReadAll(res.Body)
+	if err != nil {
+		return nil, fmt.Errorf("reading jobs response: %w", err)
+	}
+
+	var jobsResp gogithub.Jobs
+	if err := json.Unmarshal(rawData, &jobsResp); err != nil {
+		return nil, fmt.Errorf("unmarshalling jobs response: %w", err)
+	}
+
+	return jobsResp.Jobs, nil
+}
+
 func Download(url string, f io.Writer) error {
 	agent := NewAgent()
 	return agent.GetToWriter(f, url)