diff --git a/benchmarks/rps/.gitignore b/benchmarks/rps/.gitignore index b9e99e68..e69de29b 100644 --- a/benchmarks/rps/.gitignore +++ b/benchmarks/rps/.gitignore @@ -1,2 +0,0 @@ -orb-rps-client -orb-rps-server \ No newline at end of file diff --git a/benchmarks/rps/cmd/orb-rps-client/config.go b/benchmarks/rps/cmd/orb-rps-client/config.go new file mode 100644 index 00000000..7f22205c --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-client/config.go @@ -0,0 +1,131 @@ +package main + +import ( + "errors" + "runtime" + + "github.com/go-orb/go-orb/config/source/cli" +) + +const ( + configSection = "bench_client" + + defaultBypassRegistry = 1 + defaultConnections = 256 + defaultDuration = 15 + defaultTimeout = 8 + defaultTransport = "grpc" + defaultPackageSize = 1000 + defaultContentType = "application/x-protobuf" +) + +//nolint:gochecknoglobals +var ( + defaultThreads = runtime.NumCPU() +) + +func init() { + err := cli.Flags.Add(cli.NewFlag( + "bypass_registry", + defaultBypassRegistry, + cli.ConfigPathSlice([]string{configSection, "bypassRegistry"}), + cli.Usage("Bypasses the registry by caching it, set to 0 to disable"), + cli.EnvVars("BYPASS_REGISTRY"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "connections", + defaultConnections, + cli.ConfigPathSlice([]string{configSection, "connections"}), + cli.Usage("Connections to keep open"), + cli.EnvVars("CONNECTIONS"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "duration", + defaultDuration, + cli.ConfigPathSlice([]string{configSection, "duration"}), + cli.Usage("Duration in seconds"), + cli.EnvVars("DURATION"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "timeout", + defaultTimeout, + cli.ConfigPathSlice([]string{configSection, "timeout"}), + cli.Usage("Timeout in seconds"), + cli.EnvVars("TIMEOUT"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + // function init is to long..... + init2() +} + +func init2() { + err := cli.Flags.Add(cli.NewFlag( + "threads", + defaultThreads, + cli.ConfigPathSlice([]string{configSection, "threads"}), + cli.Usage("Number of threads to use = runtime.GOMAXPROCS()"), + cli.EnvVars("THREADS"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "transport", + defaultTransport, + cli.ConfigPathSlice([]string{configSection, "transport"}), + cli.Usage("Transport to use (grpc, hertzhttp, http, uvm.)"), + cli.EnvVars("TRANSPORT"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "package_size", + defaultPackageSize, + cli.ConfigPathSlice([]string{configSection, "packageSize"}), + cli.Usage("Per request package size"), + cli.EnvVars("PACKAGE_SIZE"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } + + err = cli.Flags.Add(cli.NewFlag( + "content_type", + defaultContentType, + cli.ConfigPathSlice([]string{configSection, "contentType"}), + cli.Usage("Content-Type (application/x-protobuf, application/json)"), + cli.EnvVars("CONTENT_TYPE"), + )) + if err != nil && !errors.Is(err, cli.ErrFlagExists) { + panic(err) + } +} + +type clientConfig struct { + BypassRegistry int `json:"bypassRegistry" yaml:"bypassRegistry"` + Connections int `json:"connections" yaml:"connections"` + Duration int `json:"duration" yaml:"duration"` + Timeout int `json:"timeout" yaml:"timeout"` + Threads int `json:"threads" yaml:"threads"` + Transport string `json:"transport" yaml:"transport"` + PackageSize int `json:"packageSize" yaml:"packageSize"` + ContentType string `json:"contentType" yaml:"contentType"` +} diff --git a/benchmarks/rps/cmd/orb-rps-client/config.yaml b/benchmarks/rps/cmd/orb-rps-client/config.yaml new file mode 100644 index 00000000..1e5e6202 --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-client/config.yaml @@ -0,0 +1,16 @@ +benchmarks: + rps: + client: + logger: + plugin: slog + format: json + level: INFO + registry: + plugin: consul + # client: + # middlewares: + # - name: log + # logger: + # plugin: slog + # format: json + # level: INFO diff --git a/benchmarks/rps/cmd/orb-rps-client/main.go b/benchmarks/rps/cmd/orb-rps-client/main.go new file mode 100644 index 00000000..e67d77b3 --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-client/main.go @@ -0,0 +1,320 @@ +// bench_client contains a client to benchmark `tests_server`. +package main + +import ( + "bytes" + "context" + "crypto/rand" + "errors" + "fmt" + "os" + "os/signal" + "runtime" + "sync" + "syscall" + "time" + + // go-orb. + "github.com/go-orb/go-orb/client" + "github.com/go-orb/go-orb/config" + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/types" + + // Own imports. + echoproto "github.com/go-orb/plugins/benchmarks/rps/proto/echo" + + _ "github.com/go-orb/plugins/client/orb" + _ "github.com/go-orb/plugins/codecs/jsonpb" + _ "github.com/go-orb/plugins/codecs/proto" + _ "github.com/go-orb/plugins/codecs/yaml" + _ "github.com/go-orb/plugins/config/source/cli/urfave" + _ "github.com/go-orb/plugins/config/source/file" + _ "github.com/go-orb/plugins/log/slog" + _ "github.com/go-orb/plugins/registry/consul" + _ "github.com/go-orb/plugins/registry/mdns" + + _ "github.com/go-orb/plugins/client/middleware/log" + + // All transports. + _ "github.com/go-orb/plugins/client/orb_transport/drpc" + _ "github.com/go-orb/plugins/client/orb_transport/grpc" + _ "github.com/go-orb/plugins/client/orb_transport/h2c" + _ "github.com/go-orb/plugins/client/orb_transport/hertzh2c" + _ "github.com/go-orb/plugins/client/orb_transport/hertzhttp" + _ "github.com/go-orb/plugins/client/orb_transport/http" + _ "github.com/go-orb/plugins/client/orb_transport/http3" + _ "github.com/go-orb/plugins/client/orb_transport/https" +) + +const serverName = "benchmarks.rps.server" + +type stats struct { + Ok uint64 + Error uint64 +} + +func connection( + ctx context.Context, + wg *sync.WaitGroup, + cli client.Type, + logger log.Logger, + msg []byte, + opts []client.CallOption, + connectionNum int, + statsChan chan stats, +) { + var ( + reqsOk uint64 + reqsError uint64 + ) + + for { + select { + case <-ctx.Done(): + logger.Debug("Connection results", "connection", connectionNum, "reqsOk", reqsOk, "reqsError", reqsError) + wg.Done() + + statsChan <- stats{Ok: reqsOk, Error: reqsError} + + return + default: + } + + // Create a request. + req := &echoproto.Req{Payload: msg} + + // Run the query. + resp, err := client.Call[echoproto.Resp]( + ctx, + cli, + serverName, + "echo.Echo/Echo", + req, + opts..., + ) + if err != nil { + if errors.Is(err, context.Canceled) { + continue + } + + logger.Error("while requesting", "error", err) + + reqsError++ + + continue + } + + // Check if response equals. + if !bytes.Equal(req.GetPayload(), resp.GetPayload()) { + logger.Error("request and response are not the same") + + reqsError++ + + continue + } + + reqsOk++ + } +} + +// bench. +// +//nolint:funlen +func bench( + sn types.ServiceName, + configs types.ConfigData, + logger log.Logger, + cli client.Type, +) error { + cfg := &clientConfig{ + BypassRegistry: defaultBypassRegistry, + Connections: defaultConnections, + Duration: defaultDuration, + Timeout: defaultTimeout, + Threads: defaultThreads, + Transport: defaultTransport, + PackageSize: defaultPackageSize, + ContentType: defaultContentType, + } + + sections := append(types.SplitServiceName(sn), configSection) + if err := config.Parse(sections, configs, &cfg); err != nil { + return err + } + + logger.Info( + "Config", + "bypass_registry", cfg.BypassRegistry, + "connections", cfg.Connections, + "duration", cfg.Duration, + "timeout", cfg.Timeout, + "threads", cfg.Threads, + "transport", cfg.Transport, + "package_size", cfg.PackageSize, + "content_type", cfg.ContentType, + ) + + runtime.GOMAXPROCS(cfg.Threads) + + // Setup client options. + opts := []client.CallOption{ + client.WithPoolSize(cfg.Connections), + client.WithPreferredTransports(cfg.Transport), + client.WithContentType(cfg.ContentType), + } + + if err := cli.With(client.WithClientPoolSize(cfg.Connections)); err != nil { + return err + } + + wCtx, wCancel := context.WithCancel(context.Background()) + + // Cache URL + if cfg.BypassRegistry == 1 { + logger.Debug("Resolving", "server", serverName) + + nodes, err := cli.ResolveService(wCtx, serverName, cfg.Transport) + if err != nil { + logger.Error("Failed to resolve service, did you start the server?", "error", err, "server", serverName) + wCancel() + + return err + } + + var preferredTransports []string + if len(cfg.Transport) != 0 { + preferredTransports = []string{cfg.Transport} + } else { + preferredTransports = cli.Config().PreferredTransports + } + + node, err := cli.Config().Selector(wCtx, serverName, nodes, preferredTransports, false) + if err != nil { + logger.Error("Failed to resolve service, did you start the server?", "error", err, "server", serverName) + wCancel() + + return err + } + + opts = append(opts, client.WithURL(fmt.Sprintf("%s://%s", node.Transport, node.Address))) + + logger.Info("Using transport", "transport", node.Transport) + } + + // Create random bytes to ping-pong on each request. + msg := make([]byte, cfg.PackageSize) + if _, err := rand.Reader.Read(msg); err != nil { + logger.Error("Failed to make a request", "error", err) + wCancel() + + return err + } + + var wg sync.WaitGroup + + quit := make(chan os.Signal, 1) + done := make(chan os.Signal, 1) + + // End requests on SIGINT/SIGTERM. + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + + // + // Warmup + // + + timer := time.AfterFunc(time.Second*time.Duration(cfg.Duration), func() { + done <- syscall.SIGINT + }) + + logger.Info("Warming up...") + + nullChan := make(chan stats, cfg.Connections) + + for i := 0; i < cfg.Connections; i++ { + wg.Add(1) + + go connection(wCtx, &wg, cli, logger, msg, opts, i, nullChan) + } + + select { + case <-done: + wCancel() + timer.Stop() + case <-quit: + timer.Stop() + os.Exit(1) + } + + // + // Bench + // + logger.Info("Now running the benchmark") + + ctx, cancel := context.WithCancel(context.Background()) + + // Timer to end requests + timer = time.AfterFunc(time.Second*time.Duration(cfg.Duration), func() { + done <- syscall.SIGINT + }) + + // Statistics channel + statsChan := make(chan stats, cfg.Connections) + + // Run the requests. + for i := 0; i < cfg.Connections; i++ { + wg.Add(1) + + go connection(ctx, &wg, cli, logger, msg, opts, i, statsChan) + } + + // Blocks until timer/signal happened + select { + case <-done: + timer.Stop() + // stops requesting + cancel() + + // Wait for all goroutines to exit properly. + wg.Wait() + case <-quit: + timer.Stop() + os.Exit(0) + } + + // Calculate stats + mStats := stats{} + + for i := 0; i < cfg.Connections; i++ { + cStat := <-statsChan + + mStats.Ok += cStat.Ok + mStats.Error += cStat.Error + } + + logger.Info("Summary", + "bypass_registry", cfg.BypassRegistry, + "connections", cfg.Connections, + "duration", cfg.Duration, + "timeout", cfg.Timeout, + "threads", cfg.Threads, + "transport", cfg.Transport, + "package_size", cfg.PackageSize, + "content_type", cfg.ContentType, + "reqsOk", mStats.Ok, + "reqsError", mStats.Error, + ) + + return nil +} + +func main() { + var ( + serviceName = types.ServiceName("benchmarks.rps.client") + serviceVersion = types.ServiceVersion("v0.0.1") + ) + + if _, err := run(serviceName, serviceVersion, bench); err != nil { + log.Error("While running", err) + os.Exit(1) + } +} diff --git a/benchmarks/rps/cmd/orb-rps-client/wire.go b/benchmarks/rps/cmd/orb-rps-client/wire.go new file mode 100644 index 00000000..7b3c510b --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-client/wire.go @@ -0,0 +1,123 @@ +//go:build wireinject +// +build wireinject + +// The build tag makes sure the stub is not built in the final build. +package main + +import ( + "context" + "net/url" + "os" + "os/signal" + "syscall" + + "github.com/go-orb/go-orb/client" + "github.com/go-orb/go-orb/config" + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/registry" + "github.com/go-orb/go-orb/types" + + "github.com/google/wire" +) + +// provideConfigData reads the config from cli and returns it. +func provideConfigData( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, +) (types.ConfigData, error) { + u, err := url.Parse("cli://urfave") + if err != nil { + return nil, err + } + + cfgSections := types.SplitServiceName(serviceName) + + data, err := config.Read([]*url.URL{u}, cfgSections) + + return data, err +} + +// provideComponents creates a slice of components out of the arguments. +func provideComponents( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, + cfgData types.ConfigData, + logger log.Logger, + reg registry.Type, + client client.Type, +) ([]types.Component, error) { + components := []types.Component{} + components = append(components, logger) + components = append(components, reg) + + return components, nil +} + +type wireRunResult string + +type wireRunCallback func( + serviceName types.ServiceName, + configs types.ConfigData, + logger log.Logger, + cli client.Type, +) error + +func wireRun( + serviceName types.ServiceName, + components []types.Component, + configs types.ConfigData, + logger log.Logger, + cli client.Type, + cb wireRunCallback, +) (wireRunResult, error) { + // + // Orb start + for _, c := range components { + err := c.Start() + if err != nil { + log.Error("Failed to start", err, "component", c.Type()) + os.Exit(1) + } + } + + done := make(chan os.Signal, 1) + signal.Notify(done, syscall.SIGINT, syscall.SIGTERM) + + // + // Actual code + runErr := cb(serviceName, configs, logger, cli) + + // + // Orb shutdown. + ctx := context.Background() + + for k := range components { + c := components[len(components)-1-k] + + err := c.Stop(ctx) + if err != nil { + log.Error("Failed to stop", err, "component", c.Type()) + } + } + + return "", runErr +} + +// newComponents combines everything above and returns a slice of components. +func run( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, + cb wireRunCallback, +) (wireRunResult, error) { + panic(wire.Build( + provideConfigData, + wire.Value([]log.Option{}), + log.ProvideLogger, + wire.Value([]registry.Option{}), + registry.ProvideRegistry, + wire.Value([]client.Option{}), + client.ProvideClient, + provideComponents, + wireRun, + )) +} diff --git a/benchmarks/rps/cmd/orb-rps-client/wire_gen.go b/benchmarks/rps/cmd/orb-rps-client/wire_gen.go new file mode 100644 index 00000000..8548c087 --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-client/wire_gen.go @@ -0,0 +1,161 @@ +// Code generated by Wire. DO NOT EDIT. + +//go:generate go run -mod=mod github.com/google/wire/cmd/wire +//go:build !wireinject +// +build !wireinject + +package main + +import ( + "context" + "github.com/go-orb/go-orb/client" + "github.com/go-orb/go-orb/config" + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/registry" + "github.com/go-orb/go-orb/types" + "net/url" + "os" + "os/signal" + "syscall" +) + +import ( + _ "github.com/go-orb/plugins/client/middleware/log" + _ "github.com/go-orb/plugins/client/orb" + _ "github.com/go-orb/plugins/client/orb_transport/drpc" + _ "github.com/go-orb/plugins/client/orb_transport/grpc" + _ "github.com/go-orb/plugins/client/orb_transport/h2c" + _ "github.com/go-orb/plugins/client/orb_transport/hertzh2c" + _ "github.com/go-orb/plugins/client/orb_transport/hertzhttp" + _ "github.com/go-orb/plugins/client/orb_transport/http" + _ "github.com/go-orb/plugins/client/orb_transport/http3" + _ "github.com/go-orb/plugins/client/orb_transport/https" + _ "github.com/go-orb/plugins/codecs/jsonpb" + _ "github.com/go-orb/plugins/codecs/proto" + _ "github.com/go-orb/plugins/codecs/yaml" + _ "github.com/go-orb/plugins/config/source/cli/urfave" + _ "github.com/go-orb/plugins/config/source/file" + _ "github.com/go-orb/plugins/log/slog" + _ "github.com/go-orb/plugins/registry/consul" + _ "github.com/go-orb/plugins/registry/mdns" +) + +// Injectors from wire.go: + +// newComponents combines everything above and returns a slice of components. +func run(serviceName types.ServiceName, serviceVersion types.ServiceVersion, cb wireRunCallback) (wireRunResult, error) { + configData, err := provideConfigData(serviceName, serviceVersion) + if err != nil { + return "", err + } + v := _wireValue + logger, err := log.ProvideLogger(serviceName, configData, v...) + if err != nil { + return "", err + } + v2 := _wireValue2 + registryType, err := registry.ProvideRegistry(serviceName, serviceVersion, configData, logger, v2...) + if err != nil { + return "", err + } + v3 := _wireValue3 + clientType, err := client.ProvideClient(serviceName, configData, logger, registryType, v3...) + if err != nil { + return "", err + } + v4, err := provideComponents(serviceName, serviceVersion, configData, logger, registryType, clientType) + if err != nil { + return "", err + } + mainWireRunResult, err := wireRun(serviceName, v4, configData, logger, clientType, cb) + if err != nil { + return "", err + } + return mainWireRunResult, nil +} + +var ( + _wireValue = []log.Option{} + _wireValue2 = []registry.Option{} + _wireValue3 = []client.Option{} +) + +// wire.go: + +// provideConfigData reads the config from cli and returns it. +func provideConfigData( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, +) (types.ConfigData, error) { + u, err := url.Parse("cli://urfave") + if err != nil { + return nil, err + } + + cfgSections := types.SplitServiceName(serviceName) + + data, err := config.Read([]*url.URL{u}, cfgSections) + + return data, err +} + +// provideComponents creates a slice of components out of the arguments. +func provideComponents( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, + cfgData types.ConfigData, + logger log.Logger, + reg registry.Type, client2 client.Type, + +) ([]types.Component, error) { + components := []types.Component{} + components = append(components, logger) + components = append(components, reg) + + return components, nil +} + +type wireRunResult string + +type wireRunCallback func( + serviceName types.ServiceName, + configs types.ConfigData, + logger log.Logger, + cli client.Type, +) error + +func wireRun( + serviceName types.ServiceName, + components []types.Component, + configs types.ConfigData, + logger log.Logger, + cli client.Type, + cb wireRunCallback, +) (wireRunResult, error) { + + for _, c := range components { + err := c.Start() + if err != nil { + log.Error("Failed to start", err, "component", c.Type()) + os.Exit(1) + } + } + + done := make(chan os.Signal, 1) + signal.Notify(done, syscall.SIGINT, syscall.SIGTERM) + + runErr := cb(serviceName, configs, logger, cli) + + ctx := context.Background() + + for k := range components { + c := components[len(components)-1-k] + + err := c.Stop(ctx) + if err != nil { + log.Error("Failed to stop", err, "component", c.Type()) + } + } + + return "", runErr +} diff --git a/benchmarks/rps/cmd/orb-rps-server/main.go b/benchmarks/rps/cmd/orb-rps-server/main.go new file mode 100644 index 00000000..a6ce4779 --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-server/main.go @@ -0,0 +1,62 @@ +// Package main contains a server for running tests on. +package main + +import ( + "context" + "os" + "os/signal" + "syscall" + + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/types" + + _ "github.com/go-orb/plugins/codecs/jsonpb" + _ "github.com/go-orb/plugins/codecs/proto" + _ "github.com/go-orb/plugins/codecs/yaml" + _ "github.com/go-orb/plugins/config/source/cli/urfave" + _ "github.com/go-orb/plugins/config/source/file" + _ "github.com/go-orb/plugins/log/lumberjack" + _ "github.com/go-orb/plugins/log/slog" + _ "github.com/go-orb/plugins/registry/consul" + _ "github.com/go-orb/plugins/registry/mdns" + _ "github.com/go-orb/plugins/server/http/router/chi" +) + +func main() { + var ( + serviceName = types.ServiceName("benchmarks.rps.server") + serviceVersion = types.ServiceVersion("v0.0.1") + ) + + components, err := newComponents(serviceName, serviceVersion) + if err != nil { + log.Error("while creating components", err) + os.Exit(1) + } + + for _, c := range components { + err := c.Start() + if err != nil { + log.Error("Failed to start", err, "component", c.Type()) + os.Exit(1) + } + } + + done := make(chan os.Signal, 1) + signal.Notify(done, syscall.SIGINT, syscall.SIGTERM) + + // Blocks until we get a sigint/sigterm + <-done + + // Shutdown. + ctx := context.Background() + + for k := range components { + c := components[len(components)-1-k] + + err := c.Stop(ctx) + if err != nil { + log.Error("Failed to stop", err, "component", c.Type()) + } + } +} diff --git a/benchmarks/rps/cmd/orb-rps-server/wire.go b/benchmarks/rps/cmd/orb-rps-server/wire.go new file mode 100644 index 00000000..274f8bcb --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-server/wire.go @@ -0,0 +1,141 @@ +//go:build wireinject +// +build wireinject + +// The build tag makes sure the stub is not built in the final build. +package main + +import ( + "fmt" + "net/url" + + "github.com/go-orb/go-orb/config" + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/registry" + "github.com/go-orb/go-orb/server" + "github.com/go-orb/go-orb/types" + echohandler "github.com/go-orb/plugins/benchmarks/rps/handler/echo" + echopb "github.com/go-orb/plugins/benchmarks/rps/proto/echo" + + "github.com/go-orb/plugins/server/drpc" + mgrpc "github.com/go-orb/plugins/server/grpc" + mhertz "github.com/go-orb/plugins/server/hertz" + mhttp "github.com/go-orb/plugins/server/http" + + "github.com/google/wire" + "github.com/hashicorp/consul/sdk/freeport" +) + +// provideConfigData reads the config from cli and returns it. +func provideConfigData( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, +) (types.ConfigData, error) { + u, err := url.Parse("cli://urfave") + if err != nil { + return nil, err + } + + cfgSections := types.SplitServiceName(serviceName) + + data, err := config.Read([]*url.URL{u}, cfgSections) + + return data, err +} + +// provideServerOpts provides options for the go-orb server. +// TODO(jochumdev): We should simplify server opts. +func provideServerOpts() ([]server.Option, error) { + // Get some free ports + ports, err := freeport.Take(8) + if err != nil { + return nil, err + } + + // Our lonely handler + hInstance := new(echohandler.Handler) + + return []server.Option{ + mgrpc.WithEntrypoint( + mgrpc.WithName("grpc"), + mgrpc.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[0])), + mgrpc.WithInsecure(true), + mgrpc.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhertz.WithEntrypoint( + mhertz.WithName("hertzhttp"), + mhertz.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[1])), + mhertz.WithInsecure(), + mhertz.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhertz.WithEntrypoint( + mhertz.WithName("hertzh2c"), + mhertz.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[2])), + mhertz.WithInsecure(), + mhertz.WithAllowH2C(), + mhertz.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhttp.WithEntrypoint( + mhttp.WithName("http"), + mhttp.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[3])), + mhttp.WithInsecure(), + mhttp.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhttp.WithEntrypoint( + mhttp.WithName("h2c"), + mhttp.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[4])), + mhttp.WithInsecure(), + mhttp.WithAllowH2C(), + mhttp.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhttp.WithEntrypoint( + mhttp.WithName("http3"), + mhttp.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[5])), + mhttp.WithHTTP3(), + mhttp.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + mhttp.WithEntrypoint( + mhttp.WithName("https"), + mhttp.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[6])), + mhttp.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + drpc.WithEntrypoint( + drpc.WithName("dprc"), + drpc.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[7])), + drpc.WithRegistration("Streams", echopb.RegisterEchoService(hInstance)), + ), + }, nil +} + +// provideComponents creates a slice of components out of the arguments. +func provideComponents( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, + cfgData types.ConfigData, + logger log.Logger, + reg registry.Type, + srv server.Server, +) ([]types.Component, error) { + components := []types.Component{} + components = append(components, logger) + components = append(components, reg) + components = append(components, &srv) + + return components, nil +} + +// newComponents combines everything above and returns a slice of components. +func newComponents( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, +) ([]types.Component, error) { + panic(wire.Build( + provideConfigData, + wire.Value([]log.Option{}), + log.ProvideLogger, + wire.Value([]registry.Option{}), + registry.ProvideRegistry, + provideServerOpts, + server.ProvideServer, + provideComponents, + )) +} diff --git a/benchmarks/rps/cmd/orb-rps-server/wire_gen.go b/benchmarks/rps/cmd/orb-rps-server/wire_gen.go new file mode 100644 index 00000000..be72082b --- /dev/null +++ b/benchmarks/rps/cmd/orb-rps-server/wire_gen.go @@ -0,0 +1,125 @@ +// Code generated by Wire. DO NOT EDIT. + +//go:generate go run -mod=mod github.com/google/wire/cmd/wire +//go:build !wireinject +// +build !wireinject + +package main + +import ( + "fmt" + "github.com/go-orb/go-orb/config" + "github.com/go-orb/go-orb/log" + "github.com/go-orb/go-orb/registry" + "github.com/go-orb/go-orb/server" + "github.com/go-orb/go-orb/types" + "github.com/go-orb/plugins/benchmarks/rps/handler/echo" + echo2 "github.com/go-orb/plugins/benchmarks/rps/proto/echo" + "github.com/go-orb/plugins/server/drpc" + "github.com/go-orb/plugins/server/grpc" + "github.com/go-orb/plugins/server/hertz" + "github.com/go-orb/plugins/server/http" + "github.com/hashicorp/consul/sdk/freeport" + "net/url" +) + +import ( + _ "github.com/go-orb/plugins/codecs/jsonpb" + _ "github.com/go-orb/plugins/codecs/proto" + _ "github.com/go-orb/plugins/codecs/yaml" + _ "github.com/go-orb/plugins/config/source/cli/urfave" + _ "github.com/go-orb/plugins/config/source/file" + _ "github.com/go-orb/plugins/log/lumberjack" + _ "github.com/go-orb/plugins/log/slog" + _ "github.com/go-orb/plugins/registry/consul" + _ "github.com/go-orb/plugins/registry/mdns" + _ "github.com/go-orb/plugins/server/http/router/chi" +) + +// Injectors from wire.go: + +// newComponents combines everything above and returns a slice of components. +func newComponents(serviceName types.ServiceName, serviceVersion types.ServiceVersion) ([]types.Component, error) { + configData, err := provideConfigData(serviceName, serviceVersion) + if err != nil { + return nil, err + } + v := _wireValue + logger, err := log.ProvideLogger(serviceName, configData, v...) + if err != nil { + return nil, err + } + v2 := _wireValue2 + registryType, err := registry.ProvideRegistry(serviceName, serviceVersion, configData, logger, v2...) + if err != nil { + return nil, err + } + v3, err := provideServerOpts() + if err != nil { + return nil, err + } + serverServer, err := server.ProvideServer(serviceName, configData, logger, registryType, v3...) + if err != nil { + return nil, err + } + v4, err := provideComponents(serviceName, serviceVersion, configData, logger, registryType, serverServer) + if err != nil { + return nil, err + } + return v4, nil +} + +var ( + _wireValue = []log.Option{} + _wireValue2 = []registry.Option{} +) + +// wire.go: + +// provideConfigData reads the config from cli and returns it. +func provideConfigData( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, +) (types.ConfigData, error) { + u, err := url.Parse("cli://urfave") + if err != nil { + return nil, err + } + + cfgSections := types.SplitServiceName(serviceName) + + data, err := config.Read([]*url.URL{u}, cfgSections) + + return data, err +} + +// provideServerOpts provides options for the go-orb server. +// TODO(jochumdev): We should simplify server opts. +func provideServerOpts() ([]server.Option, error) { + + ports, err := freeport.Take(8) + if err != nil { + return nil, err + } + + hInstance := new(echo.Handler) + + return []server.Option{grpc.WithEntrypoint(grpc.WithName("grpc"), grpc.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[0])), grpc.WithInsecure(true), grpc.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), hertz.WithEntrypoint(hertz.WithName("hertzhttp"), hertz.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[1])), hertz.WithInsecure(), hertz.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), hertz.WithEntrypoint(hertz.WithName("hertzh2c"), hertz.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[2])), hertz.WithInsecure(), hertz.WithAllowH2C(), hertz.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), http.WithEntrypoint(http.WithName("http"), http.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[3])), http.WithInsecure(), http.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), http.WithEntrypoint(http.WithName("h2c"), http.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[4])), http.WithInsecure(), http.WithAllowH2C(), http.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), http.WithEntrypoint(http.WithName("http3"), http.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[5])), http.WithHTTP3(), http.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), http.WithEntrypoint(http.WithName("https"), http.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[6])), http.WithRegistration("Streams", echo2.RegisterEchoService(hInstance))), drpc.WithEntrypoint(drpc.WithName("dprc"), drpc.WithAddress(fmt.Sprintf("127.0.0.1:%d", ports[7])), drpc.WithRegistration("Streams", echo2.RegisterEchoService(hInstance)))}, nil +} + +// provideComponents creates a slice of components out of the arguments. +func provideComponents( + serviceName types.ServiceName, + serviceVersion types.ServiceVersion, + cfgData types.ConfigData, + logger log.Logger, + reg registry.Type, + srv server.Server, +) ([]types.Component, error) { + components := []types.Component{} + components = append(components, logger) + components = append(components, reg) + components = append(components, &srv) + + return components, nil +}