Merge branch 'main' into issue-4960

conf/example.yaml

@@ -155,8 +155,8 @@ querier:
   cache_ttl: 168h
   # The interval for recent data that should not be cached.
   flux_interval: 5m
-  # The maximum number of concurrent queries for missing ranges.
-  max_concurrent_queries: 4
+  # The maximum number of queries a single query range request runs at once.
+  max_concurrent_queries: 8
   # When filtering logs by trace_id, clamp the query window to the trace time
   # range with padding to include slightly delayed log exports. Logs only; set
   # to 0 to disable.

docs/contributing/go/authz.md

@@ -1,149 +0,0 @@
-# Authz
-
-SigNoz uses [OpenFGA](https://openfga.dev/), a relationship-based access control (ReBAC) system, to authorize every request. Permissions are never attached to users directly — they are attached to **roles** (as relationship tuples in OpenFGA), and users or service accounts are made **assignees** of those roles. The central interface is `AuthZ` in [pkg/authz/authz.go](/pkg/authz/authz.go), backed by an embedded OpenFGA server in [pkg/authz/openfgaserver](/pkg/authz/openfgaserver/server.go).
-
-As a feature author, you will rarely touch OpenFGA directly. You interact with two layers:
-
-1. **The registries** in [pkg/types/coretypes](/pkg/types/coretypes) — where every resource, verb, and managed-role permission is declared in code.
-2. **The route wiring** in [pkg/apiserver/signozapiserver](/pkg/apiserver/signozapiserver) — where each route declares what resource it touches and which verb it needs, and the middleware turns that into a check.
-
-## What are the building blocks?
-
-### Type
-
-A `Type` is the coarse FGA type of a resource. The set is finite and defined in [pkg/types/coretypes/registry_type.go](/pkg/types/coretypes/registry_type.go): `user`, `serviceaccount`, `anonymous`, `role`, `organization`, `metaresource`, and `telemetryresource`. Each type carries a selector regex (what a valid ID looks like) and the verbs allowed on it.
-
-Almost every feature resource is a `metaresource` (dashboards, rules, pipelines, ...) or a `telemetryresource` (logs, traces, metrics). You should almost never need a new type.
-
-### Verb
-
-A `Verb` is the action being authorized. All verbs are defined in [pkg/types/coretypes/registry_verb.go](/pkg/types/coretypes/registry_verb.go): `create`, `read`, `update`, `delete`, `list`, `assignee`, `attach`, and `detach`.
-
-- `assignee` is special: it is the membership relation between a subject and a role ("user X is an assignee of role Y"), not an action a route checks for.
-- `attach`/`detach` authorize linking two resources together (e.g. assigning a role to a service account).
-
-### Kind
-
-A `Kind` is the fine-grained name of your resource — `dashboard`, `rule`, `quick-filter`. Kinds are registered in [pkg/types/coretypes/registry_kind.go](/pkg/types/coretypes/registry_kind.go). A kind rides on top of an existing type, so adding one does **not** require any OpenFGA schema change.
-
-### Resource
-
-A `Resource` ties a type and a kind together and knows how to render itself as an FGA object string. The interface lives in [pkg/types/coretypes/resource.go](/pkg/types/coretypes/resource.go):
-
-```go
-type Resource interface {
-    Type() Type
-    Kind() Kind
-    Prefix(orgId valuer.UUID) string      // metaresource:organization/<orgID>/dashboard
-    Object(orgId valuer.UUID, selector string) string
-    Scope(verb Verb) string               // dashboard:read
-    AllowedVerbs() []Verb
-}
-```
-
-All resources are registered in [pkg/types/coretypes/registry_resource.go](/pkg/types/coretypes/registry_resource.go) using constructors like `NewResourceMetaResource(KindDashboard)` or `NewResourceTelemetryResource(KindLogs)`.
-
-### Selector
-
-A `Selector` identifies *which* instance(s) of a resource a check is about — a UUID, a role name, or the wildcard `*`. A `SelectorFunc` maps the extracted resource ID to selectors at request time. Two prebuilt ones cover most routes ([pkg/types/coretypes/selector.go](/pkg/types/coretypes/selector.go)):
-
-- `WildcardSelector` — the check is against all instances of the resource (`create`, `list`).
-- `IDSelector` — the check is against the specific instance *or* the wildcard (`read`, `update`, `delete` of one object). A subject authorized on `*` is authorized on every instance.
-
-When the ID in the request is not what FGA needs (e.g. routes receive a role UUID but FGA objects use role names), write a custom `SelectorFunc` — see `roleSelector` in [pkg/apiserver/signozapiserver/serviceaccount.go](/pkg/apiserver/signozapiserver/serviceaccount.go).
-
-### Roles, transactions, and tuples
-
-SigNoz ships four managed roles, declared in [pkg/types/coretypes/registry_managed_role.go](/pkg/types/coretypes/registry_managed_role.go): `signoz-admin`, `signoz-editor`, `signoz-viewer`, and `signoz-anonymous`. Their permissions are declared in code as `Transaction`s (a verb on an object) in `ManagedRoleToTransactions` — this map is the single source of truth for what each managed role can do.
-
-At organization bootstrap, `CreateManagedRoles` and `CreateManagedUserRoleTransactions` (see [pkg/authz/authz.go](/pkg/authz/authz.go)) persist the role rows and write one OpenFGA tuple per transaction, linking `role:organization/<orgID>/role/<name>#assignee` to each permitted object. Users and service accounts are then granted roles via `Grant`/`Revoke`, which writes `assignee` tuples. Custom roles (enterprise) are managed through the roles API in [pkg/authz/signozauthzapi/handler.go](/pkg/authz/signozauthzapi/handler.go).
-
-### Schema
-
-The OpenFGA authorization model is a hand-written DSL, embedded at build time: [pkg/authz/openfgaschema/base.fga](/pkg/authz/openfgaschema/base.fga) for community and [ee/authz/openfgaschema/base.fga](/ee/authz/openfgaschema/base.fga) for enterprise. The community model only supports role assignment; the enterprise model defines per-verb relations on every type, enabling genuine per-resource checks. This split is why `CheckWithTupleCreation` behaves differently per edition (see [How does a check work at runtime?](#how-does-a-check-work-at-runtime)). You only touch these files when introducing a brand-new **type** — never for a new kind.
-
-## How do I add authz to my feature?
-
-### 1. Register the kind
-
-Add your kind in [pkg/types/coretypes/registry_kind.go](/pkg/types/coretypes/registry_kind.go) and append it to `Kinds`:
-
-```go
-KindThing = MustNewKind("thing")
-```
-
-### 2. Register the resource
-
-Add the resource in [pkg/types/coretypes/registry_resource.go](/pkg/types/coretypes/registry_resource.go) and append it to `Resources`:
-
-```go
-ResourceMetaResourceThing = NewResourceMetaResource(KindThing)
-```
-
-Pass an explicit verb list to `NewResourceMetaResource` only if your resource supports fewer verbs than the type default.
-
-### 3. Grant permissions to managed roles
-
-Decide what each managed role can do with your resource and add the transactions in [pkg/types/coretypes/registry_managed_role.go](/pkg/types/coretypes/registry_managed_role.go):
-
-```go
-// thing — editors manage, viewers read
-{Verb: VerbCreate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindThing}, WildCardSelectorString)},
-```
-
-The convention so far: admin gets everything, editor gets CRUD on day-to-day observability resources, viewer gets `read`/`list`, anonymous gets nothing (except public dashboards).
-
-### 4. Wire the route
-
-Register the route in [pkg/apiserver/signozapiserver](/pkg/apiserver/signozapiserver), wrapping your handler with `CheckResources` and declaring what the route touches via a `ResourceDef` ([pkg/http/handler/resourcedef.go](/pkg/http/handler/resourcedef.go)). A complete example from [pkg/apiserver/signozapiserver/serviceaccount.go](/pkg/apiserver/signozapiserver/serviceaccount.go):
-
-```go
-router.Handle("/api/v1/service_accounts", handler.New(
-    provider.authzMiddleware.CheckResources(provider.serviceAccountHandler.Create, authtypes.SigNozAdminRoleName),
-    handler.OpenAPIDef{
-        ID: "CreateServiceAccount",
-        // ...
-        SecuritySchemes: newScopedSecuritySchemes([]string{coretypes.ResourceServiceAccount.Scope(coretypes.VerbCreate)}),
-    },
-    handler.WithResourceDefs(handler.BasicResourceDef{
-        Resource: coretypes.ResourceServiceAccount,
-        Verb:     coretypes.VerbCreate,
-        Category: coretypes.ActionCategoryAccessControl,
-        ID:       coretypes.ResponseJSONPath("data.id"),
-        Selector: coretypes.WildcardSelector,
-    }),
-)).Methods(http.MethodPost)
-```
-
-The pieces:
-
-- **`CheckResources(handlerFn, roles...)`** — the resource-aware authorization wrapper from [pkg/http/middleware/authz.go](/pkg/http/middleware/authz.go). The role list is the community-edition fallback: which managed roles may call this route when per-resource checks are unavailable.
-- **`ResourceDef`** — declares the resource, verb, audit category, how to extract the instance ID, and how to turn that ID into selectors. ID extractors live in [pkg/types/coretypes/extractor.go](/pkg/types/coretypes/extractor.go): `PathParam("id")`, `BodyJSONPath("data.id")`, `BodyJSONArray("ids")`, and `ResponseJSONPath("data.id")` for IDs only known after the handler runs (e.g. `create`).
-- **`SecuritySchemes`** — advertises the required scope (`resource.Scope(verb)`, e.g. `serviceaccount:create`) in the OpenAPI spec.
-
-For routes that link two resources, use `AttachDetachSiblingResourceDef` (both sides are authz-checked, e.g. attaching a role to a service account requires `attach` on **both** the service account and the role) or `AttachDetachParentChildResourceDef` (only the parent is checked; the child is recorded for audit, e.g. creating an API key under a service account).
-
-Prefer `CheckResources` with a `ResourceDef` for anything resource-shaped. The older coarse gates `ViewAccess`/`EditAccess`/`AdminAccess` only check "does the caller hold one of these roles" and give up per-resource granularity; `OpenAccess` performs no authorization (authentication still applies); `CheckWithoutClaims` serves anonymous routes such as public dashboards.
-
-### 5. Backfill existing organizations (only if needed)
-
-Managed-role tuples are written from the registry at organization creation, so **new resources need no migration for new organizations**. If existing organizations must get the new permissions, add a migration in [pkg/sqlmigration](/pkg/sqlmigration) that inserts the tuples — see [083_add_role_crud_tuples.go](/pkg/sqlmigration/083_add_role_crud_tuples.go) for the pattern.
-
-## How does a check work at runtime?
-
-1. The resource middleware ([pkg/http/middleware/resource.go](/pkg/http/middleware/resource.go)) runs on every request. It reads the matched handler's `ResourceDef`s, extracts the resource IDs from path/body, and stores the resolved resources in the request context.
-2. `CheckResources` reads them back, runs each `SelectorFunc`, and calls `AuthZ.CheckWithTupleCreation(ctx, claims, orgID, relation, resource, selectors, roleSelectors)`.
-3. What happens next depends on the edition:
-   - **Community** ([pkg/authz/openfgaserver/server.go](/pkg/authz/openfgaserver/server.go)) ignores the resource and selectors and only checks whether the subject is an `assignee` of one of the allowed roles — a plain role gate.
-   - **Enterprise** ([ee/authz/openfgaserver/server.go](/ee/authz/openfgaserver/server.go)) builds tuples via `authtypes.NewTuples` — subject `user:organization/<orgID>/user/<userID>`, relation `create`, object `serviceaccount:organization/<orgID>/serviceaccount/*` — and batch-checks them against OpenFGA: genuine per-resource authorization, including custom roles.
-
-Because both paths go through the same middleware and the same `ResourceDef` declarations, a route wired once works correctly in both editions.
-
-## What should I remember?
-
-- Declare authz in the registries ([pkg/types/coretypes](/pkg/types/coretypes)), not in migrations — tuples for new organizations are derived from code at bootstrap.
-- A new kind never needs an OpenFGA schema change; only a new type does, and then **both** [pkg/authz/openfgaschema/base.fga](/pkg/authz/openfgaschema/base.fga) and [ee/authz/openfgaschema/base.fga](/ee/authz/openfgaschema/base.fga) must be updated together.
-- Prefer `CheckResources` + `ResourceDef` over the coarse `ViewAccess`/`EditAccess`/`AdminAccess` gates for new routes.
-- Use `WildcardSelector` for `create`/`list`, `IDSelector` for instance operations, and a custom `SelectorFunc` when the request ID is not the FGA selector.
-- Attach/detach routes between peer resources must check **both** sides (`AttachDetachSiblingResourceDef`); parent-child creation checks only the parent (`AttachDetachParentChildResourceDef`).
-- Changing `ManagedRoleToTransactions` only affects organizations created afterwards — add a [pkg/sqlmigration](/pkg/sqlmigration) migration to backfill existing ones.

docs/contributing/go/readme.md

@@ -11,7 +11,6 @@ We adhere to three primary style guides as our foundation:
 We **recommend** (almost enforce) reviewing these guides before contributing to the codebase. They provide valuable insights into writing idiomatic Go code and will help you understand our approach to backend development. In addition, we have a few additional rules that make certain areas stricter than the above which can be found in area-specific files in this package:
 
 - [Abstractions](abstractions.md) - When to introduce new types and intermediate representations
-- [Authz](authz.md) - Authorization, roles, and access control
 - [Errors](errors.md) - Structured error handling
 - [Endpoint](endpoint.md) - HTTP endpoint patterns
 - [Flagger](flagger.md) - Feature flag patterns

pkg/querier/config.go

@@ -7,6 +7,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 )
 
+const DefaultMaxConcurrentQueries = 8
+
 type SkipResourceFingerprint struct {
 	Enabled bool `yaml:"enabled" mapstructure:"enabled"`
 	// If count of fingerprint is above threshold, skip the fingerprint subquery and filter on main table instead.
@@ -37,7 +39,7 @@ func newConfig() factory.Config {
 		// Default values
 		CacheTTL:             168 * time.Hour,
 		FluxInterval:         5 * time.Minute,
-		MaxConcurrentQueries: 4,
+		MaxConcurrentQueries: DefaultMaxConcurrentQueries,
 		SkipResourceFingerprint: SkipResourceFingerprint{
 			Enabled:   false,
 			Threshold: 100000,

pkg/querier/querier.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
@@ -35,20 +36,21 @@ var (
 )
 
 type querier struct {
-	logger                    *slog.Logger
-	fl                        flagger.Flagger
-	telemetryStore            telemetrystore.TelemetryStore
-	metadataStore             telemetrytypes.MetadataStore
-	promEngine                prometheus.Prometheus
-	traceStmtBuilder          qbtypes.StatementBuilder[qbtypes.TraceAggregation]
-	logStmtBuilder            qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	auditStmtBuilder          qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	metricStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
-	meterStmtBuilder          qbtypes.StatementBuilder[qbtypes.MetricAggregation]
-	traceOperatorStmtBuilder  qbtypes.TraceOperatorStatementBuilder
-	bucketCache               BucketCache
-	liveDataRefresh           time.Duration
-	builderConfig             builderConfig
+	logger                   *slog.Logger
+	fl                       flagger.Flagger
+	telemetryStore           telemetrystore.TelemetryStore
+	metadataStore            telemetrytypes.MetadataStore
+	promEngine               prometheus.Prometheus
+	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
+	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
+	meterStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
+	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
+	bucketCache              BucketCache
+	liveDataRefresh          time.Duration
+	builderConfig            builderConfig
+	maxConcurrentQueries     int
 }
 
 var _ Querier = (*querier)(nil)
@@ -67,25 +69,30 @@ func New(
 	bucketCache BucketCache,
 	flagger flagger.Flagger,
 	logTraceIDWindowPadding time.Duration,
+	maxConcurrentQueries int,
 ) *querier {
 	querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
+	if maxConcurrentQueries <= 0 {
+		maxConcurrentQueries = DefaultMaxConcurrentQueries
+	}
 	return &querier{
-		logger:                    querierSettings.Logger(),
-		fl:                        flagger,
-		telemetryStore:            telemetryStore,
-		metadataStore:             metadataStore,
-		promEngine:                promEngine,
-		traceStmtBuilder:          traceStmtBuilder,
-		logStmtBuilder:            logStmtBuilder,
-		auditStmtBuilder:          auditStmtBuilder,
-		metricStmtBuilder:         metricStmtBuilder,
-		meterStmtBuilder:          meterStmtBuilder,
-		traceOperatorStmtBuilder:  traceOperatorStmtBuilder,
-		bucketCache:               bucketCache,
-		liveDataRefresh:           5 * time.Second,
+		logger:                   querierSettings.Logger(),
+		fl:                       flagger,
+		telemetryStore:           telemetryStore,
+		metadataStore:            metadataStore,
+		promEngine:               promEngine,
+		traceStmtBuilder:         traceStmtBuilder,
+		logStmtBuilder:           logStmtBuilder,
+		auditStmtBuilder:         auditStmtBuilder,
+		metricStmtBuilder:        metricStmtBuilder,
+		meterStmtBuilder:         meterStmtBuilder,
+		traceOperatorStmtBuilder: traceOperatorStmtBuilder,
+		bucketCache:              bucketCache,
+		liveDataRefresh:          5 * time.Second,
 		builderConfig: builderConfig{
 			logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
 		},
+		maxConcurrentQueries: maxConcurrentQueries,
 	}
 }
 
@@ -607,30 +614,40 @@ func (q *querier) run(
 		return false
 	}
 
-	for name, query := range qs {
-		// Skip cache if NoCache is set, or if cache is not available
-		if req.NoCache || q.bucketCache == nil || query.Fingerprint() == "" {
-			if req.NoCache {
-				q.logger.DebugContext(ctx, "NoCache flag set, bypassing cache", slog.String("query", name))
-			} else {
-				q.logger.InfoContext(ctx, "no bucket cache or fingerprint, executing query", slog.String("fingerprint", query.Fingerprint()))
+	names := maps.Keys(qs)
+	slices.Sort(names)
+	queryResults := make([]*qbtypes.Result, len(names))
+
+	// sem limits how many queries run at once for this request. The same
+	// limit covers the missing-range queries in executeWithCache. sem is held
+	// only while a query is running, never while waiting for other
+	// goroutines, so the two levels cannot deadlock.
+	sem := make(chan struct{}, q.maxConcurrentQueries)
+
+	eg, egCtx := errgroup.WithContext(ctx)
+	for i, name := range names {
+		query := qs[name]
+		eg.Go(func() error {
+			// Skip cache if NoCache is set, or if cache is not available
+			if req.NoCache || q.bucketCache == nil || query.Fingerprint() == "" {
+				if req.NoCache {
+					q.logger.DebugContext(egCtx, "NoCache flag set, bypassing cache", slog.String("query", name))
+				} else {
+					q.logger.InfoContext(egCtx, "no bucket cache or fingerprint, executing query", slog.String("fingerprint", query.Fingerprint()))
+				}
+				sem <- struct{}{}
+				result, err := query.Execute(egCtx)
+				<-sem
+				if err != nil {
+					return err
+				}
+				queryResults[i] = result
+				return nil
 			}
-			result, err := query.Execute(ctx)
-			qbEvent.HasData = qbEvent.HasData || hasData(result)
+
+			result, err := q.executeWithCache(egCtx, orgID, query, steps[name], sem)
 			if err != nil {
-				return nil, err
-			}
-			results[name] = result.Value
-			warnings = append(warnings, result.Warnings...)
-			warningsDocURL = result.WarningsDocURL
-			stats.RowsScanned += result.Stats.RowsScanned
-			stats.BytesScanned += result.Stats.BytesScanned
-			stats.DurationMS += result.Stats.DurationMS
-		} else {
-			result, err := q.executeWithCache(ctx, orgID, query, steps[name], req.NoCache)
-			qbEvent.HasData = qbEvent.HasData || hasData(result)
-			if err != nil {
-				return nil, err
+				return err
 			}
 			switch v := result.Value.(type) {
 			case *qbtypes.TimeSeriesData:
@@ -640,14 +657,23 @@ func (q *querier) run(
 			case *qbtypes.RawData:
 				v.QueryName = name
 			}
+			queryResults[i] = result
+			return nil
+		})
+	}
+	if err := eg.Wait(); err != nil {
+		return nil, err
+	}
 
-			results[name] = result.Value
-			warnings = append(warnings, result.Warnings...)
-			warningsDocURL = result.WarningsDocURL
-			stats.RowsScanned += result.Stats.RowsScanned
-			stats.BytesScanned += result.Stats.BytesScanned
-			stats.DurationMS += result.Stats.DurationMS
-		}
+	for i, name := range names {
+		result := queryResults[i]
+		qbEvent.HasData = qbEvent.HasData || hasData(result)
+		results[name] = result.Value
+		warnings = append(warnings, result.Warnings...)
+		warningsDocURL = result.WarningsDocURL
+		stats.RowsScanned += result.Stats.RowsScanned
+		stats.BytesScanned += result.Stats.BytesScanned
+		stats.DurationMS += result.Stats.DurationMS
 	}
 
 	gomaps.Copy(results, preseededResults)
@@ -707,8 +733,9 @@ func (q *querier) run(
 	return resp, nil
 }
 
-// executeWithCache executes a query using the bucket cache.
-func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query qbtypes.Query, step qbtypes.Step, _ bool) (*qbtypes.Result, error) {
+// executeWithCache executes a query using the bucket cache. sem limits how
+// many queries run at once for the whole request.
+func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query qbtypes.Query, step qbtypes.Step, sem chan struct{}) (*qbtypes.Result, error) {
 	// Get cached data and missing ranges
 	cachedResult, missingRanges := q.bucketCache.GetMissRanges(ctx, orgID, query, step)
 
@@ -721,7 +748,9 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 	if cachedResult == nil && len(missingRanges) == 1 {
 		startMs, endMs := query.Window()
 		if missingRanges[0].From == startMs && missingRanges[0].To == endMs {
+			sem <- struct{}{}
 			result, err := query.Execute(ctx)
+			<-sem
 			if err != nil {
 				return nil, err
 			}
@@ -740,7 +769,6 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 		slog.Int("missing_ranges_count", len(missingRanges)),
 		slog.Any("ranges", missingRanges))
 
-	sem := make(chan struct{}, 4)
 	var wg sync.WaitGroup
 
 	for i, timeRange := range missingRanges {
@@ -777,7 +805,9 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 		if err != nil {
 			// If any query failed, fall back to full execution
 			q.logger.ErrorContext(ctx, "parallel query execution failed", errors.Attr(err))
+			sem <- struct{}{}
 			result, err := query.Execute(ctx)
+			<-sem
 			if err != nil {
 				return nil, err
 			}

pkg/querier/querier_test.go

@@ -2,11 +2,13 @@ package querier
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	cmock "github.com/SigNoz/clickhouse-go-mock"
 
+	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -54,7 +56,8 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		nil,                // traceOperatorStmtBuilder
 		nil,                // bucketCache
 		flaggertest.New(t), // flagger
-		0,
+		0,                  // logTraceIDWindowPadding
+		0,                  // maxConcurrentQueries
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -125,7 +128,8 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		nil,                      // traceOperatorStmtBuilder
 		nil,                      // bucketCache
 		flaggertest.New(t),       // flagger
-		0,
+		0,                        // logTraceIDWindowPadding
+		0,                        // maxConcurrentQueries
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -155,3 +159,149 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, resp)
 }
+
+type fakeQuery struct {
+	execute func(ctx context.Context) (*qbtypes.Result, error)
+}
+
+func (f *fakeQuery) Fingerprint() string                                  { return "" }
+func (f *fakeQuery) Window() (uint64, uint64)                             { return 0, 0 }
+func (f *fakeQuery) Execute(ctx context.Context) (*qbtypes.Result, error) { return f.execute(ctx) }
+
+func chQueryEnvelopes(names []string) []qbtypes.QueryEnvelope {
+	envelopes := make([]qbtypes.QueryEnvelope, 0, len(names))
+	for _, name := range names {
+		envelopes = append(envelopes, qbtypes.QueryEnvelope{
+			Type: qbtypes.QueryTypeClickHouseSQL,
+			Spec: qbtypes.ClickHouseQuery{Name: name},
+		})
+	}
+	return envelopes
+}
+
+func TestRunExecutesQueriesConcurrently(t *testing.T) {
+	names := []string{"A", "B", "C", "D", "E"}
+	numQueries := len(names)
+
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: numQueries,
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var started atomic.Int32
+	allStarted := make(chan struct{})
+
+	qs := make(map[string]qbtypes.Query, numQueries)
+	for _, name := range names {
+		qs[name] = &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			if int(started.Add(1)) == numQueries {
+				close(allStarted)
+			}
+			select {
+			case <-allStarted:
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+			return &qbtypes.Result{
+				Type:  qbtypes.RequestTypeScalar,
+				Value: &qbtypes.ScalarData{QueryName: name},
+				Stats: qbtypes.ExecStats{RowsScanned: 1, BytesScanned: 2, DurationMS: 3},
+			}, nil
+		}}
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes(names)},
+	}
+	resp, err := q.run(ctx, valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Len(t, resp.Data.Results, numQueries)
+	assert.Equal(t, uint64(numQueries), resp.Meta.RowsScanned)
+	assert.Equal(t, uint64(2*numQueries), resp.Meta.BytesScanned)
+	assert.Equal(t, uint64(3*numQueries), resp.Meta.DurationMS)
+}
+
+func TestRunRespectsMaxConcurrentQueries(t *testing.T) {
+	const limit = 2
+	names := []string{"A", "B", "C", "D", "E", "F", "G", "H"}
+
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: limit,
+	}
+
+	var running, maxRunning atomic.Int32
+	qs := make(map[string]qbtypes.Query, len(names))
+	for _, name := range names {
+		qs[name] = &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			cur := running.Add(1)
+			defer running.Add(-1)
+			for {
+				m := maxRunning.Load()
+				if cur <= m || maxRunning.CompareAndSwap(m, cur) {
+					break
+				}
+			}
+			time.Sleep(20 * time.Millisecond)
+			return &qbtypes.Result{
+				Type:  qbtypes.RequestTypeScalar,
+				Value: &qbtypes.ScalarData{QueryName: name},
+			}, nil
+		}}
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes(names)},
+	}
+	resp, err := q.run(context.Background(), valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Len(t, resp.Data.Results, len(names))
+	assert.LessOrEqual(t, maxRunning.Load(), int32(limit), "running queries must not exceed maxConcurrentQueries")
+}
+
+func TestRunQueryErrorCancelsSiblings(t *testing.T) {
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: 4,
+	}
+
+	bStarted := make(chan struct{})
+	var bCanceled atomic.Bool
+
+	qs := map[string]qbtypes.Query{
+		// fails once B is running.
+		"A": &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			select {
+			case <-bStarted:
+			case <-ctx.Done():
+			}
+			return nil, errors.NewInternalf(errors.CodeInternal, "query A failed")
+		}},
+		// blocks until its context is canceled by A's failure.
+		"B": &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			close(bStarted)
+			select {
+			case <-ctx.Done():
+				bCanceled.Store(true)
+				return nil, ctx.Err()
+			case <-time.After(10 * time.Second):
+				return nil, errors.NewInternalf(errors.CodeInternal, "query B was never canceled")
+			}
+		}},
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes([]string{"A", "B"})},
+	}
+	_, err := q.run(context.Background(), valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.ErrorContains(t, err, "query A failed")
+	assert.True(t, bCanceled.Load(), "query B should be canceled once query A fails")
+}

pkg/querier/signozquerier/provider.go

@@ -193,5 +193,6 @@ func newProvider(
 		bucketCache,
 		flagger,
 		cfg.LogTraceIDWindowPadding,
+		cfg.MaxConcurrentQueries,
 	), nil
 }

pkg/query-service/rules/setups_test.go

@@ -56,6 +56,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		nil, // bucketCache
 		flagger,
 		0,
+		0, // maxConcurrentQueries (0 means default)
 	), metadataStore
 }
 
@@ -110,6 +111,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		nil,            // bucketCache
 		fl,
 		5*time.Minute, // logTraceIDWindowPadding
+		0,             // maxConcurrentQueries (0 means default)
 	)
 }
 
@@ -158,5 +160,6 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		nil,              // bucketCache
 		fl,
 		0,
+		0, // maxConcurrentQueries (0 means default)
 	)
 }