Create Distributed, Scalable, Durable, and Highly Available Software— With Cadence

version: '3'
services:
  mysql:
    image: mysql:5.7
    ports:
      - "3306:3306"
    environment:
      - "MYSQL_ROOT_PASSWORD=root"
    volumes:
      - ./data/mysql/:/var/lib/mysql
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus:/etc/prometheus
      - ./data/prometheus/:/prometheus
    user: "1000:1000"
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
    ports:
      - '9090:9090'
  cadence:
    image: ubercadence/server:0.23.2-auto-setup
    ports:
      - "8000:8000"
      - "8001:8001"
      - "8002:8002"
      - "8003:8003"
      - "7933:7933"
      - "7934:7934"
      - "7935:7935"
      - "7939:7939"
      - "7833:7833"
    environment:
      - "DB=mysql"
      - "MYSQL_USER=root"
      - "MYSQL_PWD=root"
      - "MYSQL_SEEDS=mysql"
      - "PROMETHEUS_ENDPOINT_0=0.0.0.0:8000"
      - "PROMETHEUS_ENDPOINT_1=0.0.0.0:8001"
      - "PROMETHEUS_ENDPOINT_2=0.0.0.0:8002"
      - "PROMETHEUS_ENDPOINT_3=0.0.0.0:8003"
      - "DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development.yaml"
    depends_on:
      - mysql
      - prometheus
  cadence-web:
    image: ubercadence/web:v3.29.4
    environment:
      - "CADENCE_TCHANNEL_PEERS=cadence:7933"
    ports:
      - "8088:8088"
    depends_on:
      - cadence
  grafana:
    image: grafana/grafana
    user: "1000"
    depends_on:
      - prometheus
    ports:
      - '3000:3000'

├── /app/
├── /cadence/                                    # All Related Cadence stuff goes here
│     ├── /data/                                 # 
|     |     └── /mysql/                          # The folder mounted onto the Docker to persist data for Mysql
|     |     └── /prometheus/                     # The folder mounted onto the Docker to persist data for Prometheus
│     ├── /prometheus/                           # Folder related to Prometheus
|     |     └── prometheus.yml                  # Prometheus config file
|     |     └── prometheus_config_multicluster.yml     # Prometheus config file for multi clusters
│     └── docker-compose.yml                     # Your docker-compose selected from the Cadence repo based on your persistent storage   

├── /app/
│     ├── main.go                               # the place where you will put the worker service code
│     ├── go.mod                                # the go module delaraction 
│     ├── go.sum                                # the go sum with versions 
├── /cadence/                                    # All Related Cadence stuff goes here
│     ├── /data/                                 # 
|     |     └── /mysql/                          # The folder mounted onto the Docker to persist data for Mysql
|     |     └── /prometheus/                     # The folder mounted onto the Docker to persist data for Prometheus
│     ├── /prometheus/                           # Folder related to Prometheus
|     |     └── prometheus.yml                  # Prometheus config file
|     |     └── prometheus_config_multicluster.yml     # Prometheus config file for multi clusters
│     └── docker-compose.yml                     # Your docker-compose selected from the Cadence repo based on your persistent storage  

func New(
	service workflowserviceclient.Interface,
	domain string,
	taskList string,
	options Options,
) Worker {
	return internal.NewWorker(service, domain, taskList, options)
}

package main

import (
	"fmt"

	_ "go.uber.org/cadence/.gen/go/cadence"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/worker"

	"github.com/uber-go/tally"
	"go.uber.org/yarpc"
	_ "go.uber.org/yarpc/api/transport"
	"go.uber.org/yarpc/transport/tchannel"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)

const (
	// cadenceService should always be cadence-frontend
	CadenceService = "cadence-frontend"
	// ClientName is the identifier for the service
	ClientName = "greetings-worker"
	// Domain is the domain you have registered and want to operate in
	Domain = "tavern"
	// Host is the Cadence server IP:Port
	Host = "127.0.0.1:7933"
	// TaskList is the identifier for tasks, activites and workflows
	TaskList = "greetings"
)

func main() {
	// Create the Worker service
	worker, logger, err := newWorkerServiceClient()
	if err != nil {
		panic(err)
	}

	// Start worker
	if err := worker.Start(); err != nil {
		panic(fmt.Errorf("failed to start the worker: %v", err))
	}

	logger.Info("Started Worker.", zap.String("worker", TaskList))

	// Block Forever
	select {}

}

// newWorkerServiceClient is used to initialize a new Worker service
// It will handle Connecting and configuration of the client
// Returns a Worker, the logger applied or an error
// TODO expand this function to allow more configurations, will be done later in the article.
func newWorkerServiceClient() (worker.Worker, *zap.Logger, error) {

	// Create a logger to use for the service
	logger, err := newLogger()
	if err != nil {
		return nil, nil, err
	}
	// build the most basic Options for now
	workerOptions := worker.Options{
		Logger:       logger,
		MetricsScope: tally.NewTestScope(TaskList, map[string]string{}),
	}
	// Create the connection that the worker should use
	connection, err := newCadenceConnection(ClientName)
	if err != nil {
		return nil, nil, err
	}
	//  Create the worker and return
	return worker.New(connection, Domain, TaskList, workerOptions), logger, nil
}

