Skip to content

feat: add instrumentation in transactions #489

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vasconcelosvcd wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat: add instrumentation in transactions #489

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0f5578d
feat: initial tx instrumentation
vasconcelosvcd May 6, 2026
8048313
feat: add configurable threshold
vasconcelosvcd May 6, 2026
aebdac0
chore: rollback api change
vasconcelosvcd May 6, 2026
fd091c9
Merge branch 'main' into feat/add-instrumentation-in-trasactions
vasconcelosvcd Oct 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions configs/service.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,3 +36,4 @@ trace_sample_rate: 0.00

datastore_tx_max_attempts: 2
datastore_database_id: ""
datastore_tx_slow_threshold: "1s"
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ require (
go.opencensus.io v0.24.0 // indirect
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.26.0 // indirect
go.opentelemetry.io/otel/metric v1.29.0 // indirect
go.opentelemetry.io/otel/metric v1.29.0
go.opentelemetry.io/otel/trace v1.29.0 // indirect
go.opentelemetry.io/proto/otlp v1.2.0 // indirect
golang.org/x/crypto v0.36.0 // indirect
Expand Down
5 changes: 3 additions & 2 deletions internal/pkg/config/loader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,8 +151,9 @@ func Load(path string) (*ServiceConfig, error) {
}

datastoreConfig := DatastoreConfig{
TXMaxAttempts: viper.GetInt(DatastoreTXMaxAttempts),
DatabaseId: viper.GetString(DatastoreDatabaseId),
TXMaxAttempts: viper.GetInt(DatastoreTXMaxAttempts),
DatabaseId: viper.GetString(DatastoreDatabaseId),
SlowTXThreshold: viper.GetDuration(DatastoreTXSlowThreshold),
}

return &ServiceConfig{
Expand Down
10 changes: 6 additions & 4 deletions internal/pkg/config/model.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,9 @@ const (
TraceEnableGRPCCollector = "trace_enable_grpc_collector"
TraceEnableHTTPCollector = "trace_enable_http_collector"

DatastoreTXMaxAttempts = "datastore_tx_max_attempts"
DatastoreDatabaseId = "datastore_database_id"
DatastoreTXMaxAttempts = "datastore_tx_max_attempts"
DatastoreDatabaseId = "datastore_database_id"
DatastoreTXSlowThreshold = "datastore_tx_slow_threshold"
)

type ServiceConfig struct {
Expand Down Expand Up @@ -131,6 +132,7 @@ type GRPCServerConfig struct {
}

type DatastoreConfig struct {
TXMaxAttempts int
DatabaseId string
TXMaxAttempts int
DatabaseId string
SlowTXThreshold time.Duration
}
82 changes: 65 additions & 17 deletions internal/pkg/metadb/metadb.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,14 @@ import (
"errors"
"fmt"
"strconv"
"time"

"github.com/googleforgames/open-saves/internal/pkg/config"
"github.com/googleforgames/open-saves/internal/pkg/tracing"
log "github.com/sirupsen/logrus"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
otelmetric "go.opentelemetry.io/otel/metric"

"cloud.google.com/go/datastore"
ds "cloud.google.com/go/datastore"
Expand All @@ -49,6 +53,8 @@ const (
propertiesField = "Properties"
tagsField = "Tags"
ownerField = "OwnerID"

defaultSlowTXThreshold = time.Second
)

var ErrNoUpdate = errors.New("UpdateRecord doesn't need to commit the change")
Expand All @@ -68,6 +74,9 @@ type MetaDB struct {
client *ds.Client

config config.DatastoreConfig

txDuration otelmetric.Float64Histogram
txErrors otelmetric.Int64Counter
}

// RecordUpdater is a callback function for record updates.
Expand All @@ -89,7 +98,46 @@ func NewMetaDB(ctx context.Context, projectID string, config config.DatastoreCon
if err != nil {
return nil, datastoreErrToGRPCStatus(err)
}
return &MetaDB{client: client, config: config}, nil
meter := otel.Meter("open-saves/metadb")
txDuration, _ := meter.Float64Histogram(
"metadb.transaction.duration",
otelmetric.WithDescription("Duration of MetaDB Datastore transactions in seconds"),
otelmetric.WithUnit("s"),
)
txErrors, _ := meter.Int64Counter(
"metadb.transaction.errors",
otelmetric.WithDescription("Number of failed MetaDB Datastore transactions"),
)
if config.SlowTXThreshold == 0 {
config.SlowTXThreshold = defaultSlowTXThreshold
}
return &MetaDB{client: client, config: config, txDuration: txDuration, txErrors: txErrors}, nil
}

// runInTransaction wraps RunInTransaction with duration metrics and slow-operation warnings.
// storeKey and recordKey are included in log output but not in metric attributes to avoid high cardinality.
func (m *MetaDB) runInTransaction(ctx context.Context, operation, storeKey, recordKey string, f func(*ds.Transaction) error, opts ...ds.TransactionOption) (*ds.Commit, error) {
start := time.Now()
commit, err := m.client.RunInTransaction(ctx, f, opts...)
elapsed := time.Since(start)

attrs := []attribute.KeyValue{attribute.String("operation", operation)}
m.txDuration.Record(ctx, elapsed.Seconds(), otelmetric.WithAttributes(attrs...))
if err != nil {
m.txErrors.Add(ctx, 1, otelmetric.WithAttributes(attrs...))
}

if elapsed > m.config.SlowTXThreshold {
log.WithFields(log.Fields{
"operation": operation,
"store_key": storeKey,
"record_key": recordKey,
"duration_ms": elapsed.Milliseconds(),
"error": err,
}).Warn("slow MetaDB transaction")
}

return commit, err
}

func (m *MetaDB) newQuery(kind string) *ds.Query {
Expand Down Expand Up @@ -263,7 +311,7 @@ func (m *MetaDB) DeleteStore(ctx context.Context, key string) error {

dskey := m.createStoreKey(key)

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "DeleteStore", key, "", func(tx *ds.Transaction) error {
query := ds.NewQuery(recordKind).Transaction(tx).KeysOnly().
Ancestor(dskey).Limit(1).Namespace(m.Namespace)
iter := m.client.Run(ctx, query)
Expand All @@ -289,7 +337,7 @@ func (m *MetaDB) InsertRecord(ctx context.Context, storeKey string, record *reco
record.Timestamps = timestamps.New()
record.StoreKey = storeKey
rkey := m.createRecordKey(storeKey, record.Key)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "InsertRecord", storeKey, record.Key, func(tx *ds.Transaction) error {
dskey := m.createStoreKey(storeKey)
query := ds.NewQuery(storeKind).Transaction(tx).Namespace(m.Namespace).
KeysOnly().FilterField("__key__", "=", dskey).Limit(1)
Expand Down Expand Up @@ -320,7 +368,7 @@ func (m *MetaDB) UpdateRecord(ctx context.Context, storeKey string, key string,
return nil, status.Errorf(codes.Internal, "updater cannot be nil")
}
var toUpdate *record.Record
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "UpdateRecord", storeKey, key, func(tx *ds.Transaction) error {
rkey := m.createRecordKey(storeKey, key)

// TODO(yuryu): Consider supporting transactions in MetaDB and move
Expand Down Expand Up @@ -395,7 +443,7 @@ func (m *MetaDB) DeleteRecord(ctx context.Context, storeKey, key string) error {
defer span.End()

rkey := m.createRecordKey(storeKey, key)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "DeleteRecord", storeKey, key, func(tx *ds.Transaction) error {
record := new(record.Record)
if err := tx.Get(rkey, record); err != nil {
if err == ds.ErrNoSuchEntity {
Expand Down Expand Up @@ -427,7 +475,7 @@ func (m *MetaDB) InsertBlobRef(ctx context.Context, blob *blobref.BlobRef) (*blo

blob.Timestamps = timestamps.New()
rkey := m.createRecordKey(blob.StoreKey, blob.RecordKey)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "InsertBlobRef", blob.StoreKey, blob.RecordKey, func(tx *ds.Transaction) error {
if exists, err := m.recordExists(ctx, tx, rkey); err != nil {
return err
} else if !exists {
Expand All @@ -448,7 +496,7 @@ func (m *MetaDB) UpdateBlobRef(ctx context.Context, blob *blobref.BlobRef) (*blo
_, span := otel.Tracer(tracing.ServiceName).Start(ctx, "MetaDB.UpdateBlobRef")
defer span.End()

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "UpdateBlobRef", blob.Key.String(), "", func(tx *ds.Transaction) error {
oldBlob, err := m.getBlobRef(ctx, tx, blob.Key)
if err != nil {
return err
Expand Down Expand Up @@ -493,7 +541,7 @@ func (m *MetaDB) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey stri
defer span.End()

var blob *blobref.BlobRef
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "GetCurrentBlobRef", storeKey, recordKey, func(tx *ds.Transaction) error {
var err error
blob, err = m.getCurrentBlobRef(ctx, tx, storeKey, recordKey)
return err
Expand Down Expand Up @@ -541,7 +589,7 @@ func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *blobref.Blob
defer span.End()

record := new(record.Record)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "PromoteBlobRefToCurrent", blob.StoreKey, blob.RecordKey, func(tx *ds.Transaction) error {
rkey := m.createRecordKey(blob.StoreKey, blob.RecordKey)
if err := tx.Get(rkey, record); err != nil {
return err
Expand Down Expand Up @@ -613,7 +661,7 @@ func (m *MetaDB) PromoteBlobRefWithRecordUpdater(ctx context.Context, blob *blob
defer span.End()

record := new(record.Record)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "PromoteBlobRefWithRecordUpdater", blob.StoreKey, blob.RecordKey, func(tx *ds.Transaction) error {
rkey := m.createRecordKey(blob.StoreKey, blob.RecordKey)
if err := tx.Get(rkey, record); err != nil {
return err
Expand Down Expand Up @@ -697,7 +745,7 @@ func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, reco
rkey := m.createRecordKey(storeKey, recordKey)
blob := new(blobref.BlobRef)
record := new(record.Record)
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "RemoveBlobFromRecord", storeKey, recordKey, func(tx *ds.Transaction) error {
err := tx.Get(rkey, record)
if err != nil {
return err
Expand Down Expand Up @@ -756,7 +804,7 @@ func (m *MetaDB) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
_, span := otel.Tracer(tracing.ServiceName).Start(ctx, "MetaDB.DeleteBlobRef")
defer span.End()

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "DeleteBlobRef", key.String(), "", func(tx *ds.Transaction) error {
blob, err := m.getBlobRef(ctx, tx, key)
if err != nil {
return err
Expand All @@ -781,7 +829,7 @@ func (m *MetaDB) DeleteChunkRef(ctx context.Context, blobKey, key uuid.UUID) err
_, span := otel.Tracer(tracing.ServiceName).Start(ctx, "MetaDB.DeleteChunkRef")
defer span.End()

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "DeleteChunkRef", blobKey.String(), key.String(), func(tx *ds.Transaction) error {
var chunk chunkref.ChunkRef
if err := tx.Get(m.createChunkRefKey(blobKey, key), &chunk); err != nil {
return err
Expand Down Expand Up @@ -1031,7 +1079,7 @@ func (m *MetaDB) FindBlobChunkRefsByNumber(ctx context.Context, blobKey uuid.UUI
defer span.End()

var chunks []*chunkref.ChunkRef
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "FindBlobChunkRefsByNumber", blobKey.String(), "", func(tx *ds.Transaction) error {
foundChunks, err := m.findChunkRefsByNumber(ctx, tx, blobKey, number)
chunks = foundChunks
return err
Expand All @@ -1050,7 +1098,7 @@ func (m *MetaDB) FindChunkRefByNumber(ctx context.Context, storeKey, recordKey s
defer span.End()

chunks := []*chunkref.ChunkRef{}
_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "FindChunkRefByNumber", storeKey, recordKey, func(tx *ds.Transaction) error {
blob, err := m.getCurrentBlobRef(ctx, tx, storeKey, recordKey)
if err != nil {
return err
Expand Down Expand Up @@ -1093,7 +1141,7 @@ func (m *MetaDB) InsertChunkRef(ctx context.Context, blob *blobref.BlobRef, chun
_, span := otel.Tracer(tracing.ServiceName).Start(ctx, "MetaDB.InsertChunkRef")
defer span.End()

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "InsertChunkRef", blob.StoreKey, blob.RecordKey, func(tx *ds.Transaction) error {
mut := ds.NewInsert(m.createChunkRefKey(chunk.BlobRef, chunk.Key), chunk)
if err := m.mutateSingleInTransaction(tx, mut); err != nil {
return err
Expand All @@ -1111,7 +1159,7 @@ func (m *MetaDB) MarkUncommittedBlobForDeletion(ctx context.Context, key uuid.UU
_, span := otel.Tracer(tracing.ServiceName).Start(ctx, "MetaDB.MarkUncommittedBlobForDeletion")
defer span.End()

_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
_, err := m.runInTransaction(ctx, "MarkUncommittedBlobForDeletion", key.String(), "", func(tx *ds.Transaction) error {
blob, err := m.getBlobRef(ctx, tx, key)
if err != nil {
return err
Expand Down
Loading