exp/api: Add experimental exp module; Add remote API with write client and handler. (#1658)

* api: Add remote API with write client; add remote handler.

Signed-off-by: bwplotka <[email protected]>

* Make Write message type more flexble, address some feedback (#1710)

* Address remaining feedback

Signed-off-by: Saswata Mukherjee <[email protected]>

* Make Write message type more flexible

Signed-off-by: Saswata Mukherjee <[email protected]>

---------

Signed-off-by: Saswata Mukherjee <[email protected]>

* Move remote write API to client_golang/exp (#1711)

* Move remote write API to client_golang/exp

Signed-off-by: Saswata Mukherjee <[email protected]>

* Don't use api.Client structs, add options for middleware

Signed-off-by: Saswata Mukherjee <[email protected]>

* Fix reqBuf usage

Signed-off-by: Saswata Mukherjee <[email protected]>

* Fix url path

Signed-off-by: Saswata Mukherjee <[email protected]>

* Add separate mod file (and workspace file)

Signed-off-by: Saswata Mukherjee <[email protected]>

* Hook exp tests fmt; Test handler error case; Configure backoff

Signed-off-by: Saswata Mukherjee <[email protected]>

---------

Signed-off-by: Saswata Mukherjee <[email protected]>

* exp: Add README, address feedback, use sync.Pool (#1747)

* Implement suggestion for Store interface and contentType

Signed-off-by: Saswata Mukherjee <[email protected]>

* Add README

Signed-off-by: Saswata Mukherjee <[email protected]>

* Use sync.Pool

Signed-off-by: Saswata Mukherjee <[email protected]>

* Implement review suggestions

Signed-off-by: Saswata Mukherjee <[email protected]>

* Release bufs right after compressPayload

Signed-off-by: Saswata Mukherjee <[email protected]>

---------

Signed-off-by: Saswata Mukherjee <[email protected]>

* Bump exp to go 1.22

Signed-off-by: Saswata Mukherjee <[email protected]>

---------

Signed-off-by: bwplotka <[email protected]>
Signed-off-by: Saswata Mukherjee <[email protected]>
Signed-off-by: Bartlomiej Plotka <[email protected]>
Co-authored-by: Saswata Mukherjee <[email protected]>

Author: bwplotka

.bingo/Variables.mk

@@ -1,4 +1,4 @@
-# Auto generated binary variables helper managed by https://github.com/bwplotka/bingo v0.8. DO NOT EDIT.
+# Auto generated binary variables helper managed by https://github.com/bwplotka/bingo v0.9. DO NOT EDIT.
 # All tools are designed to be build inside $GOBIN.
 BINGO_DIR := $(dir $(lastword $(MAKEFILE_LIST)))
 GOPATH ?= $(shell go env GOPATH)
@@ -7,16 +7,22 @@ GO     ?= $(shell which go)
 
 # Below generated variables ensure that every time a tool under each variable is invoked, the correct version
 # will be used; reinstalling only if needed.
-# For example for goimports variable:
+# For example for buf variable:
 #
 # In your main Makefile (for non array binaries):
 #
 #include .bingo/Variables.mk # Assuming -dir was set to .bingo .
 #
-#command: $(GOIMPORTS)
-#	@echo "Running goimports"
-#	@$(GOIMPORTS) <flags/args..>
+#command: $(BUF)
+#	@echo "Running buf"
+#	@$(BUF) <flags/args..>
 #
+BUF := $(GOBIN)/buf-v1.39.0
+$(BUF): $(BINGO_DIR)/buf.mod
+	@# Install binary/ries using Go 1.14+ build command. This is using bwplotka/bingo-controlled, separate go module with pinned dependencies.
+	@echo "(re)installing $(GOBIN)/buf-v1.39.0"
+	@cd $(BINGO_DIR) && GOWORK=off $(GO) build -mod=mod -modfile=buf.mod -o=$(GOBIN)/buf-v1.39.0 "github.com/bufbuild/buf/cmd/buf"
+
 GOIMPORTS := $(GOBIN)/goimports-v0.9.3
 $(GOIMPORTS): $(BINGO_DIR)/goimports.mod
 	@# Install binary/ries using Go 1.14+ build command. This is using bwplotka/bingo-controlled, separate go module with pinned dependencies.

.bingo/buf.mod

@@ -0,0 +1,5 @@
+module _ // Auto generated by https://github.com/bwplotka/bingo. DO NOT EDIT
+
+go 1.22.6
+
+require github.com/bufbuild/buf v1.39.0 // cmd/buf

.bingo/buf.sum

@@ -1,4 +1,4 @@
-# Auto generated binary variables helper managed by https://github.com/bwplotka/bingo v0.8. DO NOT EDIT.
+# Auto generated binary variables helper managed by https://github.com/bwplotka/bingo v0.9. DO NOT EDIT.
 # All tools are designed to be build inside $GOBIN.
 # Those variables will work only until 'bingo get' was invoked, or if tools were installed via Makefile's Variables.mk.
 GOBIN=${GOBIN:=$(go env GOBIN)}
@@ -8,5 +8,7 @@ if [ -z "$GOBIN" ]; then
 fi
 
 
+BUF="${GOBIN}/buf-v1.39.0"
+
 GOIMPORTS="${GOBIN}/goimports-v0.9.3"
 

.bingo/variables.env

@@ -14,11 +14,16 @@
 include .bingo/Variables.mk
 include Makefile.common
 
+.PHONY: deps
+deps:
+	$(GO) work sync
+	$(MAKE) common-deps
+
 .PHONY: test
-test: deps common-test
+test: deps common-test test-exp
 
 .PHONY: test-short
-test-short: deps common-test-short
+test-short: deps common-test-short test-exp-short
 
 .PHONY: generate-go-collector-test-files
 file := supported_go_versions.txt
@@ -35,5 +40,24 @@ generate-go-collector-test-files:
 	go mod tidy
 
 .PHONY: fmt
-fmt: common-format
+fmt: common-format $(GOIMPORTS)
 	$(GOIMPORTS) -local github.com/prometheus/client_golang -w .
+
+.PHONY: proto
+proto: ## Regenerate Go from remote write proto.
+proto: $(BUF)
+	@echo ">> regenerating Prometheus Remote Write proto"
+	@cd exp/api/remote/genproto && $(BUF) generate
+	@cd exp/api/remote && find genproto/ -type f -exec sed -i '' 's/protohelpers "github.com\/planetscale\/vtprotobuf\/protohelpers"/protohelpers "github.com\/prometheus\/client_golang\/exp\/internal\/github.com\/planetscale\/vtprotobuf\/protohelpers"/g' {} \;
+	# For some reasons buf generates this unused import, kill it manually for now and reformat.
+	@cd exp/api/remote && find genproto/ -type f -exec sed -i '' 's/_ "github.com\/gogo\/protobuf\/gogoproto"//g' {} \;
+	@cd exp/api/remote && go fmt ./genproto/...
+	$(MAKE) fmt
+
+.PHONY: test-exp
+test-exp:
+	cd exp && $(GOTEST) $(test-flags) $(GOOPTS) $(pkgs)
+
+.PHONY: test-exp-short
+test-exp-short:
+	cd exp && $(GOTEST) -short $(GOOPTS) $(pkgs)

Makefile

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"io"
 	"net"
 	"net/http"
 	"net/url"
@@ -133,7 +134,8 @@ func (c *httpClient) Do(ctx context.Context, req *http.Request) (*http.Response,
 	resp, err := c.client.Do(req)
 	defer func() {
 		if resp != nil {
-			resp.Body.Close()
+			_, _ = io.Copy(io.Discard, resp.Body)
+			_ = resp.Body.Close()
 		}
 	}()
 
@@ -145,6 +147,7 @@ func (c *httpClient) Do(ctx context.Context, req *http.Request) (*http.Response,
 	done := make(chan struct{})
 	go func() {
 		var buf bytes.Buffer
+		// TODO(bwplotka): Add LimitReader for too long err messages (e.g. limit by 1KB)
 		_, err = buf.ReadFrom(resp.Body)
 		body = buf.Bytes()
 		close(done)

api/client.go

@@ -0,0 +1,62 @@
+# client_golang experimental module
+
+Contains experimental utilities and APIs for Prometheus.
+The module may be contain breaking changes or be removed in the future.
+
+Packages within this module are listed below.
+
+## Remote
+
+Implements bindings from Prometheus remote APIs (remote write v1 and v2 for now).
+
+Contains flexible method for building API clients, that can send remote write protocol messages.
+
+```go
+    import (
+        "github.com/prometheus/client_golang/exp/api/remote"
+    )
+    ...
+
+	remoteAPI, err := remote.NewAPI(
+		"https://your-remote-endpoint",
+		remote.WithAPIHTTPClient(httpClient),
+		remote.WithAPILogger(logger.With("component", "remote_write_api")),
+	)
+    ...
+
+    stats, err := remoteAPI.Write(ctx, remote.WriteV2MessageType, protoWriteReq)
+```
+
+Also contains handler methods for applications that would like to handle and store remote write requests.
+
+```go
+    import (
+        "net/http"
+        "log"
+
+        "github.com/prometheus/client_golang/exp/api/remote"
+    )
+    ...
+    
+    type db {}
+
+    func NewStorage() *db {}
+
+    func (d *db) Store(ctx context.Context, msgType remote.WriteMessageType, req *http.Request) (*remote.WriteResponse, error) {}
+    ...
+
+	mux := http.NewServeMux()
+
+	remoteWriteHandler := remote.NewHandler(storage, remote.WithHandlerLogger(logger.With("component", "remote_write_handler")))
+	mux.Handle("/api/v1/write", remoteWriteHandler)
+
+	server := &http.Server{
+		Addr:    ":8080",
+		Handler: mux,
+	}
+	if err := server.ListenAndServe(); err != nil {
+		log.Fatal(err)
+	}
+```
+
+For more details, see [go doc](https://pkg.go.dev/github.com/prometheus/client_golang/exp/api/remote).

exp/README.md

@@ -0,0 +1,21 @@
+# buf.gen.yaml
+version: v2
+
+plugins:
+- remote: buf.build/protocolbuffers/go:v1.31.0
+  out: .
+  opt:
+  - Mio/prometheus/write/v2/types.proto=./v2
+
+# vtproto for efficiency utilities like pooling etc.
+# https://buf.build/community/planetscale-vtprotobuf?version=v0.6.0
+- remote: buf.build/community/planetscale-vtprotobuf:v0.6.0
+  out: .
+  opt:
+  - Mio/prometheus/write/v2/types.proto=./v2
+  - features=marshal+unmarshal+size
+
+inputs:
+- module: buf.build/prometheus/prometheus:5b212ab78fb7460e831cf7ff2d83e385
+  types:
+  - "io.prometheus.write.v2.Request"

exp/api/remote/genproto/buf.gen.yaml

@@ -0,0 +1,97 @@
+// Copyright (c) Bartłomiej Płotka @bwplotka
+// Licensed under the Apache License 2.0.
+
+// Copyright 2024 Google LLC
+//
+// 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
+//
+//     https://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.
+
+// Copyright 2024 Prometheus Team
+// 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 writev2
+
+// SymbolsTable implements table for easy symbol use.
+type SymbolsTable struct {
+	strings    []string
+	symbolsMap map[string]uint32
+}
+
+// NewSymbolTable returns a symbol table.
+func NewSymbolTable() SymbolsTable {
+	return SymbolsTable{
+		// Empty string is required as a first element.
+		symbolsMap: map[string]uint32{"": 0},
+		strings:    []string{""},
+	}
+}
+
+// Symbolize adds (if not added before) a string to the symbols table,
+// while returning its reference number.
+func (t *SymbolsTable) Symbolize(str string) uint32 {
+	if ref, ok := t.symbolsMap[str]; ok {
+		return ref
+	}
+	ref := uint32(len(t.strings))
+	t.strings = append(t.strings, str)
+	t.symbolsMap[str] = ref
+	return ref
+}
+
+// SymbolizeLabels symbolize Prometheus labels.
+func (t *SymbolsTable) SymbolizeLabels(lbls []string, buf []uint32) []uint32 {
+	result := buf[:0]
+	for i := 0; i < len(lbls); i += 2 {
+		off := t.Symbolize(lbls[i])
+		result = append(result, off)
+		off = t.Symbolize(lbls[i+1])
+		result = append(result, off)
+	}
+	return result
+}
+
+// Symbols returns computes symbols table to put in e.g. Request.Symbols.
+// As per spec, order does not matter.
+func (t *SymbolsTable) Symbols() []string {
+	return t.strings
+}
+
+// Reset clears symbols table.
+func (t *SymbolsTable) Reset() {
+	// NOTE: Make sure to keep empty symbol.
+	t.strings = t.strings[:1]
+	for k := range t.symbolsMap {
+		if k == "" {
+			continue
+		}
+		delete(t.symbolsMap, k)
+	}
+}
+
+// DesymbolizeLabels decodes label references, with given symbols to labels.
+func DesymbolizeLabels(labelRefs []uint32, symbols, buf []string) []string {
+	result := buf[:0]
+	for i := 0; i < len(labelRefs); i += 2 {
+		result = append(result, symbols[labelRefs[i]], symbols[labelRefs[i+1]])
+	}
+	return result
+}

exp/api/remote/genproto/v2/symbols.go

@@ -0,0 +1,80 @@
+// Copyright (c) Bartłomiej Płotka @bwplotka
+// Licensed under the Apache License 2.0.
+
+// Copyright 2024 Google LLC
+//
+// 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
+//
+//     https://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.
+
+// Copyright 2024 Prometheus Team
+// 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 writev2
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func requireEqual(t testing.TB, expected, got any) {
+	if diff := cmp.Diff(expected, got); diff != "" {
+		t.Fatal(diff)
+	}
+}
+
+func TestSymbolsTable(t *testing.T) {
+	s := NewSymbolTable()
+	requireEqual(t, []string{""}, s.Symbols())
+	requireEqual(t, uint32(0), s.Symbolize(""))
+	requireEqual(t, []string{""}, s.Symbols())
+
+	requireEqual(t, uint32(1), s.Symbolize("abc"))
+	requireEqual(t, []string{"", "abc"}, s.Symbols())
+
+	requireEqual(t, uint32(2), s.Symbolize("__name__"))
+	requireEqual(t, []string{"", "abc", "__name__"}, s.Symbols())
+
+	requireEqual(t, uint32(3), s.Symbolize("foo"))
+	requireEqual(t, []string{"", "abc", "__name__", "foo"}, s.Symbols())
+
+	s.Reset()
+	requireEqual(t, []string{""}, s.Symbols())
+	requireEqual(t, uint32(0), s.Symbolize(""))
+
+	requireEqual(t, uint32(1), s.Symbolize("__name__"))
+	requireEqual(t, []string{"", "__name__"}, s.Symbols())
+
+	requireEqual(t, uint32(2), s.Symbolize("abc"))
+	requireEqual(t, []string{"", "__name__", "abc"}, s.Symbols())
+
+	ls := []string{"__name__", "qwer", "zxcv", "1234"}
+	encoded := s.SymbolizeLabels(ls, nil)
+	requireEqual(t, []uint32{1, 3, 4, 5}, encoded)
+	decoded := DesymbolizeLabels(encoded, s.Symbols(), nil)
+	requireEqual(t, ls, decoded)
+
+	// Different buf.
+	ls = []string{"__name__", "qwer", "zxcv2222", "1234"}
+	encoded = s.SymbolizeLabels(ls, []uint32{1, 3, 4, 5})
+	requireEqual(t, []uint32{1, 3, 6, 5}, encoded)
+}