From 4a41f7d0ddc199aac52e94dbc03d044699af7eb6 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Fri, 14 Jun 2024 14:24:26 +0300 Subject: [PATCH] now with a base server implementation --- pkg/storage/unified/README.md | 9 + pkg/storage/unified/resource/resource.pb.go | 96 +++-- pkg/storage/unified/resource/resource.proto | 7 +- .../unified/resource/resource_grpc.pb.go | 189 +++++----- pkg/storage/unified/resource/store.go | 185 ---------- pkg/storage/unified/resource/writer.go | 336 ------------------ pkg/storage/unified/resource/writer_test.go | 111 ------ .../unified/sqlstash/sql_storage_server.go | 148 ++------ pkg/storage/unified/sqlstash/watch.go | 6 +- 9 files changed, 188 insertions(+), 899 deletions(-) delete mode 100644 pkg/storage/unified/resource/store.go delete mode 100644 pkg/storage/unified/resource/writer.go delete mode 100644 pkg/storage/unified/resource/writer_test.go diff --git a/pkg/storage/unified/README.md b/pkg/storage/unified/README.md index be8728f366a..997bab581aa 100644 --- a/pkg/storage/unified/README.md +++ b/pkg/storage/unified/README.md @@ -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 + + + + diff --git a/pkg/storage/unified/resource/resource.pb.go b/pkg/storage/unified/resource/resource.pb.go index b3130e39486..2bd8b602202 100644 --- a/pkg/storage/unified/resource/resource.pb.go +++ b/pkg/storage/unified/resource/resource.pb.go @@ -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, diff --git a/pkg/storage/unified/resource/resource.proto b/pkg/storage/unified/resource/resource.proto index 646ab1f54dd..2c4f38c7b44 100644 --- a/pkg/storage/unified/resource/resource.proto +++ b/pkg/storage/unified/resource/resource.proto @@ -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); } diff --git a/pkg/storage/unified/resource/resource_grpc.pb.go b/pkg/storage/unified/resource/resource_grpc.pb.go index 03aa2d702e1..bde23327064 100644 --- a/pkg/storage/unified/resource/resource_grpc.pb.go +++ b/pkg/storage/unified/resource/resource_grpc.pb.go @@ -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{}, diff --git a/pkg/storage/unified/resource/store.go b/pkg/storage/unified/resource/store.go deleted file mode 100644 index 41a9d6ca254..00000000000 --- a/pkg/storage/unified/resource/store.go +++ /dev/null @@ -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") -} diff --git a/pkg/storage/unified/resource/writer.go b/pkg/storage/unified/resource/writer.go deleted file mode 100644 index f0fe38420e9..00000000000 --- a/pkg/storage/unified/resource/writer.go +++ /dev/null @@ -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 -} diff --git a/pkg/storage/unified/resource/writer_test.go b/pkg/storage/unified/resource/writer_test.go deleted file mode 100644 index 57eec433156..00000000000 --- a/pkg/storage/unified/resource/writer_test.go +++ /dev/null @@ -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 -} diff --git a/pkg/storage/unified/sqlstash/sql_storage_server.go b/pkg/storage/unified/sqlstash/sql_storage_server.go index 403f964775b..5f4a5d0fc16 100644 --- a/pkg/storage/unified/sqlstash/sql_storage_server.go +++ b/pkg/storage/unified/sqlstash/sql_storage_server.go @@ -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 diff --git a/pkg/storage/unified/sqlstash/watch.go b/pkg/storage/unified/sqlstash/watch.go index 3e98c5aa2a6..9417b81d166 100644 --- a/pkg/storage/unified/sqlstash/watch.go +++ b/pkg/storage/unified/sqlstash/watch.go @@ -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"}))