hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove canary. (#477)

Browse Source
This commit is contained in:
Alex Shtin
2020-07-02 11:45:12 -07:00
committed by GitHub
parent 4d09d84fc4
commit de90455153
30 changed files with 1 additions and 3959 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
-3
View File
@@ -1,6 +1,3 @@
build/
temporal-canary
vendor/
.gobincache/
.idea/
.buildkite/
Makefile
+1 -9
View File
@@ -6,7 +6,7 @@ install: update-tools proto bins
all: proto check bins test
# Rebuild binaries.
bins: clean-bins proto temporal-server tctl temporal-cassandra-tool temporal-sql-tool temporal-canary
bins: clean-bins proto temporal-server tctl temporal-cassandra-tool temporal-sql-tool
# Delete all build artefacts.
clean: clean-bins clean-proto clean-test-results
@@ -170,7 +170,6 @@ clean-bins:
@rm -f temporal-server
@rm -f temporal-cassandra-tool
@rm -f temporal-sql-tool
@rm -f temporal-canary
temporal-server: proto
@printf $(COLOR) "Build temporal-server with OS: $(GOOS), ARCH: $(GOARCH)..."
@@ -188,10 +187,6 @@ temporal-sql-tool: proto
@printf $(COLOR) "Build temporal-sql-tool with OS: $(GOOS), ARCH: $(GOARCH)..."
go build -o temporal-sql-tool cmd/tools/sql/main.go
temporal-canary: proto
@printf $(COLOR) "Build temporal-canary with OS: $(GOOS), ARCH: $(GOARCH)..."
go build -o temporal-canary cmd/canary/main.go
##### Checks #####
copyright:
@printf $(COLOR) "Check license header..."
@@ -371,9 +366,6 @@ start-cdc-standby: temporal-server
start-cdc-other: temporal-server
./temporal-server --zone other start
start-canary: temporal-canary
./temporal-canary start
##### Auxilary #####
go-generate:
@printf $(COLOR) "Regenerate everything..."
canary/batch.go
-226
View File
@@ -1,226 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"go.temporal.io/temporal"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/client"
"go.temporal.io/temporal/workflow"
)
func registerBatch(r registrar) {
registerWorkflow(r, batchWorkflow, wfTypeBatch)
registerWorkflow(r, batchWorkflowParent, wfTypeBatchParent)
registerWorkflow(r, batchWorkflowChild, wfTypeBatchChild)
registerActivity(r, verifyBatchActivity, activityTypeVerifyBatch)
registerActivity(r, startBatchWorkflow, activityTypeStartBatch)
}
const (
// TODO: to get rid of them:
// after batch job has an API, we should use the API: https://github.com/uber/cadence/issues/2225
sysBatchWFTypeName = "temporal-sys-batch-workflow"
systemBatcherTaskQueueName = "temporal-sys-batcher-taskqueue"
// there are two level, so totally 5*5 + 5 == 30 descendants
// default batch RPS is 50, so it will takes ~1 seconds to terminate all
numChildrenPerLevel = 5
)
type (
// BatchParams is from server repo
// TODO: to get rid of it:
// after batch job has an API, we should use the API: https://github.com/uber/cadence/issues/2225
BatchParams struct {
Namespace string
Query string
Reason string
BatchType string
}
)
func batchWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeBatch, scheduledTimeNanos)
if err != nil {
return err
}
cwo := workflow.ChildWorkflowOptions{
WorkflowExecutionTimeout: childWorkflowTimeout,
WorkflowTaskTimeout: decisionTaskTimeout,
}
ctx = workflow.WithChildOptions(ctx, cwo)
var fs []workflow.ChildWorkflowFuture
for i := 0; i < numChildrenPerLevel; i++ {
f := workflow.ExecuteChildWorkflow(ctx, wfTypeBatchParent, scheduledTimeNanos)
fs = append(fs, f)
}
// waiting for all workflow started
for i := 0; i < numChildrenPerLevel; i++ {
err := fs[i].GetChildWorkflowExecution().Get(ctx, nil)
if err != nil {
return profile.end(err)
}
}
// waiting for visibility
if err := workflow.Sleep(ctx, time.Second*2); err != nil {
return profile.end(err)
}
retryPolicy := &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2,
MaximumInterval: time.Second * 12,
MaximumAttempts: 4,
}
ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
TaskQueue: taskQueueName,
ScheduleToStartTimeout: scheduleToStartTimeout,
StartToCloseTimeout: activityTaskTimeout,
RetryPolicy: retryPolicy,
})
startTime := workflow.Now(ctx).Format(time.RFC3339)
err = workflow.ExecuteActivity(ctx, activityTypeStartBatch, namespace, startTime).Get(ctx, nil)
if err != nil {
return profile.end(err)
}
// waiting for visibility
if err := workflow.Sleep(ctx, time.Second*2); err != nil {
return profile.end(err)
}
err = workflow.ExecuteActivity(ctx, activityTypeVerifyBatch, namespace, startTime).Get(ctx, nil)
return profile.end(err)
}
func batchWorkflowParent(ctx workflow.Context, scheduledTimeNanos int64) error {
profile, err := beginWorkflow(ctx, wfTypeBatchParent, scheduledTimeNanos)
if err != nil {
return err
}
cwo := workflow.ChildWorkflowOptions{
WorkflowExecutionTimeout: childWorkflowTimeout,
WorkflowTaskTimeout: decisionTaskTimeout,
}
ctx = workflow.WithChildOptions(ctx, cwo)
var fs []workflow.ChildWorkflowFuture
for i := 0; i < numChildrenPerLevel; i++ {
f := workflow.ExecuteChildWorkflow(ctx, wfTypeBatchChild, scheduledTimeNanos)
fs = append(fs, f)
}
// waiting for all workflow to finish
for i := 0; i < numChildrenPerLevel; i++ {
err := fs[i].Get(ctx, nil)
if err != nil {
return profile.end(err)
}
}
return profile.end(err)
}
func batchWorkflowChild(ctx workflow.Context, scheduledTimeNanos int64) error {
profile, err := beginWorkflow(ctx, wfTypeBatchChild, scheduledTimeNanos)
if err != nil {
return err
}
if err := workflow.Sleep(ctx, time.Hour); err != nil {
return profile.end(err)
}
return profile.end(err)
}
func startBatchWorkflow(ctx context.Context, namespace, startTime string) error {
sdkClient := getContextValue(ctx, ctxKeyActivitySystemClient).(*activityContext).cadence
params := BatchParams{
Namespace: namespace,
Query: "WorkflowType = '" + wfTypeBatchParent + "' AND CloseTime = missing AND StartTime <'" + startTime + "' ",
Reason: "batch canary",
BatchType: "terminate",
}
options := client.StartWorkflowOptions{
WorkflowExecutionTimeout: childWorkflowTimeout,
WorkflowTaskTimeout: decisionTaskTimeout,
TaskQueue: systemBatcherTaskQueueName,
SearchAttributes: map[string]interface{}{
"CustomNamespace": namespace,
"Operator": "admin",
},
}
run, err := sdkClient.ExecuteWorkflow(ctx, options, sysBatchWFTypeName, params)
if err != nil {
return err
}
err = run.Get(ctx, nil)
return err
}
func verifyBatchActivity(ctx context.Context, namespace, startTime string) error {
svClient := getActivityContext(ctx).cadence.Service
q1 := "WorkflowType = '" + wfTypeBatchParent + "' AND CloseTime = missing AND StartTime <'" + startTime + "' "
resp, err := svClient.CountWorkflowExecutions(ctx, &workflowservice.CountWorkflowExecutionsRequest{
Namespace: namespace,
Query: q1,
})
if err != nil {
return err
}
if resp.GetCount() > 0 {
return fmt.Errorf("still seeing open workflows for %v , %v, %v", wfTypeBatchParent, q1, resp.GetCount())
}
q2 := "WorkflowType = '" + wfTypeBatchChild + "' AND CloseTime = missing AND StartTime <'" + startTime + "' "
resp, err = svClient.CountWorkflowExecutions(ctx, &workflowservice.CountWorkflowExecutionsRequest{
Namespace: namespace,
Query: q2,
})
if err != nil {
return err
}
if resp.GetCount() > 0 {
return fmt.Errorf("still seeing open workflows for %v, %v, %v", wfTypeBatchChild, q2, resp.GetCount())
}
return nil
}
canary/canary.go
-203
View File
@@ -1,203 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"github.com/opentracing/opentracing-go"
enumspb "go.temporal.io/temporal-proto/enums/v1"
"go.temporal.io/temporal-proto/serviceerror"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/worker"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
type (
// Runnable is an interface for anything that exposes a Run method
Runnable interface {
Run() error
}
registrar interface {
RegisterWorkflowWithOptions(w interface{}, options workflow.RegisterOptions)
RegisterActivityWithOptions(a interface{}, options activity.RegisterOptions)
RegisterActivity(a interface{})
}
canaryImpl struct {
canaryClient cadenceClient
canaryNamespace string
archivalClient cadenceClient
systemClient cadenceClient
runtime *RuntimeContext
}
activityContext struct {
cadence cadenceClient
}
)
type contextKey string
const (
// this context key should be the same as the one defined in
// internal_worker.go, cadence go client
testTagsContextKey contextKey = "cadence-testTags"
)
// new returns a new instance of Canary runnable
func newCanary(namespace string, rc *RuntimeContext) (Runnable, error) {
canaryClient, err := newCadenceClient(namespace, rc)
if err != nil {
return nil, err
}
archivalClient, err := newCadenceClient(archivalNamespace, rc)
if err != nil {
return nil, err
}
systemClient, err := newCadenceClient(systemNamespace, rc)
if err != nil {
return nil, err
}
return &canaryImpl{
canaryClient: canaryClient,
canaryNamespace: namespace,
archivalClient: archivalClient,
systemClient: systemClient,
runtime: rc,
}, nil
}
// Run runs the canary
func (c *canaryImpl) Run() error {
var err error
log := c.runtime.logger
if err = c.createNamespace(); err != nil {
log.Error("createNamespace failed", zap.Error(err))
return err
}
if err = c.createArchivalNamespace(); err != nil {
log.Error("createArchivalNamespace failed", zap.Error(err))
return err
}
// start the initial cron workflow
c.startCronWorkflow()
err = c.startWorker()
if err != nil {
log.Error("start worker failed", zap.Error(err))
return err
}
return nil
}
func (c *canaryImpl) startWorker() error {
options := worker.Options{
BackgroundActivityContext: c.newActivityContext(),
MaxConcurrentActivityExecutionSize: activityWorkerMaxExecutors,
}
archivalWorker := worker.New(c.archivalClient.Client, archivalTaskQueueName, options)
registerHistoryArchival(archivalWorker)
defer archivalWorker.Stop()
if err := archivalWorker.Start(); err != nil {
return err
}
canaryWorker := worker.New(c.canaryClient.Client, taskQueueName, options)
registerBatch(canaryWorker)
registerCancellation(canaryWorker)
registerConcurrentExec(canaryWorker)
registerCron(canaryWorker)
registerEcho(canaryWorker)
registerLocalActivity(canaryWorker)
registerQuery(canaryWorker)
registerReset(canaryWorker)
registerRetry(canaryWorker)
registerSanity(canaryWorker)
registerSearchAttributes(canaryWorker)
registerSignal(canaryWorker)
registerTimeout(canaryWorker)
registerVisibility(canaryWorker)
return canaryWorker.Run()
}
func (c *canaryImpl) startCronWorkflow() {
wfID := "cadence.canary.cron"
opts := newWorkflowOptions(wfID, cronWFExecutionTimeout)
opts.CronSchedule = "@every 30s" // run every 30s
// create the cron workflow span
ctx := context.Background()
span := opentracing.StartSpan("start-cron-workflow-span")
defer span.Finish()
ctx = opentracing.ContextWithSpan(ctx, span)
_, err := c.canaryClient.ExecuteWorkflow(ctx, opts, cronWorkflow, c.canaryNamespace, wfTypeSanity)
if err != nil {
if _, ok := err.(*serviceerror.WorkflowExecutionAlreadyStarted); !ok {
c.runtime.logger.Error("error starting cron workflow", zap.Error(err))
}
}
}
// newActivityContext builds an activity context containing
// logger, metricsClient and cadenceClient
func (c *canaryImpl) newActivityContext() context.Context {
ctx := context.WithValue(context.Background(), ctxKeyActivityRuntime, &activityContext{cadence: c.canaryClient})
ctx = context.WithValue(ctx, ctxKeyActivityArchivalRuntime, &activityContext{cadence: c.archivalClient})
ctx = context.WithValue(ctx, ctxKeyActivitySystemClient, &activityContext{cadence: c.systemClient})
return overrideWorkerOptions(ctx)
}
func (c *canaryImpl) createNamespace() error {
name := c.canaryNamespace
desc := "Namespace for running cadence canary workflows"
owner := "canary"
return c.canaryClient.createNamespace(name, desc, owner, enumspb.ARCHIVAL_STATUS_DISABLED)
}
func (c *canaryImpl) createArchivalNamespace() error {
name := archivalNamespace
desc := "Namespace used by cadence canary workflows to verify archival"
owner := "canary"
archivalStatus := enumspb.ARCHIVAL_STATUS_ENABLED
return c.archivalClient.createNamespace(name, desc, owner, archivalStatus)
}
// Override worker options to create large number of pollers to improve the chances of activities getting sync matched
//nolint:unused
func overrideWorkerOptions(ctx context.Context) context.Context {
optionsOverride := make(map[string]map[string]string)
optionsOverride["worker-options"] = map[string]string{
"ConcurrentPollRoutineSize": "20",
}
return context.WithValue(ctx, testTagsContextKey, optionsOverride)
}
canary/cancellation.go
-232
View File
@@ -1,232 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"time"
"go.temporal.io/temporal"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
sleepDuration = time.Second * 10
)
func registerCancellation(r registrar) {
registerWorkflow(r, cancellationWorkflow, wfTypeCancellation)
registerWorkflow(r, cancellationExternalWorkflow, wfTypeCancellationExternal)
registerActivity(r, cancellationActivity, activityTypeCancellation)
registerActivity(r, cancellationChildActivity, activityTypeCancellationChild)
}
// cancellationWorkflow is the workflow implementation to test for cancellation of workflows
func cancellationWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeCancellation, scheduledTimeNanos)
if err != nil {
return err
}
// this test cancel external workflow
activityCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
err = workflow.ExecuteActivity(activityCtx, activityTypeCancellation, workflow.Now(ctx).UnixNano()).Get(activityCtx, nil)
if err != nil {
workflow.GetLogger(ctx).Info("cancel workflow failed", zap.Error(err))
return profile.end(err)
}
// this test cancellation of child workflow, inside workflow
cwo := newChildWorkflowOptions(namespace, wfTypeCancellationExternal+"-child")
childCtx := workflow.WithChildOptions(ctx, cwo)
childCtx, cancel := workflow.WithCancel(childCtx)
childFuture := workflow.ExecuteChildWorkflow(childCtx, wfTypeCancellationExternal, workflow.Now(ctx).UnixNano(), sleepDuration)
if err := workflow.Sleep(ctx, 1*time.Second); err != nil {
return profile.end(err)
}
cancel()
err = childFuture.Get(childCtx, nil)
if err == nil {
msg := "cancellationWorkflow failed: child workflow not cancelled"
workflow.GetLogger(ctx).Info(msg)
return profile.end(errors.New(msg))
}
if _, ok := err.(*temporal.CanceledError); !ok {
workflow.GetLogger(ctx).Info("cancellationWorkflow failed: child workflow return non CanceledError", zap.Error(err))
return profile.end(err)
}
// this test cancellation of child workflow, outside workflow
activityCancelChildTestFn := func(useRunID bool) error {
cwo := newChildWorkflowOptions(namespace, wfTypeCancellationExternal+"-child-outside")
childCtx := workflow.WithChildOptions(ctx, cwo)
childFuture := workflow.ExecuteChildWorkflow(childCtx, wfTypeCancellationExternal, workflow.Now(ctx).UnixNano(), sleepDuration)
childExecution := &workflow.Execution{}
err := childFuture.GetChildWorkflowExecution().Get(childCtx, childExecution)
if err != nil {
return err
}
if !useRunID {
childExecution.RunID = ""
}
// starts a activity to cancel this child workflow
activityCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
err = workflow.ExecuteActivity(activityCtx, activityTypeCancellationChild, workflow.Now(ctx).UnixNano(), childExecution).Get(activityCtx, nil)
if err != nil {
workflow.GetLogger(ctx).Info("cancel child workflow from activity failed", zap.Error(err))
return err
}
err = childFuture.Get(childCtx, nil)
if err == nil {
msg := "cancellationWorkflow failed: child workflow not cancelled"
workflow.GetLogger(ctx).Info(msg)
return errors.New(msg)
}
if _, ok := err.(*temporal.CanceledError); !ok {
workflow.GetLogger(ctx).Info("cancellationWorkflow failed: child workflow return non CanceledError", zap.Error(err))
return err
}
return nil
}
err = activityCancelChildTestFn(false)
if err != nil {
return profile.end(err)
}
err = activityCancelChildTestFn(true)
if err != nil {
return profile.end(err)
}
decisionCancelTestFn := func(useRunID bool) error {
workflowIDSuffix := "-with-run-ID"
if !useRunID {
workflowIDSuffix = "-without-run-ID"
}
cwo = newChildWorkflowOptions(namespace, wfTypeCancellationExternal+workflowIDSuffix)
childCtx = workflow.WithChildOptions(ctx, cwo)
childFuture = workflow.ExecuteChildWorkflow(childCtx, wfTypeCancellationExternal, workflow.Now(ctx).UnixNano(), sleepDuration)
childExecution := &workflow.Execution{}
err = childFuture.GetChildWorkflowExecution().Get(childCtx, childExecution)
if err != nil {
return err
}
if !useRunID {
childExecution.RunID = ""
}
cancellatinFuture := workflow.RequestCancelExternalWorkflow(ctx, childExecution.ID, childExecution.RunID)
err = cancellatinFuture.Get(ctx, nil)
if err != nil {
workflow.GetLogger(ctx).Info("cancellationWorkflow failed: fail to send cancellation to child workflow", zap.Error(err))
return err
}
err = childFuture.Get(childCtx, nil)
if err == nil {
msg := "cancellationWorkflow failed: child workflow not cancelled"
workflow.GetLogger(ctx).Info(msg)
return errors.New(msg)
}
if _, ok := err.(*temporal.CanceledError); !ok {
workflow.GetLogger(ctx).Info("cancellationWorkflow failed: child workflow return non CanceledError", zap.Error(err))
return err
}
return nil
}
err = decisionCancelTestFn(false)
if err != nil {
return profile.end(err)
}
err = decisionCancelTestFn(true)
return profile.end(err)
}
// cancellationActivity is the activity implementation to test for cancellation of non child workflow, using API
func cancellationActivity(ctx context.Context, scheduledTimeNanos int64) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeCancellation, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
apiCancelTestFn := func(useRunID bool) error {
opts := newWorkflowOptions(wfTypeCancellationExternal, childWorkflowTimeout)
workflowRun, err := client.ExecuteWorkflow(context.Background(), opts, wfTypeCancellationExternal, scheduledTimeNanos, sleepDuration)
if err != nil {
return err
}
runID := ""
if useRunID {
runID = workflowRun.GetRunID()
}
err = client.CancelWorkflow(context.Background(), wfTypeCancellationExternal, runID)
if err != nil {
return err
}
err = workflowRun.Get(ctx, nil)
if err == nil {
return errors.New("cancellationWorkflow failed: non child workflow not cancelled")
}
if _, ok := err.(*temporal.CanceledError); !ok {
return err
}
return nil
}
if err := apiCancelTestFn(false); err != nil {
return err
}
return apiCancelTestFn(true)
}
// cancellationActivity is the activity implementation to test for cancellation of non child workflow, using API
func cancellationChildActivity(ctx context.Context, scheduledTimeNanos int64, execution workflow.Execution) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeCancellationChild, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
return client.CancelWorkflow(context.Background(), execution.ID, execution.RunID)
}
// cancellationExternalWorkflow is the workflow implementation to test for cancellation of non child workflow
func cancellationExternalWorkflow(ctx workflow.Context, scheduledTimeNanos int64, sleepDuration time.Duration) error {
profile, err := beginWorkflow(ctx, wfTypeCancellationExternal, scheduledTimeNanos)
if err != nil {
return profile.end(err)
}
err = workflow.Sleep(ctx, sleepDuration)
if err != nil {
return profile.end(err)
}
return nil
}
canary/client.go
-161
View File
@@ -1,161 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"time"
"github.com/opentracing/opentracing-go"
enumspb "go.temporal.io/temporal-proto/enums/v1"
"go.temporal.io/temporal-proto/serviceerror"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/client"
"go.temporal.io/temporal/workflow"
)
// cadenceClient is an abstraction on top of
// the cadence library client that serves as
// a union of all the client interfaces that
// the library exposes
type cadenceClient struct {
client.Client
// namespaceClient only exposes namespace API
client.NamespaceClient
// this is the service needed to start the workers
Service workflowservice.WorkflowServiceClient
}
// createNamespace creates a cadence namespace with the given name and description
// if the namespace already exist, this method silently returns success
func (client *cadenceClient) createNamespace(name string, desc string, owner string, archivalStatus enumspb.ArchivalStatus) error {
emitMetric := true
isGlobalNamespace := false
retention := int32(workflowRetentionDays)
if archivalStatus == enumspb.ARCHIVAL_STATUS_ENABLED {
retention = int32(0)
}
req := &workflowservice.RegisterNamespaceRequest{
Name: name,
Description: desc,
OwnerEmail: owner,
WorkflowExecutionRetentionPeriodInDays: retention,
EmitMetric: emitMetric,
HistoryArchivalStatus: archivalStatus,
VisibilityArchivalStatus: archivalStatus,
IsGlobalNamespace: isGlobalNamespace,
}
err := client.Register(context.Background(), req)
if err != nil {
if _, ok := err.(*serviceerror.NamespaceAlreadyExists); !ok {
return err
}
}
return nil
}
// newCadenceClient builds a cadenceClient from the runtimeContext
func newCadenceClient(namespace string, runtime *RuntimeContext) (cadenceClient, error) {
tracer := opentracing.GlobalTracer()
cclient, err := client.NewClient(
client.Options{
HostPort: runtime.hostPort,
Namespace: namespace,
MetricsScope: runtime.metrics,
Tracer: tracer,
Logger: runtime.logger,
},
)
if err != nil {
return cadenceClient{}, err
}
namespaceClient, err := client.NewNamespaceClient(
client.Options{
HostPort: runtime.hostPort,
MetricsScope: runtime.metrics,
Tracer: tracer,
Logger: runtime.logger,
},
)
if err != nil {
return cadenceClient{}, err
}
return cadenceClient{
Client: cclient,
NamespaceClient: namespaceClient,
Service: runtime.service,
}, nil
}
// newWorkflowOptions builds workflowOptions with defaults for everything except startToCloseTimeout
func newWorkflowOptions(id string, executionTimeout time.Duration) client.StartWorkflowOptions {
return client.StartWorkflowOptions{
ID: id,
TaskQueue: taskQueueName,
WorkflowRunTimeout: executionTimeout,
WorkflowTaskTimeout: decisionTaskTimeout,
WorkflowIDReusePolicy: enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
}
}
// newActivityOptions builds and returns activityOptions with reasonable defaults
func newActivityOptions() workflow.ActivityOptions {
return workflow.ActivityOptions{
TaskQueue: taskQueueName,
StartToCloseTimeout: activityTaskTimeout,
ScheduleToStartTimeout: scheduleToStartTimeout,
ScheduleToCloseTimeout: scheduleToStartTimeout + activityTaskTimeout,
}
}
// newChildWorkflowOptions builds and returns childWorkflowOptions for given namespace
func newChildWorkflowOptions(namespace string, wfID string) workflow.ChildWorkflowOptions {
return workflow.ChildWorkflowOptions{
Namespace: namespace,
WorkflowID: wfID,
TaskQueue: taskQueueName,
WorkflowRunTimeout: childWorkflowTimeout,
WorkflowTaskTimeout: decisionTaskTimeout,
WorkflowIDReusePolicy: enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
}
}
// registerWorkflow registers a workflow function with a given friendly name
func registerWorkflow(r registrar, workflowFunc interface{}, name string) {
r.RegisterWorkflowWithOptions(workflowFunc, workflow.RegisterOptions{Name: name})
}
// registerActivity registers an activity function with a given friendly name
func registerActivity(r registrar, activityFunc interface{}, name string) {
if name == "" {
r.RegisterActivity(activityFunc)
} else {
r.RegisterActivityWithOptions(activityFunc, activity.RegisterOptions{Name: name})
}
}
canary/common.go
-111
View File
@@ -1,111 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
func maxInt64(a, b int64) int64 {
if a > b {
return a
}
return b
}
func absDurationDiff(d1, d2 time.Duration) time.Duration {
if d1 > d2 {
return d1 - d2
}
return d2 - d1
}
func stringPtr(v string) *string {
return &v
}
func int32Ptr(v int32) *int32 {
return &v
}
// getContextValue retrieves and returns the value corresponding
// to the given key - panics if the key does not exist
func getContextValue(ctx context.Context, key string) interface{} {
value := ctx.Value(key)
if value == nil {
panic("ctx.Value(" + key + ") returned nil")
}
return value
}
// getActivityContext retrieves and returns the activity context from the
// global context passed to the activity
func getActivityContext(ctx context.Context) *activityContext {
return getContextValue(ctx, ctxKeyActivityRuntime).(*activityContext)
}
// getActivityArchivalContext retrieves and returns the activity archival context from the
// global context passed to the activity
func getActivityArchivalContext(ctx context.Context) *activityContext {
return getContextValue(ctx, ctxKeyActivityArchivalRuntime).(*activityContext)
}
// checkWFVersionCompatibility takes a workflow.Context param and
// validates that the workflow task currently being handled
// is compatible with this version of the canary - this method
// MUST only be called within a workflow function and it MUST
// be the first line in the workflow function
// Returns an error if the version is incompatible
func checkWFVersionCompatibility(ctx workflow.Context) error {
version := workflow.GetVersion(ctx, workflowChangeID, workflowVersion, workflowVersion)
if version != workflowVersion {
workflow.GetLogger(ctx).Error("workflow version mismatch",
zap.Int("want", int(workflowVersion)), zap.Int("got", int(version)))
return fmt.Errorf("workflow version mismatch, want=%v, got=%v", workflowVersion, version)
}
return nil
}
// beginWorkflow executes the common steps involved in all the workflow functions
// It checks for workflow task version compatibility and also records the execution
// in m3. This function must be the first call in every workflow function
// Returns metrics scope on success, error on failure
func beginWorkflow(ctx workflow.Context, wfType string, scheduledTimeNanos int64) (*workflowMetricsProfile, error) {
profile := recordWorkflowStart(ctx, wfType, scheduledTimeNanos)
if err := checkWFVersionCompatibility(ctx); err != nil {
profile.scope.Counter(errIncompatibleVersion).Inc(1)
return nil, profile.end(err)
}
return profile, nil
}
func concat(first string, second string) string {
return first + "/" + second
}
canary/concurrentExec.go
-102
View File
@@ -1,102 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
// to force cuncurrent execution of activity and pagination on history events
// we need to make this number larger than the default page size, which is 1000
numConcurrentExec = 25
totalConcurrentExec = 250
)
func registerConcurrentExec(r registrar) {
registerWorkflow(r, concurrentExecWorkflow, wfTypeConcurrentExec)
registerActivity(r, concurrentExecActivity, activityTypeConcurrentExec)
}
// concurrentExecWorkflow is the workflow implementation to test
// 1. client side events pagination when reconstructing workflow state
// 2. concurrent execution of activities
func concurrentExecWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeConcurrentExec, scheduledTimeNanos)
if err != nil {
return err
}
selector := workflow.NewSelector(ctx)
errors := make([]error, totalConcurrentExec)
doActivity := func(index int) {
now := workflow.Now(ctx).UnixNano()
activityCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
future := workflow.ExecuteActivity(activityCtx, activityTypeConcurrentExec, now)
selector.AddFuture(future, func(f workflow.Future) {
// do not care about the return value
errors[index] = f.Get(activityCtx, nil)
})
}
for index := 0; index < numConcurrentExec; index++ {
doActivity(index)
}
for index := numConcurrentExec; index < totalConcurrentExec; index++ {
selector.Select(ctx)
doActivity(index)
}
for index := 0; index < numConcurrentExec; index++ {
selector.Select(ctx)
}
for _, err := range errors {
if err != nil {
workflow.GetLogger(ctx).Info("concurrentExecActivity failed", zap.Error(err))
return profile.end(err)
}
}
return profile.end(nil)
}
// concurrentExecActivity is the activity implementation for concurrent execution test
func concurrentExecActivity(ctx context.Context, scheduledTimeNanos int64) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeConcurrentExec, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
// canary test do not require any actual work to be done, so just return
return nil
}
canary/config.go
-116
View File
@@ -1,116 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"fmt"
"github.com/uber-go/tally"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.uber.org/zap"
"github.com/temporalio/temporal/common/service/config"
)
const (
// EnvKeyRoot the environment variable key for runtime root dir
EnvKeyRoot = "TEMPORAL_CANARY_ROOT"
// EnvKeyConfigDir the environment variable key for config dir
EnvKeyConfigDir = "TEMPORAL_CANARY_CONFIG_DIR"
// EnvKeyEnvironment is the environment variable key for environment
EnvKeyEnvironment = "TEMPORAL_CANARY_ENVIRONMENT"
// EnvKeyAvailabilityZone is the environment variable key for AZ
EnvKeyAvailabilityZone = "TEMPORAL_CANARY_AVAILABILITY_ZONE"
)
const (
// CadenceLocalHostPort is the default address for cadence frontend service
ServiceHostPort = "127.0.0.1:7933"
// CadenceServiceName is the default service name for cadence frontend
ServiceName = "frontend"
// CanaryServiceName is the default service name for cadence canary
CanaryServiceName = "canary"
)
type (
// Config contains the configurable yaml
// properties for the canary runtime
Config struct {
Canary Canary `yaml:"canary"`
Cadence Cadence `yaml:"temporal"`
Log config.Logger `yaml:"log"`
Metrics config.Metrics `yaml:"metrics"`
}
// Canary contains the configuration for canary tests
Canary struct {
Namespaces []string `yaml:"namespaces"`
Excludes []string `yaml:"excludes"`
PProf PProf `yaml:"pprof"`
}
// PProf contains the config items for the pprof utility
PProf struct {
Port int `yaml:"port"`
}
// Cadence contains the configuration for cadence service
Cadence struct {
ServiceName string `yaml:"service"`
HostNameAndPort string `yaml:"host"`
}
)
// Validate validates canary configration
func (c *Config) Validate() error {
if len(c.Canary.Namespaces) == 0 {
return fmt.Errorf("missing value for namespaces property")
}
return nil
}
// RuntimeContext contains all the context
// information needed to run the canary
type RuntimeContext struct {
logger *zap.Logger
metrics tally.Scope
hostPort string
service workflowservice.WorkflowServiceClient
}
// NewRuntimeContext builds a runtime context from the config
func NewRuntimeContext(
logger *zap.Logger,
scope tally.Scope,
hostPort string,
service workflowservice.WorkflowServiceClient,
) *RuntimeContext {
return &RuntimeContext{
logger: logger,
metrics: scope,
hostPort: hostPort,
service: service,
}
}
canary/const.go
-111
View File
@@ -1,111 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"time"
"go.temporal.io/temporal/workflow"
)
// global constants
const (
workflowRetentionDays = int32(1)
activityWorkerMaxExecutors = 256
scheduleToStartTimeout = 3 * time.Minute
decisionTaskTimeout = 10 * time.Second
activityTaskTimeout = 3 * time.Minute
childWorkflowTimeout = 6 * time.Minute
taskQueueName = "canary-task-queue"
ctxKeyActivityRuntime = "runtime"
ctxKeyActivityArchivalRuntime = "runtime-archival"
ctxKeyActivitySystemClient = "system-client"
archivalNamespace = "canary-archival-namespace"
systemNamespace = "temporal-system"
archivalTaskQueueName = "canary-archival-task-queue"
)
// workflowVersion represents the current version of every single
// workflow function in this canary. Every workflow function verifies
// that the decision task it is executing is compatible with this version
// Bump this version whenever a backward incompatible change for any workflow
// also see beingWorkflow function
const workflowVersion = workflow.Version(3)
const workflowChangeID = "initial version"
const (
cronJobTimeout = 9 * time.Minute
cronWFExecutionTimeout = 18 * time.Minute
)
// wfType/activityType refers to the friendly short names given to
// workflows and activities - at the time of registration, these names
// will be used to associate with a workflow or activity function
const (
wfTypeCron = "workflow.cron"
wfTypeSanity = "workflow.sanity"
wfTypeEcho = "workflow.echo"
wfTypeSignal = "workflow.signal"
wfTypeSignalExternal = "workflow.signal.external"
wfTypeVisibility = "workflow.visibility"
wfTypeSearchAttributes = "workflow.searchAttributes"
wfTypeConcurrentExec = "workflow.concurrent-execution"
wfTypeQuery = "workflow.query"
wfTypeTimeout = "workflow.timeout"
wfTypeLocalActivity = "workflow.localactivity"
wfTypeCancellation = "workflow.cancellation"
wfTypeCancellationExternal = "workflow.cancellation.external"
wfTypeRetry = "workflow.retry"
wfTypeResetBase = "workflow.reset.base"
wfTypeReset = "workflow.reset"
wfTypeHistoryArchival = "workflow.archival.history"
wfTypeVisibilityArchival = "workflow.archival.visibility"
wfTypeArchivalExternal = "workflow.archival.external"
wfTypeBatch = "workflow.batch"
wfTypeBatchParent = "workflow.batch.parent"
wfTypeBatchChild = "workflow.batch.child"
activityTypeEcho = "activity.echo"
activityTypeCron = "activity.cron"
activityTypeSignal = "activity.signal"
activityTypeVisibility = "activity.visibility"
activityTypeSearchAttributes = "activity.searchAttributes"
activityTypeConcurrentExec = "activity.concurrent-execution"
activityTypeQuery1 = "activity.query1"
activityTypeQuery2 = "activity.query2"
activityTypeTimeout = "activity.timeout"
activityTypeCancellation = "activity.cancellation"
activityTypeCancellationChild = "activity.cancellation.child"
activityTypeRetryOnTimeout = "activity.retry-on-timeout"
activityTypeRetryOnFailure = "activity.retry-on-failure"
activityTypeTriggerReset = "activity.reset.trigger"
activityTypeVerifyReset = "activity.reset.verify"
activityTypeResetBase = "activity.reset.base"
activityTypeHistoryArchival = "activity.archival.history"
activityTypeVisibilityArchival = "activity.archival.visibility"
activityTypeLargeResult = "activity.largeResult"
activityTypeVerifyBatch = "activity.batch.verify"
activityTypeStartBatch = "activity.batch.start.batch"
)
canary/cron.go
-139
View File
@@ -1,139 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"github.com/opentracing/opentracing-go"
"github.com/uber-go/tally"
"go.temporal.io/temporal-proto/serviceerror"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/client"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
var (
cronSleepTime = time.Second * 5
)
func registerCron(r registrar) {
registerWorkflow(r, cronWorkflow, wfTypeCron)
registerActivity(r, cronActivity, "")
}
// cronWorkflow is a generic cron workflow implementation that takes as input
// a tuple - (jobName, JobFrequency) where jobName refers to a workflow name
// and job frequency refers to the rate at which the job must be scheduled
// there are several assumptions that this cron makes:
// - jobFrequency is any value between 1-59 seconds
// - jobName refers to a workflow function with a well-defined function signature
// - every instance of job completes within 10 mins
func cronWorkflow(
ctx workflow.Context,
namespace string,
jobName string) error {
profile, err := beginWorkflow(ctx, wfTypeCron, workflow.Now(ctx).UnixNano())
aCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
startTime := workflow.Now(ctx).UnixNano()
workflow.ExecuteActivity(aCtx, cronActivity, startTime, namespace, jobName)
workflow.Sleep(ctx, cronSleepTime)
elapsed := time.Duration(workflow.Now(ctx).UnixNano() - startTime)
profile.scope.Timer(timerDriftLatency).Record(absDurationDiff(elapsed, cronSleepTime))
return err
}
// cronActivity starts root canary workflows at a pre-defined frequency
// this activity exits automatically a minute after it is scheduled
func cronActivity(
ctx context.Context,
scheduledTimeNanos int64,
namespace string,
jobName string) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeCron, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
cadenceClient := getActivityContext(ctx).cadence
logger := activity.GetLogger(ctx)
parentID := activity.GetInfo(ctx).WorkflowExecution.ID
jobID := fmt.Sprintf("%s-%s-%v", parentID, jobName, time.Now().Format(time.RFC3339))
wf, err := startJob(&cadenceClient, scope, jobID, jobName, namespace)
if err != nil {
logger.Error("cronActivity: failed to start job", zap.Error(err))
if _, ok := err.(*serviceerror.NamespaceNotActive); ok {
return err
}
} else {
logger.Info("cronActivity: started new job",
zap.String("wfID", jobID), zap.String("runID", wf.GetRunID()))
}
return nil
}
func startJob(
cadenceClient *cadenceClient,
scope tally.Scope,
jobID string,
jobName string,
namespace string) (client.WorkflowRun, error) {
scope.Counter(startWorkflowCount).Inc(1)
sw := scope.Timer(startWorkflowLatency).Start()
defer sw.Stop()
// start off a workflow span
ctx := context.Background()
span := opentracing.StartSpan(fmt.Sprintf("start-%v", jobName))
defer span.Finish()
ctx = opentracing.ContextWithSpan(ctx, span)
opts := newWorkflowOptions(jobID, cronJobTimeout)
wf, err := cadenceClient.ExecuteWorkflow(ctx, opts, jobName, time.Now().UnixNano(), namespace)
if err != nil {
scope.Counter(startWorkflowFailureCount).Inc(1)
switch err.(type) {
case *serviceerror.WorkflowExecutionAlreadyStarted:
scope.Counter(startWorkflowAlreadyStartedCount).Inc(1)
case *serviceerror.NamespaceNotActive:
scope.Counter(startWorkflowNamespaceNotActiveCount).Inc(1)
}
return nil, err
}
scope.Counter(startWorkflowSuccessCount).Inc(1)
return wf, err
}
canary/echo.go
-111
View File
@@ -1,111 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"math"
"reflect"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
// echoInput is the input to echoActivity
type echoInput struct {
IntVal int64
IntPtrVal *int64
FloatVal float64
StringVal string
StringPtrVal *string
SliceVal []string
MapVal map[string]string
}
// echoOutput is the output from echoActivity
type echoOutput echoInput
func registerEcho(r registrar) {
registerWorkflow(r, echoWorkflow, wfTypeEcho)
registerActivity(r, echoActivity, activityTypeEcho)
}
// echoWorkflow is a workflow implementation which simply executes an
// activity that echoes back the input as output
func echoWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeEcho, scheduledTimeNanos)
if err != nil {
return err
}
input := newEchoInput()
now := workflow.Now(ctx).UnixNano()
aCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
future := workflow.ExecuteActivity(aCtx, activityTypeEcho, now, input)
var output echoOutput
if err := future.Get(aCtx, &output); err != nil {
workflow.GetLogger(ctx).Info("echoActivity failed", zap.Error(err))
return profile.end(err)
}
if !reflect.DeepEqual(output, echoOutput(*input)) {
workflow.GetLogger(ctx).Sugar().Error("echoActivity returned wrong output", output, input)
return profile.end(errors.New("echoActivity returned invalid output"))
}
return profile.end(nil)
}
// echoActivity is an activity implementation that simply returns the input as output
// it also validates that the passed in input has the exepected values
func echoActivity(ctx context.Context, scheduledTimeNanos int64, input *echoInput) (echoOutput, error) {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeEcho, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
expected := newEchoInput()
if !reflect.DeepEqual(expected, input) {
err = errors.New("equality check failed")
return echoOutput{}, err
}
return echoOutput(*input), nil
}
func newEchoInput() *echoInput {
intVal := int64(math.MaxInt64)
stringVal := "canary_echo_test"
return &echoInput{
IntVal: intVal,
IntPtrVal: &intVal,
FloatVal: math.MaxFloat64,
StringVal: stringVal,
StringPtrVal: &stringVal,
SliceVal: []string{"canary", ".", "echoWorkflow"},
MapVal: map[string]string{"us-east-1": "dca1a", "us-west-1": "sjc1a"},
}
}
canary/historyArchival.go
-179
View File
@@ -1,179 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"fmt"
"math/rand"
"time"
"github.com/google/uuid"
commonpb "go.temporal.io/temporal-proto/common/v1"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
numHistoryArchivals = 5
resultSize = 1024 // 1KB
)
func registerHistoryArchival(r registrar) {
registerWorkflow(r, historyArchivalWorkflow, wfTypeHistoryArchival)
registerActivity(r, historyArchivalActivity, activityTypeHistoryArchival)
registerWorkflow(r, archivalExternalWorkflow, wfTypeArchivalExternal)
registerActivity(r, largeResultActivity, activityTypeLargeResult)
}
func historyArchivalWorkflow(ctx workflow.Context, scheduledTimeNanos int64, _ string) error {
profile, err := beginWorkflow(ctx, wfTypeHistoryArchival, scheduledTimeNanos)
if err != nil {
return err
}
ch := workflow.NewBufferedChannel(ctx, numHistoryArchivals)
for i := 0; i < numHistoryArchivals; i++ {
workflow.Go(ctx, func(ctx2 workflow.Context) {
aCtx := workflow.WithActivityOptions(ctx2, newActivityOptions())
err := workflow.ExecuteActivity(aCtx, activityTypeHistoryArchival, workflow.Now(ctx2).UnixNano()).Get(aCtx, nil)
errStr := ""
if err != nil {
errStr = err.Error()
}
ch.Send(ctx2, errStr)
})
}
successfulArchivalsCount := 0
for i := 0; i < numHistoryArchivals; i++ {
var errStr string
ch.Receive(ctx, &errStr)
if errStr != "" {
workflow.GetLogger(ctx).Error("at least one archival failed", zap.Int("success-count", successfulArchivalsCount), zap.String("err-string", errStr))
return profile.end(errors.New(errStr))
}
successfulArchivalsCount++
}
return profile.end(nil)
}
func historyArchivalActivity(ctx context.Context, scheduledTimeNanos int64) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeHistoryArchival, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityArchivalContext(ctx).cadence
execution, err := executeArchivalExeternalWorkflow(ctx, client, scheduledTimeNanos)
if err != nil {
return err
}
getHistoryReq := &workflowservice.GetWorkflowExecutionHistoryRequest{
Namespace: archivalNamespace,
Execution: execution,
}
failureReason := ""
attempts := 0
expireTime := time.Now().Add(activityTaskTimeout)
for {
<-time.After(5 * time.Second)
if time.Now().After(expireTime) {
break
}
attempts++
bCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
history, err := client.Service.GetWorkflowExecutionHistory(bCtx, getHistoryReq)
cancel()
if err != nil {
failureReason = fmt.Sprintf("error accessing history, %v", err.Error())
} else if !history.GetArchived() {
failureReason = "history is not archived"
} else if len(history.History.Events) == 0 {
failureReason = "got empty history"
} else {
return nil
}
}
activity.GetLogger(ctx).Error("failed to get archived history within time limit",
zap.String("failure_reason", failureReason),
zap.String("namespace", archivalNamespace),
zap.String("workflow_id", execution.GetWorkflowId()),
zap.String("run_id", execution.GetRunId()),
zap.Int("attempts", attempts))
return fmt.Errorf("failed to get archived history within time limit, %v", failureReason)
}
func executeArchivalExeternalWorkflow(
ctx context.Context,
client cadenceClient,
scheduledTimeNanos int64,
) (*commonpb.WorkflowExecution, error) {
workflowID := fmt.Sprintf("%v.%v", wfTypeArchivalExternal, uuid.New())
ops := newWorkflowOptions(workflowID, childWorkflowTimeout)
ops.TaskQueue = archivalTaskQueueName
workflowRun, err := client.ExecuteWorkflow(ctx, ops, wfTypeArchivalExternal, scheduledTimeNanos)
if err != nil {
return nil, err
}
if err := workflowRun.Get(ctx, nil); err != nil {
return nil, err
}
return &commonpb.WorkflowExecution{
WorkflowId: workflowID,
RunId: workflowRun.GetRunID(),
}, nil
}
func archivalExternalWorkflow(ctx workflow.Context, scheduledTimeNanos int64) error {
profile, err := beginWorkflow(ctx, wfTypeArchivalExternal, scheduledTimeNanos)
if err != nil {
return err
}
ao := workflow.ActivityOptions{
ScheduleToStartTimeout: time.Minute,
StartToCloseTimeout: time.Minute,
}
var result []byte
numActs := rand.Intn(10) + 1
for i := 0; i != numActs; i++ {
aCtx := workflow.WithActivityOptions(ctx, ao)
if err = workflow.ExecuteActivity(aCtx, activityTypeLargeResult).Get(aCtx, &result); err != nil {
break
}
}
if err != nil {
return profile.end(err)
}
return profile.end(nil)
}
func largeResultActivity() ([]byte, error) {
return make([]byte, resultSize, resultSize), nil
}
canary/localactivity.go
-168
View File
@@ -1,168 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"math/rand"
"time"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
)
func registerLocalActivity(r registrar) {
registerWorkflow(r, localActivityWorkfow, wfTypeLocalActivity)
registerActivity(r, activityForCondition0, "")
registerActivity(r, activityForCondition1, "")
registerActivity(r, activityForCondition2, "")
registerActivity(r, activityForCondition3, "")
registerActivity(r, activityForCondition4, "")
}
type conditionAndAction struct {
// condition is a function pointer to a local activity
condition interface{}
// action is a function pointer to a regular activity
action interface{}
}
var checks = []conditionAndAction{
{checkCondition0, activityForCondition0},
{checkCondition1, activityForCondition1},
{checkCondition2, activityForCondition2},
{checkCondition3, activityForCondition3},
{checkCondition4, activityForCondition4},
}
func localActivityWorkfow(ctx workflow.Context) (string, error) {
profile, err := beginWorkflow(ctx, wfTypeSignal, workflow.Now(ctx).UnixNano())
if err != nil {
return "", err
}
logger := workflow.GetLogger(ctx)
lao := workflow.LocalActivityOptions{
// use short timeout as local activity is execute as function locally.
ScheduleToCloseTimeout: time.Second,
}
ctx = workflow.WithLocalActivityOptions(ctx, lao)
var data int32
err = workflow.ExecuteLocalActivity(ctx, getConditionData).Get(ctx, &data)
if err != nil {
return "", profile.end(err)
}
logger.Sugar().Infof("Get condition data %v", data)
ao := workflow.ActivityOptions{
ScheduleToStartTimeout: time.Minute,
StartToCloseTimeout: time.Minute,
}
ctx = workflow.WithActivityOptions(ctx, ao)
var actionFutures []workflow.Future
for i, check := range checks {
var conditionMeet bool
err := workflow.ExecuteLocalActivity(ctx, check.condition, data).Get(ctx, &conditionMeet)
if err != nil {
return "", profile.end(err)
}
logger.Sugar().Infof("condition meet for %v: %v", i, conditionMeet)
if conditionMeet {
f := workflow.ExecuteActivity(ctx, check.action)
actionFutures = append(actionFutures, f)
}
}
var processResult string
for _, f := range actionFutures {
var actionResult string
if err := f.Get(ctx, &actionResult); err != nil {
return "", profile.end(err)
}
if len(processResult) > 0 {
processResult += " and "
}
processResult += actionResult
}
logger.Sugar().Infof("Processed condition %v: %v", data, processResult)
return processResult, profile.end(nil)
}
func getConditionData() (int32, error) {
return rand.Int31n(100), nil
}
func checkCondition0(ctx context.Context, data int32) (bool, error) {
return data < 10, nil
}
func checkCondition1(ctx context.Context, data int32) (bool, error) {
return data >= 90, nil
}
func checkCondition2(ctx context.Context, data int32) (bool, error) {
return data%2 == 0, nil
}
func checkCondition3(ctx context.Context, data int32) (bool, error) {
return data%3 == 0, nil
}
func checkCondition4(ctx context.Context, data int32) (bool, error) {
return data%5 == 0, nil
}
func activityForCondition0(ctx context.Context) (string, error) {
activity.GetLogger(ctx).Info("process for condition 0")
return "data < 10", nil
}
func activityForCondition1(ctx context.Context) (string, error) {
activity.GetLogger(ctx).Info("process for condition 1")
return "data >= 90", nil
}
func activityForCondition2(ctx context.Context) (string, error) {
activity.GetLogger(ctx).Info("process for condition 2")
return "data%2 == 0", nil
}
func activityForCondition3(ctx context.Context) (string, error) {
activity.GetLogger(ctx).Info("process for condition 3")
return "data%3 == 0", nil
}
func activityForCondition4(ctx context.Context) (string, error) {
activity.GetLogger(ctx).Info("process for condition 4")
return "data%5 == 0", nil
}
canary/metrics.go
-140
View File
@@ -1,140 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"time"
"github.com/uber-go/tally"
"go.temporal.io/temporal"
"go.temporal.io/temporal/workflow"
)
// counters go here
const (
startedCount = "started"
failedCount = "failed"
successCount = "succeeded"
startWorkflowCount = "startworkflow"
startWorkflowSuccessCount = "startworkflow.success"
startWorkflowFailureCount = "startworkflow.failures"
startWorkflowAlreadyStartedCount = "startworkflow.failures.alreadystarted"
startWorkflowNamespaceNotActiveCount = "startworkflow.failures.namespacenotactive"
listOpenWorkflowsCount = "list-open-workflows"
listOpenWorkflowsFailureCount = "list-open-workflows.failures"
listWorkflowsCount = "list-workflows"
listWorkflowsFailureCount = "list-workflows.failures"
listArchivedWorkflowCount = "list-archived-workflows"
listArchivedWorkflowFailureCount = "list-archived-workflows.failures"
getWorkflowHistoryCount = "get-workflow-history"
getWorkflowHistoryFailureCount = "get-workflow-history.failures"
errTimeoutCount = "errors.timeout"
errIncompatibleVersion = "errors.incompatibleversion"
)
// latency metrics go here
const (
latency = "latency"
startLatency = "latency.schedule-to-start"
startWorkflowLatency = "latency.startworkflow"
listOpenWorkflowsLatency = "latency.list-open-workflows"
listWorkflowsLatency = "latency.list-workflows"
listArchivedWorkflowsLatency = "latency.list-archived-workflows"
getWorkflowHistoryLatency = "latency.get-workflow-history"
timerDriftLatency = "latency.timer-drift"
)
// workflowMetricsProfile is the state that's needed to
// record success/failed and latency metrics at the end
// of a workflow
type workflowMetricsProfile struct {
ctx workflow.Context
startTimestamp int64
scope tally.Scope
}
// workflowMetricScope creates and returns a child metric scope with tags
// that identify the current workflow type
func workflowMetricScope(ctx workflow.Context, wfType string) tally.Scope {
parent := workflow.GetMetricsScope(ctx)
return parent.Tagged(map[string]string{"operation": wfType})
}
// recordWorkflowStart emits metrics at the beginning of a workflow function
func recordWorkflowStart(ctx workflow.Context, wfType string, scheduledTimeNanos int64) *workflowMetricsProfile {
now := workflow.Now(ctx).UnixNano()
scope := workflowMetricScope(ctx, wfType)
elapsed := maxInt64(0, now-scheduledTimeNanos)
scope.Timer(startLatency).Record(time.Duration(elapsed))
scope.Counter(startedCount).Inc(1)
return &workflowMetricsProfile{
ctx: ctx,
startTimestamp: now,
scope: scope,
}
}
// recordWorkflowEnd emits metrics at the end of a workflow function
func recordWorkflowEnd(scope tally.Scope, elapsed time.Duration, err error) error {
scope.Timer(latency).Record(elapsed)
if err == nil {
scope.Counter(successCount).Inc(1)
return err
}
scope.Counter(failedCount).Inc(1)
if _, ok := err.(*temporal.TimeoutError); ok {
scope.Counter(errTimeoutCount).Inc(1)
}
return err
}
// end records the elapsed time and reports the latency,
// success, failed counts to m3
func (profile *workflowMetricsProfile) end(err error) error {
now := workflow.Now(profile.ctx).UnixNano()
elapsed := time.Duration(now - profile.startTimestamp)
return recordWorkflowEnd(profile.scope, elapsed, err)
}
// recordActivityStart emits metrics at the beginning of an activity function
func recordActivityStart(
scope tally.Scope, activityType string, scheduledTimeNanos int64) (tally.Scope, tally.Stopwatch) {
scope = scope.Tagged(map[string]string{"operation": activityType})
elapsed := maxInt64(0, time.Now().UnixNano()-scheduledTimeNanos)
scope.Timer(startLatency).Record(time.Duration(elapsed))
scope.Counter(startedCount).Inc(1)
sw := scope.Timer(latency).Start()
return scope, sw
}
// recordActivityEnd emits metrics at the end of an activity function
func recordActivityEnd(scope tally.Scope, sw tally.Stopwatch, err error) {
sw.Stop()
if err != nil {
scope.Counter(failedCount).Inc(1)
return
}
scope.Counter(successCount).Inc(1)
}
canary/pprof.go
-17
View File
@@ -1,17 +0,0 @@
package canary
import (
"fmt"
"log"
"net/http"
_ "net/http/pprof"
)
func StartPProf(port int) {
go func() {
err := http.ListenAndServe(fmt.Sprintf("localhost:%d", port), nil)
if err != nil {
log.Printf("Unable to start profiler on port %d: %v", port, err)
}
}()
}
canary/query.go
-116
View File
@@ -1,116 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
queryType = "QueryWorkflow.QueryType"
queryStatusStarted = "QueryWorkflow.Status.Started"
queryStatusCloseToFinished = "QueryWorkflow.Status.CloseToFinished"
queryStatusQueryHandlerFailed = "QueryWorkflow.Status.QueryHandlerFailed"
)
func registerQuery(r registrar) {
registerWorkflow(r, queryWorkflow, wfTypeQuery)
registerActivity(r, queryActivity, activityTypeQuery1)
registerActivity(r, queryActivity, activityTypeQuery2)
}
// queryWorkflow is the workflow implementation to test for querying workflow status
func queryWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
status := queryStatusStarted
err := workflow.SetQueryHandler(ctx, queryType, func() (string, error) {
return status, nil
})
if err != nil {
status = queryStatusQueryHandlerFailed
return err
}
profile, err := beginWorkflow(ctx, wfTypeQuery, scheduledTimeNanos)
if err != nil {
return err
}
execution := workflow.GetInfo(ctx).WorkflowExecution
var queryStatus string
now := workflow.Now(ctx).UnixNano()
activityCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
err = workflow.ExecuteActivity(activityCtx, activityTypeQuery1, now, execution).Get(activityCtx, &queryStatus)
if err != nil {
workflow.GetLogger(ctx).Info("queryWorkflow failed", zap.Error(err))
return profile.end(err)
}
if queryStatus != status {
err := fmt.Errorf("queryWorkflow status expected %s, but is %s", status, queryStatus)
workflow.GetLogger(ctx).Info("queryWorkflow failed, status is not expected", zap.Error(err))
return profile.end(err)
}
status = queryStatusCloseToFinished
activityCtx = workflow.WithActivityOptions(ctx, newActivityOptions())
err = workflow.ExecuteActivity(activityCtx, activityTypeQuery2, now, execution).Get(activityCtx, &queryStatus)
if err != nil {
workflow.GetLogger(ctx).Info("queryWorkflow failed", zap.Error(err))
return profile.end(err)
}
if queryStatus != status {
err := fmt.Errorf("queryWorkflow status expected %s, but is %s", status, queryStatus)
workflow.GetLogger(ctx).Info("queryWorkflow failed, status is not expected", zap.Error(err))
return profile.end(err)
}
return profile.end(nil)
}
// queryActivity is the activity implementation for querying workflow status
func queryActivity(ctx context.Context, scheduledTimeNanos int64, execution workflow.Execution) (string, error) {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeConcurrentExec, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
encodedValue, err := client.QueryWorkflow(context.Background(), execution.ID, execution.RunID, queryType)
if err != nil {
return "", err
}
var result string
err = encodedValue.Get(&result)
return result, err
}
canary/reset.go
-253
View File
@@ -1,253 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"go.temporal.io/temporal"
commonpb "go.temporal.io/temporal-proto/common/v1"
enumspb "go.temporal.io/temporal-proto/enums/v1"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
)
const (
smallWait = 1
bigWait = 30
signalToTrigger = "signalToTrigger"
signalBeforeReset = "signalBeforeReset"
signalAfterReset = "signalAfterReset"
)
func registerReset(r registrar) {
registerWorkflow(r, resetWorkflow, wfTypeReset)
registerWorkflow(r, resetBaseWorkflow, wfTypeResetBase)
registerActivity(r, triggerResetActivity, activityTypeTriggerReset)
registerActivity(r, verifyResetActivity, activityTypeVerifyReset)
registerActivity(r, resetBaseActivity, activityTypeResetBase)
}
func resetWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeReset, scheduledTimeNanos)
if err != nil {
return err
}
info := workflow.GetInfo(ctx)
cwo := newChildWorkflowOptions(namespace, wfTypeResetBase+"-"+info.WorkflowExecution.RunID)
baseCtx := workflow.WithChildOptions(ctx, cwo)
baseFuture := workflow.ExecuteChildWorkflow(baseCtx, wfTypeResetBase, workflow.Now(ctx).UnixNano(), info.WorkflowExecution.ID, info.WorkflowExecution.RunID, namespace)
var baseWE workflow.Execution
if err := baseFuture.GetChildWorkflowExecution().Get(baseCtx, &baseWE); err != nil {
return profile.end(err)
}
signalFuture1 := baseFuture.SignalChildWorkflow(baseCtx, signalBeforeReset, "signalValue")
err = signalFuture1.Get(baseCtx, nil)
if err != nil {
return profile.end(err)
}
// use signal to wait for baseWF to get reach reset point
var value string
signalCh := workflow.GetSignalChannel(ctx, signalToTrigger)
signalCh.Receive(ctx, &value)
signalFuture2 := baseFuture.SignalChildWorkflow(baseCtx, signalAfterReset, "signalValue")
err = signalFuture2.Get(baseCtx, nil)
if err != nil {
return profile.end(err)
}
expiration := time.Duration(info.WorkflowRunTimeoutSeconds) * time.Second
activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
TaskQueue: taskQueueName,
ScheduleToStartTimeout: expiration,
StartToCloseTimeout: expiration,
})
var newWE workflow.Execution
err = workflow.ExecuteActivity(activityCtx, activityTypeTriggerReset, namespace, baseWE).Get(activityCtx, &newWE)
if err != nil {
return profile.end(err)
}
if err := workflow.Sleep(ctx, time.Duration(bigWait*2)*time.Second); err != nil {
return profile.end(err)
}
err = workflow.ExecuteActivity(activityCtx, activityTypeVerifyReset, namespace, newWE).Get(activityCtx, nil)
return profile.end(err)
}
func resetBaseWorkflow(ctx workflow.Context, scheduledTimeNanos int64, parentID, parentRunID, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeResetBase, scheduledTimeNanos)
if err != nil {
return err
}
info := workflow.GetInfo(ctx)
expiration := time.Duration(info.WorkflowRunTimeoutSeconds) * time.Second
retryPolicy := &temporal.RetryPolicy{
InitialInterval: time.Second * 5,
BackoffCoefficient: 1,
MaximumInterval: time.Second * 5,
MaximumAttempts: 5,
}
activityCtxWithRetry := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
TaskQueue: taskQueueName,
ScheduleToCloseTimeout: expiration,
ScheduleToStartTimeout: expiration,
StartToCloseTimeout: expiration,
RetryPolicy: retryPolicy,
})
activityCtxNoRetry := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
TaskQueue: taskQueueName,
ScheduleToStartTimeout: expiration,
StartToCloseTimeout: expiration,
})
var f workflow.Future
futures1 := []workflow.Future{}
futures2 := []workflow.Future{}
// completed before reset
f = workflow.ExecuteActivity(activityCtxWithRetry, activityTypeResetBase, smallWait)
futures1 = append(futures1, f)
// started before reset
f = workflow.ExecuteActivity(activityCtxNoRetry, activityTypeResetBase, bigWait)
futures2 = append(futures2, f)
// not started before reset(because it uses retry)
f = workflow.ExecuteActivity(activityCtxWithRetry, activityTypeResetBase, bigWait)
futures2 = append(futures2, f)
// fired before reset
f = workflow.NewTimer(ctx, time.Duration(smallWait*2)*time.Second)
futures1 = append(futures1, f)
// fired after reset
f = workflow.NewTimer(ctx, time.Duration(bigWait)*time.Second)
futures2 = append(futures2, f)
// wait until first set of futures1 are done: 1 act, 1 timer
for _, future := range futures1 {
if err := future.Get(ctx, nil); err != nil {
return profile.end(err)
}
}
signalFuture := workflow.SignalExternalWorkflow(ctx, parentID, parentRunID, signalToTrigger, "signalValue")
err = signalFuture.Get(ctx, nil)
if err != nil {
return profile.end(err)
}
var value string
signalCh := workflow.GetSignalChannel(ctx, signalBeforeReset)
signalCh.Receive(ctx, &value)
signalCh = workflow.GetSignalChannel(ctx, signalAfterReset)
signalCh.Receive(ctx, &value)
return profile.end(err)
}
func triggerResetActivity(ctx context.Context, namespace string, baseWE workflow.Execution) (workflow.Execution, error) {
svClient := getActivityContext(ctx).cadence.Service
reason := "reset canary"
client := getActivityContext(ctx).cadence
scope := activity.GetMetricsScope(ctx)
resetEventID := int64(0)
seenTrigger := false
// reset to last decisionCompleted before signalToTrigger was sent
// Since we are in the trigger activity, baseWF must have reached there
events, err := getMyHistory(client, baseWE, scope)
if err != nil {
return workflow.Execution{}, err
}
for _, event := range events {
if event.GetEventType() == enumspb.EVENT_TYPE_DECISION_TASK_COMPLETED {
resetEventID = event.GetEventId()
}
if event.GetEventType() == enumspb.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED {
seenTrigger = true
break
}
}
if resetEventID == 0 || !seenTrigger {
return workflow.Execution{}, fmt.Errorf("something went wrong...base workflow has not reach reset point, %v, %v", resetEventID, seenTrigger)
}
req := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: namespace,
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: baseWE.ID,
RunId: baseWE.RunID,
},
Reason: reason,
DecisionFinishEventId: resetEventID,
RequestId: baseWE.RunID,
}
resp, err := svClient.ResetWorkflowExecution(ctx, req)
if err != nil {
return workflow.Execution{}, err
}
baseWE.RunID = resp.RunId
return baseWE, nil
}
func verifyResetActivity(ctx context.Context, namespace string, newWE workflow.Execution) error {
svClient := getActivityContext(ctx).cadence.Service
resp, err := svClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
Namespace: namespace,
Execution: &commonpb.WorkflowExecution{
WorkflowId: newWE.ID,
RunId: newWE.RunID,
},
})
if err != nil {
return err
}
if resp.WorkflowExecutionInfo.GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING || resp.WorkflowExecutionInfo.GetStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
return fmt.Errorf("new execution triggered by reset is not completed")
}
return nil
}
func resetBaseActivity(ctx context.Context, waitSecs int64) error {
time.Sleep(time.Second * time.Duration(waitSecs))
return nil
}
canary/retry.go
-135
View File
@@ -1,135 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"time"
"go.temporal.io/temporal"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
var (
errRetryableActivityError = errors.New("Retry me")
errUnexpectedProgress = errors.New("Unexpected progress")
errUnexpectedResult = errors.New("Unexpected result")
)
func registerRetry(r registrar) {
registerWorkflow(r, retryWorkflow, wfTypeRetry)
registerActivity(r, retryOnTimeoutActivity, activityTypeRetryOnTimeout)
registerActivity(r, retryOnFailureActivity, activityTypeRetryOnFailure)
}
func retryWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeRetry, scheduledTimeNanos)
if err != nil {
return err
}
info := workflow.GetInfo(ctx)
now := workflow.Now(ctx).UnixNano()
expiration := time.Duration(info.WorkflowRunTimeoutSeconds) * time.Second
retryPolicy := &temporal.RetryPolicy{
InitialInterval: time.Second * 5,
BackoffCoefficient: 1,
MaximumInterval: time.Second * 5,
MaximumAttempts: 5,
}
activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
TaskQueue: taskQueueName,
ScheduleToCloseTimeout: expiration,
HeartbeatTimeout: 5 * time.Second,
RetryPolicy: retryPolicy,
})
f1 := workflow.ExecuteActivity(activityCtx, activityTypeRetryOnTimeout, now)
f2 := workflow.ExecuteActivity(activityCtx, activityTypeRetryOnFailure, now)
var progress int
err = f1.Get(ctx, &progress)
if err != nil {
workflow.GetLogger(ctx).Error("retryWorkflow failed", zap.Error(err))
return profile.end(err)
}
if progress < 200 {
workflow.GetLogger(ctx).Error("Unexpected activity progress.", zap.Int("Progress", progress))
return profile.end(errUnexpectedProgress)
}
var result int32
err = f2.Get(ctx, &result)
if err != nil {
workflow.GetLogger(ctx).Error("retryWorkflow failed", zap.Error(err))
return profile.end(err)
}
if result < 3 {
workflow.GetLogger(ctx).Error("Unexpected activity result.", zap.Int32("Result", result))
return profile.end(errUnexpectedResult)
}
return profile.end(nil)
}
func retryOnTimeoutActivity(ctx context.Context, scheduledTimeNanos int64) (int, error) {
var err error
progress := 0
if activity.HasHeartbeatDetails(ctx) {
err = activity.GetHeartbeatDetails(ctx, &progress)
if err != nil {
activity.GetLogger(ctx).Error("GetProgress failed.", zap.Error(err))
return 0, err
}
}
info := activity.GetInfo(ctx)
if info.Attempt < 3 {
activity.RecordHeartbeat(ctx, info.Attempt*100)
time.Sleep(2 * info.HeartbeatTimeout)
// Currently we have a server bug which accepts completion from different attempt
// For now fail the activity
return 0, errRetryableActivityError
}
return progress, nil
}
func retryOnFailureActivity(ctx context.Context, scheduledTimeNanos int64) (int32, error) {
info := activity.GetInfo(ctx)
if info.Attempt < 3 {
return 0, errRetryableActivityError
}
return info.Attempt, nil
}
canary/runner.go
-143
View File
@@ -1,143 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"fmt"
"os"
"os/signal"
"sync"
"syscall"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.uber.org/zap"
"github.com/temporalio/temporal/common/log/loggerimpl"
"github.com/temporalio/temporal/common/rpc"
)
type canaryRunner struct {
*RuntimeContext
config *Canary
}
// NewCanaryRunner creates and returns a runnable which spins
// up a set of canaries based on supplied config
func NewCanaryRunner(cfg *Config) (Runnable, error) {
logger := cfg.Log.NewZapLogger()
metricsScope := cfg.Metrics.NewScope(loggerimpl.NewLogger(logger))
if cfg.Cadence.HostNameAndPort == "" {
cfg.Cadence.HostNameAndPort = ServiceHostPort
}
connection, err := rpc.Dial(cfg.Cadence.HostNameAndPort, nil)
if err != nil {
return nil, fmt.Errorf("failed to create connection: %v", err)
}
runtimeContext := NewRuntimeContext(
logger,
metricsScope,
cfg.Cadence.HostNameAndPort,
workflowservice.NewWorkflowServiceClient(connection),
)
return &canaryRunner{
RuntimeContext: runtimeContext,
config: &cfg.Canary,
}, nil
}
// Run runs the canaries
func (r *canaryRunner) Run() error {
r.metrics.Counter("restarts").Inc(1)
if len(r.config.Excludes) != 0 {
updateSanityChildWFList(r.config.Excludes)
}
var wg sync.WaitGroup
for _, d := range r.config.Namespaces {
canary, err := newCanary(d, r.RuntimeContext)
if err != nil {
return err
}
r.logger.Info("starting canary", zap.String("namespace", d))
r.execute(canary, &wg)
}
awaitWaitGroup(&wg)
return nil
}
func awaitWaitGroup(wg *sync.WaitGroup) {
doneC := make(chan struct{})
go func() {
wg.Wait()
close(doneC)
}()
select {
case <-doneC:
return
case <-getKillSignal():
return
}
}
func getKillSignal() <-chan os.Signal {
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
return c
}
func (r *canaryRunner) execute(task Runnable, wg *sync.WaitGroup) {
wg.Add(1)
go func() {
task.Run()
wg.Done()
}()
}
func updateSanityChildWFList(excludes []string) {
var temp []string
for _, childName := range sanityChildWFList {
if !isStringInList(childName, excludes) {
temp = append(temp, childName)
}
}
sanityChildWFList = temp
}
func isStringInList(str string, list []string) bool {
for _, l := range list {
if l == str {
return true
}
}
return false
}
canary/sanity.go
-130
View File
@@ -1,130 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
// sanityChildWFList is the list of tests / child workflows invoked by the sanity canary
var sanityChildWFList = []string{
wfTypeEcho,
wfTypeSignal,
wfTypeVisibility,
wfTypeSearchAttributes,
wfTypeConcurrentExec,
wfTypeQuery,
wfTypeTimeout,
wfTypeLocalActivity,
wfTypeCancellation,
wfTypeRetry,
wfTypeReset,
//wfTypeHistoryArchival,
//wfTypeVisibilityArchival,
wfTypeBatch,
}
func registerSanity(r registrar) {
registerWorkflow(r, sanityWorkflow, wfTypeSanity)
}
// sanityWorkflow represents a canary implementation that tests the
// sanity of a cadence cluster - it performs probes / tests that
// exercises the frontend APIs and the basic functionality of the
// client library - each probe / test MUST be implemented as a
// child workflow of this workflow
func sanityWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
var err error
profile, err := beginWorkflow(ctx, wfTypeSanity, scheduledTimeNanos)
if err != nil {
return err
}
childNames, err := getChildWorkflowNames(ctx)
if err != nil {
return profile.end(err)
}
selector, resultC := forkChildWorkflows(ctx, namespace, childNames)
err = joinChildWorkflows(ctx, childNames, selector, resultC)
if err != nil {
workflow.GetLogger(ctx).Error("sanity workflow failed", zap.Error(err))
return profile.end(err)
}
workflow.GetLogger(ctx).Info("sanity workflow finished successfully")
return profile.end(err)
}
// forkChildWorkflows spawns child workflows with the given names
// this method assumes that all child workflows have the same method signature
func forkChildWorkflows(ctx workflow.Context, namespace string, names []string) (workflow.Selector, workflow.Channel) {
now := workflow.Now(ctx).UnixNano()
selector := workflow.NewSelector(ctx)
resultC := workflow.NewBufferedChannel(ctx, len(names))
myID := workflow.GetInfo(ctx).WorkflowExecution.ID
for _, childName := range names {
cwo := newChildWorkflowOptions(namespace, concat(myID, childName))
childCtx := workflow.WithChildOptions(ctx, cwo)
future := workflow.ExecuteChildWorkflow(childCtx, childName, now, namespace)
selector.AddFuture(future, func(f workflow.Future) {
if err := f.Get(ctx, nil); err != nil {
workflow.GetLogger(ctx).Error("child workflow failed", zap.Error(err))
resultC.Send(ctx, err)
return
}
resultC.Send(ctx, nil)
})
}
return selector, resultC
}
// joinChildWorkflows waits for all the given child workflows to complete
// returns error even if atleast one of the child workflow returns an error
func joinChildWorkflows(ctx workflow.Context, names []string, selector workflow.Selector, resultC workflow.Channel) error {
var err error
for i := 0; i < len(names); i++ {
selector.Select(ctx)
var err1 error
resultC.Receive(ctx, &err1)
if err1 != nil {
err = err1
}
}
return err
}
// getChildWorkflowNames exist mainly to make sure replays result in
// deterministic behavior for the workflow - the set of child workflows
// to be spawned are recorded in history as a side effect and result
// from the side effect is returned as a result
func getChildWorkflowNames(ctx workflow.Context) ([]string, error) {
var names []string
err := workflow.SideEffect(ctx, func(workflow.Context) interface{} { return sanityChildWFList }).Get(&names)
return names, err
}
canary/searchAttributes.go
-132
View File
@@ -1,132 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"github.com/uber-go/tally"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
timeToWaitVisibilityDataOnESInDuration = 2 * time.Second
)
func registerSearchAttributes(r registrar) {
registerWorkflow(r, searchAttributesWorkflow, wfTypeSearchAttributes)
registerActivity(r, searchAttributesActivity, activityTypeSearchAttributes)
}
// searchAttributesWorkflow tests the search attributes apis
func searchAttributesWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
var err error
profile, err := beginWorkflow(ctx, wfTypeSearchAttributes, scheduledTimeNanos)
if err != nil {
return err
}
attr := map[string]interface{}{
"CustomKeywordField": "canaryTest",
}
err = workflow.UpsertSearchAttributes(ctx, attr)
if err != nil {
workflow.GetLogger(ctx).Error("searchAttributes test failed", zap.Error(err))
return profile.end(err)
}
execInfo := workflow.GetInfo(ctx).WorkflowExecution
aCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
// wait for visibility on ES
if err := workflow.Sleep(ctx, timeToWaitVisibilityDataOnESInDuration); err != nil {
return profile.end(err)
}
now := workflow.Now(ctx).UnixNano()
err = workflow.ExecuteActivity(aCtx, activityTypeSearchAttributes, now, execInfo).Get(ctx, nil)
if err != nil {
workflow.GetLogger(ctx).Error("searchAttributes test failed", zap.Error(err))
return profile.end(err)
}
return profile.end(nil)
}
// searchAttributesActivity exercises the visibility apis
func searchAttributesActivity(ctx context.Context, scheduledTimeNanos int64, parentInfo workflow.Execution) error {
var err error
scope := activity.GetMetricsScope(ctx)
scope, sw := recordActivityStart(scope, activityTypeSearchAttributes, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
if err := listWorkflow(client, parentInfo.ID, scope); err != nil {
return err
}
if _, err := getMyHistory(client, parentInfo, scope); err != nil {
return err
}
return err
}
func listWorkflow(client cadenceClient, wfID string, scope tally.Scope) error {
pageSz := int32(1)
startTime := time.Now().UnixNano() - int64(timeSkewToleranceDuration)
endTime := time.Now().UnixNano() + int64(timeSkewToleranceDuration)
queryStr := fmt.Sprintf("WorkflowId = '%s' and CustomKeywordField = '%s' and StartTime between %d and %d",
wfID, "canaryTest", startTime, endTime)
request := &workflowservice.ListWorkflowExecutionsRequest{
PageSize: pageSz,
Query: queryStr,
}
scope.Counter(listWorkflowsCount).Inc(1)
sw := scope.Timer(listWorkflowsLatency).Start()
resp, err := client.ListWorkflow(context.Background(), request)
sw.Stop()
if err != nil {
scope.Counter(listWorkflowsFailureCount).Inc(1)
return err
}
if len(resp.Executions) != 1 {
scope.Counter(listWorkflowsFailureCount).Inc(1)
err := fmt.Errorf("listWorkflow returned %d executions, expected=1", len(resp.Executions))
return err
}
id := resp.Executions[0].Execution.GetWorkflowId()
if id != wfID {
scope.Counter(listWorkflowsFailureCount).Inc(1)
err := fmt.Errorf("listWorkflow returned wrong workflow id %v", id)
return err
}
return nil
}
canary/signal.go
-164
View File
@@ -1,164 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"fmt"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
signalName = "signal-name"
signalValue = "canary.signal"
)
func registerSignal(r registrar) {
registerWorkflow(r, signalWorkflow, wfTypeSignal)
registerWorkflow(r, signalExternalWorkflow, wfTypeSignalExternal)
registerActivity(r, signalActivity, activityTypeSignal)
}
// signalWorkflow is the workflow implementation to test SignalWorkflowExecution API
// it simply spins up an activity which sends a signal to the parent
func signalWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
var err error
profile, err := beginWorkflow(ctx, wfTypeSignal, scheduledTimeNanos)
if err != nil {
return err
}
execInfo := workflow.GetInfo(ctx).WorkflowExecution
sigName := fmt.Sprintf("sig.%v", execInfo.RunID)
sigCh := workflow.GetSignalChannel(ctx, sigName)
aCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
err = workflow.ExecuteActivity(aCtx, activityTypeSignal, workflow.Now(ctx).UnixNano(), execInfo, sigName).Get(ctx, nil)
if err != nil {
workflow.GetLogger(ctx).Error("signal activity failed", zap.Error(err))
return profile.end(err)
}
var sigValue string
sigCh.Receive(ctx, &sigValue)
if sigValue != signalValue {
workflow.GetLogger(ctx).Error("wrong signal value received", zap.String("value", sigValue))
return profile.end(errors.New("invalid signal value"))
}
sigCh.Receive(ctx, &sigValue)
if sigValue != signalValue {
workflow.GetLogger(ctx).Error("wrong signal value received", zap.String("value", sigValue))
return profile.end(errors.New("invalid signal value"))
}
cwo := newChildWorkflowOptions(namespace, wfTypeSignalExternal+"-child")
childCtx := workflow.WithChildOptions(ctx, cwo)
childFuture := workflow.ExecuteChildWorkflow(childCtx, wfTypeSignalExternal, workflow.Now(ctx).UnixNano(), sleepDuration)
signalFuture := childFuture.SignalChildWorkflow(childCtx, signalName, signalValue)
err = signalFuture.Get(childCtx, nil)
if err != nil {
return profile.end(err)
}
decisionSignalTestFn := func(useRunID bool) error {
workflowIDSuffix := "-with-run-ID"
if !useRunID {
workflowIDSuffix = "-without-run-ID"
}
cwo = newChildWorkflowOptions(namespace, wfTypeSignalExternal+workflowIDSuffix)
childCtx = workflow.WithChildOptions(ctx, cwo)
childFuture = workflow.ExecuteChildWorkflow(childCtx, wfTypeSignalExternal, workflow.Now(ctx).UnixNano(), sleepDuration)
childExecution := &workflow.Execution{}
err = childFuture.GetChildWorkflowExecution().Get(childCtx, childExecution)
if err != nil {
return err
}
if !useRunID {
childExecution.RunID = ""
}
signalFuture = workflow.SignalExternalWorkflow(ctx, childExecution.ID, childExecution.RunID, signalName, signalValue)
err = signalFuture.Get(ctx, nil)
if err != nil {
workflow.GetLogger(ctx).Info("signalWorkflow failed: fail to send siganl to child workflow", zap.Error(err))
return err
}
err = childFuture.Get(childCtx, nil)
if err != nil {
workflow.GetLogger(ctx).Info("signalWorkflow failed: child workflow return err", zap.Error(err))
return err
}
return nil
}
err = decisionSignalTestFn(false)
if err != nil {
return profile.end(err)
}
err = decisionSignalTestFn(true)
return profile.end(err)
}
// signalActivity sends a signal to the parent workflow
// that created this activity
func signalActivity(ctx context.Context, scheduledTimeNanos int64, execInfo workflow.Execution, signalName string) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeSignal, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
err = client.SignalWorkflow(context.Background(), execInfo.ID, execInfo.RunID, signalName, signalValue)
if err != nil {
return err
}
err = client.SignalWorkflow(context.Background(), execInfo.ID, "", signalName, signalValue)
if err != nil {
return err
}
return nil
}
// signalExternalWorkflow receive a signal from the parent workflow
func signalExternalWorkflow(ctx workflow.Context, scheduledTimeNanos int64) error {
profile, err := beginWorkflow(ctx, wfTypeSignalExternal, scheduledTimeNanos)
if err != nil {
return profile.end(err)
}
var value string
signalCh := workflow.GetSignalChannel(ctx, signalName)
signalCh.Receive(ctx, &value)
if value != signalValue {
workflow.GetLogger(ctx).Error("wrong signal value received", zap.String("value", value))
return profile.end(errors.New("invalid signal value"))
}
return profile.end(nil)
}
canary/timeout.go
-94
View File
@@ -1,94 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"fmt"
"time"
"go.temporal.io/temporal"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
activityDelay = 4 * time.Second
activityStartToCloseTimeoutThreshold = 1 * time.Second
activityScheduleToStartTimeoutThreshold = 1 * time.Second
activityScheduleToCloseTimeoutThreshold = activityStartToCloseTimeoutThreshold + activityScheduleToStartTimeoutThreshold
)
func registerTimeout(r registrar) {
registerWorkflow(r, timeoutWorkflow, wfTypeTimeout)
registerActivity(r, timeoutActivity, activityTypeTimeout)
}
// timeoutWorkflow is the workflow implementation to test for querying workflow status
func timeoutWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
profile, err := beginWorkflow(ctx, wfTypeConcurrentExec, scheduledTimeNanos)
if err != nil {
return err
}
now := workflow.Now(ctx).UnixNano()
activityOptions := newActivityOptions()
activityOptions.StartToCloseTimeout = activityStartToCloseTimeoutThreshold
activityOptions.ScheduleToStartTimeout = activityScheduleToStartTimeoutThreshold
activityOptions.ScheduleToCloseTimeout = activityScheduleToCloseTimeoutThreshold
activityCtx := workflow.WithActivityOptions(ctx, activityOptions)
activityFuture := workflow.ExecuteActivity(activityCtx, timeoutActivity, now)
activityErr := activityFuture.Get(ctx, nil)
if activityErr != nil {
if _, ok := activityErr.(*temporal.TimeoutError); !ok {
workflow.GetLogger(ctx).Info("activity timeout failed", zap.Error(activityErr))
} else {
activityErr = nil
}
} else {
activityErr = fmt.Errorf("activity timeout error expected, but no error received")
workflow.GetLogger(ctx).Info("activity timeout failed", zap.Error(activityErr))
}
return profile.end(activityErr)
}
// timeoutActivity is the activity implementation for timeout test
func timeoutActivity(ctx context.Context, scheduledTimeNanos int64) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeConcurrentExec, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
timer := time.NewTimer(activityDelay)
select {
case <-timer.C:
}
timer.Stop()
return nil
}
canary/visibility.go
-152
View File
@@ -1,152 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"fmt"
"time"
"github.com/uber-go/tally"
enumspb "go.temporal.io/temporal-proto/enums/v1"
filterpb "go.temporal.io/temporal-proto/filter/v1"
historypb "go.temporal.io/temporal-proto/history/v1"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
const (
timeSkewToleranceDuration = 5 * time.Minute
)
func registerVisibility(r registrar) {
registerWorkflow(r, visibilityWorkflow, wfTypeVisibility)
registerActivity(r, visibilityActivity, activityTypeVisibility)
}
// visibilityWorkflow tests the visibility apis
func visibilityWorkflow(ctx workflow.Context, scheduledTimeNanos int64, namespace string) error {
var err error
profile, err := beginWorkflow(ctx, wfTypeVisibility, scheduledTimeNanos)
if err != nil {
return err
}
execInfo := workflow.GetInfo(ctx).WorkflowExecution
aCtx := workflow.WithActivityOptions(ctx, newActivityOptions())
// wait for visibility on ES {}
if err := workflow.Sleep(ctx, 2*time.Second); err != nil {
return profile.end(err)
}
now := workflow.Now(ctx).UnixNano()
err = workflow.ExecuteActivity(aCtx, activityTypeVisibility, now, execInfo).Get(ctx, nil)
if err != nil {
workflow.GetLogger(ctx).Error("visibility test failed", zap.Error(err))
return profile.end(err)
}
return profile.end(nil)
}
// visibilityActivity exercises the visibility apis
func visibilityActivity(ctx context.Context, scheduledTimeNanos int64, parentInfo workflow.Execution) error {
var err error
scope := activity.GetMetricsScope(ctx)
scope, sw := recordActivityStart(scope, activityTypeVisibility, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityContext(ctx).cadence
if err := listMyWorkflow(client, parentInfo.ID, scope); err != nil {
return err
}
if _, err := getMyHistory(client, parentInfo, scope); err != nil {
return err
}
return err
}
func listMyWorkflow(client cadenceClient, wfID string, scope tally.Scope) error {
pageSz := int32(1)
startTime := time.Now().UnixNano() - int64(timeSkewToleranceDuration)
endTime := time.Now().UnixNano() + int64(timeSkewToleranceDuration)
request := &workflowservice.ListOpenWorkflowExecutionsRequest{
MaximumPageSize: pageSz,
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{WorkflowId: wfID}},
StartTimeFilter: &filterpb.StartTimeFilter{
EarliestTime: startTime,
LatestTime: endTime,
},
}
scope.Counter(listOpenWorkflowsCount).Inc(1)
sw := scope.Timer(listOpenWorkflowsLatency).Start()
resp, err := client.ListOpenWorkflow(context.Background(), request)
sw.Stop()
if err != nil {
scope.Counter(listOpenWorkflowsFailureCount).Inc(1)
return err
}
if len(resp.Executions) != 1 {
scope.Counter(listOpenWorkflowsFailureCount).Inc(1)
err := fmt.Errorf("listOpenWorkflow returned %d executions, expected=1", len(resp.Executions))
return err
}
id := resp.Executions[0].Execution.GetWorkflowId()
if id != wfID {
scope.Counter(listOpenWorkflowsFailureCount).Inc(1)
err := fmt.Errorf("listOpenWorkflow returned wrong workflow id %v", id)
return err
}
return nil
}
func getMyHistory(client cadenceClient, execInfo workflow.Execution, scope tally.Scope) ([]*historypb.HistoryEvent, error) {
scope.Counter(getWorkflowHistoryCount).Inc(1)
sw := scope.Timer(getWorkflowHistoryLatency).Start()
defer sw.Stop()
var events []*historypb.HistoryEvent
iter := client.GetWorkflowHistory(context.Background(), execInfo.ID, execInfo.RunID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
for iter.HasNext() {
event, err := iter.Next()
if err != nil {
scope.Counter(getWorkflowHistoryFailureCount).Inc(1)
return nil, err
}
events = append(events, event)
}
if len(events) == 0 {
return nil, errors.New("getWorkflowHistory returned history with 0 events")
}
return events, nil
}
canary/visibilityArchival.go
-249
View File
@@ -1,249 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/uber-go/tally"
enumspb "go.temporal.io/temporal-proto/enums/v1"
"go.temporal.io/temporal-proto/serviceerror"
workflowpb "go.temporal.io/temporal-proto/workflow/v1"
"go.temporal.io/temporal-proto/workflowservice/v1"
"go.temporal.io/temporal/activity"
"go.temporal.io/temporal/workflow"
"go.uber.org/zap"
)
type (
visibilityArchivalValidator interface {
shouldRun() bool
getQuery(workflowID, runID, workflowType string, startTime, closeTime time.Time) string
validateExecutions([]*workflowpb.WorkflowExecutionInfo) error
}
filestoreVisibilityArchivalValidator struct {
expectedRunID string
}
)
const (
schemeFilestore = "file"
queryPageSize = 100
)
func registerVisibilityArchival(r registrar) {
registerWorkflow(r, visibilityArchivalWorkflow, wfTypeVisibilityArchival)
registerActivity(r, visibilityArchivalActivity, activityTypeVisibilityArchival)
}
func visibilityArchivalWorkflow(ctx workflow.Context, scheduledTimeNanos int64) error {
profile, err := beginWorkflow(ctx, wfTypeHistoryArchival, scheduledTimeNanos)
if err != nil {
return err
}
if err := workflow.ExecuteActivity(
workflow.WithActivityOptions(ctx, newActivityOptions()),
activityTypeVisibilityArchival,
workflow.Now(ctx).UnixNano(),
).Get(ctx, nil); err != nil {
workflow.GetLogger(ctx).Error("failed to list archived workflows", zap.Error(err))
return profile.end(err)
}
return profile.end(nil)
}
func visibilityArchivalActivity(ctx context.Context, scheduledTimeNanos int64) error {
scope := activity.GetMetricsScope(ctx)
var err error
scope, sw := recordActivityStart(scope, activityTypeVisibilityArchival, scheduledTimeNanos)
defer recordActivityEnd(scope, sw, err)
client := getActivityArchivalContext(ctx).cadence
resp, err := client.Describe(ctx, archivalNamespace)
if err != nil {
return err
}
if resp.Config != nil &&
resp.Config.GetVisibilityArchivalStatus() == enumspb.ARCHIVAL_STATUS_DISABLED {
return errors.New("namespace not configured for visibility archival")
}
visArchivalURI := ""
if resp.Config != nil {
visArchivalURI = resp.Config.GetVisibilityArchivalUri()
}
var validator visibilityArchivalValidator
scheme := getURIScheme(visArchivalURI)
switch scheme {
case schemeFilestore:
validator = newFilestoreVisibilityArchivalValidator()
default:
return fmt.Errorf("unknown visibility archival scheme: %s", scheme)
}
if !validator.shouldRun() {
return nil
}
startTime := time.Now()
execution, err := executeArchivalExeternalWorkflow(ctx, client, startTime.UnixNano())
if err != nil {
return err
}
query := validator.getQuery(
execution.GetWorkflowId(),
execution.GetRunId(),
wfTypeArchivalExternal,
startTime,
time.Now(),
)
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
executions, err := listarchivedWorkflow(
ctx,
scope,
client,
&workflowservice.ListArchivedWorkflowExecutionsRequest{
Namespace: archivalNamespace,
PageSize: queryPageSize,
Query: query,
},
)
if err != nil && isInvalidArgumentError(err) {
return err
}
if err == nil && validator.validateExecutions(executions) == nil {
return nil
}
<-time.After(5 * time.Second)
}
}
func listarchivedWorkflow(
ctx context.Context,
scope tally.Scope,
client cadenceClient,
request *workflowservice.ListArchivedWorkflowExecutionsRequest,
) ([]*workflowpb.WorkflowExecutionInfo, error) {
var executions []*workflowpb.WorkflowExecutionInfo
for {
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
scope.Counter(listArchivedWorkflowCount).Inc(1)
sw := scope.Timer(listArchivedWorkflowsLatency).Start()
response, err := client.ListArchivedWorkflow(ctx, request)
sw.Stop()
if err != nil {
scope.Counter(listArchivedWorkflowFailureCount).Inc(1)
return nil, err
}
if len(response.Executions) != 0 {
executions = append(executions, response.Executions...)
}
if response.NextPageToken == nil {
break
}
request.NextPageToken = response.NextPageToken
}
return executions, nil
}
func newFilestoreVisibilityArchivalValidator() visibilityArchivalValidator {
return &filestoreVisibilityArchivalValidator{}
}
func (v *filestoreVisibilityArchivalValidator) shouldRun() bool {
return true
}
func (v *filestoreVisibilityArchivalValidator) getQuery(
workflowID, runID, workflowType string,
startTime, closeTime time.Time,
) string {
v.expectedRunID = runID
return fmt.Sprintf(
"WorkflowType = '%s' and WorkflowId = '%s' and CloseTime >= %v and CloseTime <= %v",
workflowType,
workflowID,
startTime.UnixNano(),
closeTime.UnixNano(),
)
}
func (v *filestoreVisibilityArchivalValidator) validateExecutions(
executions []*workflowpb.WorkflowExecutionInfo,
) error {
if len(executions) != 1 {
return fmt.Errorf("listarchivedWorkflow returned %d executions, expecting 1", len(executions))
}
runID := executions[0].Execution.GetRunId()
if runID != v.expectedRunID {
return fmt.Errorf("listarchivedWorkflow returned wrong runID %v, expecting %s", runID, v.expectedRunID)
}
return nil
}
func getURIScheme(URI string) string {
if idx := strings.Index(URI, "://"); idx != -1 {
return URI[:idx]
}
return ""
}
func isInvalidArgumentError(
err error,
) bool {
_, ok := err.(*serviceerror.InvalidArgument)
return ok
}
canary/workflow_test.go
-181
View File
@@ -1,181 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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 canary
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/suite"
"go.temporal.io/temporal/mocks"
"go.temporal.io/temporal/testsuite"
"go.temporal.io/temporal/worker"
)
type (
workflowTestSuite struct {
suite.Suite
testsuite.WorkflowTestSuite
env *testsuite.TestWorkflowEnvironment
}
mockCadenceClient struct {
client *mocks.Client
namespaceClient *mocks.NamespaceClient
}
)
func TestWorkflowTestSuite(t *testing.T) {
suite.Run(t, new(workflowTestSuite))
}
func (s *workflowTestSuite) SetupTest() {
s.env = s.NewTestWorkflowEnvironment()
registerHistoryArchival(s.env)
registerBatch(s.env)
registerCancellation(s.env)
registerConcurrentExec(s.env)
registerCron(s.env)
registerEcho(s.env)
registerLocalActivity(s.env)
registerQuery(s.env)
registerReset(s.env)
registerRetry(s.env)
registerSanity(s.env)
registerSearchAttributes(s.env)
registerSignal(s.env)
registerTimeout(s.env)
registerVisibility(s.env)
}
func (s *workflowTestSuite) TearDownTest() {
s.env.AssertExpectations(s.T())
}
func (s *workflowTestSuite) TestConcurrentExecWorkflow() {
s.env.SetWorkerOptions(newTestWorkerOptions(newMockActivityContext(newMockCadenceClient())))
s.env.ExecuteWorkflow(wfTypeConcurrentExec, time.Now().UnixNano(), "")
s.True(s.env.IsWorkflowCompleted())
s.NoError(s.env.GetWorkflowError())
}
func (s *workflowTestSuite) TestQueryWorkflow() {
// DOTO implement this test once client test framework
// SetOnActivityStartedListener is guaranteed to
// be executed and finished before activity started
}
func (s *workflowTestSuite) TestTimeout() {
// DOTO implement this test once client test framework
// can handle timeout error correctly
}
func (s *workflowTestSuite) TestSignalWorkflow() {
mockClient := newMockCadenceClient()
mockClient.client.On("SignalWorkflow",
mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
arg3 := args.Get(3).(string)
arg4 := args.Get(4).(string)
s.env.SignalWorkflow(arg3, arg4)
}).Return(nil)
s.env.SetWorkerOptions(newTestWorkerOptions(newMockActivityContext(mockClient)))
s.env.ExecuteWorkflow(wfTypeSignal, time.Now().UnixNano(), "")
s.True(s.env.IsWorkflowCompleted())
s.NoError(s.env.GetWorkflowError())
}
func (s *workflowTestSuite) TestEchoWorkflow() {
s.env.SetWorkerOptions(newTestWorkerOptions(newMockActivityContext(newMockCadenceClient())))
s.env.ExecuteWorkflow(wfTypeEcho, time.Now().UnixNano(), "")
s.True(s.env.IsWorkflowCompleted())
s.NoError(s.env.GetWorkflowError())
}
func (s *workflowTestSuite) TestVisibilityWorkflow() {
// uncomment below when race condition on s.env.SetOnActivityStartedListener is resolved
// mockClient := newMockCadenceClient()
// // setup the mock for visibility apis after the activity is invoked, because
// // we need the workflow id and run id to construct a response for the mock
// s.env.SetOnActivityStartedListener(func(activityInfo *temporal.ActivityInfo, ctx context.Context, args temporal.EncodedValues) {
// if activityInfo.ActivityType.Name != activityTypeVisibility {
// return
// }
// resp := newMockOpenWorkflowResponse(activityInfo.WorkflowExecution.ID, activityInfo.WorkflowExecution.RunID)
// mockClient.client.On("ListOpenWorkflow", mock.Anything, mock.Anything).Return(resp, nil)
// history := &historypb.History{Events: []*historypb.HistoryEvent{{}}}
// mockClient.client.On("GetWorkflowHistory", mock.Anything, mock.Anything, mock.Anything).Return(history, nil)
// })
// s.env.SetWorkerOptions(newTestWorkerOptions(newMockActivityContext(mockClient)))
// s.env.ExecuteWorkflow(wfTypeVisibility, time.Now().UnixNano(), &ServiceConfig{})
// s.True(s.env.IsWorkflowCompleted())
// s.NoError(s.env.GetWorkflowError())
}
func (s *workflowTestSuite) TestSanityWorkflow() {
oldWFList := sanityChildWFList
sanityChildWFList = []string{wfTypeEcho}
s.env.ExecuteWorkflow(wfTypeSanity, time.Now().UnixNano(), "")
sanityChildWFList = oldWFList
s.True(s.env.IsWorkflowCompleted())
s.NoError(s.env.GetWorkflowError())
}
func (s *workflowTestSuite) TestLocalActivityWorkflow() {
s.env.OnActivity(getConditionData).Return(int32(20), nil).Once()
s.env.ExecuteWorkflow(localActivityWorkfow)
s.True(s.env.IsWorkflowCompleted())
s.NoError(s.env.GetWorkflowError())
var result string
err := s.env.GetWorkflowResult(&result)
s.NoError(err)
s.Equal("data%2 == 0 and data%5 == 0", result)
}
func newTestWorkerOptions(ctx *activityContext) worker.Options {
return worker.Options{
BackgroundActivityContext: context.WithValue(context.Background(), ctxKeyActivityRuntime, ctx),
}
}
func newMockActivityContext(client mockCadenceClient) *activityContext {
return &activityContext{
cadence: cadenceClient{
Client: client.client,
NamespaceClient: client.namespaceClient,
},
}
}
func newMockCadenceClient() mockCadenceClient {
return mockCadenceClient{
client: new(mocks.Client),
namespaceClient: new(mocks.NamespaceClient),
}
}
cmd/canary/main.go
-170
View File
@@ -1,170 +0,0 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 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"
"runtime"
"runtime/pprof"
"strings"
"github.com/urfave/cli"
"github.com/temporalio/temporal/canary"
"github.com/temporalio/temporal/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)
}
if cfg.Canary.PProf.Port != 0 {
canary.StartPProf(cfg.Canary.PProf.Port)
}
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)
}
dumpMemoryProfile(c)
}
func dumpMemoryProfile(c *cli.Context) {
if !c.GlobalIsSet("memprofile") {
return
}
memprofile := c.GlobalString("memprofile")
log.Printf("Writing memory profile to: %s\n", memprofile)
f, err := os.Create(memprofile)
if err != nil {
log.Fatal("could not create memory profile: ", err)
}
defer f.Close() // error handling omitted for example
runtime.GC() // get up-to-date statistics
if err = pprof.WriteHeapProfile(f); err != nil {
log.Fatal("could not write memory profile: ", 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 = "canary"
app.Usage = "Temporal 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,
},
cli.StringFlag{
Name: "memprofile, mp",
Value: "",
Usage: "memprofile",
},
}
app.Commands = []cli.Command{
{
Name: "start",
Usage: "start temporal canary",
Action: func(c *cli.Context) {
startHandler(c)
},
},
}
return app
}
func main() {
app := buildCLI()
app.Run(os.Args)
}
config/canary/base.yaml
-3
View File
@@ -1,3 +0,0 @@
log:
stdout: true
level: info
config/canary/development.yaml
-9
View File
@@ -1,9 +0,0 @@
canary:
namespaces: ["canary"]
excludes: ["workflow.searchAttributes", "workflow.batch", "workflow.visibility", "workflow.cancellation" ]
pprof:
port: 6060
temporal:
service: "frontend"
host: "127.0.0.1:7233"