mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekay
40351cce6f
Goes from 135 MB binary with 1153 files importing go.temporal.io and 224
importing google.golang.org/grpc to a 10 MB native binary with ZERO of
either.
Deleted:
- tasks/, tasktests/, service/, chasm/, client/, api/, proto/, tools/
- cmd/tools/, components/, docker/
- schema/{cassandra,elasticsearch,mysql,postgresql,sqlite}/
- 40 tainted common/ subdirs + 7 top-level common files
- tests/testutils (no longer referenced after surviving deps drop common)
New cmd/tasksd (~100 lines) boots tasks.Embed (zap.Node) + http.Server.
New pkg/tasks/embed.go registers ZAP handlers for the SDK opcode surface
(0x0060-0x00A5). Workflow handlers return a 501 "not yet implemented"
envelope; native engine lands behind this contract.
Build: go build ./cmd/tasksd → 10 MB
Vet: go vet ./... → clean
Boot: tasksd --zap-port 9999 --http :7243
Probes: /healthz, /v1/tasks/health → 200
UI: /_/tasks/, / → 200 (embedded Vite bundle)
ZAP: _tasks._tcp on :9999 (mDNS disabled; direct connect)
go.mod: tidy. No go.temporal.io. No google.golang.org/grpc.
174 lines
5.6 KiB
Go
174 lines
5.6 KiB
Go
package history
|
|
|
|
import (
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/persistence/visibility/manager"
|
|
ctasks "go.temporal.io/server/common/tasks"
|
|
"go.temporal.io/server/common/telemetry"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/queues"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
"go.uber.org/fx"
|
|
)
|
|
|
|
const (
|
|
visibilityQueuePersistenceMaxRPSRatio = 0.15
|
|
)
|
|
|
|
type (
|
|
visibilityQueueFactoryParams struct {
|
|
fx.In
|
|
|
|
QueueFactoryBaseParams
|
|
|
|
VisibilityMgr manager.VisibilityManager
|
|
}
|
|
|
|
visibilityQueueFactory struct {
|
|
visibilityQueueFactoryParams
|
|
QueueFactoryBase
|
|
}
|
|
)
|
|
|
|
func NewVisibilityQueueFactory(
|
|
params visibilityQueueFactoryParams,
|
|
) QueueFactory {
|
|
return &visibilityQueueFactory{
|
|
visibilityQueueFactoryParams: params,
|
|
QueueFactoryBase: QueueFactoryBase{
|
|
HostScheduler: queues.NewScheduler(
|
|
params.ClusterMetadata.GetCurrentClusterName(),
|
|
queues.SchedulerOptions{
|
|
WorkerCount: params.Config.VisibilityProcessorSchedulerWorkerCount,
|
|
ActiveNamespaceWeights: params.Config.VisibilityProcessorSchedulerActiveRoundRobinWeights,
|
|
StandbyNamespaceWeights: params.Config.VisibilityProcessorSchedulerStandbyRoundRobinWeights,
|
|
InactiveNamespaceDeletionDelay: params.Config.TaskSchedulerInactiveChannelDeletionDelay,
|
|
ExecutionAwareSchedulerOptions: ctasks.ExecutionAwareSchedulerOptions{
|
|
Enabled: params.Config.TaskSchedulerEnableExecutionQueueScheduler,
|
|
MaxQueues: params.Config.TaskSchedulerExecutionQueueSchedulerMaxQueues,
|
|
QueueTTL: params.Config.TaskSchedulerExecutionQueueSchedulerQueueTTL,
|
|
QueueConcurrency: params.Config.TaskSchedulerExecutionQueueSchedulerQueueConcurrency,
|
|
},
|
|
},
|
|
params.NamespaceRegistry,
|
|
params.Logger,
|
|
params.MetricsHandler,
|
|
params.TimeSource,
|
|
),
|
|
HostPriorityAssigner: queues.NewPriorityAssigner(
|
|
params.NamespaceRegistry,
|
|
params.ClusterMetadata.GetCurrentClusterName(),
|
|
),
|
|
HostReaderRateLimiter: queues.NewReaderPriorityRateLimiter(
|
|
NewHostRateLimiterRateFn(
|
|
params.Config.VisibilityProcessorMaxPollHostRPS,
|
|
params.Config.PersistenceMaxQPS,
|
|
visibilityQueuePersistenceMaxRPSRatio,
|
|
),
|
|
int64(params.Config.VisibilityQueueMaxReaderCount()),
|
|
),
|
|
Tracer: params.TracerProvider.Tracer(telemetry.ComponentQueueVisibility),
|
|
},
|
|
}
|
|
}
|
|
|
|
func (f *visibilityQueueFactory) CreateQueue(
|
|
shard historyi.ShardContext,
|
|
) queues.Queue {
|
|
logger := log.With(shard.GetLogger(), tag.ComponentVisibilityQueue)
|
|
metricsHandler := f.MetricsHandler.WithTags(metrics.OperationTag(metrics.OperationVisibilityQueueProcessorScope))
|
|
|
|
shardScheduler := queues.NewRateLimitedScheduler(
|
|
f.HostScheduler,
|
|
queues.RateLimitedSchedulerOptions{
|
|
Enabled: f.Config.TaskSchedulerEnableRateLimiter,
|
|
EnableShadowMode: f.Config.TaskSchedulerEnableRateLimiterShadowMode,
|
|
StartupDelay: f.Config.TaskSchedulerRateLimiterStartupDelay,
|
|
},
|
|
f.ClusterMetadata.GetCurrentClusterName(),
|
|
f.NamespaceRegistry,
|
|
f.SchedulerRateLimiter,
|
|
f.TimeSource,
|
|
f.ChasmRegistry,
|
|
logger,
|
|
metricsHandler,
|
|
)
|
|
|
|
rescheduler := queues.NewRescheduler(
|
|
shardScheduler,
|
|
shard.GetTimeSource(),
|
|
logger,
|
|
metricsHandler,
|
|
)
|
|
|
|
executor := newVisibilityQueueTaskExecutor(
|
|
shard,
|
|
f.WorkflowCache,
|
|
f.VisibilityMgr,
|
|
logger,
|
|
f.MetricsHandler,
|
|
f.Config.VisibilityProcessorEnsureCloseBeforeDelete,
|
|
f.Config.VisibilityProcessorEnableCloseWorkflowCleanup,
|
|
f.Config.VisibilityProcessorRelocateAttributesMinBlobSize,
|
|
f.Config.ExternalPayloadsEnabled,
|
|
)
|
|
if f.ExecutorWrapper != nil {
|
|
executor = f.ExecutorWrapper.Wrap(executor)
|
|
}
|
|
|
|
factory := queues.NewExecutableFactory(
|
|
executor,
|
|
shardScheduler,
|
|
rescheduler,
|
|
f.HostPriorityAssigner,
|
|
shard.GetTimeSource(),
|
|
shard.GetNamespaceRegistry(),
|
|
shard.GetClusterMetadata(),
|
|
f.ChasmRegistry,
|
|
queues.GetTaskTypeTagValue,
|
|
logger,
|
|
metricsHandler,
|
|
f.Tracer,
|
|
f.DLQWriter,
|
|
f.Config.TaskDLQEnabled,
|
|
f.Config.TaskDLQUnexpectedErrorAttempts,
|
|
f.Config.TaskDLQInternalErrors,
|
|
f.Config.TaskDLQErrorPattern,
|
|
)
|
|
return queues.NewImmediateQueue(
|
|
shard,
|
|
tasks.CategoryVisibility,
|
|
shardScheduler,
|
|
rescheduler,
|
|
&queues.Options{
|
|
ReaderOptions: queues.ReaderOptions{
|
|
BatchSize: f.Config.VisibilityTaskBatchSize,
|
|
MaxPendingTasksCount: f.Config.QueuePendingTaskMaxCount,
|
|
PollBackoffInterval: f.Config.VisibilityProcessorPollBackoffInterval,
|
|
MaxPredicateSize: f.Config.QueueMaxPredicateSize,
|
|
},
|
|
MonitorOptions: queues.MonitorOptions{
|
|
PendingTasksCriticalCount: f.Config.QueuePendingTaskCriticalCount,
|
|
ReaderStuckCriticalAttempts: f.Config.QueueReaderStuckCriticalAttempts,
|
|
SliceCountCriticalThreshold: f.Config.QueueCriticalSlicesCount,
|
|
},
|
|
MaxPollRPS: f.Config.VisibilityProcessorMaxPollRPS,
|
|
MaxPollInterval: f.Config.VisibilityProcessorMaxPollInterval,
|
|
MaxPollIntervalJitterCoefficient: f.Config.VisibilityProcessorMaxPollIntervalJitterCoefficient,
|
|
CheckpointInterval: f.Config.VisibilityProcessorUpdateAckInterval,
|
|
CheckpointIntervalJitterCoefficient: f.Config.VisibilityProcessorUpdateAckIntervalJitterCoefficient,
|
|
MaxReaderCount: f.Config.VisibilityQueueMaxReaderCount,
|
|
MoveGroupTaskCountBase: f.Config.QueueMoveGroupTaskCountBase,
|
|
MoveGroupTaskCountMultiplier: f.Config.QueueMoveGroupTaskCountMultiplier,
|
|
},
|
|
f.HostReaderRateLimiter,
|
|
queues.GrouperNamespaceID{},
|
|
logger,
|
|
metricsHandler,
|
|
factory,
|
|
nil, // taskPostProcessor
|
|
)
|
|
}
|