Merge 7a774d04ea into 13851e9287

Signed-off-by: prombot <prometheus-team@googlegroups.com>

.github/workflows/golangci-lint.yml

@@ -36,4 +36,4 @@ jobs:
         uses: golangci/golangci-lint-action@971e284b6050e8a5849b72094c50ab08da042db8 # v6.1.1
         with:
           args: --verbose
-          version: v1.60.2
+          version: v1.61.0

.golangci.yml

@@ -31,6 +31,7 @@ linters:
     - staticcheck
     - unconvert
     - unused
+    - usestdlibvars
     - wastedassign
 
 issues:

Makefile.common

@@ -61,7 +61,7 @@ PROMU_URL     := https://github.com/prometheus/promu/releases/download/v$(PROMU_
 SKIP_GOLANGCI_LINT :=
 GOLANGCI_LINT :=
 GOLANGCI_LINT_OPTS ?=
-GOLANGCI_LINT_VERSION ?= v1.60.2
+GOLANGCI_LINT_VERSION ?= v1.61.0
 # golangci-lint only supports linux, darwin and windows platforms on i386/amd64/arm64.
 # windows isn't included here because of the path separator being different.
 ifeq ($(GOHOSTOS),$(filter $(GOHOSTOS),linux darwin))

examples/dbstatscollectorexample/main.go

@@ -0,0 +1,64 @@
+// Copyright 2022 The Prometheus 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.
+
+// A minimal example of how to include Prometheus instrumentation for database stats.
+package main
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+var addr = flag.String("listen-address", ":8080", "The address to listen on for HTTP requests.")
+
+func main() {
+	flag.Parse()
+
+	// Set up an in-memory SQLite DB.
+	db, err := sql.Open("sqlite3", ":memory:") // In-memory SQLite database
+	if err != nil {
+		log.Fatalf("Failed to connect to in-memory database: %v", err)
+	}
+	defer db.Close()
+
+	// Set connection pool limits to simulate more activity.
+	db.SetMaxOpenConns(10)
+	db.SetMaxIdleConns(5)
+	db.SetConnMaxIdleTime(5 * time.Minute)
+	db.SetConnMaxLifetime(30 * time.Minute)
+
+	// Create a new Prometheus registry.
+	reg := prometheus.NewRegistry()
+
+	// Create and register the DB stats collector.
+	dbStatsCollector := collectors.NewDBStatsCollector(db, "sqlite_in_memory")
+	reg.MustRegister(dbStatsCollector)
+
+	// Expose the registered metrics via HTTP.
+	http.Handle("/metrics", promhttp.HandlerFor(
+		reg,
+		promhttp.HandlerOpts{},
+	))
+
+	fmt.Println("Server is running, metrics are available at /metrics")
+	log.Fatal(http.ListenAndServe(*addr, nil))
+}

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.17.11
 	github.com/kylelemons/godebug v1.1.0
+	github.com/mattn/go-sqlite3 v1.14.24
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.60.1
 	github.com/prometheus/procfs v0.15.1

go.sum

@@ -21,6 +21,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/mattn/go-sqlite3 v1.14.24 h1:tpSp2G2KyMnnQu99ngJ47EIkWVmliIizyZBfPrBWDRM=
+github.com/mattn/go-sqlite3 v1.14.24/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

prometheus/atomic_update.go