// newCadenceConnection is used to create a new YARPC connection to the Cadence server
// @clientName - used to identify the connection on YARPC
func newCadenceConnection(clientName string) (workflowserviceclient.Interface, error) {
	// Create a new Channel to communicate through
	// Set the service name to our Client name so we can Identify the connection
	ch, err := tchannel.NewChannelTransport(tchannel.ServiceName(ClientName))
	if err != nil {
		return nil, fmt.Errorf("failed to set up Transport channel: %v", err)
	}
	// Set up the dispatcher
	dispatcher := yarpc.NewDispatcher(yarpc.Config{
		Name: ClientName,
		Outbounds: yarpc.Outbounds{
			CadenceService: {Unary: ch.NewSingleOutbound(Host)},
		},
	})
	// Start the dispatcher to allow incomming/outgoing messages
	if err := dispatcher.Start(); err != nil {
		return nil, fmt.Errorf("failed to start dispatcher: %v", err)
	}
	// Return a new workflowserviceclient with the connection assigned
	return workflowserviceclient.New(dispatcher.ClientConfig(CadenceService)), nil
}

// newLogger will create a new logger to be used by the Worker Services
// For now use DevelopmentConfig and Info level
func newLogger() (*zap.Logger, error) {
	config := zap.NewDevelopmentConfig()

	config.Level.SetLevel(zapcore.InfoLevel)

	var err error
	logger, err := config.Build()
	if err != nil {
		return nil, fmt.Errorf("failed to build logger: %v", err)
	}

	return logger, nil
}

percy@awesome:~/development/cadence-demo/app$ go run main.go 
2022-02-27T08:08:08.411+0100    INFO    internal/internal_worker.go:826 Worker has no workflows registered, so workflow worker will not be started.     {"Domain": "tavern", "TaskList": "greetings", "WorkerID": "16765@DESKTOP-3DP516F@greetings@b2b05411-9dbc-4bf3-a241-2a949da322b6"}
2022-02-27T08:08:08.411+0100    INFO    internal/internal_worker.go:834 Started Workflow Worker {"Domain": "tavern", "TaskList": "greetings", "WorkerID": "16765@DESKTOP-3DP516F@greetings@b2b05411-9dbc-4bf3-a241-2a949da322b6"}
2022-02-27T08:08:08.412+0100    INFO    internal/internal_worker.go:838 Worker has no activities registered, so activity worker will not be started.    {"Domain": "tavern", "TaskList": "greetings", "WorkerID": "16765@DESKTOP-3DP516F@greetings@b2b05411-9dbc-4bf3-a241-2a949da322b6"}
2022-02-27T08:08:08.412+0100    INFO    app/main.go:44  Started Worker. {"worker": "greetings"}

package customer

import (
	"fmt"
	"time"
)


var (
	// Bad Solution for in mem during tutorial
	Database = NewMemoryCustomers()
)

// Customer is representation of a client in the Tavern
type Customer struct {
	Name string `json:"name"`
	// LastVisit is a timestamp of the last time this visitor came by the tavern
	LastVisit time.Time `json:"lastVisit"`
	// TimesVisited is how many times a user has visited
	TimesVisited int `json:"timesVisited"`
}

// Repository is the needed methods to be a customer repo
type Repository interface {
	Get(string) (Customer, error)
	Update(Customer) error
}

// MemoryCustomers is used to store information in Memory
type MemoryCustomers struct {
	Customers map[string]Customer
}

// NewMemoryCustomers will init a new in memory storage for customers
func NewMemoryCustomers() MemoryCustomers {
	return MemoryCustomers{
		Customers: make(map[string]Customer),
	}
}

// Get is used to fetch a customer by Name
func (mc *MemoryCustomers) Get(name string) (Customer, error) {
	if mc.Customers == nil {
		mc.Customers = make(map[string]Customer)
	}
	if cust, ok := mc.Customers[name]; ok {
		return cust, nil
	}
	return Customer{}, fmt.Errorf("no such customer: %s", name)
}

// Update will override the information about a customer in storage
func (mc *MemoryCustomers) Update(customer Customer) error {
	if mc.Customers == nil {
		mc.Customers = make(map[string]Customer)
	}

	mc.Customers[customer.Name] = customer
	return nil
}

├── /app/
│     ├── main.go                               # the place where you will put the worker service code
│     ├── go.mod                                # the go module delaraction 
│     ├── go.sum                                # the go sum with versions 
│     ├── /customer                             # Customer folder holds any code related to customer domain
|     |     └── repository.go                   # Repository holds the inMemory Cache for Customeres, Replace with a DB if wanted
│     ├── /workflows                            # workflows holds all the available workflows for the Domain
|     |     └── /greetings/                     # Workflow and Activities for Greetings
|     |          └── greetings.go  
├── /cadence/                                    # All Related Cadence stuff goes here
│     ├── /data/                                 # 
|     |     └── /mysql/                          # The folder mounted onto the Docker to persist data for Mysql
|     |     └── /prometheus/                     # The folder mounted onto the Docker to persist data for Prometheus
│     ├── /prometheus/                           # Folder related to Prometheus
|     |     └── prometheus.yml                  # Prometheus config file
|     |     └── prometheus_config_multicluster.yml     # Prometheus config file for multi clusters
│     └── docker-compose.yml                     # Your docker-compose selected from the Cadence repo based on your persistent storage 

// This function is valid, accepts a Customer in JSON format as Input, Outputs a Customer and an error
func workflowGreetings(ctx workflow.Context, visitor customer.Customer) (customer.Customer, error) 
// This could be valid, accept No input and Output a Customer and Error.
func workflowGreetings(ctx workflow.Context) (customer.Customer, error) 
// Accept string as input, output only an Error  
func workflowGreetings(ctx workflow.Context, input string) error 

