24640136570

Committed 19 Apr 2026 09:58PM UTC coverage: 56.328% (+1.9%) from 54.432%

Build # 24640136570

Build Type

Pull #1394

github

Committed by

joschi

Commit Message

fix: address second round of Copilot review comments

- README.md: fix db.system → db.system.name in database span table
- otel.go: call otel.Handle(err) on instrument creation errors instead
  of silently discarding them (doc comment already promised this)
- migrate.go: use attribute.Int64 for uint version fields to avoid
  int overflow on 32-bit platforms
- migrate.go: sanitize database.Error in otelSpanSetError to avoid
  leaking migration SQL into trace span status descriptions
- database/oteldriver.go: same SQL-leak fix in endSpan; add errors import
- source/oteldriver.go: use attribute.Int64 for uint version fields
- source/oteldriver_test.go: assert span.Status().Code != codes.Error
  directly rather than comparing full sdktrace.Status struct
- source/pkger/pkger.go: fix import grouping (stdlib before third-party)
- source/google_cloud_storage/storage.go: merge context into stdlib group

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Pull Request Pull Request #1394: feat: OpenTelemetry instrumentation

Coverage Stats

1020 of 1236 new or added lines in 47 files covered. (82.52%)

8 existing lines in 7 files now uncovered.

4731 of 8399 relevant lines covered (56.33%)

50.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.61

/database/sqlite3/sqlite3.go

package sqlite3

import (
        "context"
        "database/sql"
        "errors"
        "fmt"
        "io"
        nurl "net/url"
        "strconv"
        "strings"
        "sync/atomic"

        "github.com/XSAM/otelsql"
        "github.com/golang-migrate/migrate/v4"
        "github.com/golang-migrate/migrate/v4/database"
        _ "github.com/mattn/go-sqlite3"
        semconv "go.opentelemetry.io/otel/semconv/v1.39.0"
)

func init() {
        database.Register("sqlite3", &Sqlite{})
}

var DefaultMigrationsTable = "schema_migrations"
var (
        ErrDatabaseDirty  = fmt.Errorf("database is dirty")
        ErrNilConfig      = fmt.Errorf("no config")
        ErrNoDatabaseName = fmt.Errorf("no database name")
)

type Config struct {
        MigrationsTable string
        DatabaseName    string
        NoTxWrap        bool
}

type Sqlite struct {
        db       *sql.DB
        isLocked atomic.Bool

        config *Config
}

func WithInstance(ctx context.Context, instance *sql.DB, config *Config) (database.Driver, error) {
        if config == nil {
                return nil, ErrNilConfig
        }

        if err := instance.Ping(); err != nil {
                return nil, err
        }

        if len(config.MigrationsTable) == 0 {
                config.MigrationsTable = DefaultMigrationsTable
        }

        mx := &Sqlite{
                db:     instance,
                config: config,
        }
        if err := mx.ensureVersionTable(ctx); err != nil {
                return nil, err
        }
        return mx, nil
}

// ensureVersionTable checks if versions table exists and, if not, creates it.
// Note that this function locks the database, which deviates from the usual
// convention of "caller locks" in the Sqlite type.
func (m *Sqlite) ensureVersionTable(ctx context.Context) (err error) {
        if err = m.Lock(ctx); err != nil {
                return err
        }

        defer func() {
                if e := m.Unlock(ctx); e != nil {
                        err = errors.Join(err, e)
                }
        }()

        query := fmt.Sprintf(`
        CREATE TABLE IF NOT EXISTS %s (version uint64,dirty bool);
  CREATE UNIQUE INDEX IF NOT EXISTS version_unique ON %s (version);
  `, m.config.MigrationsTable, m.config.MigrationsTable)

        if _, err := m.db.ExecContext(ctx, query); err != nil {
                return err
        }
        return nil
}

