now with a base server implementation

This commit is contained in:
Ryan McKinley 2024-06-14 14:24:26 +03:00
parent 68003738fd
commit 4a41f7d0dd
9 changed files with 188 additions and 899 deletions

View File

@ -6,6 +6,11 @@ this is a go module that can be imported into external projects
This includes the protobuf based client+server and all the logic required to convert requests into write events.
Protobuf TODO?
- can/should we use upstream k8s proto for query object?
- starting a project today... should we use proto3?
== apistore
The apiserver storage.Interface that links the storage to kubernetes
@ -13,3 +18,7 @@ The apiserver storage.Interface that links the storage to kubernetes
== sqlstash
SQL based implementation of the unified storage server

View File

@ -2654,7 +2654,7 @@ var file_resource_proto_rawDesc = []byte{
0x52, 0x4b, 0x10, 0x04, 0x2a, 0x33, 0x0a, 0x14, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x4d, 0x61, 0x74, 0x63, 0x68, 0x12, 0x10, 0x0a, 0x0c,
0x4e, 0x6f, 0x74, 0x4f, 0x6c, 0x64, 0x65, 0x72, 0x54, 0x68, 0x61, 0x6e, 0x10, 0x00, 0x12, 0x09,
0x0a, 0x05, 0x45, 0x78, 0x61, 0x63, 0x74, 0x10, 0x01, 0x32, 0xba, 0x03, 0x0a, 0x0d, 0x52, 0x65,
0x0a, 0x05, 0x45, 0x78, 0x61, 0x63, 0x74, 0x10, 0x01, 0x32, 0xf0, 0x02, 0x0a, 0x0d, 0x52, 0x65,
0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x65, 0x12, 0x35, 0x0a, 0x04, 0x52,
0x65, 0x61, 0x64, 0x12, 0x15, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52,
0x65, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x72, 0x65, 0x73,
@ -2677,37 +2677,33 @@ var file_resource_proto_rawDesc = []byte{
0x12, 0x3a, 0x0a, 0x05, 0x57, 0x61, 0x74, 0x63, 0x68, 0x12, 0x16, 0x2e, 0x72, 0x65, 0x73, 0x6f,
0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
0x74, 0x1a, 0x17, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74,
0x63, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x30, 0x01, 0x12, 0x48, 0x0a, 0x09,
0x49, 0x73, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x79, 0x12, 0x1c, 0x2e, 0x72, 0x65, 0x73, 0x6f,
0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65,
0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0xce, 0x02, 0x0a, 0x0e, 0x52, 0x65, 0x73, 0x6f, 0x75,
0x72, 0x63, 0x65, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x12, 0x35, 0x0a, 0x04, 0x52, 0x65, 0x61,
0x64, 0x12, 0x15, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61,
0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75,
0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
0x12, 0x3e, 0x0a, 0x07, 0x47, 0x65, 0x74, 0x42, 0x6c, 0x6f, 0x62, 0x12, 0x18, 0x2e, 0x72, 0x65,
0x63, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x30, 0x01, 0x32, 0x84, 0x02, 0x0a,
0x0e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x12,
0x35, 0x0a, 0x04, 0x52, 0x65, 0x61, 0x64, 0x12, 0x15, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16,
0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65,
0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3e, 0x0a, 0x07, 0x47, 0x65, 0x74, 0x42, 0x6c, 0x6f,
0x62, 0x12, 0x18, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x47, 0x65, 0x74,
0x42, 0x6c, 0x6f, 0x62, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x72, 0x65,
0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x47, 0x65, 0x74, 0x42, 0x6c, 0x6f, 0x62, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
0x2e, 0x47, 0x65, 0x74, 0x42, 0x6c, 0x6f, 0x62, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
0x12, 0x3e, 0x0a, 0x07, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x12, 0x18, 0x2e, 0x72, 0x65,
0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3e, 0x0a, 0x07, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72,
0x79, 0x12, 0x18, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x69, 0x73,
0x74, 0x6f, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x72, 0x65,
0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
0x12, 0x3b, 0x0a, 0x06, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x12, 0x17, 0x2e, 0x72, 0x65, 0x73,
0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4f,
0x72, 0x69, 0x67, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x48, 0x0a,
0x09, 0x49, 0x73, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x79, 0x12, 0x1c, 0x2e, 0x72, 0x65, 0x73,
0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63,
0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75,
0x72, 0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52,
0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x39, 0x5a, 0x37, 0x67, 0x69, 0x74, 0x68, 0x75,
0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x67, 0x72,
0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
0x65, 0x2f, 0x75, 0x6e, 0x69, 0x66, 0x69, 0x65, 0x64, 0x2f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
0x63, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3b, 0x0a, 0x06, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e,
0x12, 0x17, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4f, 0x72, 0x69, 0x67,
0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x72, 0x65, 0x73, 0x6f,
0x75, 0x72, 0x63, 0x65, 0x2e, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
0x6e, 0x73, 0x65, 0x32, 0x57, 0x0a, 0x0b, 0x44, 0x69, 0x61, 0x67, 0x6e, 0x6f, 0x73, 0x74, 0x69,
0x63, 0x73, 0x12, 0x48, 0x0a, 0x09, 0x49, 0x73, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x79, 0x12,
0x1c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74,
0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e,
0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43,
0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x39, 0x5a, 0x37,
0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61,
0x6e, 0x61, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x73,
0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x75, 0x6e, 0x69, 0x66, 0x69, 0x65, 0x64, 0x2f, 0x72,
0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
@ -2800,26 +2796,24 @@ var file_resource_proto_depIdxs = []int32{
15, // 35: resource.ResourceStore.Delete:input_type -> resource.DeleteRequest
24, // 36: resource.ResourceStore.List:input_type -> resource.ListRequest
26, // 37: resource.ResourceStore.Watch:input_type -> resource.WatchRequest
33, // 38: resource.ResourceStore.IsHealthy:input_type -> resource.HealthCheckRequest
17, // 39: resource.ResourceSearch.Read:input_type -> resource.ReadRequest
19, // 40: resource.ResourceSearch.GetBlob:input_type -> resource.GetBlobRequest
28, // 41: resource.ResourceSearch.History:input_type -> resource.HistoryRequest
30, // 42: resource.ResourceSearch.Origin:input_type -> resource.OriginRequest
33, // 43: resource.ResourceSearch.IsHealthy:input_type -> resource.HealthCheckRequest
18, // 44: resource.ResourceStore.Read:output_type -> resource.ReadResponse
12, // 45: resource.ResourceStore.Create:output_type -> resource.CreateResponse
14, // 46: resource.ResourceStore.Update:output_type -> resource.UpdateResponse
16, // 47: resource.ResourceStore.Delete:output_type -> resource.DeleteResponse
25, // 48: resource.ResourceStore.List:output_type -> resource.ListResponse
27, // 49: resource.ResourceStore.Watch:output_type -> resource.WatchResponse
34, // 50: resource.ResourceStore.IsHealthy:output_type -> resource.HealthCheckResponse
18, // 51: resource.ResourceSearch.Read:output_type -> resource.ReadResponse
20, // 52: resource.ResourceSearch.GetBlob:output_type -> resource.GetBlobResponse
29, // 53: resource.ResourceSearch.History:output_type -> resource.HistoryResponse
32, // 54: resource.ResourceSearch.Origin:output_type -> resource.OriginResponse
34, // 55: resource.ResourceSearch.IsHealthy:output_type -> resource.HealthCheckResponse
44, // [44:56] is the sub-list for method output_type
32, // [32:44] is the sub-list for method input_type
17, // 38: resource.ResourceSearch.Read:input_type -> resource.ReadRequest
19, // 39: resource.ResourceSearch.GetBlob:input_type -> resource.GetBlobRequest
28, // 40: resource.ResourceSearch.History:input_type -> resource.HistoryRequest
30, // 41: resource.ResourceSearch.Origin:input_type -> resource.OriginRequest
33, // 42: resource.Diagnostics.IsHealthy:input_type -> resource.HealthCheckRequest
18, // 43: resource.ResourceStore.Read:output_type -> resource.ReadResponse
12, // 44: resource.ResourceStore.Create:output_type -> resource.CreateResponse
14, // 45: resource.ResourceStore.Update:output_type -> resource.UpdateResponse
16, // 46: resource.ResourceStore.Delete:output_type -> resource.DeleteResponse
25, // 47: resource.ResourceStore.List:output_type -> resource.ListResponse
27, // 48: resource.ResourceStore.Watch:output_type -> resource.WatchResponse
18, // 49: resource.ResourceSearch.Read:output_type -> resource.ReadResponse
20, // 50: resource.ResourceSearch.GetBlob:output_type -> resource.GetBlobResponse
29, // 51: resource.ResourceSearch.History:output_type -> resource.HistoryResponse
32, // 52: resource.ResourceSearch.Origin:output_type -> resource.OriginResponse
34, // 53: resource.Diagnostics.IsHealthy:output_type -> resource.HealthCheckResponse
43, // [43:54] is the sub-list for method output_type
32, // [32:43] is the sub-list for method input_type
32, // [32:32] is the sub-list for extension type_name
32, // [32:32] is the sub-list for extension extendee
0, // [0:32] is the sub-list for field type_name
@ -3200,7 +3194,7 @@ func file_resource_proto_init() {
NumEnums: 5,
NumMessages: 30,
NumExtensions: 0,
NumServices: 2,
NumServices: 3,
},
GoTypes: file_resource_proto_goTypes,
DependencyIndexes: file_resource_proto_depIdxs,

View File

@ -428,13 +428,12 @@ service ResourceStore {
rpc Delete(DeleteRequest) returns (DeleteResponse);
rpc List(ListRequest) returns (ListResponse);
rpc Watch(WatchRequest) returns (stream WatchResponse);
rpc IsHealthy(HealthCheckRequest) returns (HealthCheckResponse);
}
// Clients can use this service directly
// NOTE: This is read only, and no read afer write guarantees
service ResourceSearch {
// rpc Search(...) ...
// TODO: rpc Search(...) ... eventually a typed response
rpc Read(ReadRequest) returns (ReadResponse); // Duplicated -- for client read only usage
@ -446,7 +445,11 @@ service ResourceSearch {
// Used for efficient provisioning
rpc Origin(OriginRequest) returns (OriginResponse);
}
// Clients can use this service directly
// NOTE: This is read only, and no read afer write guarantees
service Diagnostics {
// Check if the service is healthy
rpc IsHealthy(HealthCheckRequest) returns (HealthCheckResponse);
}

View File

@ -19,13 +19,12 @@ import (
const _ = grpc.SupportPackageIsVersion8
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_List_FullMethodName = "/resource.ResourceStore/List"
ResourceStore_Watch_FullMethodName = "/resource.ResourceStore/Watch"
ResourceStore_IsHealthy_FullMethodName = "/resource.ResourceStore/IsHealthy"
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.
@ -43,7 +42,6 @@ type ResourceStoreClient interface {
Delete(ctx context.Context, in *DeleteRequest, opts ...grpc.CallOption) (*DeleteResponse, error)
List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error)
Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (ResourceStore_WatchClient, error)
IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error)
}
type resourceStoreClient struct {
@ -137,16 +135,6 @@ func (x *resourceStoreWatchClient) Recv() (*WatchResponse, error) {
return m, nil
}
func (c *resourceStoreClient) IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(HealthCheckResponse)
err := c.cc.Invoke(ctx, ResourceStore_IsHealthy_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// ResourceStoreServer is the server API for ResourceStore service.
// All implementations should embed UnimplementedResourceStoreServer
// for forward compatibility
@ -162,7 +150,6 @@ type ResourceStoreServer interface {
Delete(context.Context, *DeleteRequest) (*DeleteResponse, error)
List(context.Context, *ListRequest) (*ListResponse, error)
Watch(*WatchRequest, ResourceStore_WatchServer) error
IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
}
// UnimplementedResourceStoreServer should be embedded to have forward compatible implementations.
@ -187,9 +174,6 @@ func (UnimplementedResourceStoreServer) List(context.Context, *ListRequest) (*Li
func (UnimplementedResourceStoreServer) Watch(*WatchRequest, ResourceStore_WatchServer) error {
return status.Errorf(codes.Unimplemented, "method Watch not implemented")
}
func (UnimplementedResourceStoreServer) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method IsHealthy not implemented")
}
// 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
@ -313,24 +297,6 @@ func (x *resourceStoreWatchServer) Send(m *WatchResponse) error {
return x.ServerStream.SendMsg(m)
}
func _ResourceStore_IsHealthy_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(HealthCheckRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ResourceStoreServer).IsHealthy(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ResourceStore_IsHealthy_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ResourceStoreServer).IsHealthy(ctx, req.(*HealthCheckRequest))
}
return interceptor(ctx, in, info, handler)
}
// ResourceStore_ServiceDesc is the grpc.ServiceDesc for ResourceStore service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@ -358,10 +324,6 @@ var ResourceStore_ServiceDesc = grpc.ServiceDesc{
MethodName: "List",
Handler: _ResourceStore_List_Handler,
},
{
MethodName: "IsHealthy",
Handler: _ResourceStore_IsHealthy_Handler,
},
},
Streams: []grpc.StreamDesc{
{
@ -374,11 +336,10 @@ var ResourceStore_ServiceDesc = grpc.ServiceDesc{
}
const (
ResourceSearch_Read_FullMethodName = "/resource.ResourceSearch/Read"
ResourceSearch_GetBlob_FullMethodName = "/resource.ResourceSearch/GetBlob"
ResourceSearch_History_FullMethodName = "/resource.ResourceSearch/History"
ResourceSearch_Origin_FullMethodName = "/resource.ResourceSearch/Origin"
ResourceSearch_IsHealthy_FullMethodName = "/resource.ResourceSearch/IsHealthy"
ResourceSearch_Read_FullMethodName = "/resource.ResourceSearch/Read"
ResourceSearch_GetBlob_FullMethodName = "/resource.ResourceSearch/GetBlob"
ResourceSearch_History_FullMethodName = "/resource.ResourceSearch/History"
ResourceSearch_Origin_FullMethodName = "/resource.ResourceSearch/Origin"
)
// ResourceSearchClient is the client API for ResourceSearch service.
@ -395,8 +356,6 @@ type ResourceSearchClient interface {
History(ctx context.Context, in *HistoryRequest, opts ...grpc.CallOption) (*HistoryResponse, error)
// Used for efficient provisioning
Origin(ctx context.Context, in *OriginRequest, opts ...grpc.CallOption) (*OriginResponse, error)
// Check if the service is healthy
IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error)
}
type resourceSearchClient struct {
@ -447,16 +406,6 @@ func (c *resourceSearchClient) Origin(ctx context.Context, in *OriginRequest, op
return out, nil
}
func (c *resourceSearchClient) IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(HealthCheckResponse)
err := c.cc.Invoke(ctx, ResourceSearch_IsHealthy_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// ResourceSearchServer is the server API for ResourceSearch service.
// All implementations should embed UnimplementedResourceSearchServer
// for forward compatibility
@ -471,8 +420,6 @@ type ResourceSearchServer interface {
History(context.Context, *HistoryRequest) (*HistoryResponse, error)
// Used for efficient provisioning
Origin(context.Context, *OriginRequest) (*OriginResponse, error)
// Check if the service is healthy
IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
}
// UnimplementedResourceSearchServer should be embedded to have forward compatible implementations.
@ -491,9 +438,6 @@ func (UnimplementedResourceSearchServer) History(context.Context, *HistoryReques
func (UnimplementedResourceSearchServer) Origin(context.Context, *OriginRequest) (*OriginResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method Origin not implemented")
}
func (UnimplementedResourceSearchServer) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method IsHealthy not implemented")
}
// UnsafeResourceSearchServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ResourceSearchServer will
@ -578,24 +522,6 @@ func _ResourceSearch_Origin_Handler(srv interface{}, ctx context.Context, dec fu
return interceptor(ctx, in, info, handler)
}
func _ResourceSearch_IsHealthy_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(HealthCheckRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ResourceSearchServer).IsHealthy(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ResourceSearch_IsHealthy_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ResourceSearchServer).IsHealthy(ctx, req.(*HealthCheckRequest))
}
return interceptor(ctx, in, info, handler)
}
// ResourceSearch_ServiceDesc is the grpc.ServiceDesc for ResourceSearch service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@ -619,9 +545,102 @@ var ResourceSearch_ServiceDesc = grpc.ServiceDesc{
MethodName: "Origin",
Handler: _ResourceSearch_Origin_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "resource.proto",
}
const (
Diagnostics_IsHealthy_FullMethodName = "/resource.Diagnostics/IsHealthy"
)
// DiagnosticsClient is the client API for Diagnostics service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// Clients can use this service directly
// NOTE: This is read only, and no read afer write guarantees
type DiagnosticsClient interface {
// Check if the service is healthy
IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error)
}
type diagnosticsClient struct {
cc grpc.ClientConnInterface
}
func NewDiagnosticsClient(cc grpc.ClientConnInterface) DiagnosticsClient {
return &diagnosticsClient{cc}
}
func (c *diagnosticsClient) IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(HealthCheckResponse)
err := c.cc.Invoke(ctx, Diagnostics_IsHealthy_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// DiagnosticsServer is the server API for Diagnostics service.
// All implementations should embed UnimplementedDiagnosticsServer
// for forward compatibility
//
// Clients can use this service directly
// NOTE: This is read only, and no read afer write guarantees
type DiagnosticsServer interface {
// Check if the service is healthy
IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
}
// UnimplementedDiagnosticsServer should be embedded to have forward compatible implementations.
type UnimplementedDiagnosticsServer struct {
}
func (UnimplementedDiagnosticsServer) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method IsHealthy not implemented")
}
// 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
// result in compilation errors.
type UnsafeDiagnosticsServer interface {
mustEmbedUnimplementedDiagnosticsServer()
}
func RegisterDiagnosticsServer(s grpc.ServiceRegistrar, srv DiagnosticsServer) {
s.RegisterService(&Diagnostics_ServiceDesc, srv)
}
func _Diagnostics_IsHealthy_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(HealthCheckRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(DiagnosticsServer).IsHealthy(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Diagnostics_IsHealthy_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(DiagnosticsServer).IsHealthy(ctx, req.(*HealthCheckRequest))
}
return interceptor(ctx, in, info, handler)
}
// Diagnostics_ServiceDesc is the grpc.ServiceDesc for Diagnostics service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var Diagnostics_ServiceDesc = grpc.ServiceDesc{
ServiceName: "resource.Diagnostics",
HandlerType: (*DiagnosticsServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "IsHealthy",
Handler: _ResourceSearch_IsHealthy_Handler,
Handler: _Diagnostics_IsHealthy_Handler,
},
},
Streams: []grpc.StreamDesc{},