// workflowGreetings is the Workflow that is used to handle new Customers in the Tavern.
// our Workflow accepts a customer as Input, and Outputs a Customer, and an Error
func workflowGreetings(ctx workflow.Context, visitor customer.Customer) (customer.Customer, error) {
	// workflow Options for HeartBeat Timeout and other Timeouts.
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
		HeartbeatTimeout:       time.Second * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)
	// Grab the Logger that is configured on the Workflow
	logger := workflow.GetLogger(ctx)
	logger.Info("greetings workflow started")

	// Execute the activityGreetings and Wait for the Response with GET
	// GET() will Block until the activitiy is Completed.
	// Get accepts input to marshal result to,
	// ExecuteActivity returns a FUTURE, so if you want async you can simply Skip .Get
	// Get takes in a interface{} as input that we can use to Scan the result into.
	err := workflow.ExecuteActivity(ctx, activityGreetings, visitor).Get(ctx, &visitor)
	if err != nil {
		logger.Error("Greetings Activity failed", zap.Error(err))
		return customer.Customer{}, err
	}

	err = workflow.ExecuteActivity(ctx, activityStoreCustomer, visitor).Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to update customer", zap.Error(err))
		return customer.Customer{}, err
	}
	// The output of the Workflow is a Visitor with filled information
	return visitor, nil
}

// activityGreetings is used to say Hello to a Customer and change their LastVisit and TimesVisisted
// The returned value will be a Customer struct filled with this information
func activityGreetings(ctx context.Context, visitor customer.Customer) (customer.Customer, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Greetings activity started")
	logger.Info("New Visitor", zap.String("customer", visitor.Name), zap.Int("visitorCount", visitorCount))
	visitorCount++

	oldCustomerInfo, _ := customer.Database.Get(visitor.Name)

	visitor.LastVisit = time.Now()
	visitor.TimesVisited = oldCustomerInfo.TimesVisited + 1
	return visitor, nil
}

// activityStoreCustomer is used to store the Customer in the configured Customer Storage.
func activityStoreCustomer(ctx context.Context, visitor customer.Customer) error {
	logger := activity.GetLogger(ctx)
	logger.Info("Store Customer activity started")
	logger.Info("Updating Customer", zap.String("customer", visitor.Name), zap.Time("lastVisit", visitor.LastVisit),
		zap.Int("timesVisited", visitor.TimesVisited))

	// Store Customer in Database (Memory Cache during this Example)
	err := customer.Database.Update(visitor)
	if err != nil {
		return err
	}
	return nil
}

func init() {
	// init will be called once the workflow file is imported
	// this will Register the workflow to the Worker service
	workflow.Register(workflowGreetings)
	// Register the activities also
	activity.Register(activityGreetings)
	activity.Register(activityStoreCustomer)
}

package main

import (
	"fmt"
  // Here we Import the Package, which will make Register Kick in
	_ "programmingpercy/cadence-tavern/workflows/greetings"

	_ "go.uber.org/cadence/.gen/go/cadence"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/worker"

	"github.com/uber-go/tally"
	"go.uber.org/yarpc"
	_ "go.uber.org/yarpc/api/transport"
	"go.uber.org/yarpc/transport/tchannel"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)

........
unchanged

package greetings

import (
	"context"
	"programmingpercy/cadence-tavern/customer"
	"time"

	"go.uber.org/cadence/activity"
	"go.uber.org/cadence/workflow"
	"go.uber.org/zap"
)

var (
	visitorCount = 0
)

func init() {
	// init will be called once the workflow file is imported
	// this will Register the workflow to the Worker service
	workflow.Register(workflowGreetings)
	// Register the activities also
	activity.Register(activityGreetings)
	activity.Register(activityStoreCustomer)
}

// workflowGreetings is the Workflow that is used to handle new Customers in the Tavern.
// our Workflow accepts a customer as Input, and Outputs a Customer, and an Error
func workflowGreetings(ctx workflow.Context, visitor customer.Customer) (customer.Customer, error) {
	// workflow Options for HeartBeat Timeout and other Timeouts.
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
		HeartbeatTimeout:       time.Second * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)
	// Grab the Logger that is configured on the Workflow
	logger := workflow.GetLogger(ctx)
	logger.Info("greetings workflow started")

	// Execute the activityGreetings and Wait for the Response with GET
	// GET() will Block until the activitiy is Completed.
	// Get accepts input to marshal result to,
	// ExecuteActivity returns a FUTURE, so if you want async you can simply Skip .Get
	// Get takes in a interface{} as input that we can use to Scan the result into.
	err := workflow.ExecuteActivity(ctx, activityGreetings, visitor).Get(ctx, &visitor)
	if err != nil {
		logger.Error("Greetings Activity failed", zap.Error(err))
		return customer.Customer{}, err
	}

	err = workflow.ExecuteActivity(ctx, activityStoreCustomer, visitor).Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to update customer", zap.Error(err))
		return customer.Customer{}, err
	}
	// The output of the Workflow is a Visitor with filled information
	return visitor, nil
}

// activityGreetings is used to say Hello to a Customer and change their LastVisit and TimesVisisted
// The returned value will be a Customer struct filled with this information
func activityGreetings(ctx context.Context, visitor customer.Customer) (customer.Customer, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Greetings activity started")
	logger.Info("New Visitor", zap.String("customer", visitor.Name), zap.Int("visitorCount", visitorCount))
	visitorCount++

	oldCustomerInfo, _ := customer.Database.Get(visitor.Name)

	visitor.LastVisit = time.Now()
	visitor.TimesVisited = oldCustomerInfo.TimesVisited + 1
	return visitor, nil
}

