Add component metrics to /metrics endpoint

cmd/agent/flow.go

@@ -11,6 +11,8 @@ import (
 	"os/signal"
 	"sync"
 
+	"go.uber.org/atomic"
+
 	"github.com/fatih/color"
 	"github.com/go-kit/log/level"
 	"github.com/gorilla/mux"
@@ -43,6 +45,7 @@ func runFlow() error {
 		httpListenAddr = "127.0.0.1:12345"
 		configFile     string
 		storagePath    = "data-agent/"
+		ready          = atomic.NewBool(true)
 	)
 
 	fs := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
@@ -111,6 +114,15 @@ func runFlow() error {
 		}
 
 		r := mux.NewRouter()
+		r.HandleFunc("/-/ready", func(w http.ResponseWriter, r *http.Request) {
+			if ready.Load() {
+				w.WriteHeader(http.StatusOK)
+				fmt.Fprintf(w, "Agent is Ready.\n")
+			} else {
+				w.WriteHeader(http.StatusServiceUnavailable)
+				fmt.Fprint(w, "Config failed to load.\n")
+			}
+		})
 		r.Handle("/metrics", promhttp.Handler())
 		r.Handle("/debug/config", f.ConfigHandler())
 		r.Handle("/debug/graph", f.GraphHandler())
@@ -119,6 +131,7 @@ func runFlow() error {
 
 		r.HandleFunc("/-/reload", func(w http.ResponseWriter, _ *http.Request) {
 			err := reload()
+			ready.Store(err == nil)
 			if err != nil {
 				http.Error(w, err.Error(), http.StatusBadRequest)
 				return

component/local/file/file.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/go-kit/log/level"
 	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/pkg/flow/rivertypes"
@@ -84,7 +86,8 @@ type Component struct {
 
 	// reloadCh is a buffered channel which is written to when the watched file
 	// should be reloaded by the component.
-	reloadCh chan struct{}
+	reloadCh     chan struct{}
+	lastAccessed prometheus.Gauge
 }
 
 var (
@@ -98,11 +101,19 @@ func New(o component.Options, args Arguments) (*Component, error) {
 		opts: o,
 
 		reloadCh: make(chan struct{}, 1),
+		lastAccessed: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "agent_local_file_timestamp_last_accessed_unix_seconds",
+			Help: "The last successful access in unix seconds",
+		}),
 	}
 
+	err := o.Registerer.Register(c.lastAccessed)
+	if err != nil {
+		return nil, err
+	}
 	// Perform an update which will immediately set our exports to the initial
 	// contents of the file.
-	if err := c.Update(args); err != nil {
+	if err = c.Update(args); err != nil {
 		return nil, err
 	}
 	return c, nil
@@ -162,6 +173,7 @@ func (c *Component) readFile() error {
 		return err
 	}
 	c.latestContent = string(bb)
+	c.lastAccessed.SetToCurrentTime()
 
 	c.opts.OnStateChange(Exports{
 		Content: rivertypes.OptionalSecret{

component/metrics/mutate/mutate.go

@@ -8,6 +8,7 @@ import (
 	fa "github.com/grafana/agent/component/common/appendable"
 	flow_relabel "github.com/grafana/agent/component/common/relabel"
 	"github.com/grafana/agent/component/metrics"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/model/relabel"
 	"github.com/prometheus/prometheus/storage"
 )
@@ -17,7 +18,6 @@ func init() {
 		Name:    "metrics.mutate",
 		Args:    Arguments{},
 		Exports: Exports{},
-
 		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
 			return New(opts, args.(Arguments))
 		},
@@ -44,8 +44,9 @@ type Component struct {
 	opts component.Options
 	mrc  []*relabel.Config
 
-	appendable *fa.FlowAppendable
-	receiver   *metrics.Receiver
+	appendable       *fa.FlowAppendable
+	receiver         *metrics.Receiver
+	metricsProcessed prometheus.Counter
 }
 
 var (
@@ -57,9 +58,17 @@ func New(o component.Options, args Arguments) (*Component, error) {
 	c := &Component{opts: o}
 	c.appendable = fa.NewFlowAppendable(args.ForwardTo...)
 	c.receiver = &metrics.Receiver{Receive: c.Receive}
+	c.metricsProcessed = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "agent_metrics_mutate_metrics_processed",
+		Help: "Total number of metrics processed",
+	})
 
+	err := o.Registerer.Register(c.metricsProcessed)
+	if err != nil {
+		return nil, err
+	}
 	// Call to Update() to set the relabelling rules once at the start.
-	if err := c.Update(args); err != nil {
+	if err = c.Update(args); err != nil {
 		return nil, err
 	}
 
@@ -69,6 +78,7 @@ func New(o component.Options, args Arguments) (*Component, error) {
 // Run implements component.Component.
 func (c *Component) Run(ctx context.Context) error {
 	<-ctx.Done()
+	c.opts.Registerer.Unregister(c.metricsProcessed)
 	return nil
 }
 

component/metrics/remotewrite/remote_write.go

@@ -20,7 +20,6 @@ import (
 	"github.com/grafana/agent/pkg/build"
 	"github.com/grafana/agent/pkg/flow/rivertypes"
 	"github.com/grafana/agent/pkg/metrics/wal"
-	"github.com/prometheus/client_golang/prometheus"
 	common "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/config"
@@ -85,7 +84,6 @@ type BasicAuthConfig struct {
 type Component struct {
 	log  log.Logger
 	opts component.Options
-	reg  *metrics.CollectorRegistry
 
 	walStore    *wal.Storage
 	remoteStore *remote.Storage
@@ -99,23 +97,19 @@ type Component struct {
 
 // NewComponent creates a new metrics_forwarder component.
 func NewComponent(o component.Options, c RemoteConfig) (*Component, error) {
-	reg := metrics.NewCollectorRegistry()
-
 	walLogger := log.With(o.Logger, "subcomponent", "wal")
 	dataPath := filepath.Join(o.DataPath, "wal", o.ID)
-	walStorage, err := wal.NewStorage(walLogger, reg, dataPath)
+	walStorage, err := wal.NewStorage(walLogger, o.Registerer, dataPath)
 	if err != nil {
 		return nil, err
 	}
 
 	remoteLogger := log.With(o.Logger, "subcomponent", "rw")
-	remoteStore := remote.NewStorage(remoteLogger, reg, startTime, dataPath, remoteFlushDeadline, nil)
+	remoteStore := remote.NewStorage(remoteLogger, o.Registerer, startTime, dataPath, remoteFlushDeadline, nil)
 
 	res := &Component{
-		log:  o.Logger,
-		opts: o,
-		reg:  reg,
-
+		log:         o.Logger,
+		opts:        o,
 		walStore:    walStorage,
 		remoteStore: remoteStore,
 		storage:     storage.NewFanout(o.Logger, walStorage, remoteStore),
@@ -278,13 +272,3 @@ func (c *Component) Config() RemoteConfig {
 	defer c.mut.RUnlock()
 	return c.cfg
 }
-
-// Describe implements prometheus.Collector.
-func (c *Component) Describe(ch chan<- *prometheus.Desc) {
-	c.reg.Describe(ch)
-}
-
-// Collect implements prometheus.Collector.
-func (c *Component) Collect(ch chan<- prometheus.Metric) {
-	c.reg.Collect(ch)
-}

component/metrics/scrape/scrape_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/component/metrics"
 	"github.com/grafana/agent/pkg/flow/logging"
@@ -16,7 +18,10 @@ import (
 func TestForwardingToAppendable(t *testing.T) {
 	l, err := logging.New(os.Stderr, logging.DefaultOptions)
 	require.NoError(t, err)
-	opts := component.Options{Logger: l}
+	opts := component.Options{
+		Logger:     l,
+		Registerer: prometheus.NewRegistry(),
+	}
 
 	nilReceivers := []*metrics.Receiver{nil, nil}
 

component/registry.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/grafana/regexp"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // The parsedName of a component is the parts of its name ("remote.http") split
@@ -49,6 +50,9 @@ type Options struct {
 	// by the component; a component must use the same Exports type for its
 	// lifetime.
 	OnStateChange func(e Exports)
+
+	// Registerer allows components to add their own metrics. The register will come pre-wrapped with the component ID. It is not necessary for components to unregister metrics on shutdown.
+	Registerer prometheus.Registerer
 }
 
 // Registration describes a single component.