peridot/vendor/go.temporal.io/sdk/internal/activity.go

// 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 internal

import (
	"context"
	"fmt"
	"time"

	commonpb "go.temporal.io/api/common/v1"
	"go.temporal.io/api/workflowservice/v1"

	"go.temporal.io/sdk/converter"
	"go.temporal.io/sdk/internal/common"
	"go.temporal.io/sdk/internal/common/metrics"
	"go.temporal.io/sdk/log"
)

type (
	// ActivityType identifies a activity type.
	ActivityType struct {
		Name string
	}

	// ActivityInfo contains information about currently executing activity.
	ActivityInfo struct {
		TaskToken         []byte
		WorkflowType      *WorkflowType
		WorkflowNamespace string
		WorkflowExecution WorkflowExecution
		ActivityID        string
		ActivityType      ActivityType
		TaskQueue         string
		HeartbeatTimeout  time.Duration // Maximum time between heartbeats. 0 means no heartbeat needed.
		ScheduledTime     time.Time     // Time of activity scheduled by a workflow
		StartedTime       time.Time     // Time of activity start
		Deadline          time.Time     // Time of activity timeout
		Attempt           int32         // Attempt starts from 1, and increased by 1 for every retry if retry policy is specified.
	}

	// RegisterActivityOptions consists of options for registering an activity
	RegisterActivityOptions struct {
		// When an activity is a function the name is an actual activity type name.
		// When an activity is part of a structure then each member of the structure becomes an activity with
		// this Name as a prefix + activity function name.
		Name                          string
		DisableAlreadyRegisteredCheck bool

		// When registering a struct with activities, skip functions that are not valid activities. If false,
		// registration panics.
		SkipInvalidStructFunctions bool
	}

	// ActivityOptions stores all activity-specific parameters that will be stored inside of a context.
	// The current timeout resolution implementation is in seconds and uses math.Ceil(d.Seconds()) as the duration. But is
	// subjected to change in the future.
	ActivityOptions struct {
		// TaskQueue that the activity needs to be scheduled on.
		// optional: The default task queue with the same name as the workflow task queue.
		TaskQueue string

		// ScheduleToCloseTimeout - Total time that a workflow is willing to wait for Activity to complete.
		// ScheduleToCloseTimeout limits the total time of an Activity's execution including retries
		// 		(use StartToCloseTimeout to limit the time of a single attempt).
		// The zero value of this uses default value.
		// Either this option or StartToClose is required: Defaults to unlimited.
		ScheduleToCloseTimeout time.Duration

		// ScheduleToStartTimeout - Time that the Activity Task can stay in the Task Queue before it is picked up by
		// a Worker. Do not specify this timeout unless using host specific Task Queues for Activity Tasks are being
		// used for routing. In almost all situations that don't involve routing activities to specific hosts it is
		// better to rely on the default value.
		// ScheduleToStartTimeout is always non-retryable. Retrying after this timeout doesn't make sense as it would
		// just put the Activity Task back into the same Task Queue.
		// If ScheduleToClose is not provided then this timeout is required.
		// Optional: Defaults to unlimited.
		ScheduleToStartTimeout time.Duration

		// StartToCloseTimeout - Maximum time of a single Activity execution attempt.
		// Note that the Temporal Server doesn't detect Worker process failures directly. It relies on this timeout
		// to detect that an Activity that didn't complete on time. So this timeout should be as short as the longest
		// possible execution of the Activity body. Potentially long running Activities must specify HeartbeatTimeout
		// and call Activity.RecordHeartbeat(ctx, "my-heartbeat") periodically for timely failure detection.
		// If ScheduleToClose is not provided then this timeout is required: Defaults to the ScheduleToCloseTimeout value.
		StartToCloseTimeout time.Duration

		// HeartbeatTimeout - Heartbeat interval. Activity must call Activity.RecordHeartbeat(ctx, "my-heartbeat")
		// before this interval passes after the last heartbeat or the Activity starts.
		HeartbeatTimeout time.Duration

		// WaitForCancellation - Whether to wait for canceled activity to be completed(
		// activity can be failed, completed, cancel accepted)
		// Optional: default false
		WaitForCancellation bool

		// ActivityID - Business level activity ID, this is not needed for most of the cases if you have
		// to specify this then talk to temporal team. This is something will be done in future.
		// Optional: default empty string
		ActivityID string

		// RetryPolicy specifies how to retry an Activity if an error occurs.
		// More details are available at docs.temporal.io.
		// RetryPolicy is optional. If one is not specified a default RetryPolicy is provided by the server.
		// The default RetryPolicy provided by the server specifies:
		// - InitialInterval of 1 second
		// - BackoffCoefficient of 2.0
		// - MaximumInterval of 100 x InitialInterval
		// - MaximumAttempts of 0 (unlimited)
		// To disable retries set MaximumAttempts to 1.
		// The default RetryPolicy provided by the server can be overridden by the dynamic config.
		RetryPolicy *RetryPolicy
	}

	// LocalActivityOptions stores local activity specific parameters that will be stored inside of a context.
	LocalActivityOptions struct {
		// ScheduleToCloseTimeout - The end to end timeout for the local activity including retries.
		// This field is required.
		ScheduleToCloseTimeout time.Duration

		// StartToCloseTimeout - The timeout for a single execution of the local activity.
		// Optional: defaults to ScheduleToClose
		StartToCloseTimeout time.Duration

		// RetryPolicy specify how to retry activity if error happens.
		// Optional: default is to retry according to the default retry policy up to ScheduleToCloseTimeout
		// with 1sec initial delay between retries and 2x backoff.
		RetryPolicy *RetryPolicy
	}
)