// activityStoreCustomer is used to store the Customer in the configured Customer Storage.
func activityStoreCustomer(ctx context.Context, visitor customer.Customer) error {
	logger := activity.GetLogger(ctx)
	logger.Info("Store Customer activity started")
	logger.Info("Updating Customer", zap.String("customer", visitor.Name), zap.Time("lastVisit", visitor.LastVisit),
		zap.Int("timesVisited", visitor.TimesVisited))

	// Store Customer in Database (Memory Cache during this Example)
	err := customer.Database.Update(visitor)
	if err != nil {
		return err
	}
	return nil
}

percy@tavern: go run main.go
2022-02-28T20:12:23.215+0100    INFO    internal/internal_worker.go:834 Started Workflow Worker {"Domain": "tavern", "TaskList": "greetings", "WorkerID": "10484@DESKTOP-3DP516F@greetings@b2156771-c3f8-4e3e-845f-375b1f58081a"}
2022-02-28T20:12:23.239+0100    INFO    internal/internal_worker.go:859 Started Activity Worker {"Domain": "tavern", "TaskList": "greetings", "WorkerID": "10484@DESKTOP-3DP516F@greetings@b2156771-c3f8-4e3e-845f-375b1f58081a"}
2022-02-28T20:12:23.239+0100    INFO    app/main.go:44  Started Worker. {"worker": "greetings"}

cadence --domain tavern workflow run --tl greetings --wt programmingpercy/cadence-tavern/workflows/greetings.workflowGreetings --et 60

Running execution:
  Workflow Id : 21e30ae8-8246-412d-a6d2-cc0782cf66d4                                   
  Run Id      : 7c3514b4-6839-43ee-9d45-d9d54c7c9a44                                   
  Type        : programmingpercy/cadence-tavern/workflows/greetings.workflowGreetings  
  Domain      : tavern                                                                 
  Task List   : greetings                                                              
  Args        :                                                                        
Progress:
  1, 2022-02-28T20:27:25+01:00, WorkflowExecutionStarted
  2, 2022-02-28T20:27:25+01:00, DecisionTaskScheduled
  3, 2022-02-28T20:27:25+01:00, DecisionTaskStarted
  4, 2022-02-28T20:27:25+01:00, DecisionTaskCompleted
  5, 2022-02-28T20:27:25+01:00, WorkflowExecutionFailed

Result:
  Run Time: 1 seconds
  Status: FAILED
  Reason: cadenceInternal:Generic
  Detail: unable to decode the workflow function input bytes with error: unable to decode argument: 0, *customer.Customer, with json error: EOF, function name: programmingpercy/cadence-tavern/workflows/greetings.workflowGreetings

cadence --domain tavern workflow run --tl greetings --wt programmingpercy/cadence-tavern/workflows/greetings.workflowGreetings --et 60 -i '{"name": "Percy"}'

Result:
  Run Time: 1 seconds
  Status: COMPLETED
  Output: {"name":"Percy","lastVisit":"2022-02-28T20:38:49.5461618+01:00","timesVisited":1}

package orders

import (
	"time"

	"go.uber.org/cadence/workflow"
	"go.uber.org/zap"
)

// Order is a simple type to represent orders made
type Order struct {
	Item  string  `json:"item"`
	Price float32 `json:"price"`
	By    string  `json:"by"`
}

func init() {
	workflow.Register(workflowOrder)
}

// workflowOrder will handle incomming Orders
func workflowOrder(ctx workflow.Context) error {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute * 60,
		StartToCloseTimeout:    time.Minute * 60,
		HeartbeatTimeout:       time.Hour * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)

	logger := workflow.GetLogger(ctx)
	logger.Info("Waiting for Orders")

	// Grab the Selector from the workflow Context,
	selector := workflow.NewSelector(ctx)
	// For ever running loop
	for {
		// Get the Signal used to identify an Event, we named our Order event into order
		signalChan := workflow.GetSignalChannel(ctx, "order")

		// We add a "Receiver" to the Selector, The receiver is a function that will trigger once a new Signal is recieved
		selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
			// Create the Order to marshal the Input into
			var order Order
			// Receive will read input data into the struct
			c.Receive(ctx, &order)

			logger.Info("Order made", zap.String("item", order.Item), zap.Float32("price", order.Price))
		})

		selector.Select(ctx)
		// TODO Add Signal for Leaving the Tavern, Close this workflow

	}

}

percy@awesome:~/development/cadence-demo$ cadence --domain tavern workflow run --tl greetings --wt programmingpercy/cadence-tavern/workflows/orders.workflowOrder --et 60
Running execution:
  Workflow Id : a49df98b-34ed-46b6-ac74-8a2d6e39db0e                            
  Run Id      : bd989409-8688-4eed-b08f-d871d035c77b                            
  Type        : programmingpercy/cadence-tavern/workflows/orders.workflowOrder  
  Domain      : tavern                                                          
  Task List   : greetings                                                       
  Args        :                                                                 
Progress:
  1, 2022-03-01T19:34:15+01:00, WorkflowExecutionStarted
  2, 2022-03-01T19:34:15+01:00, DecisionTaskScheduled
  3, 2022-03-01T19:34:15+01:00, DecisionTaskStarted
  4, 2022-03-01T19:34:15+01:00, DecisionTaskCompleted
  5, 2022-03-01T19:34:30+01:00, WorkflowExecutionSignaled
  6, 2022-03-01T19:34:30+01:00, DecisionTaskScheduled
  7, 2022-03-01T19:34:30+01:00, DecisionTaskStarted
  8, 2022-03-01T19:34:30+01:00, DecisionTaskCompleted

