blob: 2de976257d5c41b87e19eaafff57a5bbc4131edb [file] [log] [blame]
// Copyright 2022 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package worker provides functionality for running a worker service.
package worker
import (
"context"
"errors"
"fmt"
"net/http"
"os"
"strings"
"sync"
"time"
"cloud.google.com/go/errorreporting"
"golang.org/x/pkgsite-metrics/internal/analysis"
"golang.org/x/pkgsite-metrics/internal/bigquery"
"golang.org/x/pkgsite-metrics/internal/config"
"golang.org/x/pkgsite-metrics/internal/derrors"
"golang.org/x/pkgsite-metrics/internal/govulncheck"
"golang.org/x/pkgsite-metrics/internal/jobs"
"golang.org/x/pkgsite-metrics/internal/log"
"golang.org/x/pkgsite-metrics/internal/observe"
"golang.org/x/pkgsite-metrics/internal/proxy"
"golang.org/x/pkgsite-metrics/internal/queue"
)
type Server struct {
cfg *config.Config
observer *observe.Observer
bqClient *bigquery.Client
proxyClient *proxy.Client
queue queue.Queue
jobDB *jobs.DB
devMode bool
mu sync.Mutex
}
func NewServer(ctx context.Context, cfg *config.Config) (_ *Server, err error) {
defer derrors.WrapAndReport(&err, "NewServer")
var bq *bigquery.Client
if strings.EqualFold(cfg.BigQueryDataset, "disable") {
log.Infof(ctx, "BigQuery disabled")
} else {
bq, err = bigquery.NewClientCreate(ctx, cfg.ProjectID, cfg.BigQueryDataset)
if err != nil {
return nil, err
}
}
q, err := queue.New(ctx, cfg,
func(ctx context.Context, t queue.Task) (int, error) {
// When running locally, only the module path and version are
// printed for now.
log.Infof(ctx, "enqueuing %s?%s", t.Path(), t.Params())
return 0, nil
})
log.Debugf(ctx, "queue.New returned err %v", err)
if err != nil {
return nil, err
}
if err != nil {
return nil, err
}
proxyClient, err := proxy.New(cfg.ProxyURL)
log.Debugf(ctx, "proxy.New returned err %v", err)
if err != nil {
return nil, err
}
var jdb *jobs.DB
if cfg.ProjectID != "" {
var err error
jdb, err = jobs.NewDB(ctx, cfg.ProjectID, cfg.BigQueryDataset)
if err != nil {
return nil, err
}
}
s := &Server{
cfg: cfg,
bqClient: bq,
queue: q,
proxyClient: proxyClient,
devMode: cfg.DevMode,
jobDB: jdb,
}
if cfg.ProjectID != "" && cfg.ServiceID != "" {
s.observer, err = observe.NewObserver(ctx, cfg.ProjectID, cfg.ServiceID)
log.Debugf(ctx, "observe.NewObserver returned err %v", err)
if err != nil {
return nil, err
}
}
if cfg.UseErrorReporting {
reportingClient, err := errorreporting.NewClient(ctx, cfg.ProjectID, errorreporting.Config{
ServiceName: cfg.ServiceID,
OnError: func(err error) {
log.Errorf(ctx, err, "error-reporting failed")
},
})
log.Debugf(ctx, "errorreporting.NewClient returned err %v", err)
if err != nil {
return nil, err
}
derrors.SetReportingClient(reportingClient)
}
if err := ensureTable(ctx, bq, govulncheck.TableName); err != nil {
return nil, err
}
s.registerGovulncheckHandlers()
if err := ensureTable(ctx, bq, analysis.TableName); err != nil {
return nil, err
}
if err := s.registerAnalysisHandlers(ctx); err != nil {
return nil, err
}
// compute vulndb entries
s.handle("/vulndb", s.handleVulnDB)
// compute missing vuln.go.dev request counts
s.handle("/compute-requests", s.handleComputeRequests)
s.handle("/jobs/", s.handleJobs)
return s, nil
}
func ensureTable(ctx context.Context, bq *bigquery.Client, name string) error {
if bq == nil {
return nil
}
created, err := bq.CreateOrUpdateTable(ctx, name)
if err != nil {
return err
}
verb := "updated"
if created {
verb = "created"
}
log.Infof(ctx, "%s table %s\n", verb, name)
return nil
}
const metricNamespace = "ecosystem/worker"
type handlerFunc func(w http.ResponseWriter, r *http.Request) error
func (s *Server) handle(pattern string, handler handlerFunc) {
h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
ctx := r.Context()
logger := log.FromContext(ctx)
if t := r.Header.Get("X-Cloud-Trace-Context"); t != "" {
logger = logger.With("traceID", t)
}
ctx = log.NewContext(ctx, logger)
r = r.WithContext(ctx)
// For logging, construct a string with the entire URL except scheme and host.
url2 := *r.URL
url2.Scheme = ""
url2.Host = ""
urlString := url2.String()
log.Infof(ctx, "starting %s", urlString)
w2 := &responseWriter{ResponseWriter: w}
if err := handler(w2, r); err != nil {
derrors.Report(err)
s.serveError(ctx, w2, r, err)
}
logger.Info(fmt.Sprintf("ending %s", urlString),
"latency", time.Since(start),
"status", translateStatus(w2.status))
})
http.Handle(pattern, s.observer.Observe(h))
}
func (s *Server) registerGovulncheckHandlers() {
h := newGovulncheckServer(s)
s.handle("/govulncheck/enqueueall", h.handleEnqueueAll)
s.handle("/govulncheck/enqueue", h.handleEnqueue)
s.handle("/govulncheck/scan/", h.handleScan)
}
func (s *Server) registerAnalysisHandlers(ctx context.Context) error {
h, err := newAnalysisServer(ctx, s)
if err != nil {
return err
}
s.handle("/analysis/scan/", h.handleScan)
s.handle("/analysis/enqueue", h.handleEnqueue)
return nil
}
type serverError struct {
status int // HTTP status code
err error // wrapped error
}
func (s *serverError) Error() string {
return fmt.Sprintf("%d (%s): %v", s.status, http.StatusText(s.status), s.err)
}
func (s *Server) serveError(ctx context.Context, w http.ResponseWriter, _ *http.Request, err error) {
if errors.Is(err, derrors.InvalidArgument) {
err = &serverError{err: err, status: http.StatusBadRequest}
}
if errors.Is(err, derrors.NotFound) {
err = &serverError{err: err, status: http.StatusNotFound}
}
if errors.Is(err, derrors.BadModule) {
err = &serverError{err: err, status: http.StatusNotAcceptable}
}
serr, ok := err.(*serverError)
if !ok {
serr = &serverError{status: http.StatusInternalServerError, err: err}
}
if serr.status == http.StatusInternalServerError {
log.Errorf(ctx, err, "internal server error")
} else {
log.Warnf(ctx, "returning %v", err)
}
http.Error(w, serr.err.Error(), serr.status)
}
type responseWriter struct {
http.ResponseWriter
status int
}
func (rw *responseWriter) WriteHeader(code int) {
rw.status = code
rw.ResponseWriter.WriteHeader(code)
}
func translateStatus(code int) int64 {
if code == 0 {
return http.StatusOK
}
return int64(code)
}
var locNewYork *time.Location
func init() {
var err error
locNewYork, err = time.LoadLocation("America/New_York")
if err != nil {
log.Errorf(context.Background(), err, "time.LoadLocation")
os.Exit(1)
}
}
func FormatTime(t time.Time) string {
if t.IsZero() {
return "-"
}
return t.In(locNewYork).Format("2006-01-02 15:04:05")
}