public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

spanner: Fix insert or update in anon_device when running on Spanner. (#102991)

Browse Source 
Fix insert or update in anon_device when running on Spanner.
This commit is contained in:
Peter Štibraný
2025-03-27 16:58:27 +01:00
committed by GitHub
parent 157edf92a0
commit 57cc4f101d
1 changed files with 44 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/anonymous/anonimpl/anonstore/database.go
+44 -20
View File
@@ -15,6 +15,7 @@ import (
const cacheKeyPrefix = "anon-device"
const anonymousDeviceExpiration = 30 * 24 * time.Hour
const tableName = "anon_device"
var ErrDeviceLimitReached = fmt.Errorf("device limit reached")
@@ -25,7 +26,7 @@ type AnonDBStore struct {
}
type Device struct {
ID int64 `json:"-" xorm:"id" db:"id"`
ID int64 `json:"-" xorm:"pk autoincr 'id'" db:"id"`
DeviceID string `json:"deviceId" xorm:"device_id" db:"device_id"`
ClientIP string `json:"clientIp" xorm:"client_ip" db:"client_ip"`
UserAgent string `json:"userAgent" xorm:"user_agent" db:"user_agent"`
@@ -144,31 +145,34 @@ func (s *AnonDBStore) CreateOrUpdateDevice(ctx context.Context, device *Device)
}
}
args := []any{device.DeviceID, device.ClientIP, device.UserAgent,
device.CreatedAt.UTC(), device.UpdatedAt.UTC()}
if s.sqlStore.GetDBType() == migrator.Spanner {
return s.insertIntoSpanner(ctx, device)
}
args := []any{device.DeviceID, device.ClientIP, device.UserAgent, device.CreatedAt.UTC(), device.UpdatedAt.UTC()}
switch s.sqlStore.GetDBType() {
case migrator.Postgres:
query = `INSERT INTO anon_device (device_id, client_ip, user_agent, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (device_id) DO UPDATE SET
client_ip = $2,
user_agent = $3,
updated_at = $5
RETURNING id`
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (device_id) DO UPDATE SET
client_ip = $2,
user_agent = $3,
updated_at = $5
RETURNING id`
case migrator.MySQL:
query = `INSERT INTO anon_device (device_id, client_ip, user_agent, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
client_ip = VALUES(client_ip),
user_agent = VALUES(user_agent),
updated_at = VALUES(updated_at)`
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
client_ip = VALUES(client_ip),
user_agent = VALUES(user_agent),
updated_at = VALUES(updated_at)`
case migrator.SQLite:
query = `INSERT INTO anon_device (device_id, client_ip, user_agent, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (device_id) DO UPDATE SET
client_ip = excluded.client_ip,
user_agent = excluded.user_agent,
updated_at = excluded.updated_at`
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (device_id) DO UPDATE SET
client_ip = excluded.client_ip,
user_agent = excluded.user_agent,
updated_at = excluded.updated_at`
default:
return fmt.Errorf("unsupported database driver: %s", s.sqlStore.GetDBType())
}
@@ -182,6 +186,26 @@ updated_at = excluded.updated_at`
return err
}
// In Spanner INSERT OR UPDATE only works when conflict is on primary key. However here we expect conflict on non-PK
// column "device_id", so we need to use a transaction instead.
func (s *AnonDBStore) insertIntoSpanner(ctx context.Context, dev *Device) error {
return s.sqlStore.WithTransactionalDbSession(ctx, func(dbSession *sqlstore.DBSession) error {
prev := &Device{DeviceID: dev.DeviceID}
ok, err := dbSession.Table(tableName).Get(prev)
if err != nil {
return err
}
if !ok {
_, err = dbSession.Table(tableName).Insert(dev)
return err
}
// Include all columns in the update, even when empty (it's what inserts for other databases do too).
_, err = dbSession.Table(tableName).Where("device_id=?", dev.DeviceID).MustCols("client_ip", "user_agent", "created_at", "updated_at").Update(dev)
return err
})
}
func (s *AnonDBStore) CountDevices(ctx context.Context, from time.Time, to time.Time) (int64, error) {
var count int64
err := s.sqlStore.WithDbSession(ctx, func(dbSession *sqlstore.DBSession) error {
@@ -229,7 +253,7 @@ func (s *AnonDBStore) SearchDevices(ctx context.Context, query *SearchDeviceQuer
query.To = time.Now()
}
sess := dbSess.Table("anon_device").Alias("d")
sess := dbSess.Table(tableName).Alias("d")
if query.Limit > 0 {
offset := query.Limit * (query.Page - 1)