@@ -0,0 +1,50 @@
+// Copyright 2014 The Prometheus 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 (
+	"math"
+	"sync/atomic"
+	"time"
+)
+
+// atomicUpdateFloat atomically updates the float64 value pointed to by bits
+// using the provided updateFunc, with an exponential backoff on contention.
+func atomicUpdateFloat(bits *uint64, updateFunc func(float64) float64) {
+	const (
+		// both numbers are derived from empirical observations
+		// documented in this PR: https://github.com/prometheus/client_golang/pull/1661
+		maxBackoff     = 320 * time.Millisecond
+		initialBackoff = 10 * time.Millisecond
+	)
+	backoff := initialBackoff
+
+	for {
+		loadedBits := atomic.LoadUint64(bits)
+		oldFloat := math.Float64frombits(loadedBits)
+		newFloat := updateFunc(oldFloat)
+		newBits := math.Float64bits(newFloat)
+
+		if atomic.CompareAndSwapUint64(bits, loadedBits, newBits) {
+			break
+		} else {
+			// Exponential backoff with sleep and cap to avoid infinite wait
+			time.Sleep(backoff)
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
+	}
+}

prometheus/atomic_update_test.go

@@ -0,0 +1,167 @@
+// Copyright 2014 The Prometheus 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 (
+	"math"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"unsafe"
+)
+
+var output float64
+
+func TestAtomicUpdateFloat(t *testing.T) {
+	var val float64 = 0.0
+	bits := (*uint64)(unsafe.Pointer(&val))
+	var wg sync.WaitGroup
+	numGoroutines := 100000
+	increment := 1.0
+
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			atomicUpdateFloat(bits, func(f float64) float64 {
+				return f + increment
+			})
+		}()
+	}
+
+	wg.Wait()
+	expected := float64(numGoroutines) * increment
+	if val != expected {
+		t.Errorf("Expected %f, got %f", expected, val)
+	}
+}
+
+// Benchmark for atomicUpdateFloat with single goroutine (no contention).
+func BenchmarkAtomicUpdateFloat_SingleGoroutine(b *testing.B) {
+	var val float64 = 0.0
+	bits := (*uint64)(unsafe.Pointer(&val))
+
+	for i := 0; i < b.N; i++ {
+		atomicUpdateFloat(bits, func(f float64) float64 {
+			return f + 1.0
+		})
+	}
+
+	output = val
+}
+
+// Benchmark for old implementation with single goroutine (no contention) -> to check overhead of backoff
+func BenchmarkAtomicNoBackoff_SingleGoroutine(b *testing.B) {
+	var val float64 = 0.0
+	bits := (*uint64)(unsafe.Pointer(&val))
+
+	for i := 0; i < b.N; i++ {
+		for {
+			loadedBits := atomic.LoadUint64(bits)
+			newBits := math.Float64bits(math.Float64frombits(loadedBits) + 1.0)
+			if atomic.CompareAndSwapUint64(bits, loadedBits, newBits) {
+				break
+			}
+		}
+	}
+
+	output = val
+}
+
+// Benchmark varying the number of goroutines.
+func benchmarkAtomicUpdateFloatConcurrency(b *testing.B, numGoroutines int) {
+	var val float64 = 0.0
+	bits := (*uint64)(unsafe.Pointer(&val))
+	b.SetParallelism(numGoroutines)
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			atomicUpdateFloat(bits, func(f float64) float64 {
+				return f + 1.0
+			})
+		}
+	})
+
+	output = val
+}
+
+func benchmarkAtomicNoBackoffFloatConcurrency(b *testing.B, numGoroutines int) {
+	var val float64 = 0.0
+	bits := (*uint64)(unsafe.Pointer(&val))
+	b.SetParallelism(numGoroutines)
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			for {
+				loadedBits := atomic.LoadUint64(bits)
+				newBits := math.Float64bits(math.Float64frombits(loadedBits) + 1.0)
+				if atomic.CompareAndSwapUint64(bits, loadedBits, newBits) {
+					break
+				}
+			}
+		}
+	})
+
+	output = val
+}
+
+func BenchmarkAtomicUpdateFloat_1Goroutine(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 1)
+}
+
+func BenchmarkAtomicNoBackoff_1Goroutine(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 1)
+}
+
+func BenchmarkAtomicUpdateFloat_2Goroutines(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 2)
+}
+
+func BenchmarkAtomicNoBackoff_2Goroutines(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 2)
+}
+
+func BenchmarkAtomicUpdateFloat_4Goroutines(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 4)
+}
+
+func BenchmarkAtomicNoBackoff_4Goroutines(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 4)
+}
+
+func BenchmarkAtomicUpdateFloat_8Goroutines(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 8)
+}
+
+func BenchmarkAtomicNoBackoff_8Goroutines(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 8)
+}
+
+func BenchmarkAtomicUpdateFloat_16Goroutines(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 16)
+}
+
+func BenchmarkAtomicNoBackoff_16Goroutines(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 16)
+}
+
+func BenchmarkAtomicUpdateFloat_32Goroutines(b *testing.B) {
+	benchmarkAtomicUpdateFloatConcurrency(b, 32)
+}
+
+func BenchmarkAtomicNoBackoff_32Goroutines(b *testing.B) {
+	benchmarkAtomicNoBackoffFloatConcurrency(b, 32)
+}

prometheus/counter.go

@@ -134,13 +134,9 @@ func (c *counter) Add(v float64) {
 		return
 	}
 
