From 843dfbacbffbe05520709206ff7b39334624776f Mon Sep 17 00:00:00 2001 From: func25 Date: Tue, 21 Jul 2026 19:15:37 +0700 Subject: [PATCH] update --- app/vlselect/internalselect/internalselect.go | 17 +- app/vlselect/logsql/logsql.go | 81 ++++---- app/vlselect/main.go | 188 ++++++++---------- app/vlstorage/netselect/netselect.go | 15 +- docs/victorialogs/CHANGELOG.md | 2 + docs/victorialogs/README.md | 23 +++ docs/victorialogs/cluster.md | 7 + docs/victorialogs/logsql.md | 4 + docs/victorialogs/querying/README.md | 3 + docs/victorialogs/security-and-lb.md | 4 + lib/logstorage/block_result.go | 28 ++- lib/logstorage/block_search.go | 9 + lib/logstorage/filter_generic.go | 25 ++- lib/logstorage/pipe_field_names.go | 11 + lib/logstorage/pipe_stream_context.go | 2 +- lib/logstorage/storage.go | 4 +- lib/logstorage/storage_search.go | 120 +++++++++-- lib/logstorage/storage_search_test.go | 130 +++++++++++- lib/logstorage/storage_test.go | 2 +- lib/logstorage/tenant_id.go | 14 ++ 20 files changed, 523 insertions(+), 166 deletions(-) diff --git a/app/vlselect/internalselect/internalselect.go b/app/vlselect/internalselect/internalselect.go index ac759d3493..fa77d90885 100644 --- a/app/vlselect/internalselect/internalselect.go +++ b/app/vlselect/internalselect/internalselect.go @@ -444,8 +444,9 @@ func processTenantIDsRequest(ctx context.Context, w http.ResponseWriter, r *http } type commonParams struct { - TenantIDs []logstorage.TenantID - Query *logstorage.Query + TenantIDs []logstorage.TenantID + IsMultiTenant bool + Query *logstorage.Query // Whether to disable compression of the response sent to the vlselect. DisableCompression bool @@ -461,7 +462,7 @@ type commonParams struct { } func (cp *commonParams) NewQueryContext(ctx context.Context) *logstorage.QueryContext { - return logstorage.NewQueryContext(ctx, &cp.qs, cp.TenantIDs, cp.Query, cp.AllowPartialResponse, cp.HiddenFieldsFilters) + return logstorage.NewQueryContext(ctx, &cp.qs, cp.TenantIDs, cp.IsMultiTenant, cp.Query, cp.AllowPartialResponse, cp.HiddenFieldsFilters) } func (cp *commonParams) UpdatePerQueryStatsMetrics() { @@ -479,6 +480,11 @@ func getCommonParams(r *http.Request, expectedProtocolVersion string) (*commonPa return nil, fmt.Errorf("cannot unmarshal tenant_ids=%q: %w", tenantIDsStr, err) } + isMultiTenant, err := getBoolFromRequest(r, "multitenant") + if err != nil { + return nil, err + } + timestamp, err := getInt64FromRequest(r, "timestamp") if err != nil { return nil, err @@ -506,8 +512,9 @@ func getCommonParams(r *http.Request, expectedProtocolVersion string) (*commonPa } cp := &commonParams{ - TenantIDs: tenantIDs, - Query: q, + TenantIDs: tenantIDs, + IsMultiTenant: isMultiTenant, + Query: q, DisableCompression: disableCompression, diff --git a/app/vlselect/logsql/logsql.go b/app/vlselect/logsql/logsql.go index 6745abf971..7b7a538de8 100644 --- a/app/vlselect/logsql/logsql.go +++ b/app/vlselect/logsql/logsql.go @@ -126,8 +126,8 @@ func timestampToRFC3339Nano(nsec int64) string { // ProcessFacetsRequest handles /select/logsql/facets request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-facets -func ProcessFacetsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessFacetsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -211,8 +211,8 @@ type facetEntry struct { // ProcessHitsRequest handles /select/logsql/hits request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats -func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -423,8 +423,8 @@ func (hs *hitsSeries) Less(i, j int) bool { // ProcessFieldNamesRequest handles /select/logsql/field_names request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-field-names -func ProcessFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -457,8 +457,8 @@ func ProcessFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *htt // ProcessFieldValuesRequest handles /select/logsql/field_values request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-field-values -func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -505,8 +505,8 @@ func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *ht // ProcessStreamFieldNamesRequest processes /select/logsql/stream_field_names request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-stream-field-names -func ProcessStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -539,8 +539,8 @@ func ProcessStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter, // ProcessStreamFieldValuesRequest processes /select/logsql/stream_field_values request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-stream-field-values -func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -587,8 +587,8 @@ func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter, // ProcessStreamIDsRequest processes /select/logsql/stream_ids request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-stream_ids -func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -625,8 +625,8 @@ func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http // ProcessStreamsRequest processes /select/logsql/streams request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-streams -func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -663,11 +663,11 @@ func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.R // ProcessLiveTailRequest processes live tailing request to /select/logsq/tail // // See https://docs.victoriametrics.com/victorialogs/querying/#live-tailing -func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { +func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { liveTailRequests.Inc() defer liveTailRequests.Dec() - ca, err := parseCommonArgsExt(r, true) + ca, err := parseCommonArgsExt(r, isMultiTenant, true) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -872,8 +872,8 @@ func (tp *tailProcessor) getTailRows() ([][]logstorage.Field, error) { // ProcessStatsQueryRangeRequest handles /select/logsql/stats_query_range request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-log-range-stats -func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.SendPrometheusError(w, r, err) return @@ -1050,8 +1050,8 @@ type statsPoint struct { // ProcessStatsQueryRequest handles /select/logsql/stats_query request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-log-stats -func ProcessStatsQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessStatsQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.SendPrometheusError(w, r, err) return @@ -1172,8 +1172,8 @@ type histogramBucket struct { // ProcessQueryRequest handles /select/logsql/query request. // // See https://docs.victoriametrics.com/victorialogs/querying/#querying-logs -func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) { - ca, err := parseCommonArgs(r) +func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, isMultiTenant bool) { + ca, err := parseCommonArgs(r, isMultiTenant) if err != nil { httpserver.Errorf(w, r, "%s", err) return @@ -1430,6 +1430,9 @@ type commonArgs struct { // tenantIDs is the list of tenantIDs to query. tenantIDs []logstorage.TenantID + // isMultiTenant indicates whether the request is for a multitenant endpoint. + isMultiTenant bool + // Whether to allow partial response when some of vlstorage nodes are unavailable for querying. // This option makes sense only for cluster setup when vlselect queries vlstorage nodes. allowPartialResponse bool @@ -1448,15 +1451,15 @@ type commonArgs struct { } func (ca *commonArgs) newQueryContext(ctx context.Context) *logstorage.QueryContext { - return logstorage.NewQueryContext(ctx, &ca.qs, ca.tenantIDs, ca.q, ca.allowPartialResponse, ca.hiddenFieldsFilters) + return logstorage.NewQueryContext(ctx, &ca.qs, ca.tenantIDs, ca.isMultiTenant, ca.q, ca.allowPartialResponse, ca.hiddenFieldsFilters) } func (ca *commonArgs) updatePerQueryStatsMetrics() { vlstorage.UpdatePerQueryStatsMetrics(&ca.qs) } -func parseCommonArgs(r *http.Request) (*commonArgs, error) { - return parseCommonArgsExt(r, false) +func parseCommonArgs(r *http.Request, isMultiTenant bool) (*commonArgs, error) { + return parseCommonArgsExt(r, isMultiTenant, false) } // parseCommonArgsExt parses commonArgs from r. @@ -1464,13 +1467,20 @@ func parseCommonArgs(r *http.Request) (*commonArgs, error) { // If skipMaxQueryTimeRangeCheck is set, then the query time range isn't limited by -search.maxQueryTimeRange. // This is used by live tailing, since it processes only newly ingested logs // and doesn't scan historical data even if the query contains a wide time filter. -func parseCommonArgsExt(r *http.Request, skipMaxQueryTimeRangeCheck bool) (*commonArgs, error) { - // Extract tenantID - tenantID, err := logstorage.GetTenantIDFromRequest(r) - if err != nil { - return nil, fmt.Errorf("cannot obtain tenantID: %w", err) +func parseCommonArgsExt(r *http.Request, isMultiTenant, skipMaxQueryTimeRangeCheck bool) (*commonArgs, error) { + // Validate tenant IDs + var tenantIDs []logstorage.TenantID + if isMultiTenant { + if logstorage.HasTenantIDFromRequest(r) { + return nil, fmt.Errorf("accountID/projectID should not be set for multitenant request") + } + } else { + tenantID, err := logstorage.GetTenantIDFromRequest(r) + if err != nil { + return nil, fmt.Errorf("cannot obtain tenantID: %w", err) + } + tenantIDs = []logstorage.TenantID{tenantID} } - tenantIDs := []logstorage.TenantID{tenantID} // Parse optional start and end args start, startOK, err := getTimeNsec(r, "start") @@ -1597,8 +1607,9 @@ func parseCommonArgsExt(r *http.Request, skipMaxQueryTimeRangeCheck bool) (*comm } ca := &commonArgs{ - q: q, - tenantIDs: tenantIDs, + q: q, + tenantIDs: tenantIDs, + isMultiTenant: isMultiTenant, allowPartialResponse: allowPartialResponse, hiddenFieldsFilters: hiddenFieldsFilters, diff --git a/app/vlselect/main.go b/app/vlselect/main.go index 02f626c25e..86c9560735 100644 --- a/app/vlselect/main.go +++ b/app/vlselect/main.go @@ -34,8 +34,9 @@ var ( disableSelect = flag.Bool("select.disable", false, "Whether to disable both /select/* and /internal/select/* HTTP endpoints. Useful for dedicated vlinsert nodes. See also -internalselect.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security") disableInternalSelect = flag.Bool("internalselect.disable", false, "Whether to disable /internal/select/* HTTP endpoints. See also -select.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security") - enableDelete = flag.Bool("delete.enable", false, "Whether to enable /delete/* HTTP endpoints; see https://docs.victoriametrics.com/victorialogs/#how-to-delete-logs") - enableInternalDelete = flag.Bool("internaldelete.enable", false, "Whether to enable /internal/delete/* HTTP endpoints, which are used by vlselect for deleting logs "+ + enableMultitenantSelect = flag.Bool("multitenantselect.enable", false, "Whether to enable /select/multitenant/logsql/* HTTP endpoints. See https://docs.victoriametrics.com/victorialogs/#multitenant-querying") + enableDelete = flag.Bool("delete.enable", false, "Whether to enable /delete/* HTTP endpoints; see https://docs.victoriametrics.com/victorialogs/#how-to-delete-logs") + enableInternalDelete = flag.Bool("internaldelete.enable", false, "Whether to enable /internal/delete/* HTTP endpoints, which are used by vlselect for deleting logs "+ "via delete API at vlstorage nodes; see https://docs.victoriametrics.com/victorialogs/#how-to-delete-logs") logSlowQueryDuration = flag.Duration("search.logSlowQueryDuration", 5*time.Second, "Log queries with execution time exceeding this value. Zero disables slow query logging") @@ -192,13 +193,6 @@ func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool { return true } - if path == "/select/logsql/tail" { - logsqlTailRequests.Inc() - // Process live tailing request without timeout, since it is OK to run live tailing requests for very long time. - // Also do not apply concurrency limit to tail requests, since these limits are intended for non-tail requests. - logsql.ProcessLiveTailRequest(ctx, w, r) - return true - } if strings.HasPrefix(path, "/select/vmalert/") { vmalertRequests.Inc() if len(*vmalertProxyURL) == 0 { @@ -213,6 +207,23 @@ func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool { return true } + isMultiTenant := strings.HasPrefix(path, "/select/multitenant/logsql/") + if isMultiTenant && !*enableMultitenantSelect { + httpserver.Errorf(w, r, "requests to /select/multitenant/logsql/* endpoints are disabled; pass -multitenantselect.enable command-line flag for enabling them; "+ + "see https://docs.victoriametrics.com/victorialogs/#multitenant-querying") + return true + } + + // Process live tailing requests without timeout, since it is OK to run live tailing requests for very long time. + // Also do not apply concurrency limit to tail requests, since these limits are intended for non-tail requests. + switch path { + case "/select/logsql/tail", + "/select/multitenant/logsql/tail": + httpRequestTotalCounter(path).Inc() + logsql.ProcessLiveTailRequest(ctx, w, r, isMultiTenant) + return true + } + // Limit the number of concurrent queries, which can consume big amounts of CPU time. startTime := time.Now() d, err := getMaxQueryDuration(r) @@ -228,7 +239,7 @@ func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool { } defer decRequestConcurrency() - ok := processSelectRequest(ctxWithTimeout, w, r, path) + ok := processSelectRequest(ctxWithTimeout, w, r, path, isMultiTenant) if !ok { return false } @@ -313,79 +324,98 @@ func decRequestConcurrency() { <-concurrencyLimitCh } -func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, path string) bool { +func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, path string, isMultiTenant bool) bool { httpserver.EnableCORS(w, r) startTime := time.Now() switch path { case "/select/logsql/query_time_range": - logsqlQueryTimeRangeRequests.Inc() + httpRequestTotalCounter(path).Inc() logsql.ProcessQueryTimeRangeRequest(ctx, w, r) return true - case "/select/logsql/facets": - logsqlFacetsRequests.Inc() - logsql.ProcessFacetsRequest(ctx, w, r) - logsqlFacetsDuration.UpdateDuration(startTime) + case "/select/logsql/facets", + "/select/multitenant/logsql/facets": + httpRequestTotalCounter(path).Inc() + logsql.ProcessFacetsRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/field_names": - logsqlFieldNamesRequests.Inc() - logsql.ProcessFieldNamesRequest(ctx, w, r) - logsqlFieldNamesDuration.UpdateDuration(startTime) + case "/select/logsql/field_names", + "/select/multitenant/logsql/field_names": + httpRequestTotalCounter(path).Inc() + logsql.ProcessFieldNamesRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/field_values": - logsqlFieldValuesRequests.Inc() - logsql.ProcessFieldValuesRequest(ctx, w, r) - logsqlFieldValuesDuration.UpdateDuration(startTime) + case "/select/logsql/field_values", + "/select/multitenant/logsql/field_values": + httpRequestTotalCounter(path).Inc() + logsql.ProcessFieldValuesRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/hits": - logsqlHitsRequests.Inc() - logsql.ProcessHitsRequest(ctx, w, r) - logsqlHitsDuration.UpdateDuration(startTime) + case "/select/logsql/hits", + "/select/multitenant/logsql/hits": + httpRequestTotalCounter(path).Inc() + logsql.ProcessHitsRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/query": - logsqlQueryRequests.Inc() - logsql.ProcessQueryRequest(ctx, w, r) - logsqlQueryDuration.UpdateDuration(startTime) + case "/select/logsql/query", + "/select/multitenant/logsql/query": + httpRequestTotalCounter(path).Inc() + logsql.ProcessQueryRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/stats_query": - logsqlStatsQueryRequests.Inc() - logsql.ProcessStatsQueryRequest(ctx, w, r) - logsqlStatsQueryDuration.UpdateDuration(startTime) + case "/select/logsql/stats_query", + "/select/multitenant/logsql/stats_query": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStatsQueryRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/stats_query_range": - logsqlStatsQueryRangeRequests.Inc() - logsql.ProcessStatsQueryRangeRequest(ctx, w, r) - logsqlStatsQueryRangeDuration.UpdateDuration(startTime) + case "/select/logsql/stats_query_range", + "/select/multitenant/logsql/stats_query_range": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStatsQueryRangeRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/stream_field_names": - logsqlStreamFieldNamesRequests.Inc() - logsql.ProcessStreamFieldNamesRequest(ctx, w, r) - logsqlStreamFieldNamesDuration.UpdateDuration(startTime) + case "/select/logsql/stream_field_names", + "/select/multitenant/logsql/stream_field_names": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStreamFieldNamesRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/stream_field_values": - logsqlStreamFieldValuesRequests.Inc() - logsql.ProcessStreamFieldValuesRequest(ctx, w, r) - logsqlStreamFieldValuesDuration.UpdateDuration(startTime) + case "/select/logsql/stream_field_values", + "/select/multitenant/logsql/stream_field_values": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStreamFieldValuesRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/stream_ids": - logsqlStreamIDsRequests.Inc() - logsql.ProcessStreamIDsRequest(ctx, w, r) - logsqlStreamIDsDuration.UpdateDuration(startTime) + case "/select/logsql/stream_ids", + "/select/multitenant/logsql/stream_ids": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStreamIDsRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true - case "/select/logsql/streams": - logsqlStreamsRequests.Inc() - logsql.ProcessStreamsRequest(ctx, w, r) - logsqlStreamsDuration.UpdateDuration(startTime) + case "/select/logsql/streams", + "/select/multitenant/logsql/streams": + httpRequestTotalCounter(path).Inc() + logsql.ProcessStreamsRequest(ctx, w, r, isMultiTenant) + httpRequestDuration(path, startTime) return true case "/select/tenant_ids": - tenantIDsRequests.Inc() + httpRequestTotalCounter(path).Inc() logsql.ProcessTenantIDsRequest(ctx, w, r) - tenantIDsDuration.UpdateDuration(startTime) + httpRequestDuration(path, startTime) return true default: return false } } +func httpRequestTotalCounter(path string) *metrics.Counter { + return metrics.GetOrCreateCounter(fmt.Sprintf(`vl_http_requests_total{path=%q}`, path)) +} + +func httpRequestDuration(path string, startTime time.Time) { + metrics.GetOrCreateSummary(fmt.Sprintf(`vl_http_request_duration_seconds{path=%q}`, path)).UpdateDuration(startTime) +} + func deleteHandler(w http.ResponseWriter, r *http.Request, path string) { ctx := r.Context() @@ -479,48 +509,6 @@ func getMaxQueryDuration(r *http.Request) (time.Duration, error) { } var ( - logsqlFacetsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/facets"}`) - logsqlFacetsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/facets"}`) - - logsqlFieldNamesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_names"}`) - logsqlFieldNamesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/field_names"}`) - - logsqlFieldValuesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_values"}`) - logsqlFieldValuesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/field_values"}`) - - logsqlHitsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/hits"}`) - logsqlHitsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/hits"}`) - - logsqlQueryRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/query"}`) - logsqlQueryDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/query"}`) - - logsqlStatsQueryRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query"}`) - logsqlStatsQueryDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stats_query"}`) - - logsqlStatsQueryRangeRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query_range"}`) - logsqlStatsQueryRangeDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stats_query_range"}`) - - logsqlStreamFieldNamesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_field_names"}`) - logsqlStreamFieldNamesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_field_names"}`) - - logsqlStreamFieldValuesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_field_values"}`) - logsqlStreamFieldValuesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_field_values"}`) - - logsqlStreamIDsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_ids"}`) - logsqlStreamIDsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_ids"}`) - - logsqlStreamsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/streams"}`) - logsqlStreamsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/streams"}`) - - tenantIDsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/tenant_ids"}`) - tenantIDsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/tenant_ids"}`) - - // no need to track duration for tail requests, as they usually take long time - logsqlTailRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/tail"}`) - - // no need to track the duration for query_time_range requests, since they are instant - logsqlQueryTimeRangeRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/query_time_range"}`) - vmalertRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/vmalert"}`) // no need to track duration for /delete/* requests, because they are asynchronous diff --git a/app/vlstorage/netselect/netselect.go b/app/vlstorage/netselect/netselect.go index 5f5d464f73..058f143eb2 100644 --- a/app/vlstorage/netselect/netselect.go +++ b/app/vlstorage/netselect/netselect.go @@ -31,37 +31,37 @@ const ( // FieldNamesProtocolVersion is the version of the protocol used for /internal/select/field_names HTTP endpoint. // // It must be updated every time the protocol changes. - FieldNamesProtocolVersion = "v5" + FieldNamesProtocolVersion = "v6" // FieldValuesProtocolVersion is the version of the protocol used for /internal/select/field_values HTTP endpoint. // // It must be updated every time the protocol changes. - FieldValuesProtocolVersion = "v5" + FieldValuesProtocolVersion = "v6" // StreamFieldNamesProtocolVersion is the version of the protocol used for /internal/select/stream_field_names HTTP endpoint. // // It must be updated every time the protocol changes. - StreamFieldNamesProtocolVersion = "v5" + StreamFieldNamesProtocolVersion = "v6" // StreamFieldValuesProtocolVersion is the version of the protocol used for /internal/select/stream_field_values HTTP endpoint. // // It must be updated every time the protocol changes. - StreamFieldValuesProtocolVersion = "v5" + StreamFieldValuesProtocolVersion = "v6" // StreamsProtocolVersion is the version of the protocol used for /internal/select/streams HTTP endpoint. // // It must be updated every time the protocol changes. - StreamsProtocolVersion = "v5" + StreamsProtocolVersion = "v6" // StreamIDsProtocolVersion is the version of the protocol used for /internal/select/stream_ids HTTP endpoint. // // It must be updated every time the protocol changes. - StreamIDsProtocolVersion = "v5" + StreamIDsProtocolVersion = "v6" // QueryProtocolVersion is the version of the protocol used for /internal/select/query HTTP endpoint. // // It must be updated every time the protocol changes. - QueryProtocolVersion = "v5" + QueryProtocolVersion = "v6" // DeleteRunTaskProtocolVersion is the version of the protocol used for /internal/delete/run_task HTTP endpoint. // @@ -278,6 +278,7 @@ func (sn *storageNode) getCommonArgs(version string, qctx *logstorage.QueryConte args := url.Values{} args.Set("version", version) args.Set("tenant_ids", string(logstorage.MarshalTenantIDsToJSON(qctx.TenantIDs))) + args.Set("multitenant", fmt.Sprintf("%v", qctx.IsMultiTenant)) args.Set("query", qctx.Query.String()) args.Set("timestamp", fmt.Sprintf("%d", qctx.Query.GetTimestamp())) args.Set("disable_compression", fmt.Sprintf("%v", sn.s.disableCompression)) diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 41e90b6874..449eab187c 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/docs/victorialogs/CHANGELOG.md @@ -22,6 +22,8 @@ according to the following docs: ## tip +**Update note:** the internal protocol for `/internal/select/*` endpoints has been updated in order to support [multitenant querying](https://docs.victoriametrics.com/victorialogs/#multitenant-querying) in [VictoriaLogs cluster](https://docs.victoriametrics.com/victorialogs/cluster/). All VictoriaLogs cluster components must be upgraded together. The feature adds `/select/multitenant/logsql/*` endpoints for querying logs across multiple tenants in a single request. Enable these endpoints with `-multitenantselect.enable` and filter by `vl_account_id` / `vl_project_id` fields in LogsQL to select the needed tenants. These fields are also returned in query results unless they are removed by LogsQL pipes. See [#91](https://github.com/VictoriaMetrics/VictoriaLogs/issues/91). + * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): evenly spread rerouted data across available `vlstorage` nodes. Previously, healthy nodes adjacent to unavailable nodes in the `-storageNode` list could receive much more data, resulting in uneven resource usage. See [#1548](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1548). * BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/) and [querying](https://docs.victoriametrics.com/victorialogs/querying/): properly handle logs containing duplicate [stream field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) names. Previously, [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0) could panic when ingesting such logs in single-node VictoriaLogs, drop them during ingestion in VictoriaLogs cluster, or panic when querying such data written by earlier releases. See [#1603](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1603) and [#1604](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1604). diff --git a/docs/victorialogs/README.md b/docs/victorialogs/README.md index 00c225af5a..c03d784127 100644 --- a/docs/victorialogs/README.md +++ b/docs/victorialogs/README.md @@ -491,6 +491,29 @@ VictoriaLogs has very low overhead for per-tenant management, so it is OK to hav VictoriaLogs doesn't perform per-tenant authorization. Use [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/) or similar tools for per-tenant authorization. See [Security and Load balancing docs](https://docs.victoriametrics.com/victorialogs/security-and-lb/) for details. +### Multitenant querying + +By default, VictoriaLogs queries a single tenant per request. The requested tenant is selected via `AccountID` and `ProjectID` request headers. + +It is also possible to query multiple tenants in a single request via `/select/multitenant/logsql/*` endpoints. +These endpoints are disabled by default and can be enabled by passing `-multitenantselect.enable` command-line flag to VictoriaLogs or `vlselect`. +Requests to `/select/multitenant/logsql/*` endpoints must not contain `AccountID` and `ProjectID` request headers. + +Use `vl_account_id` and `vl_project_id` fields in LogsQL filters for limiting the query to the needed tenants. +For example, the following query searches logs containing the `error` word across `(AccountID=12, ProjectID=34)` tenant: + +```sh +curl http://localhost:9428/select/multitenant/logsql/query \ + -d 'query=vl_account_id:=12 vl_project_id:=34 error' +``` + +The `vl_account_id` and `vl_project_id` fields are generated from tenant IDs during multitenant querying. +They are returned in query results unless they are removed by LogsQL pipes such as [`fields`](https://docs.victoriametrics.com/victorialogs/logsql/#fields-pipe) or [`delete`](https://docs.victoriametrics.com/victorialogs/logsql/#delete-pipe). +They can also be used in LogsQL filters for selecting the needed tenants. + +The `/select/multitenant/logsql/*` endpoints can query all the stored tenants, so they must be protected with proper authorization. +See [Security and Load balancing docs](https://docs.victoriametrics.com/victorialogs/security-and-lb/) for details. + ### Multitenancy access control Enforce access control for tenants by using [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/). Access control can be configured for each tenant by setting up the following rules: diff --git a/docs/victorialogs/cluster.md b/docs/victorialogs/cluster.md index 16b920888d..6738f2551d 100644 --- a/docs/victorialogs/cluster.md +++ b/docs/victorialogs/cluster.md @@ -162,6 +162,8 @@ See [single-node vs cluster](https://victoriametrics.com/blog/victorialogs-archi This endpoint can be disabled via `-internalinsert.disable` command-line flag. See [security docs](https://docs.victoriametrics.com/victorialogs/cluster/#security) for details. - It accepts queries from `vlselect` via `/internal/select/*` HTTP endpoints at the TCP port specified via `-httpListenAddr` command-line flag. These endpoints can be disabled via `-internalselect.disable` command-line flag. See [security docs](https://docs.victoriametrics.com/victorialogs/cluster/#security) for details. + The `/internal/select/*` endpoints can receive both single-tenant and multitenant queries from `vlselect`; + they are internal cluster APIs and must not be exposed to untrusted clients. Every `vlstorage` node can be used as a single-node VictoriaLogs instance: @@ -178,6 +180,7 @@ Every `vlstorage` node can be used as a single-node VictoriaLogs instance: - `vlselect` can send queries to other `vlselect` nodes if they are specified via `-storageNode` command-line flag. This allows building multi-level cluster schemes when top-level `vlselect` queries multiple lower-level clusters of VictoriaLogs. If you don't want accepting queries from other `vlselect` nodes, then the `vlselect` node with `-internalselect.disable` command-line flag. + These `/internal/select/*` requests can be single-tenant or multitenant, depending on the original query. See [these docs](https://docs.victoriametrics.com/victorialogs/cluster/#tls) on how to protect communications between multiple levels of `vlinsert` and `vlselect` nodes. @@ -189,6 +192,10 @@ without direct access from the Internet. HTTP authorization proxies such as [vma must be used in front of `vlinsert` and `vlselect` for authorizing access to these components from the Internet. See [these docs](https://docs.victoriametrics.com/victorialogs/security-and-lb/) for details. +`vlselect` can serve `/select/multitenant/logsql/*` endpoints when `-multitenantselect.enable` command-line flag is set. +These endpoints can query multiple tenants in a single request, so they must be exposed only to trusted users. +See [multitenancy docs](https://docs.victoriametrics.com/victorialogs/#multitenant-querying) for details. + It is possible to disallow access to `/internal/insert` endpoint and `/internal/select/*` endpoints by running VictoriaLogs with `-internalinsert.disable` and `-internalselect.disable` command-line flags. diff --git a/docs/victorialogs/logsql.md b/docs/victorialogs/logsql.md index f83e9eaa19..1df30742dd 100644 --- a/docs/victorialogs/logsql.md +++ b/docs/victorialogs/logsql.md @@ -286,6 +286,10 @@ _time:5m | filter error_rate:>0.1; ``` +During [multitenant querying](https://docs.victoriametrics.com/victorialogs/#multitenant-querying), LogsQL can filter and return tenant IDs +via `vl_account_id` and `vl_project_id` fields. These fields are returned in query results unless they are removed by LogsQL pipes +such as [`fields`](https://docs.victoriametrics.com/victorialogs/logsql/#fields-pipe) or [`delete`](https://docs.victoriametrics.com/victorialogs/logsql/#delete-pipe). + ## Filters LogsQL supports various filters for searching for log messages (see below). diff --git a/docs/victorialogs/querying/README.md b/docs/victorialogs/querying/README.md index b71cb4b14f..2a733f3643 100644 --- a/docs/victorialogs/querying/README.md +++ b/docs/victorialogs/querying/README.md @@ -138,6 +138,9 @@ for log messages at `(AccountID=12, ProjectID=34)` tenant: curl http://localhost:9428/select/logsql/query -H 'AccountID: 12' -H 'ProjectID: 34' -d 'query=error' ``` +Use `/select/multitenant/logsql/query` endpoint for querying multiple tenants in a single request. +See [multitenancy docs](https://docs.victoriametrics.com/victorialogs/#multitenant-querying) for details. + The number of requests to `/select/logsql/query` can be [monitored](https://docs.victoriametrics.com/victorialogs/metrics/) with [`vl_http_requests_total{path="/select/logsql/query"}`](https://docs.victoriametrics.com/victorialogs/metrics/#vl_http_requests_total) metric. diff --git a/docs/victorialogs/security-and-lb.md b/docs/victorialogs/security-and-lb.md index e592708113..863b3e9340 100644 --- a/docs/victorialogs/security-and-lb.md +++ b/docs/victorialogs/security-and-lb.md @@ -144,6 +144,10 @@ similar to [this one](https://github.com/VictoriaMetrics/VictoriaLogs/issues/15# See [these docs](https://docs.victoriametrics.com/victoriametrics/vmauth/#routing-by-header) on how to setup request routing in `vmauth` by request headers. See [these docs](https://docs.victoriametrics.com/victoriametrics/vmauth/#modifying-http-headers) on how to modify request headers before proxying the requests to backends. +Requests to `/select/multitenant/logsql/*` endpoints can query multiple tenants in a single request when +[`-multitenantselect.enable`](https://docs.victoriametrics.com/victorialogs/#multitenant-querying) is set. +Expose these endpoints only to trusted users, or block them in `vmauth` configs for users who must access just a single tenant. + See also [tenant-based data ingestion request proxying](https://docs.victoriametrics.com/victorialogs/security-and-lb/#tenant-based-proxying-of-data-ingestion-requests). ### Proxying requests to the given tenants diff --git a/lib/logstorage/block_result.go b/lib/logstorage/block_result.go index 1c9bb6369c..452199ab10 100644 --- a/lib/logstorage/block_result.go +++ b/lib/logstorage/block_result.go @@ -431,6 +431,19 @@ func (br *blockResult) initColumnsByFilter(pf *prefixfilter.Filter) { } } + // Add tenant ID columns if multi-tenant search is enabled + bs := br.bs + so := bs.bsw.pso + bh := bs.bsw.bh + if so.isMultiTenant { + if pf.MatchString("vl_account_id") { + br.addConstColumn("vl_account_id", bh.streamID.tenantID.accountIDString()) + } + if pf.MatchString("vl_project_id") { + br.addConstColumn("vl_project_id", bh.streamID.tenantID.projectIDString()) + } + } + if pf.MatchString("_msg") { // Add _msg column v := br.bs.getConstColumnValue("_msg") @@ -444,13 +457,16 @@ func (br *blockResult) initColumnsByFilter(pf *prefixfilter.Filter) { } // Add other const columns - bs := br.bs csh := bs.getColumnsHeader() for _, cc := range csh.constColumns { if isSpecialColumn(cc.Name) { // Special columns have been added above. continue } + if so.isMultiTenant && isTenantColumn(cc.Name) { + continue + } + if pf.MatchString(cc.Name) && !bs.isHiddenField(cc.Name) { br.addConstColumn(cc.Name, cc.Value) } @@ -464,6 +480,10 @@ func (br *blockResult) initColumnsByFilter(pf *prefixfilter.Filter) { // Special columns have been added above. continue } + if so.isMultiTenant && isTenantColumn(ch.name) { + continue + } + if pf.MatchString(ch.name) && !bs.isHiddenField(ch.name) { br.addColumn(ch) } @@ -483,6 +503,12 @@ func isSpecialColumn(c string) bool { var specialColumns = []string{"_msg", "_time", "_stream", "_stream_id"} +func isTenantColumn(c string) bool { + return slices.Contains(tenantColumns, c) +} + +var tenantColumns = []string{"vl_account_id", "vl_project_id"} + // mustInit initializes br with the given bs and bm. // // br is valid until bs or bm changes. diff --git a/lib/logstorage/block_search.go b/lib/logstorage/block_search.go index 50c748bb71..3b86ec9b11 100644 --- a/lib/logstorage/block_search.go +++ b/lib/logstorage/block_search.go @@ -258,6 +258,15 @@ func (bs *blockSearch) getConstColumnValue(name string) string { } name = getCanonicalFieldName(name) + if bs.bsw.pso.isMultiTenant { + tenantID := bs.bsw.bh.streamID.tenantID + if name == "vl_account_id" { + return tenantID.accountIDString() + } + if name == "vl_project_id" { + return tenantID.projectIDString() + } + } if bs.partFormatVersion() < 1 { csh := bs.getColumnsHeader() diff --git a/lib/logstorage/filter_generic.go b/lib/logstorage/filter_generic.go index c7abc00f45..311ca6a4a6 100644 --- a/lib/logstorage/filter_generic.go +++ b/lib/logstorage/filter_generic.go @@ -179,12 +179,32 @@ func (fg *filterGeneric) applyToBlockSearch(bs *blockSearch, bm *bitmap) { } } - csh := bs.getColumnsHeader() + isMultiTenant := bs.bsw.pso.isMultiTenant + if isMultiTenant { + for _, cc := range tenantColumns { + if !strings.HasPrefix(cc, prefix) { + continue + } + if bs.isHiddenField(cc) { + continue + } + bmTmp.copyFrom(bmResult) + fg.f.applyToBlockSearchByField(bs, bmTmp, cc) + bmResult.andNot(bmTmp) + if bmResult.isZero() { + return + } + } + } + csh := bs.getColumnsHeader() for _, cc := range csh.constColumns { if isSpecialColumn(cc.Name) { continue } + if isMultiTenant && isTenantColumn(cc.Name) { + continue + } if !strings.HasPrefix(cc.Name, prefix) { continue } @@ -206,6 +226,9 @@ func (fg *filterGeneric) applyToBlockSearch(bs *blockSearch, bm *bitmap) { if isSpecialColumn(ch.name) { continue } + if isMultiTenant && isTenantColumn(ch.name) { + continue + } if !strings.HasPrefix(ch.name, prefix) { continue } diff --git a/lib/logstorage/pipe_field_names.go b/lib/logstorage/pipe_field_names.go index a03a4484ce..9017f6f93f 100644 --- a/lib/logstorage/pipe_field_names.go +++ b/lib/logstorage/pipe_field_names.go @@ -136,6 +136,13 @@ func (pfp *pipeFieldNamesProcessor) writeBlock(workerID uint, br *blockResult) { shard.updateColumnHits("_time", filter, hits) shard.updateColumnHits("_stream", filter, hits) shard.updateColumnHits("_stream_id", filter, hits) + if br.bs.bsw.pso.isMultiTenant { + for _, columnName := range tenantColumns { + if !br.bs.isHiddenField(columnName) { + shard.updateColumnHits(columnName, filter, hits) + } + } + } } } @@ -145,6 +152,10 @@ func (shard *pipeFieldNamesProcessorShard) updateHits(refs []columnHeaderRef, br if br.bs.isHiddenField(columnName) { continue } + if br.bs.bsw.pso.isMultiTenant && isTenantColumn(columnName) { + // The stored column is shadowed by the automatically generated tenant column. + continue + } shard.updateColumnHits(columnName, filter, hits) } } diff --git a/lib/logstorage/pipe_stream_context.go b/lib/logstorage/pipe_stream_context.go index cdae9f0033..5fadb67db3 100644 --- a/lib/logstorage/pipe_stream_context.go +++ b/lib/logstorage/pipe_stream_context.go @@ -327,7 +327,7 @@ func (pcp *pipeStreamContextProcessor) executeQuery(streamID, qStr string, neede } qctxOrig := pcp.pc.qctx - qctx := NewQueryContext(ctxWithCancel, qctxOrig.QueryStats, []TenantID{tenantID}, q, qctxOrig.AllowPartialResponse, qctxOrig.HiddenFieldsFilters) + qctx := NewQueryContext(ctxWithCancel, qctxOrig.QueryStats, []TenantID{tenantID}, qctxOrig.IsMultiTenant, q, qctxOrig.AllowPartialResponse, qctxOrig.HiddenFieldsFilters) if err := pcp.pc.runQuery(qctx, writeBlock); err != nil { return nil, 0, err } diff --git a/lib/logstorage/storage.go b/lib/logstorage/storage.go index 884b56c6a8..56f8199419 100644 --- a/lib/logstorage/storage.go +++ b/lib/logstorage/storage.go @@ -1000,7 +1000,7 @@ func (s *Storage) processDeleteTask(ctx context.Context, dt *DeleteTask) bool { q.AddTimeFilter(math.MinInt64, now) var qs QueryStats - qctx := NewQueryContext(ctx, &qs, dt.TenantIDs, q, false, nil) + qctx := NewQueryContext(ctx, &qs, dt.TenantIDs, false, q, false, nil) // Initialize subqueries qNew, err := initSubqueries(qctx, s.runQuery, false) @@ -1010,7 +1010,7 @@ func (s *Storage) processDeleteTask(ctx context.Context, dt *DeleteTask) bool { } q = qNew - sso := s.getSearchOptions(dt.TenantIDs, q, qctx.HiddenFieldsFilters) + sso := s.getSearchOptions(dt.TenantIDs, false, q, qctx.HiddenFieldsFilters) // reset fieldsFilter in order to avoid loading all the log fields // during search for parts which contain rows to delete, since these fields aren't needed. diff --git a/lib/logstorage/storage_search.go b/lib/logstorage/storage_search.go index c48ecfafb1..e6433e26d3 100644 --- a/lib/logstorage/storage_search.go +++ b/lib/logstorage/storage_search.go @@ -46,6 +46,11 @@ type QueryContext struct { // Prefixes match all the fields starting with the given prefix. HiddenFieldsFilters []string + // IsMultiTenant indicates whether the query is multitenant. + // + // If true, the query is executed as a multitenant query, i.e. the tenant ids are taken from the query itself. + IsMultiTenant bool + // startTime is creation time for the QueryContext. // // It is used for calculating query duration. @@ -53,24 +58,24 @@ type QueryContext struct { } // NewQueryContext returns new context for the given query. -func NewQueryContext(ctx context.Context, qs *QueryStats, tenantIDs []TenantID, q *Query, allowPartialResponse bool, hiddenFieldsFilters []string) *QueryContext { +func NewQueryContext(ctx context.Context, qs *QueryStats, tenantIDs []TenantID, isMultiTenant bool, q *Query, allowPartialResponse bool, hiddenFieldsFilters []string) *QueryContext { startTime := time.Now() - return newQueryContext(ctx, qs, tenantIDs, q, allowPartialResponse, hiddenFieldsFilters, startTime) + return newQueryContext(ctx, qs, tenantIDs, isMultiTenant, q, allowPartialResponse, hiddenFieldsFilters, startTime) } // WithQuery returns new QueryContext with the given q, while preserving other fields from qctx. func (qctx *QueryContext) WithQuery(q *Query) *QueryContext { - return newQueryContext(qctx.Context, qctx.QueryStats, qctx.TenantIDs, q, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) + return newQueryContext(qctx.Context, qctx.QueryStats, qctx.TenantIDs, qctx.IsMultiTenant, q, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) } // WithContext returns new QueryContext with the given ctx, while preserving other fields from qctx. func (qctx *QueryContext) WithContext(ctx context.Context) *QueryContext { - return newQueryContext(ctx, qctx.QueryStats, qctx.TenantIDs, qctx.Query, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) + return newQueryContext(ctx, qctx.QueryStats, qctx.TenantIDs, qctx.IsMultiTenant, qctx.Query, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) } // WithContextAndQuery returns new QueryContext with the given ctx and q, while preserving other fields from qctx. func (qctx *QueryContext) WithContextAndQuery(ctx context.Context, q *Query) *QueryContext { - return newQueryContext(ctx, qctx.QueryStats, qctx.TenantIDs, q, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) + return newQueryContext(ctx, qctx.QueryStats, qctx.TenantIDs, qctx.IsMultiTenant, q, qctx.AllowPartialResponse, qctx.HiddenFieldsFilters, qctx.startTime) } // QueryDurationNsecs returns the duration in nanoseconds since the NewQueryContext call. @@ -78,17 +83,18 @@ func (qctx *QueryContext) QueryDurationNsecs() int64 { return time.Since(qctx.startTime).Nanoseconds() } -func newQueryContext(ctx context.Context, qs *QueryStats, tenantIDs []TenantID, q *Query, allowPartialResponse bool, hiddenFieldsFilters []string, startTime time.Time) *QueryContext { +func newQueryContext(ctx context.Context, qs *QueryStats, tenantIDs []TenantID, isMultiTenant bool, q *Query, allowPartialResponse bool, hiddenFieldsFilters []string, startTime time.Time) *QueryContext { if q.opts.allowPartialResponse != nil { // query options override other settings for allowPartialResponse. allowPartialResponse = *q.opts.allowPartialResponse } return &QueryContext{ - Context: ctx, - QueryStats: qs, - TenantIDs: tenantIDs, - Query: q, + Context: ctx, + QueryStats: qs, + TenantIDs: tenantIDs, + IsMultiTenant: isMultiTenant, + Query: q, AllowPartialResponse: allowPartialResponse, HiddenFieldsFilters: hiddenFieldsFilters, @@ -104,6 +110,9 @@ type storageSearchOptions struct { // tenantIDs must contain the list of tenantIDs for the search. tenantIDs []TenantID + // isMultiTenant indicates whether search is performed from multitenant query + isMultiTenant bool + // streamIDs is an optional sorted list of streamIDs for the search. // If it is empty, then the search is performed by tenantIDs streamIDs []streamID @@ -141,6 +150,10 @@ type partitionSearchOptions struct { // If it is empty, then the search is performed by streamIDs tenantIDs []TenantID + // isMultiTenant indicates whether search is performed from multitenant query + // If it is true, then the blockSearch should init vl_account_id and vl_project_id columns + isMultiTenant bool + // Optional sorted list of streamIDs for the search. // If it is empty, then the search is performed by tenantIDs streamIDs []streamID @@ -223,8 +236,18 @@ func (s *Storage) runQuery(qctx *QueryContext, writeBlock writeBlockResultFunc) } q := qNew - sso := s.getSearchOptions(qctx.TenantIDs, q, qctx.HiddenFieldsFilters) + // A non-empty TenantIDs means the caller already resolved the exact tenants to search + // (e.g. the stream_context pipe scopes its subquery to a single tenant derived from the streamID). + // Only resolve tenants from the query filter when the caller left TenantIDs unresolved. + tenantIDs := qctx.TenantIDs + if qctx.IsMultiTenant && len(qctx.TenantIDs) == 0 { + tenantIDs, err = s.getTenantIDsForQuery(qctx.Context, q) + if err != nil { + return fmt.Errorf("cannot get tenant IDs: %w", err) + } + } + sso := s.getSearchOptions(tenantIDs, qctx.IsMultiTenant, q, qctx.HiddenFieldsFilters) search := func(stopCh <-chan struct{}, writeBlockToPipes writeBlockResultFunc) error { workersCount := q.GetParallelReaders(s.defaultParallelReaders) s.searchParallel(workersCount, sso, qctx.QueryStats, stopCh, writeBlockToPipes) @@ -235,7 +258,78 @@ func (s *Storage) runQuery(qctx *QueryContext, writeBlock writeBlockResultFunc) return runPipes(qctx, q.pipes, search, writeBlock, concurrency) } -func (s *Storage) getSearchOptions(tenantIDs []TenantID, q *Query, hiddenFieldsFilters []string) *storageSearchOptions { +func (s *Storage) getTenantIDsForQuery(ctx context.Context, q *Query) ([]TenantID, error) { + start, end := q.GetFilterTimeRange() + allTenantIDs, err := s.getTenantIDs(ctx, start, end) + if err != nil { + return nil, fmt.Errorf("cannot get tenant IDs: %w", err) + } + + if len(allTenantIDs) == 0 { + return allTenantIDs, nil + } + + f := getTenantFilter(q) + var tenantIDs []TenantID + for _, tenantID := range allTenantIDs { + rows := []Field{ + { + Name: "vl_account_id", + Value: tenantID.accountIDString(), + }, + { + Name: "vl_project_id", + Value: tenantID.projectIDString(), + }, + } + if f.matchRow(rows) { + tenantIDs = append(tenantIDs, tenantID) + } + } + + return tenantIDs, nil +} + +// getTenantFilter returns a tenant-only filter for finding tenants that MAY match q. +// +// It is used only for selecting candidate tenants before block search. The full +// query filter is still applied later during block scan. +func getTenantFilter(q *Query) filter { + qf := q.getFinalFilter() + + isNonTenantFieldFilter := func(fg *filterGeneric) bool { + return fg.isWildcard || !isTenantColumn(fg.fieldName) + } + + visitFunc := func(f filter) bool { + switch t := f.(type) { + case *filterAnd, *filterOr: + return false + case *filterGeneric: + return isNonTenantFieldFilter(t) + case *filterNot: + // keep only simple negated tenant filters. For complex NOT filters, + // replacing inner non-tenant filters with noop may change the result. + fg, ok := t.f.(*filterGeneric) + return !ok || isNonTenantFieldFilter(fg) + default: + return true + } + } + + copyFunc := func(f filter) (filter, error) { + return newFilterNoop(), nil + } + + f, err := copyFilter(qf, visitFunc, copyFunc) + if err != nil { + logger.Panicf("BUG: copyFunc is noop, error shouldn't be happened") + } + + return f +} + +func (s *Storage) getSearchOptions(tenantIDs []TenantID, isMultitenant bool, q *Query, hiddenFieldsFilters []string) *storageSearchOptions { // tenantIDs must be sorted, since block search performs binary search over them. sortTenantIDs(tenantIDs) @@ -260,6 +354,7 @@ func (s *Storage) getSearchOptions(tenantIDs []TenantID, q *Query, hiddenFieldsF return &storageSearchOptions{ tenantIDs: tenantIDs, streamIDs: streamIDs, + isMultiTenant: isMultitenant, minTimestamp: minTimestamp, maxTimestamp: maxTimestamp, streamFilter: sf, @@ -1448,6 +1543,7 @@ func (pt *partition) getSearchOptions(sso *storageSearchOptions) *partitionSearc } return &partitionSearchOptions{ tenantIDs: tenantIDs, + isMultiTenant: sso.isMultiTenant, streamIDs: streamIDs, minTimestamp: sso.minTimestamp, maxTimestamp: sso.maxTimestamp, diff --git a/lib/logstorage/storage_search_test.go b/lib/logstorage/storage_search_test.go index a33313c8b3..ef0c31ff80 100644 --- a/lib/logstorage/storage_search_test.go +++ b/lib/logstorage/storage_search_test.go @@ -175,6 +175,35 @@ func TestStorageRunQuery(t *testing.T) { t.Fatalf("unexpected number of matching rows; got %d; want %d", n, expectedRowsCount) } }) + t.Run("matching-multitenant-query-tenant-filter", func(t *testing.T) { + f := func(t *testing.T, query string, rowsExpected int) { + t.Helper() + + q := mustParseQuery(query) + qs := &QueryStats{} + qctx := NewQueryContext(context.Background(), qs, nil, true, q, false, nil) + + var rowsCountTotal atomic.Uint32 + writeBlock := func(_ uint, db *DataBlock) { + rowsCountTotal.Add(uint32(db.RowsCount())) + } + if err := s.RunQuery(qctx, writeBlock); err != nil { + t.Fatalf("unexpected error returned from the query [%s]: %s", q, err) + } + + if n := rowsCountTotal.Load(); n != uint32(rowsExpected) { + t.Fatalf("unexpected number of matching rows for query %q; got %d; want %d", query, n, rowsExpected) + } + } + + rowsPerTenant := streamsPerTenant * blocksPerStream * rowsPerBlock + rowsPerTenantStream := blocksPerStream * rowsPerBlock + rowsTotal := tenantsCount * rowsPerTenant + f(t, `vl_account_id:2`, rowsPerTenant) + f(t, `vl_account_id:2 AND _stream:{job="foobar",instance="host-1:234"}`, rowsPerTenantStream) + f(t, `vl_account_id:2 OR vl_project_id:31`, 2*rowsPerTenant) + f(t, `!vl_account_id:2`, rowsTotal-rowsPerTenant) + }) t.Run("matching-in-filter", func(t *testing.T) { q := mustParseQuery(`source-file:in(foobar,/foo/bar/baz)`) var rowsCountTotal atomic.Uint32 @@ -1716,5 +1745,104 @@ func storeRowsForSearchHiddenFieldsFilters(s *Storage, tenantIDs []TenantID, now func newTestQueryContext(tenantIDs []TenantID, q *Query) *QueryContext { qs := &QueryStats{} - return NewQueryContext(context.Background(), qs, tenantIDs, q, false, nil) + return NewQueryContext(context.Background(), qs, tenantIDs, false, q, false, nil) +} + +func TestStorageGetFieldNamesMultitenant(t *testing.T) { + t.Parallel() + + path := t.TempDir() + s := MustOpenStorage(path, &StorageConfig{ + Retention: 24 * time.Hour, + }) + defer s.MustClose() + + now := time.Now().UnixNano() + lr := GetLogRows(nil, nil, nil, nil, "") + for i := range 3 { + tenantID := TenantID{ + AccountID: uint32(i + 1), + ProjectID: uint32(i + 11), + } + fields := []Field{ + {Name: "_msg", Value: fmt.Sprintf("message-%d", i)}, + } + if i < 2 { + fields = append(fields, + Field{Name: "vl_account_id", Value: "stored-account-id"}, + Field{Name: "vl_project_id", Value: "stored-project-id"}, + ) + } + lr.mustAdd(tenantID, now+int64(i), fields) + } + s.MustAddRows(lr) + PutLogRows(lr) + s.DebugFlush() + + q := mustParseQuery("*") + qs := &QueryStats{} + qctx := NewQueryContext(context.Background(), qs, nil, true, q, false, nil) + results, err := s.GetFieldNames(qctx, "vl_") + if err != nil { + t.Fatalf("unexpected error: %s", err) + } + + resultsExpected := []ValueWithHits{ + {"vl_account_id", 3}, + {"vl_project_id", 3}, + } + if !reflect.DeepEqual(results, resultsExpected) { + t.Fatalf("unexpected result; got\n%v\nwant\n%v", results, resultsExpected) + } +} + +func TestGetTenantFilter(t *testing.T) { + t.Parallel() + + tenantRows := [][]Field{ + { + {Name: "vl_account_id", Value: "1"}, + {Name: "vl_project_id", Value: "11"}, + }, + { + {Name: "vl_account_id", Value: "2"}, + {Name: "vl_project_id", Value: "21"}, + }, + { + {Name: "vl_account_id", Value: "1"}, + {Name: "vl_project_id", Value: "21"}, + }, + } + + f := func(query string, matchingTenantIndexesExpected []int) { + t.Helper() + + q := mustParseQuery(query) + tf := getTenantFilter(q) + + var matchingTenantIndexes []int + for i, rows := range tenantRows { + if tf.matchRow(rows) { + matchingTenantIndexes = append(matchingTenantIndexes, i) + } + } + if !reflect.DeepEqual(matchingTenantIndexes, matchingTenantIndexesExpected) { + t.Fatalf("unexpected tenant filter matches for query %q; got %v; want %v", query, matchingTenantIndexes, matchingTenantIndexesExpected) + } + } + + f(`vl_account_id:1`, []int{0, 2}) + f(`vl_account_id:=1`, []int{0, 2}) + f(`vl_account_id:~"1|2"`, []int{0, 1, 2}) + f(`vl_account_id:in(1,2)`, []int{0, 1, 2}) + f(`vl_project_id:21`, []int{1, 2}) + f(`vl_account_id:1 AND vl_project_id:21`, []int{2}) + f(`vl_account_id:1 AND foo:bar`, []int{0, 2}) + f(`vl_account_id:2 OR vl_project_id:11`, []int{0, 1}) + f(`vl_account_id:2 OR foo:bar`, []int{0, 1, 2}) + f(`foo:bar`, []int{0, 1, 2}) + f(`!vl_account_id:1`, []int{1}) + f(`vl_account_id:!~"1"`, []int{1}) + f(`!foo:bar`, []int{0, 1, 2}) + f(`!(vl_account_id:1 OR vl_project_id:21)`, []int{0, 1, 2}) } diff --git a/lib/logstorage/storage_test.go b/lib/logstorage/storage_test.go index 69f34c0333..970a979bcf 100644 --- a/lib/logstorage/storage_test.go +++ b/lib/logstorage/storage_test.go @@ -515,7 +515,7 @@ func checkQueryResults(t *testing.T, s *Storage, now int64, tenantIDs []TenantID ctx := t.Context() var qs QueryStats - qctx := NewQueryContext(ctx, &qs, tenantIDs, q, false, hiddenFieldsFilters) + qctx := NewQueryContext(ctx, &qs, tenantIDs, false, q, false, hiddenFieldsFilters) var buf []byte var bufLock sync.Mutex diff --git a/lib/logstorage/tenant_id.go b/lib/logstorage/tenant_id.go index be9471c59e..40cea4eeff 100644 --- a/lib/logstorage/tenant_id.go +++ b/lib/logstorage/tenant_id.go @@ -34,6 +34,16 @@ func (tid TenantID) String() string { return fmt.Sprintf("{accountID=%d,projectID=%d}", tid.AccountID, tid.ProjectID) } +// accountIDString returns the string representation of tid.AccountID. +func (tid TenantID) accountIDString() string { + return strconv.FormatUint(uint64(tid.AccountID), 10) +} + +// projectIDString returns the string representation of tid.ProjectID. +func (tid TenantID) projectIDString() string { + return strconv.FormatUint(uint64(tid.ProjectID), 10) +} + // Equal returns true if tid equals to a. func (tid *TenantID) Equal(a *TenantID) bool { return tid.AccountID == a.AccountID && tid.ProjectID == a.ProjectID @@ -96,6 +106,10 @@ func (tid *TenantID) unmarshal(src []byte) ([]byte, error) { return src[8:], nil } +func HasTenantIDFromRequest(r *http.Request) bool { + return r.Header.Get("AccountID") != "" || r.Header.Get("ProjectID") != "" +} + // GetTenantIDFromRequest returns tenantID from r. func GetTenantIDFromRequest(r *http.Request) (TenantID, error) { var tenantID TenantID