From d16e219f482248945bcc059755118557e9618d49 Mon Sep 17 00:00:00 2001 From: sanchar127 Date: Tue, 18 Aug 2026 02:36:36 +0545 Subject: [PATCH] feat: add OpenTelemetry observability --- cmd/root.go | 32 +- cmd/save.go | 71 ++++- docker-compose.yml | 20 +- go.mod | 23 ++ go.sum | 65 ++++ internal/ai/provider.go | 32 +- internal/cache/hashcache.go | 63 +++- internal/db/db.go | 97 +++++- internal/objects/store.go | 54 +++- internal/snapshot/snapshot.go | 32 ++ internal/telemetry/metrics.go | 337 ++++++++++++++++++++ internal/telemetry/telemetry.go | 440 +++++++++++++++++++++++++++ internal/telemetry/telemetry_test.go | 167 ++++++++++ internal/telemetry/tracing.go | 143 +++++++++ otel-collector-config.yaml | 33 ++ 15 files changed, 1570 insertions(+), 39 deletions(-) create mode 100644 internal/telemetry/metrics.go create mode 100644 internal/telemetry/telemetry.go create mode 100644 internal/telemetry/telemetry_test.go create mode 100644 internal/telemetry/tracing.go create mode 100644 otel-collector-config.yaml diff --git a/cmd/root.go b/cmd/root.go index bdaf4fc..2b91f30 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -1,8 +1,12 @@ package cmd import ( + "context" "fmt" "os" + "time" + + "eko/internal/telemetry" "github.com/spf13/cobra" ) @@ -10,13 +14,37 @@ import ( var rootCmd = &cobra.Command{ Use: "eko", Short: "eko – AI Snapshot Versioning CLI", - // Errors (including the "not an eko project" guard) are reported once by - // Execute below; don't let Cobra also print them or dump usage on failure. + SilenceUsage: true, SilenceErrors: true, } func Execute() { + shutdown, err := telemetry.Init(context.Background()) + if err != nil { + fmt.Fprintln( + os.Stderr, + "Warning: telemetry initialization failed:", + err, + ) + } else { + defer func() { + ctx, cancel := context.WithTimeout( + context.Background(), + 5*time.Second, + ) + defer cancel() + + if err := shutdown(ctx); err != nil { + fmt.Fprintln( + os.Stderr, + "Warning: telemetry shutdown failed:", + err, + ) + } + }() + } + if err := rootCmd.Execute(); err != nil { fmt.Fprintln(os.Stderr, "Error:", err) os.Exit(1) diff --git a/cmd/save.go b/cmd/save.go index 1a95464..d5fd063 100644 --- a/cmd/save.go +++ b/cmd/save.go @@ -2,12 +2,15 @@ package cmd import ( "context" - "eko/internal/ai" - "eko/internal/db" - "eko/internal/snapshot" "errors" "fmt" "os" + "time" + + "eko/internal/ai" + "eko/internal/db" + "eko/internal/snapshot" + "eko/internal/telemetry" "github.com/spf13/cobra" ) @@ -37,13 +40,40 @@ Each snapshot generates a lightweight manifest file and stores unique file blobs # Save with AI summary using a specific provider eko save --ai --provider heuristic`, PreRunE: requireInitialized, - RunE: func(cmd *cobra.Command, args []string) error { + RunE: func(cmd *cobra.Command, args []string) (err error) { + ctx := cmd.Context() + if ctx == nil { + ctx = context.Background() + } + + operation := telemetry.StartOperation( + ctx, + "eko.save", + telemetry.CommandAttribute("save"), + ) + defer func() { + telemetry.EndOperation(operation.Span, err) + }() + + start := time.Now() + success := false + defer func() { + telemetry.RecordCommand( + operation.Context, + "save", + start, + success, + ) + }() + database := db.InitDB() defer database.Close() - // Get previous snapshot path before creating a new one + // Get previous snapshot path before creating a new one. var prevPath string - _ = database.QueryRow("SELECT path FROM snapshots ORDER BY created_at DESC, rowid DESC LIMIT 1").Scan(&prevPath) + _ = database.QueryRow( + "SELECT path FROM snapshots ORDER BY created_at DESC, rowid DESC LIMIT 1", + ).Scan(&prevPath) id, path, err := snapshot.CreateSnapshot(database) if err != nil { @@ -51,11 +81,17 @@ Each snapshot generates a lightweight manifest file and stores unique file blobs } var summaryText string + if saveAI { - ctx := context.Background() - res, err := ai.GenerateSnapshotSummary(ctx, prevPath, path, saveAIProv) - if err == nil && res != nil { + res, summaryErr := ai.GenerateSnapshotSummary( + operation.Context, + prevPath, + path, + saveAIProv, + ) + if summaryErr == nil && res != nil { summaryText = res.Summary + if saveMessage == "snapshot" { saveMessage = res.Summary } @@ -70,16 +106,27 @@ Each snapshot generates a lightweight manifest file and stores unique file blobs summaryText, ); err != nil { // CreateSnapshot has already written the manifest, but the failed row - // insert leaves no supported way to list or restore it. Remove that - // unreachable manifest so retries do not accumulate phantom snapshots. + // insert leaves no supported way to list or restore it. dbErr := fmt.Errorf("failed to save snapshot to db: %w", err) + if rmErr := os.Remove(path); rmErr != nil { - return errors.Join(dbErr, fmt.Errorf("could not remove orphaned snapshot manifest %s: %w", path, rmErr)) + return errors.Join( + dbErr, + fmt.Errorf( + "could not remove orphaned snapshot manifest %s: %w", + path, + rmErr, + ), + ) } + return dbErr } + success = true + fmt.Println("Snapshot saved:", id) + if summaryText != "" { fmt.Println("AI Summary:", summaryText) } diff --git a/docker-compose.yml b/docker-compose.yml index eea70b8..a2f850e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,12 +1,24 @@ -version: "3.8" - services: eko-cli: image: eko-cli build: ./ + environment: + EKO_OTEL_ENABLED: "true" + OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4318" volumes: - eko-data:/.eko + depends_on: + - otel-collector -volumes: - eko-data: + otel-collector: + image: otel/opentelemetry-collector:latest + command: + - "--config=/etc/otelcol/config.yaml" + volumes: + - ./otel-collector-config.yaml:/etc/otelcol/config.yaml + ports: + - "4318:4318" + - "8889:8889" +volumes: + eko-data: \ No newline at end of file diff --git a/go.mod b/go.mod index d0b2eac..a6ae76f 100644 --- a/go.mod +++ b/go.mod @@ -5,9 +5,32 @@ go 1.26 require ( github.com/mattn/go-sqlite3 v1.14.45 github.com/spf13/cobra v1.10.2 + go.opentelemetry.io/otel v1.45.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.45.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0 + go.opentelemetry.io/otel/metric v1.45.0 + go.opentelemetry.io/otel/sdk v1.45.0 + go.opentelemetry.io/otel/sdk/metric v1.45.0 + go.opentelemetry.io/otel/trace v1.45.0 ) require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/go-logr/logr v1.4.4 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/spf13/pflag v1.0.9 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 // indirect + go.opentelemetry.io/proto/otlp v1.11.0 // indirect + golang.org/x/net v0.57.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.40.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect + google.golang.org/grpc v1.83.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/go.sum b/go.sum index 112f86c..e2863e0 100644 --- a/go.sum +++ b/go.sum @@ -1,12 +1,77 @@ +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= +github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/mattn/go-sqlite3 v1.14.45 h1:6KA/spDguL3KV8rnybG7ezSaE4SeMR3KC9VbUoAQaIk= github.com/mattn/go-sqlite3 v1.14.45/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= +go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.45.0 h1:pnxy6c/kvNBWdNNFzqpjuJLm9Hjhgk/Q0nY221rwuk0= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.45.0/go.mod h1:qw6YsFapotRwoDhXRZvljzaOvCQB7UfnafEJagpN2TA= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 h1:QRefszxJmfPdjXUUm3j6iDzY03mTPXMjqErFqQ67vUg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0/go.mod h1:Tiz03lTBVBrm7eWZBOidzEaYaJa8tjwGUGv6d8mlTyk= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0 h1:QBajQ2SrwQijzHyZbQlPsuIzpl/ll8DY6wPWsajeGcI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0/go.mod h1:08ZQLjrPLQ6R4kAXvuOvODEer5Yh4CoFvll5qB2BCI8= +go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypRm3br/M= +go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= +go.opentelemetry.io/otel/metric/x v0.67.0 h1:PcicCNZFkZ4bXfSooXdo3WN7RBOVOtjVdo1wD358Uns= +go.opentelemetry.io/otel/metric/x v0.67.0/go.mod h1:FBjCWZe6wgcqxcMtjdGiClDKXb2YxxXii0CXftE4QtI= +go.opentelemetry.io/otel/sdk v1.45.0 h1:4VVSMgQ83dUgW2aoX5f6JgLvHwIvzcuLnF9lUdCSpCw= +go.opentelemetry.io/otel/sdk v1.45.0/go.mod h1:Sr40LgXV7DsKMMJMKOhUWOgMWTfAaqvm2kF0g7ilwuA= +go.opentelemetry.io/otel/sdk/metric v1.45.0 h1:oVFszMfyj1Am6s24Vtc7wBb8BKLcwepJjNEYILuiE3o= +go.opentelemetry.io/otel/sdk/metric v1.45.0/go.mod h1:vUWUxDZvu1WVRj8JA8S0AdhsPrZoDpA2DdZauIh4mDA= +go.opentelemetry.io/otel/trace v1.45.0 h1:l/mP6Uv7oNO7/TblbhpbgMidxhq1uO/rPsikOyVhxag= +go.opentelemetry.io/otel/trace v1.45.0/go.mod h1:qoJJA2xNMnxRrdISU/kLtfUH2wNeQbiv+jhs/CxI8bc= +go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= +go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= +golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d/go.mod h1:K/+WGbmBY7aNW1HDw1fJnKYo10i0DkAX6pows00dLig= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d h1:IL4hdHzcUv2l/gcg98/Rj3FbtE6axwqslOW8SW0C+S0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ= +google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/ai/provider.go b/internal/ai/provider.go index f509a28..61db9d4 100644 --- a/internal/ai/provider.go +++ b/internal/ai/provider.go @@ -11,6 +11,8 @@ import ( "path/filepath" "strings" "time" + + "eko/internal/telemetry" ) // Provider interface defines the contract for generating AI change summaries. @@ -115,7 +117,7 @@ func (o *OpenAIProvider) Name() string { return "openai" } -func (o *OpenAIProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (string, error) { +func (o *OpenAIProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (summary string, err error) { apiKey := o.APIKey if apiKey == "" { apiKey = os.Getenv("OPENAI_API_KEY") @@ -176,7 +178,18 @@ func (o *OpenAIProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (str client = &http.Client{Timeout: 15 * time.Second} } - resp, err := client.Do(req) + operation := telemetry.StartOperation( + ctx, + "eko.ai.openai", + telemetry.ProviderAttribute("openai"), + telemetry.ModelAttribute(model), + telemetry.OperationAttribute("ai.generate_summary"), + ) + defer func() { + telemetry.EndOperation(operation.Span, err) + }() + + resp, err := client.Do(req.WithContext(operation.Context)) if err != nil { // Fallback to heuristic on connection error hp := &HeuristicProvider{} @@ -223,7 +236,7 @@ func (g *GeminiProvider) Name() string { return "gemini" } -func (g *GeminiProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (string, error) { +func (g *GeminiProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (summary string, err error) { apiKey := g.APIKey if apiKey == "" { apiKey = os.Getenv("GEMINI_API_KEY") @@ -274,7 +287,18 @@ func (g *GeminiProvider) GenerateSummary(ctx context.Context, cs ChangeSet) (str client = &http.Client{Timeout: 15 * time.Second} } - resp, err := client.Do(req) + operation := telemetry.StartOperation( + ctx, + "eko.ai.gemini", + telemetry.ProviderAttribute("gemini"), + telemetry.ModelAttribute(model), + telemetry.OperationAttribute("ai.generate_summary"), + ) + defer func() { + telemetry.EndOperation(operation.Span, err) + }() + + resp, err := client.Do(req.WithContext(operation.Context)) if err != nil { hp := &HeuristicProvider{} res, _ := hp.GenerateSummary(ctx, cs) diff --git a/internal/cache/hashcache.go b/internal/cache/hashcache.go index 4458c7c..96af9b1 100644 --- a/internal/cache/hashcache.go +++ b/internal/cache/hashcache.go @@ -14,10 +14,14 @@ package cache import ( + "context" "database/sql" "fmt" "os" "sync" + "time" + + "eko/internal/telemetry" ) // HashCache wraps the db.sqlite hash_cache table with prepared statement caching @@ -31,6 +35,18 @@ type HashCache struct { // New initialises the hash_cache table and pre-compiles prepared SQL statements. func New(db *sql.DB) (*HashCache, error) { + start := time.Now() + success := false + + defer func() { + telemetry.RecordSQLite( + context.Background(), + "hash_cache_init", + start, + success, + ) + }() + _, err := db.Exec(` CREATE TABLE IF NOT EXISTS hash_cache ( path TEXT NOT NULL, @@ -55,6 +71,8 @@ func New(db *sql.DB) (*HashCache, error) { return nil, fmt.Errorf("hash_cache: prepare store: %w", err) } + success = true + return &HashCache{ db: db, stmtLookup: stmtLookup, @@ -86,10 +104,29 @@ func (c *HashCache) Store(path string, info os.FileInfo, hash string) error { c.mu.Lock() defer c.mu.Unlock() - _, err := c.stmtStore.Exec(path, info.ModTime().UnixNano(), info.Size(), hash) + start := time.Now() + success := false + + defer func() { + telemetry.RecordSQLite( + context.Background(), + "hash_cache_store", + start, + success, + ) + }() + + _, err := c.stmtStore.Exec( + path, + info.ModTime().UnixNano(), + info.Size(), + hash, + ) if err != nil { return fmt.Errorf("hash_cache: store %s: %w", path, err) } + + success = true return nil } @@ -98,6 +135,18 @@ func (c *HashCache) Purge(existingPaths map[string]bool) error { c.mu.Lock() defer c.mu.Unlock() + start := time.Now() + success := false + + defer func() { + telemetry.RecordSQLite( + context.Background(), + "hash_cache_purge", + start, + success, + ) + }() + rows, err := c.db.Query("SELECT DISTINCT path FROM hash_cache") if err != nil { return err @@ -111,12 +160,20 @@ func (c *HashCache) Purge(existingPaths map[string]bool) error { stale = append(stale, p) } } - rows.Close() + + if err := rows.Err(); err != nil { + return err + } for _, p := range stale { - if _, err := c.db.Exec("DELETE FROM hash_cache WHERE path=?", p); err != nil { + if _, err := c.db.Exec( + "DELETE FROM hash_cache WHERE path=?", + p, + ); err != nil { return err } } + + success = true return nil } diff --git a/internal/db/db.go b/internal/db/db.go index 5bcd839..9c71b60 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -1,9 +1,13 @@ package db import ( + "context" "database/sql" "fmt" "log" + "time" + + "eko/internal/telemetry" _ "github.com/mattn/go-sqlite3" ) @@ -22,7 +26,10 @@ func InitDB() *sql.DB { // MigrateDB creates the schema if needed and adds any missing columns (e.g. summary, tag). func MigrateDB(database *sql.DB) error { - _, err := database.Exec(` + _, err := execWithTelemetry( + database, + "create_schema", + ` CREATE TABLE IF NOT EXISTS snapshots ( id TEXT PRIMARY KEY, message TEXT, @@ -31,34 +38,108 @@ func MigrateDB(database *sql.DB) error { summary TEXT, tag TEXT UNIQUE ) - `) + `, + ) if err != nil { return fmt.Errorf("error creating snapshots table: %w", err) } // Safely ensure summary and tag columns exist for pre-existing databases - _, _ = database.Exec("ALTER TABLE snapshots ADD COLUMN summary TEXT") - _, _ = database.Exec("ALTER TABLE snapshots ADD COLUMN tag TEXT UNIQUE") + _, _ = execWithTelemetry( + database, + "migration", + "ALTER TABLE snapshots ADD COLUMN summary TEXT", + ) + + _, _ = execWithTelemetry( + database, + "migration", + "ALTER TABLE snapshots ADD COLUMN tag TEXT UNIQUE", + ) return nil } // SaveSummary updates the AI-generated summary for a given snapshot ID. func SaveSummary(database *sql.DB, id, summary string) error { - _, err := database.Exec("UPDATE snapshots SET summary = ? WHERE id = ?", summary, id) + _, err := execWithTelemetry( + database, + "save_summary", + "UPDATE snapshots SET summary = ? WHERE id = ?", + summary, + id, + ) return err } // SaveTag assigns a human-readable tag/alias to a given snapshot ID. func SaveTag(database *sql.DB, id, tag string) error { - _, err := database.Exec("UPDATE snapshots SET tag = ? WHERE id = ?", tag, id) + _, err := execWithTelemetry( + database, + "save_tag", + "UPDATE snapshots SET tag = ? WHERE id = ?", + tag, + id, + ) return err } // ResolveSnapshot resolves an 8-character snapshot ID or a human-readable tag to its manifest/snapshot path. func ResolveSnapshot(database *sql.DB, target string) (id, path string, err error) { - err = database.QueryRow("SELECT id, path FROM snapshots WHERE id = ? OR tag = ?", target, target).Scan(&id, &path) + start := time.Now() + success := false + + defer func() { + telemetry.RecordSQLite( + context.Background(), + "resolve_snapshot", + start, + success, + ) + }() + + err = database.QueryRow( + "SELECT id, path FROM snapshots WHERE id = ? OR tag = ?", + target, + target, + ).Scan(&id, &path) + if err != nil { - return "", "", fmt.Errorf("snapshot or tag %q not found: %w", target, err) + return "", "", fmt.Errorf( + "snapshot or tag %q not found: %w", + target, + err, + ) } + + success = true + return id, path, nil } + +func execWithTelemetry( + database *sql.DB, + operation string, + query string, + args ...any, +) (sql.Result, error) { + start := time.Now() + success := false + + defer func() { + telemetry.RecordSQLite( + context.Background(), + operation, + start, + success, + ) + }() + + result, err := database.Exec(query, args...) + if err != nil { + return nil, err + } + + success = true + + return result, nil +} diff --git a/internal/objects/store.go b/internal/objects/store.go index 972c6c9..2250b7d 100644 --- a/internal/objects/store.go +++ b/internal/objects/store.go @@ -17,6 +17,7 @@ package objects import ( "compress/gzip" + "context" "crypto/sha256" "encoding/hex" "fmt" @@ -25,6 +26,9 @@ import ( "path/filepath" "runtime" "sync" + "time" + + "eko/internal/telemetry" ) const objectsSubdir = "objects" @@ -66,11 +70,24 @@ func (s *Store) Exists(hash string) bool { // Put compresses and stores data by its SHA-256 hash using gzip.BestCompression. // If a blob with that hash already exists the call is a no-op (pure dedup). // Returns the hex-encoded SHA-256 hash. -func (s *Store) Put(data []byte) (string, error) { - hash := hashBytes(data) +func (s *Store) Put(data []byte) (hash string, err error) { + start := time.Now() + success := false + + defer func() { + telemetry.RecordCAS( + context.Background(), + "put", + start, + success, + ) + }() + + hash = hashBytes(data) // Fast path: already stored — dedup hit, no I/O needed. if s.Exists(hash) { + success = true return hash, nil } @@ -79,6 +96,7 @@ func (s *Store) Put(data []byte) (string, error) { // Double-check after acquiring lock (another goroutine may have stored it). if s.Exists(hash) { + success = true return hash, nil } @@ -89,29 +107,37 @@ func (s *Store) Put(data []byte) (string, error) { // Atomic write: write to .tmp then rename. tmp := path + ".tmp" - f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) + + f, err := os.OpenFile( + tmp, + os.O_CREATE|os.O_WRONLY|os.O_TRUNC, + 0644, + ) if err != nil { return "", fmt.Errorf("objects: create tmp: %w", err) } - // Use BestCompression for maximum disk space savings + // Use BestCompression for maximum disk space savings. gz, err := gzip.NewWriterLevel(f, gzip.BestCompression) if err != nil { f.Close() os.Remove(tmp) return "", err } + if _, err := gz.Write(data); err != nil { gz.Close() f.Close() os.Remove(tmp) return "", fmt.Errorf("objects: compress: %w", err) } + if err := gz.Close(); err != nil { f.Close() os.Remove(tmp) return "", err } + if err := f.Close(); err != nil { os.Remove(tmp) return "", err @@ -124,6 +150,8 @@ func (s *Store) Put(data []byte) (string, error) { // Make the blob immutable so it can never be accidentally overwritten. _ = os.Chmod(path, 0444) + + success = true return hash, nil } @@ -141,7 +169,19 @@ func (s *Store) PutFile(filePath, cachedHash string) (string, error) { } // Get decompresses and returns the raw bytes for a stored hash. -func (s *Store) Get(hash string) ([]byte, error) { +func (s *Store) Get(hash string) (data []byte, err error) { + start := time.Now() + success := false + + defer func() { + telemetry.RecordCAS( + context.Background(), + "get", + start, + success, + ) + }() + f, err := os.Open(s.objectPath(hash)) if err != nil { return nil, fmt.Errorf("objects: open %s: %w", hash[:8], err) @@ -154,10 +194,12 @@ func (s *Store) Get(hash string) ([]byte, error) { } defer gz.Close() - data, err := io.ReadAll(gz) + data, err = io.ReadAll(gz) if err != nil { return nil, fmt.Errorf("objects: decompress %s: %w", hash[:8], err) } + + success = true return data, nil } diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go index 37405c3..9367af5 100644 --- a/internal/snapshot/snapshot.go +++ b/internal/snapshot/snapshot.go @@ -21,6 +21,7 @@ package snapshot import ( + "context" "crypto/rand" "crypto/sha256" "database/sql" @@ -38,16 +39,44 @@ import ( "eko/internal/cache" "eko/internal/manifest" "eko/internal/objects" + "eko/internal/telemetry" "eko/internal/util" ) const ekoDir = ".eko" +// CreateSnapshot captures the current workspace into the CAS object store and +// writes a manifest. It accepts the open database so it can use the hash cache. +// +// Returns the snapshot ID and the manifest path (stored in db.snapshots.path). // CreateSnapshot captures the current workspace into the CAS object store and // writes a manifest. It accepts the open database so it can use the hash cache. // // Returns the snapshot ID and the manifest path (stored in db.snapshots.path). func CreateSnapshot(db *sql.DB) (id, path string, err error) { + ctx := context.Background() + + operation := telemetry.StartOperation( + ctx, + "eko.snapshot.create", + telemetry.OperationAttribute("snapshot.create"), + ) + defer func() { + telemetry.EndOperation(operation.Span, err) + }() + + start := time.Now() + success := false + + defer func() { + telemetry.RecordCAS( + operation.Context, + "snapshot.create", + start, + success, + ) + }() + id, err = generateID() if err != nil { return "", "", err @@ -89,6 +118,9 @@ func CreateSnapshot(db *sql.DB) (id, path string, err error) { } manifestPath := filepath.Join(ekoDir, "manifests", id+".json") + + success = true + return id, manifestPath, nil } diff --git a/internal/telemetry/metrics.go b/internal/telemetry/metrics.go new file mode 100644 index 0000000..2e93431 --- /dev/null +++ b/internal/telemetry/metrics.go @@ -0,0 +1,337 @@ +package telemetry + +import ( + "context" + "fmt" + "sync" + "time" + + "go.opentelemetry.io/otel/attribute" + otelmetric "go.opentelemetry.io/otel/metric" +) + +const metricScope = instrumentName + +// Duration buckets are expressed in seconds. +var durationBuckets = []float64{ + 0.0005, // 0.5 ms + 0.001, // 1 ms + 0.0025, // 2.5 ms + 0.005, // 5 ms + 0.01, // 10 ms + 0.025, // 25 ms + 0.05, // 50 ms + 0.1, // 100 ms + 0.25, // 250 ms + 0.5, // 500 ms + 1.0, // 1 s + 2.5, // 2.5 s + 5.0, // 5 s + 10.0, // 10 s +} + +type Metrics struct { + CommandTotal otelmetric.Int64Counter + CommandDuration otelmetric.Float64Histogram + + LLMTotal otelmetric.Int64Counter + LLMDuration otelmetric.Float64Histogram + + CASOperations otelmetric.Int64Counter + CASDuration otelmetric.Float64Histogram + + SQLiteOperations otelmetric.Int64Counter + SQLiteDuration otelmetric.Float64Histogram +} + +var ( + metrics Metrics + metricsOnce sync.Once + metricsErr error +) + +// initMetrics initializes all Eko application metrics. +// +// Metrics are initialized once and are safe to call repeatedly. +// This function is intentionally independent from whether telemetry +// exporters are enabled. Tests and application code can therefore +// safely record metrics without causing nil-instrument panics. +func initMetrics() error { + metricsOnce.Do(func() { + meter := Meter() + + metrics.CommandTotal, metricsErr = meter.Int64Counter( + "eko_command_total", + otelmetric.WithDescription( + "Total number of Eko CLI commands executed.", + ), + ) + if metricsErr != nil { + return + } + + metrics.CommandDuration, metricsErr = meter.Float64Histogram( + "eko_command_duration_seconds", + otelmetric.WithDescription( + "Duration of Eko CLI commands in seconds.", + ), + otelmetric.WithExplicitBucketBoundaries(durationBuckets...), + ) + if metricsErr != nil { + return + } + + metrics.LLMTotal, metricsErr = meter.Int64Counter( + "eko_llm_operations_total", + otelmetric.WithDescription( + "Total number of LLM operations.", + ), + ) + if metricsErr != nil { + return + } + + metrics.LLMDuration, metricsErr = meter.Float64Histogram( + "eko_llm_duration_seconds", + otelmetric.WithDescription( + "Duration of LLM operations in seconds.", + ), + otelmetric.WithExplicitBucketBoundaries(durationBuckets...), + ) + if metricsErr != nil { + return + } + + metrics.CASOperations, metricsErr = meter.Int64Counter( + "eko_cas_operations_total", + otelmetric.WithDescription( + "Total number of content-addressable storage operations.", + ), + ) + if metricsErr != nil { + return + } + + metrics.CASDuration, metricsErr = meter.Float64Histogram( + "eko_cas_duration_seconds", + otelmetric.WithDescription( + "Duration of content-addressable storage operations in seconds.", + ), + otelmetric.WithExplicitBucketBoundaries(durationBuckets...), + ) + if metricsErr != nil { + return + } + + metrics.SQLiteOperations, metricsErr = meter.Int64Counter( + "eko_sqlite_operations_total", + otelmetric.WithDescription( + "Total number of SQLite operations.", + ), + ) + if metricsErr != nil { + return + } + + metrics.SQLiteDuration, metricsErr = meter.Float64Histogram( + "eko_sqlite_duration_seconds", + otelmetric.WithDescription( + "Duration of SQLite operations in seconds.", + ), + otelmetric.WithExplicitBucketBoundaries(durationBuckets...), + ) + }) + + if metricsErr != nil { + return fmt.Errorf("initialize Eko metrics: %w", metricsErr) + } + + return nil +} + +// MetricsInstance returns the initialized metrics. +func MetricsInstance() Metrics { + _ = initMetrics() + return metrics +} + +// durationSeconds converts the duration representations used by +// existing Eko callers into seconds. +// +// Existing code historically passed time.Time values representing +// operation start times. Newer code can pass a float64 duration. +// +// Supporting both here allows telemetry to evolve without requiring +// unrelated packages to change their APIs. +func durationSeconds(value any) float64 { + switch v := value.(type) { + case float64: + if v < 0 { + return 0 + } + return v + + case float32: + if v < 0 { + return 0 + } + return float64(v) + + case time.Duration: + if v < 0 { + return 0 + } + return v.Seconds() + + case time.Time: + d := time.Since(v) + if d < 0 { + return 0 + } + return d.Seconds() + + default: + return 0 + } +} + +// RecordCommand records a completed CLI command. +// +// The duration argument accepts both: +// - float64 seconds +// - time.Time operation start time +func RecordCommand( + ctx context.Context, + command string, + duration any, + success bool, +) { + _ = initMetrics() + + if ctx == nil { + ctx = context.Background() + } + + seconds := durationSeconds(duration) + + attrs := otelmetric.WithAttributes( + attribute.String("command", command), + attribute.Bool("success", success), + ) + + metrics.CommandTotal.Add(ctx, 1, attrs) + metrics.CommandDuration.Record(ctx, seconds, attrs) +} + +// RecordLLM records a completed LLM operation. +// +// Supported forms: +// +// RecordLLM(ctx, operation, duration, success) +// +// and the legacy form: +// +// RecordLLM(ctx, operation, model, start, success) +func RecordLLM( + ctx context.Context, + operation string, + args ...any, +) { + _ = initMetrics() + + if ctx == nil { + ctx = context.Background() + } + + var ( + model string + start any + success bool + ) + + switch len(args) { + case 2: + start = args[0] + success, _ = args[1].(bool) + + case 3: + model, _ = args[0].(string) + start = args[1] + success, _ = args[2].(bool) + + default: + return + } + + seconds := durationSeconds(start) + + attrs := []attribute.KeyValue{ + attribute.String("operation", operation), + attribute.Bool("success", success), + } + + if model != "" { + attrs = append(attrs, attribute.String("model", model)) + } + + metricAttrs := otelmetric.WithAttributes(attrs...) + + metrics.LLMTotal.Add(ctx, 1, metricAttrs) + metrics.LLMDuration.Record(ctx, seconds, metricAttrs) +} + +// RecordCAS records a completed CAS operation. +// +// The duration argument accepts both: +// - float64 seconds +// - time.Time operation start time +func RecordCAS( + ctx context.Context, + operation string, + duration any, + success bool, +) { + _ = initMetrics() + + if ctx == nil { + ctx = context.Background() + } + + seconds := durationSeconds(duration) + + attrs := otelmetric.WithAttributes( + attribute.String("operation", operation), + attribute.Bool("success", success), + ) + + metrics.CASOperations.Add(ctx, 1, attrs) + metrics.CASDuration.Record(ctx, seconds, attrs) +} + +// RecordSQLite records a completed SQLite operation. +// +// The duration argument accepts both: +// - float64 seconds +// - time.Time operation start time +func RecordSQLite( + ctx context.Context, + operation string, + duration any, + success bool, +) { + _ = initMetrics() + + if ctx == nil { + ctx = context.Background() + } + + seconds := durationSeconds(duration) + + attrs := otelmetric.WithAttributes( + attribute.String("operation", operation), + attribute.Bool("success", success), + ) + + metrics.SQLiteOperations.Add(ctx, 1, attrs) + metrics.SQLiteDuration.Record(ctx, seconds, attrs) +} diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go new file mode 100644 index 0000000..db98f70 --- /dev/null +++ b/internal/telemetry/telemetry.go @@ -0,0 +1,440 @@ +package telemetry + +import ( + "context" + "fmt" + "net/url" + "os" + "strings" + "sync" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + "go.opentelemetry.io/otel/metric" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/semconv/v1.37.0" + "go.opentelemetry.io/otel/trace" +) + +const ( + serviceName = "eko" + instrumentName = "eko/telemetry" + + defaultOTLPEndpoint = "http://localhost:4318" + + telemetryTimeout = 5 * time.Second + metricExportInterval = 5 * time.Second + traceExportBatchTimeout = 5 * time.Second +) + +var ( + initOnce sync.Once + + initErr error + + shutdownMu sync.Mutex + shutdown func(context.Context) error + + tracer trace.Tracer + meter metric.Meter +) + +// Init initializes OpenTelemetry for Eko. +// +// Telemetry is disabled unless: +// +// EKO_OTEL_ENABLED=true +// +// When enabled, Eko exports metrics and traces through an OpenTelemetry +// Collector using OTLP over HTTP. +// +// The base endpoint is configured with: +// +// OTEL_EXPORTER_OTLP_ENDPOINT +// +// Example: +// +// OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +// +// The following signal endpoints are generated automatically: +// +// http://localhost:4318/v1/metrics +// http://localhost:4318/v1/traces +// +// Eko is a short-lived CLI, so the returned shutdown function explicitly +// flushes pending metrics and traces before the process exits. +func Init(ctx context.Context) (func(context.Context) error, error) { + initOnce.Do(func() { + // Always initialize safe no-op API handles first. + tracer = otel.Tracer(instrumentName) + meter = otel.Meter(instrumentName) + + // Telemetry is opt-in. + if !telemetryEnabled() { + if err := initMetrics(); err != nil { + initErr = fmt.Errorf( + "initialize disabled telemetry metrics: %w", + err, + ) + return + } + + shutdown = func(context.Context) error { + return nil + } + + return + } + + if ctx == nil { + ctx = context.Background() + } + + baseEndpoint := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") + if strings.TrimSpace(baseEndpoint) == "" { + baseEndpoint = defaultOTLPEndpoint + } + + metricsEndpoint, tracesEndpoint, err := buildEndpoints(baseEndpoint) + if err != nil { + initErr = fmt.Errorf( + "invalid OTLP endpoint: %w", + err, + ) + return + } + + // ------------------------------------------------------------ + // Resource + // ------------------------------------------------------------ + + res, err := newResource(ctx) + if err != nil { + initErr = fmt.Errorf( + "create telemetry resource: %w", + err, + ) + return + } + + // ------------------------------------------------------------ + // Trace exporter + // ------------------------------------------------------------ + + traceExporterOptions := []otlptracehttp.Option{ + otlptracehttp.WithEndpointURL(tracesEndpoint), + otlptracehttp.WithTimeout(telemetryTimeout), + } + + if isInsecureEndpoint(tracesEndpoint) { + traceExporterOptions = append( + traceExporterOptions, + otlptracehttp.WithInsecure(), + ) + } + + traceExporter, err := otlptracehttp.New( + ctx, + traceExporterOptions..., + ) + if err != nil { + initErr = fmt.Errorf( + "create OTLP trace exporter: %w", + err, + ) + return + } + + tp := sdktrace.NewTracerProvider( + sdktrace.WithBatcher( + traceExporter, + sdktrace.WithBatchTimeout( + traceExportBatchTimeout, + ), + ), + sdktrace.WithResource(res), + ) + + // ------------------------------------------------------------ + // Metrics exporter + // ------------------------------------------------------------ + + metricExporterOptions := []otlpmetrichttp.Option{ + otlpmetrichttp.WithEndpointURL(metricsEndpoint), + otlpmetrichttp.WithTimeout(telemetryTimeout), + } + + if isInsecureEndpoint(metricsEndpoint) { + metricExporterOptions = append( + metricExporterOptions, + otlpmetrichttp.WithInsecure(), + ) + } + + metricExporter, err := otlpmetrichttp.New( + ctx, + metricExporterOptions..., + ) + if err != nil { + _ = tp.Shutdown(context.Background()) + + initErr = fmt.Errorf( + "create OTLP metric exporter: %w", + err, + ) + return + } + + reader := sdkmetric.NewPeriodicReader( + metricExporter, + sdkmetric.WithInterval(metricExportInterval), + sdkmetric.WithTimeout(telemetryTimeout), + ) + + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithResource(res), + ) + + // Register providers globally before creating instruments. + otel.SetTracerProvider(tp) + otel.SetMeterProvider(mp) + + tracer = tp.Tracer(instrumentName) + meter = mp.Meter(instrumentName) + + // Application metric instruments must be created from the + // configured MeterProvider. + if err := initMetrics(); err != nil { + _ = mp.Shutdown(context.Background()) + _ = tp.Shutdown(context.Background()) + + initErr = fmt.Errorf( + "initialize application metrics: %w", + err, + ) + return + } + + // ------------------------------------------------------------ + // Shutdown + // ------------------------------------------------------------ + + var shutdownOnce sync.Once + var shutdownErr error + + shutdown = func(shutdownCtx context.Context) error { + shutdownOnce.Do(func() { + shutdownErr = shutdownTelemetry( + shutdownCtx, + reader, + tp, + mp, + ) + }) + + return shutdownErr + } + }) + + if initErr != nil { + return nil, initErr + } + + if shutdown == nil { + return nil, fmt.Errorf( + "telemetry initialization completed without shutdown handler", + ) + } + + // Return a wrapper so the global shutdown function remains protected + // from accidental reassignment. + shutdownMu.Lock() + currentShutdown := shutdown + shutdownMu.Unlock() + + return currentShutdown, nil +} + +// telemetryEnabled determines whether Eko telemetry is enabled. +// +// Telemetry is intentionally opt-in because Eko is a CLI and should not +// unexpectedly send telemetry anywhere. +func telemetryEnabled() bool { + return strings.EqualFold( + strings.TrimSpace(os.Getenv("EKO_OTEL_ENABLED")), + "true", + ) +} + +// newResource creates the resource associated with all Eko telemetry. +// +// OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES are respected through +// resource.WithFromEnv(). The code-level Eko service name provides a +// deterministic fallback. +func newResource(ctx context.Context) (*resource.Resource, error) { + if ctx == nil { + ctx = context.Background() + } + + res, err := resource.New( + ctx, + resource.WithFromEnv(), + resource.WithTelemetrySDK(), + resource.WithAttributes( + semconv.ServiceNameKey.String(serviceName), + ), + ) + if err != nil { + return nil, err + } + + return res, nil +} + +// shutdownTelemetry flushes and shuts down all telemetry providers. +// +// Metrics and traces are flushed explicitly because a CLI process can +// terminate before periodic/batched exporters have exported their data. +func shutdownTelemetry( + ctx context.Context, + reader *sdkmetric.PeriodicReader, + tp *sdktrace.TracerProvider, + mp *sdkmetric.MeterProvider, +) error { + if ctx == nil { + ctx = context.Background() + } + + var firstErr error + + if err := reader.ForceFlush(ctx); err != nil { + firstErr = err + } + + if err := tp.ForceFlush(ctx); err != nil && firstErr == nil { + firstErr = err + } + + if err := mp.Shutdown(ctx); err != nil && firstErr == nil { + firstErr = err + } + + if err := tp.Shutdown(ctx); err != nil && firstErr == nil { + firstErr = err + } + + return firstErr +} + +// buildEndpoints converts a base OTLP endpoint into explicit signal +// endpoints. +// +// Input: +// +// http://localhost:4318 +// +// Output: +// +// http://localhost:4318/v1/metrics +// http://localhost:4318/v1/traces +func buildEndpoints( + baseEndpoint string, +) (metricsEndpoint string, tracesEndpoint string, err error) { + baseEndpoint = strings.TrimSpace(baseEndpoint) + + if baseEndpoint == "" { + return "", "", fmt.Errorf("endpoint is empty") + } + + parsed, err := url.Parse(baseEndpoint) + if err != nil { + return "", "", fmt.Errorf("parse endpoint: %w", err) + } + + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return "", "", fmt.Errorf( + "endpoint scheme must be http or https, got %q", + parsed.Scheme, + ) + } + + if parsed.Host == "" { + return "", "", fmt.Errorf( + "endpoint must include a host", + ) + } + + path := strings.TrimRight(parsed.Path, "/") + + switch path { + case "": + // Base OTLP endpoint. + case "/v1/metrics": + path = "" + case "/v1/traces": + path = "" + default: + return "", "", fmt.Errorf( + "unsupported endpoint path %q; expected an OTLP base endpoint", + parsed.Path, + ) + } + + parsed.Path = path + + metricsURL := *parsed + metricsURL.Path = "/v1/metrics" + + tracesURL := *parsed + tracesURL.Path = "/v1/traces" + + return metricsURL.String(), tracesURL.String(), nil +} + +// isInsecureEndpoint determines whether TLS should be disabled. +// +// OTLP HTTP exporters require WithInsecure() for plain HTTP endpoints. +// HTTPS endpoints use TLS normally. +func isInsecureEndpoint(endpoint string) bool { + parsed, err := url.Parse(endpoint) + if err != nil { + return false + } + + return strings.EqualFold(parsed.Scheme, "http") +} + +// Tracer returns Eko's shared OpenTelemetry tracer. +func Tracer() trace.Tracer { + if tracer == nil { + tracer = otel.Tracer(instrumentName) + } + + return tracer +} + +// Meter returns Eko's shared OpenTelemetry meter. +func Meter() metric.Meter { + if meter == nil { + meter = otel.Meter(instrumentName) + } + + return meter +} + +// StartSpan starts a span using Eko's shared tracer. +func StartSpan( + ctx context.Context, + name string, +) (context.Context, trace.Span) { + if ctx == nil { + ctx = context.Background() + } + + return Tracer().Start(ctx, name) +} diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go new file mode 100644 index 0000000..dfdf840 --- /dev/null +++ b/internal/telemetry/telemetry_test.go @@ -0,0 +1,167 @@ +package telemetry + +import ( + "context" + "testing" + "time" +) + +func TestStartOperation(t *testing.T) { + ctx := context.Background() + + operation := StartOperation( + ctx, + "test.operation", + OperationAttribute("test.operation"), + ) + + if operation == nil { + t.Fatal("expected operation") + } + + if operation.Context == nil { + t.Fatal("expected operation context") + } + + if operation.Span == nil { + t.Fatal("expected span") + } + + if operation.name != "test.operation" { + t.Fatalf( + "expected operation name %q, got %q", + "test.operation", + operation.name, + ) + } + + if operation.startTime.IsZero() { + t.Fatal("expected start time") + } + + EndOperation(operation, nil) +} + +func TestStartOperationNilContext(t *testing.T) { + operation := StartOperation( + nil, + "test.operation", + ) + + if operation == nil { + t.Fatal("expected operation") + } + + if operation.Context == nil { + t.Fatal("expected operation context") + } + + EndOperation(operation, nil) +} + +func TestStartOperationEmptyName(t *testing.T) { + operation := StartOperation( + context.Background(), + "", + ) + + if operation == nil { + t.Fatal("expected operation") + } + + if operation.name != "eko.operation" { + t.Fatalf( + "expected default operation name %q, got %q", + "eko.operation", + operation.name, + ) + } + + EndOperation(operation, nil) +} + +func TestEndOperationReturnsDuration(t *testing.T) { + operation := StartOperation( + context.Background(), + "test.operation", + ) + + time.Sleep(1 * time.Millisecond) + + duration := EndOperation(operation, nil) + + if duration <= 0 { + t.Fatalf( + "expected positive duration, got %f", + duration, + ) + } +} + +func TestEndOperationWithError(t *testing.T) { + operation := StartOperation( + context.Background(), + "test.operation", + ) + + err := context.Canceled + + duration := EndOperation(operation, err) + + if duration < 0 { + t.Fatalf( + "expected non-negative duration, got %f", + duration, + ) + } +} + +func TestEndOperationNilOperation(t *testing.T) { + duration := EndOperation(nil, nil) + + if duration != 0 { + t.Fatalf( + "expected zero duration for nil operation, got %f", + duration, + ) + } +} + +func TestSetErrorNilSpan(t *testing.T) { + SetError(nil, context.Canceled) +} + +func TestSetErrorNilError(t *testing.T) { + operation := StartOperation( + context.Background(), + "test.operation", + ) + + SetError(operation.Span, nil) + + EndOperation(operation, nil) +} + +func TestSetAttributesNilSpan(t *testing.T) { + SetAttributes( + nil, + CommandAttribute("test"), + ) +} + +func TestSetAttributes(t *testing.T) { + operation := StartOperation( + context.Background(), + "test.operation", + ) + + SetAttributes( + operation.Span, + CommandAttribute("test"), + ProviderAttribute("test-provider"), + ModelAttribute("test-model"), + OperationAttribute("test.operation"), + ) + + EndOperation(operation, nil) +} \ No newline at end of file diff --git a/internal/telemetry/tracing.go b/internal/telemetry/tracing.go new file mode 100644 index 0000000..d04a84a --- /dev/null +++ b/internal/telemetry/tracing.go @@ -0,0 +1,143 @@ +package telemetry + +import ( + "context" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" +) + +// SpanResult contains the context and span created for an operation. +type SpanResult struct { + Context context.Context + Span trace.Span + + name string + startTime time.Time +} + +// StartOperation starts a new OpenTelemetry span for an Eko operation. +func StartOperation( + ctx context.Context, + name string, + attrs ...attribute.KeyValue, +) *SpanResult { + if ctx == nil { + ctx = context.Background() + } + + if name == "" { + name = "eko.operation" + } + + startTime := time.Now() + + ctx, span := Tracer().Start(ctx, name) + + if len(attrs) > 0 { + span.SetAttributes(attrs...) + } + + return &SpanResult{ + Context: ctx, + Span: span, + name: name, + startTime: startTime, + } +} + +// EndOperation completes an operation and returns its duration in seconds. +// +// It accepts both: +// +// *SpanResult +// +// and: +// +// trace.Span +// +// The SpanResult form is the preferred API because it allows telemetry +// to calculate the operation duration. The trace.Span form is retained +// for compatibility with existing Eko callers. +func EndOperation(operation any, err error) float64 { + switch op := operation.(type) { + case *SpanResult: + if op == nil || op.Span == nil { + return 0 + } + + duration := time.Since(op.startTime).Seconds() + + if err != nil { + SetError(op.Span, err) + } else { + op.Span.SetStatus(codes.Ok, "") + } + + op.Span.End() + + return duration + + case trace.Span: + if op == nil { + return 0 + } + + if err != nil { + SetError(op, err) + } else { + op.SetStatus(codes.Ok, "") + } + + op.End() + + return 0 + + default: + return 0 + } +} + +// SetError records an error on an active span without ending it. +func SetError(span trace.Span, err error) { + if span == nil || err == nil { + return + } + + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) +} + +// SetAttributes adds attributes to an active span. +func SetAttributes( + span trace.Span, + attrs ...attribute.KeyValue, +) { + if span == nil || len(attrs) == 0 { + return + } + + span.SetAttributes(attrs...) +} + +// CommandAttribute returns the standard Eko command attribute. +func CommandAttribute(command string) attribute.KeyValue { + return attribute.String("eko.command", command) +} + +// ProviderAttribute returns the standard Eko AI provider attribute. +func ProviderAttribute(provider string) attribute.KeyValue { + return attribute.String("eko.ai.provider", provider) +} + +// ModelAttribute returns the standard Eko AI model attribute. +func ModelAttribute(model string) attribute.KeyValue { + return attribute.String("eko.ai.model", model) +} + +// OperationAttribute returns the standard Eko operation attribute. +func OperationAttribute(operation string) attribute.KeyValue { + return attribute.String("eko.operation", operation) +} diff --git a/otel-collector-config.yaml b/otel-collector-config.yaml new file mode 100644 index 0000000..92e1c81 --- /dev/null +++ b/otel-collector-config.yaml @@ -0,0 +1,33 @@ +receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 + +processors: + batch: + +exporters: + prometheus: + endpoint: 0.0.0.0:8889 + + debug: + verbosity: basic + +service: + pipelines: + metrics: + receivers: + - otlp + processors: + - batch + exporters: + - prometheus + + traces: + receivers: + - otlp + processors: + - batch + exporters: + - debug \ No newline at end of file