// GetActivityInfo returns information about currently executing activity.
func GetActivityInfo(ctx context.Context) ActivityInfo {
	return getActivityOutboundInterceptor(ctx).GetInfo(ctx)
}

// HasHeartbeatDetails checks if there is heartbeat details from last attempt.
func HasHeartbeatDetails(ctx context.Context) bool {
	return getActivityOutboundInterceptor(ctx).HasHeartbeatDetails(ctx)
}

// GetHeartbeatDetails extract heartbeat details from last failed attempt. This is used in combination with retry policy.
// An activity could be scheduled with an optional retry policy on ActivityOptions. If the activity failed then server
// would attempt to dispatch another activity task to retry according to the retry policy. If there was heartbeat
// details reported by activity from the failed attempt, the details would be delivered along with the activity task for
// retry attempt. Activity could extract the details by GetHeartbeatDetails() and resume from the progress.
func GetHeartbeatDetails(ctx context.Context, d ...interface{}) error {
	return getActivityOutboundInterceptor(ctx).GetHeartbeatDetails(ctx, d...)
}

// GetActivityLogger returns a logger that can be used in activity
func GetActivityLogger(ctx context.Context) log.Logger {
	return getActivityOutboundInterceptor(ctx).GetLogger(ctx)
}

// GetActivityMetricsHandler returns a metrics handler that can be used in activity
func GetActivityMetricsHandler(ctx context.Context) metrics.Handler {
	return getActivityOutboundInterceptor(ctx).GetMetricsHandler(ctx)
}

// GetWorkerStopChannel returns a read-only channel. The closure of this channel indicates the activity worker is stopping.
// When the worker is stopping, it will close this channel and wait until the worker stop timeout finishes. After the timeout
// hit, the worker will cancel the activity context and then exit. The timeout can be defined by worker option: WorkerStopTimeout.
// Use this channel to handle activity graceful exit when the activity worker stops.
func GetWorkerStopChannel(ctx context.Context) <-chan struct{} {
	return getActivityOutboundInterceptor(ctx).GetWorkerStopChannel(ctx)
}

// RecordActivityHeartbeat sends heartbeat for the currently executing activity
// If the activity is either canceled (or) workflow/activity doesn't exist then we would cancel
// the context with error context.Canceled.
//  TODO: we don't have a way to distinguish between the two cases when context is canceled because
//  context doesn't support overriding value of ctx.Error.
//  TODO: Implement automatic heartbeating with cancellation through ctx.
// details - the details that you provided here can be seen in the workflow when it receives TimeoutError, you
// can check error TimeoutType()/Details().
func RecordActivityHeartbeat(ctx context.Context, details ...interface{}) {
	getActivityOutboundInterceptor(ctx).RecordHeartbeat(ctx, details...)
}

// ServiceInvoker abstracts calls to the Temporal service from an activity implementation.
// Implement to unit test activities.
type ServiceInvoker interface {
	// Returns ActivityTaskCanceledError if activity is canceled
	Heartbeat(ctx context.Context, details *commonpb.Payloads, skipBatching bool) error
	Close(ctx context.Context, flushBufferedHeartbeat bool)
	GetClient(options ClientOptions) Client
}