Result:
  Run Time: 41 seconds

cadence --domain tavern workflow signal -w 71b24355–4d9a-4460–818a-50c9058a837e -n order -i '{ "item": "Beer", "by": "Percy" }'

package orders

import (
	"time"

	"go.uber.org/cadence/workflow"
	"go.uber.org/zap"
)

// Order is a simple type to represent orders made
type Order struct {
	Item  string  `json:"item"`
	Price float32 `json:"price"`
	By    string  `json:"by"`
}

func init() {
	workflow.Register(workflowOrder)
}

// MaxSignalsAmount is how many signals we accept before restart
// Cadence recommends a production workflow to have <1000
const MaxSignalsAmount = 3

// workflowOrder will handle incomming Orders
func workflowOrder(ctx workflow.Context) error {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute * 60,
		StartToCloseTimeout:    time.Minute * 60,
		HeartbeatTimeout:       time.Hour * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)

	logger := workflow.GetLogger(ctx)
	logger.Info("Waiting for Orders")

	// restartWorkflow
	var restartWorkflow bool
	// signalCounter
	signalCount := 0

	// Grab the Selector from the workflow Context,
	selector := workflow.NewSelector(ctx)
	// For ever running loop
	for {
		// Get the Signal used to identify an Event, we named our Order event into order
		signalChan := workflow.GetSignalChannel(ctx, "order")

		// We add a "Receiver" to the Selector, The receiver is a function that will trigger once a new Signal is recieved
		selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
			// Create the Order to marshal the Input into
			var order Order
			// Receive will read input data into the struct
			c.Receive(ctx, &order)

			// increment signal counter
			signalCount++

			logger.Info("Order made", zap.String("item", order.Item), zap.Float32("price", order.Price))
		})

		if signalCount >= MaxSignalsAmount {
			// We should restart
			// Add a Default to the selector, which will make sure that this is triggered once all jobs in queue are done
			selector.AddDefault(func() {
				restartWorkflow = true
			})
		}

		selector.Select(ctx)

		// If its time to restart, return the ContinueAsNew
		if restartWorkflow {
			return workflow.NewContinueAsNewError(ctx, workflowOrder)
		}

	}

}

	// Preconfigure ChildWorkflow Options
	orderWaiterCfg := workflow.ChildWorkflowOptions{
		ExecutionStartToCloseTimeout: time.Minute * 2, // Each Order can tops take 2 min
	}

	// Grab the Selector from the workflow Context,
	selector := workflow.NewSelector(ctx)
	// For ever running loop
	for {
		// Get the Signal used to identify an Event, we named our Order event into order
		signalChan := workflow.GetSignalChannel(ctx, "order")

		// We add a "Receiver" to the Selector, The receiver is a function that will trigger once a new Signal is recieved
		selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
			// Create the Order to marshal the Input into
			var order Order
			// Receive will read input data into the struct
			c.Receive(ctx, &order)

			// increment signal counter
			signalCount++
			// Create ctx for Child flow
			orderCtx := workflow.WithChildOptions(ctx, orderWaiterCfg)
			// Trigger the child workflow
			waiter := workflow.ExecuteChildWorkflow(orderCtx, workflowProcessOrder, order)
			if err := waiter.Get(ctx, nil); err != nil {
				workflow.GetLogger(ctx).Error("Order has failed.", zap.Error(err))
			}

		})

package orders

import (
	"context"
	"errors"
	"programmingpercy/cadence-tavern/customer"
	"time"

	"go.uber.org/cadence/activity"
	"go.uber.org/cadence/workflow"
	"go.uber.org/zap"
)

// Order is a simple type to represent orders made
type Order struct {
	Item  string  `json:"item"`
	Price float32 `json:"price"`
	By    string  `json:"by"`
}

func init() {
	workflow.Register(workflowOrder)
	workflow.Register(workflowProcessOrder)

	activity.Register(activityIsCustomerLegal)
	activity.Register(activitiyFindCustomerByName)
}

// MaxSignalsAmount is how many signals we accept before restart
// Cadence recommends a production workflow to have <1000
const MaxSignalsAmount = 3

// workflowOrder will handle incomming Orders
func workflowOrder(ctx workflow.Context) error {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute * 60,
		StartToCloseTimeout:    time.Minute * 60,
		HeartbeatTimeout:       time.Hour * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)

	logger := workflow.GetLogger(ctx)
	logger.Info("Waiting for Orders")

	// restartWorkflow
	var restartWorkflow bool
	// signalCounter
	signalCount := 0

	// Preconfigure ChildWorkflow Options
	orderWaiterCfg := workflow.ChildWorkflowOptions{
		ExecutionStartToCloseTimeout: time.Minute * 2, // Each Order can tops take 2 min
	}

	// Grab the Selector from the workflow Context,
	selector := workflow.NewSelector(ctx)
	// For ever running loop
	for {
		// Get the Signal used to identify an Event, we named our Order event into order
		signalChan := workflow.GetSignalChannel(ctx, "order")

		// We add a "Receiver" to the Selector, The receiver is a function that will trigger once a new Signal is recieved
		selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
			// Create the Order to marshal the Input into
			var order Order
			// Receive will read input data into the struct
			c.Receive(ctx, &order)

			// increment signal counter
			signalCount++
			// Create ctx for Child flow
			orderCtx := workflow.WithChildOptions(ctx, orderWaiterCfg)
			// Trigger the child workflow
			waiter := workflow.ExecuteChildWorkflow(orderCtx, workflowProcessOrder, order)
			if err := waiter.Get(ctx, nil); err != nil {
				workflow.GetLogger(ctx).Error("Order has failed.", zap.Error(err))
			}

		})

		if signalCount >= MaxSignalsAmount {
			// We should restart
			// Add a Default to the selector, which will make sure that this is triggered once all jobs in queue are done
			selector.AddDefault(func() {
				restartWorkflow = true
			})
		}

		selector.Select(ctx)

		// If its time to restart, return the ContinueAsNew
		if restartWorkflow {
			return workflow.NewContinueAsNewError(ctx, workflowOrder)
		}

	}
}