View File

@ -1,185 +0,0 @@
package resource
import (
"context"
"encoding/json"
"fmt"
"path/filepath"
"sort"
"strings"
"github.com/hack-pad/hackpadfs"
"github.com/hack-pad/hackpadfs/mem"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime/schema"
)
type FileSystemStoreOptions struct {
// OTel tracer
Tracer trace.Tracer
// Get the next EventID. When not set, this will default to snowflake IDs
NextEventID func() int64
// Root file system -- null will be in memory
Root hackpadfs.FS
}
func NewFSStore(opts FileSystemStoreOptions) (ResourceStoreServer, error) {
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("testing")
}
var err error
root := opts.Root
if root == nil {
root, err = mem.NewFS()
if err != nil {
return nil, err
}
}
store := &fsStore{root: root}
store.writer, err = NewResourceWriter(WriterOptions{
Tracer: opts.Tracer,
Reader: store.Read,
Appender: store.append,
})
return store, err
}
var _ ResourceStoreServer = &fsStore{}
type fsStore struct {
writer ResourceWriter
root hackpadfs.FS
}
type fsEvent struct {
ResourceVersion int64 `json:"resourceVersion"`
Message string `json:"message,omitempty"`
Operation string `json:"operation,omitempty"`
Value json.RawMessage `json:"value,omitempty"`
BlobPath string `json:"blob,omitempty"`
}
// The only write command
func (f *fsStore) append(ctx context.Context, event *WriteEvent) (int64, error) {
body := fsEvent{
ResourceVersion: event.EventID,
Message: event.Message,
Operation: event.Operation.String(),
Value: event.Value,
// Blob...
}
// For this case, we will treat them the same
event.Key.ResourceVersion = 0
dir := event.Key.NamespacedPath()
err := hackpadfs.MkdirAll(f.root, dir, 0750)
if err != nil {
return 0, err
}
bytes, err := json.Marshal(&body)
if err != nil {
return 0, err
}
fpath := filepath.Join(dir, fmt.Sprintf("%d.json", event.EventID))
file, err := hackpadfs.OpenFile(f.root, fpath, hackpadfs.FlagWriteOnly|hackpadfs.FlagCreate, 0750)
if err != nil {
return 0, err
}
_, err = hackpadfs.WriteFile(file, bytes)
return event.EventID, err
}
func (f *fsStore) Create(ctx context.Context, req *CreateRequest) (*CreateResponse, error) {
return f.writer.Create(ctx, req)
}
// Update implements ResourceStoreServer.
func (f *fsStore) Update(ctx context.Context, req *UpdateRequest) (*UpdateResponse, error) {
return f.writer.Update(ctx, req)
}
// Delete implements ResourceStoreServer.
func (f *fsStore) Delete(ctx context.Context, req *DeleteRequest) (*DeleteResponse, error) {
return f.writer.Delete(ctx, req)
}
// Read implements ResourceStoreServer.
func (f *fsStore) Read(ctx context.Context, req *ReadRequest) (*ReadResponse, error) {
rv := req.Key.ResourceVersion
req.Key.ResourceVersion = 0
fname := "--x--"
dir := req.Key.NamespacedPath()
if rv > 0 {
fname = fmt.Sprintf("%d.json", rv)
} else {
files, err := hackpadfs.ReadDir(f.root, dir)
if err != nil {
return nil, err
}
// Sort by name
sort.Slice(files, func(i, j int) bool {
a := files[i].Name()
b := files[j].Name()
return a > b // ?? should we parse the numbers ???
})
// The first matching file
for _, v := range files {
fname = v.Name()
if strings.HasSuffix(fname, ".json") {
break
}
}
}
evt, err := f.open(filepath.Join(dir, fname))
if err != nil || evt.Operation == ResourceOperation_DELETED.String() {
return nil, apierrors.NewNotFound(schema.GroupResource{
Group: req.Key.Group,
Resource: req.Key.Resource,
}, req.Key.Name)
}
return &ReadResponse{
ResourceVersion: evt.ResourceVersion,
Value: evt.Value,
Message: evt.Message,
}, nil
}
func (f *fsStore) open(p string) (*fsEvent, error) {
raw, err := hackpadfs.ReadFile(f.root, p)
if err != nil {
return nil, err
}
evt := &fsEvent{}
err = json.Unmarshal(raw, evt)
return evt, err
}
// IsHealthy implements ResourceStoreServer.
func (f *fsStore) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return &HealthCheckResponse{Status: HealthCheckResponse_SERVING}, nil
}
// List implements ResourceStoreServer.
func (f *fsStore) List(ctx context.Context, req *ListRequest) (*ListResponse, error) {
panic("unimplemented")
}
// Watch implements ResourceStoreServer.
func (f *fsStore) Watch(*WatchRequest, ResourceStore_WatchServer) error {
panic("unimplemented")
}

