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

Storage: Use int64 for resourceVersion (#79735)

Browse Source
This commit is contained in:
Ryan McKinley
2023-12-20 11:26:35 -08:00
committed by GitHub
parent 67bbdd7c05
commit d195c3807e
8 changed files with 489 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/store/entity/sqlstash/sql_storage_server.go
+65 -62
View File
@@ -92,7 +92,7 @@ func (s *sqlEntityServer) getReadFields(r *entity.ReadEntityRequest) []string {
"guid",
"key",
"namespace", "group", "group_version", "resource", "name", "folder",
"version", "size", "etag", "errors", // errors are always returned
"resource_version", "size", "etag", "errors", // errors are always returned
"created_at", "created_by",
"updated_at", "updated_by",
"origin", "origin_key", "origin_ts",
@@ -137,7 +137,7 @@ func (s *sqlEntityServer) rowToEntity(ctx context.Context, rows *sql.Rows, r *en
&raw.Guid,
&raw.Key,
&raw.Namespace, &raw.Group, &raw.GroupVersion, &raw.Resource, &raw.Name, &raw.Folder,
&raw.Version, &raw.Size, &raw.ETag, &errors,
&raw.ResourceVersion, &raw.Size, &raw.ETag, &errors,
&raw.CreatedAt, &raw.CreatedBy,
&raw.UpdatedAt, &raw.UpdatedBy,
&raw.Origin.Source, &raw.Origin.Key, &raw.Origin.Time,
@@ -195,10 +195,10 @@ func (s *sqlEntityServer) read(ctx context.Context, tx session.SessionQuerier, r
where = append(where, s.dialect.Quote("namespace")+"=?", s.dialect.Quote("group")+"=?", s.dialect.Quote("resource")+"=?", s.dialect.Quote("name")+"=?")
args = append(args, key.Namespace, key.Group, key.Resource, key.Name)
if r.Version != "" {
if r.ResourceVersion != 0 {
table = "entity_history"
where = append(where, s.dialect.Quote("version")+"=?")
args = append(args, r.Version)
where = append(where, s.dialect.Quote("resource_version")+"=?")
args = append(args, r.ResourceVersion)
}
query, err := s.getReadSelect(r)
@@ -249,7 +249,7 @@ func (s *sqlEntityServer) BatchRead(ctx context.Context, b *entity.BatchReadEnti
constraints = append(constraints, s.dialect.Quote("key")+"=?")
args = append(args, r.Key)
if r.Version != "" {
if r.ResourceVersion != 0 {
return nil, fmt.Errorf("version not supported for batch read (yet?)")
}
}
@@ -409,37 +409,37 @@ func (s *sqlEntityServer) Create(ctx context.Context, r *entity.CreateEntityRequ
}
// Update version
current.Version = s.snowflake.Generate().String()
current.ResourceVersion = s.snowflake.Generate().Int64()
values := map[string]any{
"guid": current.Guid,
"key": current.Key,
"namespace": current.Namespace,
"group": current.Group,
"resource": current.Resource,
"name": current.Name,
"created_at": createdAt,
"created_by": createdBy,
"group_version": current.GroupVersion,
"folder": current.Folder,
"slug": current.Slug,
"updated_at": updatedAt,
"updated_by": updatedBy,
"body": current.Body,
"meta": current.Meta,
"status": current.Status,
"size": current.Size,
"etag": current.ETag,
"version": current.Version,
"title": current.Title,
"description": current.Description,
"labels": labels,
"fields": fields,
"errors": errors,
"origin": current.Origin.Source,
"origin_key": current.Origin.Key,
"origin_ts": current.Origin.Time,
"message": current.Message,
"guid": current.Guid,
"key": current.Key,
"namespace": current.Namespace,
"group": current.Group,
"resource": current.Resource,
"name": current.Name,
"created_at": createdAt,
"created_by": createdBy,
"group_version": current.GroupVersion,
"folder": current.Folder,
"slug": current.Slug,
"updated_at": updatedAt,
"updated_by": updatedBy,
"body": current.Body,
"meta": current.Meta,
"status": current.Status,
"size": current.Size,
"etag": current.ETag,
"resource_version": current.ResourceVersion,
"title": current.Title,
"description": current.Description,
"labels": labels,
"fields": fields,
"errors": errors,
"origin": current.Origin.Source,
"origin_key": current.Origin.Key,
"origin_ts": current.Origin.Time,
"message": current.Message,
}
// 1. Add row to the `entity_history` values
@@ -535,7 +535,7 @@ func (s *sqlEntityServer) Update(ctx context.Context, r *entity.UpdateEntityRequ
}
// Optimistic locking
if r.PreviousVersion != "" && r.PreviousVersion != current.Version {
if r.PreviousVersion > 0 && r.PreviousVersion != current.ResourceVersion {
return fmt.Errorf("optimistic lock failed")
}
@@ -630,7 +630,7 @@ func (s *sqlEntityServer) Update(ctx context.Context, r *entity.UpdateEntityRequ
}
// Update version
current.Version = s.snowflake.Generate().String()
current.ResourceVersion = s.snowflake.Generate().Int64()
values := map[string]any{
// below are only set in history table
@@ -643,26 +643,26 @@ func (s *sqlEntityServer) Update(ctx context.Context, r *entity.UpdateEntityRequ
"created_at": current.CreatedAt,
"created_by": current.CreatedBy,
// below are updated
"group_version": current.GroupVersion,
"folder": current.Folder,
"slug": current.Slug,
"updated_at": updatedAt,
"updated_by": updatedBy,
"body": current.Body,
"meta": current.Meta,
"status": current.Status,
"size": current.Size,
"etag": current.ETag,
"version": current.Version,
"title": current.Title,
"description": current.Description,
"labels": labels,
"fields": fields,
"errors": errors,
"origin": current.Origin.Source,
"origin_key": current.Origin.Key,
"origin_ts": current.Origin.Time,
"message": current.Message,
"group_version": current.GroupVersion,
"folder": current.Folder,
"slug": current.Slug,
"updated_at": updatedAt,
"updated_by": updatedBy,
"body": current.Body,
"meta": current.Meta,
"status": current.Status,
"size": current.Size,
"etag": current.ETag,
"resource_version": current.ResourceVersion,
"title": current.Title,
"description": current.Description,
"labels": labels,
"fields": fields,
"errors": errors,
"origin": current.Origin.Source,
"origin_key": current.Origin.Key,
"origin_ts": current.Origin.Time,
"message": current.Message,
}
// 1. Add the `entity_history` values
@@ -788,7 +788,7 @@ func (s *sqlEntityServer) Delete(ctx context.Context, r *entity.DeleteEntityRequ
return err
}
if r.PreviousVersion != "" && r.PreviousVersion != rsp.Entity.Version {
if r.PreviousVersion > 0 && r.PreviousVersion != rsp.Entity.ResourceVersion {
rsp.Status = entity.DeleteEntityResponse_ERROR
return fmt.Errorf("optimistic lock failed")
}
@@ -878,13 +878,16 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
args = append(args, key.Namespace, key.Group, key.Resource, key.Name)
if r.NextPageToken != "" {
if true {
return nil, fmt.Errorf("tokens not yet supported")
}
where = append(where, "version <= ?")
args = append(args, r.NextPageToken)
}
query += " FROM entity_history" +
" WHERE " + strings.Join(where, " AND ") +
" ORDER BY version DESC" +
" ORDER BY resource_version DESC" +
// select 1 more than we need to see if there is a next page
" LIMIT " + fmt.Sprint(limit+1)
@@ -905,7 +908,7 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
// found more than requested
if int64(len(rsp.Versions)) >= limit {
rsp.NextPageToken = v.Version
rsp.NextPageToken = fmt.Sprintf("rv:%d", v.ResourceVersion)
break
}
@@ -1064,7 +1067,7 @@ func (s *sqlEntityServer) FindReferences(ctx context.Context, r *entity.Referenc
fields := []string{
"e.guid", "e.guid",
"e.namespace", "e.group", "e.group_version", "e.resource", "e.name",
"e.version", "e.folder", "e.slug", "e.errors", // errors are always returned
"e.resource_version", "e.folder", "e.slug", "e.errors", // errors are always returned
"e.size", "e.updated_at", "e.updated_by",
"e.title", "e.description", "e.meta",
}
@@ -1086,7 +1089,7 @@ func (s *sqlEntityServer) FindReferences(ctx context.Context, r *entity.Referenc
args := []any{
&token, &result.Guid,
&result.Namespace, &result.Group, &result.GroupVersion, &result.Resource, &result.Name,
&result.Version, &result.Folder, &result.Slug, &result.Errors,
&result.ResourceVersion, &result.Folder, &result.Slug, &result.Errors,
&result.Size, &result.UpdatedAt, &result.UpdatedBy,
&result.Title, &result.Description, &result.Meta,
}