diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 4478877..34d39d8 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -41,4 +41,4 @@ jobs: - name: Lint uses: golangci/golangci-lint-action@v9 with: - version: v2.10.1 + version: v2.12.2 diff --git a/.golangci.yaml b/.golangci.yaml index b892a71..a9a0658 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -2,7 +2,7 @@ --- version: "2" linters: - enable: # list taken from https://golangci-lint.run/usage/linters/ - last updated 2026-01-09 for v2.8.0 + enable: # list taken from https://golangci-lint.run/usage/linters/ - last updated 2026-05-11 for v2.12.2 # enabled by default, but list them here to be explicit - errcheck - govet @@ -16,6 +16,7 @@ linters: - bidichk # checks for dangerous unicode character sequences - bodyclose # checks whether HTTP response body is closed successfully - canonicalheader # checks for canonical names in HTTP headers + - clickhouselint # detects common mistakes with the ClickHouse native Go driver API - containedctx # detects struct contained context.Context field #- contextcheck # checks for inherited context.Context - copyloopvar # detects places where loop variables are copied @@ -45,7 +46,7 @@ linters: - gochecknoinits # checks that no init functions are present in Go code - gochecksumtype # checks exhaustiveness on Go "sum types" #- gocognit # Computes and checks the cognitive complexity of functions - - goconst # finds repeated strings that could be replaced by a constant + #- goconst # finds repeated strings that could be replaced by a constant - gocritic # provides diagnostics that check for bugs, performance and style issues #- gocyclo # Computes and checks the cyclomatic complexity of functions - godoclint # Checks golang docs best practices (godoc) @@ -53,7 +54,8 @@ linters: #- godox # Tool for detection of FIXME, TODO and other comment keywords #- goheader # Checks is file header matches to pattern - gomoddirectives # manages the use of 'replace', 'retract', and 'excludes' directives in go.mod - - gomodguard # allow and block lists linter for direct Go module dependencies. This is different from depguard where there are different block types for example version constraints and module recommendations + #- gomodguard # allow and block lists linter for direct Go module dependencies. This is different from depguard where there are different block types for example version constraints and module recommendations + - gomodguard_v2 # Allow and blocklist linter for direct Go module dependencies - goprintffuncname # checks that printf-like functions are named with f at the end - gosec # inspects source code for security problems - gosmopolitan # Report certain i18n/l10n anti-patterns in your Go codebase. @@ -171,4 +173,4 @@ formatters: - gci - gofmt - gofumpt - - goimports \ No newline at end of file + - goimports diff --git a/CHANGELOG.md b/CHANGELOG.md index c50f011..a1ccdbd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,9 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.5.0] - 2026-05-11 + ### Added - `workflows`: Added `client.Jobs.QueryLogs` and `client.Jobs.QuerySpans` to query logs and trace spans for a job, plus a telemetry query example. +- `workflows`: Added `ConfigureConsoleLogging` to enable console log output that composes with the Tilebox OpenTelemetry log exporter. +- `workflows`: Added `WithSpan` and `WithSpanResult` helpers to start spans from the current task execution context without manually passing a tracer. + +### Changed + +- `workflows`: Correlate context-aware task logs with traces by adding `trace_id`, `span_id`, and `task_id` attributes and recording log messages as span events. +- `examples`: Updated workflow and dataset examples to use context-aware `slog` methods. ## [0.4.0] - 2026-03-06 @@ -81,7 +90,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added support for Tilebox Observability, including logging and tracing helpers. - Added examples for using the library. -[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.4.0...HEAD +[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.5.0...HEAD +[0.5.0]: https://github.com/tilebox/tilebox-go/compare/v0.4.0...v0.5.0 [0.4.0]: https://github.com/tilebox/tilebox-go/compare/v0.3.2...v0.4.0 [0.3.2]: https://github.com/tilebox/tilebox-go/compare/v0.3.1...v0.3.2 [0.3.1]: https://github.com/tilebox/tilebox-go/compare/v0.3.0...v0.3.1 diff --git a/README.md b/README.md index 868202b..58d23a1 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ For examples on how to use the library, see the [examples](examples) directory. ### Writing a Task -Here we define a simple task that prints "Hello World!" to the console: +Here we define a simple task that logs "Hello World!": ```go package helloworld @@ -65,8 +65,8 @@ type HelloTask struct { } // The Execute method isn't needed to submit a task but is required to run a task. -func (t *HelloTask) Execute(context.Context) error { - slog.Info("Hello World!", "Name", t.Name) +func (t *HelloTask) Execute(ctx context.Context) error { + slog.InfoContext(ctx, "Hello World!", slog.String("Name", t.Name)) return nil } @@ -96,6 +96,7 @@ type HelloTask struct { func main() { ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) client := workflows.NewClient() job, err := client.Jobs.Submit(ctx, "hello-world", @@ -106,14 +107,16 @@ func main() { }, ) if err != nil { - slog.Error("Failed to submit job", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to submit job", slog.Any("error", err)) return } - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) } ``` +`workflows.NewClient()` configures Tilebox OpenTelemetry export when an API key is available. `workflows.ConfigureConsoleLogging` adds console output without replacing the Tilebox exporter, so logs are written to both destinations when both are configured. + ### Running a Worker Here we create a TaskRunner and run a worker that is capable of executing `HelloTask` tasks: @@ -133,24 +136,25 @@ type HelloTask struct { } // The Execute method is required to run a task. -func (t *HelloTask) Execute(context.Context) error { - slog.Info("Hello World!", "Name", t.Name) +func (t *HelloTask) Execute(ctx context.Context) error { + slog.InfoContext(ctx, "Hello World!", slog.String("Name", t.Name)) return nil } func main() { ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) client := workflows.NewClient() - + runner, err := client.NewTaskRunner(ctx) if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) return } err = runner.RegisterTasks(&HelloTask{}) if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) return } @@ -158,7 +162,6 @@ func main() { } ``` - ## License Distributed under the MIT License (`The MIT License`). diff --git a/datasets/v1/datapoints.go b/datasets/v1/datapoints.go index 043eb8c..7f74318 100644 --- a/datasets/v1/datapoints.go +++ b/datasets/v1/datapoints.go @@ -335,7 +335,7 @@ func validateDatapoints(datapoints any) error { if slice.Kind() != reflect.Slice { return fmt.Errorf("datapoints must be a pointer to a slice, got %v", reflect.TypeOf(datapoints)) } - if slice.Type().Elem().Kind() != reflect.Ptr { + if slice.Type().Elem().Kind() != reflect.Pointer { return fmt.Errorf("datapoints must be a pointer to a slice of proto.Message, got %v", reflect.TypeOf(datapoints)) } messageType := reflect.TypeFor[proto.Message]() diff --git a/examples/README.md b/examples/README.md index 45574b5..fa2e3bb 100644 --- a/examples/README.md +++ b/examples/README.md @@ -12,21 +12,24 @@ Head over to [Tilebox Console](https://console.tilebox.com/account/api-keys) if Each example can be run using the following command: ```bash -go run ./ +go run ./examples/ ``` -For example, to run the `workflows/helloworld/submitter` example: -x +For example, to run the `workflows/helloworld` example: + ```bash -go run ./workflows/helloworld/submitter +go run ./examples/workflows/helloworld ``` +Workflow examples submit a job first, then start a task runner in the same process. + ## Workflows examples - [Hello world](workflows/helloworld): How to submit a task and run it. +- [MapReduce](workflows/mapreduce): How to fan out tasks and submit dependent reduce tasks. +- [Progress](workflows/progress): How to report workflow progress. - [Protobuf tasks](workflows/protobuf-task): How to use Protobuf tasks. -- [Axiom Observability](workflows/axiom): How to set up tracing and logging for workflows using [Axiom](https://axiom.co/) observability platform. -- [OpenTelemetry Observability](workflows/opentelemetry): How to set up tracing and logging for workflows using [OpenTelemetry](https://opentelemetry.io/). +- [Observability](workflows/observability): How to query Tilebox workflow telemetry and export traces/logs to custom Axiom or OpenTelemetry backends. ## Datasets examples diff --git a/examples/datasets/create/main.go b/examples/datasets/create/main.go index f1e2caf..de0b013 100644 --- a/examples/datasets/create/main.go +++ b/examples/datasets/create/main.go @@ -26,7 +26,7 @@ func main() { }, ) if err != nil { - slog.Error("Failed to create dataset", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to create dataset", slog.Any("error", err)) return } slog.InfoContext(ctx, "Created dataset", slog.String("dataset_id", dataset.ID.String())) @@ -44,7 +44,7 @@ func main() { }, ) if err != nil { - slog.Error("Failed to update dataset", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to update dataset", slog.Any("error", err)) return } diff --git a/examples/datasets/ingest/main.go b/examples/datasets/ingest/main.go index 1e26b75..38dba7b 100644 --- a/examples/datasets/ingest/main.go +++ b/examples/datasets/ingest/main.go @@ -20,14 +20,14 @@ func main() { // Select a dataset dataset, err := client.Datasets.Get(ctx, "open_data.copernicus.sentinel2_msi") if err != nil { - slog.Error("Failed to get dataset", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to get dataset", slog.Any("error", err)) return } // Create a collection collection, err := client.Collections.Create(ctx, dataset.ID, "My First Collection") if err != nil { - slog.Error("Failed to create collection", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to create collection", slog.Any("error", err)) return } @@ -47,16 +47,16 @@ func main() { // Ingest datapoints ingestResponse, err := client.Datapoints.Ingest(ctx, collection.ID, &datapoints, false) if err != nil { - slog.Error("Failed to ingest datapoints", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to ingest datapoints", slog.Any("error", err)) return } - slog.Info("Ingested datapoints", slog.Int64("created", ingestResponse.NumCreated)) + slog.InfoContext(ctx, "Ingested datapoints", slog.Int64("created", ingestResponse.NumCreated)) // Delete datapoints again numDeleted, err := client.Datapoints.DeleteIDs(ctx, collection.ID, ingestResponse.DatapointIDs) if err != nil { - slog.Error("Failed to delete datapoints", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to delete datapoints", slog.Any("error", err)) return } - slog.Info("Deleted datapoints", slog.Int64("deleted", numDeleted)) + slog.InfoContext(ctx, "Deleted datapoints", slog.Int64("deleted", numDeleted)) } diff --git a/examples/datasets/query/main.go b/examples/datasets/query/main.go index 91b2e46..97f0f6d 100644 --- a/examples/datasets/query/main.go +++ b/examples/datasets/query/main.go @@ -21,14 +21,14 @@ func main() { // Select a dataset dataset, err := client.Datasets.Get(ctx, "open_data.copernicus.sentinel2_msi") if err != nil { - slog.Error("Failed to get dataset", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to get dataset", slog.Any("error", err)) return } // Select a collection collection, err := client.Collections.Get(ctx, dataset.ID, "S2A_S2MSI1C") if err != nil { - slog.Error("Failed to get collection", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to get collection", slog.Any("error", err)) return } @@ -52,13 +52,13 @@ func main() { datasets.WithSpatialExtent(area), ) if err != nil { - slog.Error("Failed to query datapoints", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to query datapoints", slog.Any("error", err)) return } - slog.Info("Found datapoints over Colorado in March 2025", slog.Int("count", len(foundDatapoints))) + slog.InfoContext(ctx, "Found datapoints over Colorado in March 2025", slog.Int("count", len(foundDatapoints))) if len(foundDatapoints) > 0 { - slog.Info("First datapoint over Colorado", + slog.InfoContext(ctx, "First datapoint over Colorado", slog.String("id", foundDatapoints[0].GetId().AsUUID().String()), slog.Time("event time", foundDatapoints[0].GetTime().AsTime()), slog.Time("ingestion time", foundDatapoints[0].GetIngestionTime().AsTime()), diff --git a/examples/workflows/axiom/submitter/main.go b/examples/workflows/axiom/submitter/main.go deleted file mode 100644 index 0b05fc8..0000000 --- a/examples/workflows/axiom/submitter/main.go +++ /dev/null @@ -1,55 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/axiom" - "github.com/tilebox/tilebox-go/observability" - "github.com/tilebox/tilebox-go/observability/logger" - "github.com/tilebox/tilebox-go/observability/tracer" - "github.com/tilebox/tilebox-go/workflows/v1" - "go.opentelemetry.io/otel" -) - -// specify a service name and version to identify the instrumenting application in traces and logs -var service = &observability.Service{Name: "submitter", Version: "dev"} - -func main() { - ctx := context.Background() - - // Setup OpenTelemetry logging and slog - axiomHandler, shutdownLogger, err := logger.NewAxiomHandlerFromEnv(ctx, service, - logger.WithLevel(slog.LevelInfo), // export logs at info level and above as OTEL logs - ) - defer shutdownLogger(ctx) - if err != nil { - slog.Error("failed to set up axiom log handler", slog.Any("error", err)) - return - } - tileboxLogger := logger.New( // initialize a slog.Logger - axiomHandler, // export logs to Axiom - logger.NewConsoleHandler(logger.WithLevel(slog.LevelWarn)), // and additionally, export WARN and ERROR logs to stdout - ) - slog.SetDefault(tileboxLogger) // all future slog calls will be forwarded to the tilebox logger - - // Setup an OpenTelemetry trace span processor, exporting traces and spans to Axiom - tileboxTracerProvider, shutdown, err := tracer.NewAxiomProviderFromEnv(ctx, service) - defer shutdown(ctx) - if err != nil { - slog.Error("failed to set up axiom tracer provider", slog.Any("error", err)) - return - } - otel.SetTracerProvider(tileboxTracerProvider) // set the tilebox tracer provider as the global OTEL tracer provider - - client := workflows.NewClient() - job, err := client.Jobs.Submit(ctx, "Axiom Observability", - []workflows.Task{&axiom.MyAxiomTask{}}, - ) - if err != nil { - slog.Error("Failed to submit job", "error", err) - return - } - - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) -} diff --git a/examples/workflows/axiom/task.go b/examples/workflows/axiom/task.go deleted file mode 100644 index b4f9a62..0000000 --- a/examples/workflows/axiom/task.go +++ /dev/null @@ -1,25 +0,0 @@ -package axiom - -import ( - "context" - "fmt" - "log/slog" - - "github.com/tilebox/tilebox-go/workflows/v1" -) - -type MyAxiomTask struct{} - -func (t *MyAxiomTask) Execute(ctx context.Context) error { - // Start an openTelemetry tracing span that will be exported to Axiom - result, err := workflows.WithTaskSpanResult(ctx, "Expensive Compute", func(ctx context.Context) (int, error) { - return 6 * 7, nil - }) - if err != nil { - return fmt.Errorf("failed to compute: %w", err) - } - - // Log the result to Axiom - slog.InfoContext(ctx, "Computed result", slog.Int("result", result)) - return nil -} diff --git a/examples/workflows/helloworld/README.md b/examples/workflows/helloworld/README.md new file mode 100644 index 0000000..961f960 --- /dev/null +++ b/examples/workflows/helloworld/README.md @@ -0,0 +1,11 @@ +# Hello World Workflow + +Minimal workflow example that submits and runs a single `HelloTask`. + +- [main.go](main.go) contains the task definition, submits a job, and then starts a task runner to execute it. + +Run the example: + +```bash +go run ./examples/workflows/helloworld +``` diff --git a/examples/workflows/helloworld/helloworld.go b/examples/workflows/helloworld/helloworld.go deleted file mode 100644 index d08be60..0000000 --- a/examples/workflows/helloworld/helloworld.go +++ /dev/null @@ -1,16 +0,0 @@ -package helloworld - -import ( - "context" - "log/slog" -) - -type HelloTask struct { - Name string -} - -// The Execute method isn't needed to submit a task but is required to run a task on a task runner. -func (t *HelloTask) Execute(context.Context) error { - slog.Info("Hello World!", slog.String("Name", t.Name)) - return nil -} diff --git a/examples/workflows/helloworld/main.go b/examples/workflows/helloworld/main.go new file mode 100644 index 0000000..3f8aa5f --- /dev/null +++ b/examples/workflows/helloworld/main.go @@ -0,0 +1,51 @@ +package main + +import ( + "context" + "log/slog" + + "github.com/tilebox/tilebox-go/workflows/v1" +) + +type HelloTask struct { + Name string +} + +// The Execute method isn't needed to submit a task but is required to run a task on a task runner. +func (t *HelloTask) Execute(ctx context.Context) error { + slog.InfoContext(ctx, "Hello World!", slog.String("Name", t.Name)) + return nil +} + +func main() { + ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) + client := workflows.NewClient() + + job, err := client.Jobs.Submit(ctx, "hello-world", + []workflows.Task{ + &HelloTask{ + Name: "Tilebox", + }, + }, + ) + if err != nil { + slog.ErrorContext(ctx, "Failed to submit job", slog.Any("error", err)) + return + } + + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) + + runner, err := client.NewTaskRunner(ctx) + if err != nil { + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) + return + } + + if err := runner.RegisterTasks(&HelloTask{}); err != nil { + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) + return + } + + runner.RunForever(ctx) +} diff --git a/examples/workflows/helloworld/runner/main.go b/examples/workflows/helloworld/runner/main.go deleted file mode 100644 index 5fd1428..0000000 --- a/examples/workflows/helloworld/runner/main.go +++ /dev/null @@ -1,28 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/helloworld" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - runner, err := client.NewTaskRunner(ctx) - if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) - return - } - - err = runner.RegisterTasks(&helloworld.HelloTask{}) - if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) - return - } - - runner.RunForever(ctx) -} diff --git a/examples/workflows/helloworld/submitter/main.go b/examples/workflows/helloworld/submitter/main.go deleted file mode 100644 index 1bc402e..0000000 --- a/examples/workflows/helloworld/submitter/main.go +++ /dev/null @@ -1,28 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/helloworld" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - job, err := client.Jobs.Submit(ctx, "hello-world", - []workflows.Task{ - &helloworld.HelloTask{ - Name: "Tilebox", - }, - }, - ) - if err != nil { - slog.Error("Failed to submit job", slog.Any("error", err)) - return - } - - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) -} diff --git a/examples/workflows/mapreduce/README.md b/examples/workflows/mapreduce/README.md new file mode 100644 index 0000000..05e2eaa --- /dev/null +++ b/examples/workflows/mapreduce/README.md @@ -0,0 +1,11 @@ +# MapReduce Workflow + +Example workflow that fans out many map tasks and then submits reduce tasks that depend on the map tasks. + +- [main.go](main.go) contains the task definitions, submits a job, and then starts a task runner to execute it. + +Run the example: + +```bash +go run ./examples/workflows/mapreduce +``` diff --git a/examples/workflows/mapreduce/mapreduce.go b/examples/workflows/mapreduce/main.go similarity index 52% rename from examples/workflows/mapreduce/mapreduce.go rename to examples/workflows/mapreduce/main.go index ab46f22..69aae08 100644 --- a/examples/workflows/mapreduce/mapreduce.go +++ b/examples/workflows/mapreduce/main.go @@ -1,4 +1,4 @@ -package mapreduce +package main import ( "context" @@ -15,7 +15,7 @@ type RootTask struct { } func (t *RootTask) Execute(ctx context.Context) error { - slog.Info("Starting MapReduce job") + slog.InfoContext(ctx, "Starting MapReduce job") mapTasks := make([]workflows.Task, t.NumMapTasks) for i := range t.NumMapTasks { @@ -27,7 +27,7 @@ func (t *RootTask) Execute(ctx context.Context) error { return err } - slog.Info("Submitted map tasks", slog.Int("count", len(mapFutures))) + slog.InfoContext(ctx, "Submitted map tasks", slog.Int("count", len(mapFutures))) mapProgress := workflows.Progress("map") _ = mapProgress.Add(ctx, uint64(len(mapFutures))) @@ -41,7 +41,7 @@ func (t *RootTask) Execute(ctx context.Context) error { } } - slog.Info("Submitted reduce tasks", slog.Uint64("count", t.NumReduceTasks)) + slog.InfoContext(ctx, "Submitted reduce tasks", slog.Uint64("count", t.NumReduceTasks)) reduceProgress := workflows.Progress("reduce") _ = reduceProgress.Add(ctx, t.NumReduceTasks) @@ -55,7 +55,7 @@ type MapTask struct { } func (t *MapTask) Execute(ctx context.Context) error { - slog.Info("Executing map task", slog.Uint64("index", t.Index)) + slog.InfoContext(ctx, "Executing map task", slog.Uint64("index", t.Index)) _ = workflows.Progress("map").Done(ctx, 1) return nil } @@ -66,7 +66,45 @@ type ReduceTask struct { } func (t *ReduceTask) Execute(ctx context.Context) error { - slog.Info("Executing reduce task", slog.Uint64("reducer_index", t.ReducerIndex)) + slog.InfoContext(ctx, "Executing reduce task", slog.Uint64("reducer_index", t.ReducerIndex)) _ = workflows.Progress("reduce").Done(ctx, 1) return nil } + +func main() { + ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) + client := workflows.NewClient() + + job, err := client.Jobs.Submit(ctx, "mapreduce-example", + []workflows.Task{ + &RootTask{ + NumMapTasks: 100, + NumReduceTasks: 8, + }, + }, + ) + if err != nil { + slog.ErrorContext(ctx, "Failed to submit job", slog.Any("error", err)) + return + } + + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) + + runner, err := client.NewTaskRunner(ctx) + if err != nil { + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) + return + } + + if err := runner.RegisterTasks( + &RootTask{}, + &MapTask{}, + &ReduceTask{}, + ); err != nil { + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) + return + } + + runner.RunForever(ctx) +} diff --git a/examples/workflows/mapreduce/runner/main.go b/examples/workflows/mapreduce/runner/main.go deleted file mode 100644 index 0939c0d..0000000 --- a/examples/workflows/mapreduce/runner/main.go +++ /dev/null @@ -1,32 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/mapreduce" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - runner, err := client.NewTaskRunner(ctx) - if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) - return - } - - err = runner.RegisterTasks( - &mapreduce.RootTask{}, - &mapreduce.MapTask{}, - &mapreduce.ReduceTask{}, - ) - if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) - return - } - - runner.RunForever(ctx) -} diff --git a/examples/workflows/mapreduce/submitter/main.go b/examples/workflows/mapreduce/submitter/main.go deleted file mode 100644 index 5f1e203..0000000 --- a/examples/workflows/mapreduce/submitter/main.go +++ /dev/null @@ -1,29 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/mapreduce" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - job, err := client.Jobs.Submit(ctx, "mapreduce-example", - []workflows.Task{ - &mapreduce.RootTask{ - NumMapTasks: 100, - NumReduceTasks: 8, - }, - }, - ) - if err != nil { - slog.Error("Failed to submit job", slog.Any("error", err)) - return - } - - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) -} diff --git a/examples/workflows/observability/README.md b/examples/workflows/observability/README.md new file mode 100644 index 0000000..31bb1ee --- /dev/null +++ b/examples/workflows/observability/README.md @@ -0,0 +1,9 @@ +# Workflow Observability + +Tilebox-native workflow observability is automatically configured by `workflows.NewClient()` for the standard examples. + +This folder contains examples for querying workflow telemetry and for exporting telemetry to custom third-party backends: + +- [query](query): Query logs and traces for an existing workflow job. +- [axiom](axiom): Export workflow traces and logs to [Axiom](https://axiom.co/). +- [opentelemetry](opentelemetry): Export workflow traces and logs to an OpenTelemetry-compatible endpoint. diff --git a/examples/workflows/axiom/README.md b/examples/workflows/observability/axiom/README.md similarity index 69% rename from examples/workflows/axiom/README.md rename to examples/workflows/observability/axiom/README.md index e72dee6..cdbb3cc 100644 --- a/examples/workflows/axiom/README.md +++ b/examples/workflows/observability/axiom/README.md @@ -4,12 +4,16 @@ How to set up tracing and logging for workflows using [Axiom](https://axiom.co/) It uses [OpenTelemetry](https://opentelemetry.io/) to add tracing and the standard [slog](https://pkg.go.dev/log/slog) package for logging. -- [submitter/main.go](submitter/main.go) sets up tracing and logging and submits a job. -- [runner/main.go](runner/main.go) sets up tracing and logging for a runner. -- [task.go](task.go) shows how to use logs and traces within a task. +- [main.go](main.go) sets up tracing and logging, submits a job, starts a task runner to execute it, and shows how to use logs and traces within a task. To run the example, you need to set the additional following environment variables: - `AXIOM_API_KEY` to your Axiom API key. - `AXIOM_TRACES_DATASET` to the dataset where traces should be stored. - `AXIOM_LOGS_DATASET` to the dataset where logs should be stored. + +Run the example: + +```bash +go run ./examples/workflows/observability/axiom +``` diff --git a/examples/workflows/axiom/runner/main.go b/examples/workflows/observability/axiom/main.go similarity index 51% rename from examples/workflows/axiom/runner/main.go rename to examples/workflows/observability/axiom/main.go index 0b82881..a57490e 100644 --- a/examples/workflows/axiom/runner/main.go +++ b/examples/workflows/observability/axiom/main.go @@ -2,9 +2,9 @@ package main import ( "context" + "fmt" "log/slog" - "github.com/tilebox/tilebox-go/examples/workflows/axiom" "github.com/tilebox/tilebox-go/observability" "github.com/tilebox/tilebox-go/observability/logger" "github.com/tilebox/tilebox-go/observability/tracer" @@ -13,7 +13,23 @@ import ( ) // specify a service name and version to identify the instrumenting application in traces and logs -var service = &observability.Service{Name: "task-runner", Version: "dev"} +var service = &observability.Service{Name: "axiom-example", Version: "dev"} + +type MyAxiomTask struct{} + +func (t *MyAxiomTask) Execute(ctx context.Context) error { + // Start an openTelemetry tracing span that will be exported to Axiom + result, err := workflows.WithSpanResult(ctx, "compute", func(ctx context.Context) (int, error) { + return 6 * 7, nil + }) + if err != nil { + return fmt.Errorf("failed to compute: %w", err) + } + + // Log the result to Axiom + slog.InfoContext(ctx, "Computed result", slog.Int("result", result)) + return nil +} func main() { ctx := context.Background() @@ -24,34 +40,44 @@ func main() { ) defer shutdownLogger(ctx) if err != nil { - slog.Error("failed to set up axiom log handler", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to set up axiom log handler", slog.Any("error", err)) return } tileboxLogger := logger.New( // initialize a slog.Logger - axiomHandler, // export logs to the Axiom handler - logger.NewConsoleHandler(logger.WithLevel(slog.LevelWarn)), // and additionally, export WARN and ERROR logs to stdout + axiomHandler, // export logs to Axiom ) slog.SetDefault(tileboxLogger) // all future slog calls will be forwarded to the tilebox logger + workflows.ConfigureConsoleLogging(slog.LevelWarn) // Setup an OpenTelemetry trace span processor, exporting traces and spans to Axiom tileboxTracerProvider, shutdown, err := tracer.NewAxiomProviderFromEnv(ctx, service) defer shutdown(ctx) if err != nil { - slog.Error("failed to set up axiom tracer provider", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to set up axiom tracer provider", slog.Any("error", err)) return } otel.SetTracerProvider(tileboxTracerProvider) // set the tilebox tracer provider as the global OTEL tracer provider client := workflows.NewClient() - taskRunner, err := client.NewTaskRunner(ctx) + + job, err := client.Jobs.Submit(ctx, "Axiom Observability", + []workflows.Task{&MyAxiomTask{}}, + ) if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to submit job", "error", err) return } - err = taskRunner.RegisterTasks(&axiom.MyAxiomTask{}) + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) + + taskRunner, err := client.NewTaskRunner(ctx) if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) + return + } + + if err := taskRunner.RegisterTasks(&MyAxiomTask{}); err != nil { + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) return } diff --git a/examples/workflows/opentelemetry/README.md b/examples/workflows/observability/opentelemetry/README.md similarity index 51% rename from examples/workflows/opentelemetry/README.md rename to examples/workflows/observability/opentelemetry/README.md index a798c93..ab0e263 100644 --- a/examples/workflows/opentelemetry/README.md +++ b/examples/workflows/observability/opentelemetry/README.md @@ -4,6 +4,10 @@ How to set up tracing and logging for workflows using OpenTelemetry. It uses [OpenTelemetry](https://opentelemetry.io/) to add tracing and logging and the standard [slog](https://pkg.go.dev/log/slog) package for logging. -- [submitter/main.go](submitter/main.go) sets up tracing and logging and submits a job. -- [runner/main.go](runner/main.go) sets up tracing and logging for a runner. -- [task.go](task.go) shows how to use logs and traces within a task. +- [main.go](main.go) sets up tracing and logging, submits a job, starts a task runner to execute it, and shows how to use logs and traces within a task. + +Run the example: + +```bash +go run ./examples/workflows/observability/opentelemetry +``` diff --git a/examples/workflows/opentelemetry/runner/main.go b/examples/workflows/observability/opentelemetry/main.go similarity index 56% rename from examples/workflows/opentelemetry/runner/main.go rename to examples/workflows/observability/opentelemetry/main.go index 5f454db..6a49e4e 100644 --- a/examples/workflows/opentelemetry/runner/main.go +++ b/examples/workflows/observability/opentelemetry/main.go @@ -2,9 +2,9 @@ package main import ( "context" + "fmt" "log/slog" - "github.com/tilebox/tilebox-go/examples/workflows/opentelemetry" "github.com/tilebox/tilebox-go/observability" "github.com/tilebox/tilebox-go/observability/logger" "github.com/tilebox/tilebox-go/observability/tracer" @@ -13,7 +13,23 @@ import ( ) // specify a service name and version to identify the instrumenting application in traces and logs -var service = &observability.Service{Name: "task-runner", Version: "dev"} +var service = &observability.Service{Name: "opentelemetry-example", Version: "dev"} + +type MyOpenTelemetryTask struct{} + +func (t *MyOpenTelemetryTask) Execute(ctx context.Context) error { + // Start an openTelemetry tracing span that will be exported to the OpenTelemetry endpoint + result, err := workflows.WithSpanResult(ctx, "compute", func(ctx context.Context) (int, error) { + return 6 * 7, nil + }) + if err != nil { + return fmt.Errorf("failed to compute: %w", err) + } + + // Log the result to the OpenTelemetry endpoint + slog.InfoContext(ctx, "Computed result", slog.Int("result", result)) + return nil +} func main() { ctx := context.Background() @@ -31,14 +47,14 @@ func main() { ) defer shutdownLogger(ctx) if err != nil { - slog.Error("failed to set up otel log handler", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to set up otel log handler", slog.Any("error", err)) return } tileboxLogger := logger.New( // initialize a slog.Logger otelHandler, // export logs to the OTEL handler - logger.NewConsoleHandler(logger.WithLevel(slog.LevelWarn)), // and additionally, export WARN and ERROR logs to stdout ) slog.SetDefault(tileboxLogger) // all future slog calls will be forwarded to the tilebox logger + workflows.ConfigureConsoleLogging(slog.LevelWarn) // Setup an OpenTelemetry trace span processor, exporting traces and spans to an OTEL compatible trace endpoint tileboxTracerProvider, shutdown, err := tracer.NewOtelProvider(ctx, service, @@ -47,21 +63,31 @@ func main() { ) defer shutdown(ctx) if err != nil { - slog.Error("failed to set up otel span processor", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to set up otel span processor", slog.Any("error", err)) return } otel.SetTracerProvider(tileboxTracerProvider) // set the tilebox tracer provider as the global OTEL tracer provider client := workflows.NewClient() - taskRunner, err := client.NewTaskRunner(ctx) + + job, err := client.Jobs.Submit(ctx, "OpenTelemetry Observability", + []workflows.Task{&MyOpenTelemetryTask{}}, + ) if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to submit job", "error", err) return } - err = taskRunner.RegisterTasks(&opentelemetry.MyOpenTelemetryTask{}) + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) + + taskRunner, err := client.NewTaskRunner(ctx) if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) + return + } + + if err := taskRunner.RegisterTasks(&MyOpenTelemetryTask{}); err != nil { + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) return } diff --git a/examples/workflows/telemetry/query/main.go b/examples/workflows/observability/query/main.go similarity index 88% rename from examples/workflows/telemetry/query/main.go rename to examples/workflows/observability/query/main.go index ee8224b..5e193e3 100644 --- a/examples/workflows/telemetry/query/main.go +++ b/examples/workflows/observability/query/main.go @@ -12,6 +12,7 @@ import ( func main() { ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) client := workflows.NewClient() jobID := uuid.MustParse("019e070c-63ba-1c7e-5f1d-65be3e22d52a") @@ -19,7 +20,7 @@ func main() { fmt.Printf("%-8s %-30s %s\n", "LEVEL", "TIME", "BODY") //nolint:forbidigo // This example intentionally prints query results to stdout. for logRecord, err := range client.Jobs.QueryLogs(ctx, jobID, workflows.WithSortDirection(workflows.Ascending)) { if err != nil { - slog.Error("failed to query job logs", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to query job logs", slog.Any("error", err)) return } @@ -35,7 +36,7 @@ func main() { fmt.Printf("%-30s %-40s %s\n", "TIME", "SPAN NAME", "DURATION") //nolint:forbidigo // This example intentionally prints query results to stdout. for span, err := range client.Jobs.QuerySpans(ctx, jobID) { if err != nil { - slog.Error("failed to query job spans", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to query job spans", slog.Any("error", err)) return } diff --git a/examples/workflows/opentelemetry/submitter/main.go b/examples/workflows/opentelemetry/submitter/main.go deleted file mode 100644 index dd816ac..0000000 --- a/examples/workflows/opentelemetry/submitter/main.go +++ /dev/null @@ -1,65 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - "github.com/tilebox/tilebox-go/examples/workflows/axiom" - "github.com/tilebox/tilebox-go/observability" - "github.com/tilebox/tilebox-go/observability/logger" - "github.com/tilebox/tilebox-go/observability/tracer" - "github.com/tilebox/tilebox-go/workflows/v1" - "go.opentelemetry.io/otel" -) - -// specify a service name and version to identify the instrumenting application in traces and logs -var service = &observability.Service{Name: "submitter", Version: "dev"} - -func main() { - ctx := context.Background() - - endpoint := "http://localhost:4318" - headers := map[string]string{ - "Authorization": "Bearer ", - } - - // Setup an OpenTelemetry log handler, exporting logs to an OTEL compatible log endpoint - otelHandler, shutdownLogger, err := logger.NewOtelHandler(ctx, service, - logger.WithEndpointURL(endpoint), - logger.WithHeaders(headers), - logger.WithLevel(slog.LevelInfo), // export logs at info level and above as OTEL logs - ) - defer shutdownLogger(ctx) - if err != nil { - slog.Error("failed to set up otel log handler", slog.Any("error", err)) - return - } - tileboxLogger := logger.New( // initialize a slog.Logger - otelHandler, // export logs to the OTEL handler - logger.NewConsoleHandler(logger.WithLevel(slog.LevelWarn)), // and additionally, export WARN and ERROR logs to stdout - ) - slog.SetDefault(tileboxLogger) // all future slog calls will be forwarded to the tilebox logger - - // Setup an OpenTelemetry trace span processor, exporting traces and spans to an OTEL compatible trace endpoint - tileboxTracerProvider, shutdown, err := tracer.NewOtelProvider(ctx, service, - tracer.WithEndpointURL(endpoint), - tracer.WithHeaders(headers), - ) - defer shutdown(ctx) - if err != nil { - slog.Error("failed to set up otel span processor", slog.Any("error", err)) - return - } - otel.SetTracerProvider(tileboxTracerProvider) // set the tilebox tracer provider as the global OTEL tracer provider - - client := workflows.NewClient() - job, err := client.Jobs.Submit(ctx, "Axiom Observability", - []workflows.Task{&axiom.MyAxiomTask{}}, - ) - if err != nil { - slog.Error("Failed to submit job", "error", err) - return - } - - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) -} diff --git a/examples/workflows/opentelemetry/task.go b/examples/workflows/opentelemetry/task.go deleted file mode 100644 index bb467f2..0000000 --- a/examples/workflows/opentelemetry/task.go +++ /dev/null @@ -1,25 +0,0 @@ -package opentelemetry - -import ( - "context" - "fmt" - "log/slog" - - "github.com/tilebox/tilebox-go/workflows/v1" -) - -type MyOpenTelemetryTask struct{} - -func (t *MyOpenTelemetryTask) Execute(ctx context.Context) error { - // Start an openTelemetry tracing span that will be exported to the OpenTelemetry endpoint - result, err := workflows.WithTaskSpanResult(ctx, "Expensive Compute", func(ctx context.Context) (int, error) { - return 6 * 7, nil - }) - if err != nil { - return fmt.Errorf("failed to compute: %w", err) - } - - // Log the result to the OpenTelemetry endpoint - slog.InfoContext(ctx, "Computed result", slog.Int("result", result)) - return nil -} diff --git a/examples/workflows/progress/README.md b/examples/workflows/progress/README.md new file mode 100644 index 0000000..2738ff1 --- /dev/null +++ b/examples/workflows/progress/README.md @@ -0,0 +1,11 @@ +# Progress Workflow + +Example workflow that reports task progress while running subtasks. + +- [main.go](main.go) submits the workflow, starts a task runner with `RunAll` to execute it, and prints the final progress. + +Run the example: + +```bash +go run ./examples/workflows/progress +``` diff --git a/examples/workflows/progress/progress.go b/examples/workflows/progress/main.go similarity index 62% rename from examples/workflows/progress/progress.go rename to examples/workflows/progress/main.go index 55067e2..0dba852 100644 --- a/examples/workflows/progress/progress.go +++ b/examples/workflows/progress/main.go @@ -3,7 +3,6 @@ package main import ( "context" "log/slog" - "os" "github.com/tilebox/tilebox-go/workflows/v1" ) @@ -13,7 +12,7 @@ type ProgressRootTask struct { } func (t *ProgressRootTask) Execute(ctx context.Context) error { - // report that 10 units of work need to be done + // report how many units of work need to be done err := workflows.DefaultProgress().Add(ctx, t.N) if err != nil { return err @@ -42,33 +41,32 @@ func (t *ProgressSubTask) Execute(ctx context.Context) error { } func main() { - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))) - ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelDebug) client := workflows.NewClient() job, err := client.Jobs.Submit(ctx, "progress-example", []workflows.Task{&ProgressRootTask{N: 5}}, ) if err != nil { - slog.Error("Failed to submit job", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to submit job", slog.Any("error", err)) return } - slog.Info("Job submitted", slog.String("job_id", job.ID.String())) + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) if len(job.Progress) == 0 { - slog.Info("No job progress yet") + slog.InfoContext(ctx, "No job progress yet") } runner, err := client.NewTaskRunner(ctx) if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) return } err = runner.RegisterTasks(&ProgressRootTask{}, &ProgressSubTask{}) if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) return } @@ -76,12 +74,12 @@ func main() { job, err = client.Jobs.Get(ctx, job.ID) // fetch up-to-date job progress if err != nil { - slog.Error("Failed to get job", slog.Any("error", err)) + slog.ErrorContext(ctx, "Failed to get job", slog.Any("error", err)) return } - slog.Info("Job completed", slog.String("job_id", job.ID.String())) + slog.InfoContext(ctx, "Job completed", slog.String("job_id", job.ID.String())) for _, progress := range job.Progress { - slog.Info("Job progress", slog.String("label", progress.Label), slog.Uint64("total", progress.Total), slog.Uint64("done", progress.Done)) + slog.InfoContext(ctx, "Job progress", slog.String("label", progress.Label), slog.Uint64("total", progress.Total), slog.Uint64("done", progress.Done)) } } diff --git a/examples/workflows/protobuf-task/README.md b/examples/workflows/protobuf-task/README.md index b64071b..b80bf83 100644 --- a/examples/workflows/protobuf-task/README.md +++ b/examples/workflows/protobuf-task/README.md @@ -2,7 +2,7 @@ This example uses a protobuf message to define `SpawnWorkflowTreeTask` task. -The task definition is defined in [workflow.proto](../../../apis/examples/v1/workflow.proto) and the task implementation is in [example_workflow.go](tasks.go). +The task definition is defined in [workflow.proto](../../../apis/examples/v1/workflow.proto) and the task implementation is in [main.go](main.go). The generated code of `workflow.proto` is in [workflow.pb.go](../../../protogen/examples/v1/workflow.pb.go) and can be (re-)generated using the following command: @@ -10,6 +10,10 @@ The generated code of `workflow.proto` is in [workflow.pb.go](../../../protogen/ go generate ./... ``` -- [submitter/main.go](submitter/main.go) submits a workflow to Tilebox Workflows with a protobuf root task. -- [runner/main.go](runner/main.go) starts a task runner that can execute both protobuf and regular struct tasks. -- [tasks.go](tasks.go) shows the protobuf task implementation and how to submit protobuf sub-tasks. +- [main.go](main.go) submits a workflow to Tilebox Workflows, starts a task runner to execute it, and shows the protobuf task implementation and how to submit protobuf sub-tasks. + +Run the example: + +```bash +go run ./examples/workflows/protobuf-task +``` diff --git a/examples/workflows/protobuf-task/tasks.go b/examples/workflows/protobuf-task/main.go similarity index 64% rename from examples/workflows/protobuf-task/tasks.go rename to examples/workflows/protobuf-task/main.go index 984fdd6..8c062be 100644 --- a/examples/workflows/protobuf-task/tasks.go +++ b/examples/workflows/protobuf-task/main.go @@ -1,4 +1,4 @@ -package pbtask +package main import ( "context" @@ -61,3 +61,41 @@ func (t *SpawnWorkflowTreeTask) Execute(ctx context.Context) error { _, err := workflows.SubmitSubtasks(ctx, subtasks) return err } + +func main() { + ctx := context.Background() + workflows.ConfigureConsoleLogging(slog.LevelInfo) + client := workflows.NewClient() + + job, err := client.Jobs.Submit(ctx, "spawn-workflow-tree", + []workflows.Task{ + &SampleTask{ // regular struct task + Message: "hello go runner!", + Depth: 8, + BranchFactor: 4, + }, + }, + ) + if err != nil { + slog.ErrorContext(ctx, "Failed to submit job", "error", err) + return + } + + slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) + + taskRunner, err := client.NewTaskRunner(ctx) + if err != nil { + slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err)) + return + } + + if err := taskRunner.RegisterTasks( + &SampleTask{}, // regular struct task + &SpawnWorkflowTreeTask{}, // protobuf task + ); err != nil { + slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err)) + return + } + + taskRunner.RunForever(ctx) +} diff --git a/examples/workflows/protobuf-task/runner/main.go b/examples/workflows/protobuf-task/runner/main.go deleted file mode 100644 index c27c85d..0000000 --- a/examples/workflows/protobuf-task/runner/main.go +++ /dev/null @@ -1,31 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - pbtask "github.com/tilebox/tilebox-go/examples/workflows/protobuf-task" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - taskRunner, err := client.NewTaskRunner(ctx) - if err != nil { - slog.Error("failed to create task runner", slog.Any("error", err)) - return - } - - err = taskRunner.RegisterTasks( - &pbtask.SampleTask{}, // regular struct task - &pbtask.SpawnWorkflowTreeTask{}, // protobuf task - ) - if err != nil { - slog.Error("failed to register tasks", slog.Any("error", err)) - return - } - - taskRunner.RunForever(ctx) -} diff --git a/examples/workflows/protobuf-task/submitter/main.go b/examples/workflows/protobuf-task/submitter/main.go deleted file mode 100644 index 9f7cf3a..0000000 --- a/examples/workflows/protobuf-task/submitter/main.go +++ /dev/null @@ -1,30 +0,0 @@ -package main - -import ( - "context" - "log/slog" - - pbtask "github.com/tilebox/tilebox-go/examples/workflows/protobuf-task" - "github.com/tilebox/tilebox-go/workflows/v1" -) - -func main() { - ctx := context.Background() - client := workflows.NewClient() - - job, err := client.Jobs.Submit(ctx, "spawn-workflow-tree", - []workflows.Task{ - &pbtask.SampleTask{ // regular struct task - Message: "hello go runner!", - Depth: 8, - BranchFactor: 4, - }, - }, - ) - if err != nil { - slog.ErrorContext(ctx, "Failed to submit job", "error", err) - return - } - - slog.InfoContext(ctx, "Job submitted", slog.String("job_id", job.ID.String())) -} diff --git a/observability/logger/logging.go b/observability/logger/logging.go index fbcf7f1..f4d1691 100644 --- a/observability/logger/logging.go +++ b/observability/logger/logging.go @@ -15,8 +15,6 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" "go.opentelemetry.io/otel/sdk/log" - "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.30.0" oteltrace "go.opentelemetry.io/otel/trace" ) @@ -56,14 +54,8 @@ func NewOtelLogProcessor(ctx context.Context, options ...Option) (log.Processor, // NewLoggingProviderWithProcessors creates a new OpenTelemetry logger provider with the given processors. func NewLoggingProviderWithProcessors(otelService *observability.Service, processors ...log.Processor) *log.LoggerProvider { - rs := resource.NewWithAttributes( - semconv.SchemaURL, - semconv.ServiceNameKey.String(otelService.Name), - semconv.ServiceVersionKey.String(otelService.Version), - ) - opts := make([]log.LoggerProviderOption, 0, len(processors)+1) - opts = append(opts, log.WithResource(rs)) + opts = append(opts, log.WithResource(observability.NewResource(otelService))) for _, processor := range processors { opts = append(opts, log.WithProcessor(processor)) @@ -75,12 +67,35 @@ func NewLoggingProviderWithProcessors(otelService *observability.Service, proces // tracingAndlevelFilterHandler is a slog.Handler that filters log messages based on the configured level // and adds tracing information to each log record. type tracingAndlevelFilterHandler struct { - level slog.Level - next slog.Handler + level slog.Level + next slog.Handler + addSpanEvent bool } var _ slog.Handler = &tracingAndlevelFilterHandler{} +type contextSlogAttributesKeyType string + +const contextSlogAttributesKey contextSlogAttributesKeyType = "x-tilebox-ctx-slog-attributes" + +// ContextWithSlogAttributes returns a context whose log records include attrs. +func ContextWithSlogAttributes(ctx context.Context, attributes ...slog.Attr) context.Context { + if len(attributes) == 0 { + return ctx + } + + existing := contextAttributes(ctx) + contextAttributes := make([]slog.Attr, 0, len(existing)+len(attributes)) + contextAttributes = append(contextAttributes, existing...) + contextAttributes = append(contextAttributes, attributes...) + return context.WithValue(ctx, contextSlogAttributesKey, contextAttributes) +} + +func contextAttributes(ctx context.Context) []slog.Attr { + attrs, _ := ctx.Value(contextSlogAttributesKey).([]slog.Attr) + return attrs +} + // Enabled checks if a log message with the given level should be logged. func (h *tracingAndlevelFilterHandler) Enabled(ctx context.Context, level slog.Level) bool { if level < h.level { // if we are below the configured level, we for sure don't want to log this @@ -92,11 +107,26 @@ func (h *tracingAndlevelFilterHandler) Enabled(ctx context.Context, level slog.L // Handle processes a log record. func (h *tracingAndlevelFilterHandler) Handle(ctx context.Context, record slog.Record) error { - // add log info to span event + attrs := make([]slog.Attr, 0, 2) + attrs = append(attrs, contextAttributes(ctx)...) + + spanContext := oteltrace.SpanContextFromContext(ctx) + if spanContext.IsValid() { + attrs = append(attrs, + slog.String("trace_id", spanContext.TraceID().String()), + slog.String("span_id", spanContext.SpanID().String()), + ) + } + + if len(attrs) > 0 { + record = record.Clone() + record.AddAttrs(attrs...) + } + span := oteltrace.SpanFromContext(ctx) - if span != nil && span.IsRecording() { + if h.addSpanEvent && span.IsRecording() { eventAttrs := make([]attribute.KeyValue, 0, record.NumAttrs()) - eventAttrs = append(eventAttrs, attribute.String(slog.MessageKey, record.Message)) + eventAttrs = append(eventAttrs, attribute.String("body", record.Message)) eventAttrs = append(eventAttrs, attribute.String(slog.LevelKey, record.Level.String())) eventAttrs = append(eventAttrs, attribute.String(slog.TimeKey, record.Time.Format(time.RFC3339Nano))) record.Attrs(func(attr slog.Attr) bool { @@ -108,19 +138,20 @@ func (h *tracingAndlevelFilterHandler) Handle(ctx context.Context, record slog.R return true }) - span.AddEvent("log_record", oteltrace.WithAttributes(eventAttrs...)) + span.AddEvent("log.message", oteltrace.WithAttributes(eventAttrs...)) } + return h.next.Handle(ctx, record) } // WithGroup creates a new handler with the given group name. func (h *tracingAndlevelFilterHandler) WithGroup(name string) slog.Handler { - return &tracingAndlevelFilterHandler{level: h.level, next: h.next.WithGroup(name)} + return &tracingAndlevelFilterHandler{level: h.level, next: h.next.WithGroup(name), addSpanEvent: h.addSpanEvent} } // WithAttrs creates a new handler with the given attributes. func (h *tracingAndlevelFilterHandler) WithAttrs(attrs []slog.Attr) slog.Handler { - return &tracingAndlevelFilterHandler{level: h.level, next: h.next.WithAttrs(attrs)} + return &tracingAndlevelFilterHandler{level: h.level, next: h.next.WithAttrs(attrs), addSpanEvent: h.addSpanEvent} } func noShutdown(context.Context) {} @@ -140,8 +171,9 @@ func NewOtelHandler(ctx context.Context, otelService *observability.Service, opt loggerProvider := NewLoggingProviderWithProcessors(otelService, processor) return &tracingAndlevelFilterHandler{ - level: opts.Level, - next: otelslog.NewHandler(otelLoggerName, otelslog.WithLoggerProvider(loggerProvider)), + level: opts.Level, + next: otelslog.NewHandler(otelLoggerName, otelslog.WithLoggerProvider(loggerProvider)), + addSpanEvent: true, }, func(ctx context.Context) { _ = loggerProvider.Shutdown(ctx) @@ -187,7 +219,10 @@ func NewConsoleHandler(options ...Option) slog.Handler { option(opts) } - return tint.NewHandler(os.Stdout, &tint.Options{Level: opts.Level}) + return &tracingAndlevelFilterHandler{ + level: opts.Level, + next: tint.NewHandler(os.Stdout, &tint.Options{Level: opts.Level}), + } } // New creates a new slog.Logger with the given handlers. diff --git a/observability/logger/logging_test.go b/observability/logger/logging_test.go new file mode 100644 index 0000000..c3b2abb --- /dev/null +++ b/observability/logger/logging_test.go @@ -0,0 +1,86 @@ +package logger + +import ( + "context" + "log/slog" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +type recordingHandler struct { + records []slog.Record +} + +func (h *recordingHandler) Enabled(context.Context, slog.Level) bool { + return true +} + +func (h *recordingHandler) Handle(_ context.Context, record slog.Record) error { + h.records = append(h.records, record.Clone()) + return nil +} + +func (h *recordingHandler) WithAttrs([]slog.Attr) slog.Handler { + return h +} + +func (h *recordingHandler) WithGroup(string) slog.Handler { + return h +} + +func TestTracingAndLevelFilterHandlerAddsTraceAndContextAttrsAndSpanEvent(t *testing.T) { + spanRecorder := tracetest.NewSpanRecorder() + tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spanRecorder)) + tracer := tracerProvider.Tracer("test") + + ctx, span := tracer.Start(context.Background(), "test-span") + ctx = ContextWithSlogAttributes(ctx, slog.String("task_id", "019e070c-63b8-ce26-7a15-cec85a85de2b")) + + sink := &recordingHandler{} + logHandler := &tracingAndlevelFilterHandler{level: slog.LevelInfo, next: sink, addSpanEvent: true} + slog.New(logHandler).InfoContext(ctx, "hello", slog.String("custom", "value")) + span.End() + + require.Len(t, sink.records, 1) + logAttrs := slogRecordAttrs(sink.records[0]) + spanContext := span.SpanContext() + assert.Equal(t, spanContext.TraceID().String(), logAttrs["trace_id"]) + assert.Equal(t, spanContext.SpanID().String(), logAttrs["span_id"]) + assert.Equal(t, "019e070c-63b8-ce26-7a15-cec85a85de2b", logAttrs["task_id"]) + assert.Equal(t, "value", logAttrs["custom"]) + + endedSpans := spanRecorder.Ended() + require.Len(t, endedSpans, 1) + events := endedSpans[0].Events() + require.Len(t, events, 1) + assert.Equal(t, "log.message", events[0].Name) + eventAttrs := otelAttrs(events[0].Attributes) + assert.Equal(t, "hello", eventAttrs["body"]) + assert.Equal(t, "INFO", eventAttrs[slog.LevelKey]) + assert.Equal(t, spanContext.TraceID().String(), eventAttrs["trace_id"]) + assert.Equal(t, spanContext.SpanID().String(), eventAttrs["span_id"]) + assert.Equal(t, "019e070c-63b8-ce26-7a15-cec85a85de2b", eventAttrs["task_id"]) + assert.Equal(t, "value", eventAttrs["custom"]) +} + +func slogRecordAttrs(record slog.Record) map[string]string { + attrs := make(map[string]string, record.NumAttrs()) + record.Attrs(func(attr slog.Attr) bool { + attrs[attr.Key] = attr.Value.Resolve().String() + return true + }) + return attrs +} + +func otelAttrs(kvs []attribute.KeyValue) map[string]string { + attrs := make(map[string]string, len(kvs)) + for _, kv := range kvs { + attrs[string(kv.Key)] = kv.Value.AsString() + } + return attrs +} diff --git a/observability/resource.go b/observability/resource.go new file mode 100644 index 0000000..10e6c27 --- /dev/null +++ b/observability/resource.go @@ -0,0 +1,51 @@ +package observability + +import ( + "os" + "runtime" + + "github.com/google/uuid" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/resource" + semconv "go.opentelemetry.io/otel/semconv/v1.30.0" +) + +const ( + defaultServiceName = "tilebox-go" + defaultServiceNamespace = "tilebox.workflows" + defaultServiceVersion = "dev" +) + +// processInstanceID is a process-unique identifier that distinguishes different +// instances of the same service running on the same host. +var processInstanceID = uuid.NewString() + +// NewResource creates the default OpenTelemetry resource used for Tilebox workflow observability. +func NewResource(otelService *Service) *resource.Resource { + serviceName := defaultServiceName + serviceVersion := defaultServiceVersion + if otelService != nil { + if otelService.Name != "" { + serviceName = otelService.Name + } + if otelService.Version != "" { + serviceVersion = otelService.Version + } + } + + attrs := []attribute.KeyValue{ + semconv.ServiceNamespaceKey.String(defaultServiceNamespace), + semconv.ServiceNameKey.String(serviceName), + semconv.ServiceInstanceIDKey.String(processInstanceID), + semconv.ServiceVersionKey.String(serviceVersion), + semconv.ProcessPIDKey.Int(os.Getpid()), + semconv.HostArchKey.String(runtime.GOARCH), + semconv.OSTypeKey.String(runtime.GOOS), + } + + if hostname, err := os.Hostname(); err == nil && hostname != "" { + attrs = append(attrs, semconv.HostNameKey.String(hostname)) + } + + return resource.NewWithAttributes(semconv.SchemaURL, attrs...) +} diff --git a/observability/resource_test.go b/observability/resource_test.go new file mode 100644 index 0000000..6dc79da --- /dev/null +++ b/observability/resource_test.go @@ -0,0 +1,44 @@ +package observability + +import ( + "os" + "runtime" + "testing" + + "github.com/stretchr/testify/assert" + "go.opentelemetry.io/otel/attribute" + semconv "go.opentelemetry.io/otel/semconv/v1.30.0" +) + +func Test_NewResource(t *testing.T) { + res := NewResource(&Service{Name: "test-service", Version: "v1.2.3"}) + + attrs := make(map[attribute.Key]attribute.Value) + for _, attr := range res.Attributes() { + attrs[attr.Key] = attr.Value + } + + assert.Equal(t, "tilebox.workflows", attrs[semconv.ServiceNamespaceKey].AsString()) + assert.Equal(t, "test-service", attrs[semconv.ServiceNameKey].AsString()) + assert.Equal(t, processInstanceID, attrs[semconv.ServiceInstanceIDKey].AsString()) + assert.Equal(t, "v1.2.3", attrs[semconv.ServiceVersionKey].AsString()) + assert.Equal(t, int64(os.Getpid()), attrs[semconv.ProcessPIDKey].AsInt64()) + assert.Equal(t, runtime.GOARCH, attrs[semconv.HostArchKey].AsString()) + assert.Equal(t, runtime.GOOS, attrs[semconv.OSTypeKey].AsString()) + + if hostname, err := os.Hostname(); err == nil && hostname != "" { + assert.Equal(t, hostname, attrs[semconv.HostNameKey].AsString()) + } +} + +func Test_NewResource_Defaults(t *testing.T) { + res := NewResource(nil) + + attrs := make(map[attribute.Key]attribute.Value) + for _, attr := range res.Attributes() { + attrs[attr.Key] = attr.Value + } + + assert.Equal(t, defaultServiceName, attrs[semconv.ServiceNameKey].AsString()) + assert.Equal(t, defaultServiceVersion, attrs[semconv.ServiceVersionKey].AsString()) +} diff --git a/observability/tracer/tracing.go b/observability/tracer/tracing.go index dae93b5..46b38e5 100644 --- a/observability/tracer/tracing.go +++ b/observability/tracer/tracing.go @@ -9,9 +9,7 @@ import ( "github.com/tilebox/tilebox-go/observability" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" - "go.opentelemetry.io/otel/sdk/resource" "go.opentelemetry.io/otel/sdk/trace" - semconv "go.opentelemetry.io/otel/semconv/v1.30.0" ) const axiomTracesEndpoint = "https://api.axiom.co/v1/traces" @@ -50,14 +48,8 @@ func NewOtelSpanProcessor(ctx context.Context, options ...Option) (trace.SpanPro // NewOtelProviderWithProcessors creates a new OpenTelemetry tracer provider with the given processors. func NewOtelProviderWithProcessors(otelService *observability.Service, processors ...trace.SpanProcessor) *trace.TracerProvider { - rs := resource.NewWithAttributes( - semconv.SchemaURL, - semconv.ServiceNameKey.String(otelService.Name), - semconv.ServiceVersionKey.String(otelService.Version), - ) - opts := make([]trace.TracerProviderOption, 0, len(processors)+2) - opts = append(opts, trace.WithResource(rs)) + opts = append(opts, trace.WithResource(observability.NewResource(otelService))) // Ensure all traces are sampled. It a good default when generating little data (< 100 traces per second). // https://opentelemetry.io/docs/concepts/sampling/#when-not-to-sample opts = append(opts, trace.WithSampler(trace.AlwaysSample())) diff --git a/protogen/datasets/v1/core.pb.go b/protogen/datasets/v1/core.pb.go index 917d70e..fdc0752 100644 --- a/protogen/datasets/v1/core.pb.go +++ b/protogen/datasets/v1/core.pb.go @@ -1,10 +1,13 @@ // //Core message type definitions for the Tilebox datasets API. +// //Conventions: //- A dataset is a set of data points of the same type. //- A single data point (entry) is referred to in all messages as datapoint. //- Data points for a single dataset are grouped into collections. +// //- All time fields use the google.protobuf.Timestamp message type. +// //- When a wrapper message is needed to return a list of a given message type, use the pluralized message name //- e.g. message Collections contains a list of Collection messages. diff --git a/workflows/v1/client.go b/workflows/v1/client.go index 0af064e..31c72c1 100644 --- a/workflows/v1/client.go +++ b/workflows/v1/client.go @@ -38,10 +38,13 @@ type Client struct { // - environment variable TILEBOX_API_KEY as the API key // - a grpc.RetryHTTPClient HTTP client // - the global tracer provider +// - Tilebox OpenTelemetry exporters when an API key and HTTP(S) API URL are configured // // The passed options are used to override these default values and configure the returned Client appropriately. func NewClient(options ...ClientOption) *Client { cfg := newClientConfig(options) + configureTileboxTelemetry(context.Background(), cfg) + jobConnectClient := newConnectClient(workflowsv1connect.NewJobServiceClient, cfg) telemetryConnectClient := newConnectClient(workflowsv1connect.NewTelemetryQueryServiceClient, cfg) taskConnectClient := newConnectClient(workflowsv1connect.NewTaskServiceClient, cfg) @@ -76,6 +79,7 @@ type clientConfig struct { connectOptions []connect.ClientOption tracerProvider trace.TracerProvider + disableTracing bool } // ClientOption is an interface for configuring a client. Using such options helpers is a @@ -122,6 +126,7 @@ func WithConnectClientOptions(options ...connect.ClientOption) ClientOption { func WithDisableTracing() ClientOption { return func(cfg *clientConfig) { cfg.tracerProvider = noop.NewTracerProvider() + cfg.disableTracing = true } } diff --git a/workflows/v1/clusters_test.go b/workflows/v1/clusters_test.go index dbfe2dc..27cab5a 100644 --- a/workflows/v1/clusters_test.go +++ b/workflows/v1/clusters_test.go @@ -41,6 +41,7 @@ func NewRecordClient(tb testing.TB, filename string) *Client { WithURL("https://api.tilebox.com"), WithAPIKey(apiKey), WithHTTPClient(httpClient), + WithDisableTracing(), ) } @@ -61,6 +62,7 @@ func NewReplayClient(tb testing.TB, filename string) *Client { WithURL("https://api.tilebox.com"), // url/key doesn't matter WithAPIKey("key"), WithHTTPClient(httpClient), + WithDisableTracing(), ) } diff --git a/workflows/v1/jobs.go b/workflows/v1/jobs.go index 670d1bd..cc62da2 100644 --- a/workflows/v1/jobs.go +++ b/workflows/v1/jobs.go @@ -154,36 +154,36 @@ type SpanEvent struct { Attributes map[string]any } -// SortDirection specifies the sort direction for telemetry query results. +// SortDirection specifies the sort direction for observability query results. type SortDirection int32 // SortDirection values. const ( _ SortDirection = iota - // Ascending sorts telemetry query results oldest first. + // Ascending sorts observability query results oldest first. Ascending - // Descending sorts telemetry query results newest first. + // Descending sorts observability query results newest first. Descending ) -// TelemetryQueryOption contains options for querying job telemetry. +// TelemetryQueryOption contains options for querying job observability. type TelemetryQueryOption struct { - // Limit is the maximum number of telemetry records to return. + // Limit is the maximum number of observability records to return. // Leave unset or set to 0 to paginate through and return all records. Limit int64 - // SortDirection is the direction in which telemetry records should be sorted. + // SortDirection is the direction in which observability records should be sorted. // Leave unset to let the server choose its default sort direction. SortDirection SortDirection } -// WithLimit limits the total number of telemetry records returned. +// WithLimit limits the total number of observability records returned. // // Defaults to unlimited. func WithLimit(limit int64) TelemetryQueryOption { return TelemetryQueryOption{Limit: limit} } -// WithSortDirection sets the sort direction for telemetry query results. +// WithSortDirection sets the sort direction for observability query results. // // Defaults to the server default. func WithSortDirection(direction SortDirection) TelemetryQueryOption { diff --git a/workflows/v1/runner.go b/workflows/v1/runner.go index f7aff88..cd46b8f 100644 --- a/workflows/v1/runner.go +++ b/workflows/v1/runner.go @@ -19,7 +19,7 @@ import ( "github.com/google/uuid" "github.com/samber/lo" "github.com/tilebox/tilebox-go/internal/span" - "github.com/tilebox/tilebox-go/observability" + obslogger "github.com/tilebox/tilebox-go/observability/logger" tileboxv1 "github.com/tilebox/tilebox-go/protogen/tilebox/v1" workflowsv1 "github.com/tilebox/tilebox-go/protogen/workflows/v1" "github.com/tilebox/tilebox-go/workflows/v1/runner" @@ -371,7 +371,11 @@ func (t *TaskRunner) executeTask(ctx context.Context, task *workflowsv1.Task) (* } return span.StartJobSpan(ctx, t.tracer, fmt.Sprintf("task/%s", identifier.Name()), task.GetJob(), func(ctx context.Context) (taskExecutionContext *taskExecutionContext, err error) { //nolint:nonamedreturns // needed to return a value in case of panic - t.logger.DebugContext(ctx, "executing task", slog.String("task", identifier.Name()), slog.String("version", identifier.Version())) + t.logger.DebugContext(ctx, "executing task", + slog.String("task_id", task.GetId().AsUUID().String()), + slog.String("task", identifier.Name()), + slog.String("version", identifier.Version()), + ) taskStruct := reflect.New(reflect.ValueOf(taskPrototype).Elem().Type()).Interface().(ExecutableTask) _, isProtobuf := taskStruct.(proto.Message) @@ -414,6 +418,7 @@ func (t *TaskRunner) executeTask(ctx context.Context, task *workflowsv1.Task) (* t.metrics.tasksExecutedMetric.Add(ctx, 1, taskMetricAttributes) executionContext := t.withTaskExecutionContext(ctx, task) + executionContext = obslogger.ContextWithSlogAttributes(executionContext, slog.String("task_id", task.GetId().AsUUID().String())) err = taskStruct.Execute(executionContext) executionTime := time.Since(beforeTime) @@ -426,7 +431,7 @@ func (t *TaskRunner) executeTask(ctx context.Context, task *workflowsv1.Task) (* t.metrics.tasksFailedMetric.Add(ctx, 1, taskMetricAttributes) t.metrics.taskExecutionDurationMetric.Record(ctx, executionTime.Seconds(), taskMetricAttributes, metric.WithAttributes(attribute.String("state", "failed"))) - log.ErrorContext(ctx, "task execution failed", slog.Any("error", err), slog.Int64("retry_attempt", task.GetRetryCount())) + log.ErrorContext(executionContext, "task execution failed", slog.Any("error", err), slog.Int64("retry_attempt", task.GetRetryCount())) return getTaskExecutionContext(executionContext), fmt.Errorf("failed to execute task: %w", err) } @@ -750,22 +755,12 @@ func Progress(label string) ProgressTracker { // WithTaskSpanResult is a helper function that wraps a function with a tracing span. // It returns the result of the function and an error if any. func WithTaskSpanResult[Result any](ctx context.Context, name string, f func(ctx context.Context) (Result, error)) (Result, error) { - executionContext := getTaskExecutionContext(ctx) - if executionContext == nil || executionContext.runner.tracer == nil { - // if we don't have a task execution context or the tracer is not configured, just run the function without creating a span - return f(ctx) - } - return observability.WithSpanResult(ctx, executionContext.runner.tracer, name, f) + return WithSpanResult(ctx, name, f) } // WithTaskSpan is a helper function that wraps a function with a tracing span. func WithTaskSpan(ctx context.Context, name string, f func(ctx context.Context) error) error { - executionContext := getTaskExecutionContext(ctx) - if executionContext == nil || executionContext.runner.tracer == nil { - // if we don't have a task execution context or the tracer is not configured, just run the function without creating a span - return f(ctx) - } - return observability.WithSpan(ctx, executionContext.runner.tracer, name, f) + return WithSpan(ctx, name, f) } var divs = []time.Duration{ diff --git a/workflows/v1/telemetry.go b/workflows/v1/telemetry.go new file mode 100644 index 0000000..37610c7 --- /dev/null +++ b/workflows/v1/telemetry.go @@ -0,0 +1,196 @@ +package workflows + +import ( + "context" + "fmt" + "log/slog" + "net/url" + "path" + "reflect" + "strings" + "sync" + + "github.com/tilebox/tilebox-go/observability" + "github.com/tilebox/tilebox-go/observability/logger" + obstracer "github.com/tilebox/tilebox-go/observability/tracer" + "go.opentelemetry.io/otel" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/trace" +) + +const ( + otlpLogsPath = "/v1/logs" + otlpTracesPath = "/v1/traces" +) + +var ( + tileboxTelemetryService = &observability.Service{Name: "tilebox-go"} + tileboxLogHandlerOnce sync.Once + consoleLogHandlerOnce sync.Once + tileboxTraceProviders sync.Map + slogHandlersMu sync.Mutex + slogHandlers []slog.Handler +) + +// ConfigureConsoleLogging configures the default slog logger to write logs to stdout at the given level. +// +// Console logging is composed with the Tilebox log exporter configured by NewClient. Calling +// ConfigureConsoleLogging before or after NewClient causes future slog records to be sent to both +// destinations. If NewClient is not called, logs are only written to the console. +func ConfigureConsoleLogging(level slog.Level) { + consoleLogHandlerOnce.Do(func() { + configureSlogHandler(logger.NewConsoleHandler(logger.WithLevel(level))) + }) +} + +// WithSpanResult wraps f with a tracing span using the current task runner's tracer. +// If ctx is not a task execution context or the task runner has no tracer, f is called without creating a span. +func WithSpanResult[Result any](ctx context.Context, name string, f func(ctx context.Context) (Result, error)) (Result, error) { + executionContext := getTaskExecutionContext(ctx) + if executionContext == nil || executionContext.runner.tracer == nil { + return f(ctx) + } + + return observability.WithSpanResult(ctx, executionContext.runner.tracer, name, f) +} + +// WithSpan wraps f with a tracing span using the current task runner's tracer. +// If ctx is not a task execution context or the task runner has no tracer, f is called without creating a span. +func WithSpan(ctx context.Context, name string, f func(ctx context.Context) error) error { + executionContext := getTaskExecutionContext(ctx) + if executionContext == nil || executionContext.runner.tracer == nil { + return f(ctx) + } + + return observability.WithSpan(ctx, executionContext.runner.tracer, name, f) +} + +func configureTileboxTelemetry(ctx context.Context, cfg *clientConfig) { + if cfg.apiKey == "" || !isHTTPURL(cfg.url) { + return + } + + if !cfg.disableTracing { + cfg.tracerProvider = configureTileboxTracing(ctx, cfg) + } + + configureTileboxLogging(ctx, cfg) +} + +func configureTileboxTracing(ctx context.Context, cfg *clientConfig) trace.TracerProvider { + endpoint, ok := otlpEndpointURL(cfg.url, otlpTracesPath) + if !ok { + return cfg.tracerProvider + } + + globalTracerProvider := otel.GetTracerProvider() + if tracerProvider, ok := globalTracerProvider.(*sdktrace.TracerProvider); ok { + if _, loaded := tileboxTraceProviders.Load(tracerProvider); loaded { + return tracerProvider + } + } + + spanProcessor, err := obstracer.NewOtelSpanProcessor(ctx, + obstracer.WithEndpointURL(endpoint), + obstracer.WithHeaders(tileboxTelemetryHeaders(cfg.apiKey)), + ) + if err != nil { + otel.Handle(fmt.Errorf("failed to configure Tilebox trace exporter: %w", err)) + return cfg.tracerProvider + } + + if tracerProvider, ok := globalTracerProvider.(*sdktrace.TracerProvider); ok { + tracerProvider.RegisterSpanProcessor(spanProcessor) + tileboxTraceProviders.Store(tracerProvider, endpoint) + return tracerProvider + } + + if isDefaultTracerProvider(globalTracerProvider) { + tracerProvider := obstracer.NewOtelProviderWithProcessors(tileboxTelemetryService, spanProcessor) + otel.SetTracerProvider(tracerProvider) + tileboxTraceProviders.Store(tracerProvider, endpoint) + return tracerProvider + } + + return cfg.tracerProvider +} + +func configureTileboxLogging(ctx context.Context, cfg *clientConfig) { + endpoint, ok := otlpEndpointURL(cfg.url, otlpLogsPath) + if !ok { + return + } + + tileboxLogHandlerOnce.Do(func() { + otelHandler, _, err := logger.NewOtelHandler(ctx, tileboxTelemetryService, + logger.WithEndpointURL(endpoint), + logger.WithHeaders(tileboxTelemetryHeaders(cfg.apiKey)), + logger.WithLevel(slog.LevelInfo), + ) + if err != nil { + otel.Handle(fmt.Errorf("failed to configure Tilebox log exporter: %w", err)) + return + } + + configureSlogHandler(otelHandler) + }) +} + +func configureSlogHandler(handler slog.Handler) { + slogHandlersMu.Lock() + defer slogHandlersMu.Unlock() + + if len(slogHandlers) == 0 { + existingHandler := slog.Default().Handler() + if !isDefaultSlogHandler(existingHandler) { + slogHandlers = append(slogHandlers, existingHandler) + } + } + + slogHandlers = append(slogHandlers, handler) + slog.SetDefault(logger.New(slogHandlers...)) +} + +func isDefaultSlogHandler(handler slog.Handler) bool { + handlerType := reflect.TypeOf(handler) + if handlerType == nil || handlerType.Kind() != reflect.Pointer { + return false + } + + handlerType = handlerType.Elem() + return handlerType.PkgPath() == "log/slog" && handlerType.Name() == "defaultHandler" +} + +func isDefaultTracerProvider(tracerProvider trace.TracerProvider) bool { + tracerProviderType := reflect.TypeOf(tracerProvider) + if tracerProviderType == nil || tracerProviderType.Kind() != reflect.Pointer { + return false + } + + tracerProviderType = tracerProviderType.Elem() + return tracerProviderType.PkgPath() == "go.opentelemetry.io/otel/internal/global" && tracerProviderType.Name() == "tracerProvider" +} + +func tileboxTelemetryHeaders(apiKey string) map[string]string { + return map[string]string{"Authorization": "Bearer " + apiKey} +} + +func isHTTPURL(rawURL string) bool { + return strings.HasPrefix(rawURL, "https://") || strings.HasPrefix(rawURL, "http://") +} + +func otlpEndpointURL(baseURL, signalPath string) (string, bool) { + parsedURL, err := url.Parse(baseURL) + if err != nil || parsedURL.Scheme == "" || parsedURL.Host == "" { + return "", false + } + + parsedURL.RawQuery = "" + parsedURL.Fragment = "" + parsedURL.Path = path.Join(parsedURL.Path, signalPath) + if !strings.HasPrefix(parsedURL.Path, "/") { + parsedURL.Path = "/" + parsedURL.Path + } + + return parsedURL.String(), true +} diff --git a/workflows/v1/telemetry_test.go b/workflows/v1/telemetry_test.go new file mode 100644 index 0000000..8e041fb --- /dev/null +++ b/workflows/v1/telemetry_test.go @@ -0,0 +1,126 @@ +package workflows + +import ( + "context" + "log/slog" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + oteltrace "go.opentelemetry.io/otel/trace" +) + +var initialDefaultSlogLogger = slog.Default() + +type recordingSlogHandler struct { + name string + records *[]string +} + +func (h *recordingSlogHandler) Enabled(context.Context, slog.Level) bool { + return true +} + +func (h *recordingSlogHandler) Handle(_ context.Context, record slog.Record) error { + *h.records = append(*h.records, h.name+":"+record.Message) + return nil +} + +func (h *recordingSlogHandler) WithAttrs([]slog.Attr) slog.Handler { + return h +} + +func (h *recordingSlogHandler) WithGroup(string) slog.Handler { + return h +} + +func TestConfigureSlogHandlerFansOutToConfiguredHandlers(t *testing.T) { + resetSlogConfigurationForTest(t) + + records := make([]string, 0) + configureSlogHandler(&recordingSlogHandler{name: "first", records: &records}) + slog.Info("one") + + configureSlogHandler(&recordingSlogHandler{name: "second", records: &records}) + slog.Info("two") + + assert.Equal(t, []string{"first:one", "first:two", "second:two"}, records) +} + +func TestConfigureSlogHandlerPreservesExistingCustomHandler(t *testing.T) { + resetSlogConfigurationForTest(t) + + records := make([]string, 0) + slog.SetDefault(slog.New(&recordingSlogHandler{name: "existing", records: &records})) + + configureSlogHandler(&recordingSlogHandler{name: "configured", records: &records}) + slog.Info("hello") + + assert.Equal(t, []string{"existing:hello", "configured:hello"}, records) +} + +func TestConfigureConsoleLoggingIsIdempotent(t *testing.T) { + resetSlogConfigurationForTest(t) + + ConfigureConsoleLogging(slog.LevelInfo) + ConfigureConsoleLogging(slog.LevelDebug) + + slogHandlersMu.Lock() + defer slogHandlersMu.Unlock() + assert.Len(t, slogHandlers, 1) +} + +func TestWithSpanResultUsesTaskRunnerTracer(t *testing.T) { + spanRecorder := tracetest.NewSpanRecorder() + tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spanRecorder)) + ctx := context.WithValue(context.Background(), contextKeyTaskExecution, &taskExecutionContext{ + runner: &TaskRunner{tracer: tracerProvider.Tracer("test")}, + }) + + result, err := WithSpanResult(ctx, "expensive-compute", func(ctx context.Context) (string, error) { + assert.True(t, oteltrace.SpanContextFromContext(ctx).IsValid()) + return "ok", nil + }) + + require.NoError(t, err) + assert.Equal(t, "ok", result) + endedSpans := spanRecorder.Ended() + require.Len(t, endedSpans, 1) + assert.Equal(t, "expensive-compute", endedSpans[0].Name()) +} + +func TestWithSpanRunsWithoutTracerOutsideTaskContext(t *testing.T) { + called := false + err := WithSpan(context.Background(), "without-task-context", func(ctx context.Context) error { + called = true + assert.False(t, oteltrace.SpanContextFromContext(ctx).IsValid()) + return nil + }) + + require.NoError(t, err) + assert.True(t, called) +} + +func resetSlogConfigurationForTest(t *testing.T) { + t.Helper() + + defaultLogger := slog.Default() + slog.SetDefault(initialDefaultSlogLogger) + slogHandlersMu.Lock() + previousHandlers := slogHandlers + slogHandlers = nil + consoleLogHandlerOnce = sync.Once{} + slogHandlersMu.Unlock() + + t.Cleanup(func() { + slog.SetDefault(defaultLogger) + + slogHandlersMu.Lock() + defer slogHandlersMu.Unlock() + slogHandlers = previousHandlers + consoleLogHandlerOnce = sync.Once{} + }) +}