mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Lina JodoinandGitHub
19fc1ad532
## What changed? - Frontend workflow_handler can now route to CHASM scheduler, depending on experiment flag/dynamic config - Scheduler functional test has been updated to exercise both V1 and V2 scheduler - Small fixes throughout scheduler codebase to address differences noticed during functional testing - Nexus callback mechanisms updated to allow the CHASM internal callback URL I plan to address the TODO around LastCompletionResult's assertion in a separate follow-up PR. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [x] added new functional test(s) ## Potential risks - Existing V1 tests are preserved, to show that the V1 codepath still works as-is. All new CHASM handling logic falls back to V1 in absence of a positive response. CHASM can only be opted in for a schedule during creation, controlled by dynamic config/experimental header.
105 lines
3.3 KiB
Go
105 lines
3.3 KiB
Go
// Code generated by protoc-gen-go-chasm. DO NOT EDIT.
|
|
package testspb
|
|
|
|
import (
|
|
"context"
|
|
"math/rand"
|
|
"time"
|
|
|
|
"go.temporal.io/server/client/history"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/backoff"
|
|
"go.temporal.io/server/common/config"
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/headers"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/membership"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/primitives"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
// TestServiceLayeredClient is a client for TestService.
|
|
type TestServiceLayeredClient struct {
|
|
metricsHandler metrics.Handler
|
|
numShards int32
|
|
redirector history.Redirector[TestServiceClient]
|
|
retryPolicy backoff.RetryPolicy
|
|
}
|
|
|
|
// NewTestServiceLayeredClient initializes a new TestServiceLayeredClient.
|
|
func NewTestServiceLayeredClient(
|
|
dc *dynamicconfig.Collection,
|
|
rpcFactory common.RPCFactory,
|
|
monitor membership.Monitor,
|
|
config *config.Persistence,
|
|
logger log.Logger,
|
|
metricsHandler metrics.Handler,
|
|
) (TestServiceClient, error) {
|
|
resolver, err := monitor.GetResolver(primitives.HistoryService)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
connections := history.NewConnectionPool(resolver, rpcFactory, NewTestServiceClient)
|
|
var redirector history.Redirector[TestServiceClient]
|
|
if dynamicconfig.HistoryClientOwnershipCachingEnabled.Get(dc)() {
|
|
redirector = history.NewCachingRedirector(
|
|
connections,
|
|
resolver,
|
|
logger,
|
|
dynamicconfig.HistoryClientOwnershipCachingStaleTTL.Get(dc),
|
|
)
|
|
} else {
|
|
redirector = history.NewBasicRedirector(connections, resolver)
|
|
}
|
|
return &TestServiceLayeredClient{
|
|
metricsHandler: metricsHandler,
|
|
redirector: redirector,
|
|
numShards: config.NumHistoryShards,
|
|
retryPolicy: common.CreateHistoryClientRetryPolicy(),
|
|
}, nil
|
|
}
|
|
func (c *TestServiceLayeredClient) callTestNoRetry(
|
|
ctx context.Context,
|
|
request *TestRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*TestResponse, error) {
|
|
var response *TestResponse
|
|
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("TestService.Test"),
|
|
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 := int32(rand.Intn(int(c.numShards)) + 1)
|
|
op := func(ctx context.Context, client TestServiceClient) error {
|
|
var err error
|
|
ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
|
|
defer cancel()
|
|
response, err = client.Test(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err = c.redirector.Execute(ctx, shardID, op)
|
|
return response, err
|
|
}
|
|
func (c *TestServiceLayeredClient) Test(
|
|
ctx context.Context,
|
|
request *TestRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*TestResponse, error) {
|
|
call := func(ctx context.Context) (*TestResponse, error) {
|
|
return c.callTestNoRetry(ctx, request, opts...)
|
|
}
|
|
return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
|
|
}
|