func (m *Sqlite) Open(ctx context.Context, url string) (database.Driver, error) {
        purl, err := nurl.Parse(url)
        if err != nil {
                return nil, err
        }
        dbfile := strings.Replace(migrate.FilterCustomQuery(purl).String(), "sqlite3://", "", 1)
        db, err := otelsql.Open("sqlite3", dbfile,
                otelsql.WithAttributes(semconv.DBSystemNameSQLite),
        )
        if err != nil {
                return nil, err
        }

        qv := purl.Query()

        migrationsTable := qv.Get("x-migrations-table")
        if len(migrationsTable) == 0 {
                migrationsTable = DefaultMigrationsTable
        }

        noTxWrap := false
        if v := qv.Get("x-no-tx-wrap"); v != "" {
                noTxWrap, err = strconv.ParseBool(v)
                if err != nil {
                        return nil, fmt.Errorf("x-no-tx-wrap: %s", err)
                }
        }

        mx, err := WithInstance(ctx, db, &Config{
                DatabaseName:    purl.Path,
                MigrationsTable: migrationsTable,
                NoTxWrap:        noTxWrap,
        })
        if err != nil {
                return nil, err
        }
        return mx, nil
}

func (m *Sqlite) Close(ctx context.Context) error {
        return m.db.Close()
}

func (m *Sqlite) Drop(ctx context.Context) (err error) {
        query := `SELECT name FROM sqlite_master WHERE type = 'table';`
        tables, err := m.db.QueryContext(ctx, query)
        if err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }
        defer func() {
                if errClose := tables.Close(); errClose != nil {
                        err = errors.Join(err, errClose)
                }
        }()

        tableNames := make([]string, 0)
        for tables.Next() {
                var tableName string
                if err := tables.Scan(&tableName); err != nil {
                        return err
                }
                if len(tableName) > 0 {
                        tableNames = append(tableNames, tableName)
                }
        }
        if err := tables.Err(); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        if len(tableNames) > 0 {
                for _, t := range tableNames {
                        query := "DROP TABLE " + t
                        err = m.executeQuery(ctx, query)
                        if err != nil {
                                return &database.Error{OrigErr: err, Query: []byte(query)}
                        }
                }
                query := "VACUUM"
                _, err = m.db.QueryContext(ctx, query)
                if err != nil {
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
        }

        return nil
}

func (m *Sqlite) Lock(ctx context.Context) error {
        if !m.isLocked.CompareAndSwap(false, true) {
                return database.ErrLocked
        }
        return nil
}

func (m *Sqlite) Unlock(ctx context.Context) error {
        if !m.isLocked.CompareAndSwap(true, false) {
                return database.ErrNotLocked
        }
        return nil
}

func (m *Sqlite) Run(ctx context.Context, migration io.Reader) error {
        migr, err := io.ReadAll(migration)
        if err != nil {
                return err
        }
        query := string(migr[:])

        if m.config.NoTxWrap {
                return m.executeQueryNoTx(ctx, query)
        }
        return m.executeQuery(ctx, query)
}

func (m *Sqlite) executeQuery(ctx context.Context, query string) error {
        tx, err := m.db.BeginTx(ctx, nil)
        if err != nil {
                return &database.Error{OrigErr: err, Err: "transaction start failed"}
        }
        if _, err := tx.ExecContext(ctx, query); err != nil {
                if errRollback := tx.Rollback(); errRollback != nil {
                        err = errors.Join(err, errRollback)
                }
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }
        if err := tx.Commit(); err != nil {
                return &database.Error{OrigErr: err, Err: "transaction commit failed"}
        }
        return nil
}

func (m *Sqlite) executeQueryNoTx(ctx context.Context, query string) error {
        if _, err := m.db.ExecContext(ctx, query); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }
        return nil
}

