pkg/expr/transform.go - main - githubmirror/grafana

pkg/expr/transform.go
141 строка3 KB
Nick Richmond
Expressions: Isolate broken SSE pipeline nodes to enable partial results (#121043)
01 апр 2026, 20:32
01 апр 2026, 20:3242a3f3a
package expr

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/grafana/grafana-plugin-sdk-go/backend"

	"github.com/grafana/grafana/pkg/apimachinery/identity"
	"github.com/grafana/grafana/pkg/services/datasources"
)

// Request is similar to plugins.DataQuery but with the Time Ranges is per Query.
type Request struct {
	Headers map[string]string
	Debug   bool
	OrgId   int64
	Queries []Query
	User    identity.Requester
}

// Query is like plugins.DataSubQuery, but with a a time range, and only the UID
// for the data source. Also interval is a time.Duration.
type Query struct {
	RefID         string
	TimeRange     TimeRange
	DataSource    *datasources.DataSource `json:"datasource"`
	JSON          json.RawMessage
	Interval      time.Duration
	QueryType     string
	MaxDataPoints int64
}

// TimeRange is a time.Time based TimeRange.
type TimeRange interface {
	AbsoluteTime(now time.Time) backend.TimeRange
}

type AbsoluteTimeRange struct {
	From time.Time
	To   time.Time
}

func (r AbsoluteTimeRange) AbsoluteTime(_ time.Time) backend.TimeRange {
	return backend.TimeRange{
		From: r.From,
		To:   r.To,
	}
}

// RelativeTimeRange is a time range relative to some absolute time.
type RelativeTimeRange struct {
	From time.Duration
	To   time.Duration
}

func (r RelativeTimeRange) AbsoluteTime(t time.Time) backend.TimeRange {
	return backend.TimeRange{
		From: t.Add(r.From),
		To:   t.Add(r.To),
	}
}

// TransformData takes Queries which are either expressions nodes
// or are datasource requests.
func (s *Service) TransformData(ctx context.Context, now time.Time, req *Request) (r *backend.QueryDataResponse, err error) {
	if s.isDisabled() {
		return nil, fmt.Errorf("server side expressions are disabled")
	}

	start := time.Now()
	ctx, span := s.tracer.Start(ctx, "SSE.TransformData")
	defer func() {
		var respStatus string
		switch err {
		case nil:
			respStatus = "success"
		default:
			respStatus = "failure"
		}
		duration := float64(time.Since(start).Nanoseconds()) / float64(time.Millisecond)
		s.metrics.ExpressionsQuerySummary.WithLabelValues(respStatus).Observe(duration)

		span.End()
	}()

	// Build the pipeline from the request, checking for ordering issues (e.g. loops)
	// and parsing graph nodes from the queries.
	pipeline, err := s.buildPipeline(ctx, req)
	if err != nil {
		return nil, err
	}

	// Execute the pipeline. Disabled nodes (e.g. those with missing dependencies)
	// inject their errors into the result vars during execution; downstream nodes
	// fail via the existing Mode 1 dependency-error path.
	responses, err := s.ExecutePipeline(ctx, now, pipeline)
	if err != nil {
		return nil, err
	}

	// Get which queries have the Hide property so those queries' results
	// can be excluded from the response.
	hidden, err := hiddenRefIDs(req.Queries)
	if err != nil {
		return nil, err
	}

	if len(hidden) != 0 {
		filteredRes := backend.NewQueryDataResponse()
		for refID, res := range responses.Responses {
			if _, ok := hidden[refID]; !ok {
				filteredRes.Responses[refID] = res
			}
		}
		responses = filteredRes
	}

	return responses, nil
}

func hiddenRefIDs(queries []Query) (map[string]struct{}, error) {
	hidden := make(map[string]struct{})

	for _, query := range queries {
		hide := struct {
			Hide bool `json:"hide"`
		}{}

		if err := json.Unmarshal(query.JSON, &hide); err != nil {
			return nil, err
		}

		if hide.Hide {
			hidden[query.RefID] = struct{}{}
		}
	}
	return hidden, nil
}