feat: add --flux-log-enabled flag for detailed flux logs

CHANGELOG.md

@@ -47,6 +47,7 @@ This release adds an embedded SQLite database for storing metadata required by t
 1. [21936](https://github.com/influxdata/influxdb/pull/21936): Ported the `influxd inspect build-tsi` command from 1.x.
 1. [21938](https://github.com/influxdata/influxdb/pull/21938): Added route to delete individual secret.
 1. [21972](https://github.com/influxdata/influxdb/pull/21972): Added support for notebooks and annotations.
+1. [22072](https://github.com/influxdata/influxdb/pull/22072): Added `--flux-log-enabled` option to `influxd` to show detail logs for flux queries.
 1. [22135](https://github.com/influxdata/influxdb/pull/22135): Added route to return known resources.
 1. [22311](https://github.com/influxdata/influxdb/pull/22311): Add `storage-no-validate-field-size` config to `influxd` to disable enforcement of max field size.
 

cmd/influxd/launcher/cmd.go

@@ -123,6 +123,7 @@ type InfluxdOpts struct {
 	TestingAlwaysAllowSetup bool
 
 	LogLevel          zapcore.Level
+	FluxLogEnabled    bool
 	TracingType       string
 	ReportingDisabled bool
 
@@ -184,6 +185,7 @@ func NewOpts(viper *viper.Viper) *InfluxdOpts {
 		CoordinatorConfig: coordinator.NewConfig(),
 
 		LogLevel:          zapcore.InfoLevel,
+		FluxLogEnabled:    false,
 		ReportingDisabled: false,
 
 		BoltPath:   filepath.Join(dir, bolt.DefaultFilename),
@@ -231,6 +233,12 @@ func (o *InfluxdOpts) BindCliOpts() []cli.Opt {
 			Default: o.LogLevel,
 			Desc:    "supported log levels are debug, info, and error",
 		},
+		{
+			DestP:   &o.FluxLogEnabled,
+			Flag:    "flux-log-enabled",
+			Default: o.FluxLogEnabled,
+			Desc:    "enables detailed logging for flux queries",
+		},
 		{
 			DestP: &o.TracingType,
 			Flag:  "tracing-type",

cmd/influxd/launcher/launcher.go

@@ -766,6 +766,7 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 		UIDisabled:           opts.UIDisabled,
 		HTTPErrorHandler:     kithttp.ErrorHandler(0),
 		Logger:               m.log,
+		FluxLogEnabled:       opts.FluxLogEnabled,
 		SessionRenewDisabled: opts.SessionRenewDisabled,
 		NewQueryService:      source.NewQueryService,
 		PointsWriter: &storage.LoggingPointsWriter{

http/api_handler.go

@@ -33,9 +33,10 @@ type APIHandler struct {
 // APIBackend is all services and associated parameters required to construct
 // an APIHandler.
 type APIBackend struct {
-	AssetsPath string // if empty then assets are served from bindata.
-	UIDisabled bool   // if true requests for the UI will return 404
-	Logger     *zap.Logger
+	AssetsPath     string // if empty then assets are served from bindata.
+	UIDisabled     bool   // if true requests for the UI will return 404
+	Logger         *zap.Logger
+	FluxLogEnabled bool
 	errors.HTTPErrorHandler
 	SessionRenewDisabled bool
 	// MaxBatchSizeBytes is the maximum number of bytes which can be written

http/query_handler.go

@@ -17,6 +17,7 @@ import (
 	"github.com/influxdata/flux/ast"
 	"github.com/influxdata/flux/csv"
 	"github.com/influxdata/flux/iocounter"
+	"github.com/influxdata/flux/lang"
 	"github.com/influxdata/httprouter"
 	"github.com/influxdata/influxdb/v2"
 	pcontext "github.com/influxdata/influxdb/v2/context"
@@ -46,6 +47,7 @@ const (
 type FluxBackend struct {
 	errors2.HTTPErrorHandler
 	log                *zap.Logger
+	FluxLogEnabled     bool
 	QueryEventRecorder metric.EventRecorder
 
 	AlgoWProxy          FeatureProxyHandler
@@ -60,6 +62,7 @@ func NewFluxBackend(log *zap.Logger, b *APIBackend) *FluxBackend {
 	return &FluxBackend{
 		HTTPErrorHandler:   b.HTTPErrorHandler,
 		log:                log,
+		FluxLogEnabled:     b.FluxLogEnabled,
 		QueryEventRecorder: b.QueryEventRecorder,
 		AlgoWProxy:         b.AlgoWProxy,
 		ProxyQueryService: routingQueryService{
@@ -81,7 +84,8 @@ type HTTPDialect interface {
 type FluxHandler struct {
 	*httprouter.Router
 	errors2.HTTPErrorHandler
-	log *zap.Logger
+	log            *zap.Logger
+	FluxLogEnabled bool
 
 	Now                 func() time.Time
 	OrganizationService influxdb.OrganizationService
@@ -105,6 +109,7 @@ func NewFluxHandler(log *zap.Logger, b *FluxBackend) *FluxHandler {
 		Now:              time.Now,
 		HTTPErrorHandler: b.HTTPErrorHandler,
 		log:              log,
+		FluxLogEnabled:   b.FluxLogEnabled,
 
 		ProxyQueryService:   b.ProxyQueryService,
 		OrganizationService: b.OrganizationService,
@@ -196,7 +201,8 @@ func (h *FluxHandler) handleQuery(w http.ResponseWriter, r *http.Request) {
 	hd.SetHeaders(w)
 
 	cw := iocounter.Writer{Writer: w}
-	if _, err := h.ProxyQueryService.Query(ctx, &cw, req); err != nil {
+	stats, err := h.ProxyQueryService.Query(ctx, &cw, req)
+	if err != nil {
 		if cw.Count() == 0 {
 			// Only record the error headers IFF nothing has been written to w.
 			h.HandleHTTPError(ctx, err, w)
@@ -208,6 +214,33 @@ func (h *FluxHandler) handleQuery(w http.ResponseWriter, r *http.Request) {
 			zap.Error(err),
 		)
 	}
+
+	// Detailed logging for flux queries if enabled
+	if h.FluxLogEnabled {
+		h.logFluxQuery(cw.Count(), stats, req.Request.Compiler, err)
+	}
+
+}
+
+func (h *FluxHandler) logFluxQuery(n int64, stats flux.Statistics, compiler flux.Compiler, err error) {
+	var q string
+	c, ok := compiler.(lang.FluxCompiler)
+	if !ok {
+		q = "unknown"
+	}
+	q = c.Query
+
+	h.log.Info("Executed Flux query",
+		zap.String("compiler_type", string(compiler.CompilerType())),
+		zap.Int64("response_size", n),
+		zap.String("query", q),
+		zap.Error(err),
+		zap.Duration("stat_total_duration", stats.TotalDuration),
+		zap.Duration("stat_compile_duration", stats.CompileDuration),
+		zap.Duration("stat_execute_duration", stats.ExecuteDuration),
+		zap.Int64("stat_max_allocated", stats.MaxAllocated),
+		zap.Int64("stat_total_allocated", stats.TotalAllocated),
+	)
 }
 
 type langRequest struct {