HBASE-24718 : Generic NamedQueue framework for multiple use-cases (Refactor SlowLog responses) #2052
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,24 +17,24 @@ | |
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.hadoop.hbase.namequeues; | ||
|
|
||
| import com.lmax.disruptor.EventHandler; | ||
| import com.lmax.disruptor.RingBuffer; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Queue; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.apache.hadoop.conf.Configuration; | ||
| import org.apache.hadoop.hbase.HConstants; | ||
| import org.apache.hadoop.hbase.client.SlowLogParams; | ||
| import org.apache.hadoop.hbase.ipc.RpcCall; | ||
| import org.apache.yetus.audience.InterfaceAudience; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
@@ -50,64 +50,83 @@ | |
| import org.apache.hadoop.hbase.shaded.protobuf.generated.TooSlowLog.SlowLogPayload; | ||
|
|
||
| /** | ||
| * Event Handler run by disruptor ringbuffer consumer. | ||
| * Although this is generic implementation for namedQueue, it can have individual queue specific | ||
| * logic. | ||
| */ | ||
| @InterfaceAudience.Private | ||
| class LogEventHandler implements EventHandler<RingBufferEnvelope> { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(LogEventHandler.class); | ||
|
|
||
| private static final String SLOW_LOG_RING_BUFFER_SIZE = | ||
| "hbase.regionserver.slowlog.ringbuffer.size"; | ||
|
|
||
| // Map that binds namedQueues. | ||
| // If NamedQueue of specific type is enabled, corresponding Queue will be used to | ||
| // insert and retrieve records. | ||
| // Individual queue sizes should be determined based on their individual configs. | ||
| private final Map<NamedQueuePayload.NamedQueueEvent, Queue> namedQueues = new HashMap<>(); | ||
|
Contributor
There was a problem hiding this comment. For example, the caller should be able to determine which queue implementations (and its related services) should be loaded. That would isolate LogEventHandler from being constantly modified by additional services requiring this type of loggings.
Contributor
Author
There was a problem hiding this comment. Strategy pattern is good thought but we don't have common payload coming from client so far. Specifically when admin client will talk to RSRpcServer and that will talk to |
||
|
|
||
| private final boolean isSlowLogTableEnabled; | ||
| private final SlowLogPersistentService slowLogPersistentService; | ||
|
|
||
| LogEventHandler(final Configuration conf) { | ||
| // Initialize SlowLog Queue | ||
| int slowLogQueueSize = conf.getInt(SLOW_LOG_RING_BUFFER_SIZE, | ||
| HConstants.DEFAULT_SLOW_LOG_RING_BUFFER_SIZE); | ||
| EvictingQueue<SlowLogPayload> evictingQueue = EvictingQueue.create(slowLogQueueSize); | ||
|
|
||
| // Add slowLog queue in the namedQueue map | ||
| Queue<SlowLogPayload> slowLogQueue = Queues.synchronizedQueue(evictingQueue); | ||
| namedQueues.put(NamedQueuePayload.NamedQueueEvent.SLOW_LOG, slowLogQueue); | ||
|
|
||
| this.isSlowLogTableEnabled = conf.getBoolean(HConstants.SLOW_LOG_SYS_TABLE_ENABLED_KEY, | ||
| HConstants.DEFAULT_SLOW_LOG_SYS_TABLE_ENABLED_KEY); | ||
| if (isSlowLogTableEnabled) { | ||
| slowLogPersistentService = new SlowLogPersistentService(conf); | ||
| } else { | ||
| slowLogPersistentService = null; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Called when a publisher has published an event to the {@link RingBuffer}. | ||
| * This is generic consumer of disruptor ringbuffer and for each new namedQueue that we | ||
| * add, we should also provide specific consumer logic here. | ||
| * | ||
| * @param event published to the {@link RingBuffer} | ||
| * @param sequence of the event being processed | ||
| * @param endOfBatch flag to indicate if this is the last event in a batch from | ||
| * the {@link RingBuffer} | ||
| */ | ||
| @Override | ||
| public void onEvent(RingBufferEnvelope event, long sequence, boolean endOfBatch) { | ||
| final NamedQueuePayload namedQueuePayload = event.getPayload(); | ||
| // consume ringbuffer payload based on event type | ||
| if (NamedQueuePayload.NamedQueueEvent.SLOW_LOG | ||
| .equals(namedQueuePayload.getNamedQueueEvent())) { | ||
| consumeSlowLogEvent((RpcLogDetails) namedQueuePayload); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * This implementation is specific to slowLog event. This consumes slowLog event from | ||
| * disruptor and inserts records to EvictingQueue. | ||
| * | ||
| * @param rpcLogDetails Input for slow/largeLog events | ||
| */ | ||
| private void consumeSlowLogEvent(RpcLogDetails rpcLogDetails) { | ||
| final RpcCall rpcCall = rpcLogDetails.getRpcCall(); | ||
| final String clientAddress = rpcLogDetails.getClientAddress(); | ||
| final long responseSize = rpcLogDetails.getResponseSize(); | ||
| final String className = rpcLogDetails.getClassName(); | ||
| final SlowLogPayload.Type type = getLogType(rpcLogDetails); | ||
| if (type == null) { | ||
| return; | ||
| } | ||
| Descriptors.MethodDescriptor methodDescriptor = rpcCall.getMethod(); | ||
| Message param = rpcLogDetails.getParam(); | ||
| long receiveTime = rpcCall.getReceiveTime(); | ||
| long startTime = rpcCall.getStartTime(); | ||
| long endTime = System.currentTimeMillis(); | ||
|
|
@@ -153,10 +172,10 @@ public void onEvent(RingBufferEnvelope event, long sequence, boolean endOfBatch) | |
| .setType(type) | ||
| .setUserName(userName) | ||
| .build(); | ||
| namedQueues.get(NamedQueuePayload.NamedQueueEvent.SLOW_LOG).add(slowLogPayload); | ||
| if (isSlowLogTableEnabled) { | ||
| if (!slowLogPayload.getRegionName().startsWith("hbase:slowlog")) { | ||
| slowLogPersistentService.addToQueueForSysTable(slowLogPayload); | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -183,13 +202,12 @@ private SlowLogPayload.Type getLogType(RpcLogDetails rpcCallDetails) { | |
| /** | ||
| * Cleans up slow log payloads | ||
| * | ||
| * @param namedQueueEvent type of queue to clear | ||
| * @return true if slow log payloads are cleaned up, false otherwise | ||
| */ | ||
| boolean clearNamedQueue(NamedQueuePayload.NamedQueueEvent namedQueueEvent) { | ||
| LOG.debug("Received request to clean up online slowlog buffer.."); | ||
| namedQueues.get(namedQueueEvent).clear(); | ||
| return true; | ||
| } | ||
|
|
||
|
|
@@ -200,8 +218,10 @@ boolean clearSlowLogs() { | |
| * @return list of slow log payloads | ||
| */ | ||
| List<SlowLogPayload> getSlowLogPayloads(final AdminProtos.SlowLogResponseRequest request) { | ||
| Queue<SlowLogPayload> slowLogQueue = | ||
| namedQueues.get(NamedQueuePayload.NamedQueueEvent.SLOW_LOG); | ||
| List<SlowLogPayload> slowLogPayloadList = | ||
| Arrays.stream(slowLogQueue.toArray(new SlowLogPayload[0])) | ||
| .filter(e -> e.getType() == SlowLogPayload.Type.ALL | ||
| || e.getType() == SlowLogPayload.Type.SLOW_LOG) | ||
| .collect(Collectors.toList()); | ||
|
|
@@ -219,8 +239,10 @@ List<SlowLogPayload> getSlowLogPayloads(final AdminProtos.SlowLogResponseRequest | |
| * @return list of large log payloads | ||
| */ | ||
| List<SlowLogPayload> getLargeLogPayloads(final AdminProtos.SlowLogResponseRequest request) { | ||
| Queue<SlowLogPayload> largeLogQueue = | ||
| namedQueues.get(NamedQueuePayload.NamedQueueEvent.SLOW_LOG); | ||
| List<SlowLogPayload> slowLogPayloadList = | ||
| Arrays.stream(largeLogQueue.toArray(new SlowLogPayload[0])) | ||
| .filter(e -> e.getType() == SlowLogPayload.Type.ALL | ||
| || e.getType() == SlowLogPayload.Type.LARGE_LOG) | ||
| .collect(Collectors.toList()); | ||
|
|
@@ -232,34 +254,12 @@ List<SlowLogPayload> getLargeLogPayloads(final AdminProtos.SlowLogResponseReques | |
| } | ||
|
|
||
| /** | ||
| * Add all slowLog events to system table. This is only for slowLog event's persistence on | ||
| * system table. | ||
| */ | ||
| void addAllSlowLogsToSysTable() { | ||
| if (slowLogPersistentService != null) { | ||
| slowLogPersistentService.addAllLogsToSysTable(); | ||
| } | ||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.