// workflowProcessOrder is used to handle orders and will be ran as a CHILD
func workflowProcessOrder(ctx workflow.Context, order Order) error {

	logger := workflow.GetLogger(ctx)
	logger.Info("process order workflow started")
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
		HeartbeatTimeout:       time.Second * 20,
		// Here we will Add Retry policies etc later
	}
	// Add the Options to Context to apply configurations
	ctx = workflow.WithActivityOptions(ctx, ao)

	// Find Customer from Repo
	var cust customer.Customer
	err := workflow.ExecuteActivity(ctx, activitiyFindCustomerByName, order.By).Get(ctx, &cust)

	if err != nil {
		logger.Error("Customer is not in the Tavern", zap.Error(err))
		return err
	}

	var allowed bool
	err = workflow.ExecuteActivity(ctx, activityIsCustomerLegal, cust).Get(ctx, &allowed)
	if err != nil {
		logger.Error("Customer is not of age", zap.Error(err))
		return err
	}

	logger.Info("Order made", zap.String("item", order.Item), zap.Float32("price", order.Price))
	return nil

}

// activityFindCustomerByName is used to find the Customer is in the Tavern
func activitiyFindCustomerByName(ctx context.Context, name string) (customer.Customer, error) {
	return customer.Database.Get(name)
}

// activityIsCustomerLegal is used to check the age of the customer
func activityIsCustomerLegal(ctx context.Context, visitor customer.Customer) (bool, error) {

	if visitor.Age < 18 {
		return false, errors.New("customer is not old enough, dont serve him")
	}
	return true, nil
}

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"programmingpercy/cadence-tavern/customer"
	"programmingpercy/cadence-tavern/workflows/orders"
)

func main() {

	mux := http.NewServeMux()
	mux.HandleFunc("/greetings", GreetUser)
	mux.HandleFunc("/order", Order)

	log.Fatal(http.ListenAndServe("localhost:8080", mux))
}

func GreetUser(w http.ResponseWriter, r *http.Request) {

	// Grab user info from body
	var visitor customer.Customer

	err := json.NewDecoder(r.Body).Decode(&visitor)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	// Trigger Workflow here
	log.Print(visitor)
}

func Order(w http.ResponseWriter, r *http.Request) {
	// Grab order info from body
	var orderInfo orders.Order

	err := json.NewDecoder(r.Body).Decode(&orderInfo)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	log.Print(orderInfo)
	// Trigger Workflow here
}

const (
	cadenceClientName = "cadence-client"
	cadenceService    = "cadence-frontend"
)

const (
	// The names of the Workflows we will be using
	OrderWorkflow     = "programmingpercy/cadence-tavern/workflows/orders.WorkflowOrder"
	GreetingsWorkflow = "programmingpercy/cadence-tavern/workflows/greetings.workflowGreetings"
)

type CadenceClient struct {
	//dispatcher used to communicate
	dispatcher *yarpc.Dispatcher
	// wfClient is the workflow Client
	wfClient workflowserviceclient.Interface
	// client is the client used for cadence
	client client.Client
	// orderWorkflowID is used to remember the workflow id
	orderWorkflowID string
	// orderWorkflowRunID is the run id of the order workflow
	orderWorkflowRunID string
}

// SetupCadenceClient is used to create the client we can use
func SetupCadenceClient() (*CadenceClient, error) {
	// Create a dispatcher used to communicate with server
	dispatcher := yarpc.NewDispatcher(yarpc.Config{
		Name: cadenceClientName,
		Outbounds: yarpc.Outbounds{
			// This shouldnt be hard coded in real app
			// This is a map, so we store this communication channel on "cadence-frontend"
			cadenceService: {Unary: grpc.NewTransport().NewSingleOutbound("localhost:7833")},
		},
	})
	// Start dispatcher
	if err := dispatcher.Start(); err != nil {
		return nil, err
	}
	// Grab the Configurations from the Dispatcher based on cadenceService name
	yarpConfig := dispatcher.ClientConfig(cadenceService)
	// Build the workflowserviceClient that handles the workflows
	wfClient := workflowserviceclient.New(yarpConfig)
	// clientoptions used to control metrics etc
	opts := &client.Options{
		MetricsScope: tally.NoopScope, // Do nothing for now
	}
	// Build the Cadence Client
	cadenceClient := client.NewClient(wfClient, "tavern", opts)

	return &CadenceClient{
		dispatcher: dispatcher,
		wfClient:   wfClient,
		client:     cadenceClient,
	}, nil

}

// SetOrderWorkflowIds is used to store workflows IDS in Memory
func (cc *CadenceClient) SetOrderWorkflowIds(id, runID string) {
	cc.orderWorkflowID = id
	cc.orderWorkflowRunID = runID
}

