diff --git a/pkg/prometheus/BUILD b/pkg/prometheus/BUILD
index 5a2734784..95e7133ef 100644
--- a/pkg/prometheus/BUILD
+++ b/pkg/prometheus/BUILD
@@ -6,12 +6,15 @@ go_library(
     name = "prometheus",
     srcs = [
         "prometheus.go",
+        "prometheus_verify.go",
     ],
     visibility = ["//:sandbox"],
+    deps = ["//pkg/metric:metric_go_proto"],
 )
 
 go_test(
     name = "prometheus_test",
+    size = "small",
     srcs = ["prometheus_test.go"],
     library = ":prometheus",
     deps = [
diff --git a/pkg/prometheus/prometheus.go b/pkg/prometheus/prometheus.go
index 725d8b3fd..4df47d5b6 100644
--- a/pkg/prometheus/prometheus.go
+++ b/pkg/prometheus/prometheus.go
@@ -96,6 +96,19 @@ type Number struct {
 	Int int64 `json:"int,omitempty"`
 }
 
+// IsInteger returns whether this number contains an integer value.
+// This is defined as either having the `Float` part set to zero (in which case the `Int` part takes
+// precedence), or having `Float` be a value equal to its own rounding and not a special float.
+func (n *Number) IsInteger() bool {
+	if n.Float == 0 {
+		return true
+	}
+	if math.IsNaN(n.Float) || n.Float == math.Inf(-1) || n.Float == math.Inf(1) {
+		return false
+	}
+	return math.Round(n.Float) == n.Float
+}
+
 // String returns a string representation of this number.
 func (n *Number) String() string {
 	var s strings.Builder
@@ -105,6 +118,26 @@ func (n *Number) String() string {
 	return s.String()
 }
 
+// SameType returns true if `n` and `other` are either both floating-point or both integers.
+// If a `Number` is zero, it is considered of the same type as any other zero `Number`.
+func (n *Number) SameType(other *Number) bool {
+	// Within `n` and `other`, at least one of `Int` or `Float` must be set to zero.
+	// Therefore, this verifies that there is at least one shared zero between the two.
+	return n.Float == other.Float || n.Int == other.Int
+}
+
+// GreaterThan returns true if n > other.
+// Precondition: n.SameType(other) is true. Panics otherwise.
+func (n *Number) GreaterThan(other *Number) bool {
+	if !n.SameType(other) {
+		panic("tried to compare two numbers of different types")
+	}
+	if n.IsInteger() {
+		return n.Int > other.Int
+	}
+	return n.Float > other.Float
+}
+
 // writeTo writes the number to the given writer.
 func (n *Number) writeTo(w io.Writer) error {
 	var s string
diff --git a/pkg/prometheus/prometheus_test.go b/pkg/prometheus/prometheus_test.go
index 8d1ed5ab9..ebc855bc4 100644
--- a/pkg/prometheus/prometheus_test.go
+++ b/pkg/prometheus/prometheus_test.go
@@ -74,6 +74,16 @@ func (m *metricMetadata) clone() *metricMetadata {
 	return m2
 }
 
+// withField returns a copy of this *metricMetadata with the given field added
+// to its metadata.
+func (m *metricMetadata) withField(fields ...*pb.MetricMetadata_Field) *metricMetadata {
+	m2 := m.clone()
+	m2.PB.Fields = make([]*pb.MetricMetadata_Field, 0, len(m.Fields)+len(fields))
+	copy(m2.PB.Fields, m.PB.Fields)
+	m2.PB.Fields = append(m2.PB.Fields, fields...)
+	return m2
+}
+
 // metric returns the Metric metadata struct for this metric metadata.
 func (m *metricMetadata) metric() *Metric {
 	var metricType Type
@@ -226,6 +236,677 @@ var (
 	}
 )
 
+// newMetricRegistration returns a new *metricRegistration.
+func newMetricRegistration(metricMetadata ...*metricMetadata) *pb.MetricRegistration {
+	metadatas := make([]*pb.MetricMetadata, len(metricMetadata))
+	for i, mm := range metricMetadata {
+		metadatas[i] = mm.PB
+	}
+	return &pb.MetricRegistration{
+		Metrics: metadatas,
+	}
+}
+
+func TestVerifier(t *testing.T) {
+	testStart := time.Now()
+	epsilon := func(n int) time.Time {
+		return testStart.Add(time.Duration(n) * time.Millisecond)
+	}
+	for _, test := range []struct {
+		Name string
+		// At is the time at which the test executes.
+		// If unset, `testStart` is assumed.
+		At time.Time
+		// Registration is the metric registration data.
+		Registration *pb.MetricRegistration
+		// WantVerifierCreationErr is true if the test expects the
+		// creation of the Verifier to fail. All the fields below it
+		// are ignored in this case.
+		WantVerifierCreationErr bool
+		// WantSuccess is a sequence of Snapshots to present to
+		// the verifier. The test expects all of them to pass verification.
+		// If unset, the test simply presents the WantFail Snapshot.
+		// If both WantSuccess and WantFail are unset, the test presents
+		// an empty snapshot and expects it to succeed.
+		WantSuccess []*Snapshot
+		// WantFail is a Snapshot to present to the verifier after all
+		// snapshots in WantSuccess have been presented.
+		// The test expects this Snapshot to fail verification.
+		// If unset, the test does not present any snapshot after
+		// having presented the WantSuccess Snapshots.
+		WantFail *Snapshot
+	}{
+		{
+			Name: "no metrics, empty snapshot",
+		},
+		{
+			Name:                    "duplicate metric",
+			Registration:            newMetricRegistration(fooInt, fooInt),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name:                    "duplicate metric with different field set",
+			Registration:            newMetricRegistration(fooInt, fooInt.withField(field1)),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name:                    "duplicate field in metric",
+			Registration:            newMetricRegistration(fooInt.withField(field1, field1)),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "no field allowed value",
+			Registration: newMetricRegistration(fooInt.withField(&pb.MetricMetadata_Field{
+				FieldName: "field1",
+			})),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "duplicate field allowed value",
+			Registration: newMetricRegistration(fooInt.withField(&pb.MetricMetadata_Field{
+				FieldName:     "field1",
+				AllowedValues: []string{"val1", "val1"},
+			})),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "invalid metric type",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name:           "fooBar",
+					PrometheusName: "foo_bar",
+					Type:           pb.MetricMetadata_Type(1337),
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "empty metric name",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					PrometheusName: "foo_bar",
+					Type:           pb.MetricMetadata_TYPE_UINT64,
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "empty Prometheus metric name",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name: "fooBar",
+					Type: pb.MetricMetadata_TYPE_UINT64,
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "bad Prometheus metric name",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name:           "fooBar",
+					PrometheusName: "fooBar",
+					Type:           pb.MetricMetadata_TYPE_UINT64,
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "bad first Prometheus metric name character",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name:           "fooBar",
+					PrometheusName: "_foo_bar",
+					Type:           pb.MetricMetadata_TYPE_UINT64,
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "no buckets",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name:                          "fooBar",
+					PrometheusName:                "foo_bar",
+					Type:                          pb.MetricMetadata_TYPE_DISTRIBUTION,
+					DistributionBucketLowerBounds: []int64{},
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "too many buckets",
+			Registration: newMetricRegistration(&metricMetadata{
+				PB: &pb.MetricMetadata{
+					Name:                          "fooBar",
+					PrometheusName:                "foo_bar",
+					Type:                          pb.MetricMetadata_TYPE_DISTRIBUTION,
+					DistributionBucketLowerBounds: make([]int64, 999),
+				}},
+			),
+			WantVerifierCreationErr: true,
+		},
+		{
+			Name: "successful registration of complex set of metrics",
+			Registration: newMetricRegistration(
+				fooInt,
+				fooCounter.withField(field1, field2),
+				fooDist.withField(field2),
+			),
+		},
+		{
+			Name: "snapshot time ordering",
+			At:   epsilon(0),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)),
+				newSnapshotAt(epsilon(-2)),
+				newSnapshotAt(epsilon(-1)),
+			},
+			WantFail: newSnapshotAt(epsilon(-2)),
+		},
+		{
+			Name: "same snapshot time is ok",
+			At:   epsilon(0),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)),
+				newSnapshotAt(epsilon(-2)),
+				newSnapshotAt(epsilon(-1)),
+				newSnapshotAt(epsilon(-1)),
+				newSnapshotAt(epsilon(-1)),
+				newSnapshotAt(epsilon(-1)),
+				newSnapshotAt(epsilon(0)),
+				newSnapshotAt(epsilon(0)),
+				newSnapshotAt(epsilon(0)),
+				newSnapshotAt(epsilon(0)),
+			},
+		},
+		{
+			Name:     "snapshot from the future",
+			At:       epsilon(0),
+			WantFail: newSnapshotAt(epsilon(1)),
+		},
+		{
+			Name:     "snapshot from the long past",
+			At:       testStart,
+			WantFail: newSnapshotAt(testStart.Add(-25 * time.Hour)),
+		},
+		{
+			Name:         "simple metric update",
+			Registration: newMetricRegistration(fooInt),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-1)).Add(
+					fooInt.int(2),
+				),
+			},
+		},
+		{
+			Name:         "simple metric update multiple times",
+			Registration: newMetricRegistration(fooInt),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)).Add(fooInt.int(2)),
+				newSnapshotAt(epsilon(-2)).Add(fooInt.int(-1)),
+				newSnapshotAt(epsilon(-1)).Add(fooInt.int(4)),
+			},
+		},
+		{
+			Name:         "counter can go forwards",
+			Registration: newMetricRegistration(fooCounter),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)).Add(fooCounter.int(1)),
+				newSnapshotAt(epsilon(-2)).Add(fooCounter.int(3)),
+				newSnapshotAt(epsilon(-1)).Add(fooCounter.int(3)),
+			},
+		},
+		{
+			Name:         "counter cannot go backwards",
+			Registration: newMetricRegistration(fooCounter),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)).Add(fooCounter.int(1)),
+				newSnapshotAt(epsilon(-2)).Add(fooCounter.int(3)),
+			},
+			WantFail: newSnapshotAt(epsilon(-1)).Add(fooCounter.int(2)),
+		},
+		{
+			Name:         "counter cannot change type",
+			Registration: newMetricRegistration(fooCounter),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)).Add(fooCounter.int(1)),
+				newSnapshotAt(epsilon(-2)).Add(fooCounter.int(3)),
+			},
+			WantFail: newSnapshotAt(epsilon(-1)).Add(fooCounter.float(4)),
+		},
+		{
+			Name:         "update for unknown metric",
+			Registration: newMetricRegistration(fooInt),
+			WantFail:     newSnapshotAt(epsilon(-1)).Add(fooCounter.int(2)),
+		},
+		{
+			Name:         "update for mismatching metric definition: type",
+			Registration: newMetricRegistration(fooInt),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				(&metricMetadata{PB: &pb.MetricMetadata{
+					PrometheusName: fooInt.PB.GetPrometheusName(),
+					Type:           pb.MetricMetadata_TYPE_DISTRIBUTION,
+					Description:    fooInt.PB.GetDescription(),
+				}}).int(2),
+			),
+		},
+		{
+			Name:         "update for mismatching metric definition: name",
+			Registration: newMetricRegistration(fooInt),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				(&metricMetadata{PB: &pb.MetricMetadata{
+					PrometheusName: "not_foo_int",
+					Type:           fooInt.PB.GetType(),
+					Description:    fooInt.PB.GetDescription(),
+				}}).int(2),
+			),
+		},
+		{
+			Name:         "update for mismatching metric definition: description",
+			Registration: newMetricRegistration(fooInt),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				(&metricMetadata{PB: &pb.MetricMetadata{
+					PrometheusName: fooInt.PB.GetPrometheusName(),
+					Type:           fooInt.PB.GetType(),
+					Description:    "not fooInt's description",
+				}}).int(2),
+			),
+		},
+		{
+			Name:         "update with no fields for metric with fields",
+			Registration: newMetricRegistration(fooInt.withField(field1)),
+			WantFail:     newSnapshotAt(epsilon(-1)).Add(fooInt.int(2)),
+		},
+		{
+			Name:         "update with fields for metric without fields",
+			Registration: newMetricRegistration(fooInt),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooInt.fieldVal(field1, "val1a").int(2),
+			),
+		},
+		{
+			Name:         "update with invalid field value",
+			Registration: newMetricRegistration(fooInt.withField(field1)),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooInt.fieldVal(field1, "not_val1a").int(2),
+			),
+		},
+		{
+			Name:         "update with valid field value for wrong field",
+			Registration: newMetricRegistration(fooInt.withField(field1)),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooInt.fieldVal(field2, "val1a").int(2),
+			),
+		},
+		{
+			Name:         "update with valid field values provided twice",
+			Registration: newMetricRegistration(fooInt.withField(field1)),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooInt.fieldVal(field1, "val1a").int(2),
+				fooInt.fieldVal(field1, "val1a").int(2),
+			),
+		},
+		{
+			Name:         "update with valid field value",
+			Registration: newMetricRegistration(fooInt.withField(field1)),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-1)).Add(
+					fooInt.fieldVal(field1, "val1a").int(7),
+					fooInt.fieldVal(field1, "val1b").int(2),
+				),
+			},
+		},
+		{
+			Name:         "update with multiple valid field value",
+			Registration: newMetricRegistration(fooCounter.withField(field1, field2)),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-1)).Add(
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(3),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1b",
+						field2: "val2a",
+					}).int(2),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2b",
+					}).int(1),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1b",
+						field2: "val2b",
+					}).int(4),
+				),
+			},
+		},
+		{
+			Name:         "update with multiple valid field values but duplicated",
+			Registration: newMetricRegistration(fooCounter.withField(field1, field2)),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+					field1: "val1b",
+					field2: "val2b",
+				}).int(4),
+				fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+					field1: "val1b",
+					field2: "val2b",
+				}).int(4),
+			),
+		},
+		{
+			Name: "update with same valid field values across two metrics",
+			Registration: newMetricRegistration(
+				fooInt.withField(field1, field2),
+				fooCounter.withField(field1, field2),
+			),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-1)).Add(
+					fooInt.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(3),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(3),
+				),
+			},
+		},
+		{
+			Name:         "update with multiple value types",
+			Registration: newMetricRegistration(fooInt),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooInt.metric(),
+					Number: &Number{Int: 2},
+					HistogramValue: &Histogram{
+						Total: Number{Int: 5},
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 1},
+							{UpperBound: Number{Int: 1}, Samples: 1},
+						},
+					},
+				},
+			),
+		},
+		{
+			Name:         "integer metric gets float value",
+			Registration: newMetricRegistration(fooInt),
+			WantFail:     newSnapshotAt(epsilon(-1)).Add(fooInt.float(2.5)),
+		},
+		{
+			Name:         "metric gets no value",
+			Registration: newMetricRegistration(fooInt),
+			WantFail:     newSnapshotAt(epsilon(-1)).Add(&Data{Metric: fooInt.metric()}),
+		},
+		{
+			Name:         "distribution gets integer value",
+			Registration: newMetricRegistration(fooDist),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooDist.int(2),
+			),
+		},
+		{
+			Name:         "successful distribution",
+			Registration: newMetricRegistration(fooDist),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-1)).Add(
+					fooDist.dist(1, 2, 3, 4, 5, 6),
+				),
+			},
+		},
+		{
+			Name:         "distribution updates",
+			Registration: newMetricRegistration(fooDist),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-2)).Add(
+					fooDist.dist(1, 2, 3, 4, 5, 6),
+				),
+				newSnapshotAt(epsilon(-1)).Add(
+					fooDist.dist(0, 1, 1, 2, 2, 3, 4, 5, 5, 6, 7, 8, 9, 25),
+				),
+			},
+		},
+		{
+			Name:         "distribution updates with fields",
+			Registration: newMetricRegistration(fooDist.withField(field1)),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-2)).Add(
+					fooDist.fieldVal(field1, "val1a").dist(1, 2, 3, 4, 5, 6),
+				),
+				newSnapshotAt(epsilon(-1)).Add(
+					fooDist.fieldVal(field1, "val1a").dist(0, 1, 1, 2, 2, 3, 4, 5, 5, 6, 7, 8, 9, 25),
+				),
+			},
+		},
+		{
+			Name:         "distribution cannot have number of samples regress",
+			Registration: newMetricRegistration(fooDist),
+			WantSuccess: []*Snapshot{
+				newSnapshotAt(epsilon(-3)).Add(
+					fooDist.dist(1, 2, 3, 4, 5, 6),
+				),
+				newSnapshotAt(epsilon(-2)).Add(
+					fooDist.dist(0, 1, 1, 2, 2, 3, 4, 5, 5, 6, 7, 8, 9, 25),
+				),
+			},
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				fooDist.dist(0, 1, 2, 2, 3, 4, 5, 5, 6, 7, 8, 9),
+			),
+		},
+		{
+			Name:         "distribution with zero samples",
+			Registration: newMetricRegistration(fooDist),
+			WantSuccess: []*Snapshot{newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooDist.metric(),
+					HistogramValue: &Histogram{
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 0},
+							{UpperBound: Number{Int: 1}, Samples: 0},
+							{UpperBound: Number{Int: 2}, Samples: 0},
+							{UpperBound: Number{Int: 4}, Samples: 0},
+							{UpperBound: Number{Int: 8}, Samples: 0},
+							{UpperBound: Number{Float: math.Inf(1)}, Samples: 0},
+						},
+					},
+				},
+			)},
+		},
+		{
+			Name:         "distribution with manual samples",
+			Registration: newMetricRegistration(fooDist),
+			WantSuccess: []*Snapshot{newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooDist.metric(),
+					HistogramValue: &Histogram{
+						Total: Number{Int: 10},
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 2},
+							{UpperBound: Number{Int: 1}, Samples: 1},
+							{UpperBound: Number{Int: 2}, Samples: 3},
+							{UpperBound: Number{Int: 4}, Samples: 1},
+							{UpperBound: Number{Int: 8}, Samples: 4},
+							{UpperBound: Number{Float: math.Inf(1)}, Samples: 1},
+						},
+					},
+				},
+			)},
+		},
+		{
+			Name:         "distribution gets bad number of buckets",
+			Registration: newMetricRegistration(fooDist),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooDist.metric(),
+					HistogramValue: &Histogram{
+						Total: Number{Int: 10},
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 2},
+							{UpperBound: Number{Int: 1}, Samples: 1},
+							{UpperBound: Number{Int: 2}, Samples: 3},
+							// Missing: {UpperBound: Number{Int: 4}, Samples: 1},
+							{UpperBound: Number{Int: 8}, Samples: 4},
+							{UpperBound: Number{Float: math.Inf(1)}, Samples: 1},
+						},
+					},
+				},
+			),
+		},
+		{
+			Name:         "distribution gets unexpected bucket boundary",
+			Registration: newMetricRegistration(fooDist),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooDist.metric(),
+					HistogramValue: &Histogram{
+						Total: Number{Int: 10},
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 2},
+							{UpperBound: Number{Int: 1}, Samples: 1},
+							{UpperBound: Number{Int: 3 /* Should be 2 */}, Samples: 3},
+							{UpperBound: Number{Int: 4}, Samples: 1},
+							{UpperBound: Number{Int: 8}, Samples: 4},
+							{UpperBound: Number{Float: math.Inf(1)}, Samples: 1},
+						},
+					},
+				},
+			),
+		},
+		{
+			Name:         "distribution gets unexpected last bucket boundary",
+			Registration: newMetricRegistration(fooDist),
+			WantFail: newSnapshotAt(epsilon(-1)).Add(
+				&Data{
+					Metric: fooDist.metric(),
+					HistogramValue: &Histogram{
+						Total: Number{Int: 10},
+						Buckets: []Bucket{
+							{UpperBound: Number{Int: 0}, Samples: 2},
+							{UpperBound: Number{Int: 1}, Samples: 1},
+							{UpperBound: Number{Int: 2}, Samples: 3},
+							{UpperBound: Number{Int: 4}, Samples: 1},
+							{UpperBound: Number{Int: 8}, Samples: 4},
+							{
+								UpperBound: Number{Float: math.Inf(-1) /* Should be +inf */},
+								Samples:    1,
+							},
+						},
+					},
+				},
+			),
+		},
+		{
+			Name: "worked example",
+			Registration: newMetricRegistration(
+				fooInt,
+				fooDist.withField(field1),
+				fooCounter.withField(field1, field2),
+			),
+			WantSuccess: []*Snapshot{
+				// Empty snapshot.
+				newSnapshotAt(epsilon(-6)),
+				// Simple snapshot.
+				newSnapshotAt(epsilon(-5)).Add(
+					fooInt.int(3),
+					fooDist.fieldVal(field1, "val1a").dist(1, 2, 3, 4, 5, 6),
+					fooDist.fieldVal(field1, "val1b").dist(-1, -8, 100),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(6),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1b",
+						field2: "val2a",
+					}).int(3),
+				),
+				// And another.
+				newSnapshotAt(epsilon(-4)).Add(
+					fooInt.int(1),
+					fooDist.fieldVal(field1, "val1a").dist(1, 2, 3, 4, 5, 6, 7),
+					fooDist.fieldVal(field1, "val1b").dist(-1, -8, 100, 42),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(6),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1b",
+						field2: "val2a",
+					}).int(4),
+				),
+				// And another one, partial this time.
+				newSnapshotAt(epsilon(-3)).Add(
+					fooDist.fieldVal(field1, "val1b").dist(-1, -8, 100, 42, 1337),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(6),
+				),
+				// An empty one.
+				newSnapshotAt(epsilon(-2)),
+				// Another empty one at the same timestamp.
+				newSnapshotAt(epsilon(-1)),
+				// Another full one which doesn't change any value.
+				newSnapshotAt(epsilon(0)).Add(
+					fooInt.int(1),
+					fooDist.fieldVal(field1, "val1a").dist(1, 2, 3, 4, 5, 6, 7),
+					fooDist.fieldVal(field1, "val1b").dist(-1, -8, 100, 42, 1337),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1a",
+						field2: "val2a",
+					}).int(6),
+					fooCounter.fieldVals(map[*pb.MetricMetadata_Field]string{
+						field1: "val1b",
+						field2: "val2a",
+					}).int(4),
+				),
+			},
+		},
+	} {
+		t.Run(test.Name, func(t *testing.T) {
+			testTime := test.At
+			if testTime.IsZero() {
+				testTime = testStart
+			}
+			at(testTime, func() {
+				t.Logf("Test is running with simulated time: %v", testTime)
+				verifier, err := NewVerifier(test.Registration)
+				if err != nil && !test.WantVerifierCreationErr {
+					t.Fatalf("unexpected verifier creation error: %v", err)
+				}
+				if err == nil && test.WantVerifierCreationErr {
+					t.Fatal("verifier creation unexpectedly succeeded")
+				}
+				if err != nil {
+					t.Logf("Verifier creation failed (as expected by this test): %v", err)
+					return
+				}
+
+				if len(test.WantSuccess) == 0 && test.WantFail == nil {
+					if err = verifier.Verify(NewSnapshot()); err != nil {
+						t.Errorf("empty snapshot failed verification: %v", err)
+					}
+				} else {
+					for i, snapshot := range test.WantSuccess {
+						if err = verifier.Verify(snapshot); err != nil {
+							t.Fatalf("snapshot WantSuccess[%d] failed verification: %v", i, err)
+						}
+					}
+					if test.WantFail != nil {
+						if err = verifier.Verify(test.WantFail); err == nil {
+							t.Error("WantFail snapshot unexpectedly succeeded verification")
+						} else {
+							t.Logf("WantFail snapshot failed verification (as expected by this test): %v", err)
+						}
+					}
+				}
+			})
+		})
+	}
+}
+
 // shortWriter implements io.Writer but fails after a given number of bytes.
 type shortWriter struct {
 	buf     bytes.Buffer
diff --git a/pkg/prometheus/prometheus_verify.go b/pkg/prometheus/prometheus_verify.go
new file mode 100644
index 000000000..7b77409af
--- /dev/null
+++ b/pkg/prometheus/prometheus_verify.go
@@ -0,0 +1,332 @@
+// Copyright 2022 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package prometheus
+
+import (
+	"errors"
+	"fmt"
+	"math"
+	"strings"
+	"sync"
+	"time"
+	"unicode"
+
+	pb "gvisor.dev/gvisor/pkg/metric/metric_go_proto"
+)
+
+const (
+	// maxExportStaleness is the maximum allowed age of a snapshot when it is verified.
+	// Used to avoid exporting snapshots from bogus times from ages past.
+	maxExportStaleness = 10 * time.Second
+)
+
+// verifiableMetric verifies a single metric within a Verifier.
+type verifiableMetric struct {
+	metadata              *pb.MetricMetadata
+	wantMetric            Metric
+	numFields             int
+	allowedFieldValues    map[string]map[string]struct{}
+	wantBucketUpperBounds []Number
+
+	// The following fields are used to verify that values are actually increasing monotonically.
+	// They are only read and modified when the parent Verifier.mu is held.
+	// They are mapped by their combination of field values.
+
+	// lastCounterValue is used for counter metrics.
+	lastCounterValue map[string]Number
+
+	// lastBucketSamples is used for distribution ("histogram") metrics.
+	lastBucketSamples map[string][]uint64
+}
+
+// newVerifiableMetric creates a new verifiableMetric that can verify the
+// values of a metric with the given metadata.
+func newVerifiableMetric(metadata *pb.MetricMetadata) (*verifiableMetric, error) {
+	if metadata.GetName() == "" || metadata.GetPrometheusName() == "" {
+		return nil, errors.New("metric has no name")
+	}
+	if !unicode.IsLower(rune(metadata.GetPrometheusName()[0])) {
+		return nil, fmt.Errorf("invalid initial character in prometheus metric name: %q", metadata.GetPrometheusName())
+	}
+	for _, r := range metadata.GetPrometheusName() {
+		if !unicode.IsLower(r) && !unicode.IsDigit(r) && r != '_' {
+			return nil, fmt.Errorf("invalid character %c in prometheus metric name %q", r, metadata.GetPrometheusName())
+		}
+	}
+	numFields := len(metadata.GetFields())
+	var allowedFieldValues map[string]map[string]struct{}
+	if numFields > 0 {
+		seenFields := make(map[string]struct{}, numFields)
+		allowedFieldValues = make(map[string]map[string]struct{}, numFields)
+		for _, field := range metadata.GetFields() {
+			fieldName := field.GetFieldName()
+			if _, alreadyExists := seenFields[fieldName]; alreadyExists {
+				return nil, fmt.Errorf("field %s is defined twice", fieldName)
+			}
+			seenFields[fieldName] = struct{}{}
+			if len(field.GetAllowedValues()) == 0 {
+				return nil, fmt.Errorf("field %s has no allowed values", fieldName)
+			}
+			fieldValues := make(map[string]struct{}, len(field.GetAllowedValues()))
+			for _, value := range field.GetAllowedValues() {
+				if _, alreadyExists := fieldValues[value]; alreadyExists {
+					return nil, fmt.Errorf("field %s has duplicate allowed value %q", fieldName, value)
+				}
+				fieldValues[value] = struct{}{}
+			}
+			allowedFieldValues[fieldName] = fieldValues
+		}
+	}
+	v := &verifiableMetric{
+		metadata: metadata,
+		wantMetric: Metric{
+			Name: metadata.GetPrometheusName(),
+			Help: metadata.GetDescription(),
+		},
+		numFields:          numFields,
+		allowedFieldValues: allowedFieldValues,
+	}
+	numFieldCombinations := len(allowedFieldValues)
+	switch metadata.GetType() {
+	case pb.MetricMetadata_TYPE_UINT64:
+		v.wantMetric.Type = TypeGauge
+		if metadata.GetCumulative() {
+			v.wantMetric.Type = TypeCounter
+			v.lastCounterValue = make(map[string]Number, numFieldCombinations)
+		}
+	case pb.MetricMetadata_TYPE_DISTRIBUTION:
+		v.wantMetric.Type = TypeHistogram
+		numBuckets := len(metadata.GetDistributionBucketLowerBounds()) + 1
+		if numBuckets <= 1 || numBuckets > 256 {
+			return nil, fmt.Errorf("unsupported number of buckets: %d", numBuckets)
+		}
+		v.wantBucketUpperBounds = make([]Number, numBuckets)
+		for i, boundary := range metadata.GetDistributionBucketLowerBounds() {
+			v.wantBucketUpperBounds[i] = Number{Int: boundary}
+		}
+		v.wantBucketUpperBounds[numBuckets-1] = Number{Float: math.Inf(1)}
+		v.lastBucketSamples = make(map[string][]uint64, numFieldCombinations)
+	default:
+		return nil, fmt.Errorf("invalid type: %v", metadata.GetType())
+	}
+	return v, nil
+}
+
+func (v *verifiableMetric) numFieldCombinations() int {
+	return len(v.allowedFieldValues)
+}
+
+// verify does read-only checks on `data`.
+// `metricFieldsSeen` is passed across calls to `verify`. It is used to track the set of metric
+// field values that have already been seen. `verify` should populate this.
+// `dataToFieldsSeen` is passed across calls to `verify` and other methods of `verifiableMetric`.
+// It is used to store the canonical representation of the field values seen for each *Data.
+func (v *verifiableMetric) verify(data *Data, metricFieldsSeen map[string]struct{}, dataToFieldsSeen map[*Data]string) error {
+	if *data.Metric != v.wantMetric {
+		return fmt.Errorf("invalid metric definition: got %+v want %+v", data.Metric, v.wantMetric)
+	}
+
+	// Verify fields.
+	if len(data.Labels) != v.numFields {
+		return fmt.Errorf("invalid number of fields: got %d want %d", len(data.Labels), v.numFields)
+	}
+	var fieldValues strings.Builder
+	firstField := true
+	for _, field := range v.metadata.GetFields() {
+		fieldName := field.GetFieldName()
+		value, found := data.Labels[fieldName]
+		if !found {
+			return fmt.Errorf("did not specify field %q", fieldName)
+		}
+		if _, allowed := v.allowedFieldValues[fieldName][value]; !allowed {
+			return fmt.Errorf("value %q is not allowed for field %s", value, fieldName)
+		}
+		if !firstField {
+			fieldValues.WriteRune(',')
+		}
+		fieldValues.WriteString(value)
+		firstField = false
+	}
+	fieldValuesStr := fieldValues.String()
+	if _, alreadySeen := metricFieldsSeen[fieldValuesStr]; alreadySeen {
+		return fmt.Errorf("combination of field values %q was already seen", fieldValuesStr)
+	}
+
+	// Verify value.
+	gotNumber := data.Number != nil
+	gotHistogram := data.HistogramValue != nil
+	numSpecified := 0
+	if gotNumber {
+		numSpecified++
+	}
+	if gotHistogram {
+		numSpecified++
+	}
+	if numSpecified != 1 {
+		return fmt.Errorf("invalid number of value fields specified: %d", numSpecified)
+	}
+	switch v.metadata.GetType() {
+	case pb.MetricMetadata_TYPE_UINT64:
+		if !gotNumber {
+			return errors.New("expected number value for gauge or counter")
+		}
+		if !data.Number.IsInteger() {
+			return fmt.Errorf("integer metric got non-integer value: %v", data.Number)
+		}
+	case pb.MetricMetadata_TYPE_DISTRIBUTION:
+		if !gotHistogram {
+			return errors.New("expected histogram value for histogram")
+		}
+		if len(data.HistogramValue.Buckets) != len(v.wantBucketUpperBounds) {
+			return fmt.Errorf("invalid number of buckets: got %d want %d", len(data.HistogramValue.Buckets), len(v.wantBucketUpperBounds))
+		}
+		for i, b := range data.HistogramValue.Buckets {
+			if want := v.wantBucketUpperBounds[i]; b.UpperBound != want {
+				return fmt.Errorf("invalid upper bound for bucket %d (0-based): got %v want %v", i, b.UpperBound, want)
+			}
+		}
+	default:
+		return fmt.Errorf("invalid metric type: %v", v.wantMetric.Type)
+	}
+
+	// All passed. Update the maps that are shared across calls.
+	dataToFieldsSeen[data] = fieldValuesStr
+	metricFieldsSeen[fieldValuesStr] = struct{}{}
+	return nil
+}
+
+// verifyIncrement verifies that incremental metrics are monotonically increasing.
+// Preconditions: `verify` has succeeded on the given `data`, and `Verifier.mu` is held.
+func (v *verifiableMetric) verifyIncrement(data *Data, fieldValues string) error {
+	switch v.wantMetric.Type {
+	case TypeCounter:
+		last := v.lastCounterValue[fieldValues]
+		if !last.SameType(data.Number) {
+			return fmt.Errorf("counter number type changed: %v vs %v", last, data.Number)
+		}
+		if last.GreaterThan(data.Number) {
+			return fmt.Errorf("counter value decreased from %v to %v", last, data.Number)
+		}
+	case TypeHistogram:
+		lastBucketSamples := v.lastBucketSamples[fieldValues]
+		if lastBucketSamples == nil {
+			lastBucketSamples = make([]uint64, len(v.wantBucketUpperBounds))
+			v.lastBucketSamples[fieldValues] = lastBucketSamples
+		}
+		for i, b := range data.HistogramValue.Buckets {
+			if lastBucketSamples[i] > b.Samples {
+				return fmt.Errorf("number of samples in bucket %d (0-based) decreased from %d to %d", i, lastBucketSamples[i], b.Samples)
+			}
+		}
+	}
+	return nil
+}
+
+// update updates incremental metrics' "last seen" data.
+// Preconditions: `verifyIncrement` has succeeded on the given `data`, and `Verifier.mu` is held.
+func (v *verifiableMetric) update(data *Data, fieldValues string) {
+	switch v.wantMetric.Type {
+	case TypeCounter:
+		v.lastCounterValue[fieldValues] = *data.Number
+	case TypeHistogram:
+		lastBucketSamples := v.lastBucketSamples[fieldValues]
+		for i, b := range data.HistogramValue.Buckets {
+			lastBucketSamples[i] = b.Samples
+		}
+	}
+}
+
+// Verifier allows verifying metric snapshot against metric registration data.
+// The aim is to prevent a compromised Sentry from emitting bogus data or DoS'ing metric ingestion.
+// A single Verifier should be used per sandbox. It is expected to be reused across exports such
+// that it can enforce the export snapshot timestamp is strictly monotonically increasing.
+type Verifier struct {
+	knownMetrics  map[string]*verifiableMetric
+	mu            sync.Mutex
+	lastTimestamp time.Time
+}
+
+// NewVerifier returns a new metric verifier that can verify the integrity of snapshots against
+// the given metric registration data.
+func NewVerifier(registration *pb.MetricRegistration) (*Verifier, error) {
+	knownMetrics := make(map[string]*verifiableMetric)
+	for _, metric := range registration.GetMetrics() {
+		metricName := metric.GetPrometheusName()
+		if _, alreadyExists := knownMetrics[metricName]; alreadyExists {
+			return nil, fmt.Errorf("metric %q registered twice", metricName)
+		}
+		verifiableM, err := newVerifiableMetric(metric)
+		if err != nil {
+			return nil, fmt.Errorf("metric %q: %v", metricName, err)
+		}
+		knownMetrics[metricName] = verifiableM
+	}
+	return &Verifier{
+		knownMetrics: knownMetrics,
+	}, nil
+}
+
+// Verify verifies the integrity of a snapshot against the metric registration data of the Verifier.
+// It assumes that it will be called on snapshots obtained chronologically over time.
+func (v *Verifier) Verify(snapshot *Snapshot) error {
+	var err error
+
+	// Basic timestamp checks.
+	now := timeNow()
+	if snapshot.When.After(now) {
+		return errors.New("snapshot is from the future")
+	}
+	if snapshot.When.Before(now.Add(-maxExportStaleness)) {
+		return fmt.Errorf("snapshot is too old; it is from %v, expected at least %v (%v from now)", snapshot.When, now.Add(-maxExportStaleness), maxExportStaleness)
+	}
+
+	// Metrics checks.
+	fieldsSeen := make(map[string]map[string]struct{}, len(v.knownMetrics))
+	dataToFieldsSeen := make(map[*Data]string, len(snapshot.Data))
+	for _, data := range snapshot.Data {
+		metricName := data.Metric.Name
+		verifiableM, found := v.knownMetrics[metricName]
+		if !found {
+			return fmt.Errorf("snapshot contains unknown metric %q", metricName)
+		}
+		metricFieldsSeen, found := fieldsSeen[metricName]
+		if !found {
+			metricFieldsSeen = make(map[string]struct{}, verifiableM.numFieldCombinations())
+			fieldsSeen[metricName] = metricFieldsSeen
+		}
+		if err = verifiableM.verify(data, metricFieldsSeen, dataToFieldsSeen); err != nil {
+			return fmt.Errorf("metric %q: %v", metricName, err)
+		}
+	}
+
+	// Start the critical section.
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if v.lastTimestamp.After(snapshot.When) {
+		return fmt.Errorf("consecutive snapshots are not chronologically ordered: last verified snapshot was exported at %v, this one is from %v", v.lastTimestamp, snapshot.When)
+	}
+	for _, data := range snapshot.Data {
+		if err = v.knownMetrics[data.Metric.Name].verifyIncrement(data, dataToFieldsSeen[data]); err != nil {
+			return fmt.Errorf("metric %q: %v", data.Metric.Name, err)
+		}
+	}
+
+	// All checks succeeded, update last-seen data.
+	v.lastTimestamp = snapshot.When
+	for _, data := range snapshot.Data {
+		v.knownMetrics[data.Metric.Name].update(data, dataToFieldsSeen[data])
+	}
+	return nil
+}