mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
## 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.
167 lines
5.1 KiB
Go
167 lines
5.1 KiB
Go
package callbacks
|
|
|
|
import (
|
|
"net/url"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/common/backoff"
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/nexus"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
var RequestTimeout = dynamicconfig.NewDestinationDurationSetting(
|
|
"component.callbacks.request.timeout",
|
|
time.Second*10,
|
|
`RequestTimeout is the timeout for executing a single callback request.`,
|
|
)
|
|
|
|
var RetryPolicyInitialInterval = dynamicconfig.NewGlobalDurationSetting(
|
|
"component.callbacks.retryPolicy.initialInterval",
|
|
time.Second,
|
|
`The initial backoff interval between every callback request attempt for a given callback.`,
|
|
)
|
|
|
|
var RetryPolicyMaximumInterval = dynamicconfig.NewGlobalDurationSetting(
|
|
"component.callbacks.retryPolicy.maxInterval",
|
|
time.Hour,
|
|
`The maximum backoff interval between every callback request attempt for a given callback.`,
|
|
)
|
|
|
|
type Config struct {
|
|
RequestTimeout dynamicconfig.DurationPropertyFnWithDestinationFilter
|
|
RetryPolicy func() backoff.RetryPolicy
|
|
}
|
|
|
|
func ConfigProvider(dc *dynamicconfig.Collection) *Config {
|
|
return &Config{
|
|
RequestTimeout: RequestTimeout.Get(dc),
|
|
RetryPolicy: func() backoff.RetryPolicy {
|
|
return backoff.NewExponentialRetryPolicy(
|
|
RetryPolicyInitialInterval.Get(dc)(),
|
|
).WithMaximumInterval(
|
|
RetryPolicyMaximumInterval.Get(dc)(),
|
|
).WithExpirationInterval(
|
|
backoff.NoInterval,
|
|
)
|
|
},
|
|
}
|
|
}
|
|
|
|
var AllowedAddresses = dynamicconfig.NewNamespaceTypedSettingWithConverter(
|
|
"component.callbacks.allowedAddresses",
|
|
allowedAddressConverter,
|
|
AddressMatchRules{},
|
|
`The per-namespace list of addresses that are allowed for callbacks and whether secure connections (https) are required.
|
|
URL: "temporal://system" is always allowed for worker callbacks. The default is no address rules.
|
|
URLs are checked against each in order when starting a workflow with attached callbacks and only need to match one to pass validation.
|
|
This configuration is required for external endpoint targets; any invalid entries are ignored. Each entry is a map with possible values:
|
|
- "Pattern":string (required) the host:port pattern to which this config applies.
|
|
Wildcards, '*', are supported and can match any number of characters (e.g. '*' matches everything, 'prefix.*.domain' matches 'prefix.a.domain' as well as 'prefix.a.b.domain').
|
|
- "AllowInsecure":bool (optional, default=false) indicates whether https is required`)
|
|
|
|
type AddressMatchRules struct {
|
|
Rules []AddressMatchRule
|
|
}
|
|
|
|
func (a AddressMatchRules) Validate(rawURL string) error {
|
|
// Exact match only; no path, query, or fragment allowed for system URL
|
|
if rawURL == nexus.SystemCallbackURL || rawURL == chasm.NexusCompletionHandlerURL {
|
|
return nil
|
|
}
|
|
u, err := url.Parse(rawURL)
|
|
if err != nil {
|
|
return status.Errorf(codes.InvalidArgument, "invalid callback url: %v", err)
|
|
}
|
|
if u.Scheme != "http" && u.Scheme != "https" {
|
|
return status.Errorf(codes.InvalidArgument, "invalid url: unknown scheme: %v", u)
|
|
}
|
|
if u.Host == "" {
|
|
return status.Errorf(codes.InvalidArgument, "invalid url: missing host")
|
|
}
|
|
for _, rule := range a.Rules {
|
|
allow, err := rule.Allow(u)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if allow {
|
|
return nil
|
|
}
|
|
}
|
|
return status.Errorf(codes.InvalidArgument, "invalid url: url does not match any configured callback address: %v", u)
|
|
}
|
|
|
|
type AddressMatchRule struct {
|
|
Regexp *regexp.Regexp
|
|
AllowInsecure bool
|
|
}
|
|
|
|
// Allow validates the URL by:
|
|
// 1. true, nil if the provided url matches the rule and passed validation
|
|
// for the given rule.
|
|
// 2. false, nil if the URL does not match the rule.
|
|
// 3. It false, error if there is a match and the URL fails validation
|
|
func (a AddressMatchRule) Allow(u *url.URL) (bool, error) {
|
|
if !a.Regexp.MatchString(u.Host) {
|
|
return false, nil
|
|
}
|
|
if a.AllowInsecure {
|
|
return true, nil
|
|
}
|
|
if u.Scheme != "https" {
|
|
return false,
|
|
status.Errorf(codes.InvalidArgument,
|
|
"invalid url: callback address does not allow insecure connections: %v", u)
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func allowedAddressConverter(val any) (AddressMatchRules, error) {
|
|
type entry struct {
|
|
Pattern string
|
|
AllowInsecure bool
|
|
}
|
|
intermediate, err := dynamicconfig.ConvertStructure[[]entry](nil)(val)
|
|
if err != nil {
|
|
return AddressMatchRules{}, err
|
|
}
|
|
|
|
configs := []AddressMatchRule{}
|
|
for _, e := range intermediate {
|
|
if e.Pattern == "" {
|
|
// Skip configs with missing / unparsable Pattern
|
|
continue
|
|
}
|
|
re, err := regexp.Compile(addressPatternToRegexp(e.Pattern))
|
|
if err != nil {
|
|
// Skip configs with malformed Pattern
|
|
continue
|
|
}
|
|
configs = append(configs, AddressMatchRule{
|
|
Regexp: re,
|
|
AllowInsecure: e.AllowInsecure,
|
|
})
|
|
}
|
|
return AddressMatchRules{Rules: configs}, nil
|
|
}
|
|
|
|
func addressPatternToRegexp(pattern string) string {
|
|
var result strings.Builder
|
|
result.WriteString("^")
|
|
first := true
|
|
for literal := range strings.SplitSeq(pattern, "*") {
|
|
if !first {
|
|
// Replace * with .*
|
|
result.WriteString(".*")
|
|
}
|
|
result.WriteString(regexp.QuoteMeta(literal))
|
|
first = false
|
|
}
|
|
result.WriteString("$")
|
|
return result.String()
|
|
}
|