| 1346 | |
| 1347 | |
| 1348 | def start_server( |
| 1349 | store: FeatureStore, |
| 1350 | port: int, |
| 1351 | wait_for_termination: bool = True, |
| 1352 | tls_key_path: str = "", |
| 1353 | tls_cert_path: str = "", |
| 1354 | ): |
| 1355 | auth_manager_type = str_to_auth_manager_type(store.config.auth_config.type) |
| 1356 | init_security_manager(auth_type=auth_manager_type, fs=store) |
| 1357 | init_auth_manager( |
| 1358 | auth_type=auth_manager_type, |
| 1359 | server_type=ServerType.GRPC, |
| 1360 | auth_config=store.config.auth_config, |
| 1361 | ) |
| 1362 | |
| 1363 | server = grpc.server( |
| 1364 | futures.ThreadPoolExecutor(max_workers=10), |
| 1365 | interceptors=_grpc_interceptors(auth_manager_type), |
| 1366 | ) |
| 1367 | RegistryServer_pb2_grpc.add_RegistryServerServicer_to_server( |
| 1368 | RegistryServer(store.registry), server |
| 1369 | ) |
| 1370 | # Add health check service to server |
| 1371 | health_servicer = health.HealthServicer() |
| 1372 | health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server) |
| 1373 | health_servicer.set("", health_pb2.HealthCheckResponse.SERVING) |
| 1374 | |
| 1375 | service_names_available_for_reflection = ( |
| 1376 | RegistryServer_pb2.DESCRIPTOR.services_by_name["RegistryServer"].full_name, |
| 1377 | health_pb2.DESCRIPTOR.services_by_name["Health"].full_name, |
| 1378 | reflection.SERVICE_NAME, |
| 1379 | ) |
| 1380 | reflection.enable_server_reflection(service_names_available_for_reflection, server) |
| 1381 | |
| 1382 | if tls_cert_path and tls_key_path: |
| 1383 | with ( |
| 1384 | open(tls_cert_path, "rb") as cert_file, |
| 1385 | open(tls_key_path, "rb") as key_file, |
| 1386 | ): |
| 1387 | certificate_chain = cert_file.read() |
| 1388 | private_key = key_file.read() |
| 1389 | server_credentials = grpc.ssl_server_credentials( |
| 1390 | ((private_key, certificate_chain),) |
| 1391 | ) |
| 1392 | logger.info("Starting grpc registry server in TLS(SSL) mode") |
| 1393 | server.add_secure_port(f"[::]:{port}", server_credentials) |
| 1394 | else: |
| 1395 | logger.info("Starting grpc registry server in non-TLS(SSL) mode") |
| 1396 | server.add_insecure_port(f"[::]:{port}") |
| 1397 | server.start() |
| 1398 | if wait_for_termination: |
| 1399 | logger.info(f"Grpc server started at localhost:{port}") |
| 1400 | server.wait_for_termination() |
| 1401 | else: |
| 1402 | return server |
| 1403 | |
| 1404 | |
| 1405 | def _grpc_interceptors( |