func (m *Sqlite) SetVersion(ctx context.Context, version int, dirty bool) error {
        tx, err := m.db.BeginTx(ctx, nil)
        if err != nil {
                return &database.Error{OrigErr: err, Err: "transaction start failed"}
        }

        query := "DELETE FROM " + m.config.MigrationsTable
        if _, err := tx.ExecContext(ctx, query); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        // Also re-write the schema version for nil dirty versions to prevent
        // empty schema version for failed down migration on the first migration
        // See: https://github.com/golang-migrate/migrate/issues/330
        if version >= 0 || (version == database.NilVersion && dirty) {
                query := fmt.Sprintf(`INSERT INTO %s (version, dirty) VALUES (?, ?)`, m.config.MigrationsTable)
                if _, err := tx.ExecContext(ctx, query, version, dirty); err != nil {
                        if errRollback := tx.Rollback(); errRollback != nil {
                                err = errors.Join(err, errRollback)
                        }
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
        }

        if err := tx.Commit(); err != nil {
                return &database.Error{OrigErr: err, Err: "transaction commit failed"}
        }

        return nil
}

func (m *Sqlite) Version(ctx context.Context) (version int, dirty bool, err error) {
        query := "SELECT version, dirty FROM " + m.config.MigrationsTable + " LIMIT 1"
        err = m.db.QueryRowContext(ctx, query).Scan(&version, &dirty)
        if err != nil {
                return database.NilVersion, false, nil
        }
        return version, dirty, nil
}

1	package sqlite3
2
3	import (
4	"context"
5	"database/sql"
6	"errors"
7	"fmt"
8	"io"
9	nurl "net/url"
10	"strconv"
11	"strings"
12	"sync/atomic"
13
14	"github.com/XSAM/otelsql"
15	"github.com/golang-migrate/migrate/v4"
16	"github.com/golang-migrate/migrate/v4/database"
17	_ "github.com/mattn/go-sqlite3"
18	semconv "go.opentelemetry.io/otel/semconv/v1.39.0"
19	)
20
21	func init() {	2✔
22	database.Register("sqlite3", &Sqlite{})	2✔
23	}	2✔
24
25	var DefaultMigrationsTable = "schema_migrations"
26	var (
27	ErrDatabaseDirty = fmt.Errorf("database is dirty")
28	ErrNilConfig = fmt.Errorf("no config")
29	ErrNoDatabaseName = fmt.Errorf("no database name")
30	)
31
32	type Config struct {
33	MigrationsTable string
34	DatabaseName string
35	NoTxWrap bool
36	}
37
38	type Sqlite struct {
39	db *sql.DB
40	isLocked atomic.Bool
41
42	config *Config
43	}
44
45	func WithInstance(ctx context.Context, instance sql.DB, config Config) (database.Driver, error) {	10✔
46	if config == nil {	10✔
47	return nil, ErrNilConfig	×
48	}	×
49
50	if err := instance.Ping(); err != nil {	10✔
51	return nil, err	×
52	}	×
53
54	if len(config.MigrationsTable) == 0 {	12✔
55	config.MigrationsTable = DefaultMigrationsTable	2✔
56	}	2✔
57
58	mx := &Sqlite{	10✔
59	db: instance,	10✔
60	config: config,	10✔
61	}	10✔
62	if err := mx.ensureVersionTable(ctx); err != nil {	10✔
63	return nil, err	×
64	}	×
65	return mx, nil	10✔
66	}
67
68	// ensureVersionTable checks if versions table exists and, if not, creates it.
69	// Note that this function locks the database, which deviates from the usual
70	// convention of "caller locks" in the Sqlite type.
71	func (m *Sqlite) ensureVersionTable(ctx context.Context) (err error) {	10✔
72	if err = m.Lock(ctx); err != nil {	10✔
73	return err	×
74	}	×
75
76	defer func() {	20✔
77	if e := m.Unlock(ctx); e != nil {	10✔
78	err = errors.Join(err, e)	×
79	}	×
80	}()
81
82	query := fmt.Sprintf(`	10✔
83	CREATE TABLE IF NOT EXISTS %s (version uint64,dirty bool);	10✔
84	CREATE UNIQUE INDEX IF NOT EXISTS version_unique ON %s (version);	10✔
85	`, m.config.MigrationsTable, m.config.MigrationsTable)	10✔
86		10✔
87	if _, err := m.db.ExecContext(ctx, query); err != nil {	10✔
88	return err	×
89	}	×
90	return nil	10✔
91	}
92
93	func (m *Sqlite) Open(ctx context.Context, url string) (database.Driver, error) {	8✔
94	purl, err := nurl.Parse(url)	8✔
95	if err != nil {	8✔
96	return nil, err	×
97	}	×
98	dbfile := strings.Replace(migrate.FilterCustomQuery(purl).String(), "sqlite3://", "", 1)	8✔
99	db, err := otelsql.Open("sqlite3", dbfile,	8✔
100	otelsql.WithAttributes(semconv.DBSystemNameSQLite),	8✔
101	)	8✔
102	if err != nil {	8✔
103	return nil, err	×
104	}	×
105
106	qv := purl.Query()	8✔
107		8✔
108	migrationsTable := qv.Get("x-migrations-table")	8✔
109	if len(migrationsTable) == 0 {	16✔
110	migrationsTable = DefaultMigrationsTable	8✔
111	}	8✔
112
113	noTxWrap := false	8✔
114	if v := qv.Get("x-no-tx-wrap"); v != "" {	12✔
115	noTxWrap, err = strconv.ParseBool(v)	4✔
116	if err != nil {	6✔
117	return nil, fmt.Errorf("x-no-tx-wrap: %s", err)	2✔
118	}	2✔
119	}
120
121	mx, err := WithInstance(ctx, db, &Config{	6✔
122	DatabaseName: purl.Path,	6✔
123	MigrationsTable: migrationsTable,	6✔
124	NoTxWrap: noTxWrap,	6✔
125	})	6✔
126	if err != nil {	6✔
127	return nil, err	×
128	}	×
129	return mx, nil	6✔
130	}
131
NEW 132	func (m *Sqlite) Close(ctx context.Context) error {	×
133	return m.db.Close()	×
134	}	×
135
136	func (m *Sqlite) Drop(ctx context.Context) (err error) {	8✔
137	query := `SELECT name FROM sqlite_master WHERE type = 'table';`	8✔
138	tables, err := m.db.QueryContext(ctx, query)	8✔
139	if err != nil {	8✔
140	return &database.Error{OrigErr: err, Query: []byte(query)}	×
141	}	×
142	defer func() {	16✔
143	if errClose := tables.Close(); errClose != nil {	8✔
144	err = errors.Join(err, errClose)	×
145	}	×
146	}()
147
148	tableNames := make([]string, 0)	8✔
149	for tables.Next() {	24✔
150	var tableName string	16✔
151	if err := tables.Scan(&tableName); err != nil {	16✔
152	return err	×
153	}	×
154	if len(tableName) > 0 {	32✔
155	tableNames = append(tableNames, tableName)	16✔
156	}	16✔
157	}
158	if err := tables.Err(); err != nil {	8✔
159	return &database.Error{OrigErr: err, Query: []byte(query)}	×
160	}	×
161
162	if len(tableNames) > 0 {	16✔
163	for _, t := range tableNames {	24✔
164	query := "DROP TABLE " + t	16✔
165	err = m.executeQuery(ctx, query)	16✔
166	if err != nil {	16✔
167	return &database.Error{OrigErr: err, Query: []byte(query)}	×
168	}	×
169	}
170	query := "VACUUM"	8✔
171	_, err = m.db.QueryContext(ctx, query)	8✔
172	if err != nil {	8✔
173	return &database.Error{OrigErr: err, Query: []byte(query)}	×
174	}	×
175	}
176
177	return nil	8✔
178	}
179
180	func (m *Sqlite) Lock(ctx context.Context) error {	34✔
181	if !m.isLocked.CompareAndSwap(false, true) {	40✔
182	return database.ErrLocked	6✔
183	}	6✔
184	return nil	28✔
185	}
186
187	func (m *Sqlite) Unlock(ctx context.Context) error {	28✔
188	if !m.isLocked.CompareAndSwap(true, false) {	28✔
189	return database.ErrNotLocked	×
190	}	×
191	return nil	28✔
192	}
193
194	func (m *Sqlite) Run(ctx context.Context, migration io.Reader) error {	14✔
195	migr, err := io.ReadAll(migration)	14✔
196	if err != nil {	14✔
197	return err	×
198	}	×
199	query := string(migr[:])	14✔
200		14✔
201	if m.config.NoTxWrap {	16✔
202	return m.executeQueryNoTx(ctx, query)	2✔
203	}	2✔
204	return m.executeQuery(ctx, query)	12✔
205	}
206
207	func (m *Sqlite) executeQuery(ctx context.Context, query string) error {	28✔
208	tx, err := m.db.BeginTx(ctx, nil)	28✔
209	if err != nil {	28✔
210	return &database.Error{OrigErr: err, Err: "transaction start failed"}	×
211	}	×
212	if _, err := tx.ExecContext(ctx, query); err != nil {	28✔
213	if errRollback := tx.Rollback(); errRollback != nil {	×
214	err = errors.Join(err, errRollback)	×
215	}	×
216	return &database.Error{OrigErr: err, Query: []byte(query)}	×
217	}
218	if err := tx.Commit(); err != nil {	28✔
219	return &database.Error{OrigErr: err, Err: "transaction commit failed"}	×
220	}	×
221	return nil	28✔
222	}
223
224	func (m *Sqlite) executeQueryNoTx(ctx context.Context, query string) error {	2✔
225	if _, err := m.db.ExecContext(ctx, query); err != nil {	2✔
226	return &database.Error{OrigErr: err, Query: []byte(query)}	×
227	}	×
228	return nil	2✔
229	}
230
231	func (m *Sqlite) SetVersion(ctx context.Context, version int, dirty bool) error {	52✔
232	tx, err := m.db.BeginTx(ctx, nil)	52✔
233	if err != nil {	52✔
234	return &database.Error{OrigErr: err, Err: "transaction start failed"}	×
235	}	×
236
237	query := "DELETE FROM " + m.config.MigrationsTable	52✔
238	if _, err := tx.ExecContext(ctx, query); err != nil {	52✔
239	return &database.Error{OrigErr: err, Query: []byte(query)}	×
240	}	×
241
242	// Also re-write the schema version for nil dirty versions to prevent
243	// empty schema version for failed down migration on the first migration
244	// See: https://github.com/golang-migrate/migrate/issues/330
245	if version >= 0 \|\| (version == database.NilVersion && dirty) {	98✔
246	query := fmt.Sprintf(`INSERT INTO %s (version, dirty) VALUES (?, ?)`, m.config.MigrationsTable)	46✔
247	if _, err := tx.ExecContext(ctx, query, version, dirty); err != nil {	46✔
248	if errRollback := tx.Rollback(); errRollback != nil {	×
249	err = errors.Join(err, errRollback)	×
250	}	×
251	return &database.Error{OrigErr: err, Query: []byte(query)}	×
252	}
253	}
254
255	if err := tx.Commit(); err != nil {	52✔
256	return &database.Error{OrigErr: err, Err: "transaction commit failed"}	×
257	}	×
258
259	return nil	52✔
260	}
261
262	func (m *Sqlite) Version(ctx context.Context) (version int, dirty bool, err error) {	46✔
263	query := "SELECT version, dirty FROM " + m.config.MigrationsTable + " LIMIT 1"	46✔
264	err = m.db.QueryRowContext(ctx, query).Scan(&version, &dirty)	46✔
265	if err != nil {	62✔
266	return database.NilVersion, false, nil	16✔
267	}	16✔
268	return version, dirty, nil	30✔
269	}

golang-migrate / migrate / 24640136570

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous