feat: instrument error reporting and tracing for go jobs (#4819)

1. Makes error level (and the hypothetical levels above) logs create a
new error in the error reporting dashboard, which should hopefully link
to the logs
2. adds traces to all our go jobs, which should maybe let us use the
trace explorer to see how long things take in a fancy waterfall graph
(related to #3547)
- The exporter will create one big span for the exporter, and sub-spans
for each ecosystem
- The relations will create one big span for the relations, and a
sub-span for alias/upstream/related
- gitter traces each http request that is made to it. We can add
propagate the same trace from the worker/importer to correlate requests,
which should be fun.

This doesn't affect vulnfeeds (yet), since it uses a duplicate logging
thing.
To instrument tracing, you have to propagate the context correctly, and
the new `logger.[LEVEL]Context()` commands.

Author: michaelkedar

deployment/clouddeploy/gke-workers/base/exporter.yaml

@@ -31,4 +31,7 @@ spec:
               limits:
                 cpu: "20"
                 memory: "64Gi"
+            env:
+              - name: TRACE_SAMPLE_RATE
+                value: "1.0"
           restartPolicy: Never

deployment/clouddeploy/gke-workers/base/generate-sitemap.yaml

@@ -22,4 +22,7 @@ spec:
               limits:
                 cpu: "1"
                 memory: "6G"
+            env:
+              - name: TRACE_SAMPLE_RATE
+                value: "1.0"
           restartPolicy: OnFailure

deployment/clouddeploy/gke-workers/base/record-checker.yaml

@@ -22,4 +22,7 @@ spec:
               limits:
                 cpu: "1"
                 memory: "2G"
+            env:
+              - name: TRACE_SAMPLE_RATE
+                value: "1.0"
           restartPolicy: Never

deployment/clouddeploy/gke-workers/base/relations.yaml

@@ -22,4 +22,7 @@ spec:
               limits:
                 cpu: "1"
                 memory: "13G"
+            env:
+              - name: TRACE_SAMPLE_RATE
+                value: "1.0"
           restartPolicy: Never

go/cmd/custommetrics/main.go

@@ -12,6 +12,7 @@ import (
 	monitoring "cloud.google.com/go/monitoring/apiv3/v2"
 	"cloud.google.com/go/monitoring/apiv3/v2/monitoringpb"
 	"github.com/google/osv.dev/go/logger"
+	"go.opentelemetry.io/otel"
 	"google.golang.org/api/iterator"
 	"google.golang.org/genproto/googleapis/api/metric"
 	"google.golang.org/genproto/googleapis/api/monitoredres"
@@ -24,14 +25,19 @@ const (
 )
 
 func main() {
+	logger.InitGlobalLogger()
+	defer logger.Close()
+
+	ctx, span := otel.Tracer("custommetrics").Start(context.Background(), "custommetrics")
+	defer span.End()
+
 	project := os.Getenv("GOOGLE_CLOUD_PROJECT")
 	if project == "" {
-		logger.Fatal("GOOGLE_CLOUD_PROJECT must be set")
+		logger.FatalContext(ctx, "GOOGLE_CLOUD_PROJECT must be set")
 	}
-	ctx := context.Background()
 	cl, err := monitoring.NewMetricClient(ctx)
 	if err != nil {
-		logger.Fatal("failed to create monitoring client", slog.Any("err", err))
+		logger.FatalContext(ctx, "failed to create monitoring client", slog.Any("err", err))
 	}
 	defer cl.Close()
 
@@ -40,10 +46,10 @@ func main() {
 	for _, cron := range crons {
 		timeSince, err := getCronFreshness(ctx, cl, project, cron)
 		if err != nil {
-			logger.Fatal("error getting freshness", slog.String("cronjob", cron), slog.Any("err", err))
+			logger.FatalContext(ctx, "error getting freshness", slog.String("cronjob", cron), slog.Any("err", err))
 		}
 		if err := writeCronFreshness(ctx, cl, project, cron, timeSince); err != nil {
-			logger.Fatal("error writing freshness", slog.String("cronjob", cron), slog.Any("err", err))
+			logger.FatalContext(ctx, "error writing freshness", slog.String("cronjob", cron), slog.Any("err", err))
 		}
 	}
 }
@@ -59,14 +65,14 @@ func getCronFreshness(ctx context.Context, cl *monitoring.MetricClient, project
 	it := cl.ListTimeSeries(ctx, req)
 	ts, err := it.Next()
 	if errors.Is(err, iterator.Done) {
-		logger.Warn("last_successful_time was not found for past day", slog.String("cronjob", cronjob))
+		logger.WarnContext(ctx, "last_successful_time was not found for past day", slog.String("cronjob", cronjob))
 		return int64(lookbackDuration / time.Second), nil
 	} else if err != nil {
 		return 0, err
 	}
 	points := ts.GetPoints()
 	if len(points) == 0 { // I'm pretty sure iterator.Done would be returned instead.
-		logger.Warn("time series has no points", slog.String("cronjob", cronjob))
+		logger.WarnContext(ctx, "time series has no points", slog.String("cronjob", cronjob))
 		return int64(lookbackDuration / time.Second), nil
 	}
 	val := points[0].GetValue().GetDoubleValue()

go/cmd/exporter/exporter.go

@@ -15,6 +15,7 @@ import (
 	"github.com/google/osv.dev/go/logger"
 	"github.com/google/osv.dev/go/osv/clients"
 	"github.com/ossf/osv-schema/bindings/go/osvschema"
+	"go.opentelemetry.io/otel"
 )
 
 const gcsProtoPrefix = "all/pb/"
@@ -23,6 +24,10 @@ const gcsProtoPrefix = "all/pb/"
 // sets up the worker pipeline, and starts the GCS object iteration.
 func main() {
 	logger.InitGlobalLogger()
+	defer logger.Close()
+
+	ctx, span := otel.Tracer("exporter").Start(context.Background(), "exporter")
+	defer span.End()
 
 	outBucketName := flag.String("bucket", "osv-test-vulnerabilities", "Output bucket or directory name. If -local is true, this is a local path; otherwise, it's a GCS bucket name.")
 	vulnBucketName := flag.String("osv-vulns-bucket", os.Getenv("OSV_VULNERABILITIES_BUCKET"), "GCS bucket to read vulnerability protobufs from. Can also be set with the OSV_VULNERABILITIES_BUCKET environment variable.")
@@ -31,22 +36,22 @@ func main() {
 
 	flag.Parse()
 
-	logger.Info("exporter starting",
+	logger.InfoContext(ctx, "exporter starting",
 		slog.String("bucket", *outBucketName),
 		slog.String("osv-vulns-bucket", *vulnBucketName),
 		slog.Bool("upload-to-gcs", *uploadToGCS),
 		slog.Int("workers", *numWorkers))
 
 	if *vulnBucketName == "" {
-		logger.Fatal("OSV_VULNERABILITIES_BUCKET must be set")
+		logger.FatalContext(ctx, "OSV_VULNERABILITIES_BUCKET must be set")
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	storageClient, err := storage.NewClient(ctx)
 	if err != nil {
-		logger.Fatal("failed to create storage client", slog.Any("err", err))
+		logger.FatalContext(ctx, "failed to create storage client", slog.Any("err", err))
 	}
 	defer storageClient.Close()
 
@@ -93,13 +98,13 @@ func main() {
 MainLoop:
 	for path, err := range vulnClient.Objects(ctx, gcsProtoPrefix) {
 		if err != nil {
-			logger.Fatal("failed to list objects", slog.Any("err", err))
+			logger.FatalContext(ctx, "failed to list objects", slog.Any("err", err))
 		}
 		// Only log when we see a new ID prefix (i.e. roughly once per data source)
 		prefix := filepath.Base(path)
 		prefix, _, _ = strings.Cut(prefix, "-")
 		if prefix != prevPrefix {
-			logger.Info("iterating vulnerabilities", slog.String("now_at", path))
+			logger.InfoContext(ctx, "iterating vulnerabilities", slog.String("now_at", path))
 			prevPrefix = prefix
 		}
 		select {
@@ -117,17 +122,17 @@ MainLoop:
 	writerWg.Wait()
 
 	if ctx.Err() != nil {
-		logger.Fatal("exporter cancelled")
+		logger.FatalContext(ctx, "exporter cancelled")
 	}
-	logger.Info("export completed successfully")
+	logger.InfoContext(ctx, "export completed successfully")
 }
 
 // ecosystemRouter receives vulnerabilities from inCh and fans them out to the
 // appropriate ecosystemWorker. It creates workers on-demand for each new
 // ecosystem encountered. It also sends every vulnerability to the allEcosystemWorker.
 func ecosystemRouter(ctx context.Context, inCh <-chan *osvschema.Vulnerability, outCh chan<- writeMsg, wg *sync.WaitGroup) {
 	defer wg.Done()
-	logger.Info("ecosystem router starting")
+	logger.InfoContext(ctx, "ecosystem router starting")
 	workers := make(map[string]*ecosystemWorker)
 	var workersWg sync.WaitGroup
 	vulnCounter := 0
@@ -190,8 +195,8 @@ RouterLoop:
 	allEcosystemWorker.Finish()
 	workersWg.Wait()
 	if ctx.Err() == nil {
-		logger.Info("ecosystem router finished, all vulnerabilities dispatched", slog.Int("total_vulnerabilities", vulnCounter))
+		logger.InfoContext(ctx, "ecosystem router finished, all vulnerabilities dispatched", slog.Int("total_vulnerabilities", vulnCounter))
 	} else {
-		logger.Info("ecosystem router cancelled", slog.Any("err", ctx.Err()))
+		logger.InfoContext(ctx, "ecosystem router cancelled", slog.Any("err", ctx.Err()))
 	}
 }

go/cmd/exporter/worker.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/google/osv.dev/go/logger"
 	"github.com/ossf/osv-schema/bindings/go/osvschema"
+	"go.opentelemetry.io/otel"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -60,7 +61,10 @@ type vulnData struct {
 // zip, csv, and (for GIT) vanir files.
 func (w *ecosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sync.WaitGroup) {
 	defer wg.Done()
-	logger.Info("new ecosystem worker started", slog.String("ecosystem", w.ecosystem))
+	ctx, span := otel.Tracer("exporter").Start(ctx, w.ecosystem)
+	defer span.End()
+
+	logger.InfoContext(ctx, "new ecosystem worker started", slog.String("ecosystem", w.ecosystem))
 	var allVulns []vulnData
 	var csvData [][]string
 	var vanirVulns []vulnData
@@ -72,7 +76,7 @@ WorkLoop:
 		// Wait to receive a vulnerability, or be cancelled.
 		select {
 		case <-ctx.Done():
-			logger.Warn("ecosystem worker cancelled", slog.String("ecosystem", w.ecosystem), slog.Any("err", ctx.Err()))
+			logger.WarnContext(ctx, "ecosystem worker cancelled", slog.String("ecosystem", w.ecosystem), slog.Any("err", ctx.Err()))
 			return
 		case v, ok = <-w.inCh:
 			if !ok {
@@ -82,15 +86,15 @@ WorkLoop:
 		// Process vulnerability.
 		b, err := marshalToJSON(v)
 		if err != nil {
-			logger.Error("failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err))
+			logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err))
 			continue
 		}
 
 		// Wait to send the result, or be cancelled.
 		select {
 		case outCh <- writeMsg{path: filepath.Join(w.ecosystem, v.GetId()) + ".json", mimeType: "application/json", data: b}:
 		case <-ctx.Done():
-			logger.Warn("ecosystem worker cancelled", slog.String("ecosystem", w.ecosystem), slog.Any("err", ctx.Err()))
+			logger.WarnContext(ctx, "ecosystem worker cancelled", slog.String("ecosystem", w.ecosystem), slog.Any("err", ctx.Err()))
 			return
 		}
 
@@ -110,13 +114,13 @@ WorkLoop:
 		}
 	}
 
-	logger.Info("All vulnerabilities processed", slog.String("ecosystem", w.ecosystem))
+	logger.InfoContext(ctx, "All vulnerabilities processed", slog.String("ecosystem", w.ecosystem))
 	writeModifiedIDCSV(ctx, filepath.Join(w.ecosystem, modifiedCSVFilename), csvData, outCh)
 	writeZIP(ctx, filepath.Join(w.ecosystem, allZipFilename), allVulns, outCh)
 	if w.ecosystem == gitEcosystem {
 		writeVanir(ctx, vanirVulns, outCh)
 	}
-	logger.Info("ecosystem worker finished processing", slog.String("ecosystem", w.ecosystem))
+	logger.InfoContext(ctx, "ecosystem worker finished processing", slog.String("ecosystem", w.ecosystem))
 }
 
 // Finish signals the worker to stop processing by closing its input channel.
@@ -153,23 +157,26 @@ func newAllEcosystemWorker(ctx context.Context, outCh chan<- writeMsg, wg *sync.
 // and generates the global all.zip, modified_id.csv, and ecosystems.txt files.
 func (w *allEcosystemWorker) run(ctx context.Context, outCh chan<- writeMsg, wg *sync.WaitGroup) {
 	defer wg.Done()
-	logger.Info("all-ecosystem worker started")
+	ctx, span := otel.Tracer("exporter").Start(ctx, "all-ecosystems")
+	defer span.End()
+
+	logger.InfoContext(ctx, "all-ecosystem worker started")
 	var allVulns []vulnData
 	var csvData [][]string
 	ecosystems := make(map[string]struct{})
 WorkLoop:
 	for {
 		select {
 		case <-ctx.Done():
-			logger.Warn("all-ecosystem worker cancelled", slog.Any("err", ctx.Err()))
+			logger.WarnContext(ctx, "all-ecosystem worker cancelled", slog.Any("err", ctx.Err()))
 			return
 		case v, ok := <-w.inCh:
 			if !ok {
 				break WorkLoop
 			}
 			b, err := marshalToJSON(v.Vulnerability)
 			if err != nil {
-				logger.Error("failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err))
+				logger.ErrorContext(ctx, "failed to marshal vulnerability to json", slog.String("id", v.GetId()), slog.Any("err", err))
 				continue
 			}
 			modified := v.GetModified().AsTime()
@@ -186,7 +193,7 @@ WorkLoop:
 	slices.Sort(ecos)
 	ecoString := strings.Join(ecos, "\n") + "\n"
 	write(ctx, ecosystemsFilename, []byte(ecoString), "text/plain", outCh)
-	logger.Info("all-ecosystem worker finished processing")
+	logger.InfoContext(ctx, "all-ecosystem worker finished processing")
 }
 
 // Finish signals the worker to stop processing by closing its input channel.
@@ -219,7 +226,7 @@ func write(ctx context.Context, path string, data []byte, mimeType string, outCh
 
 // writeModifiedIDCSV constructs and writes a modified_id.csv file.
 func writeModifiedIDCSV(ctx context.Context, path string, csvData [][]string, outCh chan<- writeMsg) {
-	logger.Info("constructing csv file", slog.String("path", path))
+	logger.InfoContext(ctx, "constructing csv file", slog.String("path", path))
 	slices.SortFunc(csvData, func(a, b []string) int {
 		return cmp.Or(
 			-cmp.Compare(a[0], b[0]), // Modified date, descending
@@ -230,17 +237,17 @@ func writeModifiedIDCSV(ctx context.Context, path string, csvData [][]string, ou
 	var buf bytes.Buffer
 	wr := csv.NewWriter(&buf)
 	if err := wr.WriteAll(csvData); err != nil {
-		logger.Error("failed writing csv", slog.String("path", path), slog.Any("err", err))
+		logger.ErrorContext(ctx, "failed writing csv", slog.String("path", path), slog.Any("err", err))
 		return
 	}
 	wr.Flush()
-	logger.Info("writing csv file", slog.String("path", path))
+	logger.InfoContext(ctx, "writing csv file", slog.String("path", path))
 	write(ctx, path, buf.Bytes(), "text/csv", outCh)
 }
 
 // writeZIP constructs and writes an all.zip file.
 func writeZIP(ctx context.Context, path string, allVulns []vulnData, outCh chan<- writeMsg) {
-	logger.Info("constructing zip file", slog.String("path", path))
+	logger.InfoContext(ctx, "constructing zip file", slog.String("path", path))
 	slices.SortFunc(allVulns, func(a, b vulnData) int {
 		return cmp.Compare(a.id, b.id)
 	})
@@ -253,18 +260,18 @@ func writeZIP(ctx context.Context, path string, allVulns []vulnData, outCh chan<
 			Method:   zip.Deflate,
 		})
 		if err != nil {
-			logger.Error("failed to create vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err))
+			logger.ErrorContext(ctx, "failed to create vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err))
 			continue
 		}
 		r := bytes.NewReader(vuln.data)
 		if _, err := io.Copy(w, r); err != nil {
-			logger.Error("failed to write vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err))
+			logger.ErrorContext(ctx, "failed to write vuln json in zip file", slog.String("id", vuln.id), slog.Any("err", err))
 		}
 	}
 	if err := wr.Close(); err != nil {
-		logger.Error("failed to close zip writer", slog.String("path", path), slog.Any("err", err))
+		logger.ErrorContext(ctx, "failed to close zip writer", slog.String("path", path), slog.Any("err", err))
 	}
-	logger.Info("writing zip file", slog.String("path", path))
+	logger.InfoContext(ctx, "writing zip file", slog.String("path", path))
 	write(ctx, path, buf.Bytes(), "application/zip", outCh)
 }
 
@@ -277,7 +284,7 @@ func writeVanir(ctx context.Context, vanirVulns []vulnData, outCh chan<- writeMs
 	}
 	finalJSON, err := json.Marshal(vulns)
 	if err != nil {
-		logger.Error("failed to marshal vanir JSON file", slog.Any("err", err))
+		logger.ErrorContext(ctx, "failed to marshal vanir JSON file", slog.Any("err", err))
 		return
 	}
 	write(ctx, filepath.Join(gitEcosystem, vanirVulnsFilename), finalJSON, "application/json", outCh)