cli to push metrics to aggregator service (#8835)

* new prometheus metrics for client-stats metrics

* adds client-stats types beacause they are
  used by some of the prometheus collection code.
* new prometheus collector for db disk size
* new prometheus collector for web3 client
  connection status

* adds client-stats api push cli in cmd/client-stats

* adds api metadata to client-stats collector posts

* appease deepsource

* mop up copypasta

* use prysm assert package for testing

beacon-chain/powchain/service.go

@@ -312,7 +312,7 @@ func (s *Service) Status() error {
 	return nil
 }
 
-func (s *Service) updateBeaconnodeStats() {
+func (s *Service) updateBeaconNodeStats() {
 	bs := clientstats.BeaconNodeStats{}
 	if len(s.httpEndpoints) > 1 {
 		bs.SyncEth1FallbackConfigured = true
@@ -329,12 +329,12 @@ func (s *Service) updateBeaconnodeStats() {
 
 func (s *Service) updateCurrHttpEndpoint(endpoint httputils.Endpoint) {
 	s.currHttpEndpoint = endpoint
-	s.updateBeaconnodeStats()
+	s.updateBeaconNodeStats()
 }
 
 func (s *Service) updateConnectedETH1(state bool) {
 	s.connectedETH1 = state
-	s.updateBeaconnodeStats()
+	s.updateBeaconNodeStats()
 }
 
 // IsConnectedToETH1 checks if the beacon node is connected to a ETH1 Node.

cmd/client-stats/BUILD.bazel

@@ -0,0 +1,31 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_binary")
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "log.go",
+        "main.go",
+        "usage.go",
+    ],
+    importpath = "github.com/prysmaticlabs/prysm/cmd/client-stats",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//cmd/client-stats/flags:go_default_library",
+        "//shared/clientstats:go_default_library",
+        "//shared/cmd:go_default_library",
+        "//shared/journald:go_default_library",
+        "//shared/logutil:go_default_library",
+        "//shared/version:go_default_library",
+        "@com_github_joonix_log//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+        "@com_github_urfave_cli_v2//:go_default_library",
+        "@com_github_x_cray_logrus_prefixed_formatter//:go_default_library",
+    ],
+)
+
+go_binary(
+    name = "client-stats",
+    embed = [":go_default_library"],
+    visibility = ["//visibility:public"],
+)

cmd/client-stats/flags/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["flags.go"],
+    importpath = "github.com/prysmaticlabs/prysm/cmd/client-stats/flags",
+    visibility = ["//visibility:public"],
+    deps = ["@com_github_urfave_cli_v2//:go_default_library"],
+)

cmd/client-stats/flags/flags.go

@@ -0,0 +1,30 @@
+// Package flags contains all configuration runtime flags for
+// the client-stats daemon.
+package flags
+
+import (
+	"github.com/urfave/cli/v2"
+)
+
+var (
+	// BeaconCertFlag defines a flag for the beacon api certificate.
+	ValidatorMetricsURLFlag = &cli.StringFlag{
+		Name:  "validator-metrics-url",
+		Usage: "Full URL to the validator /metrics prometheus endpoint to scrape. eg http://localhost:8081/metrics",
+	}
+	// BeaconRPCProviderFlag defines a flag for the beacon host ip or address.
+	BeaconnodeMetricsURLFlag = &cli.StringFlag{
+		Name:  "beacon-node-metrics-url",
+		Usage: "Full URL to the beacon-node /metrics prometheus endpoint to scrape. eg http://localhost:8080/metrics",
+	}
+	// CertFlag defines a flag for the node's TLS certificate.
+	ClientStatsAPIURLFlag = &cli.StringFlag{
+		Name:  "clientstats-api-url",
+		Usage: "Full URL to the client stats endpoint where collected metrics should be sent.",
+	}
+	// CertFlag defines a flag for the node's TLS certificate.
+	ScrapeIntervalFlag = &cli.DurationFlag{
+		Name:  "scrape-interval",
+		Usage: "Frequency of scraping expressed as a duration, eg 2m or 1m5s. Default is 60s.",
+	}
+)

cmd/client-stats/log.go

@@ -0,0 +1,5 @@
+package main
+
+import "github.com/sirupsen/logrus"
+
+var log = logrus.WithField("prefix", "main")

cmd/client-stats/main.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	runtimeDebug "runtime/debug"
+	"time"
+
+	joonix "github.com/joonix/log"
+	"github.com/prysmaticlabs/prysm/cmd/client-stats/flags"
+	"github.com/prysmaticlabs/prysm/shared/clientstats"
+	"github.com/prysmaticlabs/prysm/shared/cmd"
+	"github.com/prysmaticlabs/prysm/shared/journald"
+	"github.com/prysmaticlabs/prysm/shared/logutil"
+	"github.com/prysmaticlabs/prysm/shared/version"
+	"github.com/sirupsen/logrus"
+	"github.com/urfave/cli/v2"
+	prefixed "github.com/x-cray/logrus-prefixed-formatter"
+)
+
+var appFlags = []cli.Flag{
+	cmd.VerbosityFlag,
+	cmd.LogFormat,
+	cmd.LogFileName,
+	cmd.ConfigFileFlag,
+	flags.BeaconnodeMetricsURLFlag,
+	flags.ValidatorMetricsURLFlag,
+	flags.ClientStatsAPIURLFlag,
+	flags.ScrapeIntervalFlag,
+}
+var scrapeInterval = 60 * time.Second
+
+func main() {
+	app := cli.App{}
+	app.Name = "client-stats"
+	app.Usage = "daemon to scrape client-stats from prometheus and ship to a remote endpoint"
+	app.Action = run
+	app.Version = version.Version()
+
+	app.Flags = appFlags
+
+	// logging/config setup cargo-culted from beaconchain
+	app.Before = func(ctx *cli.Context) error {
+		// Load flags from config file, if specified.
+		if err := cmd.LoadFlagsFromConfig(ctx, app.Flags); err != nil {
+			return err
+		}
+
+		verbosity := ctx.String(cmd.VerbosityFlag.Name)
+		level, err := logrus.ParseLevel(verbosity)
+		if err != nil {
+			return err
+		}
+		logrus.SetLevel(level)
+
+		format := ctx.String(cmd.LogFormat.Name)
+		switch format {
+		case "text":
+			formatter := new(prefixed.TextFormatter)
+			formatter.TimestampFormat = "2006-01-02 15:04:05"
+			formatter.FullTimestamp = true
+			// If persistent log files are written - we disable the log messages coloring because
+			// the colors are ANSI codes and seen as gibberish in the log files.
+			formatter.DisableColors = ctx.String(cmd.LogFileName.Name) != ""
+			logrus.SetFormatter(formatter)
+		case "fluentd":
+			f := joonix.NewFormatter()
+			if err := joonix.DisableTimestampFormat(f); err != nil {
+				panic(err)
+			}
+			logrus.SetFormatter(f)
+		case "json":
+			logrus.SetFormatter(&logrus.JSONFormatter{})
+		case "journald":
+			if err := journald.Enable(); err != nil {
+				return err
+			}
+		default:
+			return fmt.Errorf("unknown log format %s", format)
+		}
+
+		logFileName := ctx.String(cmd.LogFileName.Name)
+		if logFileName != "" {
+			if err := logutil.ConfigurePersistentLogging(logFileName); err != nil {
+				log.WithError(err).Error("Failed to configuring logging to disk.")
+			}
+		}
+		return nil
+	}
+
+	defer func() {
+		if x := recover(); x != nil {
+			log.Errorf("Runtime panic: %v\n%v", x, string(runtimeDebug.Stack()))
+			panic(x)
+		}
+	}()
+
+	if err := app.Run(os.Args); err != nil {
+		log.Error(err.Error())
+	}
+}
+
+func run(ctx *cli.Context) error {
+	if ctx.IsSet(flags.ScrapeIntervalFlag.Name) {
+		scrapeInterval = ctx.Duration(flags.ScrapeIntervalFlag.Name)
+	}
+
+	var upd clientstats.Updater
+	if ctx.IsSet(flags.ClientStatsAPIURLFlag.Name) {
+		u := ctx.String(flags.ClientStatsAPIURLFlag.Name)
+		upd = clientstats.NewClientStatsHTTPPostUpdater(u)
+	} else {
+		log.Warn("No --clientstats-api-url flag set, writing to stdout as default metrics sink.")
+		upd = clientstats.NewGenericClientStatsUpdater(os.Stdout)
+	}
+
+	scrapers := make([]clientstats.Scraper, 0)
+	if ctx.IsSet(flags.BeaconnodeMetricsURLFlag.Name) {
+		u := ctx.String(flags.BeaconnodeMetricsURLFlag.Name)
+		scrapers = append(scrapers, clientstats.NewBeaconNodeScraper(u))
+	}
+	if ctx.IsSet(flags.ValidatorMetricsURLFlag.Name) {
+		u := ctx.String(flags.ValidatorMetricsURLFlag.Name)
+		scrapers = append(scrapers, clientstats.NewValidatorScraper(u))
+	}
+
+	ticker := time.NewTicker(scrapeInterval)
+	for {
+		select {
+		case <-ticker.C:
+			for _, s := range scrapers {
+				r, err := s.Scrape()
+				if err != nil {
+					log.Errorf("Scraper error: %s", err)
+					continue
+				}
+				err = upd.Update(r)
+				if err != nil {
+					log.Errorf("client-stats collector error: %s", err)
+					continue
+				}
+			}
+		case <-ctx.Done():
+			ticker.Stop()
+			return nil
+		}
+	}
+}

