blob: 387ebe76b2c77b744fe9035be4829606a4c451a7 [file] [log] [blame]
// Copyright 2023 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.
// Telemetrygodev serves the telemetry.go.dev website.
package main
import (
"context"
_ "embed"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"io/fs"
"log"
"net/http"
"os"
"path"
"strings"
"time"
"golang.org/x/mod/semver"
"golang.org/x/telemetry"
"golang.org/x/telemetry/godev"
"golang.org/x/telemetry/godev/internal/content"
"golang.org/x/telemetry/godev/internal/middleware"
"golang.org/x/telemetry/godev/internal/storage"
"golang.org/x/telemetry/godev/internal/unionfs"
tconfig "golang.org/x/telemetry/internal/config"
)
func main() {
flag.Parse()
ctx := context.Background()
cfg := newConfig()
buckets, err := buckets(ctx, cfg)
if err != nil {
log.Fatal(err)
}
ucfg, err := tconfig.ReadConfig(cfg.UploadConfig)
if err != nil {
log.Fatal(err)
}
fsys := fsys(cfg.DevMode)
cserv := content.Server(fsys)
mux := http.NewServeMux()
mux.Handle("/", cserv)
mux.Handle("/upload/", handleUpload(ucfg, buckets))
mux.Handle("/charts/", handleChart(fsys, ucfg, buckets))
mw := middleware.Chain(
middleware.Log,
middleware.Timeout(cfg.RequestTimeout),
middleware.RequestSize(cfg.MaxRequestBytes),
middleware.Recover,
)
fmt.Printf("server listening at http://localhost:%s\n", cfg.Port)
log.Fatal(http.ListenAndServe(":"+cfg.Port, mw(mux)))
}
type chartPage struct {
Charts map[string]any
}
func handleChart(fsys fs.FS, ucfg *tconfig.Config, buckets *stores) content.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) error {
ctx := r.Context()
p := strings.TrimPrefix(path.Clean(r.URL.Path), "/charts")
reader, err := buckets.chart.Reader(ctx, p+".json")
if errors.Is(err, storage.ErrObjectNotExist) {
return content.Status(w, http.StatusNotFound)
} else if err != nil {
return err
}
defer reader.Close()
data, err := io.ReadAll(reader)
if err != nil {
return err
}
var charts map[string]any
if err := json.Unmarshal(data, &charts); err != nil {
return err
}
page := chartPage{
Charts: charts,
}
return content.Template(w, fsys, "charts.html", page, http.StatusOK)
}
}
func handleUpload(ucfg *tconfig.Config, buckets *stores) content.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) error {
if r.Method == "POST" {
var report telemetry.Report
if err := json.NewDecoder(r.Body).Decode(&report); err != nil {
return content.Error(err, http.StatusBadRequest)
}
if err := validate(&report, ucfg); err != nil {
return content.Error(err, http.StatusBadRequest)
}
// TODO: capture metrics for collisions.
ctx := r.Context()
name := fmt.Sprintf("%s/%g.json", report.Week, report.X)
f, err := buckets.upload.Writer(ctx, name)
if err != nil {
return err
}
defer f.Close()
if err := json.NewEncoder(f).Encode(report); err != nil {
return err
}
if err := f.Close(); err != nil {
return err
}
return content.Status(w, http.StatusOK)
}
return content.Status(w, http.StatusMethodNotAllowed)
}
}
// validate validates the telemetry report data against the latest config.
func validate(r *telemetry.Report, cfg *tconfig.Config) error {
// TODO: reject/drop data arrived too early or too late.
if _, err := time.Parse("2006-01-02", r.Week); err != nil {
return fmt.Errorf("invalid week %s", r.Week)
}
if !semver.IsValid(r.Config) {
return fmt.Errorf("invalid config %s", r.Config)
}
if r.X == 0 {
return fmt.Errorf("invalid X %g", r.X)
}
// TODO: We can probably keep known programs and counters even when a report
// includes something that has been removed from the latest config.
for _, p := range r.Programs {
if !cfg.HasGOARCH(p.GOARCH) ||
!cfg.HasGOOS(p.GOOS) ||
!cfg.HasGoVersion(p.GoVersion) ||
!cfg.HasProgram(p.Program) ||
!cfg.HasVersion(p.Program, p.Version) {
return fmt.Errorf("unknown program build %s@%s %s %s/%s", p.Program, p.Version, p.GoVersion, p.GOOS, p.GOARCH)
}
for c := range p.Counters {
if !cfg.HasCounter(p.Program, c) {
return fmt.Errorf("unknown counter %s", c)
}
}
for s := range p.Stacks {
if !cfg.HasStack(p.Program, s) {
return fmt.Errorf("unknown stack %s", s)
}
}
}
return nil
}
func fsys(fromOS bool) fs.FS {
var f fs.FS = godev.FS
if fromOS {
f = os.DirFS(".")
}
f, err := unionfs.Sub(f, "content/telemetrygodev", "content/shared", "third_party")
if err != nil {
log.Fatal(err)
}
return f
}
type stores struct {
upload storage.Store
merge storage.Store
chart storage.Store
}
func buckets(ctx context.Context, cfg *config) (*stores, error) {
if cfg.UseGCS && !cfg.onCloudRun() {
if err := os.Setenv("STORAGE_EMULATOR_HOST", cfg.StorageEmulatorHost); err != nil {
return nil, err
}
}
var upload storage.Store
var merge storage.Store
var chart storage.Store
var err error
if cfg.UseGCS {
upload, err = storage.NewGCStore(ctx, cfg.ProjectID, cfg.UploadBucket)
if err != nil {
return nil, err
}
merge, err = storage.NewGCStore(ctx, cfg.ProjectID, cfg.MergedBucket)
if err != nil {
return nil, err
}
chart, err = storage.NewGCStore(ctx, cfg.ProjectID, cfg.ChartDataBucket)
if err != nil {
return nil, err
}
} else {
upload, err = storage.NewFSStore(ctx, cfg.LocalStorage, cfg.UploadBucket)
if err != nil {
return nil, err
}
merge, err = storage.NewFSStore(ctx, cfg.LocalStorage, cfg.MergedBucket)
if err != nil {
return nil, err
}
chart, err = storage.NewFSStore(ctx, cfg.LocalStorage, cfg.ChartDataBucket)
if err != nil {
return nil, err
}
}
return &stores{upload, merge, chart}, nil
}