// WithActivityTask adds activity specific information into context.
// Use this method to unit test activity implementations that use context extractor methodshared.
func WithActivityTask(
	ctx context.Context,
	task *workflowservice.PollActivityTaskQueueResponse,
	taskQueue string,
	invoker ServiceInvoker,
	logger log.Logger,
	metricsHandler metrics.Handler,
	dataConverter converter.DataConverter,
	workerStopChannel <-chan struct{},
	contextPropagators []ContextPropagator,
	interceptors []WorkerInterceptor,
) (context.Context, error) {
	var deadline time.Time
	scheduled := common.TimeValue(task.GetScheduledTime())
	started := common.TimeValue(task.GetStartedTime())
	scheduleToCloseTimeout := common.DurationValue(task.GetScheduleToCloseTimeout())
	startToCloseTimeout := common.DurationValue(task.GetStartToCloseTimeout())
	heartbeatTimeout := common.DurationValue(task.GetHeartbeatTimeout())

	startToCloseDeadline := started.Add(startToCloseTimeout)
	if scheduleToCloseTimeout > 0 {
		scheduleToCloseDeadline := scheduled.Add(scheduleToCloseTimeout)
		// Minimum of the two deadlines.
		if scheduleToCloseDeadline.Before(startToCloseDeadline) {
			deadline = scheduleToCloseDeadline
		} else {
			deadline = startToCloseDeadline
		}
	} else {
		deadline = startToCloseDeadline
	}

	logger = log.With(logger,
		tagActivityID, task.ActivityId,
		tagActivityType, task.ActivityType.GetName(),
		tagAttempt, task.Attempt,
		tagWorkflowType, task.WorkflowType.GetName(),
		tagWorkflowID, task.WorkflowExecution.WorkflowId,
		tagRunID, task.WorkflowExecution.RunId,
	)

	return newActivityContext(ctx, interceptors, &activityEnvironment{
		taskToken:      task.TaskToken,
		serviceInvoker: invoker,
		activityType:   ActivityType{Name: task.ActivityType.GetName()},
		activityID:     task.ActivityId,
		workflowExecution: WorkflowExecution{
			RunID: task.WorkflowExecution.RunId,
			ID:    task.WorkflowExecution.WorkflowId},
		logger:           logger,
		metricsHandler:   metricsHandler,
		deadline:         deadline,
		heartbeatTimeout: heartbeatTimeout,
		scheduledTime:    scheduled,
		startedTime:      started,
		taskQueue:        taskQueue,
		dataConverter:    dataConverter,
		attempt:          task.GetAttempt(),
		heartbeatDetails: task.HeartbeatDetails,
		workflowType: &WorkflowType{
			Name: task.WorkflowType.GetName(),
		},
		workflowNamespace:  task.WorkflowNamespace,
		workerStopChannel:  workerStopChannel,
		contextPropagators: contextPropagators,
	})
}

// WithLocalActivityTask adds local activity specific information into context.
func WithLocalActivityTask(
	ctx context.Context,
	task *localActivityTask,
	logger log.Logger,
	metricsHandler metrics.Handler,
	dataConverter converter.DataConverter,
	interceptors []WorkerInterceptor,
) (context.Context, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	workflowTypeLocal := task.params.WorkflowInfo.WorkflowType
	workflowType := task.params.WorkflowInfo.WorkflowType.Name
	activityType := task.params.ActivityType
	logger = log.With(logger,
		tagActivityID, task.activityID,
		tagActivityType, activityType,
		tagAttempt, task.attempt,
		tagWorkflowType, workflowType,
		tagWorkflowID, task.params.WorkflowInfo.WorkflowExecution.ID,
		tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID,
	)
	return newActivityContext(ctx, interceptors, &activityEnvironment{
		workflowType:      &workflowTypeLocal,
		workflowNamespace: task.params.WorkflowInfo.Namespace,
		taskQueue:         task.params.WorkflowInfo.TaskQueueName,
		activityType:      ActivityType{Name: activityType},
		activityID:        fmt.Sprintf("%v", task.activityID),
		workflowExecution: task.params.WorkflowInfo.WorkflowExecution,
		logger:            logger,
		metricsHandler:    metricsHandler,
		isLocalActivity:   true,
		dataConverter:     dataConverter,
		attempt:           task.attempt,
	})
}

func newActivityContext(
	ctx context.Context,
	interceptors []WorkerInterceptor,
	env *activityEnvironment,
) (context.Context, error) {
	ctx = context.WithValue(ctx, activityEnvContextKey, env)

	// Create interceptor with default inbound and outbound values and put on
	// context
	envInterceptor := &activityEnvironmentInterceptor{env: env}
	envInterceptor.inboundInterceptor = envInterceptor
	envInterceptor.outboundInterceptor = envInterceptor
	ctx = context.WithValue(ctx, activityEnvInterceptorContextKey, envInterceptor)
	ctx = context.WithValue(ctx, activityInterceptorContextKey, envInterceptor.outboundInterceptor)

	// Intercept, run init, and put the new outbound interceptor on the context
	for i := len(interceptors) - 1; i >= 0; i-- {
		envInterceptor.inboundInterceptor = interceptors[i].InterceptActivity(ctx, envInterceptor.inboundInterceptor)
	}
	err := envInterceptor.inboundInterceptor.Init(envInterceptor)
	if err != nil {
		return nil, err
	}
	ctx = context.WithValue(ctx, activityInterceptorContextKey, envInterceptor.outboundInterceptor)

	return ctx, nil
}