Move Query locking back into private query function (#4694)

* Add parallel benchmarks to silence_bench_test

Add benchmarks about parallel queries, parallel queries in concurrency
with adds, and about running mutes.

Signed-off-by: Guido Trotter <[email protected]>

* Move Query locking back into private query function

This was changed in b67bde8 because Set
called query, while already holding a lock, but that call was removed in
dbe6312 and replaced with a simple
len(s.st) call, so locking can move back to the inner function.

Signed-off-by: Guido Trotter <[email protected]>

---------

Signed-off-by: Guido Trotter <[email protected]>
Co-authored-by: Guido Trotter <[email protected]>

Author: ultrotter

silence/silence.go

@@ -794,9 +794,6 @@ func (s *Silences) QueryOne(params ...QueryParam) (*pb.Silence, error) {
 // Query for silences based on the given query parameters. It returns the
 // resulting silences and the state version the result is based on.
 func (s *Silences) Query(params ...QueryParam) ([]*pb.Silence, int, error) {
-	s.mtx.Lock()
-	defer s.mtx.Unlock()
-
 	s.metrics.queriesTotal.Inc()
 	defer prometheus.NewTimer(s.metrics.queryDuration).ObserveDuration()
 
@@ -836,6 +833,9 @@ func (s *Silences) query(q *query, now time.Time) ([]*pb.Silence, int, error) {
 	// the use of post-filter functions is the trivial solution for now.
 	var res []*pb.Silence
 
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
 	if q.ids != nil {
 		for _, id := range q.ids {
 			if s, ok := s.st[id]; ok {

silence/silence_bench_test.go

@@ -15,6 +15,7 @@ package silence
 
 import (
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -153,3 +154,257 @@ func benchmarkQuery(b *testing.B, numSilences int) {
 		require.Len(b, sils, numSilences/10)
 	}
 }
+
+// BenchmarkQueryParallel benchmarks concurrent queries to demonstrate
+// the performance improvement from using read locks (RLock) instead of
+// write locks (Lock). With the pre-compiled matcher cache, multiple
+// queries can now execute in parallel.
+func BenchmarkQueryParallel(b *testing.B) {
+	b.Run("100 silences, 1 goroutine", func(b *testing.B) {
+		benchmarkQueryParallel(b, 100, 1)
+	})
+	b.Run("100 silences, 2 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 100, 2)
+	})
+	b.Run("100 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 100, 4)
+	})
+	b.Run("100 silences, 8 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 100, 8)
+	})
+	b.Run("1000 silences, 1 goroutine", func(b *testing.B) {
+		benchmarkQueryParallel(b, 1000, 1)
+	})
+	b.Run("1000 silences, 2 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 1000, 2)
+	})
+	b.Run("1000 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 1000, 4)
+	})
+	b.Run("1000 silences, 8 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 1000, 8)
+	})
+	b.Run("10000 silences, 1 goroutine", func(b *testing.B) {
+		benchmarkQueryParallel(b, 10000, 1)
+	})
+	b.Run("10000 silences, 2 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 10000, 2)
+	})
+	b.Run("10000 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 10000, 4)
+	})
+	b.Run("10000 silences, 8 goroutines", func(b *testing.B) {
+		benchmarkQueryParallel(b, 10000, 8)
+	})
+}
+
+func benchmarkQueryParallel(b *testing.B, numSilences, numGoroutines int) {
+	s, err := New(Options{})
+	require.NoError(b, err)
+
+	clock := quartz.NewMock(b)
+	s.clock = clock
+	now := clock.Now()
+
+	lset := model.LabelSet{"aaaa": "AAAA", "bbbb": "BBBB", "cccc": "CCCC"}
+
+	// Create silences with pre-compiled matchers
+	for i := 0; i < numSilences; i++ {
+		id := strconv.Itoa(i)
+		patA := "A{4}|" + id
+		patB := id
+		if i%10 == 0 {
+			patB = "B(B|C)B.|" + id
+		}
+
+		sil := &silencepb.Silence{
+			Matchers: []*silencepb.Matcher{
+				{Type: silencepb.Matcher_REGEXP, Name: "aaaa", Pattern: patA},
+				{Type: silencepb.Matcher_REGEXP, Name: "bbbb", Pattern: patB},
+			},
+			StartsAt:  now.Add(-time.Minute),
+			EndsAt:    now.Add(time.Hour),
+			UpdatedAt: now.Add(-time.Hour),
+		}
+		require.NoError(b, s.Set(sil))
+	}
+
+	// Verify initial query works
+	sils, _, err := s.Query(
+		QState(types.SilenceStateActive),
+		QMatches(lset),
+	)
+	require.NoError(b, err)
+	require.Len(b, sils, numSilences/10)
+
+	b.ResetTimer()
+
+	// Run queries in parallel across multiple goroutines
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			sils, _, err := s.Query(
+				QState(types.SilenceStateActive),
+				QMatches(lset),
+			)
+			if err != nil {
+				b.Error(err)
+			}
+			if len(sils) != numSilences/10 {
+				b.Errorf("expected %d silences, got %d", numSilences/10, len(sils))
+			}
+		}
+	})
+}
+
+// BenchmarkQueryWithConcurrentAdds benchmarks the behavior when queries
+// are running concurrently with silence additions. This demonstrates how
+// the system handles read-heavy workloads with occasional writes.
+func BenchmarkQueryWithConcurrentAdds(b *testing.B) {
+	b.Run("1000 initial silences, 10% add rate", func(b *testing.B) {
+		benchmarkQueryWithConcurrentAdds(b, 1000, 0.1)
+	})
+	b.Run("1000 initial silences, 1% add rate", func(b *testing.B) {
+		benchmarkQueryWithConcurrentAdds(b, 1000, 0.01)
+	})
+	b.Run("1000 initial silences, 0.1% add rate", func(b *testing.B) {
+		benchmarkQueryWithConcurrentAdds(b, 1000, 0.001)
+	})
+	b.Run("10000 initial silences, 1% add rate", func(b *testing.B) {
+		benchmarkQueryWithConcurrentAdds(b, 10000, 0.01)
+	})
+	b.Run("10000 initial silences, 0.1% add rate", func(b *testing.B) {
+		benchmarkQueryWithConcurrentAdds(b, 10000, 0.001)
+	})
+}
+
+func benchmarkQueryWithConcurrentAdds(b *testing.B, initialSilences int, addRatio float64) {
+	s, err := New(Options{})
+	require.NoError(b, err)
+
+	clock := quartz.NewMock(b)
+	s.clock = clock
+	now := clock.Now()
+
+	lset := model.LabelSet{"aaaa": "AAAA", "bbbb": "BBBB", "cccc": "CCCC"}
+
+	// Create initial silences
+	for i := 0; i < initialSilences; i++ {
+		id := strconv.Itoa(i)
+		patA := "A{4}|" + id
+		patB := id
+		if i%10 == 0 {
+			patB = "B(B|C)B.|" + id
+		}
+
+		sil := &silencepb.Silence{
+			Matchers: []*silencepb.Matcher{
+				{Type: silencepb.Matcher_REGEXP, Name: "aaaa", Pattern: patA},
+				{Type: silencepb.Matcher_REGEXP, Name: "bbbb", Pattern: patB},
+			},
+			StartsAt:  now.Add(-time.Minute),
+			EndsAt:    now.Add(time.Hour),
+			UpdatedAt: now.Add(-time.Hour),
+		}
+		require.NoError(b, s.Set(sil))
+	}
+
+	var addCounter int
+	var mu sync.Mutex
+
+	b.ResetTimer()
+
+	// Run parallel operations
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			// Determine if this iteration should add a silence
+			mu.Lock()
+			shouldAdd := float64(addCounter) < float64(b.N)*addRatio
+			if shouldAdd {
+				addCounter++
+			}
+			localCounter := addCounter + initialSilences
+			mu.Unlock()
+
+			if shouldAdd {
+				// Add a new silence
+				id := strconv.Itoa(localCounter)
+				patA := "A{4}|" + id
+				patB := "B(B|C)B.|" + id
+
+				sil := &silencepb.Silence{
+					Matchers: []*silencepb.Matcher{
+						{Type: silencepb.Matcher_REGEXP, Name: "aaaa", Pattern: patA},
+						{Type: silencepb.Matcher_REGEXP, Name: "bbbb", Pattern: patB},
+					},
+					StartsAt:  now.Add(-time.Minute),
+					EndsAt:    now.Add(time.Hour),
+					UpdatedAt: now.Add(-time.Hour),
+				}
+				if err := s.Set(sil); err != nil {
+					b.Error(err)
+				}
+			} else {
+				// Query silences (the common operation)
+				_, _, err := s.Query(
+					QState(types.SilenceStateActive),
+					QMatches(lset),
+				)
+				if err != nil {
+					b.Error(err)
+				}
+			}
+		}
+	})
+}
+
+// BenchmarkMutesParallel benchmarks concurrent Mutes calls to demonstrate
+// the performance improvement from parallel query execution.
+func BenchmarkMutesParallel(b *testing.B) {
+	b.Run("100 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkMutesParallel(b, 100, 4)
+	})
+	b.Run("1000 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkMutesParallel(b, 1000, 4)
+	})
+	b.Run("10000 silences, 4 goroutines", func(b *testing.B) {
+		benchmarkMutesParallel(b, 10000, 4)
+	})
+	b.Run("10000 silences, 8 goroutines", func(b *testing.B) {
+		benchmarkMutesParallel(b, 10000, 8)
+	})
+}
+
+func benchmarkMutesParallel(b *testing.B, numSilences, numGoroutines int) {
+	silences, err := New(Options{})
+	require.NoError(b, err)
+
+	clock := quartz.NewMock(b)
+	silences.clock = clock
+	now := clock.Now()
+
+	// Create silences that will match the alert
+	for i := 0; i < numSilences; i++ {
+		s := &silencepb.Silence{
+			Matchers: []*silencepb.Matcher{{
+				Type:    silencepb.Matcher_EQUAL,
+				Name:    "foo",
+				Pattern: "bar",
+			}},
+			StartsAt: now,
+			EndsAt:   now.Add(time.Minute),
+		}
+		require.NoError(b, silences.Set(s))
+	}
+
+	m := types.NewMarker(prometheus.NewRegistry())
+	silencer := NewSilencer(silences, m, promslog.NewNopLogger())
+
+	b.ResetTimer()
+
+	// Run Mutes in parallel
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			silencer.Mutes(model.LabelSet{"foo": "bar"})
+		}
+	})
+}