From: Martina Ferrari Date: Sun, 9 Feb 2025 23:02:10 +0000 Subject: Disable OTLP Ingestion endpoint This depends on code not yet in Debian. --- cmd/prometheus/main.go | 5 +- storage/remote/codec.go | 64 +---------------------- storage/remote/write_handler.go | 54 -------------------- storage/remote/write_test.go | 110 ---------------------------------------- web/api/v1/api.go | 21 +------- web/api/v1/errors_test.go | 2 - web/web.go | 5 +- 7 files changed, 5 insertions(+), 256 deletions(-) diff --git a/cmd/prometheus/main.go b/cmd/prometheus/main.go index 4d8031d..da750e6 100644 --- a/cmd/prometheus/main.go +++ b/cmd/prometheus/main.go @@ -178,9 +178,6 @@ func (c *flagConfig) setFeatureListOptions(logger log.Logger) error { case "remote-write-receiver": c.web.EnableRemoteWriteReceiver = true level.Warn(logger).Log("msg", "Remote write receiver enabled via feature flag remote-write-receiver. This is DEPRECATED. Use --web.enable-remote-write-receiver.") - case "otlp-write-receiver": - c.web.EnableOTLPWriteReceiver = true - level.Info(logger).Log("msg", "Experimental OTLP write receiver enabled") case "expand-external-labels": c.enableExpandExternalLabels = true level.Info(logger).Log("msg", "Experimental expand-external-labels enabled") @@ -465,7 +462,7 @@ func main() { a.Flag("scrape.discovery-reload-interval", "Interval used by scrape manager to throttle target groups updates."). Hidden().Default("5s").SetValue(&cfg.scrape.DiscoveryReloadInterval) - a.Flag("enable-feature", "Comma separated feature names to enable. Valid options: agent, auto-gomemlimit, exemplar-storage, expand-external-labels, memory-snapshot-on-shutdown, promql-per-step-stats, promql-experimental-functions, remote-write-receiver (DEPRECATED), extra-scrape-metrics, new-service-discovery-manager, auto-gomaxprocs, no-default-scrape-port, native-histograms, otlp-write-receiver, created-timestamp-zero-ingestion, concurrent-rule-eval. See https://prometheus.io/docs/prometheus/latest/feature_flags/ for more details."). + a.Flag("enable-feature", "Comma separated feature names to enable. Valid options: agent, auto-gomemlimit, exemplar-storage, expand-external-labels, memory-snapshot-on-shutdown, promql-per-step-stats, promql-experimental-functions, remote-write-receiver (DEPRECATED), extra-scrape-metrics, new-service-discovery-manager, auto-gomaxprocs, no-default-scrape-port, native-histograms, created-timestamp-zero-ingestion, concurrent-rule-eval. See https://prometheus.io/docs/prometheus/latest/feature_flags/ for more details."). Default("").StringsVar(&cfg.featureList) promlogflag.AddFlags(a, &cfg.promlogConfig) diff --git a/storage/remote/codec.go b/storage/remote/codec.go index 1228b23..4893340 100644 --- a/storage/remote/codec.go +++ b/storage/remote/codec.go @@ -14,7 +14,6 @@ package remote import ( - "compress/gzip" "errors" "fmt" "io" @@ -28,7 +27,6 @@ import ( "github.com/gogo/protobuf/proto" "github.com/golang/snappy" "github.com/prometheus/common/model" - "go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp" "github.com/prometheus/prometheus/model/exemplar" "github.com/prometheus/prometheus/model/histogram" @@ -40,13 +38,8 @@ import ( "github.com/prometheus/prometheus/util/annotations" ) -const ( - // decodeReadLimit is the maximum size of a read request body in bytes. - decodeReadLimit = 32 * 1024 * 1024 - - pbContentType = "application/x-protobuf" - jsonContentType = "application/json" -) +// decodeReadLimit is the maximum size of a read request body in bytes. +const decodeReadLimit = 32 * 1024 * 1024 type HTTPError struct { msg string @@ -814,56 +807,3 @@ func DecodeWriteRequest(r io.Reader) (*prompb.WriteRequest, error) { return &req, nil } - -func DecodeOTLPWriteRequest(r *http.Request) (pmetricotlp.ExportRequest, error) { - contentType := r.Header.Get("Content-Type") - var decoderFunc func(buf []byte) (pmetricotlp.ExportRequest, error) - switch contentType { - case pbContentType: - decoderFunc = func(buf []byte) (pmetricotlp.ExportRequest, error) { - req := pmetricotlp.NewExportRequest() - return req, req.UnmarshalProto(buf) - } - - case jsonContentType: - decoderFunc = func(buf []byte) (pmetricotlp.ExportRequest, error) { - req := pmetricotlp.NewExportRequest() - return req, req.UnmarshalJSON(buf) - } - - default: - return pmetricotlp.NewExportRequest(), fmt.Errorf("unsupported content type: %s, supported: [%s, %s]", contentType, jsonContentType, pbContentType) - } - - reader := r.Body - // Handle compression. - switch r.Header.Get("Content-Encoding") { - case "gzip": - gr, err := gzip.NewReader(reader) - if err != nil { - return pmetricotlp.NewExportRequest(), err - } - reader = gr - - case "": - // No compression. - - default: - return pmetricotlp.NewExportRequest(), fmt.Errorf("unsupported compression: %s. Only \"gzip\" or no compression supported", r.Header.Get("Content-Encoding")) - } - - body, err := io.ReadAll(reader) - if err != nil { - r.Body.Close() - return pmetricotlp.NewExportRequest(), err - } - if err = r.Body.Close(); err != nil { - return pmetricotlp.NewExportRequest(), err - } - otlpReq, err := decoderFunc(body) - if err != nil { - return pmetricotlp.NewExportRequest(), err - } - - return otlpReq, nil -} diff --git a/storage/remote/write_handler.go b/storage/remote/write_handler.go index ff22729..4bb43be 100644 --- a/storage/remote/write_handler.go +++ b/storage/remote/write_handler.go @@ -28,7 +28,6 @@ import ( "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/prompb" "github.com/prometheus/prometheus/storage" - otlptranslator "github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite" ) type writeHandler struct { @@ -180,56 +179,3 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err return nil } - -// NewOTLPWriteHandler creates a http.Handler that accepts OTLP write requests and -// writes them to the provided appendable. -func NewOTLPWriteHandler(logger log.Logger, appendable storage.Appendable) http.Handler { - rwHandler := &writeHandler{ - logger: logger, - appendable: appendable, - } - - return &otlpWriteHandler{ - logger: logger, - rwHandler: rwHandler, - } -} - -type otlpWriteHandler struct { - logger log.Logger - rwHandler *writeHandler -} - -func (h *otlpWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - req, err := DecodeOTLPWriteRequest(r) - if err != nil { - level.Error(h.logger).Log("msg", "Error decoding remote write request", "err", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - converter := otlptranslator.NewPrometheusConverter() - if err := converter.FromMetrics(req.Metrics(), otlptranslator.Settings{ - AddMetricSuffixes: true, - }); err != nil { - level.Warn(h.logger).Log("msg", "Error translating OTLP metrics to Prometheus write request", "err", err) - } - - err = h.rwHandler.write(r.Context(), &prompb.WriteRequest{ - Timeseries: converter.TimeSeries(), - }) - - switch { - case err == nil: - case errors.Is(err, storage.ErrOutOfOrderSample), errors.Is(err, storage.ErrOutOfBounds), errors.Is(err, storage.ErrDuplicateSampleForTimestamp): - // Indicated an out of order sample is a bad request to prevent retries. - http.Error(w, err.Error(), http.StatusBadRequest) - return - default: - level.Error(h.logger).Log("msg", "Error appending remote write", "err", err.Error()) - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - - w.WriteHeader(http.StatusOK) -} diff --git a/storage/remote/write_test.go b/storage/remote/write_test.go index c79ac3a..1d33ba3 100644 --- a/storage/remote/write_test.go +++ b/storage/remote/write_test.go @@ -14,9 +14,6 @@ package remote import ( - "bytes" - "net/http" - "net/http/httptest" "net/url" "testing" "time" @@ -25,9 +22,6 @@ import ( common_config "github.com/prometheus/common/config" "github.com/prometheus/common/model" "github.com/stretchr/testify/require" - "go.opentelemetry.io/collector/pdata/pcommon" - "go.opentelemetry.io/collector/pdata/pmetric" - "go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp" "github.com/prometheus/prometheus/config" "github.com/prometheus/prometheus/model/labels" @@ -379,107 +373,3 @@ func TestWriteStorageApplyConfigsPartialUpdate(t *testing.T) { err = s.Close() require.NoError(t, err) } - -func TestOTLPWriteHandler(t *testing.T) { - exportRequest := generateOTLPWriteRequest(t) - - buf, err := exportRequest.MarshalProto() - require.NoError(t, err) - - req, err := http.NewRequest("", "", bytes.NewReader(buf)) - require.NoError(t, err) - req.Header.Set("Content-Type", "application/x-protobuf") - - appendable := &mockAppendable{} - handler := NewOTLPWriteHandler(nil, appendable) - - recorder := httptest.NewRecorder() - handler.ServeHTTP(recorder, req) - - resp := recorder.Result() - require.Equal(t, http.StatusOK, resp.StatusCode) - - require.Len(t, appendable.samples, 12) // 1 (counter) + 1 (gauge) + 1 (target_info) + 7 (hist_bucket) + 2 (hist_sum, hist_count) - require.Len(t, appendable.histograms, 1) // 1 (exponential histogram) - require.Len(t, appendable.exemplars, 1) // 1 (exemplar) -} - -func generateOTLPWriteRequest(t *testing.T) pmetricotlp.ExportRequest { - d := pmetric.NewMetrics() - - // Generate One Counter, One Gauge, One Histogram, One Exponential-Histogram - // with resource attributes: service.name="test-service", service.instance.id="test-instance", host.name="test-host" - // with metric attribute: foo.bar="baz" - - timestamp := time.Now() - - resourceMetric := d.ResourceMetrics().AppendEmpty() - resourceMetric.Resource().Attributes().PutStr("service.name", "test-service") - resourceMetric.Resource().Attributes().PutStr("service.instance.id", "test-instance") - resourceMetric.Resource().Attributes().PutStr("host.name", "test-host") - - scopeMetric := resourceMetric.ScopeMetrics().AppendEmpty() - - // Generate One Counter - counterMetric := scopeMetric.Metrics().AppendEmpty() - counterMetric.SetName("test-counter") - counterMetric.SetDescription("test-counter-description") - counterMetric.SetEmptySum() - counterMetric.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) - counterMetric.Sum().SetIsMonotonic(true) - - counterDataPoint := counterMetric.Sum().DataPoints().AppendEmpty() - counterDataPoint.SetTimestamp(pcommon.NewTimestampFromTime(timestamp)) - counterDataPoint.SetDoubleValue(10.0) - counterDataPoint.Attributes().PutStr("foo.bar", "baz") - - counterExemplar := counterDataPoint.Exemplars().AppendEmpty() - counterExemplar.SetTimestamp(pcommon.NewTimestampFromTime(timestamp)) - counterExemplar.SetDoubleValue(10.0) - counterExemplar.SetSpanID(pcommon.SpanID{0, 1, 2, 3, 4, 5, 6, 7}) - counterExemplar.SetTraceID(pcommon.TraceID{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}) - - // Generate One Gauge - gaugeMetric := scopeMetric.Metrics().AppendEmpty() - gaugeMetric.SetName("test-gauge") - gaugeMetric.SetDescription("test-gauge-description") - gaugeMetric.SetEmptyGauge() - - gaugeDataPoint := gaugeMetric.Gauge().DataPoints().AppendEmpty() - gaugeDataPoint.SetTimestamp(pcommon.NewTimestampFromTime(timestamp)) - gaugeDataPoint.SetDoubleValue(10.0) - gaugeDataPoint.Attributes().PutStr("foo.bar", "baz") - - // Generate One Histogram - histogramMetric := scopeMetric.Metrics().AppendEmpty() - histogramMetric.SetName("test-histogram") - histogramMetric.SetDescription("test-histogram-description") - histogramMetric.SetEmptyHistogram() - histogramMetric.Histogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) - - histogramDataPoint := histogramMetric.Histogram().DataPoints().AppendEmpty() - histogramDataPoint.SetTimestamp(pcommon.NewTimestampFromTime(timestamp)) - histogramDataPoint.ExplicitBounds().FromRaw([]float64{0.0, 1.0, 2.0, 3.0, 4.0, 5.0}) - histogramDataPoint.BucketCounts().FromRaw([]uint64{2, 2, 2, 2, 2, 2}) - histogramDataPoint.SetCount(10) - histogramDataPoint.SetSum(30.0) - histogramDataPoint.Attributes().PutStr("foo.bar", "baz") - - // Generate One Exponential-Histogram - exponentialHistogramMetric := scopeMetric.Metrics().AppendEmpty() - exponentialHistogramMetric.SetName("test-exponential-histogram") - exponentialHistogramMetric.SetDescription("test-exponential-histogram-description") - exponentialHistogramMetric.SetEmptyExponentialHistogram() - exponentialHistogramMetric.ExponentialHistogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) - - exponentialHistogramDataPoint := exponentialHistogramMetric.ExponentialHistogram().DataPoints().AppendEmpty() - exponentialHistogramDataPoint.SetTimestamp(pcommon.NewTimestampFromTime(timestamp)) - exponentialHistogramDataPoint.SetScale(2.0) - exponentialHistogramDataPoint.Positive().BucketCounts().FromRaw([]uint64{2, 2, 2, 2, 2}) - exponentialHistogramDataPoint.SetZeroCount(2) - exponentialHistogramDataPoint.SetCount(10) - exponentialHistogramDataPoint.SetSum(30.0) - exponentialHistogramDataPoint.Attributes().PutStr("foo.bar", "baz") - - return pmetricotlp.NewExportRequestFromMetrics(d) -} diff --git a/web/api/v1/api.go b/web/api/v1/api.go index f088492..7385042 100644 --- a/web/api/v1/api.go +++ b/web/api/v1/api.go @@ -214,7 +214,6 @@ type API struct { remoteWriteHandler http.Handler remoteReadHandler http.Handler - otlpWriteHandler http.Handler codecs []Codec } @@ -247,8 +246,6 @@ func NewAPI( gatherer prometheus.Gatherer, registerer prometheus.Registerer, statsRenderer StatsRenderer, - rwEnabled bool, - otlpEnabled bool, ) *API { a := &API{ QueryEngine: qe, @@ -285,16 +282,9 @@ func NewAPI( a.statsRenderer = statsRenderer } - if ap == nil && (rwEnabled || otlpEnabled) { - panic("remote write or otlp write enabled, but no appender passed in.") - } - - if rwEnabled { + if ap != nil { a.remoteWriteHandler = remote.NewWriteHandler(logger, registerer, ap) } - if otlpEnabled { - a.otlpWriteHandler = remote.NewOTLPWriteHandler(logger, ap) - } return a } @@ -387,7 +377,6 @@ func (api *API) Register(r *route.Router) { r.Get("/status/walreplay", api.serveWALReplayStatus) r.Post("/read", api.ready(api.remoteRead)) r.Post("/write", api.ready(api.remoteWrite)) - r.Post("/otlp/v1/metrics", api.ready(api.otlpWrite)) r.Get("/alerts", wrapAgent(api.alerts)) r.Get("/rules", wrapAgent(api.rules)) @@ -1656,14 +1645,6 @@ func (api *API) remoteWrite(w http.ResponseWriter, r *http.Request) { } } -func (api *API) otlpWrite(w http.ResponseWriter, r *http.Request) { - if api.otlpWriteHandler != nil { - api.otlpWriteHandler.ServeHTTP(w, r) - } else { - http.Error(w, "otlp write receiver needs to be enabled with --enable-feature=otlp-write-receiver", http.StatusNotFound) - } -} - func (api *API) deleteSeries(r *http.Request) apiFuncResult { if !api.enableAdmin { return apiFuncResult{nil, &apiError{errorUnavailable, errors.New("admin APIs disabled")}, nil, nil} diff --git a/web/api/v1/errors_test.go b/web/api/v1/errors_test.go index e76a1a3..6220199 100644 --- a/web/api/v1/errors_test.go +++ b/web/api/v1/errors_test.go @@ -134,8 +134,6 @@ func createPrometheusAPI(q storage.SampleAndChunkQueryable) *route.Router { prometheus.DefaultGatherer, nil, nil, - false, - false, ) promRouter := route.New().WithPrefix("/api/v1") diff --git a/web/web.go b/web/web.go index a3f049b..5075d1e 100644 --- a/web/web.go +++ b/web/web.go @@ -261,7 +261,6 @@ type Options struct { RemoteReadConcurrencyLimit int RemoteReadBytesInFrame int EnableRemoteWriteReceiver bool - EnableOTLPWriteReceiver bool IsAgent bool AppName string @@ -320,7 +319,7 @@ func New(logger log.Logger, o *Options) *Handler { FactoryRr := func(_ context.Context) api_v1.RulesRetriever { return h.ruleManager } var app storage.Appendable - if o.EnableRemoteWriteReceiver || o.EnableOTLPWriteReceiver { + if o.EnableRemoteWriteReceiver { app = h.storage } @@ -352,8 +351,6 @@ func New(logger log.Logger, o *Options) *Handler { o.Gatherer, o.Registerer, nil, - o.EnableRemoteWriteReceiver, - o.EnableOTLPWriteReceiver, ) if o.RoutePrefix != "/" {