mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Carly de FrondevilleandGitHub
547652a9b2
## What changed? Give the option to upgrade-on-Continue-as-New with a continue-as-new option. Ensure that Pinned Overrides are inherited across the entire continue-as-new chain, regardless of the initial versioning behavior of the continue-as-new. ## Why? To unlock "Trampolining" ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [x] added new functional test(s) --- > [!NOTE] > Introduces explicit Continue-as-New tracking and version-targeting across APIs and server internals. > > - Protos: add `target_deployment_version` to `RecordWorkflowTaskStartedRequest`; add repeated `workflow_task_suggest_continue_as_new_reasons` to persistence `WorkflowExecutionInfo`; regenerate bindings and enum imports > - History: plumb `target_deployment_version` through `AddWorkflowTaskStartedEvent`; include `suggest_continue_as_new_reasons` in `WorkflowTaskStarted` events; choose non-nil `VersioningOverride` when starting executions (user or inherited) > - Metrics: remove `workflow_update_continue_as_new_suggestions`; add `workflow_continue_as_new_count` and `workflow_suggest_continue_as_new_count`; introduce tags `continue_as_new_versioning_behavior` and per-reason suggest flags; tag helpers added > - Interfaces/tests: extend `MutableState.Add/ApplyWorkflowTaskStartedEvent` and `WorkflowTaskInfo` with suggest reasons; update call sites and tests; update `common/util_test.go` scan; bump `go.temporal.io/api` > > <sup>Written by [Cursor Bugbot]() for commit 849687bd0cee86dfa8a7949e76161a5c5a090f25. This will update automatically on new commits. Configure [here]().</sup>
110 lines
3.1 KiB
Go
110 lines
3.1 KiB
Go
package update
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"go.opentelemetry.io/otel/trace"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/softassert"
|
|
"go.temporal.io/server/common/telemetry"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type (
|
|
instrumentation struct {
|
|
log log.Logger
|
|
metrics metrics.Handler
|
|
tracer trace.Tracer
|
|
}
|
|
)
|
|
|
|
var (
|
|
noopInstrumentation = instrumentation{
|
|
log: log.NewNoopLogger(),
|
|
metrics: metrics.NoopMetricsHandler,
|
|
tracer: telemetry.NoopTracer,
|
|
}
|
|
)
|
|
|
|
func (i *instrumentation) countRequestMsg() {
|
|
i.oneOf(metrics.MessageTypeRequestWorkflowExecutionUpdateCounter.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countAcceptanceMsg() {
|
|
i.oneOf(metrics.MessageTypeAcceptWorkflowExecutionUpdateCounter.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countRejectionMsg() {
|
|
i.oneOf(metrics.MessageTypeRejectWorkflowExecutionUpdateCounter.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countResponseMsg() {
|
|
i.oneOf(metrics.MessageTypeRespondWorkflowExecutionUpdateCounter.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countRateLimited() {
|
|
i.oneOf(metrics.WorkflowExecutionUpdateRequestRateLimited.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countRegistrySizeLimited(updateCount, registrySize, payloadSize int) {
|
|
i.oneOf(metrics.WorkflowExecutionUpdateRegistrySizeLimited.Name())
|
|
// TODO: remove log once limit is enforced everywhere
|
|
i.log.Warn("update registry size limit reached",
|
|
tag.NewInt("registry-size", registrySize),
|
|
tag.NewInt("payload-size", payloadSize),
|
|
tag.NewInt("update-count", updateCount))
|
|
}
|
|
|
|
func (i *instrumentation) countTooMany() {
|
|
i.oneOf(metrics.WorkflowExecutionUpdateTooMany.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countAborted(updateID string, reason AbortReason) {
|
|
i.metrics.Counter(metrics.WorkflowExecutionUpdateAborted.Name()).
|
|
Record(1, metrics.ReasonTag(metrics.ReasonString(reason.String())))
|
|
i.log.Debug("update aborted",
|
|
tag.NewStringTag("reason", reason.String()),
|
|
tag.NewStringTag("update-id", updateID),
|
|
)
|
|
}
|
|
|
|
func (i *instrumentation) countSent() {
|
|
i.oneOf(metrics.WorkflowExecutionUpdateSentToWorker.Name())
|
|
}
|
|
|
|
func (i *instrumentation) countSentAgain() {
|
|
i.oneOf(metrics.WorkflowExecutionUpdateSentToWorkerAgain.Name())
|
|
}
|
|
|
|
func (i *instrumentation) invalidStateTransition(updateID string, msg proto.Message, state state) {
|
|
i.oneOf(metrics.InvalidStateTransitionWorkflowExecutionUpdateCounter.Name())
|
|
softassert.Fail(
|
|
i.log,
|
|
"invalid state transition attempted",
|
|
tag.ComponentWorkflowUpdate,
|
|
tag.NewStringTag("update-id", updateID),
|
|
tag.NewStringTag("message", fmt.Sprintf("%T", msg)),
|
|
tag.NewStringerTag("state", state),
|
|
)
|
|
}
|
|
|
|
func (i *instrumentation) updateRegistrySize(size int) {
|
|
i.metrics.Histogram(metrics.WorkflowExecutionUpdateRegistrySize.Name(), metrics.Bytes).Record(int64(size))
|
|
}
|
|
|
|
func (i *instrumentation) oneOf(counterName string) {
|
|
i.metrics.Counter(counterName).Record(1)
|
|
}
|
|
|
|
func (i *instrumentation) stateChange(updateID string, from, to state) {
|
|
i.log.Debug(
|
|
"update state change",
|
|
tag.ComponentWorkflowUpdate,
|
|
tag.NewStringTag("update-id", updateID),
|
|
tag.NewStringerTag("from-state", from),
|
|
tag.NewStringerTag("to-state", to),
|
|
)
|
|
}
|