// GreetUser is used to Welcome a new User into the tavern
func (cc *CadenceClient) GreetUser(w http.ResponseWriter, r *http.Request) {
	// Grab user info from body
	var visitor customer.Customer

	err := json.NewDecoder(r.Body).Decode(&visitor)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	// Trigger Workflow here
	log.Print(visitor)

	// Create workflow options, this is the same as the CLI, a task list, a timeout timer
	opts := client.StartWorkflowOptions{
		TaskList:                     "greetings",
		ExecutionStartToCloseTimeout: time.Second * 10,
	}

	log.Println("Starting workflow")
	// This is how you Execute a Workflow and wait for it to finish
	// This is useful if you have synchronous workflows that you want to leverage as functions
	future, err := cc.client.ExecuteWorkflow(r.Context(), opts, GreetingsWorkflow, visitor)

	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	log.Println("Get Result from  workflow")
	// Fetch result once done and marshal into
	if err := future.Get(r.Context(), &visitor); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	data, _ := json.Marshal(visitor)
	w.WriteHeader(http.StatusOK)
	w.Write(data)
}

// Order is used to send a signal to the worker
func (cc *CadenceClient) Order(w http.ResponseWriter, r *http.Request) {
	// Grab order info from body
	var orderInfo orders.Order

	err := json.NewDecoder(r.Body).Decode(&orderInfo)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	log.Print(orderInfo)
	// Send a signal to the Workflow
	// We need to provide a Workflow ID, the RUN ID of the workflow, and the Signal type
	err = cc.client.SignalWorkflow(r.Context(), cc.orderWorkflowID, cc.orderWorkflowRunID, "order", orderInfo)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	log.Println("Signalled system of order")

	w.WriteHeader(http.StatusOK)

}

package main

import (
	"context"
	"log"
	"net/http"
	"time"

	"go.uber.org/cadence/client"
)

func main() {

	rootCtx := context.Background()
	cc, err := SetupCadenceClient()
	if err != nil {
		panic(err)
	}

	// Start long running workflow
	opts := client.StartWorkflowOptions{
		TaskList:                     "greetings",
		ExecutionStartToCloseTimeout: time.Hour * 1, // Wait 1 hours, make sure you use a high enough time
		// to make sure that the workflow does not timeout before 3 singals are recieved
	}

	// We use Start here since we want to start it but not wait for it to return
	// Execution contains information about the execution such as Workflow ID etc
	// In production, make sure you check if the WOrkflows are already running to avoid  booting up multiple unless wanted
	execution, err := cc.client.StartWorkflow(rootCtx, opts, OrderWorkflow)
	if err != nil {
		panic(err)
	}

	log.Println("Workflow ID: ", execution.ID)
	// Apply Workflows IDs
	cc.SetOrderWorkflowIds(execution.ID, execution.RunID)

	mux := http.NewServeMux()
	mux.HandleFunc("/greetings", cc.GreetUser)
	mux.HandleFunc("/order", cc.Order)

	log.Fatal(http.ListenAndServe("localhost:8080", mux))
}

# Start API
go run *.go

# New Terminal
# Send CURL
percy@dev:~/development/cadence-demo/app/api$ curl localhost:8080/greetings -d '{"name": "Percy", "age": 22}'
# RESPONSE
{"name":"Percy","lastVisit":"2022-03-04T07:21:33.1540528+01:00","timesVisited":1,"age":22}

# Send Order
percy@dev:~/development/cadence-demo/app/api$ curl localhost:8080/order -d '{"by": "Percy", "item": "Beer"}'

package prometheus

import (
	"time"

	prom "github.com/m3db/prometheus_client_golang/prometheus"
	"github.com/uber-go/tally"
	"github.com/uber-go/tally/prometheus"
	"go.uber.org/zap"
)

var (
	safeCharacters = []rune{'_'}

	sanitizeOptions = tally.SanitizeOptions{
		NameCharacters: tally.ValidCharacters{
			Ranges:     tally.AlphanumericRange,
			Characters: safeCharacters,
		},
		KeyCharacters: tally.ValidCharacters{
			Ranges:     tally.AlphanumericRange,
			Characters: safeCharacters,
		},
		ValueCharacters: tally.ValidCharacters{
			Ranges:     tally.AlphanumericRange,
			Characters: safeCharacters,
		},
		ReplacementCharacter: tally.DefaultReplacementCharacter,
	}
)

// NewPrometheusReporter is used to create a new reporter that can send info to prom
// we need a zap logger inputted to make sure we get logs on error
// addr should be the IP:PORT to send metrics
func NewPrometheusReporter(addr string, logger *zap.Logger) (prometheus.Reporter, error) {
	promCfg := prometheus.Configuration{
		ListenAddress: addr,
	}

	reporter, err := promCfg.NewReporter(
		prometheus.ConfigurationOptions{
			Registry: prom.NewRegistry(),
			OnError: func(err error) {
				logger.Warn("error in prometheus reporter", zap.Error(err))
			},
		},
	)
	if err != nil {
		return nil, err
	}

	return reporter, nil
}

// NewServiceScope is used by services and prefixed Service_
func NewServiceScope(reporter prometheus.Reporter) tally.Scope {
	serviceScope, _ := tally.NewRootScope(tally.ScopeOptions{
		Prefix:          "Service_",
		Tags:            map[string]string{},
		CachedReporter:  reporter,
		Separator:       prometheus.DefaultSeparator,
		SanitizeOptions: &sanitizeOptions,
	}, 1*time.Second)

	return serviceScope
}

// NewWorkerScope is used by Workers and prefixed Worker_
func NewWorkerScope(reporter prometheus.Reporter) tally.Scope {
	serviceScope, _ := tally.NewRootScope(tally.ScopeOptions{
		Prefix:          "Worker_",
		Tags:            map[string]string{},
		CachedReporter:  reporter,
		Separator:       prometheus.DefaultSeparator,
		SanitizeOptions: &sanitizeOptions,
	}, 1*time.Second)

	return serviceScope
}

