Merge remote-tracking branch 'origin/develop'

Author: robbert229

.github/workflows/golang.yaml

@@ -36,6 +36,9 @@ jobs:
       - name: Build
         run: go build -v ./...
       - name: Test
-        env:
-          TEST_DATABASE_URL: "postgres://postgres:password@localhost:5432/jaeger"
-        run: go test -test.parallel=1 ./...
+        uses: nick-fields/retry@v2
+        with:
+          max_attempts: 3
+          timeout_minutes: 5
+          command: |
+            TEST_DATABASE_URL='postgres://postgres:password@localhost:5432/jaeger' go test -test.parallel=1 ./...

.github/workflows/release.yaml

@@ -22,8 +22,6 @@ jobs:
     env:
       flags: ""
     steps:
-      - if: ${{ !startsWith(github.ref, 'refs/tags/v') }}
-        run: echo "flags=--snapshot" >> $GITHUB_ENV
       - name: Checkout
         uses: actions/checkout@v4
         with:

.gitignore

@@ -16,4 +16,6 @@ vendor/
 
 jaeger/
 jaeger-all-in-one
-jaeger-postgresql
+
+jaeger-postgresql
+jaeger-postgresql-cleaner

.goreleaser.yaml

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/robbert229/fxslog"
+	"github.com/robbert229/jaeger-postgresql/internal/logger"
+	"github.com/robbert229/jaeger-postgresql/internal/sql"
+	"go.uber.org/fx"
+)
+
+var (
+	databaseURLFlag      = flag.String("database.url", "", "the postgres connection url to use to connect to the database")
+	databaseMaxConnsFlag = flag.Int("database.max-conns", 20, "Max number of database connections of which the plugin will try to maintain at any given time")
+	loglevelFlag         = flag.String("log-level", "warn", "Minimal allowed log level")
+	maxSpanAgeFlag       = flag.Duration("max-span-age", time.Hour*24, "Maximum age of a span before it will be cleaned")
+)
+
+// ProvideLogger returns a function that provides a logger
+func ProvideLogger() any {
+	return func() (*slog.Logger, error) {
+		return logger.New(loglevelFlag)
+	}
+}
+
+// ProvidePgxPool returns a function that provides a pgx pool
+func ProvidePgxPool() any {
+	return func(logger *slog.Logger, lc fx.Lifecycle) (*pgxpool.Pool, error) {
+		if databaseURLFlag == nil {
+			return nil, fmt.Errorf("invalid database url")
+		}
+
+		databaseURL := *databaseURLFlag
+		if databaseURL == "" {
+			return nil, fmt.Errorf("invalid database url")
+		}
+
+		err := sql.Migrate(logger, databaseURL)
+		if err != nil {
+			return nil, fmt.Errorf("failed to migrate database: %w", err)
+		}
+
+		pgxconfig, err := pgxpool.ParseConfig(databaseURL)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse database url")
+		}
+
+		// handle max conns
+		{
+			var maxConns int32
+			if databaseMaxConnsFlag == nil {
+				maxConns = 20
+			} else {
+				maxConns = int32(*databaseMaxConnsFlag)
+			}
+
+			pgxconfig.MaxConns = maxConns
+		}
+
+		// handle timeout duration
+		connectTimeoutDuration := time.Second * 10
+		pgxconfig.ConnConfig.ConnectTimeout = connectTimeoutDuration
+
+		ctx, cancelFn := context.WithTimeout(context.Background(), connectTimeoutDuration)
+		defer cancelFn()
+
+		pool, err := pgxpool.NewWithConfig(ctx, pgxconfig)
+		if err != nil {
+			return nil, fmt.Errorf("failed to connect to the postgres database: %w", err)
+		}
+
+		logger.Info("connected to postgres")
+
+		lc.Append(fx.Hook{
+			OnStop: func(ctx context.Context) error {
+				pool.Close()
+				return nil
+			},
+		})
+
+		return pool, nil
+	}
+}
+
+// clean purges the old roles from the database
+func clean(ctx context.Context, pool *pgxpool.Pool) (int64, error) {
+	q := sql.New(pool)
+	result, err := q.CleanSpans(ctx, pgtype.Timestamp{Time: time.Now().Add(-1 * *maxSpanAgeFlag), Valid: true})
+	if err != nil {
+		return 0, err
+	}
+
+	return result, nil
+}
+
+func main() {
+	flag.Parse()
+
+	fx.New(
+		fxslog.WithLogger(func(logger *slog.Logger) *slog.Logger {
+			return logger.With("component", "uber/fx")
+		}),
+		fx.Provide(
+			ProvideLogger(),
+			ProvidePgxPool(),
+		),
+		fx.Invoke(func(pool *pgxpool.Pool, lc fx.Lifecycle, logger *slog.Logger, stopper fx.Shutdowner) error {
+			go func(ctx context.Context) {
+				ctx, cancelFn := context.WithTimeout(ctx, time.Minute)
+				defer cancelFn()
+
+				count, err := clean(ctx, pool)
+				if err != nil {
+					logger.Error("failed to clean database", "err", err)
+					stopper.Shutdown(fx.ExitCode(1))
+					return
+				}
+
+				logger.Info("successfully cleaned database", "spans", count)
+				stopper.Shutdown(fx.ExitCode(0))
+			}(context.Background())
+			return nil
+		}),
+	).Run()
+}

