SSE: Localize/Contain Errors within an Expression (#73163)

Changes SSE to not always fail all queries when one fails. Now only the query itself, and nodes that depend on it will error.
---------

Co-authored-by: Gilles De Mey <gilles.de.mey@gmail.com>

pkg/expr/nodes.go

@@ -85,6 +85,10 @@ func (gn *CMDNode) NodeType() NodeType {
 	return TypeCMDNode
 }
 
+func (gn *CMDNode) NeedsVars() []string {
+	return gn.Command.NeedsVars()
+}
+
 // Execute runs the node and adds the results to vars. If the node requires
 // other nodes they must have already been executed and their results must
 // already by in vars.
@@ -151,6 +155,11 @@ func (dn *DSNode) NodeType() NodeType {
 	return TypeDatasourceNode
 }
 
+// NodeType returns the data pipeline node type.
+func (dn *DSNode) NeedsVars() []string {
+	return []string{}
+}
+
 func (s *Service) buildDSNode(dp *simple.DirectedGraph, rn *rawNode, req *Request) (*DSNode, error) {
 	if rn.TimeRange == nil {
 		return nil, fmt.Errorf("time range must be specified for refID %s", rn.RefID)
@@ -196,7 +205,7 @@ func (s *Service) buildDSNode(dp *simple.DirectedGraph, rn *rawNode, req *Reques
 
 // executeDSNodesGrouped groups datasource node queries by the datasource instance, and then sends them
 // in a single request with one or more queries to the datasource.
-func executeDSNodesGrouped(ctx context.Context, now time.Time, vars mathexp.Vars, s *Service, nodes []*DSNode) (e error) {
+func executeDSNodesGrouped(ctx context.Context, now time.Time, vars mathexp.Vars, s *Service, nodes []*DSNode) {
 	type dsKey struct {
 		uid   string // in theory I think this all I need for the key, but rather be safe
 		id    int64
@@ -209,13 +218,16 @@ func executeDSNodesGrouped(ctx context.Context, now time.Time, vars mathexp.Vars
 	}
 
 	for _, nodeGroup := range byDS {
-		if err := func() error {
+		func() {
 			ctx, span := s.tracer.Start(ctx, "SSE.ExecuteDatasourceQuery")
 			defer span.End()
 			firstNode := nodeGroup[0]
 			pCtx, err := s.pCtxProvider.GetWithDataSource(ctx, firstNode.datasource.Type, firstNode.request.User, firstNode.datasource)
 			if err != nil {
-				return err
+				for _, dn := range nodeGroup {
+					vars[dn.refID] = mathexp.Results{Error: datasources.ErrDataSourceNotFound}
+				}
+				return
 			}
 
 			logger := logger.FromContext(ctx).New("datasourceType", firstNode.datasource.Type,
@@ -243,9 +255,9 @@ func executeDSNodesGrouped(ctx context.Context, now time.Time, vars mathexp.Vars
 				})
 			}
 
-			responseType := "unknown"
-			respStatus := "success"
-			defer func() {
+			instrument := func(e error, rt string) {
+				respStatus := "success"
+				responseType := rt
 				if e != nil {
 					responseType = "error"
 					respStatus = "failure"
@@ -258,32 +270,35 @@ func executeDSNodesGrouped(ctx context.Context, now time.Time, vars mathexp.Vars
 				logger.Debug("Data source queried", "responseType", responseType)
 				useDataplane := strings.HasPrefix(responseType, "dataplane-")
 				s.metrics.dsRequests.WithLabelValues(respStatus, fmt.Sprintf("%t", useDataplane), firstNode.datasource.Type).Inc()
-			}()
+			}
 
 			resp, err := s.dataService.QueryData(ctx, req)
 			if err != nil {
-				return MakeQueryError(firstNode.refID, firstNode.datasource.UID, err)
+				for _, dn := range nodeGroup {
+					vars[dn.refID] = mathexp.Results{Error: MakeQueryError(firstNode.refID, firstNode.datasource.UID, err)}
+				}
+				instrument(err, "")
+				return
 			}
 
 			for _, dn := range nodeGroup {
 				dataFrames, err := getResponseFrame(resp, dn.refID)
 				if err != nil {
-					return MakeQueryError(dn.refID, dn.datasource.UID, err)
+					vars[dn.refID] = mathexp.Results{Error: MakeQueryError(dn.refID, dn.datasource.UID, err)}
+					instrument(err, "")
+					return
 				}
 
 				var result mathexp.Results
-				responseType, result, err = convertDataFramesToResults(ctx, dataFrames, dn.datasource.Type, s, logger)
+				responseType, result, err := convertDataFramesToResults(ctx, dataFrames, dn.datasource.Type, s, logger)
 				if err != nil {
-					return MakeConversionError(dn.refID, err)
+					result.Error = makeConversionError(dn.RefID(), err)
 				}
+				instrument(err, responseType)
 				vars[dn.refID] = result
 			}
-			return nil
-		}(); err != nil {
-			return err
-		}
+		}()
 	}
-	return nil
 }
 
 // Execute runs the node and adds the results to vars. If the node requires
@@ -346,7 +361,7 @@ func (dn *DSNode) Execute(ctx context.Context, now time.Time, _ mathexp.Vars, s
 	var result mathexp.Results
 	responseType, result, err = convertDataFramesToResults(ctx, dataFrames, dn.datasource.Type, s, logger)
 	if err != nil {
-		err = MakeConversionError(dn.refID, err)
+		err = makeConversionError(dn.refID, err)
 	}
 	return result, err
 }