cmd/client-stats/usage.go

@@ -0,0 +1,81 @@
+// This code was adapted from https://github.com/ethereum/go-ethereum/blob/master/cmd/geth/usage.go
+package main
+
+import (
+	"io"
+	"sort"
+
+	"github.com/prysmaticlabs/prysm/cmd/client-stats/flags"
+	"github.com/prysmaticlabs/prysm/shared/cmd"
+	"github.com/urfave/cli/v2"
+)
+
+var appHelpTemplate = `NAME:
+   {{.App.Name}} - {{.App.Usage}}
+USAGE:
+   {{.App.HelpName}} [options]{{if .App.Commands}} command [command options]{{end}} {{if .App.ArgsUsage}}{{.App.ArgsUsage}}{{else}}[arguments...]{{end}}
+   {{if .App.Version}}
+AUTHOR:
+   {{range .App.Authors}}{{ . }}{{end}}
+   {{end}}{{if .App.Commands}}
+GLOBAL OPTIONS:
+   {{range .App.Commands}}{{join .Names ", "}}{{ "\t" }}{{.Usage}}
+   {{end}}{{end}}{{if .FlagGroups}}
+{{range .FlagGroups}}{{.Name}} OPTIONS:
+  {{range .Flags}}{{.}}
+  {{end}}
+{{end}}{{end}}{{if .App.Copyright }}
+COPYRIGHT:
+   {{.App.Copyright}}
+VERSION:
+   {{.App.Version}}
+   {{end}}{{if len .App.Authors}}
+   {{end}}
+`
+
+type flagGroup struct {
+	Name  string
+	Flags []cli.Flag
+}
+
+var appHelpFlagGroups = []flagGroup{
+	{
+		Name: "cmd",
+		Flags: []cli.Flag{
+			cmd.VerbosityFlag,
+			cmd.LogFormat,
+			cmd.LogFileName,
+			cmd.ConfigFileFlag,
+		},
+	},
+	{
+		Name: "client-stats",
+		Flags: []cli.Flag{
+			flags.BeaconnodeMetricsURLFlag,
+			flags.ValidatorMetricsURLFlag,
+			flags.ClientStatsAPIURLFlag,
+			flags.ScrapeIntervalFlag,
+		},
+	},
+}
+
+func init() {
+	cli.AppHelpTemplate = appHelpTemplate
+
+	type helpData struct {
+		App        interface{}
+		FlagGroups []flagGroup
+	}
+
+	originalHelpPrinter := cli.HelpPrinter
+	cli.HelpPrinter = func(w io.Writer, tmpl string, data interface{}) {
+		if tmpl == appHelpTemplate {
+			for _, group := range appHelpFlagGroups {
+				sort.Sort(cli.FlagsByName(group.Flags))
+			}
+			originalHelpPrinter(w, tmpl, helpData{data, appHelpFlagGroups})
+		} else {
+			originalHelpPrinter(w, tmpl, data)
+		}
+	}
+}