-	for {
-		oldBits := atomic.LoadUint64(&c.valBits)
-		newBits := math.Float64bits(math.Float64frombits(oldBits) + v)
-		if atomic.CompareAndSwapUint64(&c.valBits, oldBits, newBits) {
-			return
-		}
-	}
+	atomicUpdateFloat(&c.valBits, func(oldVal float64) float64 {
+		return oldVal + v
+	})
 }
 
 func (c *counter) AddWithExemplar(v float64, e Labels) {

prometheus/gauge.go

@@ -120,13 +120,9 @@ func (g *gauge) Dec() {
 }
 
 func (g *gauge) Add(val float64) {
-	for {
-		oldBits := atomic.LoadUint64(&g.valBits)
-		newBits := math.Float64bits(math.Float64frombits(oldBits) + val)
-		if atomic.CompareAndSwapUint64(&g.valBits, oldBits, newBits) {
-			return
-		}
-	}
+	atomicUpdateFloat(&g.valBits, func(oldVal float64) float64 {
+		return oldVal + val
+	})
 }
 
 func (g *gauge) Sub(val float64) {

prometheus/histogram.go

@@ -1641,13 +1641,9 @@ func waitForCooldown(count uint64, counts *histogramCounts) {
 // atomicAddFloat adds the provided float atomically to another float
 // represented by the bit pattern the bits pointer is pointing to.
 func atomicAddFloat(bits *uint64, v float64) {
-	for {
-		loadedBits := atomic.LoadUint64(bits)
-		newBits := math.Float64bits(math.Float64frombits(loadedBits) + v)
-		if atomic.CompareAndSwapUint64(bits, loadedBits, newBits) {
-			break
-		}
-	}
+	atomicUpdateFloat(bits, func(oldVal float64) float64 {
+		return oldVal + v
+	})
 }
 
 // atomicDecUint32 atomically decrements the uint32 p points to.  See

prometheus/promhttp/http_test.go

@@ -131,7 +131,7 @@ func TestHandlerErrorHandling(t *testing.T) {
 	logger := log.New(logBuf, "", 0)
 
 	writer := httptest.NewRecorder()
-	request, _ := http.NewRequest("GET", "/", nil)
+	request, _ := http.NewRequest(http.MethodGet, "/", nil)
 	request.Header.Add("Accept", "test/plain")
 
 	mReg := &mockTransactionGatherer{g: reg}
@@ -252,7 +252,7 @@ func TestInstrumentMetricHandler(t *testing.T) {
 	// Do it again to test idempotency.
 	InstrumentMetricHandler(reg, HandlerForTransactional(mReg, HandlerOpts{}))
 	writer := httptest.NewRecorder()
-	request, _ := http.NewRequest("GET", "/", nil)
+	request, _ := http.NewRequest(http.MethodGet, "/", nil)
 	request.Header.Add(acceptHeader, acceptTextPlain)
 
 	handler.ServeHTTP(writer, request)
@@ -311,7 +311,7 @@ func TestHandlerMaxRequestsInFlight(t *testing.T) {
 	w1 := httptest.NewRecorder()
 	w2 := httptest.NewRecorder()
 	w3 := httptest.NewRecorder()
-	request, _ := http.NewRequest("GET", "/", nil)
+	request, _ := http.NewRequest(http.MethodGet, "/", nil)
 	request.Header.Add(acceptHeader, acceptTextPlain)
 
 	c := blockingCollector{Block: make(chan struct{}), CollectStarted: make(chan struct{}, 1)}
@@ -348,7 +348,7 @@ func TestHandlerTimeout(t *testing.T) {
 	handler := HandlerFor(reg, HandlerOpts{Timeout: time.Millisecond})
 	w := httptest.NewRecorder()
 
-	request, _ := http.NewRequest("GET", "/", nil)
+	request, _ := http.NewRequest(http.MethodGet, "/", nil)
 	request.Header.Add("Accept", "test/plain")
 
 	c := blockingCollector{Block: make(chan struct{}), CollectStarted: make(chan struct{}, 1)}
@@ -372,7 +372,7 @@ func TestInstrumentMetricHandlerWithCompression(t *testing.T) {
 	handler := InstrumentMetricHandler(reg, HandlerForTransactional(mReg, HandlerOpts{DisableCompression: false}))
 	compression := Zstd
 	writer := httptest.NewRecorder()
-	request, _ := http.NewRequest("GET", "/", nil)
+	request, _ := http.NewRequest(http.MethodGet, "/", nil)
 	request.Header.Add(acceptHeader, acceptTextPlain)
 	request.Header.Add(acceptEncodingHeader, string(compression))
 
@@ -533,7 +533,7 @@ func TestNegotiateEncodingWriter(t *testing.T) {
 	}
 
 	for _, test := range testCases {
-		request, _ := http.NewRequest("GET", "/", nil)
+		request, _ := http.NewRequest(http.MethodGet, "/", nil)
 		request.Header.Add(acceptEncodingHeader, test.acceptEncoding)
 		rr := httptest.NewRecorder()
 		_, encodingHeader, _, err := negotiateEncodingWriter(request, rr, test.offeredCompressions)
@@ -631,7 +631,7 @@ func BenchmarkCompression(b *testing.B) {
 			b.Run(benchmark.name+"_"+size.name, func(b *testing.B) {
 				for i := 0; i < b.N; i++ {
 					writer := httptest.NewRecorder()
-					request, _ := http.NewRequest("GET", "/", nil)
+					request, _ := http.NewRequest(http.MethodGet, "/", nil)
 					request.Header.Add(acceptEncodingHeader, benchmark.compressionType)
 					handler.ServeHTTP(writer, request)
 				}

prometheus/promhttp/instrument_client_test.go

@@ -223,7 +223,7 @@ func TestClientMiddlewareAPI_WithRequestContext(t *testing.T) {
 	}))
 	defer backend.Close()
 
-	req, err := http.NewRequest("GET", backend.URL, nil)
+	req, err := http.NewRequest(http.MethodGet, backend.URL, nil)
 	if err != nil {
 		t.Fatalf("%v", err)
 	}
@@ -276,7 +276,7 @@ func TestClientMiddlewareAPIWithRequestContextTimeout(t *testing.T) {
 	}))
 	defer backend.Close()
 
-	req, err := http.NewRequest("GET", backend.URL, nil)
+	req, err := http.NewRequest(http.MethodGet, backend.URL, nil)
 	if err != nil {
 		t.Fatalf("%v", err)
 	}

prometheus/promhttp/instrument_server_test.go

@@ -418,7 +418,7 @@ func TestMiddlewareAPI(t *testing.T) {
 		_, _ = w.Write([]byte("OK"))
 	})
 
-	r, _ := http.NewRequest("GET", "www.example.com", nil)
+	r, _ := http.NewRequest(http.MethodGet, "www.example.com", nil)
 	w := httptest.NewRecorder()
 	chain.ServeHTTP(w, r)
 
@@ -432,7 +432,7 @@ func TestMiddlewareAPI_WithExemplars(t *testing.T) {
 		_, _ = w.Write([]byte("OK"))
 	}, WithExemplarFromContext(func(_ context.Context) prometheus.Labels { return exemplar }))
 
-	r, _ := http.NewRequest("GET", "www.example.com", nil)
+	r, _ := http.NewRequest(http.MethodGet, "www.example.com", nil)
 	w := httptest.NewRecorder()
 	chain.ServeHTTP(w, r)
 

prometheus/registry_test.go

@@ -714,7 +714,7 @@ collected metric "broken_metric" { label:<name:"foo" value:"bar" > label:<name:"
 		}
 		writer := httptest.NewRecorder()
 		handler := promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{})
-		request, _ := http.NewRequest("GET", "/", nil)
+		request, _ := http.NewRequest(http.MethodGet, "/", nil)
 		for key, value := range scenario.headers {
 			request.Header.Add(key, value)
 		}

prometheus/summary.go

@@ -471,13 +471,9 @@ func (s *noObjectivesSummary) Observe(v float64) {
 	n := atomic.AddUint64(&s.countAndHotIdx, 1)
 	hotCounts := s.counts[n>>63]
 
-	for {
-		oldBits := atomic.LoadUint64(&hotCounts.sumBits)
-		newBits := math.Float64bits(math.Float64frombits(oldBits) + v)
-		if atomic.CompareAndSwapUint64(&hotCounts.sumBits, oldBits, newBits) {
-			break
-		}
-	}
+	atomicUpdateFloat(&hotCounts.sumBits, func(oldVal float64) float64 {
+		return oldVal + v
+	})
 	// Increment count last as we take it as a signal that the observation
 	// is complete.
 	atomic.AddUint64(&hotCounts.count, 1)
@@ -519,14 +515,13 @@ func (s *noObjectivesSummary) Write(out *dto.Metric) error {
 	// Finally add all the cold counts to the new hot counts and reset the cold counts.
 	atomic.AddUint64(&hotCounts.count, count)
 	atomic.StoreUint64(&coldCounts.count, 0)
-	for {
-		oldBits := atomic.LoadUint64(&hotCounts.sumBits)
-		newBits := math.Float64bits(math.Float64frombits(oldBits) + sum.GetSampleSum())
-		if atomic.CompareAndSwapUint64(&hotCounts.sumBits, oldBits, newBits) {
-			atomic.StoreUint64(&coldCounts.sumBits, 0)
-			break
-		}
-	}
+
+	// Use atomicUpdateFloat to update hotCounts.sumBits atomically.
+	atomicUpdateFloat(&hotCounts.sumBits, func(oldVal float64) float64 {
+		return oldVal + sum.GetSampleSum()
+	})
+	atomic.StoreUint64(&coldCounts.sumBits, 0)
+
 	return nil
 }