From 8000528ebde8752c14efc4369a9540219e4d4014 Mon Sep 17 00:00:00 2001
From: Meysam <meysam@meysam.io>
Date: Sat, 5 Sep 2026 13:48:26 +0700
Subject: [PATCH] feat(prometheus.receive_http): accept Pushgateway format (#1)

---
 .../prometheus/prometheus.receive_http.md     |  76 ++-
 example/.gitignore                            |   3 +-
 example/pushgateway/README.md                 |  20 +
 example/pushgateway/config.alloy              |  15 +
 example/pushgateway/docker-compose.yml        |   8 +
 .../prometheus/receive_http/pushgateway.go    | 268 +++++++++++
 .../receive_http/pushgateway_test.go          | 455 ++++++++++++++++++
 .../prometheus/receive_http/receive_http.go   |   7 +-
 8 files changed, 844 insertions(+), 8 deletions(-)
 create mode 100644 example/pushgateway/README.md
 create mode 100644 example/pushgateway/config.alloy
 create mode 100644 example/pushgateway/docker-compose.yml
 create mode 100644 internal/component/prometheus/receive_http/pushgateway.go
 create mode 100644 internal/component/prometheus/receive_http/pushgateway_test.go

diff --git a/docs/sources/reference/components/prometheus/prometheus.receive_http.md b/docs/sources/reference/components/prometheus/prometheus.receive_http.md
index 346080ab35d..428712068d0 100644
--- a/docs/sources/reference/components/prometheus/prometheus.receive_http.md
+++ b/docs/sources/reference/components/prometheus/prometheus.receive_http.md
@@ -14,11 +14,15 @@ title: prometheus.receive_http
 
 `prometheus.receive_http` listens for HTTP requests containing Prometheus metric samples and forwards them to other components capable of receiving metrics.
 
-The HTTP API exposed is compatible with [Prometheus `remote_write` API][prometheus-remote-write-docs].
-This means that other [`prometheus.remote_write`][prometheus.remote_write] components can be used as a client and send requests to `prometheus.receive_http` which enables using {{< param "PRODUCT_NAME" >}} as a proxy for Prometheus metrics.
+The component exposes two HTTP APIs:
+
+* The [Prometheus `remote_write` API][prometheus-remote-write-docs], so that other [`prometheus.remote_write`][prometheus.remote_write] components can be used as a client and send requests to `prometheus.receive_http`.
+  This enables using {{< param "PRODUCT_NAME" >}} as a proxy for Prometheus metrics.
+* The [Pushgateway push API][pushgateway-docs], so that short-lived jobs can push metrics with the Prometheus client library or tooling they already use.
 
 [prometheus.remote_write]: ../prometheus.remote_write/
 [prometheus-remote-write-docs]: https://prometheus.io/docs/prometheus/latest/querying/api/#remote-write-receiver
+[pushgateway-docs]: https://github.com/prometheus/pushgateway#api
 
 ## Usage
 
@@ -32,11 +36,16 @@ prometheus.receive_http "<LABEL>" {
 }
 ```
 
-The component starts an HTTP server supporting the following endpoint:
+The component starts an HTTP server supporting the following endpoints.
+Both endpoints forward the metrics they receive to the receivers configured in the `forward_to` argument.
 
-* `POST /api/v1/metrics/write`: Sends metrics to the component, which in turn is forwarded to the receivers as configured in `forward_to` argument.
+* `POST /api/v1/metrics/write`: Sends metrics to the component.
   The request format must be compatible with the [Prometheus remote_write API][prometheus-remote-write-docs] and can use either the v1 or v2 format.
   One way to send valid requests to this component is to use another {{< param "PRODUCT_NAME" >}} with a [`prometheus.remote_write`][prometheus.remote_write] component.
+* `POST|PUT /metrics/job/<JOB_NAME>{/<LABEL_NAME>/<LABEL_VALUE>}`: Pushes metrics to the component.
+  The request body must be in the Prometheus text, OpenMetrics, or protobuf exposition format, selected with the `Content-Type` header.
+  {{< param "PRODUCT_NAME" >}} assumes the Prometheus text format when the header is missing.
+  Refer to [Push metrics](#push-metrics) for details.
 
 ## Arguments
 
@@ -51,7 +60,7 @@ You can use the following arguments with `prometheus.receive_http`:
 
 > **EXPERIMENTAL**: The `append_metadata`, `enable_type_and_unit_labels`, and using `"io.prometheus.write.v2.Request"` in `accepted_remote_write_protobuf_messages` are [experimental][] features.
 >
-> The `append_metadata` and `enable_type_and_unit_labels` arguments only apply to remote write v2 payloads and only when metadata is included in those payloads.
+> The `append_metadata` and `enable_type_and_unit_labels` arguments only apply to pushed exposition payloads and remote write v2 payloads, and only when metadata is included in those payloads.
 > Enabling support for remote write v2 payloads requires that `"io.prometheus.write.v2.Request"` is included in `accepted_remote_write_protobuf_messages`.
 > Remote write v1 payloads (`accepted_remote_write_protobuf_messages = ["prometheus.WriteRequest"]`) cannot support these features.
 >
@@ -107,6 +116,33 @@ The metrics include labels such as `status_code` where relevant, which can be us
 * `prometheus_receive_http_response_message_bytes` (histogram): Size (in bytes) of messages sent in response.
 * `prometheus_receive_http_tcp_connections` (gauge): Current number of accepted TCP connections.
 
+## Push metrics
+
+The `POST|PUT /metrics/job/<JOB_NAME>{/<LABEL_NAME>/<LABEL_VALUE>}` endpoint accepts the same requests as [Pushgateway][pushgateway-docs].
+Short-lived jobs that can't be scraped can push their metrics to it with the Prometheus client library or the tooling they already use.
+
+The following example pushes a metric to a component listening on port `9999`:
+
+```shell
+echo 'job_last_success_timestamp 1700000000' | \
+  curl --data-binary @- http://localhost:9999/metrics/job/demo_batch/instance/worker-1
+```
+
+The path following `/metrics/` sets the grouping labels {{< param "PRODUCT_NAME" >}} adds to every series in the request body.
+The `job` label is required and must come first.
+These labels replace any label of the same name in the request body.
+Add a `@base64` suffix to a label name to encode its value with the [base64url][] alphabet, which is how you push a value that contains a `/`.
+
+Unlike Pushgateway, {{< param "PRODUCT_NAME" >}} doesn't store the metrics it receives.
+It forwards them to the receivers in `forward_to` the same way it forwards a remote write request, which has the following consequences:
+
+* {{< param "PRODUCT_NAME" >}} doesn't expose the pushed metrics for scraping, and doesn't repeat a sample until the job pushes it again.
+  Push at the interval you want the samples to appear at.
+* {{< param "PRODUCT_NAME" >}} keeps the timestamp of the samples that carry one, and stamps the other samples with the time the request arrived.
+* {{< param "PRODUCT_NAME" >}} doesn't group the metrics, so `POST` and `PUT` do the same thing, and the `DELETE` method of the Pushgateway API isn't supported and responds with `405 Method Not Allowed`.
+
+[base64url]: https://datatracker.ietf.org/doc/html/rfc4648#section-5
+
 ## Example
 
 ### Receive metrics over HTTP
@@ -137,6 +173,33 @@ prometheus.remote_write "local" {
 }
 ```
 
+### Receive metrics from a batch job
+
+The following example creates a `prometheus.receive_http` component that a batch job can push to.
+
+```alloy
+prometheus.receive_http "pushes" {
+  http {
+    listen_address = "0.0.0.0"
+    listen_port = 9999
+  }
+  forward_to = [prometheus.remote_write.local.receiver]
+}
+
+prometheus.remote_write "local" {
+  endpoint {
+    url = "http://mimir:9009/api/v1/push"
+  }
+}
+```
+
+The job pushes the timestamp of its last successful run when it finishes:
+
+```shell
+printf 'job_last_success_timestamp %s\n' "$(date +%s)" \
+  | curl --data-binary @- http://alloy:9999/metrics/job/demo_batch/instance/worker-1
+```
+
 ### Proxy metrics
 
 To send metrics to the `prometheus.receive_http` component defined in the previous example, another {{< param "PRODUCT_NAME" >}} can run with the following configuration:
@@ -161,7 +224,8 @@ prometheus.remote_write "local" {
 
 ## Technical details
 
-`prometheus.receive_http` uses [snappy](<https://en.wikipedia.org/wiki/Snappy_(compression)>) for compression.
+`prometheus.receive_http` uses [snappy](<https://en.wikipedia.org/wiki/Snappy_(compression)>) for compression of remote write requests.
+Requests to the push endpoint aren't compressed.
 
 <!-- START GENERATED COMPATIBLE COMPONENTS -->
 
diff --git a/example/.gitignore b/example/.gitignore
index 08f5eb8b6df..27eb946822e 100644
--- a/example/.gitignore
+++ b/example/.gitignore
@@ -1,2 +1,3 @@
 # Add here any additional datasources for testing
-docker-compose/config/grafana/datasources/datasource-dev.yml
\ No newline at end of file
+docker-compose/config/grafana/datasources/datasource-dev.yml
+pushgateway/data/
diff --git a/example/pushgateway/README.md b/example/pushgateway/README.md
new file mode 100644
index 00000000000..c8080dd8ffc
--- /dev/null
+++ b/example/pushgateway/README.md
@@ -0,0 +1,20 @@
+# Pushgateway endpoint example
+
+Start the remote storage:
+
+    docker compose -f example/pushgateway/docker-compose.yml up -d
+
+Start Alloy (build first: `cd collector && go build -tags gore2regex -o ../build/alloy . && cd ..`):
+
+    ./build/alloy run --storage.path=./example/pushgateway/data --server.http.listen-addr=127.0.0.1:12345 example/pushgateway/config.alloy
+
+Push a metric:
+
+    echo 'job_last_success_timestamp 1700000000' \
+      | curl --data-binary @- http://127.0.0.1:9999/metrics/job/demo_batch/instance/worker-1
+
+Read it back:
+
+    curl -sG http://127.0.0.1:8428/api/v1/query --data-urlencode 'query=job_last_success_timestamp'
+
+Alloy UI: http://127.0.0.1:12345 — vmsingle UI: http://127.0.0.1:8428/vmui
diff --git a/example/pushgateway/config.alloy b/example/pushgateway/config.alloy
new file mode 100644
index 00000000000..9f55b687154
--- /dev/null
+++ b/example/pushgateway/config.alloy
@@ -0,0 +1,15 @@
+// Accepts remote write on /api/v1/metrics/write and Pushgateway-format
+// pushes on /metrics/job/<JOB_NAME>{/<LABEL_NAME>/<LABEL_VALUE>}.
+prometheus.receive_http "pushes" {
+  http {
+    listen_address = "127.0.0.1"
+    listen_port    = 9999
+  }
+  forward_to = [prometheus.remote_write.vmsingle.receiver]
+}
+
+prometheus.remote_write "vmsingle" {
+  endpoint {
+    url = "http://127.0.0.1:8428/api/v1/write"
+  }
+}
diff --git a/example/pushgateway/docker-compose.yml b/example/pushgateway/docker-compose.yml
new file mode 100644
index 00000000000..b5e24d40551
--- /dev/null
+++ b/example/pushgateway/docker-compose.yml
@@ -0,0 +1,8 @@
+services:
+  vmsingle:
+    image: victoriametrics/victoria-metrics:v1.151.0-scratch
+    network_mode: host
+    command:
+      - --httpListenAddr=:8428
+      - --retentionPeriod=1d
+      - --selfScrapeInterval=0s
diff --git a/internal/component/prometheus/receive_http/pushgateway.go b/internal/component/prometheus/receive_http/pushgateway.go
new file mode 100644
index 00000000000..0468398bbf8
--- /dev/null
+++ b/internal/component/prometheus/receive_http/pushgateway.go
@@ -0,0 +1,268 @@
+package receive_http
+
+import (
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/exemplar"
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/metadata"
+	"github.com/prometheus/prometheus/model/textparse"
+	"github.com/prometheus/prometheus/model/timestamp"
+	"github.com/prometheus/prometheus/storage"
+)
+
+const (
+	// pushPathVar is the mux variable holding the grouping labels of a push,
+	// which is everything following the /metrics/ prefix of the request path.
+	pushPathVar = "labels"
+
+	// base64Suffix marks the label value following a label name in the request
+	// path as base64 encoded.
+	base64Suffix = "@base64"
+
+	// fallbackContentType is used for pushes that don't declare a supported
+	// Content-Type. Pushgateway assumes the Prometheus text format in that case.
+	fallbackContentType = "text/plain"
+)
+
+// handlePush forwards metrics pushed in a Prometheus exposition format to the
+// configured receivers. The request path carries the grouping labels the same
+// way the Pushgateway API does, so Prometheus client libraries can push to this
+// component without changes.
+//
+// Unlike Pushgateway, nothing is stored: a push is forwarded as-is, the same
+// way a remote write request is.
+func (c *Component) handlePush(w http.ResponseWriter, r *http.Request) {
+	groupingLabels, err := parseGroupingLabels(mux.Vars(r)[pushPathVar])
+	if err != nil {
+		c.failPush(w, http.StatusBadRequest, err)
+		return
+	}
+
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		c.failPush(w, http.StatusBadRequest, fmt.Errorf("failed to read request body: %w", err))
+		return
+	}
+
+	c.updateMut.RLock()
+	appendMetadata := c.args.AppendMetadata
+	enableTypeAndUnitLabels := c.args.EnableTypeAndUnitLabels
+	c.updateMut.RUnlock()
+
+	// A Content-Type that isn't an exposition format falls back to the text
+	// format instead of failing, the same way Pushgateway does, because clients
+	// such as `curl --data-binary` send one that says nothing about the body.
+	// The error is then only informational, so a nil parser is the fatal case.
+	parser, err := textparse.New(body, r.Header.Get("Content-Type"), nil, textparse.ParserOptions{
+		EnableTypeAndUnitLabels: enableTypeAndUnitLabels,
+		FallbackContentType:     fallbackContentType,
+	})
+	if parser == nil {
+		c.failPush(w, http.StatusUnsupportedMediaType, fmt.Errorf("unsupported Content-Type %q: %v", r.Header.Get("Content-Type"), err))
+		return
+	}
+
+	app := c.fanout.Appender(r.Context())
+	status, err := c.appendExposition(app, parser, groupingLabels, timestamp.FromTime(time.Now()), appendMetadata)
+	if err != nil {
+		if rollbackErr := app.Rollback(); rollbackErr != nil {
+			c.opts.Logger.Warn("failed to roll back pushed metrics", "err", rollbackErr)
+		}
+		c.failPush(w, status, err)
+		return
+	}
+
+	if err := app.Commit(); err != nil {
+		c.failPush(w, http.StatusInternalServerError, fmt.Errorf("failed to forward pushed metrics: %w", err))
+		return
+	}
+}
+
+// appendExposition appends every sample of an exposition payload to app,
+// overriding the labels of each series with groupingLabels. Samples that carry
+// no timestamp of their own are stamped with defaultTS. It returns the HTTP
+// status to respond with when appending fails.
+func (c *Component) appendExposition(
+	app storage.Appender,
+	parser textparse.Parser,
+	groupingLabels labels.Labels,
+	defaultTS int64,
+	appendMetadata bool,
+) (int, error) {
+
+	var (
+		lset    labels.Labels
+		builder = labels.NewBuilder(labels.EmptyLabels())
+		ex      exemplar.Exemplar
+
+		// Help, type, and unit are exposed one at a time and always precede the
+		// series of the metric family they describe.
+		familyName string
+		familyMeta metadata.Metadata
+	)
+
+	for {
+		entry, err := parser.Next()
+		if errors.Is(err, io.EOF) {
+			return http.StatusOK, nil
+		}
+		if err != nil {
+			return http.StatusBadRequest, fmt.Errorf("failed to parse pushed metrics: %w", err)
+		}
+
+		var (
+			isHistogram bool
+			parsedTS    *int64
+			val         float64
+			h           *histogram.Histogram
+			fh          *histogram.FloatHistogram
+		)
+
+		switch entry {
+		case textparse.EntrySeries:
+			_, parsedTS, val = parser.Series()
+		case textparse.EntryHistogram:
+			isHistogram = true
+			_, parsedTS, h, fh = parser.Histogram()
+		case textparse.EntryHelp:
+			name, help := parser.Help()
+			familyName, familyMeta = trackFamily(familyName, string(name), familyMeta)
+			familyMeta.Help = string(help)
+			continue
+		case textparse.EntryType:
+			name, typ := parser.Type()
+			familyName, familyMeta = trackFamily(familyName, string(name), familyMeta)
+			familyMeta.Type = typ
+			continue
+		case textparse.EntryUnit:
+			name, unit := parser.Unit()
+			familyName, familyMeta = trackFamily(familyName, string(name), familyMeta)
+			familyMeta.Unit = string(unit)
+			continue
+		default: // Comments and any entry a future parser may add.
+			continue
+		}
+
+		parser.Labels(&lset)
+		builder.Reset(lset)
+		groupingLabels.Range(func(l labels.Label) { builder.Set(l.Name, l.Value) })
+		lset = builder.Labels()
+
+		ts := defaultTS
+		if parsedTS != nil {
+			ts = *parsedTS
+		}
+
+		var ref storage.SeriesRef
+		if isHistogram {
+			ref, err = app.AppendHistogram(0, lset, ts, h, fh)
+		} else {
+			ref, err = app.Append(0, lset, ts, val)
+		}
+		if err != nil {
+			return http.StatusInternalServerError, fmt.Errorf("failed to append pushed series %s: %w", lset.String(), err)
+		}
+
+		for parser.Exemplar(&ex) {
+			if !ex.HasTs {
+				ex.Ts = ts
+			}
+			// Exemplars are best effort: a rejected one must not fail the push.
+			if _, err := app.AppendExemplar(ref, lset, ex); err != nil {
+				c.opts.Logger.Debug("failed to append exemplar of pushed series", "series", lset.String(), "err", err)
+			}
+			ex = exemplar.Exemplar{}
+		}
+
+		if appendMetadata && isSeriesOfFamily(lset, familyName) {
+			if _, err := app.UpdateMetadata(ref, lset, familyMeta); err != nil {
+				c.opts.Logger.Debug("failed to append metadata of pushed series", "series", lset.String(), "err", err)
+			}
+		}
+	}
+}
+
+// failPush responds with an error and logs it. Rejected pushes are logged at
+// debug level to keep a misbehaving client from flooding the logs.
+func (c *Component) failPush(w http.ResponseWriter, status int, err error) {
+	http.Error(w, err.Error(), status)
+	if status >= http.StatusInternalServerError {
+		c.opts.Logger.Error("failed to ingest pushed metrics", "err", err)
+	} else {
+		c.opts.Logger.Debug("rejected pushed metrics", "status", status, "err", err)
+	}
+}
+
+// trackFamily returns the metadata to accumulate the next help, type, or unit
+// into, starting over whenever the parser moves on to a new metric family.
+func trackFamily(current, next string, meta metadata.Metadata) (string, metadata.Metadata) {
+	if current == next {
+		return current, meta
+	}
+	return next, metadata.Metadata{Type: model.MetricTypeUnknown}
+}
+
+// isSeriesOfFamily reports whether the metadata of the metric family belongs to
+// the series. The suffixed series of a classic histogram or summary, such as
+// _bucket or _count, share the metadata of their family.
+func isSeriesOfFamily(lset labels.Labels, familyName string) bool {
+	if familyName == "" {
+		return false
+	}
+	suffix, ok := strings.CutPrefix(lset.Get(model.MetricNameLabel), familyName)
+	return ok && (suffix == "" || strings.HasPrefix(suffix, "_"))
+}
+
+// parseGroupingLabels turns the request path following /metrics/ into the
+// grouping labels applied to every series of a push. The layout is the one of
+// the Pushgateway API, job/<JOB_NAME>{/<LABEL_NAME>/<LABEL_VALUE>}, where a
+// label name carrying a @base64 suffix marks its value as base64 encoded.
+func parseGroupingLabels(path string) (labels.Labels, error) {
+	components := strings.Split(path, "/")
+	if len(components) < 2 || len(components)%2 != 0 {
+		return labels.EmptyLabels(), fmt.Errorf("invalid path %q: expected /metrics/job/<JOB_NAME>{/<LABEL_NAME>/<LABEL_VALUE>}", path)
+	}
+	if strings.TrimSuffix(components[0], base64Suffix) != model.JobLabel {
+		return labels.EmptyLabels(), fmt.Errorf("invalid path %q: the first label must be %q", path, model.JobLabel)
+	}
+
+	builder := labels.NewBuilder(labels.EmptyLabels())
+	for i := 0; i < len(components); i += 2 {
+		name, isBase64 := strings.CutSuffix(components[i], base64Suffix)
+		// Prometheus validates label names as UTF-8 since v3, so the only names
+		// left to reject are the ones reserved for internal use.
+		if !model.UTF8Validation.IsValidLabelName(name) || strings.HasPrefix(name, model.ReservedLabelPrefix) {
+			return labels.EmptyLabels(), fmt.Errorf("invalid label name %q in path %q", name, path)
+		}
+
+		value := components[i+1]
+		if isBase64 {
+			// Padding is optional, and a lone padding character encodes the
+			// empty string.
+			decoded, err := base64.RawURLEncoding.DecodeString(strings.TrimRight(value, "="))
+			if err != nil {
+				return labels.EmptyLabels(), fmt.Errorf("invalid base64 value of label %q in path %q: %w", name, path, err)
+			}
+			value = string(decoded)
+		}
+		// Setting an empty value drops the label, matching the Prometheus
+		// convention that an empty label is an absent one.
+		builder.Set(name, value)
+	}
+
+	grouping := builder.Labels()
+	if grouping.Get(model.JobLabel) == "" {
+		return labels.EmptyLabels(), fmt.Errorf("invalid path %q: the job name must not be empty", path)
+	}
+	return grouping, nil
+}
diff --git a/internal/component/prometheus/receive_http/pushgateway_test.go b/internal/component/prometheus/receive_http/pushgateway_test.go
new file mode 100644
index 00000000000..56f514b76c0
--- /dev/null
+++ b/internal/component/prometheus/receive_http/pushgateway_test.go
@@ -0,0 +1,455 @@
+package receive_http
+
+import (
+	"context"
+	"net/http"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/exp/api/remote"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/push"
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/stretchr/testify/require"
+
+	fnet "github.com/grafana/alloy/internal/component/common/net"
+	alloyprom "github.com/grafana/alloy/internal/component/prometheus"
+)
+
+const (
+	textContentType  = "text/plain"
+	protoContentType = "application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; encoding=delimited"
+)
+
+func TestParseGroupingLabels(t *testing.T) {
+	tests := []struct {
+		name     string
+		path     string
+		expected labels.Labels
+		errorMsg string
+	}{
+		{
+			name:     "job only",
+			path:     "job/demo_batch",
+			expected: labels.FromStrings("job", "demo_batch"),
+		},
+		{
+			name:     "job and grouping labels",
+			path:     "job/demo_batch/instance/worker-1/zone/eu",
+			expected: labels.FromStrings("instance", "worker-1", "job", "demo_batch", "zone", "eu"),
+		},
+		{
+			name:     "trailing slash of a complete pair",
+			path:     "job/demo_batch/",
+			errorMsg: "expected /metrics/job/<JOB_NAME>",
+		},
+		{
+			name:     "base64 encoded job",
+			path:     "job@base64/ZGVtby9iYXRjaA",
+			expected: labels.FromStrings("job", "demo/batch"),
+		},
+		{
+			name:     "base64 encoded value with padding",
+			path:     "job/demo_batch/instance@base64/d29ya2VyLzE=",
+			expected: labels.FromStrings("instance", "worker/1", "job", "demo_batch"),
+		},
+		{
+			name:     "base64 encoded empty value is dropped",
+			path:     "job/demo_batch/instance@base64/=",
+			expected: labels.FromStrings("job", "demo_batch"),
+		},
+		{
+			name:     "empty value is dropped",
+			path:     "job/demo_batch/instance/",
+			expected: labels.FromStrings("job", "demo_batch"),
+		},
+		{
+			name:     "last value of a repeated label wins",
+			path:     "job/demo_batch/zone/eu/zone/us",
+			expected: labels.FromStrings("job", "demo_batch", "zone", "us"),
+		},
+		{
+			name:     "empty path",
+			path:     "",
+			errorMsg: "expected /metrics/job/<JOB_NAME>",
+		},
+		{
+			name:     "odd number of components",
+			path:     "job/demo_batch/instance",
+			errorMsg: "expected /metrics/job/<JOB_NAME>",
+		},
+		{
+			name:     "job is not the first label",
+			path:     "instance/worker-1/job/demo_batch",
+			errorMsg: `the first label must be "job"`,
+		},
+		{
+			name:     "empty job name",
+			path:     "job/",
+			errorMsg: "the job name must not be empty",
+		},
+		{
+			name:     "base64 encoded empty job name",
+			path:     "job@base64/=",
+			errorMsg: "the job name must not be empty",
+		},
+		{
+			name:     "reserved label name",
+			path:     "job/demo_batch/__name__/foo",
+			errorMsg: `invalid label name "__name__"`,
+		},
+		{
+			name:     "invalid base64 value",
+			path:     "job/demo_batch/instance@base64/!!!",
+			errorMsg: `invalid base64 value of label "instance"`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual, err := parseGroupingLabels(tt.path)
+			if tt.errorMsg != "" {
+				require.ErrorContains(t, err, tt.errorMsg)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tt.expected, actual)
+		})
+	}
+}
+
+func TestIsSeriesOfFamily(t *testing.T) {
+	tests := []struct {
+		metricName string
+		familyName string
+		expected   bool
+	}{
+		{metricName: "request_duration_seconds", familyName: "request_duration_seconds", expected: true},
+		{metricName: "request_duration_seconds_sum", familyName: "request_duration_seconds", expected: true},
+		{metricName: "request_duration_seconds_bucket", familyName: "request_duration_seconds", expected: true},
+		{metricName: "requests_total", familyName: "request_duration_seconds", expected: false},
+		// A family whose name is a prefix of an unrelated metric name.
+		{metricName: "requestsize", familyName: "requests", expected: false},
+		{metricName: "requests", familyName: "", expected: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.metricName+"/"+tt.familyName, func(t *testing.T) {
+			lset := labels.FromStrings("__name__", tt.metricName)
+			require.Equal(t, tt.expected, isSeriesOfFamily(lset, tt.familyName))
+		})
+	}
+}
+
+func TestPushTextFormat(t *testing.T) {
+	samples := make(chan testSample, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+	body := `
+# HELP job_last_success_timestamp Timestamp of the last successful run.
+# TYPE job_last_success_timestamp gauge
+job_last_success_timestamp 1700000000
+requests_total{code="200"} 3
+requests_total{code="500",job="overridden"} 1
+`
+
+	before := time.Now().UnixMilli()
+	resp := doPush(t, http.MethodPost, addr+"/metrics/job/demo_batch/instance/worker-1", textContentType, body)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	after := time.Now().UnixMilli()
+
+	actual := collectSamples(t, samples, 3)
+	require.Equal(t, []testSample{
+		{ts: actual[0].ts, val: 1700000000, l: labels.FromStrings("__name__", "job_last_success_timestamp", "instance", "worker-1", "job", "demo_batch")},
+		{ts: actual[1].ts, val: 3, l: labels.FromStrings("__name__", "requests_total", "code", "200", "instance", "worker-1", "job", "demo_batch")},
+		{ts: actual[2].ts, val: 1, l: labels.FromStrings("__name__", "requests_total", "code", "500", "instance", "worker-1", "job", "demo_batch")},
+	}, actual)
+
+	// Samples without a timestamp of their own are stamped on arrival.
+	for _, s := range actual {
+		require.GreaterOrEqual(t, s.ts, before)
+		require.LessOrEqual(t, s.ts, after)
+	}
+}
+
+// TestPushFallsBackToTextFormat covers the Pushgateway behavior of parsing a
+// body as the Prometheus text format when the Content-Type says nothing about
+// it, which is what `curl --data-binary` sends.
+func TestPushFallsBackToTextFormat(t *testing.T) {
+	for _, contentType := range []string{"", "application/x-www-form-urlencoded", "application/json"} {
+		t.Run(contentType, func(t *testing.T) {
+			samples := make(chan testSample, 10)
+			addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+			resp := doPush(t, http.MethodPost, addr+"/metrics/job/demo_batch", contentType, "my_metric 42\n")
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+
+			actual := collectSamples(t, samples, 1)
+			require.Equal(t, []testSample{
+				{ts: actual[0].ts, val: 42, l: labels.FromStrings("__name__", "my_metric", "job", "demo_batch")},
+			}, actual)
+		})
+	}
+}
+
+func TestPushHonorsTimestamps(t *testing.T) {
+	samples := make(chan testSample, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+	resp := doPush(t, http.MethodPost, addr+"/metrics/job/demo_batch", textContentType, "my_metric 42 1700000000000\n")
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	require.Equal(t, []testSample{
+		{ts: 1700000000000, val: 42, l: labels.FromStrings("__name__", "my_metric", "job", "demo_batch")},
+	}, collectSamples(t, samples, 1))
+}
+
+// TestPushFromClientLibrary covers a push from an unmodified Prometheus client
+// library, which uses PUT, the protobuf exposition format, and base64 encoded
+// path components.
+func TestPushFromClientLibrary(t *testing.T) {
+	samples := make(chan testSample, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+	registry := prometheus.NewRegistry()
+	duration := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "batch_duration_seconds",
+		Help: "Duration of the last batch run.",
+	})
+	duration.Set(1.5)
+	registry.MustRegister(duration)
+
+	// The job name contains a slash, so the client library base64 encodes it,
+	// and the empty instance is encoded as a lone padding character.
+	err := push.New("http://"+addr, "demo/batch").
+		Grouping("instance", "").
+		Gatherer(registry).
+		Push()
+	require.NoError(t, err)
+
+	actual := collectSamples(t, samples, 1)
+	require.Equal(t, []testSample{
+		{ts: actual[0].ts, val: 1.5, l: labels.FromStrings("__name__", "batch_duration_seconds", "job", "demo/batch")},
+	}, actual)
+}
+
+func TestPushNativeHistogram(t *testing.T) {
+	histograms := make(chan testHistogram, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testHistogramAppendable(histograms)})
+
+	registry := prometheus.NewRegistry()
+	latency := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:                            "request_latency_seconds",
+		Help:                            "Latency of the last requests.",
+		NativeHistogramBucketFactor:     1.1,
+		NativeHistogramMaxBucketNumber:  100,
+		NativeHistogramMinResetDuration: time.Hour,
+	})
+	latency.Observe(0.5)
+	latency.Observe(1.5)
+	registry.MustRegister(latency)
+
+	require.NoError(t, push.New("http://"+addr, "demo_batch").Gatherer(registry).Push())
+
+	select {
+	case actual := <-histograms:
+		require.Equal(t, labels.FromStrings("__name__", "request_latency_seconds", "job", "demo_batch"), actual.l)
+		require.NotNil(t, actual.h)
+		require.NoError(t, actual.h.Validate())
+		require.Equal(t, uint64(2), actual.h.Count)
+		require.Equal(t, 2.0, actual.h.Sum)
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for a native histogram")
+	}
+}
+
+func TestPushMetadata(t *testing.T) {
+	samples := make(chan testSample, 10)
+	metadata := make(chan testMetadata, 10)
+	addr := startPushComponent(t, Arguments{
+		ForwardTo:      testAppendableWithMetadata(samples, metadata),
+		AppendMetadata: true,
+	})
+
+	body := `
+# HELP request_duration_seconds Duration of the requests.
+# TYPE request_duration_seconds summary
+request_duration_seconds_sum 12
+request_duration_seconds_count 3
+`
+	resp := doPush(t, http.MethodPost, addr+"/metrics/job/demo_batch", textContentType, body)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	expected := testMetadata{metricName: "request_duration_seconds", metricType: "summary", help: "Duration of the requests."}
+	for range 2 {
+		select {
+		case actual := <-metadata:
+			// The suffixed series of a summary share the metadata of the family.
+			require.Equal(t, expected.metricType, actual.metricType)
+			require.Equal(t, expected.help, actual.help)
+		case <-time.After(5 * time.Second):
+			t.Fatal("timed out waiting for metadata")
+		}
+	}
+}
+
+func TestPushErrors(t *testing.T) {
+	tests := []struct {
+		name        string
+		method      string
+		path        string
+		contentType string
+		body        string
+		expected    int
+	}{
+		{
+			name:     "invalid path",
+			method:   http.MethodPost,
+			path:     "/metrics/job/demo_batch/instance",
+			body:     "my_metric 1\n",
+			expected: http.StatusBadRequest,
+		},
+		{
+			name:     "missing job name",
+			method:   http.MethodPost,
+			path:     "/metrics/job/",
+			body:     "my_metric 1\n",
+			expected: http.StatusBadRequest,
+		},
+		{
+			name:     "unparsable body",
+			method:   http.MethodPost,
+			path:     "/metrics/job/demo_batch",
+			body:     "this is not an exposition format\n",
+			expected: http.StatusBadRequest,
+		},
+		{
+			name:        "unparsable protobuf body",
+			method:      http.MethodPost,
+			path:        "/metrics/job/demo_batch",
+			contentType: protoContentType,
+			body:        "my_metric 1\n",
+			expected:    http.StatusBadRequest,
+		},
+		{
+			name:     "delete is not supported",
+			method:   http.MethodDelete,
+			path:     "/metrics/job/demo_batch",
+			expected: http.StatusMethodNotAllowed,
+		},
+		{
+			name:     "get is not supported",
+			method:   http.MethodGet,
+			path:     "/metrics/job/demo_batch",
+			expected: http.StatusMethodNotAllowed,
+		},
+	}
+
+	samples := make(chan testSample, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp := doPush(t, tt.method, addr+tt.path, tt.contentType, tt.body)
+			require.Equal(t, tt.expected, resp.StatusCode)
+			require.Empty(t, samples)
+		})
+	}
+}
+
+// TestPushDoesNotBreakRemoteWrite makes sure the push endpoint doesn't shadow
+// the remote write one.
+func TestPushDoesNotBreakRemoteWrite(t *testing.T) {
+	samples := make(chan testSample, 10)
+	addr := startPushComponent(t, Arguments{ForwardTo: testAppendable(samples)})
+
+	resp := doPush(t, http.MethodPost, addr+"/metrics/job/demo_batch", textContentType, "my_metric 1\n")
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	resp = doPush(t, http.MethodPost, addr+"/api/v1/metrics/write", "", "")
+	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+}
+
+func startPushComponent(t *testing.T, args Arguments) string {
+	t.Helper()
+
+	args.Server = &fnet.ServerConfig{
+		HTTP: &fnet.HTTPConfig{ListenAddress: "localhost", ListenPort: 0},
+		GRPC: testGRPCConfig(),
+	}
+	args.AcceptedRemoteWriteProtobufMessages = []string{string(remote.WriteV1MessageType)}
+
+	comp, err := New(testOptions(t), args)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	t.Cleanup(cancel)
+	go func() {
+		_ = comp.Run(ctx)
+	}()
+
+	addr := comp.server.HTTPListenAddr()
+	waitForServerToBeReady(t, addr, nil)
+	return addr
+}
+
+func doPush(t *testing.T, method, url, contentType, body string) *http.Response {
+	t.Helper()
+
+	req, err := http.NewRequestWithContext(t.Context(), method, "http://"+url, strings.NewReader(body))
+	require.NoError(t, err)
+	if contentType != "" {
+		req.Header.Set("Content-Type", contentType)
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = resp.Body.Close() })
+	return resp
+}
+
+// collectSamples waits for count samples and returns them sorted by labels, so
+// that assertions don't depend on the order the exposition was parsed in.
+func collectSamples(t *testing.T, samples chan testSample, count int) []testSample {
+	t.Helper()
+
+	collected := make([]testSample, 0, count)
+	for range count {
+		select {
+		case sample := <-samples:
+			collected = append(collected, sample)
+		case <-time.After(5 * time.Second):
+			t.Fatalf("timed out after %d of %d samples", len(collected), count)
+		}
+	}
+	sort.Slice(collected, func(i, j int) bool { return labels.Compare(collected[i].l, collected[j].l) < 0 })
+
+	require.Empty(t, samples, "unexpected extra samples")
+	return collected
+}
+
+type testHistogram struct {
+	l labels.Labels
+	h *histogram.Histogram
+}
+
+func testHistogramAppendable(actual chan testHistogram) []storage.Appendable {
+	hookFn := func(
+		ref storage.SeriesRef,
+		l labels.Labels,
+		ts int64,
+		h *histogram.Histogram,
+		fh *histogram.FloatHistogram,
+		next storage.Appender,
+	) (storage.SeriesRef, error) {
+
+		actual <- testHistogram{l: l, h: h}
+		return ref, nil
+	}
+
+	return []storage.Appendable{alloyprom.NewInterceptor(nil, alloyprom.WithHistogramHook(hookFn))}
+}
diff --git a/internal/component/prometheus/receive_http/receive_http.go b/internal/component/prometheus/receive_http/receive_http.go
index 0aa9f53ccb8..9e56a141347 100644
--- a/internal/component/prometheus/receive_http/receive_http.go
+++ b/internal/component/prometheus/receive_http/receive_http.go
@@ -159,7 +159,12 @@ func (c *Component) Update(args component.Arguments) error {
 	c.server = s
 
 	err = c.server.MountAndRun(func(router *mux.Router) {
-		router.Path("/api/v1/metrics/write").Methods("POST").Handler(c.handler)
+		router.Path("/api/v1/metrics/write").Methods(http.MethodPost).Handler(c.handler)
+		// The Pushgateway API puts the grouping labels in the path, so everything
+		// below /metrics/ is routed to the push handler and parsed there.
+		router.Path(fmt.Sprintf("/metrics/{%s:.+}", pushPathVar)).
+			Methods(http.MethodPost, http.MethodPut).
+			HandlerFunc(c.handlePush)
 	})
 	if err != nil {
 		return err