Merge pull request #10 from moov-io/feat-add-telemetry

meta: setup telemetry

Author: adamdecaf

configs/config.default.yml

@@ -6,3 +6,5 @@ RailMsgSQL:
     Admin:
       Bind:
         Address: ":9696"
+  Telemetry:
+    ServiceName: "rail-msg-sql"

internal/search/service.go

@@ -9,12 +9,15 @@ import (
 	"github.com/moov-io/ach"
 	"github.com/moov-io/ach/cmd/achcli/describe/mask"
 	"github.com/moov-io/base/log"
+	"github.com/moov-io/base/telemetry"
 	railmsgsql "github.com/moov-io/rail-msg-sql"
 	"github.com/moov-io/rail-msg-sql/internal/achhelp"
 	"github.com/moov-io/rail-msg-sql/internal/storage"
 
 	_ "github.com/ncruces/go-sqlite3/driver"
 	_ "github.com/ncruces/go-sqlite3/embed"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Service interface {
@@ -71,19 +74,24 @@ func (s *service) Close() error {
 }
 
 func (s *service) IngestACHFiles(ctx context.Context, params storage.FilterParams) error {
+	ctx, span := telemetry.StartSpan(ctx, "ingest-ach-files")
+	defer span.End()
+
 	files, err := s.fileStorage.ListAchFiles(ctx, params)
 	if err != nil {
 		return fmt.Errorf("ingesting ach files: %w", err)
 	}
 
 	for idx := range files {
-		if files[idx].File == nil {
+		file := files[idx]
+
+		if file.File == nil {
 			continue
 		}
 
-		err := s.IngestACHFile(ctx, files[idx].Filename, files[idx].File)
+		err := s.IngestACHFile(ctx, file.Filename, file.File)
 		if err != nil {
-			return fmt.Errorf("ingesting %s failed: %w", files[idx].Filename, err)
+			return fmt.Errorf("ingesting %s failed: %w", file.Filename, err)
 		}
 	}
 
@@ -92,6 +100,11 @@ func (s *service) IngestACHFiles(ctx context.Context, params storage.FilterParam
 
 // insertFile inserts an ACH file's header and control into ach_files.
 func (s *service) insertFile(ctx context.Context, tx *sql.Tx, filename string, file *ach.File) error {
+	ctx, span := telemetry.StartSpan(ctx, "sql-insert-file", trace.WithAttributes(
+		attribute.String("filename", filename),
+	))
+	defer span.End()
+
 	query := `
         INSERT OR IGNORE INTO ach_files (
             file_id,
@@ -412,6 +425,11 @@ func (s *service) IngestACHFile(ctx context.Context, filename string, file *ach.
 		return errors.New("nil File")
 	}
 
+	ctx, span := telemetry.StartSpan(ctx, "ingest-ach-file", trace.WithAttributes(
+		attribute.String("filename", filename),
+	))
+	defer span.End()
+
 	// Make sure to normalize the IDs
 	file = achhelp.PopulateIDs(file)
 
@@ -493,6 +511,11 @@ func (s *service) Search(ctx context.Context, query string, params storage.Filte
 		return nil, fmt.Errorf("query cannot be empty")
 	}
 
+	ctx, span := telemetry.StartSpan(ctx, "search-files", trace.WithAttributes(
+		attribute.String("sql.query", query),
+	))
+	defer span.End()
+
 	rows, err := s.db.QueryContext(ctx, query)
 	if err != nil {
 		return nil, fmt.Errorf("failed to execute query: %w", err)

internal/storage/repository.go

@@ -4,9 +4,14 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"time"
 
 	"github.com/moov-io/ach"
 	"github.com/moov-io/ach-web-viewer/pkg/filelist"
+	"github.com/moov-io/base/telemetry"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Repository struct {
@@ -32,6 +37,13 @@ func NewRepository(config Config) (*Repository, error) {
 }
 
 func (r *Repository) ListAchFiles(ctx context.Context, params FilterParams) ([]File, error) {
+	_, span := telemetry.StartSpan(ctx, "list-ach-files", trace.WithAttributes(
+		attribute.String("filter.start", params.StartDate.Format(time.RFC3339)),
+		attribute.String("filter.end", params.EndDate.Format(time.RFC3339)),
+		attribute.String("filter.pattern", params.Pattern),
+	))
+	defer span.End()
+
 	resp, err := r.ach.GetFiles(filelist.ListOpts{
 		StartDate: params.StartDate,
 		EndDate:   params.EndDate,

pkg/service/environment.go

@@ -3,8 +3,13 @@
 package service
 
 import (
+	"context"
+	"fmt"
+
 	"github.com/moov-io/base/log"
 	"github.com/moov-io/base/stime"
+	"github.com/moov-io/base/telemetry"
+	railmsgsql "github.com/moov-io/rail-msg-sql"
 
 	"github.com/gorilla/mux"
 )
@@ -44,6 +49,16 @@ func NewEnvironment(env *Environment) (*Environment, error) {
 		env.TimeService = stime.NewSystemTimeService()
 	}
 
+	telemetryShutdownFunc, err := telemetry.SetupTelemetry(context.Background(), env.Config.Telemetry, railmsgsql.Version)
+	if err != nil {
+		return env, fmt.Errorf("setting up telemetry failed: %w", err)
+	}
+	prev := env.Shutdown
+	env.Shutdown = func() {
+		prev()
+		telemetryShutdownFunc()
+	}
+
 	// router
 	if env.PublicRouter == nil {
 		env.PublicRouter = mux.NewRouter()

pkg/service/model_config.go

@@ -3,6 +3,7 @@
 package service
 
 import (
+	"github.com/moov-io/base/telemetry"
 	"github.com/moov-io/rail-msg-sql/internal/search"
 	"github.com/moov-io/rail-msg-sql/internal/storage"
 )
@@ -13,6 +14,8 @@ type GlobalConfig struct {
 
 // Config defines all the configuration for the app
 type Config struct {
+	Telemetry telemetry.Config
+
 	Servers ServerConfig
 
 	Search  search.Config