deps.bzl

@@ -2669,6 +2669,13 @@ def prysm_deps():
         sum = "h1:Uehi/mxLK0eiUc0H0++5tpMGTexB8wZ598MIgU8VpDM=",
         version = "v0.3.0",
     )
+    go_repository(
+        name = "com_github_prometheus_prom2json",
+        importpath = "github.com/prometheus/prom2json",
+        sum = "h1:BlqrtbT9lLH3ZsOVhXPsHzFrApCTKRifB7gjJuypu6Y=",
+        version = "v1.3.0",
+    )
+
     go_repository(
         name = "com_github_prometheus_tsdb",
         importpath = "github.com/prometheus/tsdb",

go.mod

@@ -82,7 +82,9 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prestonvanloon/go-recaptcha v0.0.0-20190217191114-0834cef6e8bd
 	github.com/prometheus/client_golang v1.9.0
+	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/procfs v0.3.0 // indirect
+	github.com/prometheus/prom2json v1.3.0
 	github.com/prometheus/tsdb v0.10.0 // indirect
 	github.com/prysmaticlabs/eth2-types v0.0.0-20210219172114-1da477c09a06
 	github.com/prysmaticlabs/ethereumapis v0.0.0-20210311175904-cf9f64632dd4

go.sum

@@ -49,6 +49,9 @@ github.com/Azure/go-autorest/autorest/date v0.1.0/go.mod h1:plvfp3oPSKwf2DNjlBjW
 github.com/Azure/go-autorest/autorest/date v0.1.0/go.mod h1:plvfp3oPSKwf2DNjlBjWF/7vwR+cUD/ELuzDCXwHUVA=
 github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
 github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
+github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
+github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
+github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
 github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
 github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
 github.com/Azure/go-autorest/logger v0.1.0/go.mod h1:oExouG+K6PryycPJfVSxi/koC6LSNgds39diKLz7Vrc=
@@ -123,6 +126,7 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
+github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
 github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+WjiTceGBHIoTvn60HIbs7Hm7bcHjyrSqYB9c=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/bradfitz/gomemcache v0.0.0-20170208213004-1952afaa557d/go.mod h1:PmM6Mmwb0LSuEubjR8N7PtNe1KxZLtOUHtbeikc5h60=
@@ -163,6 +167,7 @@ github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMn
 github.com/clbanning/x2j v0.0.0-20191024224557-825249438eec/go.mod h1:jMjuTZXRI4dUb/I5gc9Hdhagfvm9+RyrPryS/auMzxE=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.14.0/go.mod h1:EnwdgGMaFOruiPZRFSgn+TsQ3hQ7C/YWzIGLeu5c304=
+github.com/cloudflare/cloudflare-go v0.14.0/go.mod h1:EnwdgGMaFOruiPZRFSgn+TsQ3hQ7C/YWzIGLeu5c304=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=
@@ -170,11 +175,19 @@ github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE
 github.com/confluentinc/confluent-kafka-go v1.4.2 h1:13EK9RTujF7lVkvHQ5Hbu6bM+Yfrq8L0MkJNnjHSd4Q=
 github.com/confluentinc/confluent-kafka-go v1.4.2/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg=
 github.com/consensys/bavard v0.1.8-0.20210105233146-c16790d2aa8b/go.mod h1:Bpd0/3mZuaj6Sj+PqrmIquiOKy397AKGThQPaGzNXAQ=
+github.com/consensys/bavard v0.1.8-0.20210105233146-c16790d2aa8b/go.mod h1:Bpd0/3mZuaj6Sj+PqrmIquiOKy397AKGThQPaGzNXAQ=
+github.com/consensys/goff v0.3.10/go.mod h1:xTldOBEHmFiYS0gPXd3NsaEqZWlnmeWcRLWgD3ba3xc=
 github.com/consensys/goff v0.3.10/go.mod h1:xTldOBEHmFiYS0gPXd3NsaEqZWlnmeWcRLWgD3ba3xc=
 github.com/consensys/gurvy v0.3.8/go.mod h1:sN75xnsiD593XnhbhvG2PkOy194pZBzqShWF/kwuW/g=
+github.com/consensys/gurvy v0.3.8/go.mod h1:sN75xnsiD593XnhbhvG2PkOy194pZBzqShWF/kwuW/g=
+github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
+github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
+github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
+github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
+github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -309,6 +322,7 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me
 github.com/go-yaml/yaml v2.1.0+incompatible h1:RYi2hDdss1u4YE7GwixGzWwVo47T8UQwnTLB6vQiq+o=
 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
 github.com/gofrs/uuid v3.3.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
+github.com/gofrs/uuid v3.3.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -432,6 +446,8 @@ github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmv
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
 github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE=
 github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE=
+github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE=
+github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE=
 github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
 github.com/hashicorp/consul/sdk v0.3.0/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -477,10 +493,17 @@ github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ
 github.com/inconshreveable/log15 v0.0.0-20170622235902-74a0988b5f80/go.mod h1:cOaXtrgN4ScfRrD9Bre7U1thNq5RtJ8ZoP4iXVGRj6o=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY=
+github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY=
 github.com/influxdata/influxdb v1.8.3 h1:WEypI1BQFTT4teLM+1qkEcvUi0dAvopAI/ir0vAiBg8=
+github.com/influxdata/influxdb v1.8.3 h1:WEypI1BQFTT4teLM+1qkEcvUi0dAvopAI/ir0vAiBg8=
+github.com/influxdata/influxdb v1.8.3/go.mod h1:JugdFhsvvI8gadxOI6noqNeeBHvWNTbfYGtiAn+2jhI=
 github.com/influxdata/influxdb v1.8.3/go.mod h1:JugdFhsvvI8gadxOI6noqNeeBHvWNTbfYGtiAn+2jhI=
 github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
 github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
+github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
+github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
+github.com/influxdata/influxql v1.1.1-0.20200828144457-65d3ef77d385/go.mod h1:gHp9y86a/pxhjJ+zMjNXiQAA197Xk9wLxaz+fGG+kWk=
+github.com/influxdata/influxql v1.1.1-0.20200828144457-65d3ef77d385/go.mod h1:gHp9y86a/pxhjJ+zMjNXiQAA197Xk9wLxaz+fGG+kWk=
 github.com/influxdata/influxql v1.1.1-0.20200828144457-65d3ef77d385/go.mod h1:gHp9y86a/pxhjJ+zMjNXiQAA197Xk9wLxaz+fGG+kWk=
 github.com/influxdata/line-protocol v0.0.0-20180522152040-32c6aa80de5e/go.mod h1:4kt73NQhadE3daL3WhR5EJ/J2ocX0PZzwxQ0gXJ7oFE=
 github.com/influxdata/promql/v2 v2.12.0/go.mod h1:fxOPu+DY0bqCTCECchSRtWfc+0X19ybifQhZoQNF5D8=
@@ -545,6 +568,8 @@ github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/joonix/log v0.0.0-20200409080653-9c1d2ceb5f1d h1:k+SfYbN66Ev/GDVq39wYOXVW5RNd5kzzairbCe9dK5Q=
@@ -585,6 +610,7 @@ github.com/klauspost/crc32 v0.0.0-20161016154125-cb6bfca970f6/go.mod h1:+ZoRqAPR
 github.com/klauspost/pgzip v1.0.2-0.20170402124221-0bf5dcad4ada/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
 github.com/klauspost/reedsolomon v1.9.3/go.mod h1:CwCi+NUr9pqSVktrkN+Ondf06rkhYZ/pcNv7fu+8Un4=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJKvHsGVBgn1xWYf1NbHQhywc8=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/koron/go-ssdp v0.0.0-20191105050749-2e1c40ed0b5d/go.mod h1:5Ky9EC2xfoUKUor0Hjgi2BJhCSXJfMOFlmyYrVKGQMk=
@@ -602,6 +628,7 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/leanovate/gopter v0.2.8/go.mod h1:gNcbPWNEWRe4lm+bycKqxUYoH5uoVje5SkOJ3uoLer8=
 github.com/leanovate/gopter v0.2.9/go.mod h1:U2L/78B+KVFIx2VmW6onHJQzXtFb+p5y3y2Sh+Jxxv8=
+github.com/leanovate/gopter v0.2.9/go.mod h1:U2L/78B+KVFIx2VmW6onHJQzXtFb+p5y3y2Sh+Jxxv8=
 github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/libp2p/go-addr-util v0.0.1/go.mod h1:4ac6O7n9rIAKB1dnd+s8IbbMXkt+oBpzX4/+RACcnlQ=
 github.com/libp2p/go-addr-util v0.0.2 h1:7cWK5cdA5x72jX0g8iLrQWm5TRJZ6CzGdPEhWj7plWU=
@@ -1044,6 +1071,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/procfs v0.2.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.3.0 h1:Uehi/mxLK0eiUc0H0++5tpMGTexB8wZ598MIgU8VpDM=
 github.com/prometheus/procfs v0.3.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/prometheus/prom2json v1.3.0 h1:BlqrtbT9lLH3ZsOVhXPsHzFrApCTKRifB7gjJuypu6Y=
+github.com/prometheus/prom2json v1.3.0/go.mod h1:rMN7m0ApCowcoDlypBHlkNbp5eJQf/+1isKykIP5ZnM=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/prometheus/tsdb v0.10.0 h1:If5rVCMTp6W2SiRAQFlbpJNgVlgMEd+U2GZckwK38ic=
 github.com/prometheus/tsdb v0.10.0/go.mod h1:oi49uRhEe9dPUTlS3JRZOwJuVi6tmh10QSgwXEyGCt4=
@@ -1119,6 +1148,7 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
 github.com/spf13/viper v1.0.0/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7SrnBM=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
 github.com/spf13/viper v1.7.0/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg=
+github.com/spf13/viper v1.7.0/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg=
 github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc=
 github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q=
 github.com/status-im/keycard-go v0.0.0-20200402102358-957c09536969 h1:Oo2KZNP70KE0+IUJSidPj/BFS/RXNHmKIJOdckzml2E=
@@ -1650,6 +1680,7 @@ gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
+gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo=
 gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q=
 gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4=

shared/clientstats/BUILD.bazel

@@ -1,8 +1,26 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_test")
 load("@prysm//tools/go:def.bzl", "go_library")
 
 go_library(
     name = "go_default_library",
-    srcs = ["types.go"],
+    srcs = [
+        "interfaces.go",
+        "scrapers.go",
+        "types.go",
+        "updaters.go",
+    ],
     importpath = "github.com/prysmaticlabs/prysm/shared/clientstats",
     visibility = ["//visibility:public"],
+    deps = [
+        "@com_github_prometheus_client_model//go:go_default_library",
+        "@com_github_prometheus_prom2json//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["scrapers_test.go"],
+    embed = [":go_default_library"],
+    deps = ["//shared/testutil/assert:go_default_library"],
 )

shared/clientstats/interfaces.go

@@ -0,0 +1,19 @@
+package clientstats
+
+import "io"
+
+// A Scraper polls the data source it has been configured with
+// and interprets the content to produce a client-stats process
+// metric. Scrapers currently exist to produce 'validator' and
+// 'beaconnode' metric types.
+type Scraper interface {
+	Scrape() (io.Reader, error)
+}
+
+// An Updater can take the io.Reader created by Scraper and
+// send it to a data sink for consumption. An Updater is used
+// for instance ot send the scraped data for a beacon-node to
+// a remote client-stats endpoint.
+type Updater interface {
+	Update(io.Reader) error
+}

shared/clientstats/scrapers.go

@@ -0,0 +1,240 @@
+package clientstats
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"time"
+
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/prom2json"
+	log "github.com/sirupsen/logrus"
+)
+
+type beaconNodeScraper struct {
+	url     string
+	tripper http.RoundTripper
+}
+
+func (bc *beaconNodeScraper) Scrape() (io.Reader, error) {
+	log.Infof("Scraping beacon-node at %s", bc.url)
+	pf, err := scrapeProm(bc.url, bc.tripper)
+	if err != nil {
+		return nil, nil
+	}
+
+	bs, err := populateBeaconNodeStats(pf)
+	if err != nil {
+		return nil, err
+	}
+
+	b, err := json.Marshal(bs)
+	return bytes.NewBuffer(b), err
+}
+
+// NewBeaconNodeScraper constructs a Scaper capable of scraping
+// the prometheus endpoint of a beacon-node process and producing
+// the json body for the beaconnode client-stats process type.
+func NewBeaconNodeScraper(promExpoURL string) Scraper {
+	return &beaconNodeScraper{
+		url: promExpoURL,
+	}
+}
+
+type validatorScraper struct {
+	url     string
+	tripper http.RoundTripper
+}
+
+func (vc *validatorScraper) Scrape() (io.Reader, error) {
+	log.Infof("Scraping validator at %s", vc.url)
+	pf, err := scrapeProm(vc.url, vc.tripper)
+	if err != nil {
+		return nil, nil
+	}
+
+	vs, err := populateValidatorStats(pf)
+	if err != nil {
+		return nil, err
+	}
+
+	b, err := json.Marshal(vs)
+	return bytes.NewBuffer(b), err
+}
+
+// NewValidatorScraper constructs a Scaper capable of scraping
+// the prometheus endpoint of a validator process and producing
+// the json body for the validator client-stats process type.
+func NewValidatorScraper(promExpoURL string) Scraper {
+	return &validatorScraper{
+		url: promExpoURL,
+	}
+}
+
+// note on tripper -- under the hood FetchMetricFamilies constructs an http.Client,
+// which, if transport is nil, will just use the DefaultTransport, so we
+// really only bother specifying the transport in tests, otherwise we let
+// the zero-value (which is nil) flow through so that the default transport
+// will be used.
+func scrapeProm(url string, tripper http.RoundTripper) (map[string]*dto.MetricFamily, error) {
+	mfChan := make(chan *dto.MetricFamily)
+	errChan := make(chan error)
+	go func() {
+		// FetchMetricFamilies handles grpc flavored prometheus ez
+		// but at the cost of the awkward channel select loop below
+		err := prom2json.FetchMetricFamilies(url, mfChan, tripper)
+		if err != nil {
+			errChan <- err
+		}
+	}()
+	result := make(map[string]*dto.MetricFamily)
+	// channel select accumulates results from FetchMetricFamilies
+	// unless there is an error.
+	for {
+		select {
+		case fam, chanOpen := <-mfChan:
+			// FetchMetricFamiles will close the channel when done
+			// at which point we want to stop the goroutine
+			if fam == nil && !chanOpen {
+				return result, nil
+			}
+			ptr := fam
+			result[fam.GetName()] = ptr
+		case err := <-errChan:
+			return result, err
+		}
+		if errChan == nil && mfChan == nil {
+			return result, nil
+		}
+	}
+}
+
+type metricMap map[string]*dto.MetricFamily
+
+func (mm metricMap) getFamily(name string) (*dto.MetricFamily, error) {
+	f, ok := mm[name]
+	if !ok {
+		return nil, fmt.Errorf("Scraper did not find metric family %s", name)
+	}
+	return f, nil
+}
+
+var now = time.Now // var hook for tests to overwrite
+var nanosPerMilli = (int64(time.Millisecond) / int64(time.Nanosecond))
+
+func populateAPIMessage(processName string) APIMessage {
+	return APIMessage{
+		Timestamp:   now().UnixNano() / nanosPerMilli,
+		APIVersion:  APIVersion,
+		ProcessName: processName,
+	}
+}
+
+func populateCommonStats(pf metricMap) (CommonStats, error) {
+	cs := CommonStats{}
+	cs.ClientName = ClientName
+	var f *dto.MetricFamily
+	var m *dto.Metric
+	var err error
+
+	f, err = pf.getFamily("process_cpu_seconds_total")
+	if err != nil {
+		return cs, err
+	}
+	m = f.Metric[0]
+	// float64->int64: truncates fractional seconds
+	cs.CPUProcessSecondsTotal = int64(m.Counter.GetValue())
+
+	f, err = pf.getFamily("process_resident_memory_bytes")
+	if err != nil {
+		return cs, err
+	}
+	m = f.Metric[0]
+	cs.MemoryProcessBytes = int64(m.Gauge.GetValue())
+
+	f, err = pf.getFamily("prysm_version")
+	if err != nil {
+		return cs, err
+	}
+	m = f.Metric[0]
+	for _, l := range m.GetLabel() {
+		switch l.GetName() {
+		case "version":
+			cs.ClientVersion = l.GetValue()
+		case "buildDate":
+			buildDate, err := strconv.Atoi(l.GetValue())
+			if err != nil {
+				return cs, fmt.Errorf("error when retrieving buildDate label from the prysm_version metric: %s", err)
+			}
+			cs.ClientBuild = int64(buildDate)
+		}
+	}
+
+	return cs, nil
+}
+
+func populateBeaconNodeStats(pf metricMap) (BeaconNodeStats, error) {
+	var err error
+	bs := BeaconNodeStats{}
+	bs.CommonStats, err = populateCommonStats(pf)
+	if err != nil {
+		return bs, err
+	}
+	bs.APIMessage = populateAPIMessage(BeaconNodeProcessName)
+
+	var f *dto.MetricFamily
+	var m *dto.Metric
+
+	f, err = pf.getFamily("beacon_head_slot")
+	if err != nil {
+		return bs, err
+	}
+	m = f.Metric[0]
+	bs.SyncBeaconHeadSlot = int64(m.Gauge.GetValue())
+
+	f, err = pf.getFamily("beacon_clock_time_slot")
+	if err != nil {
+		return bs, err
+	}
+	m = f.Metric[0]
+	if int64(m.Gauge.GetValue()) == bs.SyncBeaconHeadSlot {
+		bs.SyncEth2Synced = true
+	}
+
+	f, err = pf.getFamily("bcnode_disk_beaconchain_bytes_total")
+	if err != nil {
+		return bs, err
+	}
+	m = f.Metric[0]
+	bs.DiskBeaconchainBytesTotal = int64(m.Gauge.GetValue())
+
+	f, err = pf.getFamily("p2p_peer_count")
+	if err != nil {
+		return bs, err
+	}
+	for _, m := range f.Metric {
+		for _, l := range m.GetLabel() {
+			if l.GetName() == "state" {
+				if l.GetValue() == "Connected" {
+					bs.NetworkPeersConnected = int64(m.Gauge.GetValue())
+				}
+			}
+		}
+	}
+
+	return bs, nil
+}
+
+func populateValidatorStats(pf map[string]*dto.MetricFamily) (ValidatorStats, error) {
+	var err error
+	vs := ValidatorStats{}
+	vs.CommonStats, err = populateCommonStats(pf)
+	if err != nil {
+		return vs, err
+	}
+	vs.APIMessage = populateAPIMessage(ValidatorProcessName)
+	return vs, nil
+}

shared/clientstats/scrapers_test.go

@@ -0,0 +1,167 @@
+package clientstats
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
+)
+
+type mockRT struct {
+	body       string
+	status     string
+	statusCode int
+}
+
+func (rt *mockRT) RoundTrip(req *http.Request) (*http.Response, error) {
+	return &http.Response{
+		Status:     http.StatusText(http.StatusOK),
+		StatusCode: http.StatusOK,
+		Body:       io.NopCloser(strings.NewReader(rt.body)),
+	}, nil
+}
+
+var _ http.RoundTripper = &mockRT{}
+
+func TestBeaconNodeScraper(t *testing.T) {
+	bnScraper := beaconNodeScraper{}
+	bnScraper.tripper = &mockRT{body: prometheusTestBody}
+	r, err := bnScraper.Scrape()
+	assert.NoError(t, err, "Unexpected error calling beaconNodeScraper.Scrape")
+	bs := &BeaconNodeStats{}
+	err = json.NewDecoder(r).Decode(bs)
+	assert.NoError(t, err, "Unexpected error decoding result of beaconNodeScraper.Scrape")
+	// CommonStats
+	assert.Equal(t, int64(225), bs.CPUProcessSecondsTotal)
+	assert.Equal(t, int64(1166630912), bs.MemoryProcessBytes)
+	assert.Equal(t, int64(1619586241), bs.ClientBuild)
+	assert.Equal(t, "v1.3.8-hotfix+6c0942", bs.ClientVersion)
+	assert.Equal(t, "prysm", bs.ClientName)
+
+	// BeaconNodeStats
+	assert.Equal(t, int64(256552), bs.SyncBeaconHeadSlot)
+	assert.Equal(t, true, bs.SyncEth2Synced)
+	assert.Equal(t, int64(7365341184), bs.DiskBeaconchainBytesTotal)
+	assert.Equal(t, int64(37), bs.NetworkPeersConnected)
+}
+
+func TestFalseEth2Synced(t *testing.T) {
+	bnScraper := beaconNodeScraper{}
+	eth2NotSynced := strings.Replace(prometheusTestBody, "beacon_head_slot 256552", "beacon_head_slot 256559", 1)
+	bnScraper.tripper = &mockRT{body: eth2NotSynced}
+	r, err := bnScraper.Scrape()
+	assert.NoError(t, err, "Unexpected error calling beaconNodeScraper.Scrape")
+
+	bs := &BeaconNodeStats{}
+	err = json.NewDecoder(r).Decode(bs)
+	assert.NoError(t, err, "Unexpected error decoding result of beaconNodeScraper.Scrape")
+
+	assert.Equal(t, false, bs.SyncEth2Synced)
+}
+
+func TestValidatorScraper(t *testing.T) {
+	vScraper := validatorScraper{}
+	vScraper.tripper = &mockRT{body: prometheusTestBody}
+	r, err := vScraper.Scrape()
+	assert.NoError(t, err, "Unexpected error calling validatorScraper.Scrape")
+	vs := &ValidatorStats{}
+	err = json.NewDecoder(r).Decode(vs)
+	assert.NoError(t, err, "Unexpected error decoding result of validatorScraper.Scrape")
+	// CommonStats
+	assert.Equal(t, int64(225), vs.CPUProcessSecondsTotal)
+	assert.Equal(t, int64(1166630912), vs.MemoryProcessBytes)
+	assert.Equal(t, int64(1619586241), vs.ClientBuild)
+	assert.Equal(t, "v1.3.8-hotfix+6c0942", vs.ClientVersion)
+	assert.Equal(t, "prysm", vs.ClientName)
+}
+
+func mockNowFunc(fixedTime time.Time) func() time.Time {
+	return func() time.Time {
+		return fixedTime
+	}
+}
+
+func TestValidatorAPIMessageDefaults(t *testing.T) {
+	now = mockNowFunc(time.Unix(1619811114, 123456789))
+	// 1+e6 ns per ms, so 123456789 ns rounded down should be 123 ms
+	nowMillis := int64(1619811114123)
+	vScraper := validatorScraper{}
+	vScraper.tripper = &mockRT{body: prometheusTestBody}
+	r, err := vScraper.Scrape()
+	assert.NoError(t, err, "unexpected error from validatorScraper.Scrape()")
+
+	vs := &ValidatorStats{}
+	err = json.NewDecoder(r).Decode(vs)
+	assert.NoError(t, err, "Unexpected error decoding result of validatorScraper.Scrape")
+
+	// CommonStats
+	assert.Equal(t, nowMillis, vs.Timestamp, "Unexpected 'timestamp' in client-stats APIMessage struct")
+	assert.Equal(t, APIVersion, vs.APIVersion, "Unexpected 'version' in client-stats APIMessage struct")
+	assert.Equal(t, ValidatorProcessName, vs.ProcessName, "Unexpected value for 'process' in client-stats APIMessage struct")
+}
+
+func TestBeaconNodeAPIMessageDefaults(t *testing.T) {
+	now = mockNowFunc(time.Unix(1619811114, 123456789))
+	// 1+e6 ns per ms, so 123456789 ns rounded down should be 123 ms
+	nowMillis := int64(1619811114123)
+	bScraper := beaconNodeScraper{}
+	bScraper.tripper = &mockRT{body: prometheusTestBody}
+	r, err := bScraper.Scrape()
+	assert.NoError(t, err, "unexpected error from beaconNodeScraper.Scrape()")
+
+	vs := &BeaconNodeStats{}
+	err = json.NewDecoder(r).Decode(vs)
+	assert.NoError(t, err, "Unexpected error decoding result of beaconNodeScraper.Scrape")
+
+	// CommonStats
+	assert.Equal(t, nowMillis, vs.Timestamp, "Unexpected 'timestamp' in client-stats APIMessage struct")
+	assert.Equal(t, APIVersion, vs.APIVersion, "Unexpected 'version' in client-stats APIMessage struct")
+	assert.Equal(t, BeaconNodeProcessName, vs.ProcessName, "Unexpected value for 'process' in client-stats APIMessage struct")
+}
+
+func TestBadInput(t *testing.T) {
+	bnScraper := beaconNodeScraper{}
+	bnScraper.tripper = &mockRT{body: ""}
+	_, err := bnScraper.Scrape()
+	assert.ErrorContains(t, "did not find metric family", err, "Expected errors for missing metric families on empty input.")
+}
+
+var prometheusTestBody = `
+# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds.
+# TYPE process_cpu_seconds_total counter
+process_cpu_seconds_total 225.09
+# HELP process_resident_memory_bytes Resident memory size in bytes.
+# TYPE process_resident_memory_bytes gauge
+process_resident_memory_bytes 1.166630912e+09
+# HELP prysm_version
+# TYPE prysm_version gauge
+prysm_version{buildDate="1619586241",commit="51eb1540fa838cdbe467bbeb0e36ee667d449377",version="v1.3.8-hotfix+6c0942"} 1
+# HELP validator_count The total number of validators
+# TYPE validator_count gauge
+validator_count{state="Active"} 210301
+validator_count{state="Exited"} 10
+validator_count{state="Exiting"} 0
+validator_count{state="Pending"} 0
+validator_count{state="Slashed"} 0
+validator_count{state="Slashing"} 0
+# HELP beacon_head_slot Slot of the head block of the beacon chain
+# TYPE beacon_head_slot gauge
+beacon_head_slot 256552
+# HELP beacon_clock_time_slot The current slot based on the genesis time and current clock
+# TYPE beacon_clock_time_slot gauge
+beacon_clock_time_slot 256552
+# HELP bcnode_disk_beaconchain_bytes_total Total hard disk space used by the beaconchain database, in bytes. May include mmap.
+# TYPE bcnode_disk_beaconchain_bytes_total gauge
+bcnode_disk_beaconchain_bytes_total 7.365341184e+09
+# HELP p2p_peer_count The number of peers in a given state.
+# TYPE p2p_peer_count gauge
+p2p_peer_count{state="Bad"} 1
+p2p_peer_count{state="Connected"} 37
+p2p_peer_count{state="Connecting"} 0
+p2p_peer_count{state="Disconnected"} 62
+p2p_peer_count{state="Disconnecting"} 0
+`

