Skip to content

KAFKA-7834: Extend collected logs in system test services to include heap dumps #6158

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

Closed
kkonstantine wants to merge 4 commits into from
Closed

KAFKA-7834: Extend collected logs in system test services to include heap dumps #6158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
68af2bc
KAFKA-7834: Enable and collect heap dump for ZK system test service
kkonstantine Jan 16, 2019
3286d69
KAFKA-7834: Enable and collect heap dump for Kafka system test service
kkonstantine Jan 16, 2019
c30cf90
KAFKA-7834: Enable and collect heap dump for Connect system test service
kkonstantine Jan 16, 2019
b964036
KAFKA-7834: Change Connect service to delete its root directory
kkonstantine Jan 16, 2019
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
26 changes: 22 additions & 4 deletions tests/kafkatest/services/connect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ class ConnectServiceBase(KafkaPathResolverMixin, Service):
PID_FILE = os.path.join(PERSISTENT_ROOT, "connect.pid")
EXTERNAL_CONFIGS_FILE = os.path.join(PERSISTENT_ROOT, "connect-external-configs.properties")
CONNECT_REST_PORT = 8083
HEAP_DUMP_FILE = os.path.join(PERSISTENT_ROOT, "connect_heap_dump.bin")

# Currently the Connect worker supports waiting on three modes:
STARTUP_MODE_INSTANT = 'INSTANT'
Expand All @@ -61,6 +62,9 @@ class ConnectServiceBase(KafkaPathResolverMixin, Service):
"connect_stderr": {
"path": STDERR_FILE,
"collect_default": True},
"connect_heap_dump_file": {
"path": HEAP_DUMP_FILE,
"collect_default": True}
}

def __init__(self, context, num_nodes, kafka, files, startup_timeout_sec = 60):
Expand Down Expand Up @@ -160,8 +164,8 @@ def restart_node(self, node, clean_shutdown=True):
def clean_node(self, node):
node.account.kill_process("connect", clean_shutdown=False, allow_fail=True)
self.security_config.clean_node(node)
all_files = " ".join([self.CONFIG_FILE, self.LOG4J_CONFIG_FILE, self.PID_FILE, self.LOG_FILE, self.STDOUT_FILE, self.STDERR_FILE, self.EXTERNAL_CONFIGS_FILE] + self.config_filenames() + self.files)
node.account.ssh("rm -rf " + all_files, allow_fail=False)
other_files = " ".join(self.config_filenames() + self.files)
node.account.ssh("rm -rf -- %s %s" % (ConnectServiceBase.PERSISTENT_ROOT, other_files), allow_fail=False)

@ewencp ewencp Jan 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

are these all under PERSISTENT_ROOT? if yes, can we just handle PERSISTENT_ROOT and not worry about individual files?

@kkonstantine kkonstantine Jan 23, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the other_files are not necessarily under PERSISTENT_ROOT
What is under it for sure is:
self.CONFIG_FILE, self.LOG4J_CONFIG_FILE, self.PID_FILE, self.LOG_FILE, self.STDOUT_FILE, self.STDERR_FILE, self.EXTERNAL_CONFIGS_FILE which I removed from explicit listing and I'm removing with PERSISTENT_ROOT altogether.

@ewencp ewencp Feb 4, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I get that though I think that's just tech debt. for any test related files, we really shouldn't be using anything other than PERSISTENT_ROOT so that we can at least attempt to ensure each test/service gets a clean workspace


