From d8abf8d851fec91ce896437a7ebe29fcbd89229f Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Fri, 14 Jun 2024 22:53:10 +0300 Subject: [PATCH] entity bridge --- go.work.sum | 4 + .../unified/entitybridge/entitybridge.go | 200 ++++++++++++++++++ pkg/storage/unified/resource/server.go | 9 + .../{sqlstash => sqlnext}/data/common.sql | 0 .../data/resource_get.sql | 0 .../data/resource_insert.sql | 0 .../{sqlstash => sqlnext}/data/rv_get.sql | 0 .../{sqlstash => sqlnext}/data/rv_inc.sql | 0 .../{sqlstash => sqlnext}/data/rv_insert.sql | 0 .../{sqlstash => sqlnext}/data/rv_lock.sql | 0 .../sql_resources.go} | 9 +- .../unified/{sqlstash => sqlnext}/watch.go | 2 +- pkg/storage/unified/sqlstash/utils.go | 15 -- 13 files changed, 215 insertions(+), 24 deletions(-) create mode 100644 pkg/storage/unified/entitybridge/entitybridge.go rename pkg/storage/unified/{sqlstash => sqlnext}/data/common.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/resource_get.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/resource_insert.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/rv_get.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/rv_inc.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/rv_insert.sql (100%) rename pkg/storage/unified/{sqlstash => sqlnext}/data/rv_lock.sql (100%) rename pkg/storage/unified/{sqlstash/sql_storage_server.go => sqlnext/sql_resources.go} (95%) rename pkg/storage/unified/{sqlstash => sqlnext}/watch.go (98%) delete mode 100644 pkg/storage/unified/sqlstash/utils.go diff --git a/go.work.sum b/go.work.sum index 111a6af928b..9983b4dab6f 100644 --- a/go.work.sum +++ b/go.work.sum @@ -941,6 +941,7 @@ github.com/grafana/grafana-plugin-sdk-go v0.228.0/go.mod h1:u4K9vVN6eU86loO68977 github.com/grafana/grafana-plugin-sdk-go v0.229.0/go.mod h1:6V6ikT4ryva8MrAp7Bdz5fTJx3/ztzKvpMJFfpzr4CI= github.com/grafana/grafana-plugin-sdk-go v0.231.1-0.20240523124942-62dae9836284/go.mod h1:bNgmNmub1I7Mc8dzIncgNqHC5jTgSZPPHlZ3aG8HKJQ= github.com/grafana/grafana-plugin-sdk-go v0.234.0/go.mod h1:FlXjmBESxaD6Hoi8ojWLkH007nyjtJM3XC8SpwzF/YE= +github.com/grafana/grafana/pkg/apimachinery v0.0.0-20240613114114-5e2f08de316d/go.mod h1:adT8O7k6ZSzUKjAC4WS6VfWlCE4G1VavPwSXVhvScCs= github.com/grafana/grafana/pkg/promlib v0.0.3/go.mod h1:3El4NlsfALz8QQCbEGHGFvJUG+538QLMuALRhZ3pcoo= github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 h1:pdN6V1QBWetyv/0+wjACpqVH+eVULgEjkurDLq3goeM= github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= @@ -949,6 +950,9 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.18.1/go.mod h1:YvJ2f6MplWDhfxiUC3Kp github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1/go.mod h1:5SN9VR2LTsRFsrEC6FHgRbTWrTHu6tqPeKxEQv15giM= github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645 h1:MJG/KsmcqMwFAkh8mTnAwhyKoB+sTAnY4CACC110tbU= github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645/go.mod h1:6iZfnjpejD4L/4DwD7NryNaJyCQdzwWwH2MWhCA90Kw= +github.com/hack-pad/go-indexeddb v0.3.2/go.mod h1:QvfTevpDVlkfomY498LhstjwbPW6QC4VC/lxYb0Kom0= +github.com/hack-pad/hackpadfs v0.2.1/go.mod h1:khQBuCEwGXWakkmq8ZiFUvUZz84ZkJ2KNwKvChs4OrU= +github.com/hack-pad/safejs v0.1.0/go.mod h1:HdS+bKF1NrE72VoXZeWzxFOVQVUSqZJAG0xNCnb+Tio= github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed h1:5upAirOpQc1Q53c0bnx2ufif5kANL7bfZWcc6VJWJd8= github.com/hamba/avro/v2 v2.17.2 h1:6PKpEWzJfNnvBgn7m2/8WYaDOUASxfDU+Jyb4ojDgFY= github.com/hamba/avro/v2 v2.17.2/go.mod h1:Q9YK+qxAhtVrNqOhwlZTATLgLA8qxG2vtvkhK8fJ7Jo= diff --git a/pkg/storage/unified/entitybridge/entitybridge.go b/pkg/storage/unified/entitybridge/entitybridge.go new file mode 100644 index 00000000000..520a192aad7 --- /dev/null +++ b/pkg/storage/unified/entitybridge/entitybridge.go @@ -0,0 +1,200 @@ +package sqlstash + +import ( + "context" + "fmt" + + "github.com/grafana/grafana/pkg/infra/tracing" + "github.com/grafana/grafana/pkg/services/store/entity" + "github.com/grafana/grafana/pkg/services/store/entity/db" + "github.com/grafana/grafana/pkg/services/store/entity/sqlstash" + "github.com/grafana/grafana/pkg/storage/unified/resource" +) + +// Creates a ResourceServer using the existing entity tables +// NOTE: most of the field values are ignored +func ProvideEnityStoreResources(db db.EntityDBInterface, tracer tracing.Tracer) (resource.ResourceServer, error) { + entity, err := sqlstash.ProvideSQLEntityServer(db, tracer) + if err != nil { + return nil, err + } + + store := &entityBridge{ + entity: entity, + } + + return resource.NewResourceServer(resource.ResourceServerOptions{ + Tracer: tracer, + Store: store, + NodeID: 234, // from config? used for snowflake ID + Diagnostics: store, + Lifecycle: store, + }) +} + +type entityBridge struct { + entity sqlstash.SqlEntityServer +} + +// Init implements ResourceServer. +func (b *entityBridge) Init() error { + return b.entity.Init() +} + +// Stop implements ResourceServer. +func (b *entityBridge) Stop() { + b.entity.Stop() +} + +// Convert resource key to the entity key +func toEntityKey(key *resource.ResourceKey) string { + e := entity.Key{ + Group: key.Group, + Resource: key.Resource, + Namespace: key.Namespace, + Name: key.Name, + } + return e.String() +} + +func (b *entityBridge) WriteEvent(ctx context.Context, event *resource.WriteEvent) (int64, error) { + key := toEntityKey(event.Key) + + // Delete does not need to create an entity first + if event.Operation == resource.ResourceOperation_DELETED { + rsp, err := b.entity.Delete(ctx, &entity.DeleteEntityRequest{ + Key: key, + PreviousVersion: event.PreviousRV, + }) + if err != nil { + return 0, err + } + return rsp.Entity.ResourceVersion, err + } + + obj := event.Object + msg := &entity.Entity{ + Key: key, + Group: event.Key.Group, + Resource: event.Key.Resource, + Namespace: event.Key.Namespace, + Name: event.Key.Name, + Guid: string(event.Object.GetUID()), + + // Key: fmt.Sprint("%s/%s/%s/%s", ), + Folder: obj.GetFolder(), + Body: event.Value, + Message: event.Message, + + Labels: obj.GetLabels(), + Size: int64(len(event.Value)), + } + + switch event.Operation { + case resource.ResourceOperation_CREATED: + msg.Action = entity.Entity_CREATED + rsp, err := b.entity.Create(ctx, &entity.CreateEntityRequest{Entity: msg}) + if err != nil { + return 0, err + } + return rsp.Entity.ResourceVersion, err + + case resource.ResourceOperation_UPDATED: + msg.Action = entity.Entity_UPDATED + rsp, err := b.entity.Update(ctx, &entity.UpdateEntityRequest{ + Entity: msg, + PreviousVersion: event.PreviousRV, + }) + if err != nil { + return 0, err + } + return rsp.Entity.ResourceVersion, err + } + + return 0, fmt.Errorf("unsupported operation: %s", event.Operation.String()) +} + +func (b *entityBridge) Watch(ctx context.Context, req *resource.WatchRequest) (chan *resource.WatchEvent, error) { + return nil, resource.ErrNotImplementedYet +} + +// IsHealthy implements ResourceServer. +func (b *entityBridge) IsHealthy(ctx context.Context, req *resource.HealthCheckRequest) (*resource.HealthCheckResponse, error) { + rsp, err := b.entity.IsHealthy(ctx, &entity.HealthCheckRequest{ + Service: req.Service, // ?? + }) + if err != nil { + return nil, err + } + return &resource.HealthCheckResponse{ + Status: resource.HealthCheckResponse_ServingStatus(rsp.Status), + }, nil +} + +// Read implements ResourceServer. +func (b *entityBridge) Read(ctx context.Context, req *resource.ReadRequest) (*resource.ReadResponse, error) { + v, err := b.entity.Read(ctx, &entity.ReadEntityRequest{ + Key: toEntityKey(req.Key), + WithBody: true, + }) + if err != nil { + return nil, err + } + return &resource.ReadResponse{ + ResourceVersion: v.ResourceVersion, + Value: v.Body, + Message: v.Message, + }, nil +} + +// List implements ResourceServer. +func (b *entityBridge) List(ctx context.Context, req *resource.ListRequest) (*resource.ListResponse, error) { + key := req.Options.Key + query := &entity.EntityListRequest{ + NextPageToken: req.NextPageToken, + Limit: req.Limit, + Key: []string{toEntityKey(key)}, + WithBody: true, + } + + // Assumes everything is equals + if len(req.Options.Labels) > 0 { + query.Labels = make(map[string]string) + for _, q := range req.Options.Labels { + query.Labels[q.Key] = q.Values[0] + } + } + + found, err := b.entity.List(ctx, query) + if err != nil { + return nil, err + } + + rsp := &resource.ListResponse{ + ResourceVersion: found.ResourceVersion, + NextPageToken: found.NextPageToken, + } + for _, item := range found.Results { + rsp.Items = append(rsp.Items, &resource.ResourceWrapper{ + ResourceVersion: item.ResourceVersion, + Value: item.Body, + Operation: resource.ResourceOperation(item.Action), + }) + } + return rsp, nil +} + +// GetBlob implements ResourceServer. +func (b *entityBridge) GetBlob(context.Context, *resource.GetBlobRequest) (*resource.GetBlobResponse, error) { + return nil, resource.ErrNotImplementedYet +} + +// History implements ResourceServer. +func (b *entityBridge) History(context.Context, *resource.HistoryRequest) (*resource.HistoryResponse, error) { + return nil, resource.ErrNotImplementedYet +} + +// Origin implements ResourceServer. +func (b *entityBridge) Origin(context.Context, *resource.OriginRequest) (*resource.OriginResponse, error) { + return nil, resource.ErrNotImplementedYet +} diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 501f044e932..3b31ff9630c 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -442,6 +442,15 @@ func (s *server) Read(ctx context.Context, req *ReadRequest) (*ReadResponse, err return nil, err } + if req.Key.Group == "" { + status, _ := errToStatus(apierrors.NewBadRequest("missing group")) + return &ReadResponse{Status: status}, nil + } + if req.Key.Resource == "" { + status, _ := errToStatus(apierrors.NewBadRequest("missing resource")) + return &ReadResponse{Status: status}, nil + } + rsp, err := s.store.Read(ctx, req) if err != nil { if rsp == nil { diff --git a/pkg/storage/unified/sqlstash/data/common.sql b/pkg/storage/unified/sqlnext/data/common.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/common.sql rename to pkg/storage/unified/sqlnext/data/common.sql diff --git a/pkg/storage/unified/sqlstash/data/resource_get.sql b/pkg/storage/unified/sqlnext/data/resource_get.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/resource_get.sql rename to pkg/storage/unified/sqlnext/data/resource_get.sql diff --git a/pkg/storage/unified/sqlstash/data/resource_insert.sql b/pkg/storage/unified/sqlnext/data/resource_insert.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/resource_insert.sql rename to pkg/storage/unified/sqlnext/data/resource_insert.sql diff --git a/pkg/storage/unified/sqlstash/data/rv_get.sql b/pkg/storage/unified/sqlnext/data/rv_get.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/rv_get.sql rename to pkg/storage/unified/sqlnext/data/rv_get.sql diff --git a/pkg/storage/unified/sqlstash/data/rv_inc.sql b/pkg/storage/unified/sqlnext/data/rv_inc.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/rv_inc.sql rename to pkg/storage/unified/sqlnext/data/rv_inc.sql diff --git a/pkg/storage/unified/sqlstash/data/rv_insert.sql b/pkg/storage/unified/sqlnext/data/rv_insert.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/rv_insert.sql rename to pkg/storage/unified/sqlnext/data/rv_insert.sql diff --git a/pkg/storage/unified/sqlstash/data/rv_lock.sql b/pkg/storage/unified/sqlnext/data/rv_lock.sql similarity index 100% rename from pkg/storage/unified/sqlstash/data/rv_lock.sql rename to pkg/storage/unified/sqlnext/data/rv_lock.sql diff --git a/pkg/storage/unified/sqlstash/sql_storage_server.go b/pkg/storage/unified/sqlnext/sql_resources.go similarity index 95% rename from pkg/storage/unified/sqlstash/sql_storage_server.go rename to pkg/storage/unified/sqlnext/sql_resources.go index ff05bb02b5b..e81fa39841d 100644 --- a/pkg/storage/unified/sqlstash/sql_storage_server.go +++ b/pkg/storage/unified/sqlnext/sql_resources.go @@ -1,4 +1,4 @@ -package sqlstash +package sqlnext import ( "context" @@ -152,13 +152,6 @@ func (s *sqlResourceStore) Read(ctx context.Context, req *resource.ReadRequest) _, span := s.tracer.Start(ctx, "storage_server.GetResource") defer span.End() - if req.Key.Group == "" { - return &resource.ReadResponse{Status: badRequest("missing group")}, nil - } - if req.Key.Resource == "" { - return &resource.ReadResponse{Status: badRequest("missing resource")}, nil - } - fmt.Printf("TODO, GET: %+v", req.Key) return nil, ErrNotImplementedYet diff --git a/pkg/storage/unified/sqlstash/watch.go b/pkg/storage/unified/sqlnext/watch.go similarity index 98% rename from pkg/storage/unified/sqlstash/watch.go rename to pkg/storage/unified/sqlnext/watch.go index 8dfcdb4d5ee..feb5a5da4bb 100644 --- a/pkg/storage/unified/sqlstash/watch.go +++ b/pkg/storage/unified/sqlnext/watch.go @@ -1,4 +1,4 @@ -package sqlstash +package sqlnext import ( "context" diff --git a/pkg/storage/unified/sqlstash/utils.go b/pkg/storage/unified/sqlstash/utils.go deleted file mode 100644 index afc742fd766..00000000000 --- a/pkg/storage/unified/sqlstash/utils.go +++ /dev/null @@ -1,15 +0,0 @@ -package sqlstash - -import ( - "net/http" - - "github.com/grafana/grafana/pkg/storage/unified/resource" -) - -func badRequest(msg string) *resource.StatusResult { - return &resource.StatusResult{ - Status: "Failure", - Message: msg, - Code: http.StatusBadRequest, - } -}