Skip to content

feat: report user query bytes (#27188) #27198

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davidby-influx merged 1 commit into from
Feb 6, 2026
Merged

feat: report user query bytes (#27188) #27198

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions pkg/data/gensyncmap/gensyncmap.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,3 +45,12 @@ func (m *Map[K, V]) Len() int {
})
return n
}

func (m *Map[K, V]) IsEmpty() bool {
isEmpty := true
m.m.Range(func(_, _ any) bool {
isEmpty = false
return false
})
return isEmpty
}
68 changes: 35 additions & 33 deletions services/httpd/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,39 +32,41 @@ const (

// Config represents a configuration for a HTTP service.
type Config struct {
Enabled bool `toml:"enabled"`
BindAddress string `toml:"bind-address"`
AuthEnabled bool `toml:"auth-enabled"`
LogEnabled bool `toml:"log-enabled"`
SuppressWriteLog bool `toml:"suppress-write-log"`
WriteTracing bool `toml:"write-tracing"`
FluxEnabled bool `toml:"flux-enabled"`
FluxLogEnabled bool `toml:"flux-log-enabled"`
FluxTesting bool `toml:"flux-testing"`
PprofEnabled bool `toml:"pprof-enabled"`
PprofAuthEnabled bool `toml:"pprof-auth-enabled"`
DebugPprofEnabled bool `toml:"debug-pprof-enabled"`
PingAuthEnabled bool `toml:"ping-auth-enabled"`
PromReadAuthEnabled bool `toml:"prom-read-auth-enabled"`
HTTPHeaders map[string]string `toml:"headers"`
HTTPSEnabled bool `toml:"https-enabled"`
HTTPSCertificate string `toml:"https-certificate"`
HTTPSPrivateKey string `toml:"https-private-key"`
MaxRowLimit int `toml:"max-row-limit"`
MaxConnectionLimit int `toml:"max-connection-limit"`
SharedSecret string `toml:"shared-secret"`
Realm string `toml:"realm"`
UnixSocketEnabled bool `toml:"unix-socket-enabled"`
UnixSocketGroup *toml.Group `toml:"unix-socket-group"`
UnixSocketPermissions toml.FileMode `toml:"unix-socket-permissions"`
BindSocket string `toml:"bind-socket"`
MaxBodySize int `toml:"max-body-size"`
AccessLogPath string `toml:"access-log-path"`
AccessLogStatusFilters []StatusFilter `toml:"access-log-status-filters"`
MaxConcurrentWriteLimit int `toml:"max-concurrent-write-limit"`
MaxEnqueuedWriteLimit int `toml:"max-enqueued-write-limit"`
EnqueuedWriteTimeout time.Duration `toml:"enqueued-write-timeout"`
TLS *tls.Config `toml:"-"`
Enabled bool `toml:"enabled"`
BindAddress string `toml:"bind-address"`
AuthEnabled bool `toml:"auth-enabled"`
LogEnabled bool `toml:"log-enabled"`
SuppressWriteLog bool `toml:"suppress-write-log"`
WriteTracing bool `toml:"write-tracing"`
FluxEnabled bool `toml:"flux-enabled"`
FluxLogEnabled bool `toml:"flux-log-enabled"`
FluxTesting bool `toml:"flux-testing"`
PprofEnabled bool `toml:"pprof-enabled"`
PprofAuthEnabled bool `toml:"pprof-auth-enabled"`
DebugPprofEnabled bool `toml:"debug-pprof-enabled"`
PingAuthEnabled bool `toml:"ping-auth-enabled"`
PromReadAuthEnabled bool `toml:"prom-read-auth-enabled"`
HTTPHeaders map[string]string `toml:"headers"`
HTTPSEnabled bool `toml:"https-enabled"`
HTTPSCertificate string `toml:"https-certificate"`
HTTPSPrivateKey string `toml:"https-private-key"`
HTTPSInsecureCertificate bool `toml:"https-insecure-certificate"`
MaxRowLimit int `toml:"max-row-limit"`
MaxConnectionLimit int `toml:"max-connection-limit"`
SharedSecret string `toml:"shared-secret"`
Realm string `toml:"realm"`
UnixSocketEnabled bool `toml:"unix-socket-enabled"`
UnixSocketGroup *toml.Group `toml:"unix-socket-group"`
UnixSocketPermissions toml.FileMode `toml:"unix-socket-permissions"`
BindSocket string `toml:"bind-socket"`
MaxBodySize int `toml:"max-body-size"`
AccessLogPath string `toml:"access-log-path"`
AccessLogStatusFilters []StatusFilter `toml:"access-log-status-filters"`
MaxConcurrentWriteLimit int `toml:"max-concurrent-write-limit"`
MaxEnqueuedWriteLimit int `toml:"max-enqueued-write-limit"`
EnqueuedWriteTimeout time.Duration `toml:"enqueued-write-timeout"`
UserQueryBytesEnabled bool `toml:"user-query-bytes-enabled"`
TLS *tls.Config `toml:"-"`
}

// NewConfig returns a new Config with default settings.
Expand Down
78 changes: 63 additions & 15 deletions services/httpd/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/influxdata/influxdb/models"
"github.com/influxdata/influxdb/monitor"
"github.com/influxdata/influxdb/monitor/diagnostics"
"github.com/influxdata/influxdb/pkg/data/gensyncmap"
"github.com/influxdata/influxdb/prometheus"
"github.com/influxdata/influxdb/query"
"github.com/influxdata/influxdb/services/meta"
Expand Down Expand Up @@ -143,12 +144,13 @@ type Handler struct {
Controller Controller
CompilerMappings flux.CompilerMappings

Config *Config
Logger *zap.Logger
CLFLogger *log.Logger
accessLog *os.File
accessLogFilters StatusFilters
stats *Statistics
Config *Config
Logger *zap.Logger
CLFLogger *log.Logger
accessLog *os.File
accessLogFilters StatusFilters
stats *Statistics
queryBytesPerUser gensyncmap.Map[string, *atomic.Int64]

requestTracker *RequestTracker
writeThrottler *Throttler
Expand Down Expand Up @@ -322,7 +324,12 @@ func NewHandler(c Config) *Handler {
return func(w http.ResponseWriter, r *http.Request, user meta.User) {
// TODO: This is the only place we use AuthorizeUnrestricted. It would be better to use an explicit permission
if user == nil || !user.AuthorizeUnrestricted() {
h.Logger.Info("Unauthorized request", zap.String("user", user.ID()), zap.String("path", r.URL.Path))
// Don't panic
id := ""
if user != nil {
id = user.ID()
}
h.Logger.Info("Unauthorized request", zap.String("user", id), zap.String("path", r.URL.Path))
h.httpError(w, "error authorizing admin access", http.StatusForbidden)
return
}
Expand Down Expand Up @@ -442,7 +449,7 @@ type Statistics struct {

// Statistics returns statistics for periodic monitoring.
func (h *Handler) Statistics(tags map[string]string) []models.Statistic {
return []models.Statistic{{
stats := []models.Statistic{{
Name: "httpd",
Tags: tags,
Values: map[string]interface{}{
Expand Down Expand Up @@ -472,6 +479,35 @@ func (h *Handler) Statistics(tags map[string]string) []models.Statistic {
statFluxQueryRequestBytesTransmitted: atomic.LoadInt64(&h.stats.FluxQueryRequestBytesTransmitted),
},
}}

// Add per-user query bytes as separate statistics (one per user) if enabled
if h.Config.UserQueryBytesEnabled && !h.queryBytesPerUser.IsEmpty() {
h.queryBytesPerUser.Range(func(user string, counter *atomic.Int64) bool {
userTag := user
if user == "" {
userTag = StatAnonymousUser
}
userTags := models.NewTags(tags).Merge(map[string]string{StatUserTagKey: userTag}).Map()
stats = append(stats, models.Statistic{
Name: "userquerybytes",
Tags: userTags,
Values: map[string]interface{}{statUserQueryRespBytes: counter.Load()},
})
return true
})
}

return stats
}

// addQueryBytesForUser atomically adds bytes to the per-user query bytes counter.
// This is a no-op if UserQueryBytesEnabled is false.
func (h *Handler) addQueryBytesForUser(user string, n int64) {
if !h.Config.UserQueryBytesEnabled {
return
}
counter, _ := h.queryBytesPerUser.LoadOrStore(user, &atomic.Int64{})
counter.Add(n)
}

// AddRoutes sets the provided routes on the handler.
Expand Down Expand Up @@ -782,6 +818,7 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user meta.U
Results: []*query.Result{r},
})
atomic.AddInt64(&h.stats.QueryRequestBytesTransmitted, int64(n))
h.addQueryBytesForUser(userName, int64(n))
w.(http.Flusher).Flush()
continue
}
Expand Down Expand Up @@ -873,6 +910,7 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user meta.U
if !chunked {
n, _ := rw.WriteResponse(resp)
atomic.AddInt64(&h.stats.QueryRequestBytesTransmitted, int64(n))
h.addQueryBytesForUser(userName, int64(n))
}
}

Expand Down Expand Up @@ -2031,6 +2069,11 @@ func (h *Handler) servePromRead(w http.ResponseWriter, r *http.Request, user met
return
}

var userName string
if user != nil {
userName = user.ID()
}

respond := func(resp *prompb.ReadResponse) {
data, err := resp.Marshal()
if err != nil {
Expand All @@ -2048,6 +2091,7 @@ func (h *Handler) servePromRead(w http.ResponseWriter, r *http.Request, user met
}

atomic.AddInt64(&h.stats.QueryRequestBytesTransmitted, int64(len(compressed)))
h.addQueryBytesForUser(userName, int64(len(compressed)))
}

ctx := context.Background()
Expand Down Expand Up @@ -2134,6 +2178,11 @@ func (h *Handler) serveFluxQuery(w http.ResponseWriter, r *http.Request, user me
atomic.AddInt64(&h.stats.FluxQueryRequestDuration, time.Since(start).Nanoseconds())
}(time.Now())

var userName string
if user != nil {
userName = user.ID()
}

req, err := decodeQueryRequest(r)
if err != nil {
h.httpError(w, err.Error(), http.StatusBadRequest)
Expand Down Expand Up @@ -2201,14 +2250,13 @@ func (h *Handler) serveFluxQuery(w http.ResponseWriter, r *http.Request, user me
defer results.Release()

n, err = encoder.Encode(w, results)
if err != nil {
if n == 0 {
// If the encoder did not write anything, we can write an error header.
h.httpError(w, err.Error(), http.StatusInternalServerError)
} else {
atomic.AddInt64(&h.stats.FluxQueryRequestBytesTransmitted, int64(n))
}
if err != nil && n == 0 {
// If the encoder did not write anything, we can write an error header.
h.httpError(w, err.Error(), http.StatusInternalServerError)
return
}
atomic.AddInt64(&h.stats.FluxQueryRequestBytesTransmitted, int64(n))
h.addQueryBytesForUser(userName, int64(n))
}
}

Expand Down
Loading