View File

@ -1,336 +0,0 @@
package resource
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/bwmarrin/snowflake"
"github.com/prometheus/client_golang/prometheus"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"go.opentelemetry.io/otel/trace"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/apimachinery/utils"
)
// Package-level errors.
var (
ErrNotFound = errors.New("entity not found")
ErrOptimisticLockingFailed = errors.New("optimistic locking failed")
ErrUserNotFoundInContext = errors.New("user not found in context")
ErrUnableToReadResourceJSON = errors.New("unable to read resource json")
ErrNextPageTokenNotSupported = errors.New("nextPageToken not yet supported")
ErrLimitNotSupported = errors.New("limit not yet supported")
ErrNotImplementedYet = errors.New("not implemented yet")
)
// Resource writer support
type ResourceWriter interface {
Create(context.Context, *CreateRequest) (*CreateResponse, error)
Update(context.Context, *UpdateRequest) (*UpdateResponse, error)
Delete(context.Context, *DeleteRequest) (*DeleteResponse, error)
}
type WriterOptions struct {
// OTel tracer
Tracer trace.Tracer
// When running in a cluster, each node should have a different ID
// This is used for snowflake generation and log identification
NodeID int64
// Read an individual item
Reader func(context.Context, *ReadRequest) (*ReadResponse, error)
// Add a validated write event
Appender EventAppender
// Get the next EventID. When not set, this will default to snowflake IDs
NextEventID func() int64
// Check if a user has access to write folders
// When this is nil, no resources can have folders configured
FolderAccess func(ctx context.Context, user identity.Requester, uid string) bool
// When configured, this will make sure a user is allowed to save to a given origin
OriginAccess func(ctx context.Context, user identity.Requester, origin string) bool
}
func NewResourceWriter(opts WriterOptions) (ResourceWriter, error) {
log := slog.Default().With("logger", "resource-writer")
if err := prometheus.Register(NewStorageMetrics()); err != nil {
log.Warn("error registering storage server metrics", "error", err)
}
if opts.NextEventID == nil {
eventNode, err := snowflake.NewNode(opts.NodeID)
if err != nil {
return nil, apierrors.NewInternalError(
fmt.Errorf("error initializing snowflake id generator :: %w", err))
}
opts.NextEventID = func() int64 {
return eventNode.Generate().Int64()
}
}
return &writeServer{
log: log,
opts: opts,
}, nil
}
type writeServer struct {
log *slog.Logger
opts WriterOptions
}
func (s *writeServer) newEvent(ctx context.Context, key *ResourceKey, value, oldValue []byte) (*WriteEvent, error) {
var err error
event := &WriteEvent{
EventID: s.opts.NextEventID(),
Key: key,
Value: value,
}
event.Requester, err = identity.GetRequester(ctx)
if err != nil {
return nil, ErrUserNotFoundInContext
}
dummy := &metav1.PartialObjectMetadata{}
err = json.Unmarshal(value, dummy)
if err != nil {
return nil, ErrUnableToReadResourceJSON
}
obj, err := utils.MetaAccessor(dummy)
if err != nil {
return nil, apierrors.NewBadRequest("invalid object in json")
}
if obj.GetUID() == "" {
return nil, apierrors.NewBadRequest("the UID must be set")
}
if obj.GetGenerateName() != "" {
return nil, apierrors.NewBadRequest("can not save value with generate name")
}
gvk := obj.GetGroupVersionKind()
if gvk.Kind == "" {
return nil, apierrors.NewBadRequest("expecting resources with a kind in the body")
}
if gvk.Version == "" {
return nil, apierrors.NewBadRequest("expecting resources with an apiVersion")
}
if gvk.Group != "" && gvk.Group != key.Group {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("group in key does not match group in the body (%s != %s)", key.Group, gvk.Group),
)
}
if obj.GetName() != key.Name {
return nil, apierrors.NewBadRequest("key name does not match the name in the body")
}
if obj.GetNamespace() != key.Namespace {
return nil, apierrors.NewBadRequest("key namespace does not match the namespace in the body")
}
folder := obj.GetFolder()
if folder != "" {
if s.opts.FolderAccess == nil {
return nil, apierrors.NewBadRequest("folders are not supported")
} else if !s.opts.FolderAccess(ctx, event.Requester, folder) {
return nil, apierrors.NewBadRequest("unable to add resource to folder") // 403?
}
}
origin, err := obj.GetOriginInfo()
if err != nil {
return nil, apierrors.NewBadRequest("invalid origin info")
}
if origin != nil && s.opts.OriginAccess != nil {
if !s.opts.OriginAccess(ctx, event.Requester, origin.Name) {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("not allowed to write resource to origin (%s)", origin.Name))
}
}
event.Object = obj
// This is an update
if oldValue != nil {
dummy := &metav1.PartialObjectMetadata{}
err = json.Unmarshal(oldValue, dummy)
if err != nil {
return nil, apierrors.NewBadRequest("error reading old json value")
}
old, err := utils.MetaAccessor(dummy)
if err != nil {
return nil, apierrors.NewBadRequest("invalid object inside old json")
}
if key.Name != old.GetName() {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("the old value has a different name (%s != %s)", key.Name, old.GetName()))
}
// Can not change creation timestamps+user
if obj.GetCreatedBy() != old.GetCreatedBy() {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("can not change the created by metadata (%s != %s)", obj.GetCreatedBy(), old.GetCreatedBy()))
}
if obj.GetCreationTimestamp() != old.GetCreationTimestamp() {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("can not change the CreationTimestamp metadata (%v != %v)", obj.GetCreationTimestamp(), old.GetCreationTimestamp()))
}
oldFolder := obj.GetFolder()
if oldFolder != folder {
event.FolderChanged = true
}
event.OldObject = old
} else if folder != "" {
event.FolderChanged = true
}
return event, nil
}
func (s *writeServer) Create(ctx context.Context, req *CreateRequest) (*CreateResponse, error) {
ctx, span := s.opts.Tracer.Start(ctx, "storage_server.Create")
defer span.End()
if req.Key.ResourceVersion > 0 {
return nil, apierrors.NewBadRequest("can not update a specific resource version")
}
event, err := s.newEvent(ctx, req.Key, req.Value, nil)
if err != nil {
return nil, err
}
event.Operation = ResourceOperation_CREATED
event.Blob = req.Blob
event.Message = req.Message
rsp := &CreateResponse{}
// Make sure the created by user is accurate
//----------------------------------------
val := event.Object.GetCreatedBy()
if val != "" && val != event.Requester.GetUID().String() {
return nil, apierrors.NewBadRequest("created by annotation does not match: metadata.annotations#" + utils.AnnoKeyCreatedBy)
}
// Create can not have updated properties
//----------------------------------------
if event.Object.GetUpdatedBy() != "" {
return nil, apierrors.NewBadRequest("unexpected metadata.annotations#" + utils.AnnoKeyCreatedBy)
}
ts, err := event.Object.GetUpdatedTimestamp()
if err != nil {
return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid timestamp: %s", err))
}
if ts != nil {
return nil, apierrors.NewBadRequest("unexpected metadata.annotations#" + utils.AnnoKeyUpdatedTimestamp)
}
// Append and set the resource version
rsp.ResourceVersion, err = s.opts.Appender(ctx, event)
// ?? convert the error to status?
return rsp, err
}
func (s *writeServer) Update(ctx context.Context, req *UpdateRequest) (*UpdateResponse, error) {
ctx, span := s.opts.Tracer.Start(ctx, "storage_server.Update")
defer span.End()
rsp := &UpdateResponse{}
if req.Key.ResourceVersion < 0 {
return nil, apierrors.NewBadRequest("update must include the previous version")
}
latest, err := s.opts.Reader(ctx, &ReadRequest{
Key: req.Key.WithoutResourceVersion(),
})
if err != nil {
return nil, err
}
if latest.Value == nil {
return nil, apierrors.NewBadRequest("current value does not exist")
}
event, err := s.newEvent(ctx, req.Key, req.Value, latest.Value)
if err != nil {
return nil, err
}
event.Operation = ResourceOperation_UPDATED
event.PreviousRV = latest.ResourceVersion
event.Message = req.Message
// Make sure the update user is accurate
//----------------------------------------
val := event.Object.GetUpdatedBy()
if val != "" && val != event.Requester.GetUID().String() {
return nil, apierrors.NewBadRequest("updated by annotation does not match: metadata.annotations#" + utils.AnnoKeyUpdatedBy)
}
rsp.ResourceVersion, err = s.opts.Appender(ctx, event)
return rsp, err
}
func (s *writeServer) Delete(ctx context.Context, req *DeleteRequest) (*DeleteResponse, error) {
ctx, span := s.opts.Tracer.Start(ctx, "storage_server.Delete")
defer span.End()
rsp := &DeleteResponse{}
if req.Key.ResourceVersion < 0 {
return nil, apierrors.NewBadRequest("update must include the previous version")
}
latest, err := s.opts.Reader(ctx, &ReadRequest{
Key: req.Key.WithoutResourceVersion(),
})
if err != nil {
return nil, err
}
if latest.ResourceVersion != req.Key.ResourceVersion {
return nil, ErrOptimisticLockingFailed
}
now := metav1.NewTime(time.Now())
event := &WriteEvent{
EventID: s.opts.NextEventID(),
Key: req.Key,
Operation: ResourceOperation_DELETED,
PreviousRV: latest.ResourceVersion,
}
event.Requester, err = identity.GetRequester(ctx)
if err != nil {
return nil, apierrors.NewBadRequest("unable to get user")
}
marker := &DeletedMarker{}
err = json.Unmarshal(latest.Value, marker)
if err != nil {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("unable to read previous object, %v", err))
}
event.Object, err = utils.MetaAccessor(marker)
if err != nil {
return nil, err
}
event.Object.SetDeletionTimestamp(&now)
event.Object.SetUpdatedTimestamp(&now.Time)
event.Object.SetManagedFields(nil)
event.Object.SetFinalizers(nil)
event.Object.SetUpdatedBy(event.Requester.GetUID().String())
marker.TypeMeta = metav1.TypeMeta{
Kind: "DeletedMarker",
APIVersion: "storage.grafana.app/v0alpha1", // ?? or can we stick this in common?
}
marker.Annotations["RestoreResourceVersion"] = fmt.Sprintf("%d", event.PreviousRV)
event.Value, err = json.Marshal(marker)
if err != nil {
return nil, apierrors.NewBadRequest(
fmt.Sprintf("unable creating deletion marker, %v", err))
}
rsp.ResourceVersion, err = s.opts.Appender(ctx, event)
return rsp, err
}

