Merge pull request #2322 from bnamasivayam/master
Added contrib folder to foundationdb. New tool called transaction_pr…
This commit is contained in:
Alex Miller
GitHub
commit
488b480f20
|
|
@ -201,7 +201,7 @@ add_subdirectory(tests)
|
|||
if(WITH_DOCUMENTATION)
|
||||
add_subdirectory(documentation)
|
||||
endif()
|
||||
add_subdirectory(monitoring)
|
||||
add_subdirectory(contrib/monitoring)
|
||||
|
||||
if(WIN32)
|
||||
add_subdirectory(packaging/msi)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,806 @@
|
|||
"""
|
||||
Requirements:
|
||||
python3
|
||||
fdb python bindings
|
||||
optional packages:
|
||||
dateparser (for human date parsing)
|
||||
sortedcontainers (for estimating key range read/write density)
|
||||
"""
|
||||
|
||||
|
||||
import argparse
|
||||
from collections import defaultdict
|
||||
from enum import Enum
|
||||
import fdb
|
||||
from fdb.impl import strinc
|
||||
import json
|
||||
from json import JSONEncoder
|
||||
import logging
|
||||
import struct
|
||||
from bisect import bisect_left
|
||||
import time
|
||||
|
||||
PROTOCOL_VERSION_5_2 = 0x0FDB00A552000001
|
||||
PROTOCOL_VERSION_6_0 = 0x0FDB00A570010001
|
||||
PROTOCOL_VERSION_6_1 = 0x0FDB00B061060001
|
||||
PROTOCOL_VERSION_6_2 = 0x0FDB00B062010001
|
||||
supported_protocol_versions = frozenset([PROTOCOL_VERSION_5_2, PROTOCOL_VERSION_6_0, PROTOCOL_VERSION_6_1,
|
||||
PROTOCOL_VERSION_6_2])
|
||||
|
||||
|
||||
fdb.api_version(600)
|
||||
|
||||
BASIC_FORMAT = "%(asctime)s - %(levelname)-8s %(message)s"
|
||||
LOG_PATH = "transaction_profiling_analyzer.log"
|
||||
|
||||
|
||||
def setup_logger(name):
|
||||
root = logging.getLogger(name)
|
||||
root.setLevel(logging.DEBUG)
|
||||
root.propagate = False
|
||||
|
||||
file_formatter = logging.Formatter(BASIC_FORMAT)
|
||||
|
||||
file_handler = logging.FileHandler(LOG_PATH)
|
||||
file_handler.setFormatter(file_formatter)
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
|
||||
root.addHandler(file_handler)
|
||||
|
||||
return root
|
||||
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
class ByteBuffer(object):
|
||||
def __init__(self, val):
|
||||
self._offset = 0
|
||||
self.val = val
|
||||
|
||||
def get_bytes(self, n):
|
||||
if self._offset + n > len(self.val):
|
||||
raise IndexError("Request to read %d bytes with only %d remaining" % (n, self.get_remaining_bytes()))
|
||||
ret = self.val[self._offset:self._offset + n]
|
||||
self._offset += n
|
||||
return ret
|
||||
|
||||
def get_int(self):
|
||||
return struct.unpack("<i", self.get_bytes(4))[0]
|
||||
|
||||
def get_long(self):
|
||||
return struct.unpack("<q", self.get_bytes(8))[0]
|
||||
|
||||
def get_double(self):
|
||||
return struct.unpack("<d", self.get_bytes(8))[0]
|
||||
|
||||
def get_bytes_with_length(self):
|
||||
length = self.get_int()
|
||||
return self.get_bytes(length)
|
||||
|
||||
def get_key_range(self):
|
||||
return KeyRange(self.get_bytes_with_length(), self.get_bytes_with_length())
|
||||
|
||||
def get_key_range_list(self):
|
||||
length = self.get_int()
|
||||
return [self.get_key_range() for _ in range(0, length)]
|
||||
|
||||
def get_mutation(self):
|
||||
return Mutation(ord(self.get_bytes(1)), self.get_bytes_with_length(), self.get_bytes_with_length())
|
||||
|
||||
def get_mutation_list(self):
|
||||
length = self.get_int()
|
||||
return [self.get_mutation() for _ in range(0, length)]
|
||||
|
||||
def get_remaining_bytes(self):
|
||||
return len(self.val) - self._offset
|
||||
|
||||
|
||||
class ObjJsonEncoder(JSONEncoder):
|
||||
def default(self, o):
|
||||
try:
|
||||
super().default(o)
|
||||
except TypeError:
|
||||
if isinstance(o, Enum):
|
||||
return str(o)
|
||||
if hasattr(o, "__dict__"):
|
||||
return o.__dict__
|
||||
return str(o)
|
||||
|
||||
|
||||
class TrInfoChunk(object):
|
||||
def __init__(self, num_chunks, chunk_num, key, value):
|
||||
self.num_chunks = num_chunks
|
||||
self.chunk_num = chunk_num
|
||||
self.key = key
|
||||
self.value = value
|
||||
|
||||
|
||||
class KeyRange(object):
|
||||
def __init__(self, start_key, end_key):
|
||||
self.start_key = start_key
|
||||
self.end_key = end_key
|
||||
|
||||
|
||||
class MutationType(Enum):
|
||||
SET_VALUE = 0
|
||||
CLEAR_RANGE = 1
|
||||
ADD_VALUE = 2
|
||||
DEBUG_KEY_RANGE = 3
|
||||
DEBUG_KEY = 4
|
||||
NO_OP = 5
|
||||
AND = 6
|
||||
OR = 7
|
||||
XOR = 8
|
||||
APPEND_IF_FITS = 9
|
||||
AVAILABLE_FOR_REUSE = 10
|
||||
RESERVED_FOR_LOG_PROTOCOL_MESSAGE = 11
|
||||
MAX = 12
|
||||
MIN = 13
|
||||
SET_VERSION_STAMPED_KEY = 14
|
||||
SET_VERSION_STAMPED_VALUE = 15
|
||||
|
||||
|
||||
class Mutation(object):
|
||||
def __init__(self, code, param_one, param_two):
|
||||
self.code = MutationType(code)
|
||||
self.param_one = param_one
|
||||
self.param_two = param_two
|
||||
|
||||
|
||||
class BaseInfo(object):
|
||||
def __init__(self, start_timestamp):
|
||||
self.start_timestamp = start_timestamp
|
||||
|
||||
|
||||
class GetVersionInfo(BaseInfo):
|
||||
def __init__(self, bb, protocol_version):
|
||||
super().__init__(bb.get_double())
|
||||
self.latency = bb.get_double()
|
||||
if protocol_version >= PROTOCOL_VERSION_6_2:
|
||||
self.transaction_priority_type = bb.get_int()
|
||||
|
||||
|
||||
class GetInfo(BaseInfo):
|
||||
def __init__(self, bb):
|
||||
super().__init__(bb.get_double())
|
||||
self.latency = bb.get_double()
|
||||
self.value_size = bb.get_int()
|
||||
self.key = bb.get_bytes_with_length()
|
||||
|
||||
|
||||
class GetRangeInfo(BaseInfo):
|
||||
def __init__(self, bb):
|
||||
super().__init__(bb.get_double())
|
||||
self.latency = bb.get_double()
|
||||
self.range_size = bb.get_int()
|
||||
self.key_range = bb.get_key_range()
|
||||
|
||||
|
||||
class CommitInfo(BaseInfo):
|
||||
def __init__(self, bb, full_output=True):
|
||||
super().__init__(bb.get_double())
|
||||
self.latency = bb.get_double()
|
||||
self.num_mutations = bb.get_int()
|
||||
self.commit_bytes = bb.get_int()
|
||||
|
||||
read_conflict_range = bb.get_key_range_list()
|
||||
if full_output:
|
||||
self.read_conflict_range = read_conflict_range
|
||||
write_conflict_range = bb.get_key_range_list()
|
||||
if full_output:
|
||||
self.write_conflict_range = write_conflict_range
|
||||
mutations = bb.get_mutation_list()
|
||||
if full_output:
|
||||
self.mutations = mutations
|
||||
|
||||
self.read_snapshot_version = bb.get_long()
|
||||
|
||||
|
||||
class ErrorGetInfo(BaseInfo):
|
||||
def __init__(self, bb):
|
||||
super().__init__(bb.get_double())
|
||||
self.error_code = bb.get_int()
|
||||
self.key = bb.get_bytes_with_length()
|
||||
|
||||
|
||||
class ErrorGetRangeInfo(BaseInfo):
|
||||
def __init__(self, bb):
|
||||
super().__init__(bb.get_double())
|
||||
self.error_code = bb.get_int()
|
||||
self.key_range = bb.get_key_range()
|
||||
|
||||
|
||||
class ErrorCommitInfo(BaseInfo):
|
||||
def __init__(self, bb, full_output=True):
|
||||
super().__init__(bb.get_double())
|
||||
self.error_code = bb.get_int()
|
||||
|
||||
read_conflict_range = bb.get_key_range_list()
|
||||
if full_output:
|
||||
self.read_conflict_range = read_conflict_range
|
||||
write_conflict_range = bb.get_key_range_list()
|
||||
if full_output:
|
||||
self.write_conflict_range = write_conflict_range
|
||||
mutations = bb.get_mutation_list()
|
||||
if full_output:
|
||||
self.mutations = mutations
|
||||
|
||||
self.read_snapshot_version = bb.get_long()
|
||||
|
||||
|
||||
class UnsupportedProtocolVersionError(Exception):
|
||||
def __init__(self, protocol_version):
|
||||
super().__init__("Unsupported protocol version 0x%0.2X" % protocol_version)
|
||||
|
||||
|
||||
class ClientTransactionInfo:
|
||||
def __init__(self, bb, full_output=True, type_filter=None):
|
||||
self.get_version = None
|
||||
self.gets = []
|
||||
self.get_ranges = []
|
||||
self.commit = None
|
||||
self.error_gets = []
|
||||
self.error_get_ranges = []
|
||||
self.error_commits = []
|
||||
|
||||
protocol_version = bb.get_long()
|
||||
if protocol_version not in supported_protocol_versions:
|
||||
raise UnsupportedProtocolVersionError(protocol_version)
|
||||
while bb.get_remaining_bytes():
|
||||
event = bb.get_int()
|
||||
if event == 0:
|
||||
# we need to read it to consume the buffer even if we don't want to store it
|
||||
get_version = GetVersionInfo(bb, protocol_version)
|
||||
if (not type_filter or "get_version" in type_filter):
|
||||
self.get_version = get_version
|
||||
elif event == 1:
|
||||
get = GetInfo(bb)
|
||||
if (not type_filter or "get" in type_filter):
|
||||
# because of the crappy json serializtion using __dict__ we have to set the list here otherwise
|
||||
# it doesn't print
|
||||
if not self.gets: self.gets = []
|
||||
self.gets.append(get)
|
||||
elif event == 2:
|
||||
get_range = GetRangeInfo(bb)
|
||||
if (not type_filter or "get_range" in type_filter):
|
||||
if not self.get_ranges: self.get_ranges = []
|
||||
self.get_ranges.append(get_range)
|
||||
elif event == 3:
|
||||
commit = CommitInfo(bb, full_output=full_output)
|
||||
if (not type_filter or "commit" in type_filter):
|
||||
self.commit = commit
|
||||
elif event == 4:
|
||||
error_get = ErrorGetInfo(bb)
|
||||
if (not type_filter or "error_gets" in type_filter):
|
||||
if not self.error_gets: self.error_gets = []
|
||||
self.error_gets.append(error_get)
|
||||
elif event == 5:
|
||||
error_get_range = ErrorGetRangeInfo(bb)
|
||||
if (not type_filter or "error_get_range" in type_filter):
|
||||
if not self.error_get_ranges: self.error_get_ranges = []
|
||||
self.error_get_ranges.append(error_get_range)
|
||||
elif event == 6:
|
||||
error_commit = ErrorCommitInfo(bb, full_output=full_output)
|
||||
if (not type_filter or "error_commit" in type_filter):
|
||||
if not self.error_commits: self.error_commits = []
|
||||
self.error_commits.append(error_commit)
|
||||
else:
|
||||
raise Exception("Unknown event type %d" % event)
|
||||
|
||||
def has_types(self):
|
||||
return self.get_version or self.gets or self.get_ranges or self.commit or self.error_gets \
|
||||
or self.error_get_ranges or self.error_commits
|
||||
|
||||
def to_json(self):
|
||||
return json.dumps(self, cls=ObjJsonEncoder, sort_keys=True)
|
||||
|
||||
|
||||
class TransactionInfoLoader(object):
|
||||
max_num_chunks_to_store = 1000 # Each chunk would be 100 KB in size
|
||||
|
||||
def __init__(self, db, full_output=True, type_filter=None, min_timestamp=None, max_timestamp=None):
|
||||
self.db = db
|
||||
self.full_output = full_output
|
||||
self.type_filter = type_filter
|
||||
self.min_timestamp = min_timestamp
|
||||
self.max_timestamp = max_timestamp
|
||||
'''
|
||||
Keys look like this
|
||||
FF - 2 bytes \xff\x02
|
||||
SSSSSSSSSS - 10 bytes Version Stamp
|
||||
RRRRRRRRRRRRRRRR - 16 bytes Transaction id
|
||||
NNNN - 4 Bytes Chunk number
|
||||
TTTT - 4 Bytes Total number of chunks
|
||||
'''
|
||||
sample_key = "FF/fdbClientInfo/client_latency/SSSSSSSSSS/RRRRRRRRRRRRRRRR/NNNNTTTT/"
|
||||
|
||||
self.client_latency_start = b'\xff\x02/fdbClientInfo/client_latency/'
|
||||
self.client_latency_start_key_selector = fdb.KeySelector.first_greater_than(self.client_latency_start)
|
||||
self.client_latency_end_key_selector = fdb.KeySelector.first_greater_or_equal(strinc(self.client_latency_start))
|
||||
self.version_stamp_start_idx = sample_key.index('S')
|
||||
self.version_stamp_end_idx = sample_key.rindex('S')
|
||||
self.tr_id_start_idx = sample_key.index('R')
|
||||
self.tr_id_end_idx = sample_key.rindex('R')
|
||||
self.chunk_num_start_idx = sample_key.index('N')
|
||||
self.num_chunks_start_idx = sample_key.index('T')
|
||||
|
||||
self.tr_info_map = {}
|
||||
self.num_chunks_stored = 0
|
||||
self.num_transactions_discarded = 0
|
||||
|
||||
def _check_and_adjust_chunk_cache_size(self):
|
||||
if self.num_chunks_stored > self.max_num_chunks_to_store:
|
||||
c_list = self.tr_info_map.pop(next(iter(self.tr_info_map)))
|
||||
self.num_chunks_stored -= len(c_list)
|
||||
self.num_transactions_discarded += 1
|
||||
|
||||
def parse_key(self, k):
|
||||
version_stamp_bytes = k[self.version_stamp_start_idx:self.version_stamp_end_idx + 1]
|
||||
tr_id = k[self.tr_id_start_idx:self.tr_id_end_idx + 1]
|
||||
num_chunks = struct.unpack(">i", k[self.num_chunks_start_idx:self.num_chunks_start_idx + 4])[0]
|
||||
chunk_num = struct.unpack(">i", k[self.chunk_num_start_idx:self.chunk_num_start_idx + 4])[0]
|
||||
return version_stamp_bytes, tr_id, num_chunks, chunk_num
|
||||
|
||||
def get_key_prefix_for_version_stamp(self, version_stamp):
|
||||
return self.client_latency_start + struct.pack(">Q", version_stamp) + b'\x00\x00'
|
||||
|
||||
@fdb.transactional
|
||||
def find_version_for_timestamp(self, tr, timestamp, start):
|
||||
"""
|
||||
Uses Timekeeper to find the closest version to a timestamp.
|
||||
If start is True, will find the greatest version at or before timestamp.
|
||||
If start is False, will find the smallest version at or after the timestamp.
|
||||
|
||||
:param tr:
|
||||
:param timestamp:
|
||||
:param start:
|
||||
:return:
|
||||
"""
|
||||
tr.options.set_read_system_keys()
|
||||
tr.options.set_read_lock_aware()
|
||||
timekeeper_prefix = b'\xff\x02/timeKeeper/map/'
|
||||
timestamp_packed = fdb.tuple.pack((timestamp,))
|
||||
if start:
|
||||
start_key = timekeeper_prefix
|
||||
end_key = fdb.KeySelector.first_greater_than(timekeeper_prefix + timestamp_packed)
|
||||
reverse = True
|
||||
else:
|
||||
start_key = fdb.KeySelector.first_greater_or_equal(timekeeper_prefix + timestamp_packed)
|
||||
end_key = fdb.KeySelector.first_greater_or_equal(strinc(timekeeper_prefix))
|
||||
reverse = False
|
||||
for k, v in tr.snapshot.get_range(start_key, end_key, limit=1, reverse=reverse):
|
||||
return fdb.tuple.unpack(v)[0]
|
||||
return 0 if start else 0x8000000000000000 # we didn't find any timekeeper data so find the max range
|
||||
|
||||
def fetch_transaction_info(self):
|
||||
if self.min_timestamp:
|
||||
start_version = self.find_version_for_timestamp(self.db, self.min_timestamp, True)
|
||||
logger.debug("Using start version %s" % start_version)
|
||||
start_key = self.get_key_prefix_for_version_stamp(start_version)
|
||||
else:
|
||||
start_key = self.client_latency_start_key_selector
|
||||
|
||||
if self.max_timestamp:
|
||||
end_version = self.find_version_for_timestamp(self.db, self.max_timestamp, False)
|
||||
logger.debug("Using end version %s" % end_version)
|
||||
end_key = self.get_key_prefix_for_version_stamp(end_version)
|
||||
else:
|
||||
end_key = self.client_latency_end_key_selector
|
||||
|
||||
valid_transaction_infos = 0
|
||||
invalid_transaction_infos = 0
|
||||
|
||||
def build_client_transaction_info(v):
|
||||
return ClientTransactionInfo(ByteBuffer(v), full_output=self.full_output, type_filter=self.type_filter)
|
||||
|
||||
more = True
|
||||
tr = self.db.create_transaction()
|
||||
while more:
|
||||
tr.options.set_read_system_keys()
|
||||
tr.options.set_read_lock_aware()
|
||||
found = 0
|
||||
buffer = []
|
||||
try:
|
||||
logger.debug("Querying [%s:%s]" % (start_key, end_key))
|
||||
transaction_info_range = tr.snapshot.get_range(start_key, end_key,
|
||||
streaming_mode=fdb.impl.StreamingMode.want_all)
|
||||
for k, v in transaction_info_range:
|
||||
found += 1
|
||||
#logger.debug(k)
|
||||
start_key = fdb.KeySelector.first_greater_than(k)
|
||||
|
||||
_, tr_id, num_chunks, chunk_num = self.parse_key(k)
|
||||
|
||||
#logger.debug("num_chunks=%d, chunk_num=%d" % (num_chunks,chunk_num))
|
||||
|
||||
if num_chunks == 1:
|
||||
assert chunk_num == 1
|
||||
try:
|
||||
info = build_client_transaction_info(v)
|
||||
if info.has_types():
|
||||
buffer.append(info)
|
||||
valid_transaction_infos += 1
|
||||
except UnsupportedProtocolVersionError as e:
|
||||
invalid_transaction_infos += 1
|
||||
except ValueError:
|
||||
invalid_transaction_infos += 1
|
||||
else:
|
||||
if chunk_num == 1:
|
||||
# first chunk
|
||||
assert tr_id not in self.tr_info_map
|
||||
self.tr_info_map[tr_id] = [TrInfoChunk(num_chunks, chunk_num, k, v)]
|
||||
self.num_chunks_stored += 1
|
||||
self._check_and_adjust_chunk_cache_size()
|
||||
else:
|
||||
if tr_id not in self.tr_info_map:
|
||||
logger.error("Got a middle chunk without getting beginning part. Discarding transaction id: %s\n" % tr_id)
|
||||
continue
|
||||
c_list = self.tr_info_map[tr_id]
|
||||
if c_list[-1].num_chunks != num_chunks or c_list[-1].chunk_num != chunk_num - 1:
|
||||
self.tr_info_map.pop(tr_id)
|
||||
self.num_chunks_stored -= len(c_list)
|
||||
raise Exception("Chunk numbers do not match for Transaction id: %s" % tr_id)
|
||||
c_list.append(TrInfoChunk(num_chunks, chunk_num, k, v))
|
||||
self.num_chunks_stored += 1
|
||||
if num_chunks == chunk_num:
|
||||
self.tr_info_map.pop(tr_id)
|
||||
self.num_chunks_stored -= len(c_list)
|
||||
try:
|
||||
info = build_client_transaction_info(b''.join([chunk.value for chunk in c_list]))
|
||||
if info.has_types():
|
||||
buffer.append(info)
|
||||
valid_transaction_infos += 1
|
||||
except UnsupportedProtocolVersionError as e:
|
||||
invalid_transaction_infos += 1
|
||||
except ValueError:
|
||||
invalid_transaction_infos += 1
|
||||
self._check_and_adjust_chunk_cache_size()
|
||||
if (valid_transaction_infos + invalid_transaction_infos) % 1000 == 0:
|
||||
print("Processed valid: %d, invalid: %d" % (valid_transaction_infos, invalid_transaction_infos))
|
||||
if found == 0:
|
||||
more = False
|
||||
except fdb.FDBError as e:
|
||||
# if too old then reset and don't wait
|
||||
if e.code == 1007:
|
||||
tr.reset()
|
||||
else:
|
||||
tr.on_error(e).wait()
|
||||
for item in buffer:
|
||||
yield item
|
||||
|
||||
|
||||
def has_sortedcontainers():
|
||||
try:
|
||||
import sortedcontainers
|
||||
return True
|
||||
except ImportError:
|
||||
logger.warn("Can't find sortedcontainers so disabling RangeCounter")
|
||||
return False
|
||||
|
||||
|
||||
def has_dateparser():
|
||||
try:
|
||||
import dateparser
|
||||
return True
|
||||
except ImportError:
|
||||
logger.warn("Can't find dateparser so disabling human date parsing")
|
||||
return False
|
||||
|
||||
|
||||
class RangeCounter(object):
|
||||
def __init__(self, k):
|
||||
self.k = k
|
||||
from sortedcontainers import SortedDict
|
||||
self.ranges = SortedDict()
|
||||
|
||||
def process(self, transaction_info):
|
||||
for get_range in transaction_info.get_ranges:
|
||||
self._insert_range(get_range.key_range.start_key, get_range.key_range.end_key)
|
||||
|
||||
def _insert_range(self, start_key, end_key):
|
||||
keys = self.ranges.keys()
|
||||
if len(keys) == 0:
|
||||
self.ranges[start_key] = end_key, 1
|
||||
return
|
||||
|
||||
start_pos = bisect_left(keys, start_key)
|
||||
end_pos = bisect_left(keys, end_key)
|
||||
#print("start_pos=%d, end_pos=%d" % (start_pos, end_pos))
|
||||
|
||||
possible_intersection_keys = keys[max(0, start_pos - 1):min(len(keys), end_pos+1)]
|
||||
|
||||
start_range_left = start_key
|
||||
|
||||
for key in possible_intersection_keys:
|
||||
cur_end_key, cur_count = self.ranges[key]
|
||||
#logger.debug("key=%s, cur_end_key=%s, cur_count=%d, start_range_left=%s" % (key, cur_end_key, cur_count, start_range_left))
|
||||
if start_range_left < key:
|
||||
if end_key <= key:
|
||||
self.ranges[start_range_left] = end_key, 1
|
||||
return
|
||||
self.ranges[start_range_left] = key, 1
|
||||
start_range_left = key
|
||||
assert start_range_left >= key
|
||||
if start_range_left >= cur_end_key:
|
||||
continue
|
||||
|
||||
# [key, start_range_left) = cur_count
|
||||
# if key == start_range_left this will get overwritten below
|
||||
self.ranges[key] = start_range_left, cur_count
|
||||
|
||||
if end_key <= cur_end_key:
|
||||
# [start_range_left, end_key) = cur_count+1
|
||||
# [end_key, cur_end_key) = cur_count
|
||||
self.ranges[start_range_left] = end_key, cur_count + 1
|
||||
if end_key != cur_end_key:
|
||||
self.ranges[end_key] = cur_end_key, cur_count
|
||||
start_range_left = end_key
|
||||
break
|
||||
else:
|
||||
# [start_range_left, cur_end_key) = cur_count+1
|
||||
self.ranges[start_range_left] = cur_end_key, cur_count+1
|
||||
start_range_left = cur_end_key
|
||||
assert start_range_left <= end_key
|
||||
|
||||
# there may be some range left
|
||||
if start_range_left < end_key:
|
||||
self.ranges[start_range_left] = end_key, 1
|
||||
|
||||
def get_count_for_key(self, key):
|
||||
if key in self.ranges:
|
||||
return self.ranges[key][1]
|
||||
|
||||
keys = self.ranges.keys()
|
||||
index = bisect_left(keys, key)
|
||||
if index == 0:
|
||||
return 0
|
||||
|
||||
index_key = keys[index-1]
|
||||
if index_key <= key < self.ranges[index_key][0]:
|
||||
return self.ranges[index_key][1]
|
||||
return 0
|
||||
|
||||
def get_range_boundaries(self, shard_finder=None):
|
||||
total = sum([count for _, (_, count) in self.ranges.items()])
|
||||
range_size = total // self.k
|
||||
output_range_counts = []
|
||||
|
||||
def add_boundary(start, end, count):
|
||||
if shard_finder:
|
||||
shard_count = shard_finder.get_shard_count(start, end)
|
||||
if shard_count == 1:
|
||||
addresses = shard_finder.get_addresses_for_key(start)
|
||||
else:
|
||||
addresses = None
|
||||
output_range_counts.append((start, end, count, shard_count, addresses))
|
||||
else:
|
||||
output_range_counts.append((start, end, count, None, None))
|
||||
|
||||
this_range_start_key = None
|
||||
count_this_range = 0
|
||||
for (start_key, (end_key, count)) in self.ranges.items():
|
||||
if not this_range_start_key:
|
||||
this_range_start_key = start_key
|
||||
count_this_range += count
|
||||
if count_this_range >= range_size:
|
||||
add_boundary(this_range_start_key, end_key, count_this_range)
|
||||
count_this_range = 0
|
||||
this_range_start_key = None
|
||||
if count_this_range > 0:
|
||||
add_boundary(this_range_start_key, end_key, count_this_range)
|
||||
|
||||
return output_range_counts
|
||||
|
||||
|
||||
class ShardFinder(object):
|
||||
def __init__(self, db):
|
||||
self.db = db
|
||||
|
||||
@staticmethod
|
||||
@fdb.transactional
|
||||
def _get_boundary_keys(tr, begin, end):
|
||||
tr.options.set_read_lock_aware()
|
||||
return fdb.locality.get_boundary_keys(tr, begin, end)
|
||||
|
||||
@staticmethod
|
||||
@fdb.transactional
|
||||
def _get_addresses_for_key(tr, key):
|
||||
tr.options.set_read_lock_aware()
|
||||
return fdb.locality.get_addresses_for_key(tr, key)
|
||||
|
||||
def get_shard_count(self, start_key, end_key):
|
||||
return len(list(self._get_boundary_keys(self.db, start_key, end_key))) + 1
|
||||
|
||||
def get_addresses_for_key(self, key):
|
||||
return [a.decode('ascii') for a in self._get_addresses_for_key(self.db, key).wait()]
|
||||
|
||||
|
||||
class TopKeysCounter(object):
|
||||
mutation_types_to_consider = frozenset([MutationType.SET_VALUE, MutationType.ADD_VALUE])
|
||||
|
||||
def __init__(self, k):
|
||||
self.k = k
|
||||
self.reads = defaultdict(lambda: 0)
|
||||
self.writes = defaultdict(lambda: 0)
|
||||
|
||||
def process(self, transaction_info):
|
||||
for get in transaction_info.gets:
|
||||
self.reads[get.key] += 1
|
||||
if transaction_info.commit:
|
||||
for mutation in transaction_info.commit.mutations:
|
||||
if mutation.code in self.mutation_types_to_consider:
|
||||
self.writes[mutation.param_one] += 1
|
||||
|
||||
def _get_range_boundaries(self, counts, shard_finder=None):
|
||||
total = sum([v for (k, v) in counts.items()])
|
||||
range_size = total // self.k
|
||||
key_counts_sorted = sorted(counts.items())
|
||||
output_range_counts = []
|
||||
|
||||
def add_boundary(start, end, count):
|
||||
if shard_finder:
|
||||
shard_count = shard_finder.get_shard_count(start, end)
|
||||
if shard_count == 1:
|
||||
addresses = shard_finder.get_addresses_for_key(start)
|
||||
else:
|
||||
addresses = None
|
||||
output_range_counts.append((start, end, count, shard_count, addresses))
|
||||
else:
|
||||
output_range_counts.append((start, end, count, None, None))
|
||||
|
||||
start_key = None
|
||||
count_this_range = 0
|
||||
for (k, v) in key_counts_sorted:
|
||||
if not start_key:
|
||||
start_key = k
|
||||
count_this_range += v
|
||||
if count_this_range >= range_size:
|
||||
add_boundary(start_key, k, count_this_range)
|
||||
count_this_range = 0
|
||||
start_key = None
|
||||
if count_this_range > 0:
|
||||
add_boundary(start_key, k, count_this_range)
|
||||
|
||||
return output_range_counts
|
||||
|
||||
def _get_top_k(self, counts):
|
||||
count_key_pairs = sorted([(v, k) for (k, v) in counts.items()], reverse=True)
|
||||
return count_key_pairs[0:self.k]
|
||||
|
||||
def get_top_k_reads(self):
|
||||
return self._get_top_k(self.reads)
|
||||
|
||||
def get_top_k_writes(self):
|
||||
return self._get_top_k(self.writes)
|
||||
|
||||
def get_k_read_range_boundaries(self, shard_finder=None):
|
||||
return self._get_range_boundaries(self.reads, shard_finder)
|
||||
|
||||
def get_k_write_range_boundaries(self, shard_finder=None):
|
||||
return self._get_range_boundaries(self.writes, shard_finder)
|
||||
|
||||
|
||||
def connect(cluster_file=None):
|
||||
db = fdb.open(cluster_file=cluster_file)
|
||||
return db
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="TransactionProfilingAnalyzer")
|
||||
parser.add_argument("-C", "--cluster-file", type=str, help="Cluster file")
|
||||
parser.add_argument("--full-output", action="store_true", help="Print full output from mutations")
|
||||
parser.add_argument("--filter-get-version", action="store_true",
|
||||
help="Include get_version type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-get", action="store_true",
|
||||
help="Include get type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-get-range", action="store_true",
|
||||
help="Include get_range type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-commit", action="store_true",
|
||||
help="Include commit type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-error-get", action="store_true",
|
||||
help="Include error_get type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-error-get-range", action="store_true",
|
||||
help="Include error_get_range type. If no filter args are given all will be returned.")
|
||||
parser.add_argument("--filter-error-commit", action="store_true",
|
||||
help="Include error_commit type. If no filter args are given all will be returned.")
|
||||
start_time_group = parser.add_mutually_exclusive_group()
|
||||
start_time_group.add_argument("--min-timestamp", type=int, help="Don't return events older than this epoch time")
|
||||
start_time_group.add_argument("-s", "--start-time", type=str,
|
||||
help="Don't return events older than this parsed time")
|
||||
end_time_group = parser.add_mutually_exclusive_group()
|
||||
end_time_group.add_argument("--max-timestamp", type=int, help="Don't return events newer than this epoch time")
|
||||
end_time_group.add_argument("-e", "--end-time", type=str, help="Don't return events older than this parsed time")
|
||||
parser.add_argument("--top-keys", type=int, help="If specified will output this many top keys for reads or writes", default=0)
|
||||
args = parser.parse_args()
|
||||
|
||||
type_filter = set()
|
||||
if args.filter_get_version: type_filter.add("get_version")
|
||||
if args.filter_get: type_filter.add("get")
|
||||
if args.filter_get_range: type_filter.add("get_range")
|
||||
if args.filter_commit: type_filter.add("commit")
|
||||
if args.filter_error_get: type_filter.add("error_get")
|
||||
if args.filter_error_get_range: type_filter.add("error_get_range")
|
||||
if args.filter_error_commit: type_filter.add("error_commit")
|
||||
top_keys = args.top_keys
|
||||
key_counter = TopKeysCounter(top_keys) if top_keys else None
|
||||
range_counter = RangeCounter(top_keys) if (has_sortedcontainers() and top_keys) else None
|
||||
full_output = args.full_output or (top_keys is not None)
|
||||
|
||||
if args.min_timestamp:
|
||||
min_timestamp = args.min_timestamp
|
||||
elif args.start_time:
|
||||
if not has_dateparser():
|
||||
raise Exception("Can't find dateparser needed to parse human dates")
|
||||
import dateparser
|
||||
min_timestamp = int(dateparser.parse(args.start_time).timestamp())
|
||||
else:
|
||||
raise Exception("Must specify start time")
|
||||
|
||||
if args.max_timestamp:
|
||||
max_timestamp = args.max_timestamp
|
||||
elif args.end_time:
|
||||
if not has_dateparser():
|
||||
raise Exception("Can't find dateparser needed to parse human dates")
|
||||
import dateparser
|
||||
max_timestamp = int(dateparser.parse(args.end_time).timestamp())
|
||||
else:
|
||||
raise Exception("Must specify end time")
|
||||
|
||||
now = time.time()
|
||||
if max_timestamp > now:
|
||||
raise Exception("max_timestamp is %d seconds in the future" % (max_timestamp - now))
|
||||
if min_timestamp > now:
|
||||
raise Exception("min_timestamp is %d seconds in the future" % (min_timestamp - now))
|
||||
|
||||
logger.info("Loading transactions from %d to %d" % (min_timestamp, max_timestamp))
|
||||
|
||||
db = connect(cluster_file=args.cluster_file)
|
||||
loader = TransactionInfoLoader(db, full_output=full_output, type_filter=type_filter,
|
||||
min_timestamp=min_timestamp, max_timestamp=max_timestamp)
|
||||
for info in loader.fetch_transaction_info():
|
||||
if info.has_types():
|
||||
if not key_counter and not range_counter:
|
||||
print(info.to_json())
|
||||
else:
|
||||
if key_counter:
|
||||
key_counter.process(info)
|
||||
if range_counter:
|
||||
range_counter.process(info)
|
||||
|
||||
if key_counter:
|
||||
def print_top(top):
|
||||
for (count, key) in top:
|
||||
print("%s %d" % (key, count))
|
||||
|
||||
def print_range_boundaries(range_boundaries):
|
||||
for (start, end, count, shard_count, addresses) in range_boundaries:
|
||||
if not shard_count:
|
||||
print("[%s, %s] %d" % (start, end, count))
|
||||
else:
|
||||
addresses_string = "addresses=%s" % ','.join(addresses) if addresses else ''
|
||||
print("[%s, %s] %d shards=%d %s" % (start, end, count, shard_count, addresses_string))
|
||||
|
||||
shard_finder = ShardFinder(db)
|
||||
top_reads = key_counter.get_top_k_reads()
|
||||
if top_reads:
|
||||
print("Top %d reads:" % min(top_keys, len(top_reads)))
|
||||
print_top(top_reads)
|
||||
print("Approx equal sized gets range boundaries:")
|
||||
print_range_boundaries(key_counter.get_k_read_range_boundaries(shard_finder=shard_finder))
|
||||
top_writes = key_counter.get_top_k_writes()
|
||||
if top_writes:
|
||||
print("Top %d writes:" % min(top_keys, len(top_writes)))
|
||||
print_top(top_writes)
|
||||
print("Approx equal sized commits range boundaries:")
|
||||
print_range_boundaries(key_counter.get_k_write_range_boundaries(shard_finder=shard_finder))
|
||||
if range_counter:
|
||||
range_boundaries = range_counter.get_range_boundaries(shard_finder=shard_finder)
|
||||
if range_boundaries:
|
||||
print("Approx equal sized get_ranges boundaries:")
|
||||
print_range_boundaries(range_boundaries)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Loading…
Reference in New Issue