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

Storage: Add support for listing resource history (#84331)

Browse Source 
* add support for listing resource history

* make watch handle custom label selectors properly

* fix tests

* Apply suggestions from code review

Co-authored-by: Diego Augusto Molina <diegoaugustomolina@gmail.com>

* properly handle special characters in json label matcher

* tidy up

---------

Co-authored-by: Diego Augusto Molina <diegoaugustomolina@gmail.com>
This commit is contained in:
Dan Cech
2024-03-15 19:17:54 -04:00
committed by GitHub
co-authored by Diego Augusto Molina
parent 9c7a5ed506
commit 89f3b70e17
10 changed files with 530 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/store/entity/sqlstash/querybuilder.go
+21
View File
@@ -1,6 +1,7 @@
package sqlstash
import (
"encoding/json"
"strings"
"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
@@ -71,6 +72,26 @@ func (q *selectQuery) addWhereIn(f string, vals []string) {
}
}
const sqlLikeEscape = "#"
var sqlLikeEscapeReplacer = strings.NewReplacer(
sqlLikeEscape, sqlLikeEscape+sqlLikeEscape,
"%", sqlLikeEscape+"%",
"_", sqlLikeEscape+"_",
)
func escapeJSONStringSQLLike(s string) string {
b, _ := json.Marshal(s)
return sqlLikeEscapeReplacer.Replace(string(b))
}
func (q *selectQuery) addWhereJsonContainsKV(field string, key string, value string) {
escapedKey := escapeJSONStringSQLLike(key)
escapedValue := escapeJSONStringSQLLike(value)
q.where = append(q.where, q.dialect.Quote(field)+" LIKE ? ESCAPE ?")
q.args = append(q.args, "{%"+escapedKey+":"+escapedValue+"%}", sqlLikeEscape)
}
func (q *selectQuery) addOrderBy(field string, direction Direction) {
q.orderBy = append(q.orderBy, field)
q.direction = append(q.direction, direction)
pkg/services/store/entity/sqlstash/sql_storage_server.go
+122 -75
View File
@@ -95,7 +95,12 @@ func (s *sqlEntityServer) Init() error {
return nil
}
func (s *sqlEntityServer) getReadFields(r *entity.ReadEntityRequest) []string {
type FieldSelectRequest interface {
GetWithBody() bool
GetWithStatus() bool
}
func (s *sqlEntityServer) getReadFields(r FieldSelectRequest) []string {
fields := []string{
"guid",
"key",
@@ -110,17 +115,17 @@ func (s *sqlEntityServer) getReadFields(r *entity.ReadEntityRequest) []string {
"action",
}
if r.WithBody {
if r.GetWithBody() {
fields = append(fields, `body`)
}
if r.WithStatus {
if r.GetWithStatus() {
fields = append(fields, "status")
}
return fields
}
func (s *sqlEntityServer) getReadSelect(r *entity.ReadEntityRequest) (string, error) {
func (s *sqlEntityServer) getReadSelect(r FieldSelectRequest) (string, error) {
if err := s.Init(); err != nil {
return "", err
}
@@ -134,7 +139,7 @@ func (s *sqlEntityServer) getReadSelect(r *entity.ReadEntityRequest) (string, er
return "SELECT " + strings.Join(quotedFields, ","), nil
}
func rowToEntity(rows *sql.Rows, r *entity.ReadEntityRequest) (*entity.Entity, error) {
func rowToEntity(rows *sql.Rows, r FieldSelectRequest) (*entity.Entity, error) {
raw := &entity.Entity{
Origin: &entity.EntityOriginInfo{},
}
@@ -156,10 +161,10 @@ func rowToEntity(rows *sql.Rows, r *entity.ReadEntityRequest) (*entity.Entity, e
&raw.Message,
&raw.Action,
}
if r.WithBody {
if r.GetWithBody() {
args = append(args, &raw.Body)
}
if r.WithStatus {
if r.GetWithStatus() {
args = append(args, &raw.Status)
}
@@ -820,6 +825,17 @@ func (s *sqlEntityServer) doDelete(ctx context.Context, tx *session.SessionTx, e
// Update resource version
ent.ResourceVersion = s.snowflake.Generate().Int64()
// Set updated at/by
ent.UpdatedAt = time.Now().UnixMilli()
modifier, err := appcontext.User(ctx)
if err != nil {
return err
}
if modifier == nil {
return fmt.Errorf("can not find user in context")
}
ent.UpdatedBy = store.GetUserIDString(modifier)
labels, err := json.Marshal(ent.Labels)
if err != nil {
s.log.Error("error marshalling labels", "msg", err.Error())
@@ -911,21 +927,13 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
return nil, err
}
var limit int64 = 100
if r.Limit > 0 && r.Limit < 100 {
limit = r.Limit
}
rr := &entity.ReadEntityRequest{
Key: r.Key,
WithBody: true,
WithStatus: true,
}
query, err := s.getReadSelect(rr)
user, err := appcontext.User(ctx)
if err != nil {
return nil, err
}
if user == nil {
return nil, fmt.Errorf("missing user in context")
}
if r.Key == "" {
return nil, fmt.Errorf("missing key")
@@ -936,25 +944,59 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
return nil, err
}
where := []string{}
args := []any{}
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.NextPageToken != "" {
if true {
return nil, fmt.Errorf("tokens not yet supported")
}
where = append(where, "version <= ?")
args = append(args, r.NextPageToken)
if key.Name == "" {
return nil, fmt.Errorf("missing name")
}
query += " FROM entity_history" +
" WHERE " + strings.Join(where, " AND ") +
" ORDER BY resource_version DESC" +
// select 1 more than we need to see if there is a next page
" LIMIT " + fmt.Sprint(limit+1)
var limit int64 = 100
if r.Limit > 0 && r.Limit < 100 {
limit = r.Limit
}
fields := s.getReadFields(r)
entityQuery := selectQuery{
dialect: s.dialect,
fields: fields,
from: "entity_history", // the table
args: []any{},
limit: r.Limit,
offset: 0,
oneExtra: true, // request one more than the limit (and show next token if it exists)
}
args := []any{key.Group, key.Resource}
whereclause := "(" + s.dialect.Quote("group") + "=? AND " + s.dialect.Quote("resource") + "=?"
if key.Namespace != "" {
args = append(args, key.Namespace)
whereclause += " AND " + s.dialect.Quote("namespace") + "=?"
}
args = append(args, key.Name)
whereclause += " AND " + s.dialect.Quote("name") + "=?)"
entityQuery.addWhere(whereclause, args...)
// if we have a page token, use that to specify the first record
continueToken, err := GetContinueToken(r)
if err != nil {
return nil, err
}
if continueToken != nil {
entityQuery.offset = continueToken.StartOffset
}
for _, sort := range r.Sort {
sortBy, err := ParseSortBy(sort)
if err != nil {
return nil, err
}
entityQuery.addOrderBy(sortBy.Field, sortBy.Direction)
}
entityQuery.addOrderBy("resource_version", Ascending)
query, args := entityQuery.toQuery()
s.log.Debug("history", "query", query, "args", args)
rows, err := s.sess.Query(ctx, query, args...)
if err != nil {
@@ -963,17 +1005,22 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
defer func() { _ = rows.Close() }()
rsp := &entity.EntityHistoryResponse{
Key: r.Key,
Key: r.Key,
ResourceVersion: s.snowflake.Generate().Int64(),
}
for rows.Next() {
v, err := rowToEntity(rows, rr)
v, err := rowToEntity(rows, r)
if err != nil {
return nil, err
}
// found more than requested
if int64(len(rsp.Versions)) >= limit {
rsp.NextPageToken = fmt.Sprintf("rv:%d", v.ResourceVersion)
continueToken := &ContinueToken{
Sort: r.Sort,
StartOffset: entityQuery.offset + entityQuery.limit,
}
rsp.NextPageToken = continueToken.String()
break
}
@@ -982,6 +1029,11 @@ func (s *sqlEntityServer) History(ctx context.Context, r *entity.EntityHistoryRe
return rsp, err
}
type ContinueRequest interface {
GetNextPageToken() string
GetSort() []string
}
type ContinueToken struct {
Sort []string `json:"s"`
StartOffset int64 `json:"o"`
@@ -992,12 +1044,12 @@ func (c *ContinueToken) String() string {
return base64.StdEncoding.EncodeToString(b)
}
func GetContinueToken(r *entity.EntityListRequest) (*ContinueToken, error) {
if r.NextPageToken == "" {
func GetContinueToken(r ContinueRequest) (*ContinueToken, error) {
if r.GetNextPageToken() == "" {
return nil, nil
}
continueVal, err := base64.StdEncoding.DecodeString(r.NextPageToken)
continueVal, err := base64.StdEncoding.DecodeString(r.GetNextPageToken())
if err != nil {
return nil, fmt.Errorf("error decoding continue token")
}
@@ -1008,7 +1060,7 @@ func GetContinueToken(r *entity.EntityListRequest) (*ContinueToken, error) {
return nil, err
}
if !slices.Equal(t.Sort, r.Sort) {
if !slices.Equal(t.Sort, r.GetSort()) {
return nil, fmt.Errorf("sort order changed")
}
@@ -1064,12 +1116,7 @@ func (s *sqlEntityServer) List(ctx context.Context, r *entity.EntityListRequest)
return nil, fmt.Errorf("missing user in context")
}
rr := &entity.ReadEntityRequest{
WithBody: r.WithBody,
WithStatus: r.WithStatus,
}
fields := s.getReadFields(rr)
fields := s.getReadFields(r)
entityQuery := selectQuery{
dialect: s.dialect,
@@ -1143,7 +1190,7 @@ func (s *sqlEntityServer) List(ctx context.Context, r *entity.EntityListRequest)
// if we are looking for deleted entities, we need to use the labels column
if r.Deleted {
for labelKey, labelValue := range r.Labels {
entityQuery.addWhere(s.dialect.Quote("labels")+" LIKE ?", "%\""+labelKey+"\":\""+labelValue+"\"%")
entityQuery.addWhereJsonContainsKV("labels", labelKey, labelValue)
}
// for active entities, we can use the entity_labels table
} else {
@@ -1163,6 +1210,7 @@ func (s *sqlEntityServer) List(ctx context.Context, r *entity.EntityListRequest)
entityQuery.addWhereInSubquery("guid", query, args)
}
}
for _, sort := range r.Sort {
sortBy, err := ParseSortBy(sort)
if err != nil {
@@ -1185,7 +1233,7 @@ func (s *sqlEntityServer) List(ctx context.Context, r *entity.EntityListRequest)
ResourceVersion: s.snowflake.Generate().Int64(),
}
for rows.Next() {
result, err := rowToEntity(rows, rr)
result, err := rowToEntity(rows, r)
if err != nil {
return rsp, err
}
@@ -1237,12 +1285,7 @@ func (s *sqlEntityServer) Watch(r *entity.EntityWatchRequest, w entity.EntitySto
// watchInit is a helper function to send the initial set of entities to the client
func (s *sqlEntityServer) watchInit(r *entity.EntityWatchRequest, w entity.EntityStore_WatchServer) error {
rr := &entity.ReadEntityRequest{
WithBody: r.WithBody,
WithStatus: r.WithStatus,
}
fields := s.getReadFields(rr)
fields := s.getReadFields(r)
entityQuery := selectQuery{
dialect: s.dialect,
@@ -1301,32 +1344,38 @@ func (s *sqlEntityServer) watchInit(r *entity.EntityWatchRequest, w entity.Entit
}
if len(r.Labels) > 0 {
var args []any
var conditions []string
for labelKey, labelValue := range r.Labels {
args = append(args, labelKey)
args = append(args, labelValue)
conditions = append(conditions, "(label = ? AND value = ?)")
}
query := "SELECT guid FROM entity_labels" +
" WHERE (" + strings.Join(conditions, " OR ") + ")" +
" GROUP BY guid" +
" HAVING COUNT(label) = ?"
args = append(args, len(r.Labels))
if r.Since > 0 {
for labelKey, labelValue := range r.Labels {
entityQuery.addWhereJsonContainsKV("labels", labelKey, labelValue)
}
} else {
var args []any
var conditions []string
for labelKey, labelValue := range r.Labels {
args = append(args, labelKey)
args = append(args, labelValue)
conditions = append(conditions, "(label = ? AND value = ?)")
}
query := "SELECT guid FROM entity_labels" +
" WHERE (" + strings.Join(conditions, " OR ") + ")" +
" GROUP BY guid" +
" HAVING COUNT(label) = ?"
args = append(args, len(r.Labels))
entityQuery.addWhereInSubquery("guid", query, args)
entityQuery.addWhereInSubquery("guid", query, args)
}
}
entityQuery.addOrderBy("resource_version", Ascending)
var err error
s.log.Debug("watch init", "since", r.Since)
for hasmore := true; hasmore; {
err = func() error {
query, args := entityQuery.toQuery()
s.log.Debug("watch init", "query", query, "args", args)
rows, err := s.sess.Query(w.Context(), query, args...)
if err != nil {
return err
@@ -1342,7 +1391,7 @@ func (s *sqlEntityServer) watchInit(r *entity.EntityWatchRequest, w entity.Entit
return nil
}
result, err := rowToEntity(rows, rr)
result, err := rowToEntity(rows, r)
if err != nil {
return err
}
@@ -1392,8 +1441,6 @@ func (s *sqlEntityServer) poller(stream chan *entity.Entity) {
}
func (s *sqlEntityServer) poll(since int64, out chan *entity.Entity) (int64, error) {
s.log.Debug("watch poll", "since", since)
rr := &entity.ReadEntityRequest{
WithBody: true,
WithStatus: true,