def config_filenames(self):
return [os.path.join(self.PERSISTENT_ROOT, "connect-connector-" + str(idx) + ".properties") for idx, template in enumerate(self.connector_config_templates or [])]
Expand Down Expand Up @@ -252,6 +256,14 @@ def _rest_with_retry(self, path, body=None, node=None, method="GET", retries=40,
def _base_url(self, node):
return 'http://' + node.account.externally_routable_ip + ':' + str(self.CONNECT_REST_PORT)

def append_to_environment_variable(self, envvar, value):

@ewencp ewencp Jan 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what is the logic for this, including the different ways env_opts is being encoded?

@kkonstantine kkonstantine Jan 23, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I ended up adding the heap jvm arguments in KAFKA_OPTS instead of KAFKA_HEAP_OPTS because the latter has defaults for sizes if it's empty. However, if I needed to add values to an env variable, I can't just append them to it's current value, because the value is wrapped with \"

That's why I'm adding this method - unused here - that can be used by tests that inherit this class. Let me know if you think I should remove.

@ewencp ewencp Feb 5, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah, in general the construction of commands could be cleaned up quite a bit for extensibility -- i think we tend to forget about the importance/value of doing this when writing tests in the Kafka repo, but these classes should be reusable and extensible. this approach seems pretty error-prone, although at least it should consistently fail if the quoting gets messed up since you'll have invalid extra args on the command line.

env_opts = self.environment[envvar]
if env_opts is None:
env_opts = "\"%s\"" % value
else:
env_opts = "\"%s %s\"" % (env_opts.strip('\"'), value)
self.environment[envvar] = env_opts


class ConnectStandaloneService(ConnectServiceBase):
"""Runs Kafka Connect in standalone mode."""
Expand All @@ -266,7 +278,10 @@ def node(self):

def start_cmd(self, node, connector_configs):
cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE
cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts
heap_kafka_opts = "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s" % \

@ewencp ewencp Jan 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we seem to duplicate this pattern of setting constants, is there any opportunity to relocate and reduce to one copy of these constants (given they should be same across all our java services)?

@kkonstantine kkonstantine Jan 23, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With each service setting its security configs in KAFKA_OPTS using their own way, this seems difficult at the moment. I'd prefer if I could avoid to extend this PR to a general refactoring at this point.

self.logs["connect_heap_dump_file"]["path"]
other_kafka_opts = self.security_config.kafka_opts.strip('\"')

@ewencp ewencp Jan 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what's happening with the various strip('\"') calls? why do we need this process? can we refactor to something that makes it clearer what's happening?

@kkonstantine kkonstantine Jan 23, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As I mention above, we often wrap the value of an env var in escaped double quotes. To append contents, we first need to remove those. If they don't exist, strip is a no-op.

cmd += "export KAFKA_OPTS=\"%s %s\"; " % (heap_kafka_opts, other_kafka_opts)
for envvar in self.environment:
cmd += "export %s=%s; " % (envvar, str(self.environment[envvar]))
cmd += "%s %s " % (self.path.script("connect-standalone.sh", node), self.CONFIG_FILE)
Expand Down Expand Up @@ -314,7 +329,10 @@ def __init__(self, context, num_nodes, kafka, files, offsets_topic="connect-offs
# connector_configs argument is intentionally ignored in distributed service.
def start_cmd(self, node, connector_configs):
cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE
cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts
heap_kafka_opts = "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s" % \
self.logs["connect_heap_dump_file"]["path"]
other_kafka_opts = self.security_config.kafka_opts.strip('\"')
cmd += "export KAFKA_OPTS=\"%s %s\"; " % (heap_kafka_opts, other_kafka_opts)
for envvar in self.environment:
cmd += "export %s=%s; " % (envvar, str(self.environment[envvar]))
cmd += "%s %s " % (self.path.script("connect-distributed.sh", node), self.CONFIG_FILE)
Expand Down
11 changes: 9 additions & 2 deletions tests/kafkatest/services/kafka/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "kafka.properties")
# Kafka Authorizer
SIMPLE_AUTHORIZER = "kafka.security.auth.SimpleAclAuthorizer"
HEAP_DUMP_FILE = os.path.join(PERSISTENT_ROOT, "kafka_heap_dump.bin")

logs = {
"kafka_server_start_stdout_stderr": {
Expand All @@ -65,7 +66,10 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
"collect_default": False},
"kafka_data_2": {
"path": DATA_LOG_DIR_2,
"collect_default": False}
"collect_default": False},
"kafka_heap_dump_file": {
"path": HEAP_DUMP_FILE,
"collect_default": True}
}

def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAINTEXT, interbroker_security_protocol=SecurityConfig.PLAINTEXT,
Expand Down Expand Up @@ -247,7 +251,10 @@ def render_configs(self, configs):
def start_cmd(self, node):
cmd = "export JMX_PORT=%d; " % self.jmx_port
cmd += "export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG
cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts
heap_kafka_opts = "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s" % \
self.logs["kafka_heap_dump_file"]["path"]
other_kafka_opts = self.security_config.kafka_opts.strip('\"')
cmd += "export KAFKA_OPTS=\"%s %s\"; " % (heap_kafka_opts, other_kafka_opts)
cmd += "%s %s 1>> %s 2>> %s &" % \
(self.path.script("kafka-server-start.sh", node),
KafkaService.CONFIG_FILE,
Expand Down
10 changes: 8 additions & 2 deletions tests/kafkatest/services/zookeeper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,18 @@
class ZookeeperService(KafkaPathResolverMixin, Service):
ROOT = "/mnt/zookeeper"
DATA = os.path.join(ROOT, "data")
HEAP_DUMP_FILE = os.path.join(ROOT, "zk_heap_dump.bin")

logs = {
"zk_log": {
"path": "%s/zk.log" % ROOT,
"collect_default": True},
"zk_data": {
"path": DATA,
"collect_default": False}
"collect_default": False},
"zk_heap_dump_file": {
"path": HEAP_DUMP_FILE,
"collect_default": True}
}

def __init__(self, context, num_nodes, zk_sasl = False):
Expand Down Expand Up @@ -76,8 +80,10 @@ def start_node(self, node):
self.logger.info(config_file)
node.account.create_file("%s/zookeeper.properties" % ZookeeperService.ROOT, config_file)

start_cmd = "export KAFKA_OPTS=\"%s\";" % (self.kafka_opts + ' ' + self.security_system_properties) \
heap_kafka_opts = "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s" % self.logs["zk_heap_dump_file"]["path"]
other_kafka_opts = self.kafka_opts + ' ' + self.security_system_properties \
if self.security_config.zk_sasl else self.kafka_opts
start_cmd = "export KAFKA_OPTS=\"%s %s\";" % (heap_kafka_opts, other_kafka_opts)
start_cmd += "%s " % self.path.script("zookeeper-server-start.sh", node)
start_cmd += "%s/zookeeper.properties &>> %s &" % (ZookeeperService.ROOT, self.logs["zk_log"]["path"])
node.account.ssh(start_cmd)
Expand Down