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 c610317..1a47b1f 100644 --- a/cmd/save.go +++ b/cmd/save.go @@ -2,6 +2,7 @@ package cmd import ( "context" + "eko/internal/ai" "eko/internal/db" "eko/internal/notify" @@ -11,7 +12,7 @@ import ( "fmt" "os" "time" - + "eko/internal/telemetry" "github.com/spf13/cobra" ) @@ -42,13 +43,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) if saveWithEnv { fmt.Println("Warning: Capturing environment variables may store sensitive credentials (API keys, passwords, etc.) in the snapshot.") @@ -73,11 +101,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 } @@ -101,16 +135,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 54cfc38..78c65dc 100644 --- a/go.mod +++ b/go.mod @@ -5,11 +5,34 @@ 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/klauspost/compress v1.19.2 // 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 ac1553f..97baf7e 100644 --- a/go.sum +++ b/go.sum @@ -1,16 +1,80 @@ +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/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= 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 73450f7..5fa7e1b 100644 --- a/internal/objects/store.go +++ b/internal/objects/store.go @@ -1,24 +1,25 @@ // Package objects implements a content-addressable storage (CAS) engine for Eko. // -// Every file blob is stored exactly once, identified by its SHA-256 hash and -// compressed with gzip. Identical file content across any number of snapshots -// occupies only one entry in the object store, giving Git-like deduplication -// without the complexity of delta encoding. +// Every file blob is stored exactly once, identified by its SHA-256 hash. +// Objects are compressed with zstd when beneficial, while already-compressed +// and small files are stored raw. Legacy gzip objects remain readable. // // Layout inside .eko/objects/: // // <2-char prefix>/ -// .gz ← gzip-compressed raw file bytes +// .zst <- zstd-compressed raw file bytes +// .raw <- uncompressed raw file bytes +// .gz <- legacy gzip object // // All objects are stored read-only (0444). Writes are atomic: the blob is -// written to a .tmp file first, then renamed into place — so a crashed write +// written to a .tmp file first, then renamed into place, so a crashed write // never leaves a corrupt object. package objects import ( "compress/gzip" + "context" "crypto/sha256" - "eko/internal/util" "encoding/hex" "fmt" "io" @@ -27,6 +28,10 @@ import ( "runtime" "strings" "sync" + "time" + + "eko/internal/telemetry" + "eko/internal/util" "github.com/klauspost/compress/zstd" ) @@ -50,47 +55,114 @@ type Store struct { // New creates (or opens) the object store under ekoDir/objects. func New(ekoDir string) (*Store, error) { base := filepath.Join(ekoDir, objectsSubdir) + if err := os.MkdirAll(base, 0755); err != nil { return nil, fmt.Errorf("objects: mkdir %s: %w", base, err) } - return &Store{baseDir: base}, nil + + return &Store{ + baseDir: base, + }, nil } // objectPath returns the on-disk path for a given SHA-256 hex hash. +// +// Existing objects are preferred in the following order: +// +// .zst -> .raw -> .gz +// +// The .gz format is retained for backwards compatibility with older Eko +// snapshots. func (s *Store) objectPath(hash string) string { - // Compatibility check: check if .zst or .raw exists, otherwise default to .gz - prefix := filepath.Join(s.baseDir, hash[:2], hash[2:]) - for _, ext := range []string{".zst", ".raw"} { + prefix := filepath.Join( + s.baseDir, + hash[:2], + hash[2:], + ) + + for _, ext := range []string{ + ".zst", + ".raw", + ".gz", + } { path := prefix + ext + if _, err := os.Stat(path); err == nil { return path } } + + // Default path for a missing object. return prefix + ".gz" } // Exists reports whether a blob with this hash is already stored. func (s *Store) Exists(hash string) bool { - prefix := filepath.Join(s.baseDir, hash[:2], hash[2:]) - for _, ext := range []string{".zst", ".raw", ".gz"} { + if len(hash) < 2 { + return false + } + + prefix := filepath.Join( + s.baseDir, + hash[:2], + hash[2:], + ) + + for _, ext := range []string{ + ".zst", + ".raw", + ".gz", + } { if _, err := os.Stat(prefix + ext); err == nil { return true } } + return false } -// Put compresses and stores data by its SHA-256 hash using zstd. -// 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) { - return s.putWithCompression(data, shouldCompress(data)) +// Put stores data by its SHA-256 hash. +// +// The storage format is selected automatically: +// - zstd for compressible data +// - raw for small/already-compressed data +// +// If a blob with the same hash already exists, the call is a no-op. +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, err = s.putWithCompression( + data, + shouldCompress(data), + ) + + if err == nil { + success = true + } + + return hash, err } -func (s *Store) putWithCompression(data []byte, compress bool) (string, error) { +// putWithCompression stores data either compressed with zstd or uncompressed. +// +// The hash is always calculated from the original uncompressed bytes. +func (s *Store) putWithCompression( + data []byte, + compress bool, +) (string, error) { hash := hashBytes(data) - // Fast path: already stored — dedup hit, no I/O needed. + // Fast path: already stored. if s.Exists(hash) { return hash, nil } @@ -98,7 +170,8 @@ func (s *Store) putWithCompression(data []byte, compress bool) (string, error) { s.mu.Lock() defer s.mu.Unlock() - // Double-check after acquiring lock. + // Double-check after acquiring the lock because another goroutine may + // have stored the object while we were waiting. if s.Exists(hash) { return hash, nil } @@ -108,184 +181,427 @@ func (s *Store) putWithCompression(data []byte, compress bool) (string, error) { ext = ".zst" } - path := filepath.Join(s.baseDir, hash[:2], hash[2:]+ext) + path := filepath.Join( + s.baseDir, + hash[:2], + hash[2:]+ext, + ) + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { - return "", fmt.Errorf("objects: mkdir prefix: %w", err) + return "", fmt.Errorf( + "objects: mkdir prefix: %w", + err, + ) } - // Atomic write: write to .tmp then rename. + // Atomic write: + // + // .zst.tmp + // | + // v + // .zst + // + // This prevents partially written objects from appearing as valid objects. 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) + return "", fmt.Errorf( + "objects: create tmp: %w", + err, + ) + } + + cleanup := func() { + _ = f.Close() + _ = os.Remove(tmp) } if compress { var opts []zstd.EOption + + // Small files benefit from avoiding unnecessary encoder parallelism. if len(data) < 1024*1024 { - // Single-threaded encoding for files under 1MB to minimize overhead - opts = append(opts, zstd.WithEncoderConcurrency(1)) + opts = append( + opts, + zstd.WithEncoderConcurrency(1), + ) } + encoder, err := zstd.NewWriter(f, opts...) if err != nil { - f.Close() - os.Remove(tmp) - return "", err + cleanup() + return "", fmt.Errorf( + "objects: create zstd encoder: %w", + err, + ) } + if _, err := encoder.Write(data); err != nil { - encoder.Close() - f.Close() - os.Remove(tmp) - return "", fmt.Errorf("objects: compress: %w", err) + _ = encoder.Close() + cleanup() + + return "", fmt.Errorf( + "objects: compress: %w", + err, + ) } + if err := encoder.Close(); err != nil { - f.Close() - os.Remove(tmp) - return "", err + cleanup() + + return "", fmt.Errorf( + "objects: close zstd encoder: %w", + err, + ) } } else { if _, err := f.Write(data); err != nil { - f.Close() - os.Remove(tmp) - return "", fmt.Errorf("objects: write raw: %w", err) + cleanup() + + return "", fmt.Errorf( + "objects: write raw: %w", + err, + ) } } if err := f.Close(); err != nil { - os.Remove(tmp) - return "", err + _ = os.Remove(tmp) + + return "", fmt.Errorf( + "objects: close tmp: %w", + err, + ) } if err := os.Rename(tmp, path); err != nil { - os.Remove(tmp) - return "", fmt.Errorf("objects: rename: %w", err) + _ = os.Remove(tmp) + + return "", fmt.Errorf( + "objects: rename: %w", + err, + ) } - // Make the blob immutable. + // Objects are immutable after creation. _ = os.Chmod(path, 0444) + return hash, nil } // PutFile reads filePath, stores it in the object store, and returns its hash. -// If cachedHash is non-empty it is used directly (hash-cache hit: no file read). -func (s *Store) PutFile(filePath, cachedHash string) (string, error) { +// +// If cachedHash is non-empty and the corresponding object exists, the file +// does not need to be read again. +func (s *Store) PutFile( + filePath string, + cachedHash string, +) (string, error) { if cachedHash != "" && s.Exists(cachedHash) { - return cachedHash, nil // full cache hit: nothing to do + return cachedHash, nil } + data, err := os.ReadFile(filePath) if err != nil { - return "", fmt.Errorf("objects: read %s: %w", filePath, err) + return "", fmt.Errorf( + "objects: read %s: %w", + filePath, + err, + ) } - compress := shouldCompress(data) && !hasCompressedExtension(filePath) - return s.putWithCompression(data, compress) + + compress := shouldCompress(data) && + !hasCompressedExtension(filePath) + + return s.putWithCompression( + data, + compress, + ) } +// shouldCompress determines whether the data should be compressed. +// +// Small files are kept raw because compression overhead is usually not worth +// it. Common already-compressed formats are also kept raw. func shouldCompress(data []byte) bool { if len(data) < 1024 { - return false // skip compressing very small files + return false } + if len(data) >= 4 { - // Gzip - if data[0] == 0x1f && data[1] == 0x8b { + // Gzip. + if data[0] == 0x1f && + data[1] == 0x8b { return false } - // Zip - if data[0] == 0x50 && data[1] == 0x4b && data[2] == 0x03 && data[3] == 0x04 { + + // ZIP. + if data[0] == 0x50 && + data[1] == 0x4b && + data[2] == 0x03 && + data[3] == 0x04 { return false } - // PNG - if data[0] == 0x89 && data[1] == 0x50 && data[2] == 0x4e && data[3] == 0x47 { + + // PNG. + if data[0] == 0x89 && + data[1] == 0x50 && + data[2] == 0x4e && + data[3] == 0x47 { return false } - // PDF - if data[0] == 0x25 && data[1] == 0x50 && data[2] == 0x44 && data[3] == 0x46 { + + // PDF. + if data[0] == 0x25 && + data[1] == 0x50 && + data[2] == 0x44 && + data[3] == 0x46 { return false } - // JPEG - if data[0] == 0xff && data[1] == 0xd8 && data[2] == 0xff { + + // JPEG. + if data[0] == 0xff && + data[1] == 0xd8 && + data[2] == 0xff { return false } - // Zstd - if data[0] == 0x28 && data[1] == 0xb5 && data[2] == 0x2f && data[3] == 0xfd { + + // Zstd. + if data[0] == 0x28 && + data[1] == 0xb5 && + data[2] == 0x2f && + data[3] == 0xfd { return false } } + return true } +// hasCompressedExtension avoids recompressing files that are conventionally +// already compressed. func hasCompressedExtension(path string) bool { ext := strings.ToLower(filepath.Ext(path)) + switch ext { - case ".zip", ".tar", ".gz", ".zst", ".tgz", ".png", ".jpg", ".jpeg", ".gif", ".pdf", ".mp4", ".mp3", ".dmg", ".exe", ".dll", ".so", ".dylib", ".rar", ".7z": + case ".zip", + ".tar", + ".gz", + ".zst", + ".tgz", + ".png", + ".jpg", + ".jpeg", + ".gif", + ".pdf", + ".mp4", + ".mp3", + ".dmg", + ".exe", + ".dll", + ".so", + ".dylib", + ".rar", + ".7z": return true } + return false } // Get decompresses and returns the raw bytes for a stored hash. -func (s *Store) Get(hash string) ([]byte, error) { - prefix := filepath.Join(s.baseDir, hash[:2], hash[2:]) +// +// Reading order: +// +// 1. zstd +// 2. raw +// 3. legacy gzip +func (s *Store) Get( + hash string, +) (data []byte, err error) { + start := time.Now() + success := false + + defer func() { + telemetry.RecordCAS( + context.Background(), + "get", + start, + success, + ) + }() + + if len(hash) < 2 { + return nil, fmt.Errorf( + "objects: invalid hash %q", + hash, + ) + } + + prefix := filepath.Join( + s.baseDir, + hash[:2], + hash[2:], + ) - // 1. Try Zstd - if f, err := os.Open(prefix + ".zst"); err == nil { + // 1. Zstd. + if f, openErr := os.Open(prefix + ".zst"); openErr == nil { defer f.Close() - decoder, err := zstd.NewReader(f) - if err != nil { - return nil, fmt.Errorf("objects: zstd open %s: %w", hash[:8], err) + + decoder, decoderErr := zstd.NewReader(f) + if decoderErr != nil { + return nil, fmt.Errorf( + "objects: zstd open %s: %w", + hash[:8], + decoderErr, + ) } defer decoder.Close() - return io.ReadAll(decoder) + + data, err = io.ReadAll(decoder) + if err != nil { + return nil, fmt.Errorf( + "objects: zstd decompress %s: %w", + hash[:8], + err, + ) + } + + success = true + return data, nil } - // 2. Try Raw (uncompressed) - if f, err := os.Open(prefix + ".raw"); err == nil { + // 2. Raw. + if f, openErr := os.Open(prefix + ".raw"); openErr == nil { defer f.Close() - return io.ReadAll(f) + + data, err = io.ReadAll(f) + if err != nil { + return nil, fmt.Errorf( + "objects: read raw %s: %w", + hash[:8], + err, + ) + } + + success = true + return data, nil } - // 3. Try Gzip (legacy fallback) - if f, err := os.Open(prefix + ".gz"); err == nil { + // 3. Legacy gzip. + if f, openErr := os.Open(prefix + ".gz"); openErr == nil { defer f.Close() - gz, err := gzip.NewReader(f) - if err != nil { - return nil, fmt.Errorf("objects: gzip open %s: %w", hash[:8], err) + + gz, gzipErr := gzip.NewReader(f) + if gzipErr != nil { + return nil, fmt.Errorf( + "objects: gzip open %s: %w", + hash[:8], + gzipErr, + ) } defer gz.Close() - return io.ReadAll(gz) + + data, err = io.ReadAll(gz) + if err != nil { + return nil, fmt.Errorf( + "objects: gzip decompress %s: %w", + hash[:8], + err, + ) + } + + success = true + return data, nil } - return nil, fmt.Errorf("objects: open %s: file not found", hash[:8]) + return nil, fmt.Errorf( + "objects: open %s: file not found", + hash[:8], + ) } -// ExtractTo writes the decompressed content of hash to dstPath with the given mode. -func (s *Store) ExtractTo(hash string, dstPath string, mode os.FileMode) error { - prefix := filepath.Join(s.baseDir, hash[:2], hash[2:]) +// ExtractTo writes the content of hash to dstPath with the given mode. +// +// Raw objects use CopyOrCloneFile when possible, avoiding an unnecessary +// read/write cycle. Compressed objects are decompressed normally. +func (s *Store) ExtractTo( + hash string, + dstPath string, + mode os.FileMode, +) error { + if len(hash) < 2 { + return fmt.Errorf( + "objects: invalid hash %q", + hash, + ) + } + + prefix := filepath.Join( + s.baseDir, + hash[:2], + hash[2:], + ) - // If the file is stored as a raw uncompressed file (.raw), we can use reflink/zero-copy copy! rawPath := prefix + ".raw" + + // Raw objects can be copied or cloned directly. if _, err := os.Stat(rawPath); err == nil { - err := util.CopyOrCloneFile(rawPath, dstPath) - if err != nil { + if err := os.MkdirAll( + filepath.Dir(dstPath), + 0755, + ); err != nil { return err } - return os.Chmod(dstPath, mode.Perm()) + + if err := util.CopyOrCloneFile( + rawPath, + dstPath, + ); err != nil { + return err + } + + return os.Chmod( + dstPath, + mode.Perm(), + ) } - // Otherwise, fallback to reading and decompressing normally + // Compressed objects need to be decompressed. data, err := s.Get(hash) if err != nil { return err } - if err := os.MkdirAll(filepath.Dir(dstPath), 0755); err != nil { + + if err := os.MkdirAll( + filepath.Dir(dstPath), + 0755, + ); err != nil { return err } - return os.WriteFile(dstPath, data, mode) + + return os.WriteFile( + dstPath, + data, + mode, + ) } -// RestoreTree extracts all files described by tree (path → FileEntry) into dstDir -// using a parallel worker pool for maximum throughput. -// The optional onProgress callback is invoked after each file is extracted. -func (s *Store) RestoreTree(tree map[string]FileEntry, dstDir string, onProgress func()) error { +// RestoreTree extracts all files described by tree into dstDir using a +// parallel worker pool. +// +// The optional onProgress callback is invoked after each successfully +// extracted file. +func (s *Store) RestoreTree( + tree map[string]FileEntry, + dstDir string, + onProgress func(), +) error { type job struct { rel string hash string @@ -293,20 +609,36 @@ func (s *Store) RestoreTree(tree map[string]FileEntry, dstDir string, onProgress } numWorkers := runtime.NumCPU() + if numWorkers < 1 { + numWorkers = 1 + } + jobs := make(chan job, numWorkers*2) errs := make(chan error, numWorkers) var wg sync.WaitGroup + for i := 0; i < numWorkers; i++ { wg.Add(1) + go func() { defer wg.Done() + for j := range jobs { - dst := filepath.Join(dstDir, filepath.FromSlash(j.rel)) - if err := s.ExtractTo(j.hash, dst, j.mode); err != nil { + dst := filepath.Join( + dstDir, + filepath.FromSlash(j.rel), + ) + + if err := s.ExtractTo( + j.hash, + dst, + j.mode, + ); err != nil { errs <- err return } + if onProgress != nil { onProgress() } @@ -315,37 +647,88 @@ func (s *Store) RestoreTree(tree map[string]FileEntry, dstDir string, onProgress } for rel, entry := range tree { - jobs <- job{rel: rel, hash: entry.Hash, mode: entry.Mode} + jobs <- job{ + rel: rel, + hash: entry.Hash, + mode: entry.Mode, + } } + close(jobs) + wg.Wait() close(errs) - return <-errs + for err := range errs { + return err + } + + return nil } -// AllHashes returns every hash currently stored (used by GC). +// AllHashes returns every hash currently stored. +// +// Both current formats (.zst/.raw) and legacy gzip (.gz) objects are included. func (s *Store) AllHashes() ([]string, error) { var hashes []string - err := filepath.Walk(s.baseDir, func(path string, info os.FileInfo, err error) error { - if err != nil || info.IsDir() { - return err - } - dir := filepath.Base(filepath.Dir(path)) - name := info.Name() - if len(name) > 4 && (name[len(name)-4:] == ".zst" || name[len(name)-4:] == ".raw") { - hashes = append(hashes, dir+name[:len(name)-4]) - } else if len(name) > 3 && name[len(name)-3:] == ".gz" { - hashes = append(hashes, dir+name[:len(name)-3]) - } - return nil - }) + + err := filepath.Walk( + s.baseDir, + func( + path string, + info os.FileInfo, + err error, + ) error { + if err != nil { + return err + } + + if info.IsDir() { + return nil + } + + dir := filepath.Base( + filepath.Dir(path), + ) + name := info.Name() + + if len(name) > 4 && + (name[len(name)-4:] == ".zst" || + name[len(name)-4:] == ".raw") { + hashes = append( + hashes, + dir+name[:len(name)-4], + ) + + return nil + } + + if len(name) > 3 && + name[len(name)-3:] == ".gz" { + hashes = append( + hashes, + dir+name[:len(name)-3], + ) + } + + return nil + }, + ) + return hashes, err } // GarbageCollect deletes objects that are not referenced by any hash in keep. -// Returns the number of bytes freed. -func (s *Store) GarbageCollect(keep map[string]bool, dryRun bool) (int64, int, error) { +// +// It returns: +// +// (bytes freed, objects removed, error) +// +// If dryRun is true, objects are not actually deleted. +func (s *Store) GarbageCollect( + keep map[string]bool, + dryRun bool, +) (int64, int, error) { all, err := s.AllHashes() if err != nil { return 0, 0, err @@ -353,27 +736,40 @@ func (s *Store) GarbageCollect(keep map[string]bool, dryRun bool) (int64, int, e var freed int64 var count int + for _, h := range all { if keep[h] { continue } + path := s.objectPath(h) + info, err := os.Stat(path) if err != nil { continue } + freed += info.Size() count++ - if !dryRun { - // Make writable before removing (objects stored 0444) - _ = os.Chmod(path, 0644) - _ = os.Remove(path) + + if dryRun { + continue + } + + // Objects are normally read-only. + _ = os.Chmod(path, 0644) + + if err := os.Remove(path); err != nil { + continue } } + return freed, count, nil } -// FileEntry is a reference to a stored blob. Used by RestoreTree and manifests. +// FileEntry is a reference to a stored blob. +// +// Used by RestoreTree and manifests. type FileEntry struct { Hash string `json:"hash"` Mode os.FileMode `json:"mode"` diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go index 5d1d3d0..d47236f 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,6 +39,7 @@ import ( "eko/internal/cache" "eko/internal/manifest" "eko/internal/objects" + "eko/internal/telemetry" "eko/internal/util" ) @@ -69,41 +71,67 @@ func CountFiles() (int, error) { // 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). -// The optional onProgress callback is invoked after each file is processed. func CreateSnapshot(db *sql.DB, withEnv bool, onProgress func()) (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, + ) + }() + + // Generate a unique snapshot ID. id, err = generateID() if err != nil { - return "", "", err + return "", "", fmt.Errorf("snapshot: generate ID: %w", err) } + // Open the CAS object store. store, err := objects.New(ekoDir) if err != nil { return "", "", fmt.Errorf("snapshot: open object store: %w", err) } + // Open the hash cache. Cache failure is non-fatal; we can still + // create the snapshot by hashing files normally. hc, err := cache.New(db) if err != nil { - // Hash cache failure is non-fatal: fall back to always hashing. hc = nil } else { defer hc.Close() } + // Walk the workspace and store files in the CAS. tree, err := buildTree(store, hc, onProgress) if err != nil { return "", "", fmt.Errorf("snapshot: build tree: %w", err) } - // Capture and store environment variables as a blob if requested. + // Optionally capture the current environment variables as a CAS blob. var envHash string if withEnv { - var err error envHash, err = captureEnvVars(store) if err != nil { return "", "", fmt.Errorf("snapshot: capture env: %w", err) } } + // Build the snapshot manifest. m := &manifest.Manifest{ ID: id, CreatedAt: time.Now(), @@ -111,11 +139,15 @@ func CreateSnapshot(db *sql.DB, withEnv bool, onProgress func()) (id, path strin EnvHash: envHash, } + // Persist the manifest. if err := manifest.Write(ekoDir, m); err != nil { return "", "", fmt.Errorf("snapshot: write manifest: %w", err) } 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