Skip to content

Commit b4b1af6

Browse files
pthirunpthirun
committed
[Controller] Add gRPC support for getStore API
Add gRPC support for the getStore endpoint while maintaining backward compatibility with the existing HTTP endpoint. Changes: - Add getStore RPC to StoreGrpcService.proto - Add handler method to StoreRequestHandler - Add gRPC service implementation - Update HTTP route to delegate to shared handler Tests: - Unit tests for success, error, invalid input - Integration test for full lifecycle
1 parent 8109fcd commit b4b1af6

9 files changed

Lines changed: 290 additions & 39 deletions

File tree

‎internal/venice-common/src/main/proto/controller/StoreGrpcService.proto‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ service StoreGrpcService {
1414
rpc checkResourceCleanupForStoreCreation(ClusterStoreGrpcInfo) returns (ResourceCleanupCheckGrpcResponse) {}
1515
rpc validateStoreDeleted(ValidateStoreDeletedGrpcRequest) returns (ValidateStoreDeletedGrpcResponse);
1616
rpc listStores(ListStoresGrpcRequest) returns (ListStoresGrpcResponse);
17+
rpc getStore(GetStoreGrpcRequest) returns (GetStoreGrpcResponse);
1718
}
1819

1920
message CreateStoreGrpcRequest {
@@ -82,4 +83,13 @@ message ListStoresGrpcRequest {
8283
message ListStoresGrpcResponse {
8384
string clusterName = 1;
8485
repeated string storeNames = 2;
86+
}
87+
88+
message GetStoreGrpcRequest {
89+
ClusterStoreGrpcInfo storeInfo = 1;
90+
}
91+
92+
message GetStoreGrpcResponse {
93+
ClusterStoreGrpcInfo storeInfo = 1;
94+
string storeInfoJson = 2;
8595
}

‎internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestControllerGrpcEndpoints.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import com.linkedin.venice.protocols.controller.CreateStoreGrpcResponse;
2121
import com.linkedin.venice.protocols.controller.DiscoverClusterGrpcRequest;
2222
import com.linkedin.venice.protocols.controller.DiscoverClusterGrpcResponse;
23+
import com.linkedin.venice.protocols.controller.GetStoreGrpcRequest;
24+
import com.linkedin.venice.protocols.controller.GetStoreGrpcResponse;
2325
import com.linkedin.venice.protocols.controller.LeaderControllerGrpcRequest;
2426
import com.linkedin.venice.protocols.controller.LeaderControllerGrpcResponse;
2527
import com.linkedin.venice.protocols.controller.ListStoresGrpcRequest;
@@ -337,6 +339,41 @@ public void testListStoresGrpcEndpoint() {
337339
}
338340
}
339341

342+
@Test(timeOut = TIMEOUT_MS)
343+
public void testGetStoreGrpcEndpoint() {
344+
String storeName = Utils.getUniqueString("test_get_store");
345+
String controllerGrpcUrl = veniceCluster.getLeaderVeniceController().getControllerGrpcUrl();
346+
ManagedChannel channel = Grpc.newChannelBuilder(controllerGrpcUrl, InsecureChannelCredentials.create()).build();
347+
StoreGrpcServiceGrpc.StoreGrpcServiceBlockingStub storeBlockingStub = StoreGrpcServiceGrpc.newBlockingStub(channel);
348+
349+
ClusterStoreGrpcInfo storeGrpcInfo = ClusterStoreGrpcInfo.newBuilder()
350+
.setClusterName(veniceCluster.getClusterName())
351+
.setStoreName(storeName)
352+
.build();
353+
354+
// Step 1: Create the store first
355+
CreateStoreGrpcRequest createStoreGrpcRequest = CreateStoreGrpcRequest.newBuilder()
356+
.setStoreInfo(storeGrpcInfo)
357+
.setOwner("owner")
358+
.setKeySchema(DEFAULT_KEY_SCHEMA)
359+
.setValueSchema("\"string\"")
360+
.build();
361+
CreateStoreGrpcResponse createResponse = storeBlockingStub.createStore(createStoreGrpcRequest);
362+
assertNotNull(createResponse, "Create response should not be null");
363+
assertEquals(createResponse.getStoreInfo().getStoreName(), storeName);
364+
365+
// Step 2: Get store info via gRPC
366+
GetStoreGrpcRequest getStoreRequest = GetStoreGrpcRequest.newBuilder().setStoreInfo(storeGrpcInfo).build();
367+
GetStoreGrpcResponse getStoreResponse = storeBlockingStub.getStore(getStoreRequest);
368+
369+
assertNotNull(getStoreResponse, "Response should not be null");
370+
assertEquals(getStoreResponse.getStoreInfo().getStoreName(), storeName);
371+
assertEquals(getStoreResponse.getStoreInfo().getClusterName(), veniceCluster.getClusterName());
372+
assertNotNull(getStoreResponse.getStoreInfoJson(), "Store info JSON should not be null");
373+
assertTrue(getStoreResponse.getStoreInfoJson().contains(storeName), "Store info JSON should contain store name");
374+
assertTrue(getStoreResponse.getStoreInfoJson().contains("owner"), "Store info JSON should contain owner");
375+
}
376+
340377
private static class MockDynamicAccessController extends NoOpDynamicAccessController {
341378
private final Set<String> resourcesInAllowList = ConcurrentHashMap.newKeySet();
342379

‎services/venice-controller/src/main/java/com/linkedin/venice/controller/grpc/server/StoreGrpcServiceImpl.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@
1414
import com.linkedin.venice.protocols.controller.DeleteAclForStoreGrpcResponse;
1515
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcRequest;
1616
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcResponse;
17+
import com.linkedin.venice.protocols.controller.GetStoreGrpcRequest;
18+
import com.linkedin.venice.protocols.controller.GetStoreGrpcResponse;
1719
import com.linkedin.venice.protocols.controller.ListStoresGrpcRequest;
1820
import com.linkedin.venice.protocols.controller.ListStoresGrpcResponse;
1921
import com.linkedin.venice.protocols.controller.ResourceCleanupCheckGrpcResponse;
@@ -147,4 +149,15 @@ public void listStores(ListStoresGrpcRequest grpcRequest, StreamObserver<ListSto
147149
clusterName,
148150
null);
149151
}
152+
153+
@Override
154+
public void getStore(GetStoreGrpcRequest grpcRequest, StreamObserver<GetStoreGrpcResponse> responseObserver) {
155+
LOGGER.debug("Received getStore with args: {}", grpcRequest);
156+
// No ACL check for getting store metadata - this is a read-only operation
157+
ControllerGrpcServerUtils.handleRequest(
158+
StoreGrpcServiceGrpc.getGetStoreMethod(),
159+
() -> storeRequestHandler.getStore(grpcRequest),
160+
responseObserver,
161+
grpcRequest.getStoreInfo());
162+
}
150163
}