shared/clientstats/types.go

@@ -1,5 +1,23 @@
 package clientstats
 
+const (
+	ClientName            = "prysm"
+	BeaconNodeProcessName = "beaconnode"
+	ValidatorProcessName  = "validator"
+	APIVersion            = 1
+)
+
+// APIMessage are common to all requests to the client-stats API
+// Note that there is a "system" type that we do not currently
+// support -- if we did APIMessage would be present on the system
+// messages as well as validator and beaconnode, whereas
+// CommonStats would only be part of beaconnode and validator.
+type APIMessage struct {
+	APIVersion  int    `json:"version"`
+	Timestamp   int64  `json:"timestamp"` // unix timestamp in milliseconds
+	ProcessName string `json:"process"`   // validator, beaconnode, system
+}
+
 // CommonStats represent generic metrics that are expected on both
 // beaconnode and validator metric types. This type is used for
 // marshaling metrics to the POST body sent to the metrics collcetor.
@@ -19,6 +37,7 @@ type CommonStats struct {
 	// This is different from a "fallback" configuration where
 	// the second address is treated as a failover.
 	SyncEth2FallbackConnected bool `json:"sync_eth2_fallback_connected"`
+	APIMessage                `json:",inline"`
 }
 
 // BeaconNodeStats embeds CommonStats and represents metrics specific to

