Skip to content

Extract Pinot Broker Endpoints with interfaces #8876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
dongxiaoman wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

Extract Pinot Broker Endpoints with interfaces #8876

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
599a018
partially done
Jun 9, 2022
0c31c84
full refactor
Jun 9, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 3 additions & 18 deletions pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotBrokerAppConfigs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,41 +18,26 @@
*/
package org.apache.pinot.broker.api.resources;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.Authorization;
import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Application;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import org.apache.pinot.broker.api.services.PinotBrokerAppConfigsService;
import org.apache.pinot.broker.broker.BrokerAdminApiApplication;
import org.apache.pinot.common.utils.PinotAppConfigs;
import org.apache.pinot.spi.env.PinotConfiguration;

import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;


/**
* Resource to get the app configs {@link PinotAppConfigs} for
* the broker.
*/
@Api(tags = "AppConfig", authorizations = {@Authorization(value = SWAGGER_AUTHORIZATION_KEY)})
@SwaggerDefinition(securityDefinition = @SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotBrokerAppConfigs {
public class PinotBrokerAppConfigs implements PinotBrokerAppConfigsService {

@Context
private Application _application;

@GET
@Path("/appconfigs")
@Produces(MediaType.APPLICATION_JSON)
@Override
public String getAppConfigs() {
PinotConfiguration pinotConfiguration =
(PinotConfiguration) _application.getProperties().get(BrokerAdminApiApplication.PINOT_CONFIGURATION);
Expand Down
63 changes: 8 additions & 55 deletions pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotBrokerDebug.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,28 +18,14 @@
*/
package org.apache.pinot.broker.api.resources;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Authorization;
import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.pinot.broker.api.services.PinotBrokerDebugService;
import org.apache.pinot.broker.routing.BrokerRoutingManager;
import org.apache.pinot.broker.routing.timeboundary.TimeBoundaryInfo;
import org.apache.pinot.core.routing.RoutingTable;
Expand All @@ -48,30 +34,15 @@
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.sql.parsers.CalciteSqlCompiler;

import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;


@Api(tags = "Debug", authorizations = {@Authorization(value = SWAGGER_AUTHORIZATION_KEY)})
@SwaggerDefinition(securityDefinition = @SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
// TODO: Add APIs to return the RoutingTable (with unavailable segments)
public class PinotBrokerDebug {
public class PinotBrokerDebug implements PinotBrokerDebugService {

@Inject
private BrokerRoutingManager _routingManager;

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/debug/timeBoundary/{tableName}")
@ApiOperation(value = "Get the time boundary information for a table")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Time boundary information for a table"),
@ApiResponse(code = 404, message = "Time boundary not found"),
@ApiResponse(code = 500, message = "Internal server error")
})
public TimeBoundaryInfo getTimeBoundary(
@ApiParam(value = "Name of the table") @PathParam("tableName") String tableName) {
@Override
public TimeBoundaryInfo getTimeBoundary(String tableName) {
String offlineTableName =
TableNameBuilder.OFFLINE.tableNameWithType(TableNameBuilder.extractRawTableName(tableName));
TimeBoundaryInfo timeBoundaryInfo = _routingManager.getTimeBoundaryInfo(offlineTableName);
Expand All @@ -82,17 +53,8 @@ public TimeBoundaryInfo getTimeBoundary(
}
}

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/debug/routingTable/{tableName}")
@ApiOperation(value = "Get the routing table for a table")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Routing table"),
@ApiResponse(code = 404, message = "Routing not found"),
@ApiResponse(code = 500, message = "Internal server error")
})
public Map<String, Map<ServerInstance, List<String>>> getRoutingTable(
@ApiParam(value = "Name of the table") @PathParam("tableName") String tableName) {
@Override
public Map<String, Map<ServerInstance, List<String>>> getRoutingTable(String tableName) {
Map<String, Map<ServerInstance, List<String>>> result = new TreeMap<>();
TableType tableType = TableNameBuilder.getTableTypeFromTableName(tableName);
if (tableType != TableType.REALTIME) {
Expand All @@ -118,17 +80,8 @@ public Map<String, Map<ServerInstance, List<String>>> getRoutingTable(
}
}

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/debug/routingTable/sql")
@ApiOperation(value = "Get the routing table for a query")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Routing table"),
@ApiResponse(code = 404, message = "Routing not found"),
@ApiResponse(code = 500, message = "Internal server error")
})
public Map<ServerInstance, List<String>> getRoutingTableForQuery(
@ApiParam(value = "SQL query (table name should have type suffix)") @QueryParam("query") String query) {
@Override
public Map<ServerInstance, List<String>> getRoutingTableForQuery(String query) {
RoutingTable routingTable = _routingManager.getRoutingTable(CalciteSqlCompiler.compileToBrokerRequest(query));
if (routingTable != null) {
return routingTable.getServerInstanceToSegmentsMap();
Expand Down
29 changes: 3 additions & 26 deletions pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotBrokerHealthCheck.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,51 +18,28 @@
*/
package org.apache.pinot.broker.api.resources;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Authorization;
import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import javax.inject.Inject;
import javax.inject.Named;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.pinot.broker.api.services.PinotBrokerHealthCheckService;
import org.apache.pinot.broker.broker.BrokerAdminApiApplication;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.utils.ServiceStatus;

import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;


@Api(tags = "Health", authorizations = {@Authorization(value = SWAGGER_AUTHORIZATION_KEY)})
@SwaggerDefinition(securityDefinition = @SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotBrokerHealthCheck {
public class PinotBrokerHealthCheck implements PinotBrokerHealthCheckService {
@Inject
@Named(BrokerAdminApiApplication.BROKER_INSTANCE_ID)
private String _instanceId;

@Inject
private BrokerMetrics _brokerMetrics;

@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("health")
@ApiOperation(value = "Checking broker health")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Broker is healthy"),
@ApiResponse(code = 503, message = "Broker is not healthy")
})
@Override
public String getBrokerHealth() {
ServiceStatus.Status status = ServiceStatus.getServiceStatus(_instanceId);
if (status == ServiceStatus.Status.GOOD) {
Expand Down
60 changes: 8 additions & 52 deletions pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotBrokerRouting.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,76 +18,32 @@
*/
package org.apache.pinot.broker.api.resources;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Authorization;
import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import javax.inject.Inject;
import javax.ws.rs.DELETE;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import org.apache.pinot.broker.api.services.PinotBrokerRoutingService;
import org.apache.pinot.broker.routing.BrokerRoutingManager;

import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;


@Api(tags = "Routing", authorizations = {@Authorization(value = SWAGGER_AUTHORIZATION_KEY)})
@SwaggerDefinition(securityDefinition = @SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotBrokerRouting {
public class PinotBrokerRouting implements PinotBrokerRoutingService {

@Inject
BrokerRoutingManager _routingManager;

@PUT
@Produces(MediaType.TEXT_PLAIN)
@Path("/routing/{tableName}")
@ApiOperation(value = "Build/rebuild the routing for a table", notes = "Build/rebuild the routing for a table")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Success"),
@ApiResponse(code = 500, message = "Internal server error")
})
public String buildRouting(
@ApiParam(value = "Table name (with type)") @PathParam("tableName") String tableNameWithType) {
@Override
public String buildRouting(String tableNameWithType) {
_routingManager.buildRouting(tableNameWithType);
return "Success";
}

@PUT
@Produces(MediaType.TEXT_PLAIN)
@Path("/routing/refresh/{tableName}/{segmentName}")
@ApiOperation(value = "Refresh the routing for a segment", notes = "Refresh the routing for a segment")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Success"),
@ApiResponse(code = 500, message = "Internal server error")
})
public String refreshRouting(
@ApiParam(value = "Table name (with type)") @PathParam("tableName") String tableNameWithType,
@ApiParam(value = "Segment name") @PathParam("segmentName") String segmentName) {
@Override
public String refreshRouting(String tableNameWithType, String segmentName) {
_routingManager.refreshSegment(tableNameWithType, segmentName);
return "Success";
}

@DELETE
@Produces(MediaType.TEXT_PLAIN)
@Path("/routing/{tableName}")
@ApiOperation(value = "Remove the routing for a table", notes = "Remove the routing for a table")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Success"),
@ApiResponse(code = 500, message = "Internal server error")
})
public String removeRouting(
@ApiParam(value = "Table name (with type)") @PathParam("tableName") String tableNameWithType) {
@Override
public String removeRouting(String tableNameWithType) {
_routingManager.removeRouting(tableNameWithType);
return "Success";
}
Expand Down
56 changes: 8 additions & 48 deletions pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,31 +22,15 @@
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Authorization;
import io.swagger.annotations.SecurityDefinition;
import io.swagger.annotations.SwaggerDefinition;
import java.util.HashMap;
import java.util.Map;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.pinot.broker.api.HttpRequesterIdentity;
import org.apache.pinot.broker.api.services.PinotClientRequestService;
import org.apache.pinot.broker.requesthandler.BrokerRequestHandler;
import org.apache.pinot.common.exception.QueryException;
import org.apache.pinot.common.metrics.BrokerMeter;
Expand All @@ -61,18 +45,12 @@
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.sql.parsers.PinotSqlType;
import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
import org.glassfish.jersey.server.ManagedAsync;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;


@Api(tags = "Query", authorizations = {@Authorization(value = SWAGGER_AUTHORIZATION_KEY)})
@SwaggerDefinition(securityDefinition = @SecurityDefinition(apiKeyAuthDefinitions = @ApiKeyAuthDefinition(name =
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotClientRequest {
public class PinotClientRequest implements PinotClientRequestService {
private static final Logger LOGGER = LoggerFactory.getLogger(PinotClientRequest.class);

@Inject
Expand All @@ -84,19 +62,9 @@ public class PinotClientRequest {
@Inject
private BrokerMetrics _brokerMetrics;

@GET
@ManagedAsync
@Produces(MediaType.APPLICATION_JSON)
@Path("query/sql")
@ApiOperation(value = "Querying pinot")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Query response"),
@ApiResponse(code = 500, message = "Internal Server Error")
})
public void processSqlQueryGet(@ApiParam(value = "Query", required = true) @QueryParam("sql") String query,
@ApiParam(value = "Trace enabled") @QueryParam(Request.TRACE) String traceEnabled,
@ApiParam(value = "Debug options") @QueryParam(Request.DEBUG_OPTIONS) String debugOptions,
@Suspended AsyncResponse asyncResponse, @Context org.glassfish.grizzly.http.server.Request requestContext) {
@Override
public void processSqlQueryGet(String query, String traceEnabled, String debugOptions, AsyncResponse asyncResponse,
org.glassfish.grizzly.http.server.Request requestContext) {
try {
ObjectNode requestJson = JsonUtils.newObjectNode();
requestJson.put(Request.SQL, query);
Expand All @@ -117,17 +85,9 @@ public void processSqlQueryGet(@ApiParam(value = "Query", required = true) @Quer
}
}

@POST
@ManagedAsync
@Produces(MediaType.APPLICATION_JSON)
@Path("query/sql")
@ApiOperation(value = "Querying pinot")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Query response"),
@ApiResponse(code = 500, message = "Internal Server Error")
})
public void processSqlQueryPost(String query, @Suspended AsyncResponse asyncResponse,
@Context org.glassfish.grizzly.http.server.Request requestContext) {
@Override
public void processSqlQueryPost(String query, AsyncResponse asyncResponse,
org.glassfish.grizzly.http.server.Request requestContext) {
try {
JsonNode requestJson = JsonUtils.stringToJsonNode(query);
if (!requestJson.has(Request.SQL)) {
Expand Down
Loading