View File

@ -1,111 +0,0 @@
package resource
import (
"context"
"embed"
"encoding/json"
"fmt"
"os"
"testing"
"time"
"github.com/hack-pad/hackpadfs"
hackos "github.com/hack-pad/hackpadfs/os"
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/apimachinery/utils"
)
func TestWriter(t *testing.T) {
testUserA := &identity.StaticRequester{
Namespace: identity.NamespaceUser,
UserID: 123,
UserUID: "u123",
OrgRole: identity.RoleAdmin,
IsGrafanaAdmin: true, // can do anything
}
ctx := identity.WithRequester(context.Background(), testUserA)
var root hackpadfs.FS
if false {
tmp, err := os.MkdirTemp("", "xxx-*")
require.NoError(t, err)
root, err = hackos.NewFS().Sub(tmp[1:])
require.NoError(t, err)
fmt.Printf("ROOT: %s\n\n", tmp)
}
store, err := NewFSStore(FileSystemStoreOptions{
Root: root,
})
require.NoError(t, err)
t.Run("playlist happy CRUD paths", func(t *testing.T) {
raw := testdata(t, "01_create_playlist.json")
key := &ResourceKey{
Group: "playlist.grafana.app",
Resource: "rrrr", // can be anything :(
Namespace: "default",
Name: "fdgsv37qslr0ga",
}
created, err := store.Create(ctx, &CreateRequest{
Value: raw,
Key: key,
})
require.NoError(t, err)
require.True(t, created.ResourceVersion > 0)
// The key does not include resource version
found, err := store.Read(ctx, &ReadRequest{Key: key})
require.NoError(t, err)
require.Equal(t, created.ResourceVersion, found.ResourceVersion)
// Now update the value
tmp := &unstructured.Unstructured{}
err = json.Unmarshal(raw, tmp)
require.NoError(t, err)
now := time.Now().UnixMilli()
obj, err := utils.MetaAccessor(tmp)
require.NoError(t, err)
obj.SetAnnotation("test", "hello")
obj.SetUpdatedTimestampMillis(now)
obj.SetUpdatedBy(testUserA.GetUID().String())
raw, err = json.Marshal(tmp)
require.NoError(t, err)
key.ResourceVersion = created.ResourceVersion
updated, err := store.Update(ctx, &UpdateRequest{Key: key, Value: raw})
require.NoError(t, err)
require.True(t, updated.ResourceVersion > created.ResourceVersion)
// We should still get the latest
key.ResourceVersion = 0
found, err = store.Read(ctx, &ReadRequest{Key: key})
require.NoError(t, err)
require.Equal(t, updated.ResourceVersion, found.ResourceVersion)
key.ResourceVersion = updated.ResourceVersion
deleted, err := store.Delete(ctx, &DeleteRequest{Key: key})
require.NoError(t, err)
require.True(t, deleted.ResourceVersion > updated.ResourceVersion)
// We should get not found when trying to read the latest value
key.ResourceVersion = 0
found, err = store.Read(ctx, &ReadRequest{Key: key})
require.Error(t, err)
require.Nil(t, found)
})
}
//go:embed testdata/*
var testdataFS embed.FS
func testdata(t *testing.T, filename string) []byte {
t.Helper()
b, err := testdataFS.ReadFile(`testdata/` + filename)
require.NoError(t, err)
return b
}

