@@ -25,6 +25,7 @@ import (
2525 "github.com/tilebox/tilebox-go/workflows/v1/runner"
2626 "github.com/tilebox/tilebox-go/workflows/v1/subtask"
2727 "go.opentelemetry.io/otel"
28+ "go.opentelemetry.io/otel/attribute"
2829 "go.opentelemetry.io/otel/metric"
2930 "go.opentelemetry.io/otel/trace"
3031 "google.golang.org/grpc/codes"
@@ -48,17 +49,87 @@ const (
4849 fallbackJitterInterval = 5 * time .Second
4950)
5051
52+ const (
53+ UnitSeconds = "s"
54+ UnitDimensionless = "1"
55+ UnitBytes = "By"
56+ )
57+
58+ type taskRunnerMetrics struct {
59+ tasksExecutedMetric metric.Int64Counter
60+ tasksComputedMetric metric.Int64Counter
61+ tasksFailedMetric metric.Int64Counter
62+
63+ taskInputSizeMetric metric.Int64Histogram
64+ taskExecutionDurationMetric metric.Float64Histogram
65+ }
66+
67+ func newTaskRunnerMetrics (meter metric.Meter ) (* taskRunnerMetrics , error ) {
68+ tasksExecutedMetric , err := meter .Int64Counter (
69+ "task.executed.count" ,
70+ metric .WithDescription ("Number of tasks executed" ),
71+ metric .WithUnit (UnitDimensionless ),
72+ )
73+ if err != nil {
74+ return nil , fmt .Errorf ("failed to create task count metric: %w" , err )
75+ }
76+
77+ tasksComputedMetric , err := meter .Int64Counter (
78+ "task.computed.count" ,
79+ metric .WithDescription ("Number of tasks computed" ),
80+ metric .WithUnit (UnitDimensionless ),
81+ )
82+ if err != nil {
83+ return nil , fmt .Errorf ("failed to create task computed metric: %w" , err )
84+ }
85+
86+ tasksFailedMetric , err := meter .Int64Counter (
87+ "task.failed.count" ,
88+ metric .WithDescription ("Number of tasks failed" ),
89+ metric .WithUnit (UnitDimensionless ),
90+ )
91+ if err != nil {
92+ return nil , fmt .Errorf ("failed to create task failed metric: %w" , err )
93+ }
94+
95+ taskArgsSizeMetric , err := meter .Int64Histogram (
96+ "task.input.size" ,
97+ metric .WithDescription ("Task arguments size" ),
98+ metric .WithUnit (UnitBytes ),
99+ )
100+ if err != nil {
101+ return nil , fmt .Errorf ("failed to create task input size metric: %w" , err )
102+ }
103+
104+ taskExecutionDurationMetric , err := meter .Float64Histogram (
105+ "task.execution.duration" ,
106+ metric .WithDescription ("Task execution duration" ),
107+ metric .WithUnit (UnitSeconds ),
108+ )
109+ if err != nil {
110+ return nil , fmt .Errorf ("failed to create task duration metric: %w" , err )
111+ }
112+
113+ return & taskRunnerMetrics {
114+ tasksExecutedMetric : tasksExecutedMetric ,
115+ tasksComputedMetric : tasksComputedMetric ,
116+ tasksFailedMetric : tasksFailedMetric ,
117+ taskInputSizeMetric : taskArgsSizeMetric ,
118+ taskExecutionDurationMetric : taskExecutionDurationMetric ,
119+ }, nil
120+ }
121+
51122// TaskRunner executes tasks.
52123//
53124// Documentation: https://docs.tilebox.com/workflows/concepts/task-runners
54125type TaskRunner struct {
55126 service TaskService
56127 taskDefinitions map [taskIdentifier ]ExecutableTask
57128
58- cluster string
59- tracer trace.Tracer
60- logger * slog.Logger
61- taskDurationMetric metric. Float64Histogram
129+ cluster string
130+ tracer trace.Tracer
131+ logger * slog.Logger
132+ metrics * taskRunnerMetrics
62133}
63134
64135func newTaskRunner (ctx context.Context , service TaskService , clusterClient ClusterClient , tracer trace.Tracer , options ... runner.Option ) (* TaskRunner , error ) {
@@ -76,25 +147,19 @@ func newTaskRunner(ctx context.Context, service TaskService, clusterClient Clust
76147 return nil , fmt .Errorf ("failed to get cluster: %w" , err )
77148 }
78149
79- meter := opts .MeterProvider .Meter (otelMeterName )
80-
81- taskDurationMetric , err := meter .Float64Histogram (
82- "task.execution.duration" ,
83- metric .WithDescription ("Task execution duration" ),
84- metric .WithUnit ("s" ),
85- )
150+ metrics , err := newTaskRunnerMetrics (opts .MeterProvider .Meter (otelMeterName ))
86151 if err != nil {
87- return nil , fmt .Errorf ("failed to create task execution duration metric : %w" , err )
152+ return nil , fmt .Errorf ("failed to create task runner metrics : %w" , err )
88153 }
89154
90155 return & TaskRunner {
91156 service : service ,
92157 taskDefinitions : make (map [taskIdentifier ]ExecutableTask ),
93158
94- cluster : cluster .Slug ,
95- tracer : tracer ,
96- logger : opts .Logger ,
97- taskDurationMetric : taskDurationMetric ,
159+ cluster : cluster .Slug ,
160+ tracer : tracer ,
161+ logger : opts .Logger ,
162+ metrics : metrics ,
98163 }, nil
99164}
100165
@@ -329,33 +394,44 @@ func (t *TaskRunner) executeTask(ctx context.Context, task *workflowsv1.Task) (*
329394 slog .Time ("start_time" , beforeTime ),
330395 )
331396
397+ taskMetricAttributes := metric .WithAttributes (attribute .String ("task_identifier" , identifier .Name ()), attribute .String ("task_version" , identifier .Version ()))
398+
332399 defer func () {
333400 if r := recover (); r != nil {
334401 // recover from panics during task executions, so we can still report the error to the server and continue
335402 // with other tasks
336403 log .ErrorContext (ctx , "task execution failed" , slog .String ("error" , "panic" ), slog .Int64 ("retry_attempt" , task .GetRetryCount ()))
337404 taskExecutionContext = nil
338405 err = fmt .Errorf ("task panicked: %v" , r )
406+
407+ // also instrument the panic as a failed task execution
408+ t .metrics .tasksFailedMetric .Add (ctx , 1 , taskMetricAttributes )
409+ t .metrics .taskExecutionDurationMetric .Record (ctx , time .Since (beforeTime ).Seconds (), taskMetricAttributes , metric .WithAttributes (attribute .String ("state" , "failed" )))
339410 }
340411 }()
341412
413+ t .metrics .taskInputSizeMetric .Record (ctx , int64 (len (task .GetInput ())), taskMetricAttributes )
414+ t .metrics .tasksExecutedMetric .Add (ctx , 1 , taskMetricAttributes )
415+
342416 executionContext := t .withTaskExecutionContext (ctx , task )
343417 err = taskStruct .Execute (executionContext )
344418
345419 executionTime := time .Since (beforeTime )
346-
347420 log = log .With (
348421 slog .Duration ("execution_time" , executionTime ),
349422 slog .String ("execution_time_human" , roundDuration (executionTime , 2 ).String ()),
350423 )
351424
352425 if err != nil {
426+ t .metrics .tasksFailedMetric .Add (ctx , 1 , taskMetricAttributes )
427+ t .metrics .taskExecutionDurationMetric .Record (ctx , executionTime .Seconds (), taskMetricAttributes , metric .WithAttributes (attribute .String ("state" , "failed" )))
428+
353429 log .ErrorContext (ctx , "task execution failed" , slog .Any ("error" , err ), slog .Int64 ("retry_attempt" , task .GetRetryCount ()))
354430 return getTaskExecutionContext (executionContext ), fmt .Errorf ("failed to execute task: %w" , err )
355431 }
356432
357- // record the time it took to run a successful task
358- t .taskDurationMetric . Record (ctx , executionTime .Seconds ())
433+ t . metrics . tasksComputedMetric . Add ( ctx , 1 , taskMetricAttributes )
434+ t .metrics . taskExecutionDurationMetric . Record (ctx , executionTime .Seconds (), taskMetricAttributes , metric . WithAttributes ( attribute . String ( "state" , "computed" ) ))
359435
360436 return getTaskExecutionContext (executionContext ), nil
361437 })
0 commit comments