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

K8s: Remove restore functionality; can be done with list (#102560)

Browse Source
This commit is contained in:
Stephanie Hingtgen
2025-03-20 16:38:32 -05:00
committed by GitHub
parent 92cc10f983
commit c33a53a47a
32 changed files with 732 additions and 1873 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/resource.pb.go
+606 -749
View File
File diff suppressed because it is too large Load Diff
pkg/storage/unified/resource/resource.proto
-23
View File
@@ -177,8 +177,6 @@ message ReadRequest {
// Optionally pick an explicit resource version
int64 resource_version = 2;
// Optionally decide to return the latest RV if deleted
bool include_deleted = 3;
}
message ReadResponse {
@@ -702,26 +700,6 @@ message ResourceTableRow {
bytes object = 4;
}
//----------------------------
// Restore Support
//----------------------------
message RestoreRequest {
// Full key must be set
ResourceKey key = 1;
// The resource version to restore
int64 resource_version = 2;
}
message RestoreResponse {
// Error details
ErrorResult error = 1;
// The updated resource version
int64 resource_version = 2;
}
//----------------------------
// Blob Support
//----------------------------
@@ -811,7 +789,6 @@ service ResourceStore {
rpc Create(CreateRequest) returns (CreateResponse);
rpc Update(UpdateRequest) returns (UpdateResponse);
rpc Delete(DeleteRequest) returns (DeleteResponse);
rpc Restore(RestoreRequest) returns (RestoreResponse);
// The results *may* include values that should not be returned to the user
// This will perform best-effort filtering to increase performace.
pkg/storage/unified/resource/resource_grpc.pb.go
+119 -156
View File
@@ -1,6 +1,6 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.4.0
// - protoc-gen-go-grpc v1.5.1
// - protoc (unknown)
// source: resource.proto
@@ -15,17 +15,16 @@ import (
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.62.0 or later.
const _ = grpc.SupportPackageIsVersion8
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
ResourceStore_Read_FullMethodName = "/resource.ResourceStore/Read"
ResourceStore_Create_FullMethodName = "/resource.ResourceStore/Create"
ResourceStore_Update_FullMethodName = "/resource.ResourceStore/Update"
ResourceStore_Delete_FullMethodName = "/resource.ResourceStore/Delete"
ResourceStore_Restore_FullMethodName = "/resource.ResourceStore/Restore"
ResourceStore_List_FullMethodName = "/resource.ResourceStore/List"
ResourceStore_Watch_FullMethodName = "/resource.ResourceStore/Watch"
ResourceStore_Read_FullMethodName = "/resource.ResourceStore/Read"
ResourceStore_Create_FullMethodName = "/resource.ResourceStore/Create"
ResourceStore_Update_FullMethodName = "/resource.ResourceStore/Update"
ResourceStore_Delete_FullMethodName = "/resource.ResourceStore/Delete"
ResourceStore_List_FullMethodName = "/resource.ResourceStore/List"
ResourceStore_Watch_FullMethodName = "/resource.ResourceStore/Watch"
)
// ResourceStoreClient is the client API for ResourceStore service.
@@ -41,7 +40,6 @@ type ResourceStoreClient interface {
Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error)
Update(ctx context.Context, in *UpdateRequest, opts ...grpc.CallOption) (*UpdateResponse, error)
Delete(ctx context.Context, in *DeleteRequest, opts ...grpc.CallOption) (*DeleteResponse, error)
Restore(ctx context.Context, in *RestoreRequest, opts ...grpc.CallOption) (*RestoreResponse, error)
// The results *may* include values that should not be returned to the user
// This will perform best-effort filtering to increase performace.
// NOTE: storage.Interface is ultimatly responsible for the final filtering
@@ -49,7 +47,7 @@ type ResourceStoreClient interface {
// The results *may* include values that should not be returned to the user
// This will perform best-effort filtering to increase performace.
// NOTE: storage.Interface is ultimatly responsible for the final filtering
Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (ResourceStore_WatchClient, error)
Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[WatchEvent], error)
}
type resourceStoreClient struct {
@@ -100,16 +98,6 @@ func (c *resourceStoreClient) Delete(ctx context.Context, in *DeleteRequest, opt
return out, nil
}
func (c *resourceStoreClient) Restore(ctx context.Context, in *RestoreRequest, opts ...grpc.CallOption) (*RestoreResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RestoreResponse)
err := c.cc.Invoke(ctx, ResourceStore_Restore_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *resourceStoreClient) List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListResponse)
@@ -120,13 +108,13 @@ func (c *resourceStoreClient) List(ctx context.Context, in *ListRequest, opts ..
return out, nil
}
func (c *resourceStoreClient) Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (ResourceStore_WatchClient, error) {
func (c *resourceStoreClient) Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[WatchEvent], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &ResourceStore_ServiceDesc.Streams[0], ResourceStore_Watch_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &resourceStoreWatchClient{ClientStream: stream}
x := &grpc.GenericClientStream[WatchRequest, WatchEvent]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
@@ -136,26 +124,12 @@ func (c *resourceStoreClient) Watch(ctx context.Context, in *WatchRequest, opts
return x, nil
}
type ResourceStore_WatchClient interface {
Recv() (*WatchEvent, error)
grpc.ClientStream
}
type resourceStoreWatchClient struct {
grpc.ClientStream
}
func (x *resourceStoreWatchClient) Recv() (*WatchEvent, error) {
m := new(WatchEvent)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type ResourceStore_WatchClient = grpc.ServerStreamingClient[WatchEvent]
// ResourceStoreServer is the server API for ResourceStore service.
// All implementations should embed UnimplementedResourceStoreServer
// for forward compatibility
// for forward compatibility.
//
// This provides the CRUD+List+Watch support needed for a k8s apiserver
// The semantics and behaviors of this service are constrained by kubernetes
@@ -166,7 +140,6 @@ type ResourceStoreServer interface {
Create(context.Context, *CreateRequest) (*CreateResponse, error)
Update(context.Context, *UpdateRequest) (*UpdateResponse, error)
Delete(context.Context, *DeleteRequest) (*DeleteResponse, error)
Restore(context.Context, *RestoreRequest) (*RestoreResponse, error)
// The results *may* include values that should not be returned to the user
// This will perform best-effort filtering to increase performace.
// NOTE: storage.Interface is ultimatly responsible for the final filtering
@@ -174,12 +147,15 @@ type ResourceStoreServer interface {
// The results *may* include values that should not be returned to the user
// This will perform best-effort filtering to increase performace.
// NOTE: storage.Interface is ultimatly responsible for the final filtering
Watch(*WatchRequest, ResourceStore_WatchServer) error
Watch(*WatchRequest, grpc.ServerStreamingServer[WatchEvent]) error
}
// UnimplementedResourceStoreServer should be embedded to have forward compatible implementations.
type UnimplementedResourceStoreServer struct {
}
// UnimplementedResourceStoreServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedResourceStoreServer struct{}
func (UnimplementedResourceStoreServer) Read(context.Context, *ReadRequest) (*ReadResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method Read not implemented")
@@ -193,15 +169,13 @@ func (UnimplementedResourceStoreServer) Update(context.Context, *UpdateRequest)
func (UnimplementedResourceStoreServer) Delete(context.Context, *DeleteRequest) (*DeleteResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method Delete not implemented")
}
func (UnimplementedResourceStoreServer) Restore(context.Context, *RestoreRequest) (*RestoreResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method Restore not implemented")
}
func (UnimplementedResourceStoreServer) List(context.Context, *ListRequest) (*ListResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method List not implemented")
}
func (UnimplementedResourceStoreServer) Watch(*WatchRequest, ResourceStore_WatchServer) error {
func (UnimplementedResourceStoreServer) Watch(*WatchRequest, grpc.ServerStreamingServer[WatchEvent]) error {
return status.Errorf(codes.Unimplemented, "method Watch not implemented")
}
func (UnimplementedResourceStoreServer) testEmbeddedByValue() {}
// UnsafeResourceStoreServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ResourceStoreServer will
@@ -211,6 +185,13 @@ type UnsafeResourceStoreServer interface {
}
func RegisterResourceStoreServer(s grpc.ServiceRegistrar, srv ResourceStoreServer) {
// If the following call pancis, it indicates UnimplementedResourceStoreServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&ResourceStore_ServiceDesc, srv)
}
@@ -286,24 +267,6 @@ func _ResourceStore_Delete_Handler(srv interface{}, ctx context.Context, dec fun
return interceptor(ctx, in, info, handler)
}
func _ResourceStore_Restore_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RestoreRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ResourceStoreServer).Restore(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ResourceStore_Restore_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ResourceStoreServer).Restore(ctx, req.(*RestoreRequest))
}
return interceptor(ctx, in, info, handler)
}
func _ResourceStore_List_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListRequest)
if err := dec(in); err != nil {
@@ -327,21 +290,11 @@ func _ResourceStore_Watch_Handler(srv interface{}, stream grpc.ServerStream) err
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(ResourceStoreServer).Watch(m, &resourceStoreWatchServer{ServerStream: stream})
return srv.(ResourceStoreServer).Watch(m, &grpc.GenericServerStream[WatchRequest, WatchEvent]{ServerStream: stream})
}
type ResourceStore_WatchServer interface {
Send(*WatchEvent) error
grpc.ServerStream
}
type resourceStoreWatchServer struct {
grpc.ServerStream
}
func (x *resourceStoreWatchServer) Send(m *WatchEvent) error {
return x.ServerStream.SendMsg(m)
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type ResourceStore_WatchServer = grpc.ServerStreamingServer[WatchEvent]
// ResourceStore_ServiceDesc is the grpc.ServiceDesc for ResourceStore service.
// It's only intended for direct use with grpc.RegisterService,
@@ -366,10 +319,6 @@ var ResourceStore_ServiceDesc = grpc.ServiceDesc{
MethodName: "Delete",
Handler: _ResourceStore_Delete_Handler,
},
{
MethodName: "Restore",
Handler: _ResourceStore_Restore_Handler,
},
{
MethodName: "List",
Handler: _ResourceStore_List_Handler,
@@ -396,7 +345,7 @@ type BulkStoreClient interface {
// Write multiple resources to the same Namespace/Group/Resource
// Events will not be sent until the stream is complete
// Only the *create* permissions is checked
BulkProcess(ctx context.Context, opts ...grpc.CallOption) (BulkStore_BulkProcessClient, error)
BulkProcess(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[BulkRequest, BulkResponse], error)
}
type bulkStoreClient struct {
@@ -407,58 +356,40 @@ func NewBulkStoreClient(cc grpc.ClientConnInterface) BulkStoreClient {
return &bulkStoreClient{cc}
}
func (c *bulkStoreClient) BulkProcess(ctx context.Context, opts ...grpc.CallOption) (BulkStore_BulkProcessClient, error) {
func (c *bulkStoreClient) BulkProcess(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[BulkRequest, BulkResponse], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &BulkStore_ServiceDesc.Streams[0], BulkStore_BulkProcess_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &bulkStoreBulkProcessClient{ClientStream: stream}
x := &grpc.GenericClientStream[BulkRequest, BulkResponse]{ClientStream: stream}
return x, nil
}
type BulkStore_BulkProcessClient interface {
Send(*BulkRequest) error
CloseAndRecv() (*BulkResponse, error)
grpc.ClientStream
}
type bulkStoreBulkProcessClient struct {
grpc.ClientStream
}
func (x *bulkStoreBulkProcessClient) Send(m *BulkRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *bulkStoreBulkProcessClient) CloseAndRecv() (*BulkResponse, error) {
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
m := new(BulkResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type BulkStore_BulkProcessClient = grpc.ClientStreamingClient[BulkRequest, BulkResponse]
// BulkStoreServer is the server API for BulkStore service.
// All implementations should embed UnimplementedBulkStoreServer
// for forward compatibility
// for forward compatibility.
type BulkStoreServer interface {
// Write multiple resources to the same Namespace/Group/Resource
// Events will not be sent until the stream is complete
// Only the *create* permissions is checked
BulkProcess(BulkStore_BulkProcessServer) error
BulkProcess(grpc.ClientStreamingServer[BulkRequest, BulkResponse]) error
}
// UnimplementedBulkStoreServer should be embedded to have forward compatible implementations.
type UnimplementedBulkStoreServer struct {
}
// UnimplementedBulkStoreServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedBulkStoreServer struct{}
func (UnimplementedBulkStoreServer) BulkProcess(BulkStore_BulkProcessServer) error {
func (UnimplementedBulkStoreServer) BulkProcess(grpc.ClientStreamingServer[BulkRequest, BulkResponse]) error {
return status.Errorf(codes.Unimplemented, "method BulkProcess not implemented")
}
func (UnimplementedBulkStoreServer) testEmbeddedByValue() {}
// UnsafeBulkStoreServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to BulkStoreServer will
@@ -468,34 +399,22 @@ type UnsafeBulkStoreServer interface {
}
func RegisterBulkStoreServer(s grpc.ServiceRegistrar, srv BulkStoreServer) {
// If the following call pancis, it indicates UnimplementedBulkStoreServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&BulkStore_ServiceDesc, srv)
}
func _BulkStore_BulkProcess_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(BulkStoreServer).BulkProcess(&bulkStoreBulkProcessServer{ServerStream: stream})
return srv.(BulkStoreServer).BulkProcess(&grpc.GenericServerStream[BulkRequest, BulkResponse]{ServerStream: stream})
}
type BulkStore_BulkProcessServer interface {
SendAndClose(*BulkResponse) error
Recv() (*BulkRequest, error)
grpc.ServerStream
}
type bulkStoreBulkProcessServer struct {
grpc.ServerStream
}
func (x *bulkStoreBulkProcessServer) SendAndClose(m *BulkResponse) error {
return x.ServerStream.SendMsg(m)
}
func (x *bulkStoreBulkProcessServer) Recv() (*BulkRequest, error) {
m := new(BulkRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type BulkStore_BulkProcessServer = grpc.ClientStreamingServer[BulkRequest, BulkResponse]
// BulkStore_ServiceDesc is the grpc.ServiceDesc for BulkStore service.
// It's only intended for direct use with grpc.RegisterService,
@@ -561,7 +480,7 @@ func (c *resourceIndexClient) GetStats(ctx context.Context, in *ResourceStatsReq
// ResourceIndexServer is the server API for ResourceIndex service.
// All implementations should embed UnimplementedResourceIndexServer
// for forward compatibility
// for forward compatibility.
//
// Unlike the ResourceStore, this service can be exposed to clients directly
// It should be implemented with efficient indexes and does not need read-after-write semantics
@@ -571,9 +490,12 @@ type ResourceIndexServer interface {
GetStats(context.Context, *ResourceStatsRequest) (*ResourceStatsResponse, error)
}
// UnimplementedResourceIndexServer should be embedded to have forward compatible implementations.
type UnimplementedResourceIndexServer struct {
}
// UnimplementedResourceIndexServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedResourceIndexServer struct{}
func (UnimplementedResourceIndexServer) Search(context.Context, *ResourceSearchRequest) (*ResourceSearchResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method Search not implemented")
@@ -581,6 +503,7 @@ func (UnimplementedResourceIndexServer) Search(context.Context, *ResourceSearchR
func (UnimplementedResourceIndexServer) GetStats(context.Context, *ResourceStatsRequest) (*ResourceStatsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetStats not implemented")
}
func (UnimplementedResourceIndexServer) testEmbeddedByValue() {}
// UnsafeResourceIndexServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ResourceIndexServer will
@@ -590,6 +513,13 @@ type UnsafeResourceIndexServer interface {
}
func RegisterResourceIndexServer(s grpc.ServiceRegistrar, srv ResourceIndexServer) {
// If the following call pancis, it indicates UnimplementedResourceIndexServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&ResourceIndex_ServiceDesc, srv)
}
@@ -697,7 +627,7 @@ func (c *managedObjectIndexClient) ListManagedObjects(ctx context.Context, in *L
// ManagedObjectIndexServer is the server API for ManagedObjectIndex service.
// All implementations should embed UnimplementedManagedObjectIndexServer
// for forward compatibility
// for forward compatibility.
//
// Query managed objects
// Results access control is based on access to the repository *not* the items
@@ -708,9 +638,12 @@ type ManagedObjectIndexServer interface {
ListManagedObjects(context.Context, *ListManagedObjectsRequest) (*ListManagedObjectsResponse, error)
}
// UnimplementedManagedObjectIndexServer should be embedded to have forward compatible implementations.
type UnimplementedManagedObjectIndexServer struct {
}
// UnimplementedManagedObjectIndexServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedManagedObjectIndexServer struct{}
func (UnimplementedManagedObjectIndexServer) CountManagedObjects(context.Context, *CountManagedObjectsRequest) (*CountManagedObjectsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method CountManagedObjects not implemented")
@@ -718,6 +651,7 @@ func (UnimplementedManagedObjectIndexServer) CountManagedObjects(context.Context
func (UnimplementedManagedObjectIndexServer) ListManagedObjects(context.Context, *ListManagedObjectsRequest) (*ListManagedObjectsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListManagedObjects not implemented")
}
func (UnimplementedManagedObjectIndexServer) testEmbeddedByValue() {}
// UnsafeManagedObjectIndexServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ManagedObjectIndexServer will
@@ -727,6 +661,13 @@ type UnsafeManagedObjectIndexServer interface {
}
func RegisterManagedObjectIndexServer(s grpc.ServiceRegistrar, srv ManagedObjectIndexServer) {
// If the following call pancis, it indicates UnimplementedManagedObjectIndexServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&ManagedObjectIndex_ServiceDesc, srv)
}
@@ -832,7 +773,7 @@ func (c *blobStoreClient) GetBlob(ctx context.Context, in *GetBlobRequest, opts
// BlobStoreServer is the server API for BlobStore service.
// All implementations should embed UnimplementedBlobStoreServer
// for forward compatibility
// for forward compatibility.
type BlobStoreServer interface {
// Upload a blob that will be saved in a resource
PutBlob(context.Context, *PutBlobRequest) (*PutBlobResponse, error)
@@ -841,9 +782,12 @@ type BlobStoreServer interface {
GetBlob(context.Context, *GetBlobRequest) (*GetBlobResponse, error)
}
// UnimplementedBlobStoreServer should be embedded to have forward compatible implementations.
type UnimplementedBlobStoreServer struct {
}
// UnimplementedBlobStoreServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedBlobStoreServer struct{}
func (UnimplementedBlobStoreServer) PutBlob(context.Context, *PutBlobRequest) (*PutBlobResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method PutBlob not implemented")
@@ -851,6 +795,7 @@ func (UnimplementedBlobStoreServer) PutBlob(context.Context, *PutBlobRequest) (*
func (UnimplementedBlobStoreServer) GetBlob(context.Context, *GetBlobRequest) (*GetBlobResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetBlob not implemented")
}
func (UnimplementedBlobStoreServer) testEmbeddedByValue() {}
// UnsafeBlobStoreServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to BlobStoreServer will
@@ -860,6 +805,13 @@ type UnsafeBlobStoreServer interface {
}
func RegisterBlobStoreServer(s grpc.ServiceRegistrar, srv BlobStoreServer) {
// If the following call pancis, it indicates UnimplementedBlobStoreServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&BlobStore_ServiceDesc, srv)
}
@@ -954,7 +906,7 @@ func (c *diagnosticsClient) IsHealthy(ctx context.Context, in *HealthCheckReques
// DiagnosticsServer is the server API for Diagnostics service.
// All implementations should embed UnimplementedDiagnosticsServer
// for forward compatibility
// for forward compatibility.
//
// Clients can use this service directly
// NOTE: This is read only, and no read afer write guarantees
@@ -963,13 +915,17 @@ type DiagnosticsServer interface {
IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
}
// UnimplementedDiagnosticsServer should be embedded to have forward compatible implementations.
type UnimplementedDiagnosticsServer struct {
}
// UnimplementedDiagnosticsServer should be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedDiagnosticsServer struct{}
func (UnimplementedDiagnosticsServer) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method IsHealthy not implemented")
}
func (UnimplementedDiagnosticsServer) testEmbeddedByValue() {}
// UnsafeDiagnosticsServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to DiagnosticsServer will
@@ -979,6 +935,13 @@ type UnsafeDiagnosticsServer interface {
}
func RegisterDiagnosticsServer(s grpc.ServiceRegistrar, srv DiagnosticsServer) {
// If the following call pancis, it indicates UnimplementedDiagnosticsServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&Diagnostics_ServiceDesc, srv)
}
pkg/storage/unified/resource/server.go
+1 -129
View File
@@ -10,14 +10,12 @@ import (
"sync/atomic"
"time"
"github.com/google/uuid"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/types"
claims "github.com/grafana/authlib/types"
"github.com/grafana/grafana/pkg/apimachinery/utils"
@@ -393,6 +391,7 @@ func (s *server) newEvent(ctx context.Context, user claims.AuthInfo, key *Resour
if oldValue == nil {
event.Type = WatchEvent_ADDED
} else {
event.Type = WatchEvent_MODIFIED
check.Verb = utils.VerbUpdate
temp := &unstructured.Unstructured{}
@@ -404,13 +403,6 @@ func (s *server) newEvent(ctx context.Context, user claims.AuthInfo, key *Resour
if err != nil {
return nil, AsErrorResult(err)
}
// restores will restore with a different k8s uid
if event.ObjectOld.GetUID() != obj.GetUID() {
event.Type = WatchEvent_ADDED
} else {
event.Type = WatchEvent_MODIFIED
}
}
if key.Namespace != obj.GetNamespace() {
@@ -792,126 +784,6 @@ func (s *server) List(ctx context.Context, req *ListRequest) (*ListResponse, err
return rsp, err
}
func (s *server) Restore(ctx context.Context, req *RestoreRequest) (*RestoreResponse, error) {
ctx, span := s.tracer.Start(ctx, "storage_server.List")
defer span.End()
// check that the user has access
user, ok := claims.AuthInfoFrom(ctx)
if !ok || user == nil {
return &RestoreResponse{
Error: &ErrorResult{
Message: "no user found in context",
Code: http.StatusUnauthorized,
}}, nil
}
if err := s.Init(ctx); err != nil {
return nil, err
}
checker, err := s.access.Compile(ctx, user, claims.ListRequest{
Group: req.Key.Group,
Resource: req.Key.Resource,
Namespace: req.Key.Namespace,
Verb: utils.VerbGet,
})
if err != nil {
return &RestoreResponse{Error: AsErrorResult(err)}, nil
}
if checker == nil {
return &RestoreResponse{Error: &ErrorResult{
Code: http.StatusForbidden,
}}, nil
}
// get the asked for resource version to restore
readRsp, err := s.Read(ctx, &ReadRequest{
Key: req.Key,
ResourceVersion: req.ResourceVersion,
IncludeDeleted: true,
})
if err != nil || readRsp == nil || readRsp.Error != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusNotFound,
Message: fmt.Sprintf("could not find old resource: %s", readRsp.Error.Message),
},
}, nil
}
// generate a new k8s UID when restoring. The name will remain the same
// (for dashboards, this will be the dashboard uid), but since controllers
// will see this as a create event, we do not want the same k8s UID, or
// there may be unintended behavior
newUid := types.UID(uuid.NewString())
tmp := &unstructured.Unstructured{}
err = tmp.UnmarshalJSON(readRsp.Value)
if err != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusNotFound,
Message: fmt.Sprintf("could not unmarhsal: %s", err.Error()),
},
}, nil
}
obj, err := utils.MetaAccessor(tmp)
if err != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusNotFound,
Message: fmt.Sprintf("could not get object: %s", err.Error()),
},
}, nil
}
obj.SetUID(newUid)
rtObj, ok := obj.GetRuntimeObject()
if !ok {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusNotFound,
Message: "could not get runtime object",
},
}, nil
}
newObj, err := json.Marshal(rtObj)
if err != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusNotFound,
Message: fmt.Sprintf("could not marshal object: %s", err.Error()),
},
}, nil
}
// finally, send to the backend to create & update the history of the restored object
event, errRes := s.newEvent(ctx, user, req.Key, newObj, readRsp.Value)
if errRes != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusInternalServerError,
Message: fmt.Sprintf("could not create restore resource event: %s", errRes.Message),
},
}, nil
}
rv, err := s.backend.WriteEvent(ctx, *event)
if err != nil {
return &RestoreResponse{
Error: &ErrorResult{
Code: http.StatusInternalServerError,
Message: fmt.Sprintf("could not restore resource: %s", err.Error()),
},
}, nil
}
return &RestoreResponse{
Error: nil,
ResourceVersion: rv,
}, nil
}
func (s *server) initWatcher() error {
var err error
s.broadcaster, err = NewBroadcaster(s.ctx, func(out chan<- *WrittenEvent) error {
pkg/storage/unified/resource/server_test.go
-75
View File
@@ -233,79 +233,4 @@ func TestSimpleServer(t *testing.T) {
ResourceVersion: created.ResourceVersion})
require.ErrorIs(t, err, ErrOptimisticLockingFailed)
})
t.Run("playlist restore", func(t *testing.T) {
uid := "zzz"
raw := []byte(`{
"apiVersion": "playlist.grafana.app/v0alpha1",
"kind": "Playlist",
"metadata": {
"name": "fdgsv37qslr0ga",
"namespace": "default",
"uid": "` + uid + `",
"annotations": {
"grafana.app/repoName": "elsewhere",
"grafana.app/repoPath": "path/to/item",
"grafana.app/repoTimestamp": "2024-02-02T00:00:00Z"
}
},
"spec": {
"title": "hello",
"interval": "5m",
"items": [
{
"type": "dashboard_by_uid",
"value": "vmie2cmWz"
}
]
}
}`)
key := &ResourceKey{
Group: "playlist.grafana.app",
Resource: "rrrr",
Namespace: "default",
Name: "fdgsv37qslr0ga",
}
// create
created, err := server.Create(ctx, &CreateRequest{
Value: raw,
Key: key,
})
require.NoError(t, err)
// make sure it exists
found, err := server.Read(ctx, &ReadRequest{Key: key})
require.NoError(t, err)
require.Nil(t, found.Error)
fmt.Println(found.ResourceVersion)
// delete it
deleted, err := server.Delete(ctx, &DeleteRequest{Key: key, ResourceVersion: created.ResourceVersion})
require.NoError(t, err)
require.True(t, deleted.ResourceVersion > created.ResourceVersion)
// restore it
restored, err := server.Restore(ctx, &RestoreRequest{
Key: key,
ResourceVersion: found.ResourceVersion,
})
require.NoError(t, err)
require.Nil(t, restored.Error)
require.True(t, restored.ResourceVersion > deleted.ResourceVersion)
// ensure it exists now
found, err = server.Read(ctx, &ReadRequest{Key: key})
require.NoError(t, err)
require.Nil(t, found.Error)
require.Equal(t, restored.ResourceVersion, found.ResourceVersion)
foundUnstructured := &unstructured.Unstructured{}
err = foundUnstructured.UnmarshalJSON(found.Value)
require.NoError(t, err)
foundObj, err := utils.MetaAccessor(foundUnstructured)
require.NoError(t, err)
// the UID should be different now
require.NotEqual(t, uid, string(foundObj.GetUID()))
})
}
pkg/storage/unified/sql/backend.go
-81
View File
@@ -308,9 +308,6 @@ func (b *backend) WriteEvent(ctx context.Context, event resource.WriteEvent) (in
// TODO: validate key ?
switch event.Type {
case resource.WatchEvent_ADDED:
if event.ObjectOld != nil {
return b.restore(ctx, event)
}
return b.create(ctx, event)
case resource.WatchEvent_MODIFIED:
return b.update(ctx, event)
@@ -488,73 +485,6 @@ func (b *backend) delete(ctx context.Context, event resource.WriteEvent) (int64,
return rv, nil
}
func (b *backend) restore(ctx context.Context, event resource.WriteEvent) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"Restore")
defer span.End()
guid := uuid.New().String()
folder := ""
if event.Object != nil {
folder = event.Object.GetFolder()
}
rv, err := b.rvManager.ExecWithRV(ctx, event.Key, func(tx db.Tx) (string, error) {
// 1. Re-create resource
// Note: we may want to replace the write event with a create event, tbd.
if _, err := dbutil.Exec(ctx, tx, sqlResourceInsert, sqlResourceRequest{
SQLTemplate: sqltemplate.New(b.dialect),
WriteEvent: event,
Folder: folder,
GUID: guid,
}); err != nil {
return guid, fmt.Errorf("insert into resource: %w", err)
}
// 2. Insert into resource history
if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryInsert, sqlResourceRequest{
SQLTemplate: sqltemplate.New(b.dialect),
WriteEvent: event,
Folder: folder,
GUID: guid,
}); err != nil {
return guid, fmt.Errorf("insert into resource history: %w", err)
}
_ = b.historyPruner.Add(pruningKey{
namespace: event.Key.Namespace,
group: event.Key.Group,
resource: event.Key.Resource,
name: event.Key.Name,
})
// 3. Update all resource history entries with the new UID
// Note: we do not update any history entries that have a deletion timestamp included. This will become
// important once we start using finalizers, as the initial delete will show up as an update with a deletion timestamp included.
if _, err := dbutil.Exec(ctx, tx, sqlResoureceHistoryUpdateUid, sqlResourceHistoryUpdateRequest{
SQLTemplate: sqltemplate.New(b.dialect),
WriteEvent: event,
OldUID: string(event.ObjectOld.GetUID()),
NewUID: string(event.Object.GetUID()),
}); err != nil {
return guid, fmt.Errorf("update history uid: %w", err)
}
return guid, nil
})
if err != nil {
return 0, err
}
b.notifier.send(ctx, &resource.WrittenEvent{
Type: event.Type,
Key: event.Key,
PreviousRV: event.PreviousRV,
Value: event.Value,
ResourceVersion: rv,
Folder: folder,
})
return rv, nil
}
func (b *backend) ReadResource(ctx context.Context, req *resource.ReadRequest) *resource.BackendReadResponse {
_, span := b.tracer.Start(ctx, tracePrefix+".Read")
defer span.End()
@@ -577,17 +507,6 @@ func (b *backend) ReadResource(ctx context.Context, req *resource.ReadRequest) *
err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
var err error
res, err = dbutil.QueryRow(ctx, tx, sr, readReq)
// if not found, look for latest deleted version (if requested)
if errors.Is(err, sql.ErrNoRows) && req.IncludeDeleted {
sr = sqlResourceHistoryRead
readReq2 := &sqlResourceReadRequest{
SQLTemplate: sqltemplate.New(b.dialect),
Request: req,
Response: NewReadResponse(),
}
res, err = dbutil.QueryRow(ctx, tx, sr, readReq2)
return err
}
return err
})
pkg/storage/unified/sql/backend_test.go
-82
View File
@@ -376,88 +376,6 @@ func TestBackend_delete(t *testing.T) {
})
}
func TestBackend_restore(t *testing.T) {
t.Parallel()
meta, err := utils.MetaAccessor(&unstructured.Unstructured{
Object: map[string]any{},
})
require.NoError(t, err)
meta.SetUID("new-uid")
oldMeta, err := utils.MetaAccessor(&unstructured.Unstructured{
Object: map[string]any{},
})
require.NoError(t, err)
oldMeta.SetUID("old-uid")
event := resource.WriteEvent{
Type: resource.WatchEvent_ADDED,
Key: resKey,
Object: meta,
ObjectOld: oldMeta,
}
t.Run("happy path", func(t *testing.T) {
t.Parallel()
b, ctx := setupBackendTest(t)
b.SQLMock.ExpectBegin()
expectSuccessfulResourceVersionExec(t, b.TestDBProvider,
func() { b.ExecWithResult("insert resource", 0, 1) },
func() { b.ExecWithResult("insert resource_history", 0, 1) },
func() { b.ExecWithResult("update resource_history", 0, 1) },
)
b.SQLMock.ExpectCommit()
v, err := b.restore(ctx, event)
require.NoError(t, err)
require.Equal(t, int64(200), v)
})
t.Run("error restoring resource", func(t *testing.T) {
t.Parallel()
b, ctx := setupBackendTest(t)
b.SQLMock.ExpectBegin()
b.ExecWithErr("insert resource", errTest)
b.SQLMock.ExpectRollback()
v, err := b.restore(ctx, event)
require.Zero(t, v)
require.Error(t, err)
require.ErrorContains(t, err, "insert into resource")
})
t.Run("error inserting into resource history", func(t *testing.T) {
t.Parallel()
b, ctx := setupBackendTest(t)
b.SQLMock.ExpectBegin()
b.ExecWithResult("insert resource", 0, 1)
b.ExecWithErr("insert resource_history", errTest)
b.SQLMock.ExpectRollback()
v, err := b.restore(ctx, event)
require.Zero(t, v)
require.Error(t, err)
require.ErrorContains(t, err, "insert into resource history")
})
t.Run("error updating resource history uid", func(t *testing.T) {
t.Parallel()
b, ctx := setupBackendTest(t)
b.SQLMock.ExpectBegin()
b.ExecWithResult("insert resource", 0, 1)
b.ExecWithResult("insert resource_history", 0, 1)
b.ExecWithErr("update resource_history", errTest)
b.SQLMock.ExpectRollback()
v, err := b.restore(ctx, event)
require.Zero(t, v)
require.Error(t, err)
require.ErrorContains(t, err, "update history uid")
})
}
func TestBackend_getHistory(t *testing.T) {
t.Parallel()
pkg/storage/unified/sql/data/resource_history_read.sql
+1 -5
View File
@@ -14,12 +14,8 @@ SELECT
AND {{ .Ident "group" }} = {{ .Arg .Request.Key.Group }}
AND {{ .Ident "resource" }} = {{ .Arg .Request.Key.Resource }}
AND {{ .Ident "name" }} = {{ .Arg .Request.Key.Name }}
{{ if .Request.IncludeDeleted }}
AND {{ .Ident "action" }} != 3
AND {{ .Ident "value" }} NOT LIKE '%deletionTimestamp%'
{{ end }}
{{ if gt .Request.ResourceVersion 0 }}
AND {{ .Ident "resource_version" }} {{ if .Request.IncludeDeleted }}={{ else }}<={{ end }} {{ .Arg .Request.ResourceVersion }}
AND {{ .Ident "resource_version" }} <= {{ .Arg .Request.ResourceVersion }}
{{ end }}
ORDER BY {{ .Ident "resource_version" }} DESC
LIMIT 1
pkg/storage/unified/sql/data/resource_history_update_uid.sql
-8
View File
@@ -1,8 +0,0 @@
UPDATE {{ .Ident "resource_history" }}
SET {{ .Ident "value" }} = REPLACE({{ .Ident "value" }}, CONCAT('"uid":"', {{ .Arg .OldUID }}, '"'), CONCAT('"uid":"', {{ .Arg .NewUID }}, '"'))
WHERE {{ .Ident "name" }} = {{ .Arg .WriteEvent.Key.Name }}
AND {{ .Ident "namespace" }} = {{ .Arg .WriteEvent.Key.Namespace }}
AND {{ .Ident "group" }} = {{ .Arg .WriteEvent.Key.Group }}
AND {{ .Ident "resource" }} = {{ .Arg .WriteEvent.Key.Resource }}
AND {{ .Ident "action" }} != 3
AND {{ .Ident "value" }} NOT LIKE '%deletionTimestamp%';
pkg/storage/unified/sql/queries.go
-14
View File
@@ -39,7 +39,6 @@ var (
sqlResourceUpdateRV = mustTemplate("resource_update_rv.sql")
sqlResourceHistoryRead = mustTemplate("resource_history_read.sql")
sqlResourceHistoryUpdateRV = mustTemplate("resource_history_update_rv.sql")
sqlResoureceHistoryUpdateUid = mustTemplate("resource_history_update_uid.sql")
sqlResourceHistoryInsert = mustTemplate("resource_history_insert.sql")
sqlResourceHistoryPoll = mustTemplate("resource_history_poll.sql")
sqlResourceHistoryGet = mustTemplate("resource_history_get.sql")
@@ -281,19 +280,6 @@ func (r *sqlPruneHistoryRequest) Validate() error {
return nil
}
// update resource history
type sqlResourceHistoryUpdateRequest struct {
sqltemplate.SQLTemplate
WriteEvent resource.WriteEvent
OldUID string
NewUID string
}
func (r sqlResourceHistoryUpdateRequest) Validate() error {
return nil // TODO
}
type sqlResourceBlobInsertRequest struct {
sqltemplate.SQLTemplate
Now time.Time
pkg/storage/unified/sql/queries_test.go
-20
View File
@@ -176,26 +176,6 @@ func TestUnifiedStorageQueries(t *testing.T) {
},
},
sqlResoureceHistoryUpdateUid: {
{
Name: "modify uids in history",
Data: &sqlResourceHistoryUpdateRequest{
SQLTemplate: mocks.NewTestingSQLTemplate(),
WriteEvent: resource.WriteEvent{
Key: &resource.ResourceKey{
Namespace: "nn",
Group: "gg",
Resource: "rr",
Name: "name",
},
PreviousRV: 1234,
},
OldUID: "old-uid",
NewUID: "new-uid",
},
},
},
sqlResourceHistoryInsert: {
{
Name: "insert into resource_history",
pkg/storage/unified/sql/testdata/mysql--resource_history_update_uid-modify uids in history.sql
Vendored
-8
View File
@@ -1,8 +0,0 @@
UPDATE `resource_history`
SET `value` = REPLACE(`value`, CONCAT('"uid":"', 'old-uid', '"'), CONCAT('"uid":"', 'new-uid', '"'))
WHERE `name` = 'name'
AND `namespace` = 'nn'
AND `group` = 'gg'
AND `resource` = 'rr'
AND `action` != 3
AND `value` NOT LIKE '%deletionTimestamp%';
pkg/storage/unified/sql/testdata/postgres--resource_history_update_uid-modify uids in history.sql
Vendored
-8
View File
@@ -1,8 +0,0 @@
UPDATE "resource_history"
SET "value" = REPLACE("value", CONCAT('"uid":"', 'old-uid', '"'), CONCAT('"uid":"', 'new-uid', '"'))
WHERE "name" = 'name'
AND "namespace" = 'nn'
AND "group" = 'gg'
AND "resource" = 'rr'
AND "action" != 3
AND "value" NOT LIKE '%deletionTimestamp%';
pkg/storage/unified/sql/testdata/sqlite--resource_history_update_uid-modify uids in history.sql
Vendored
-8
View File
@@ -1,8 +0,0 @@
UPDATE "resource_history"
SET "value" = REPLACE("value", CONCAT('"uid":"', 'old-uid', '"'), CONCAT('"uid":"', 'new-uid', '"'))
WHERE "name" = 'name'
AND "namespace" = 'nn'
AND "group" = 'gg'
AND "resource" = 'rr'
AND "action" != 3
AND "value" NOT LIKE '%deletionTimestamp%';