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).
97 lines
3.5 KiB
Go
97 lines
3.5 KiB
Go
package workers
|
|
|
|
import (
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
workerpb "go.temporal.io/api/worker/v1"
|
|
"github.com/hanzoai/tasks/common/dynamicconfig"
|
|
"github.com/hanzoai/tasks/common/metrics"
|
|
"github.com/hanzoai/tasks/common/namespace"
|
|
"github.com/hanzoai/tasks/common/tqid"
|
|
)
|
|
|
|
// WorkerMetricsConfig contains dynamic config flags for worker-related metrics.
|
|
type WorkerMetricsConfig struct {
|
|
EnablePluginMetrics dynamicconfig.BoolPropertyFn
|
|
EnablePollerAutoscalingMetrics dynamicconfig.BoolPropertyFn
|
|
BreakdownMetricsByTaskQueue dynamicconfig.BoolPropertyFnWithTaskQueueFilter
|
|
ExternalPayloadsEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter
|
|
}
|
|
|
|
// workerMetricsEmitter encapsulates logic for emitting metrics derived from worker heartbeats.
|
|
type workerMetricsEmitter struct {
|
|
handler metrics.Handler
|
|
config WorkerMetricsConfig
|
|
}
|
|
|
|
func (e *workerMetricsEmitter) emit(nsID namespace.ID, nsName namespace.Name, heartbeats []*workerpb.WorkerHeartbeat) {
|
|
enablePluginMetrics := e.config.EnablePluginMetrics != nil && e.config.EnablePluginMetrics()
|
|
enablePollerAutoscalingMetrics := e.config.EnablePollerAutoscalingMetrics != nil && e.config.EnablePollerAutoscalingMetrics()
|
|
enableStorageDriverMetrics := e.config.ExternalPayloadsEnabled != nil && e.config.ExternalPayloadsEnabled(nsName.String())
|
|
|
|
recordedPlugins := make(map[string]bool)
|
|
recordedDrivers := make(map[string]bool)
|
|
|
|
for _, hb := range heartbeats {
|
|
// Activity slots metric (always enabled)
|
|
if hb.ActivityTaskSlotsInfo != nil {
|
|
metrics.WorkerRegistryActivitySlotsUsed.With(e.handler).Record(int64(hb.ActivityTaskSlotsInfo.CurrentUsedSlots))
|
|
}
|
|
|
|
// Plugin metrics (if enabled)
|
|
if enablePluginMetrics {
|
|
for _, pluginInfo := range hb.Plugins {
|
|
pluginName := pluginInfo.Name
|
|
if !recordedPlugins[pluginName] {
|
|
metrics.WorkerPluginNameMetric.
|
|
With(e.handler).
|
|
Record(1, metrics.NamespaceTag(nsName.String()), metrics.WorkerPluginNameTag(pluginName))
|
|
recordedPlugins[pluginName] = true
|
|
}
|
|
}
|
|
}
|
|
|
|
// Poller autoscaling metrics (if enabled)
|
|
if enablePollerAutoscalingMetrics {
|
|
e.emitPollerAutoscaling(nsID, nsName, hb)
|
|
}
|
|
|
|
// Storage driver metrics (if external payloads enabled)
|
|
if enableStorageDriverMetrics {
|
|
for _, driver := range hb.GetDrivers() {
|
|
driverType := driver.GetType()
|
|
if !recordedDrivers[driverType] {
|
|
metrics.WorkerStorageDriverTypeMetric.
|
|
With(e.handler).
|
|
Record(1, metrics.NamespaceTag(nsName.String()), metrics.WorkerStorageDriverTypeTag(driverType))
|
|
recordedDrivers[driverType] = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *workerMetricsEmitter) emitPollerAutoscaling(nsID namespace.ID, nsName namespace.Name, hb *workerpb.WorkerHeartbeat) {
|
|
family, err := tqid.NewTaskQueueFamily(nsID.String(), hb.GetTaskQueue())
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
recordAutoscaling := func(taskType enumspb.TaskQueueType) {
|
|
tq := family.TaskQueue(taskType)
|
|
breakdownByTQ := e.config.BreakdownMetricsByTaskQueue != nil &&
|
|
e.config.BreakdownMetricsByTaskQueue(nsName.String(), hb.GetTaskQueue(), taskType)
|
|
handler := metrics.GetPerTaskQueueScope(e.handler, nsName.String(), tq, breakdownByTQ)
|
|
metrics.PollerAutoscalingHeartbeatCount.With(handler).Record(1)
|
|
}
|
|
|
|
if hb.WorkflowPollerInfo.GetIsAutoscaling() {
|
|
recordAutoscaling(enumspb.TASK_QUEUE_TYPE_WORKFLOW)
|
|
}
|
|
if hb.ActivityPollerInfo.GetIsAutoscaling() {
|
|
recordAutoscaling(enumspb.TASK_QUEUE_TYPE_ACTIVITY)
|
|
}
|
|
if hb.NexusPollerInfo.GetIsAutoscaling() {
|
|
recordAutoscaling(enumspb.TASK_QUEUE_TYPE_NEXUS)
|
|
}
|
|
}
|