package main

import (
	"fmt"
	localprom "programmingpercy/cadence-tavern/prometheus"
	_ "programmingpercy/cadence-tavern/workflows/greetings"
	_ "programmingpercy/cadence-tavern/workflows/orders"

	_ "go.uber.org/cadence/.gen/go/cadence"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/worker"

	"go.uber.org/yarpc"
	_ "go.uber.org/yarpc/api/transport"
	"go.uber.org/yarpc/transport/tchannel"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)
..... unchanged until

// newWorkerServiceClient is used to initialize a new Worker service
// It will handle Connecting and configuration of the client
// Returns a Worker, the logger applied or an error
// TODO expand this function to allow more configurations, will be done later in the article.
func newWorkerServiceClient() (worker.Worker, *zap.Logger, error) {

	// Create a logger to use for the service
	logger, err := newLogger()
	if err != nil {
		return nil, nil, err
	}

	reporter, err := localprom.NewPrometheusReporter("127.0.0.1:9098", logger)
	if err != nil {
		return nil, nil, err
	}

	metricsScope := localprom.NewServiceScope(reporter)

	// build the most basic Options for now
	workerOptions := worker.Options{
		Logger:       logger,
		MetricsScope: metricsScope,
	}
	// Create the connection that the worker should use
	connection, err := newCadenceConnection(ClientName)
	if err != nil {
		return nil, nil, err
	}
	//  Create the worker and return
	return worker.New(connection, Domain, TaskList, workerOptions), logger, nil
}

// SetupCadenceClient is used to create the client we can use
func SetupCadenceClient() (*CadenceClient, error) {
	// Create a dispatcher used to communicate with server
	dispatcher := yarpc.NewDispatcher(yarpc.Config{
		Name: cadenceClientName,
		Outbounds: yarpc.Outbounds{
			// This shouldnt be hard coded in real app
			// This is a map, so we store this communication channel on "cadence-frontend"
			cadenceService: {Unary: grpc.NewTransport().NewSingleOutbound("localhost:7833")},
		},
	})
	// Start dispatcher
	if err := dispatcher.Start(); err != nil {
		return nil, err
	}
	// Grab the Configurations from the Dispatcher based on cadenceService name
	yarpConfig := dispatcher.ClientConfig(cadenceService)
	// Build the workflowserviceClient that handles the workflows
	wfClient := workflowserviceclient.New(yarpConfig)
	// clientoptions used to control metrics etc

	config := zap.NewDevelopmentConfig()

	config.Level.SetLevel(zapcore.InfoLevel)

	var err error
	logger, err := config.Build()
	if err != nil {
		return nil, fmt.Errorf("failed to build logger: %v", err)
	}

	// Start prom scope
	reporter, err := localprom.NewPrometheusReporter("127.0.0.1:9099", logger)
	if err != nil {
		return nil, err
	}
	// use WorkerScope
	metricsScope := localprom.NewWorkerScope(reporter)

	opts := &client.Options{
		MetricsScope: metricsScope,
	}

	// Build the Cadence Client
	cadenceClient := client.NewClient(wfClient, "tavern", opts)

	return &CadenceClient{
		dispatcher: dispatcher,
		wfClient:   wfClient,
		client:     cadenceClient,
	}, nil

}// SetupCadenceClient is used to create the client we can use
func SetupCadenceClient() (*CadenceClient, error) {
	// Create a dispatcher used to communicate with server
	dispatcher := yarpc.NewDispatcher(yarpc.Config{
		Name: cadenceClientName,
		Outbounds: yarpc.Outbounds{
			// This shouldnt be hard coded in real app
			// This is a map, so we store this communication channel on "cadence-frontend"
			cadenceService: {Unary: grpc.NewTransport().NewSingleOutbound("localhost:7833")},
		},
	})
	// Start dispatcher
	if err := dispatcher.Start(); err != nil {
		return nil, err
	}
	// Grab the Configurations from the Dispatcher based on cadenceService name
	yarpConfig := dispatcher.ClientConfig(cadenceService)
	// Build the workflowserviceClient that handles the workflows
	wfClient := workflowserviceclient.New(yarpConfig)
	// clientoptions used to control metrics etc

	config := zap.NewDevelopmentConfig()

	config.Level.SetLevel(zapcore.InfoLevel)

	var err error
	logger, err := config.Build()
	if err != nil {
		return nil, fmt.Errorf("failed to build logger: %v", err)
	}

	// Start prom scope
	reporter, err := localprom.NewPrometheusReporter("127.0.0.1:9099", logger)
	if err != nil {
		return nil, err
	}
	// use WorkerScope
	metricsScope := localprom.NewWorkerScope(reporter)

	opts := &client.Options{
		MetricsScope: metricsScope,
	}

	// Build the Cadence Client
	cadenceClient := client.NewClient(wfClient, "tavern", opts)

	return &CadenceClient{
		dispatcher: dispatcher,
		wfClient:   wfClient,
		client:     cadenceClient,
	}, nil

}

global:
  scrape_interval: 5s
  external_labels:
    monitor: 'cadence-monitor'
scrape_configs:
  - job_name: 'prometheus'
    static_configs:
      - targets: # addresses to scrape
          - 'localhost:9090'
          - 'cadence:8000'
          - 'cadence:8001'
          - 'cadence:8002'
          - 'cadence:8003'
          - 'host.docker.internal:9098'
          - 'host.docker.internal:9099'