cmd/jaeger-postgresql-cleaner/main.go

@@ -7,7 +7,6 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
-	"os"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -16,40 +15,17 @@ import (
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/robbert229/fxslog"
+	"github.com/robbert229/jaeger-postgresql/internal/logger"
 	"github.com/robbert229/jaeger-postgresql/internal/sql"
 	"github.com/robbert229/jaeger-postgresql/internal/store"
-	"google.golang.org/grpc"
-
 	"go.uber.org/fx"
+	"google.golang.org/grpc"
 )
 
 // ProvideLogger returns a function that provides a logger
 func ProvideLogger() any {
 	return func() (*slog.Logger, error) {
-		levelFn := func() (slog.Level, error) {
-			if loglevelFlag == nil {
-				return slog.LevelWarn, nil
-			}
-
-			switch *loglevelFlag {
-			case "info":
-				return slog.LevelInfo, nil
-			case "warn":
-				return slog.LevelWarn, nil
-			case "error":
-				return slog.LevelError, nil
-			case "debug":
-				return slog.LevelDebug, nil
-			default:
-				return 0, fmt.Errorf("invalid log level: %s", *loglevelFlag)
-			}
-		}
-		level, err := levelFn()
-		if err != nil {
-			return nil, fmt.Errorf("failed to build logger: %w", err)
-		}
-
-		return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: level})), nil
+		return logger.New(loglevelFlag)
 	}
 }
 

main.go cmd/jaeger-postgresql/main.gomain.go renamed to cmd/jaeger-postgresql/main.go

@@ -0,0 +1,35 @@
+package logger
+
+import (
+	"fmt"
+	"log/slog"
+	"os"
+)
+
+// New returns a new logger.
+func New(loglevelStr *string) (*slog.Logger, error) {
+	levelFn := func() (slog.Level, error) {
+		if loglevelStr == nil {
+			return slog.LevelWarn, nil
+		}
+
+		switch *loglevelStr {
+		case "info":
+			return slog.LevelInfo, nil
+		case "warn":
+			return slog.LevelWarn, nil
+		case "error":
+			return slog.LevelError, nil
+		case "debug":
+			return slog.LevelDebug, nil
+		default:
+			return 0, fmt.Errorf("invalid log level: %s", *loglevelStr)
+		}
+	}
+	level, err := levelFn()
+	if err != nil {
+		return nil, fmt.Errorf("failed to build logger: %w", err)
+	}
+
+	return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: level})), nil
+}

internal/logger/logger.go

@@ -119,16 +119,7 @@ VALUES(
 )
 RETURNING spans.hack_id;
 
--- -- name: InsertSpanRefs :copyfrom
--- INSERT INTO spanrefs (
---   source_span_id,
---   child_span_id,
---   trace_id,
---   ref_type
--- )
--- VALUES (
---   sqlc.arg(source_span_id),
---   sqlc.arg(child_span_id),
---   sqlc.arg(trace_id),
---   sqlc.arg(ref_type)
--- );
+-- name: CleanSpans :execrows
+
+DELETE FROM spans
+WHERE spans.start_time < sqlc.arg(prune_before)::TIMESTAMP;

internal/sql/query.sql

@@ -184,8 +184,7 @@ func (r *Reader) FindTraceIDs(ctx context.Context, query *spanstore.TraceQueryPa
 		StartTimeMinimum:       EncodeTimestamp(query.StartTimeMin),
 		StartTimeMinimumEnable: query.StartTimeMin.After(time.Time{}),
 		StartTimeMaximum:       EncodeTimestamp(query.StartTimeMax),
-		// StartTimeMaximumEnable: query.StartTimeMax.After(time.Time{}),
-		StartTimeMaximumEnable: false, // maintaining feature parity for now.
+		StartTimeMaximumEnable: query.StartTimeMax.After(time.Time{}),
 		DurationMinimum:        EncodeInterval(query.DurationMin),
 		DurationMinimumEnable:  query.DurationMin > 0*time.Second,
 		DurationMaximum:        EncodeInterval(query.DurationMax),