Merge branch 'main' into suvij/eng-6332-store-persisted-operation-content-in-the-database-and-show

Author: JivusAyrus

router-tests/introspection_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/sebdah/goldie/v2"
+
 	"github.com/wundergraph/cosmo/router-tests/testenv"
 )
 

router-tests/testdata/introspection/base-graph-schema.json

@@ -5,15 +5,16 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
 	"github.com/wundergraph/cosmo/router/core"
 	"github.com/wundergraph/cosmo/router/internal/versioninfo"
 	"github.com/wundergraph/cosmo/router/pkg/config"
 	"github.com/wundergraph/cosmo/router/pkg/logging"
 	"github.com/wundergraph/cosmo/router/pkg/profile"
-	"log"
-	"os"
-	"os/signal"
-	"syscall"
 
 	"go.uber.org/zap"
 )
@@ -29,7 +30,6 @@ var (
 )
 
 func Main() {
-
 	// Parse flags before calling profile.Start(), since it may add flags
 	flag.Parse()
 

router-tests/testdata/introspection/feature-graph-schema.json

@@ -1,92 +1,11 @@
 package main
 
 import (
-	"flag"
-	"log"
 	"os"
-	"path/filepath"
-	"slices"
-	"strings"
-	"time"
 
-	"github.com/wundergraph/cosmo/router/core"
-)
-
-var (
-	executionConfigFilePath    = flag.String("execution-config", "config.json", "execution config file location")
-	sourceOperationFoldersPath = flag.String("operations", "operations", "source operations folder location")
-	plansOutPath               = flag.String("plans", "plans", "output plans folder location")
-	operationFilterFilePath    = flag.String("filter", "", "operation filter file location which should contain file names of operations to include")
+	"github.com/wundergraph/cosmo/router/cmd"
 )
 
 func main() {
-	flag.Parse()
-
-	queriesPath, err := filepath.Abs(*sourceOperationFoldersPath)
-	if err != nil {
-		log.Fatalf("failed to get absolute path for queries: %v", err)
-	}
-
-	outPath, err := filepath.Abs(*plansOutPath)
-	if err != nil {
-		log.Fatalf("failed to get absolute path for output: %v", err)
-	}
-	if err := os.MkdirAll(outPath, 0755); err != nil {
-		log.Fatalf("failed to create output directory: %v", err)
-	}
-
-	supergraphConfigPath, err := filepath.Abs(*executionConfigFilePath)
-	log.Println("supergraphPath:", supergraphConfigPath)
-	if err != nil {
-		log.Fatalf("failed to get absolute path for supergraph: %v", err)
-	}
-
-	queries, err := os.ReadDir(queriesPath)
-	if err != nil {
-		log.Fatalf("failed to read queries directory: %v", err)
-	}
-
-	var filter []string
-	if *operationFilterFilePath != "" {
-		filterContent, err := os.ReadFile(*operationFilterFilePath)
-		if err != nil {
-			log.Fatalf("failed to read filter file: %v", err)
-		}
-
-		filter = strings.Split(string(filterContent), "\n")
-	}
-
-	pg, err := core.NewPlanGenerator(supergraphConfigPath)
-	if err != nil {
-		log.Fatalf("failed to create plan generator: %v", err)
-	}
-
-	t := time.Now()
-
-	for i, queryFile := range queries {
-		if filepath.Ext(queryFile.Name()) != ".graphql" {
-			continue
-		}
-
-		if len(filter) > 0 && !slices.Contains(filter, queryFile.Name()) {
-			continue
-		}
-
-		log.Println("Running query #", i, " name:", queryFile.Name())
-
-		queryFilePath := filepath.Join(queriesPath, queryFile.Name())
-
-		outContent, err := pg.PlanOperation(queryFilePath)
-		if err != nil {
-			log.Printf("failed operation #%d: %s err: %v\n", i, queryFile.Name(), err.Error())
-		}
-
-		outFileName := filepath.Join(outPath, queryFile.Name())
-		err = os.WriteFile(outFileName, []byte(outContent), 0644)
-		if err != nil {
-			log.Fatalf("failed to write file: %v", err)
-		}
-	}
-
-	log.Println("Total planning time:", time.Since(t))
+	cmd.PlanGenerator(os.Args)
 }

router/cmd/main.go

@@ -0,0 +1,58 @@
+package cmd
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/wundergraph/cosmo/router/pkg/plan_generator"
+)
+
+func PlanGenerator(args []string) {
+	var planHelp bool
+
+	cfg := plan_generator.QueryPlanConfig{}
+	f := flag.NewFlagSet("router "+args[0], flag.ExitOnError)
+	f.BoolVar(&planHelp, "help", false, "Prints the help message")
+	f.StringVar(&cfg.ExecutionConfig, "execution-config", "", "required, execution config file location")
+	f.StringVar(&cfg.SourceDir, "operations", "", "required, source operations folder location")
+	f.StringVar(&cfg.OutDir, "plans", "", "required, output plans folder location")
+	f.StringVar(&cfg.Filter, "filter", "", "operation filter file location which should contain file names of operations to include")
+	f.StringVar(&cfg.Timeout, "timeout", "30s", "timeout")
+	f.IntVar(&cfg.Concurrency, "concurrency", 0, "how many query plan run concurrently")
+	f.BoolVar(&cfg.OutputFiles, "print-per-file", true, "write a file for each query, with inside the plan or the query plan error")
+	f.BoolVar(&cfg.OutputReport, "print-report", false, "write a report.json file, with all the query plans and errors sorted by file name")
+	f.BoolVar(&cfg.FailOnPlanError, "fail-on-error", false, "if at least one plan fails, the command exit code will be 1")
+	f.BoolVar(&cfg.FailFast, "fail-fast", false, "stop as soon as possible if a plan fails")
+
+	if err := f.Parse(args[1:]); err != nil {
+		f.PrintDefaults()
+		log.Fatalf("Failed to parse flags: %v", err)
+	}
+
+	if planHelp {
+		f.PrintDefaults()
+		return
+	}
+	if cfg.ExecutionConfig == "" || cfg.SourceDir == "" || cfg.OutDir == "" {
+		f.PrintDefaults()
+		log.Fatalf("missing required flags")
+	}
+
+	ctxNotify, stop := signal.NotifyContext(context.Background(), os.Interrupt,
+		syscall.SIGHUP,  // process is detached from terminal
+		syscall.SIGTERM, // default for kill
+		syscall.SIGKILL,
+		syscall.SIGQUIT, // ctrl + \
+		syscall.SIGINT,  // ctrl+c
+	)
+	defer stop()
+
+	err := plan_generator.PlanGenerator(ctxNotify, cfg)
+	if err != nil {
+		log.Fatalf("Error during command plan-generator: %s", err)
+	}
+}

