From 9b269b3d5bddce61978ca15e84125100d49e539c Mon Sep 17 00:00:00 2001 From: Yousef Khan Date: Mon, 6 Jul 2026 16:18:33 -0700 Subject: [PATCH] Update to sqlx --- go.mod | 1 + go.sum | 7 ++++ pkg/timescaledb/convert.go | 72 ++++++++++++++++++++++------------ pkg/timescaledb/timescaledb.go | 5 +-- 4 files changed, 56 insertions(+), 29 deletions(-) diff --git a/go.mod b/go.mod index 88d836c..cdf520f 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/google/uuid v1.6.0 github.com/influxdata/influxdb-client-go/v2 v2.14.0 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf + github.com/jmoiron/sqlx v1.4.0 github.com/lib/pq v1.10.9 github.com/mattn/go-sqlite3 v1.14.28 github.com/pelletier/go-toml/v2 v2.2.3 diff --git a/go.sum b/go.sum index 39333b1..0ac9f67 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk= github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ= github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk= @@ -14,6 +16,8 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= github.com/go-logr/logr v1.4.2/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/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= +github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= 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= @@ -26,6 +30,8 @@ github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjw github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI= github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf h1:7JTmneyiNEwVBOHSjoMxiWAqB992atOeepeFYegn5RU= github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo= +github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o= +github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY= github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -33,6 +39,7 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/mattn/go-sqlite3 v1.14.28 h1:ThEiQrnbtumT+QMknw63Befp/ce/nUPgBPMlRFEum7A= github.com/mattn/go-sqlite3 v1.14.28/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/oapi-codegen/runtime v1.1.1 h1:EXLHh0DXIJnWhdRPN2w4MXAzFyE4CskzhNLUmtpMYro= diff --git a/pkg/timescaledb/convert.go b/pkg/timescaledb/convert.go index 4f79bb8..6234825 100644 --- a/pkg/timescaledb/convert.go +++ b/pkg/timescaledb/convert.go @@ -2,24 +2,24 @@ package timescaledb import ( "context" - "database/sql" "encoding/json" "fmt" "strconv" stdtime "time" + "github.com/jmoiron/sqlx" "github.com/nasa/hermes/pkg/pb" ) const ( insertEventDefSQL = `INSERT INTO eventDefs (id, component, name, severity, args) - VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING` + VALUES (:id, :component, :name, :severity, :args) ON CONFLICT DO NOTHING` insertEventSQL = `INSERT INTO events (eventDefId, time, timeSclk, message, source, args, ert) - VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING` + VALUES (:eventDefId, :time, :timeSclk, :message, :source, :args, :ert) ON CONFLICT DO NOTHING` insertTelemetryDefSQL = `INSERT INTO telemetryDefs (id, name, component) - VALUES ($1, $2, $3) ON CONFLICT DO NOTHING` + VALUES (:id, :name, :component) ON CONFLICT DO NOTHING` insertTelemetrySQL = `INSERT INTO telemetry (time, telemetryDefId, timeSclk, source, labels, key, valueType, integral, floating, boolval, string, bytes, ert) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT DO NOTHING` + VALUES (:time, :telemetryDefId, :timeSclk, :source, :labels, :key, :valueType, :integral, :floating, :boolval, :string, :bytes, :ert) ON CONFLICT DO NOTHING` ) func valuesToAnys(values []*pb.Value) ([]any, error) { @@ -34,7 +34,7 @@ func valuesToAnys(values []*pb.Value) ([]any, error) { return valueAnys, nil } -func InsertEvent(ctx context.Context, db *sql.DB, msg *pb.SourcedEvent) error { +func InsertEvent(ctx context.Context, db *sqlx.DB, msg *pb.SourcedEvent) error { event := msg.GetEvent() eventArgsArray, err := valuesToAnys(event.GetArgs()) @@ -52,30 +52,39 @@ func InsertEvent(ctx context.Context, db *sql.DB, msg *pb.SourcedEvent) error { return fmt.Errorf("failed to marshal def args: %w", err) } - tx, err := db.BeginTx(ctx, nil) + tx, err := db.BeginTxx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback() ref := event.GetRef() - if _, err := tx.ExecContext(ctx, insertEventDefSQL, - ref.GetId(), ref.GetComponent(), ref.GetName(), ref.GetSeverity(), string(defArgs), - ); err != nil { + if _, err := tx.NamedExecContext(ctx, insertEventDefSQL, map[string]any{ + "id": ref.GetId(), + "component": ref.GetComponent(), + "name": ref.GetName(), + "severity": ref.GetSeverity(), + "args": string(defArgs), + }); err != nil { return fmt.Errorf("failed to insert event def: %w", err) } - if _, err := tx.ExecContext(ctx, insertEventSQL, - ref.GetId(), event.GetTime().GetUnix().AsTime(), event.GetTime().GetSclk(), - event.GetMessage(), msg.GetSource(), string(eventArgs), stdtime.Now(), - ); err != nil { + if _, err := tx.NamedExecContext(ctx, insertEventSQL, map[string]any{ + "eventDefId": ref.GetId(), + "time": event.GetTime().GetUnix().AsTime(), + "timeSclk": event.GetTime().GetSclk(), + "message": event.GetMessage(), + "source": msg.GetSource(), + "args": string(eventArgs), + "ert": stdtime.Now(), + }); err != nil { return fmt.Errorf("failed to insert event: %w", err) } return tx.Commit() } -func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry) error { +func InsertTelemetry(ctx context.Context, db *sqlx.DB, msg *pb.SourcedTelemetry) error { tlm := msg.GetTelemetry() def := tlm.GetRef() @@ -84,15 +93,17 @@ func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry) return fmt.Errorf("failed to marshal telemetry labels: %w", err) } - tx, err := db.BeginTx(ctx, nil) + tx, err := db.BeginTxx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback() - if _, err := tx.ExecContext(ctx, insertTelemetryDefSQL, - def.GetId(), def.GetName(), def.GetComponent(), - ); err != nil { + if _, err := tx.NamedExecContext(ctx, insertTelemetryDefSQL, map[string]any{ + "id": def.GetId(), + "name": def.GetName(), + "component": def.GetComponent(), + }); err != nil { return fmt.Errorf("failed to insert telemetry def: %w", err) } @@ -103,7 +114,7 @@ func InsertTelemetry(ctx context.Context, db *sql.DB, msg *pb.SourcedTelemetry) return tx.Commit() } -func insertValue(ctx context.Context, tx *sql.Tx, time *pb.Time, telemetryDefId int32, source string, labels string, path string, value *pb.Value) error { +func insertValue(ctx context.Context, tx *sqlx.Tx, time *pb.Time, telemetryDefId int32, source string, labels string, path string, value *pb.Value) error { var ( valueType string integral, floating any @@ -151,11 +162,20 @@ func insertValue(ctx context.Context, tx *sql.Tx, time *pb.Time, telemetryDefId bytes = valueTy.R.Value } - now := stdtime.Now() - _, err := tx.ExecContext(ctx, insertTelemetrySQL, - time.GetUnix().AsTime(), telemetryDefId, time.GetSclk(), - source, labels, path, valueType, - integral, floating, boolval, str, bytes, now, - ) + _, err := tx.NamedExecContext(ctx, insertTelemetrySQL, map[string]any{ + "time": time.GetUnix().AsTime(), + "telemetryDefId": telemetryDefId, + "timeSclk": time.GetSclk(), + "source": source, + "labels": labels, + "key": path, + "valueType": valueType, + "integral": integral, + "floating": floating, + "boolval": boolval, + "string": str, + "bytes": bytes, + "ert": stdtime.Now(), + }) return err } diff --git a/pkg/timescaledb/timescaledb.go b/pkg/timescaledb/timescaledb.go index bec0d5b..de60348 100644 --- a/pkg/timescaledb/timescaledb.go +++ b/pkg/timescaledb/timescaledb.go @@ -6,8 +6,7 @@ import ( _ "embed" - "database/sql" - + "github.com/jmoiron/sqlx" _ "github.com/lib/pq" "github.com/nasa/hermes/pkg/host" "github.com/nasa/hermes/pkg/pb" @@ -54,7 +53,7 @@ func (t *timescaleDbProvider) Start( dsn, ) - db, err := sql.Open("postgres", dsn) + db, err := sqlx.Open("postgres", dsn) if err != nil { return fmt.Errorf("failed to open timescaledb connection: %w", err) }