mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
zeekay
5a601d081b
Full module path rebrand from Temporal to Hanzo Tasks: - go.mod module: github.com/hanzoai/tasks - All internal Go imports updated (1906 files) - cmd/server renamed to cmd/tasksd - Binary name: tasksd (was temporal-server) - Docker images: hanzoai/tasks (was temporalio/server) - README rewritten with Hanzo Tasks branding - LLM.md created with integration notes - Makefile, goreleaser, CI workflows updated - go.temporal.io/sdk and go.temporal.io/api imports preserved (separate repos) Fork of Temporal.io (MIT license).
362 lines
14 KiB
Go
362 lines
14 KiB
Go
// Code generated by protoc-gen-go-chasm. DO NOT EDIT.
|
|
package schedulerpb
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/hanzoai/tasks/client/history"
|
|
"github.com/hanzoai/tasks/common"
|
|
"github.com/hanzoai/tasks/common/backoff"
|
|
"github.com/hanzoai/tasks/common/config"
|
|
"github.com/hanzoai/tasks/common/dynamicconfig"
|
|
"github.com/hanzoai/tasks/common/headers"
|
|
"github.com/hanzoai/tasks/common/log"
|
|
"github.com/hanzoai/tasks/common/membership"
|
|
"github.com/hanzoai/tasks/common/metrics"
|
|
"github.com/hanzoai/tasks/common/primitives"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
// SchedulerServiceLayeredClient is a client for SchedulerService.
|
|
type SchedulerServiceLayeredClient struct {
|
|
metricsHandler metrics.Handler
|
|
numShards int32
|
|
redirector history.Redirector[SchedulerServiceClient]
|
|
retryPolicy backoff.RetryPolicy
|
|
}
|
|
|
|
// NewSchedulerServiceLayeredClient initializes a new SchedulerServiceLayeredClient.
|
|
func NewSchedulerServiceLayeredClient(
|
|
dc *dynamicconfig.Collection,
|
|
rpcFactory common.RPCFactory,
|
|
monitor membership.Monitor,
|
|
config *config.Persistence,
|
|
logger log.Logger,
|
|
metricsHandler metrics.Handler,
|
|
) (SchedulerServiceClient, error) {
|
|
resolver, err := monitor.GetResolver(primitives.HistoryService)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
connections := history.NewConnectionPool(resolver, rpcFactory, NewSchedulerServiceClient)
|
|
var redirector history.Redirector[SchedulerServiceClient]
|
|
if dynamicconfig.HistoryClientOwnershipCachingEnabled.Get(dc)() {
|
|
redirector = history.NewCachingRedirector(
|
|
connections,
|
|
resolver,
|
|
logger,
|
|
dynamicconfig.HistoryClientOwnershipCachingStaleTTL.Get(dc),
|
|
)
|
|
} else {
|
|
redirector = history.NewBasicRedirector(connections, resolver)
|
|
}
|
|
return &SchedulerServiceLayeredClient{
|
|
metricsHandler: metricsHandler,
|
|
redirector: redirector,
|
|
numShards: config.NumHistoryShards,
|
|
retryPolicy: common.CreateHistoryClientRetryPolicy(),
|
|
}, nil
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callCreateScheduleNoRetry(
|
|
ctx context.Context,
|
|
request *CreateScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*CreateScheduleResponse, error) {
|
|
var response *CreateScheduleResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.CreateSchedule"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.CreateSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) CreateSchedule(
|
|
ctx context.Context,
|
|
request *CreateScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*CreateScheduleResponse, error) {
|
|
call := func(ctx context.Context) (*CreateScheduleResponse, error) {
|
|
return c.callCreateScheduleNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callUpdateScheduleNoRetry(
|
|
ctx context.Context,
|
|
request *UpdateScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*UpdateScheduleResponse, error) {
|
|
var response *UpdateScheduleResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.UpdateSchedule"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.UpdateSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) UpdateSchedule(
|
|
ctx context.Context,
|
|
request *UpdateScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*UpdateScheduleResponse, error) {
|
|
call := func(ctx context.Context) (*UpdateScheduleResponse, error) {
|
|
return c.callUpdateScheduleNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callPatchScheduleNoRetry(
|
|
ctx context.Context,
|
|
request *PatchScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*PatchScheduleResponse, error) {
|
|
var response *PatchScheduleResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.PatchSchedule"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.PatchSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) PatchSchedule(
|
|
ctx context.Context,
|
|
request *PatchScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*PatchScheduleResponse, error) {
|
|
call := func(ctx context.Context) (*PatchScheduleResponse, error) {
|
|
return c.callPatchScheduleNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callDeleteScheduleNoRetry(
|
|
ctx context.Context,
|
|
request *DeleteScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*DeleteScheduleResponse, error) {
|
|
var response *DeleteScheduleResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.DeleteSchedule"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.DeleteSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) DeleteSchedule(
|
|
ctx context.Context,
|
|
request *DeleteScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*DeleteScheduleResponse, error) {
|
|
call := func(ctx context.Context) (*DeleteScheduleResponse, error) {
|
|
return c.callDeleteScheduleNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callDescribeScheduleNoRetry(
|
|
ctx context.Context,
|
|
request *DescribeScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*DescribeScheduleResponse, error) {
|
|
var response *DescribeScheduleResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.DescribeSchedule"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.DescribeSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) DescribeSchedule(
|
|
ctx context.Context,
|
|
request *DescribeScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*DescribeScheduleResponse, error) {
|
|
call := func(ctx context.Context) (*DescribeScheduleResponse, error) {
|
|
return c.callDescribeScheduleNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callListScheduleMatchingTimesNoRetry(
|
|
ctx context.Context,
|
|
request *ListScheduleMatchingTimesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*ListScheduleMatchingTimesResponse, error) {
|
|
var response *ListScheduleMatchingTimesResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.ListScheduleMatchingTimes"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.ListScheduleMatchingTimes(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) ListScheduleMatchingTimes(
|
|
ctx context.Context,
|
|
request *ListScheduleMatchingTimesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*ListScheduleMatchingTimesResponse, error) {
|
|
call := func(ctx context.Context) (*ListScheduleMatchingTimesResponse, error) {
|
|
return c.callListScheduleMatchingTimesNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) callCreateFromMigrationStateNoRetry(
|
|
ctx context.Context,
|
|
request *CreateFromMigrationStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*CreateFromMigrationStateResponse, error) {
|
|
var response *CreateFromMigrationStateResponse
|
|
var err error
|
|
startTime := time.Now().UTC()
|
|
// the caller is a namespace, hence the tag below.
|
|
caller := headers.GetCallerInfo(ctx).CallerName
|
|
metricsHandler := c.metricsHandler.WithTags(
|
|
metrics.OperationTag("SchedulerService.CreateFromMigrationState"),
|
|
metrics.NamespaceTag(caller),
|
|
metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
|
|
)
|
|
metrics.ClientRequests.With(metricsHandler).Record(1)
|
|
defer func() {
|
|
if err != nil {
|
|
metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
|
|
}
|
|
metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
|
|
}()
|
|
shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetState().GetSchedulerState().GetScheduleId(), c.numShards)
|
|
op := func(ctx context.Context, client SchedulerServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.CreateFromMigrationState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *SchedulerServiceLayeredClient) CreateFromMigrationState(
|
|
ctx context.Context,
|
|
request *CreateFromMigrationStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*CreateFromMigrationStateResponse, error) {
|
|
call := func(ctx context.Context) (*CreateFromMigrationStateResponse, error) {
|
|
return c.callCreateFromMigrationStateNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|