router/cmd/plan-generator/main.go

@@ -1,9 +1,15 @@
 package main
 
 import (
+	"os"
+
 	routercmd "github.com/wundergraph/cosmo/router/cmd"
 )
 
 func main() {
-	routercmd.Main()
+	if len(os.Args) > 1 && os.Args[1] == "query-plan" {
+		routercmd.PlanGenerator(os.Args[1:])
+	} else {
+		routercmd.Main()
+	}
 }

router/cmd/plan_generator.go

@@ -8,12 +8,14 @@ import (
 	"os"
 
 	log "github.com/jensneuse/abstractlogger"
-
+	nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/ast"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/astnormalization"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/astparser"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/asttransform"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource"
+	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/introspection_datasource"
+	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/plan"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/postprocess"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
@@ -112,6 +114,27 @@ func (pg *PlanGenerator) loadConfiguration(configFilePath string) error {
 		return err
 	}
 
+	natSources := map[string]pubsub_datasource.NatsPubSub{}
+	kafkaSources := map[string]pubsub_datasource.KafkaPubSub{}
+	for _, ds := range routerConfig.GetEngineConfig().GetDatasourceConfigurations() {
+		if ds.GetKind() != nodev1.DataSourceKind_PUBSUB || ds.GetCustomEvents() == nil {
+			continue
+		}
+		for _, natConfig := range ds.GetCustomEvents().GetNats() {
+			providerId := natConfig.GetEngineEventConfiguration().GetProviderId()
+			if _, ok := natSources[providerId]; !ok {
+				natSources[providerId] = nil
+			}
+		}
+		for _, kafkaConfig := range ds.GetCustomEvents().GetKafka() {
+			providerId := kafkaConfig.GetEngineEventConfiguration().GetProviderId()
+			if _, ok := kafkaSources[providerId]; !ok {
+				kafkaSources[providerId] = nil
+			}
+		}
+	}
+	pubSubFactory := pubsub_datasource.NewFactory(context.Background(), natSources, kafkaSources)
+
 	var netPollConfig graphql_datasource.NetPollConfiguration
 	netPollConfig.ApplyDefaults()
 
@@ -129,6 +152,7 @@ func (pg *PlanGenerator) loadConfiguration(configFilePath string) error {
 		streamingClient:    http.DefaultClient,
 		subscriptionClient: subscriptionClient,
 		transportOptions:   &TransportOptions{SubgraphTransportOptions: NewSubgraphTransportOptions(config.TrafficShapingRules{})},
+		pubsub:             pubSubFactory,
 	})
 
 	// this generates the plan configuration using the data source factories from the config package
@@ -162,6 +186,23 @@ func (pg *PlanGenerator) loadConfiguration(configFilePath string) error {
 		return fmt.Errorf("failed to merge graphql schema with base schema: %w", err)
 	}
 
+	// by default, the engine doesn't understand how to resolve the __schema and __type queries
+	// we need to add a special datasource for that
+	// it takes the definition as the input and generates introspection data
+	// datasource is attached to Query.__schema, Query.__type, __Type.fields and __Type.enumValues fields
+	introspectionFactory, err := introspection_datasource.NewIntrospectionConfigFactory(&definition)
+	if err != nil {
+		return fmt.Errorf("failed to create introspection config factory: %w", err)
+	}
+	dataSources := introspectionFactory.BuildDataSourceConfigurations()
+
+	fieldConfigs := introspectionFactory.BuildFieldConfigurations()
+	// we need to add these fields to the config
+	// otherwise the engine wouldn't know how to resolve them
+	planConfig.Fields = append(planConfig.Fields, fieldConfigs...)
+	// finally, we add our data source for introspection to the existing data sources
+	planConfig.DataSources = append(planConfig.DataSources, dataSources...)
+
 	pg.planConfiguration = planConfig
 	pg.definition = &definition
 	return nil