RepoMirrors/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus synced 2025-03-23 03:36:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Put PRWv2 created timestamp ingestion behing feature-flag

Browse Source 
Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>
This commit is contained in:
Arthur Silva Sens 2024-08-28 16:58:46 -03:00
parent 11e4673e0b
commit 3b97a6397c
No known key found for this signature in database
6 changed files with 63 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/prometheus/main.go
View File

@ -259,6 +259,7 @@ func (c *flagConfig) setFeatureListOptions(logger *slog.Logger) error {
logger.Info("Experimental out-of-order native histogram ingestion enabled. This will only take effect if OutOfOrderTimeWindow is > 0 and if EnableNativeHistograms = true")
case "created-timestamp-zero-ingestion":
c.scrape.EnableCreatedTimestampZeroIngestion = true
c.web.CTZeroIngestionEnabled = true
// Change relevant global variables. Hacky, but it's hard to pass a new option or default to unmarshallers.
config.DefaultConfig.GlobalConfig.ScrapeProtocols = config.DefaultProtoFirstScrapeProtocols
config.DefaultGlobalConfig.ScrapeProtocols = config.DefaultProtoFirstScrapeProtocols

44
storage/remote/write_handler.go
View File

@ -48,6 +48,8 @@ type writeHandler struct {
samplesAppendedWithoutMetadata prometheus.Counter
acceptedProtoMsgs map[config.RemoteWriteProtoMsg]struct{}
ingestCTZeroSample bool
}
const maxAheadTime = 10 * time.Minute
@ -57,7 +59,7 @@ const maxAheadTime = 10 * time.Minute
//
// NOTE(bwplotka): When accepting v2 proto and spec, partial writes are possible
// as per https://prometheus.io/docs/specs/remote_write_spec_2_0/#partial-write.
func NewWriteHandler(logger *slog.Logger, reg prometheus.Registerer, appendable storage.Appendable, acceptedProtoMsgs []config.RemoteWriteProtoMsg) http.Handler {
func NewWriteHandler(logger *slog.Logger, reg prometheus.Registerer, appendable storage.Appendable, acceptedProtoMsgs []config.RemoteWriteProtoMsg, ingestCTZeroSample bool) http.Handler {
protoMsgs := map[config.RemoteWriteProtoMsg]struct{}{}
for _, acc := range acceptedProtoMsgs {
protoMsgs[acc] = struct{}{}
@ -78,6 +80,8 @@ func NewWriteHandler(logger *slog.Logger, reg prometheus.Registerer, appendable
Name: "remote_write_without_metadata_appended_samples_total",
Help: "The total number of received remote write samples (and histogram samples) which were ingested without corresponding metadata.",
}),
ingestCTZeroSample: ingestCTZeroSample,
}
return h
}
@ -393,22 +397,17 @@ func (h *writeHandler) appendV2(app storage.Appender, req *writev2.Request, rs *
allSamplesSoFar := rs.AllSamples()
var ref storage.SeriesRef
// Samples.
for i, s := range ts.Samples {
if h.ingestCTZeroSample {
// CT only needs to be ingested for the first sample, it will be considered
// out of order for the rest.
if i == 0 && ts.CreatedTimestamp != 0 {
ref, err = app.AppendCTZeroSample(ref, ls, s.Timestamp, ts.CreatedTimestamp)
if err != nil && !errors.Is(err, storage.ErrOutOfOrderCT) {
// Even for the first sample OOO is a common scenario.
// We don't fail the request and just log it.
level.Debug(h.logger).Log("msg", "Error when appending CT in remote write request", "err", err, "series", ls.String(), "created_timestamp", ts.CreatedTimestamp, "timestamp", s.Timestamp)
}
if err == nil {
rs.Samples++
}
ref, err = h.handleCTZeroSample(app, ref, ls, ts.Samples[0], ts.CreatedTimestamp, rs)
if err != nil {
h.logger.Debug("Error when appending CT in remote write request", "err", err, "series", ls.String(), "created_timestamp", ts.CreatedTimestamp, "timestamp", ts.Samples[0].Timestamp)
}
}
// Samples.
for _, s := range ts.Samples {
ref, err = app.Append(ref, ls, s.GetTimestamp(), s.GetValue())
if err == nil {
rs.Samples++
@ -493,6 +492,25 @@ func (h *writeHandler) appendV2(app storage.Appender, req *writev2.Request, rs *
return samplesWithoutMetadata, http.StatusBadRequest, errors.Join(badRequestErrs...)
}
// handleCTZeroSample appends CT as a zero-value sample with CT value as the sample timestamp.
// It doens't return errors in case of out of order CT.
func (h *writeHandler) handleCTZeroSample(app storage.Appender, ref storage.SeriesRef, l labels.Labels, sample writev2.Sample, ct int64, rs *WriteResponseStats) (storage.SeriesRef, error) {
var err error
if sample.Timestamp != 0 && ct != 0 {
ref, err = app.AppendCTZeroSample(ref, l, sample.Timestamp, ct)
if err == nil {
rs.Samples++
}
if err != nil && errors.Is(err, storage.ErrOutOfOrderCT) {
// Even for the first sample OOO is a common scenario because
// we can't tell if a CT was already ingested in a previous request.
// We ignore the error.
err = nil
}
}
return ref, err
}
// NewOTLPWriteHandler creates a http.Handler that accepts OTLP write requests and
// writes them to the provided appendable.
func NewOTLPWriteHandler(logger *slog.Logger, appendable storage.Appendable, configFunc func() config.Config) http.Handler {

41
storage/remote/write_handler_test.go
View File

@ -130,7 +130,7 @@ func TestRemoteWriteHandlerHeadersHandling_V1Message(t *testing.T) {
}
appendable := &mockAppendable{}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -231,7 +231,7 @@ func TestRemoteWriteHandlerHeadersHandling_V2Message(t *testing.T) {
}
appendable := &mockAppendable{}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -256,7 +256,7 @@ func TestRemoteWriteHandler_V1Message(t *testing.T) {
// in Prometheus, so keeping like this to not break existing 1.0 clients.
appendable := &mockAppendable{}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -316,9 +316,17 @@ func TestRemoteWriteHandler_V2Message(t *testing.T) {
appendHistogramErr error
appendExemplarErr error
updateMetadataErr error
ingestCTZeroSample bool
}{
{
desc: "All timeseries accepted",
desc: "All timeseries accepted/ct_enabled",
input: writeV2RequestFixture.Timeseries,
expectedCode: http.StatusNoContent,
ingestCTZeroSample: true,
},
{
desc: "All timeseries accepted/ct_disabled",
input: writeV2RequestFixture.Timeseries,
expectedCode: http.StatusNoContent,
},
@ -448,7 +456,7 @@ func TestRemoteWriteHandler_V2Message(t *testing.T) {
appendExemplarErr: tc.appendExemplarErr,
updateMetadataErr: tc.updateMetadataErr,
}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2}, tc.ingestCTZeroSample)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -474,8 +482,11 @@ func TestRemoteWriteHandler_V2Message(t *testing.T) {
// Double check mandatory 2.0 stats.
// writeV2RequestFixture has 2 series with 1 sample, 2 histograms, 1 exemplar each.
// We expect 3 samples because the first series has regular sample + CT.
expectHeaderValue(t, 3, resp.Header.Get(rw20WrittenSamplesHeader))
if tc.ingestCTZeroSample {
expectHeaderValue(t, 3, resp.Header.Get(rw20WrittenSamplesHeader))
} else {
expectHeaderValue(t, 2, resp.Header.Get(rw20WrittenSamplesHeader))
}
expectHeaderValue(t, 4, resp.Header.Get(rw20WrittenHistogramsHeader))
if tc.appendExemplarErr != nil {
expectHeaderValue(t, 0, resp.Header.Get(rw20WrittenExemplarsHeader))
@ -492,7 +503,7 @@ func TestRemoteWriteHandler_V2Message(t *testing.T) {
ls := ts.ToLabels(&b, writeV2RequestFixture.Symbols)
for _, s := range ts.Samples {
if ts.CreatedTimestamp != 0 {
if ts.CreatedTimestamp != 0 && tc.ingestCTZeroSample {
requireEqual(t, mockSample{ls, ts.CreatedTimestamp, 0}, appendable.samples[i])
i++
}
@ -552,7 +563,7 @@ func TestOutOfOrderSample_V1Message(t *testing.T) {
require.NoError(t, err)
appendable := &mockAppendable{latestSample: map[uint64]int64{labels.FromStrings("__name__", "test_metric").Hash(): 100}}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -594,7 +605,7 @@ func TestOutOfOrderExemplar_V1Message(t *testing.T) {
require.NoError(t, err)
appendable := &mockAppendable{latestSample: map[uint64]int64{labels.FromStrings("__name__", "test_metric").Hash(): 100}}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -632,7 +643,7 @@ func TestOutOfOrderHistogram_V1Message(t *testing.T) {
require.NoError(t, err)
appendable := &mockAppendable{latestSample: map[uint64]int64{labels.FromStrings("__name__", "test_metric").Hash(): 100}}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -663,7 +674,7 @@ func BenchmarkRemoteWriteHandler(b *testing.B) {
appendable := &mockAppendable{}
// TODO: test with other proto format(s)
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
b.ResetTimer()
@ -680,7 +691,7 @@ func TestCommitErr_V1Message(t *testing.T) {
require.NoError(t, err)
appendable := &mockAppendable{commitErr: errors.New("commit error")}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -704,7 +715,7 @@ func TestCommitErr_V2Message(t *testing.T) {
req.Header.Set(RemoteWriteVersionHeader, RemoteWriteVersion20HeaderValue)
appendable := &mockAppendable{commitErr: errors.New("commit error")}
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, appendable, []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV2}, false)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
@ -731,7 +742,7 @@ func BenchmarkRemoteWriteOOOSamples(b *testing.B) {
require.NoError(b, db.Close())
})
// TODO: test with other proto format(s)
handler := NewWriteHandler(promslog.NewNopLogger(), nil, db.Head(), []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1})
handler := NewWriteHandler(promslog.NewNopLogger(), nil, db.Head(), []config.RemoteWriteProtoMsg{config.RemoteWriteProtoMsgV1}, false)
buf, _, _, err := buildWriteRequest(nil, genSeriesWithSample(1000, 200*time.Minute.Milliseconds()), nil, nil, nil, nil, "snappy")
require.NoError(b, err)

3
web/api/v1/api.go
View File

@ -258,6 +258,7 @@ func NewAPI(
rwEnabled bool,
acceptRemoteWriteProtoMsgs []config.RemoteWriteProtoMsg,
otlpEnabled bool,
ctZeroIngestionEnabled bool,
) *API {
a := &API{
QueryEngine: qe,
@ -301,7 +302,7 @@ func NewAPI(
}
if rwEnabled {
a.remoteWriteHandler = remote.NewWriteHandler(logger, registerer, ap, acceptRemoteWriteProtoMsgs)
a.remoteWriteHandler = remote.NewWriteHandler(logger, registerer, ap, acceptRemoteWriteProtoMsgs, ctZeroIngestionEnabled)
}
if otlpEnabled {
a.otlpWriteHandler = remote.NewOTLPWriteHandler(logger, ap, configFunc)

1
web/api/v1/errors_test.go
View File

@ -142,6 +142,7 @@ func createPrometheusAPI(t *testing.T, q storage.SampleAndChunkQueryable) *route
false,
config.RemoteWriteProtoMsgs{config.RemoteWriteProtoMsgV1, config.RemoteWriteProtoMsgV2},
false,
false,
)
promRouter := route.New().WithPrefix("/api/v1")

2
web/web.go
View File

@ -290,6 +290,7 @@ type Options struct {
EnableRemoteWriteReceiver bool
EnableOTLPWriteReceiver bool
IsAgent bool
CTZeroIngestionEnabled bool
AppName string
AcceptRemoteWriteProtoMsgs []config.RemoteWriteProtoMsg
@ -386,6 +387,7 @@ func New(logger *slog.Logger, o *Options) *Handler {
o.EnableRemoteWriteReceiver,
o.AcceptRemoteWriteProtoMsgs,
o.EnableOTLPWriteReceiver,
o.CTZeroIngestionEnabled,
)
if o.RoutePrefix != "/" {