‎services/venice-controller/src/main/java/com/linkedin/venice/controller/server/AdminSparkServer.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -338,7 +338,11 @@ public boolean startInner() throws Exception {
338338
httpService.get(
339339
CLUSTER_HEALTH_STORES.getPath(),
340340
new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getAllStoresStatuses(admin)));
341-
httpService.get(STORE.getPath(), new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getStore(admin)));
341+
httpService.get(
342+
STORE.getPath(),
343+
new VeniceParentControllerRegionStateHandler(
344+
admin,
345+
storesRoutes.getStore(admin, requestHandler.getStoreRequestHandler())));
342346
httpService.get(
343347
FUTURE_VERSION.getPath(),
344348
new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getFutureVersion(admin)));

‎services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoreRequestHandler.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@
44
import com.linkedin.venice.controller.ControllerRequestHandlerDependencies;
55
import com.linkedin.venice.controller.StoreDeletedValidation;
66
import com.linkedin.venice.exceptions.VeniceException;
7+
import com.linkedin.venice.exceptions.VeniceNoStoreException;
78
import com.linkedin.venice.meta.Store;
9+
import com.linkedin.venice.meta.StoreInfo;
810
import com.linkedin.venice.meta.ZKStore;
911
import com.linkedin.venice.protocols.controller.ClusterStoreGrpcInfo;
1012
import com.linkedin.venice.protocols.controller.CreateStoreGrpcRequest;
@@ -13,13 +15,16 @@
1315
import com.linkedin.venice.protocols.controller.DeleteAclForStoreGrpcResponse;
1416
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcRequest;
1517
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcResponse;
18+
import com.linkedin.venice.protocols.controller.GetStoreGrpcRequest;
19+
import com.linkedin.venice.protocols.controller.GetStoreGrpcResponse;
1620
import com.linkedin.venice.protocols.controller.ListStoresGrpcRequest;
1721
import com.linkedin.venice.protocols.controller.ListStoresGrpcResponse;
1822
import com.linkedin.venice.protocols.controller.UpdateAclForStoreGrpcRequest;
1923
import com.linkedin.venice.protocols.controller.UpdateAclForStoreGrpcResponse;
2024
import com.linkedin.venice.protocols.controller.ValidateStoreDeletedGrpcRequest;
2125
import com.linkedin.venice.protocols.controller.ValidateStoreDeletedGrpcResponse;
2226
import com.linkedin.venice.systemstore.schemas.StoreProperties;
27+
import com.linkedin.venice.utils.ObjectMapperFactory;
2328
import java.util.ArrayList;
2429
import java.util.List;
2530
import java.util.Optional;
@@ -255,4 +260,45 @@ public ListStoresGrpcResponse listStores(ListStoresGrpcRequest request) {
255260
LOGGER.info("Found {} stores in cluster: {}", selectedStoreNames.size(), clusterName);
256261
return ListStoresGrpcResponse.newBuilder().setClusterName(clusterName).addAllStoreNames(selectedStoreNames).build();
257262
}
263+
264+
/**
265+
* Gets store information for a given store in a cluster.
266+
* @param request the request containing cluster and store name
267+
* @return response containing store information as JSON
268+
*/
269+
public GetStoreGrpcResponse getStore(GetStoreGrpcRequest request) {
270+
ClusterStoreGrpcInfo storeGrpcInfo = request.getStoreInfo();
271+
ControllerRequestParamValidator.validateClusterStoreInfo(storeGrpcInfo);
272+
String clusterName = storeGrpcInfo.getClusterName();
273+
String storeName = storeGrpcInfo.getStoreName();
274+
275+
LOGGER.info("Getting store info for store: {} in cluster: {}", storeName, clusterName);
276+
277+
Store store = admin.getStore(clusterName, storeName);
278+
if (store == null) {
279+
throw new VeniceNoStoreException(storeName);
280+
}
281+
282+
StoreInfo storeInfo = StoreInfo.fromStore(store);
283+
// Set default retention time if not set
284+
if (storeInfo.getBackupVersionRetentionMs() < 0) {
285+
storeInfo.setBackupVersionRetentionMs(admin.getBackupVersionDefaultRetentionMs());
286+
}
287+
// Set default max record size if not set
288+
if (storeInfo.getMaxRecordSizeBytes() < 0) {
289+
storeInfo.setMaxRecordSizeBytes(admin.getDefaultMaxRecordSizeBytes());
290+
}
291+
storeInfo.setColoToCurrentVersions(admin.getCurrentVersionsForMultiColos(clusterName, storeName));
292+
boolean isSSL = admin.isSSLEnabledForPush(clusterName, storeName);
293+
storeInfo.setKafkaBrokerUrl(admin.getKafkaBootstrapServers(isSSL));
294+
295+
String storeInfoJson;
296+
try {
297+
storeInfoJson = ObjectMapperFactory.getInstance().writeValueAsString(storeInfo);
298+
} catch (Exception e) {
299+
throw new RuntimeException("Failed to serialize StoreInfo to JSON", e);
300+
}
301+
302+
return GetStoreGrpcResponse.newBuilder().setStoreInfo(storeGrpcInfo).setStoreInfoJson(storeInfoJson).build();
303+
}
258304
}