shared/clientstats/updaters.go

@@ -0,0 +1,58 @@
+package clientstats
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+type genericWriter struct {
+	io.Writer
+}
+
+func (gw *genericWriter) Update(r io.Reader) error {
+	_, err := io.Copy(gw, r)
+	return err
+}
+
+// NewGenericClientStatsUpdater can Update any io.Writer.
+// It is used by the cli to write to stdout when an http endpoint
+// is not provided. The output could be piped into another program
+// or used for debugging.
+func NewGenericClientStatsUpdater(w io.Writer) Updater {
+	return &genericWriter{w}
+}
+
+type httpPoster struct {
+	url    string
+	client *http.Client
+}
+
+func (gw *httpPoster) Update(r io.Reader) error {
+	resp, err := gw.client.Post(gw.url, "application/json", r)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			return
+		}
+	}()
+	if resp.StatusCode != http.StatusOK {
+		buf := new(bytes.Buffer)
+		_, err = io.Copy(buf, resp.Body)
+		if err != nil {
+			return fmt.Errorf("error reading response body for non-200 response status code (%d), err=%s", resp.StatusCode, err)
+		}
+		return fmt.Errorf("non-200 response status code (%d). response body=%s", resp.StatusCode, buf.String())
+	}
+
+	return nil
+}
+
+// NewClientStatsHTTPPostUpdater is used when the update endpoint
+// is reachable via an HTTP POST request.
+func NewClientStatsHTTPPostUpdater(u string) Updater {
+	return &httpPoster{url: u, client: http.DefaultClient}
+}