Make Canary Runnable (cadence-workflow#2959)

Makefile

@@ -133,6 +133,10 @@ cadence-server: $(ALL_SRC)
 	@echo "compiling cadence-server with OS: $(GOOS), ARCH: $(GOARCH)"
 	go build -ldflags '$(GO_BUILD_LDFLAGS)' -i -o cadence-server cmd/server/main.go
 
+cadence-canary: $(ALL_SRC)
+	@echo "compiling cadence-canary with OS: $(GOOS), ARCH: $(GOARCH)"
+	go build -i -o cadence-canary cmd/canary/main.go
+
 go-generate:
 	GO111MODULE=off go get -u github.com/myitcv/gobin
 	GOOS= GOARCH= gobin -mod=readonly github.com/golang/mock/mockgen
@@ -159,7 +163,7 @@ fmt:
 	@echo "running goimports"
 	@goimports -local "github.com/uber/cadence" -w $(ALL_SRC)
 
-bins_nothrift: fmt lint copyright cadence-cassandra-tool cadence-sql-tool cadence cadence-server
+bins_nothrift: fmt lint copyright cadence-cassandra-tool cadence-sql-tool cadence cadence-server cadence-canary
 
 bins: thriftc bins_nothrift
 
@@ -303,3 +307,6 @@ start-cdc-standby: bins
 
 start-cdc-other: bins
 	./cadence-server --zone other start
+
+start-canary: bins
+	./cadence-canary start

canary/canary.go

@@ -144,15 +144,16 @@ func (c *canaryImpl) createDomain() error {
 	name := c.canaryDomain
 	desc := "Domain for running cadence canary workflows"
 	owner := "cadence-canary"
-	return c.canaryClient.createDomain(name, desc, owner, nil)
+	archivalStatus := shared.ArchivalStatusDisabled
+	return c.canaryClient.createDomain(name, desc, owner, &archivalStatus)
 }
 
 func (c *canaryImpl) createArchivalDomain() error {
 	name := archivalDomain
 	desc := "Domain used by cadence canary workflows to verify archival"
 	owner := "cadence-canary"
 	archivalStatus := shared.ArchivalStatusEnabled
-	return c.canaryClient.createDomain(name, desc, owner, &archivalStatus)
+	return c.archivalClient.createDomain(name, desc, owner, &archivalStatus)
 }
 
 // Override worker options to create large number of pollers to improve the chances of activities getting sync matched

canary/config.go

@@ -25,39 +25,57 @@ import (
 
 	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
-	"go.uber.org/config"
 	"go.uber.org/zap"
+
+	"github.com/uber/cadence/common/service/config"
+)
+
+const (
+	// EnvKeyRoot the environment variable key for runtime root dir
+	EnvKeyRoot = "CADENCE_CANARY_ROOT"
+	// EnvKeyConfigDir the environment variable key for config dir
+	EnvKeyConfigDir = "CADENCE_CANARY_CONFIG_DIR"
+	// EnvKeyEnvironment is the environment variable key for environment
+	EnvKeyEnvironment = "CADENCE_CANARY_ENVIRONMENT"
+	// EnvKeyAvailabilityZone is the environment variable key for AZ
+	EnvKeyAvailabilityZone = "CADENCE_CANARY_AVAILABILITY_ZONE"
+)
+
+const (
+	// CadenceLocalHostPort is the default address for cadence frontend service
+	CadenceLocalHostPort = "127.0.0.1:7933"
+	// CadenceServiceName is the default service name for cadence frontend
+	CadenceServiceName = "cadence-frontend"
+	// CanaryServiceName is the default service name for cadence canary
+	CanaryServiceName = "cadence-canary"
 )
 
 type (
 	// Config contains the configurable yaml
 	// properties for the canary runtime
 	Config struct {
+		Canary  Canary         `yaml:"canary"`
+		Cadence Cadence        `yaml:"cadence"`
+		Log     config.Logger  `yaml:"log"`
+		Metrics config.Metrics `yaml:"metrics"`
+	}
+
+	// Canary contains the configuration for canary tests
+	Canary struct {
 		Domains  []string `yaml:"domains"`
 		Excludes []string `yaml:"excludes"`
 	}
-)
-
-const (
-	// ConfigurationKey is the config YAML key for the canary module
-	ConfigurationKey = "canary"
-)
 
-// Init validates and initializes the config
-func newCanaryConfig(provider config.Provider) (*Config, error) {
-	raw := provider.Get(ConfigurationKey)
-	var cfg Config
-	if err := raw.Populate(&cfg); err != nil {
-		return nil, fmt.Errorf("failed to load canary configuration with error: %v", err)
-	}
-	if err := cfg.validate(); err != nil {
-		return nil, err
+	// Cadence contains the configuration for cadence service
+	Cadence struct {
+		ServiceName     string `yaml:"service"`
+		HostNameAndPort string `yaml:"host"`
 	}
-	return &cfg, nil
-}
+)
 
-func (c *Config) validate() error {
-	if len(c.Domains) == 0 {
+// Validate validates canary configration
+func (c *Config) Validate() error {
+	if len(c.Canary.Domains) == 0 {
 		return fmt.Errorf("missing value for domains property")
 	}
 	return nil
@@ -66,16 +84,18 @@ func (c *Config) validate() error {
 // RuntimeContext contains all the context
 // information needed to run the canary
 type RuntimeContext struct {
-	Env     string
 	logger  *zap.Logger
 	metrics tally.Scope
 	service workflowserviceclient.Interface
 }
 
 // NewRuntimeContext builds a runtime context from the config
-func newRuntimeContext(env string, logger *zap.Logger, scope tally.Scope, service workflowserviceclient.Interface) *RuntimeContext {
+func NewRuntimeContext(
+	logger *zap.Logger,
+	scope tally.Scope,
+	service workflowserviceclient.Interface,
+) *RuntimeContext {
 	return &RuntimeContext{
-		Env:     env,
 		logger:  logger,
 		metrics: scope,
 		service: service,

canary/runner.go

@@ -21,23 +21,66 @@
 package canary
 
 import (
+	"fmt"
 	"sync"
 
+	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
+	"go.uber.org/yarpc"
+	"go.uber.org/yarpc/transport/tchannel"
 	"go.uber.org/zap"
+
+	"github.com/uber/cadence/common/log/loggerimpl"
 )
 
 type canaryRunner struct {
 	*RuntimeContext
-	config *Config
+	config *Canary
 }
 
-// New creates and returns a runnable which spins
+// NewCanaryRunner creates and returns a runnable which spins
 // up a set of canaries based on supplied config
-func newCanaryRunner(cfg *Config, runtime *RuntimeContext) Runnable {
-	return &canaryRunner{
-		RuntimeContext: runtime,
-		config:         cfg,
+func NewCanaryRunner(cfg *Config) (Runnable, error) {
+	logger := cfg.Log.NewZapLogger()
+
+	metricsScope := cfg.Metrics.NewScope(loggerimpl.NewLogger(logger))
+
+	if cfg.Cadence.ServiceName == "" {
+		cfg.Cadence.ServiceName = CadenceServiceName
+	}
+
+	if cfg.Cadence.HostNameAndPort == "" {
+		cfg.Cadence.HostNameAndPort = CadenceLocalHostPort
+	}
+
+	ch, err := tchannel.NewChannelTransport(
+		tchannel.ServiceName(CanaryServiceName),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transport channel: %v", err)
+	}
+
+	dispatcher := yarpc.NewDispatcher(yarpc.Config{
+		Name: CanaryServiceName,
+		Outbounds: yarpc.Outbounds{
+			cfg.Cadence.ServiceName: {Unary: ch.NewSingleOutbound(cfg.Cadence.HostNameAndPort)},
+		},
+	})
+
+	if err := dispatcher.Start(); err != nil {
+		dispatcher.Stop()
+		return nil, fmt.Errorf("failed to create outbound transport channel: %v", err)
 	}
+
+	runtimeContext := NewRuntimeContext(
+		logger,
+		metricsScope,
+		workflowserviceclient.New(dispatcher.ClientConfig(cfg.Cadence.ServiceName)),
+	)
+
+	return &canaryRunner{
+		RuntimeContext: runtimeContext,
+		config:         &cfg.Canary,
+	}, nil
 }
 
 // Run runs the canaries

cmd/canary/main.go

@@ -0,0 +1,135 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package main
+
+import (
+	"log"
+	"os"
+	"path"
+	"strings"
+
+	"github.com/urfave/cli"
+
+	"github.com/uber/cadence/canary"
+	"github.com/uber/cadence/common/service/config"
+)
+
+func startHandler(c *cli.Context) {
+	env := getEnvironment(c)
+	zone := getZone(c)
+	configDir := getConfigDir(c)
+
+	log.Printf("Loading config; env=%v,zone=%v,configDir=%v\n", env, zone, configDir)
+
+	var cfg canary.Config
+	if err := config.Load(env, configDir, zone, &cfg); err != nil {
+		log.Fatal("Failed to load config file: ", err)
+	}
+
+	if err := cfg.Validate(); err != nil {
+		log.Fatal("Invalid config: ", err)
+	}
+
+	canary, err := canary.NewCanaryRunner(&cfg)
+	if err != nil {
+		log.Fatal("Failed to initialize canary: ", err)
+	}
+
+	if err := canary.Run(); err != nil {
+		log.Fatal("Failed to run canary: ", err)
+	}
+}
+
+func getRootDir(c *cli.Context) string {
+	rootDir := c.GlobalString("root")
+	if len(rootDir) == 0 {
+		var err error
+		if rootDir, err = os.Getwd(); err != nil {
+			rootDir = "."
+		}
+	}
+	return rootDir
+}
+
+func getConfigDir(c *cli.Context) string {
+	rootDir := getRootDir(c)
+	configDir := c.GlobalString("config")
+	return path.Join(rootDir, configDir)
+}
+
+func getEnvironment(c *cli.Context) string {
+	return strings.TrimSpace(c.GlobalString("env"))
+}
+
+func getZone(c *cli.Context) string {
+	return strings.TrimSpace(c.GlobalString("zone"))
+}
+
+func buildCLI() *cli.App {
+	app := cli.NewApp()
+	app.Name = "cadence-canary"
+	app.Usage = "Cadence canary"
+	app.Version = "0.0.1"
+
+	app.Flags = []cli.Flag{
+		cli.StringFlag{
+			Name:   "root, r",
+			Value:  ".",
+			Usage:  "root directory of execution environment",
+			EnvVar: canary.EnvKeyRoot,
+		},
+		cli.StringFlag{
+			Name:   "config, c",
+			Value:  "config/canary",
+			Usage:  "config dir path relative to root",
+			EnvVar: canary.EnvKeyConfigDir,
+		},
+		cli.StringFlag{
+			Name:   "env, e",
+			Value:  "development",
+			Usage:  "runtime environment",
+			EnvVar: canary.EnvKeyEnvironment,
+		},
+		cli.StringFlag{
+			Name:   "zone, az",
+			Value:  "",
+			Usage:  "availability zone",
+			EnvVar: canary.EnvKeyAvailabilityZone,
+		},
+	}
+
+	app.Commands = []cli.Command{
+		{
+			Name:  "start",
+			Usage: "start cadence canary",
+			Action: func(c *cli.Context) {
+				startHandler(c)
+			},
+		},
+	}
+
+	return app
+}
+
+func main() {
+	app := buildCLI()
+	app.Run(os.Args)
+}

config/canary/base.yaml

@@ -0,0 +1,3 @@
+log:
+  stdout: true
+  level: info

config/canary/development.yaml

@@ -0,0 +1,7 @@
+canary:
+  domains: ["cadence-canary"]
+  excludes: ["workflow.searchAttributes", "workflow.batch"]
+
+cadence:
+  service: "cadence-frontend"
+  host: "127.0.0.1:7933"