‎services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoresRoutes.java‎

Lines changed: 21 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,8 @@
109109
import com.linkedin.venice.meta.StoreInfo;
110110
import com.linkedin.venice.meta.Version;
111111
import com.linkedin.venice.protocols.controller.ClusterStoreGrpcInfo;
112+
import com.linkedin.venice.protocols.controller.GetStoreGrpcRequest;
113+
import com.linkedin.venice.protocols.controller.GetStoreGrpcResponse;
112114
import com.linkedin.venice.protocols.controller.ListStoresGrpcRequest;
113115
import com.linkedin.venice.protocols.controller.ListStoresGrpcResponse;
114116
import com.linkedin.venice.protocols.controller.ValidateStoreDeletedGrpcRequest;
@@ -118,6 +120,8 @@
118120
import com.linkedin.venice.pubsub.api.exceptions.PubSubTopicDoesNotExistException;
119121
import com.linkedin.venice.pubsub.manager.TopicManager;
120122
import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
123+
import com.linkedin.venice.systemstore.schemas.StoreProperties;
124+
import com.linkedin.venice.utils.ObjectMapperFactory;
121125
import com.linkedin.venice.utils.Utils;
122126
import java.util.ArrayList;
123127
import java.util.Collections;
@@ -275,34 +279,31 @@ public void internalHandle(Request request, RepushInfoResponse veniceResponse) {
275279
/**
276280
* @see Admin#getStore(String, String)
277281
*/
278-
public Route getStore(Admin admin) {
282+
public Route getStore(Admin admin, StoreRequestHandler requestHandler) {
279283
return new VeniceRouteHandler<StoreResponse>(StoreResponse.class) {
280284
@Override
281285
public void internalHandle(Request request, StoreResponse veniceResponse) {
282286
// No ACL check for getting store metadata
283287
AdminSparkServer.validateParams(request, STORE.getParams(), admin);
284288
String storeName = request.queryParams(NAME);
285289
String clusterName = request.queryParams(CLUSTER);
286-
veniceResponse.setCluster(clusterName);
287-
veniceResponse.setName(storeName);
288-
Store store = admin.getStore(clusterName, storeName);
289-
if (store == null) {
290-
throw new VeniceNoStoreException(storeName);
291-
}
292-
StoreInfo storeInfo = StoreInfo.fromStore(store);
293-
// Make sure store info will have right default retention time for Nuage UI display.
294-
if (storeInfo.getBackupVersionRetentionMs() < 0) {
295-
storeInfo.setBackupVersionRetentionMs(admin.getBackupVersionDefaultRetentionMs());
296-
}
297-
// This is the only place the default value of maxRecordSizeBytes is set for StoreResponse for VPJ and Consumer
298-
if (storeInfo.getMaxRecordSizeBytes() < 0) {
299-
storeInfo.setMaxRecordSizeBytes(admin.getDefaultMaxRecordSizeBytes());
300-
}
301-
storeInfo.setColoToCurrentVersions(admin.getCurrentVersionsForMultiColos(clusterName, storeName));
302-
boolean isSSL = admin.isSSLEnabledForPush(clusterName, storeName);
303-
storeInfo.setKafkaBrokerUrl(admin.getKafkaBootstrapServers(isSSL));
304290

305-
veniceResponse.setStore(storeInfo);
291+
ClusterStoreGrpcInfo storeGrpcInfo =
292+
ClusterStoreGrpcInfo.newBuilder().setClusterName(clusterName).setStoreName(storeName).build();
293+
GetStoreGrpcRequest grpcRequest = GetStoreGrpcRequest.newBuilder().setStoreInfo(storeGrpcInfo).build();
294+
295+
GetStoreGrpcResponse grpcResponse = requestHandler.getStore(grpcRequest);
296+
297+
veniceResponse.setCluster(grpcResponse.getStoreInfo().getClusterName());
298+
veniceResponse.setName(grpcResponse.getStoreInfo().getStoreName());
299+
300+
try {
301+
StoreInfo storeInfo =
302+
ObjectMapperFactory.getInstance().readValue(grpcResponse.getStoreInfoJson(), StoreInfo.class);
303+
veniceResponse.setStore(storeInfo);
304+
} catch (Exception e) {
305+
throw new VeniceException("Failed to deserialize StoreInfo from JSON", e);
306+
}
306307
}
307308
};
308309
}

‎services/venice-controller/src/test/java/com/linkedin/venice/controller/grpc/server/StoreGrpcServiceImplTest.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import com.linkedin.venice.protocols.controller.DeleteAclForStoreGrpcResponse;
2626
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcRequest;
2727
import com.linkedin.venice.protocols.controller.GetAclForStoreGrpcResponse;
28+
import com.linkedin.venice.protocols.controller.GetStoreGrpcRequest;
29+
import com.linkedin.venice.protocols.controller.GetStoreGrpcResponse;
2830
import com.linkedin.venice.protocols.controller.ListStoresGrpcRequest;
2931
import com.linkedin.venice.protocols.controller.ListStoresGrpcResponse;
3032
import com.linkedin.venice.protocols.controller.ResourceCleanupCheckGrpcResponse;
@@ -445,4 +447,39 @@ public void testListStoresWithFilters() {
445447
assertEquals(actualResponse.getClusterName(), TEST_CLUSTER, "Cluster name should match");
446448
assertEquals(actualResponse.getStoreNamesCount(), 1, "Should have 1 store after filtering");
447449
}
450+
451+
@Test
452+
public void testGetStoreReturnsSuccessfulResponse() {
453+
ClusterStoreGrpcInfo storeInfo =
454+
ClusterStoreGrpcInfo.newBuilder().setClusterName(TEST_CLUSTER).setStoreName(TEST_STORE).build();
455+
GetStoreGrpcRequest request = GetStoreGrpcRequest.newBuilder().setStoreInfo(storeInfo).build();
456+
String storeInfoJson = "{\"name\":\"" + TEST_STORE + "\"}";
457+
GetStoreGrpcResponse response =
458+
GetStoreGrpcResponse.newBuilder().setStoreInfo(storeInfo).setStoreInfoJson(storeInfoJson).build();
459+
when(storeRequestHandler.getStore(any(GetStoreGrpcRequest.class))).thenReturn(response);
460+
461+
GetStoreGrpcResponse actualResponse = blockingStub.getStore(request);
462+
463+
assertNotNull(actualResponse, "Response should not be null");
464+
assertEquals(actualResponse.getStoreInfo(), storeInfo, "Store info should match");
465+
assertEquals(actualResponse.getStoreInfoJson(), storeInfoJson, "Store info JSON should match");
466+
}
467+
468+
@Test
469+
public void testGetStoreReturnsErrorResponse() {
470+
ClusterStoreGrpcInfo storeInfo =
471+
ClusterStoreGrpcInfo.newBuilder().setClusterName(TEST_CLUSTER).setStoreName(TEST_STORE).build();
472+
GetStoreGrpcRequest request = GetStoreGrpcRequest.newBuilder().setStoreInfo(storeInfo).build();
473+
when(storeRequestHandler.getStore(any(GetStoreGrpcRequest.class)))
474+
.thenThrow(new VeniceException("Store not found"));
475+
476+
StatusRuntimeException e = expectThrows(StatusRuntimeException.class, () -> blockingStub.getStore(request));
477+
478+
assertNotNull(e.getStatus(), "Status should not be null");
479+
assertEquals(e.getStatus().getCode(), Status.INTERNAL.getCode());
480+
VeniceControllerGrpcErrorInfo errorInfo = GrpcRequestResponseConverter.parseControllerGrpcError(e);
481+
assertNotNull(errorInfo, "Error info should not be null");
482+
assertEquals(errorInfo.getErrorType(), ControllerGrpcErrorType.GENERAL_ERROR);
483+
assertTrue(errorInfo.getErrorMessage().contains("Store not found"));
484+
}
448485
}

0 commit comments

Comments
 (0)