View File

@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"strings"
"sync"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel/trace"
@ -25,47 +24,31 @@ var (
ErrNotImplementedYet = errors.New("not implemented yet")
)
// Make sure we implement both store and search
var _ resource.ResourceStoreServer = &sqlResourceServer{}
var _ resource.ResourceSearchServer = &sqlResourceServer{}
func ProvideSQLResourceServer(db db.EntityDBInterface, tracer tracing.Tracer) (SqlResourceServer, error) {
func ProvideSQLResourceServer(db db.EntityDBInterface, tracer tracing.Tracer) (resource.ResourceServer, error) {
ctx, cancel := context.WithCancel(context.Background())
var err error
server := &sqlResourceServer{
store := &sqlResourceStore{
db: db,
log: log.New("sql-resource-server"),
ctx: ctx,
cancel: cancel,
tracer: tracer,
}
server.writer, err = resource.NewResourceWriter(resource.WriterOptions{
NodeID: 123, // for snowflake ID generation
Tracer: tracer,
Reader: server.Read,
Appender: server.append,
})
if err != nil {
if err := prometheus.Register(sqlstash.NewStorageMetrics()); err != nil {
return nil, err
}
if err := prometheus.Register(sqlstash.NewStorageMetrics()); err != nil {
server.log.Warn("error registering storage server metrics", "error", err)
}
return server, nil
return resource.NewResourceServer(resource.ResourceServerOptions{
Tracer: tracer,
Store: store,
NodeID: 234, // from config? used for snowflake ID
Diagnostics: store,
Lifecycle: store,
})
}
type SqlResourceServer interface {
resource.ResourceStoreServer
resource.ResourceSearchServer
Init() error
Stop()
}
type sqlResourceServer struct {
type sqlResourceStore struct {
log log.Logger
db db.EntityDBInterface // needed to keep xorm engine in scope
sess *session.SessionDB
@ -76,29 +59,11 @@ type sqlResourceServer struct {
stream chan *resource.WatchResponse
tracer trace.Tracer
// Wrapper around all write events
writer resource.ResourceWriter
once sync.Once
initErr error
sqlDB db.DB
sqlDialect sqltemplate.Dialect
}
func (s *sqlResourceServer) Init() error {
s.once.Do(func() {
s.initErr = s.init()
})
if s.initErr != nil {
return fmt.Errorf("initialize Entity Server: %w", s.initErr)
}
return s.initErr
}
func (s *sqlResourceServer) init() error {
func (s *sqlResourceStore) Init() error {
if s.sess != nil {
return nil
}
@ -143,15 +108,6 @@ func (s *sqlResourceServer) init() error {
s.sess = sess
s.dialect = migrator.NewDialect(engine.DriverName())
s.writer, err = resource.NewResourceWriter(resource.WriterOptions{
NodeID: 10,
Tracer: s.tracer,
Reader: s.Read,
Appender: s.append,
})
if err != nil {
return err
}
// set up the broadcaster
s.broadcaster, err = sqlstash.NewBroadcaster(s.ctx, func(stream chan *resource.WatchResponse) error {
@ -169,12 +125,8 @@ func (s *sqlResourceServer) init() error {
return nil
}
func (s *sqlResourceServer) IsHealthy(ctx context.Context, r *resource.HealthCheckRequest) (*resource.HealthCheckResponse, error) {
ctxLogger := s.log.FromContext(log.WithContextualAttributes(ctx, []any{"method", "isHealthy"}))
if err := s.Init(); err != nil {
ctxLogger.Error("init error", "error", err)
return nil, err
}
func (s *sqlResourceStore) IsHealthy(ctx context.Context, r *resource.HealthCheckRequest) (*resource.HealthCheckResponse, error) {
// ctxLogger := s.log.FromContext(log.WithContextualAttributes(ctx, []any{"method", "isHealthy"}))
if err := s.sqlDB.PingContext(ctx); err != nil {
return nil, err
@ -183,11 +135,11 @@ func (s *sqlResourceServer) IsHealthy(ctx context.Context, r *resource.HealthChe
return &resource.HealthCheckResponse{Status: resource.HealthCheckResponse_SERVING}, nil
}
func (s *sqlResourceServer) Stop() {
func (s *sqlResourceStore) Stop() {
s.cancel()
}
func (s *sqlResourceServer) append(ctx context.Context, event *resource.WriteEvent) (int64, error) {
func (s *sqlResourceStore) WriteEvent(ctx context.Context, event *resource.WriteEvent) (int64, error) {
_, span := s.tracer.Start(ctx, "storage_server.WriteEvent")
defer span.End()
@ -196,14 +148,10 @@ func (s *sqlResourceServer) append(ctx context.Context, event *resource.WriteEve
return 0, ErrNotImplementedYet
}
func (s *sqlResourceServer) Read(ctx context.Context, req *resource.ReadRequest) (*resource.ReadResponse, error) {
func (s *sqlResourceStore) Read(ctx context.Context, req *resource.ReadRequest) (*resource.ReadResponse, error) {
_, span := s.tracer.Start(ctx, "storage_server.GetResource")
defer span.End()
if err := s.Init(); err != nil {
return nil, err
}
if req.Key.Group == "" {
return &resource.ReadResponse{Status: badRequest("missing group")}, nil
}
@ -216,92 +164,40 @@ func (s *sqlResourceServer) Read(ctx context.Context, req *resource.ReadRequest)
return nil, ErrNotImplementedYet
}
func (s *sqlResourceServer) Create(ctx context.Context, req *resource.CreateRequest) (*resource.CreateResponse, error) {
rsp, err := s.writer.Create(ctx, req)
if err != nil {
s.log.Info("create", "error", err)
rsp.Status = &resource.StatusResult{
Status: "Failure",
Message: err.Error(),
}
}
return rsp, nil
}
func (s *sqlResourceServer) Update(ctx context.Context, req *resource.UpdateRequest) (*resource.UpdateResponse, error) {
rsp, err := s.writer.Update(ctx, req)
if err != nil {
s.log.Info("create", "error", err)
rsp.Status = &resource.StatusResult{
Status: "Failure",
Message: err.Error(),
}
}
return rsp, nil
}
func (s *sqlResourceServer) Delete(ctx context.Context, req *resource.DeleteRequest) (*resource.DeleteResponse, error) {
rsp, err := s.writer.Delete(ctx, req)
if err != nil {
s.log.Info("create", "error", err)
rsp.Status = &resource.StatusResult{
Status: "Failure",
Message: err.Error(),
}
}
return rsp, nil
}
func (s *sqlResourceServer) List(ctx context.Context, req *resource.ListRequest) (*resource.ListResponse, error) {
func (s *sqlResourceStore) List(ctx context.Context, req *resource.ListRequest) (*resource.ListResponse, error) {
_, span := s.tracer.Start(ctx, "storage_server.List")
defer span.End()
if err := s.Init(); err != nil {
return nil, err
}
fmt.Printf("TODO, LIST: %+v", req.Options.Key)
return nil, ErrNotImplementedYet
}
// Get the raw blob bytes and metadata
func (s *sqlResourceServer) GetBlob(ctx context.Context, req *resource.GetBlobRequest) (*resource.GetBlobResponse, error) {
func (s *sqlResourceStore) GetBlob(ctx context.Context, req *resource.GetBlobRequest) (*resource.GetBlobResponse, error) {
_, span := s.tracer.Start(ctx, "storage_server.List")
defer span.End()
if err := s.Init(); err != nil {
return nil, err
}
fmt.Printf("TODO, GET BLOB: %+v", req.Key)
return nil, ErrNotImplementedYet
}
// Show resource history (and trash)
func (s *sqlResourceServer) History(ctx context.Context, req *resource.HistoryRequest) (*resource.HistoryResponse, error) {
func (s *sqlResourceStore) History(ctx context.Context, req *resource.HistoryRequest) (*resource.HistoryResponse, error) {
_, span := s.tracer.Start(ctx, "storage_server.History")
defer span.End()
if err := s.Init(); err != nil {
return nil, err
}
fmt.Printf("TODO, GET History: %+v", req.Key)
return nil, ErrNotImplementedYet
}
// Used for efficient provisioning
func (s *sqlResourceServer) Origin(ctx context.Context, req *resource.OriginRequest) (*resource.OriginResponse, error) {
func (s *sqlResourceStore) Origin(ctx context.Context, req *resource.OriginRequest) (*resource.OriginResponse, error) {
_, span := s.tracer.Start(ctx, "storage_server.History")
defer span.End()
if err := s.Init(); err != nil {
return nil, err
}
fmt.Printf("TODO, GET History: %+v", req.Key)
return nil, ErrNotImplementedYet

View File

@ -7,11 +7,11 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/resource"
)
func (s *sqlResourceServer) Watch(*resource.WatchRequest, resource.ResourceStore_WatchServer) error {
func (s *sqlResourceStore) Watch(*resource.WatchRequest, resource.ResourceStore_WatchServer) error {
return ErrNotImplementedYet
}
func (s *sqlResourceServer) poller(stream chan *resource.WatchResponse) {
func (s *sqlResourceStore) poller(stream chan *resource.WatchResponse) {
var err error
since := int64(0)
@ -34,7 +34,7 @@ func (s *sqlResourceServer) poller(stream chan *resource.WatchResponse) {
}
}
func (s *sqlResourceServer) poll(since int64, out chan *resource.WatchResponse) (int64, error) {
func (s *sqlResourceStore) poll(since int64, out chan *resource.WatchResponse) (int64, error) {
ctx, span := s.tracer.Start(s.ctx, "storage_server.poll")
defer span.End()
ctxLogger := s.log.FromContext(log.WithContextualAttributes(ctx, []any{"method", "poll"}))