-
Notifications
You must be signed in to change notification settings - Fork 32
/
concurrent_select.py
executable file
·2423 lines (2191 loc) · 106 KB
/
concurrent_select.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env impala-python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# This module is used to stress test Impala by running queries concurrently.
#
# Stress test outline (and notes):
# 1) Get a set of queries as requested by the user from the CLI options.
# 2) For each query, run it individually to find:
# a) Minimum mem limit to avoid spilling
# b) Minimum mem limit to successfully run the query (spilling allowed)
# c) Runtime when no mem was spilled
# d) Runtime when mem was spilled
# e) A row order independent hash of the result set.
# This is a slow process so the results will be written to disk for reuse.
# 3) Find the memory available to Impalad. This will be done by finding the minimum
# memory available across all impalads (-mem_limit startup option). Ideally, for
# maximum stress, all impalads will have the same memory configuration but this is
# not required.
# 4) Optionally, set an amount of memory that can be overcommitted. Overcommitting
# memory can increase memory pressure which can result in memory being spilled to
# disk or queries failing with out-of-memory.
# 5) Start submitting queries. There are two modes for throttling the number of
# concurrent queries, depending on --test-admission-control.
# a) test-admission-control=false: Submit queries until all available memory (as
# determined by items 3 and 4) is used. Before running the query a query mem
# limit is set between 2a and 2b. (There is a runtime option to increase the
# likelihood that a query will be given the full 2a limit to avoid spilling.)
# b) test-admission-control=true: Submit enough queries to achieve the desired
# level of overcommit, but expect that Impala's admission control will throttle
# queries. In this mode mem_limit is not set per query.
# 6) Randomly cancel queries to test cancellation. There is a runtime option to control
# the likelihood that a query will be randomly canceled.
# 7) If a query errored, verify that the error is expected. Errors are expected in the
# following cases:
# a) Memory-based admission control is not being tested (i.e.
# --test-admission-control=false), the error is an out-of-memory error and memory
# on the cluster is overcommitted.
# b) The error is an admission control rejection or timeout.
# 8) Verify the result set hash of successful queries if there are no DML queries in the
# current run.
from __future__ import print_function
import json
import logging
import os
import re
import signal
import sys
import threading
import traceback
from Queue import Empty # Must be before Queue below
from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace, SUPPRESS
from collections import defaultdict
from contextlib import contextmanager
from copy import copy
from datetime import datetime
from multiprocessing import Lock, Process, Queue, Value
from random import choice, random, randrange, shuffle
from sys import exit, maxint
from tempfile import gettempdir
from textwrap import dedent
from threading import current_thread, Thread
from time import sleep, time
import tests.comparison.cli_options as cli_options
import tests.util.test_file_parser as test_file_parser
from tests.comparison.cluster import Timeout
from tests.comparison.db_types import Int, TinyInt, SmallInt, BigInt
from tests.comparison.model_translator import SqlWriter
from tests.comparison.query_generator import QueryGenerator
from tests.comparison.query_profile import DefaultProfile
from tests.util.parse_util import (
EXPECTED_TPCDS_QUERIES_COUNT, EXPECTED_TPCH_NESTED_QUERIES_COUNT,
EXPECTED_TPCH_STRESS_QUERIES_COUNT, match_memory_estimate, parse_mem_to_mb)
from tests.util.thrift_util import op_handle_to_query_id
LOG = logging.getLogger(os.path.splitext(os.path.basename(__file__))[0])
PROFILES_DIR = "profiles"
RESULT_HASHES_DIR = "result_hashes"
# The version of the file format containing the collected query runtime info.
RUNTIME_INFO_FILE_VERSION = 3
# Metrics collected during the stress running process.
NUM_QUERIES_DEQUEUED = "num_queries_dequeued"
# The number of queries that were submitted to a query runner.
NUM_QUERIES_SUBMITTED = "num_queries_submitted"
# The number of queries that have entered the RUNNING state (i.e. got through Impala's
# admission control and started executing) or were cancelled or hit an error.
NUM_QUERIES_STARTED_RUNNING_OR_CANCELLED = "num_queries_started_running_or_cancelled"
NUM_QUERIES_FINISHED = "num_queries_finished"
NUM_QUERIES_EXCEEDED_MEM_LIMIT = "num_queries_exceeded_mem_limit"
NUM_QUERIES_AC_REJECTED = "num_queries_ac_rejected"
NUM_QUERIES_AC_TIMEDOUT = "num_queries_ac_timedout"
NUM_QUERIES_CANCELLED = "num_queries_cancelled"
NUM_RESULT_MISMATCHES = "num_result_mismatches"
NUM_OTHER_ERRORS = "num_other_errors"
class StressArgConverter(object):
def __init__(self, args):
"""
Convert arguments as returned from from argparse parse_args() into internal forms.
The purpose of this object is to do any conversions needed from the type given by
parge_args() into internal forms. For example, if a commandline option takes in a
complicated string that needs to be converted into a list or dictionary, this is the
place to do it. Access works the same as on the object returned by parse_args(),
i.e., object.option_attribute.
In most cases, simple arguments needn't be converted, because argparse handles the
type conversion already, and in most cases, type conversion (e.g., "8" <str> to 8
<int>) is all that's needed. If a property getter below doesn't exist, it means the
argument value is just passed along unconverted.
Params:
args: argparse.Namespace object (from argparse.ArgumentParser().parse_args())
"""
assert isinstance(args, Namespace), "expected Namespace, got " + str(type(args))
self._args = args
self._common_query_options = None
def __getattr__(self, attr):
# This "proxies through" all the attributes from the Namespace object that are not
# defined in this object via property getters below.
return getattr(self._args, attr)
@property
def common_query_options(self):
# Memoize this, as the integrity checking of --common-query-options need only
# happen once.
if self._common_query_options is not None:
return self._common_query_options
# The stress test sets these, so callers cannot override them.
IGNORE_QUERY_OPTIONS = frozenset([
'ABORT_ON_ERROR',
'MEM_LIMIT',
])
common_query_options = {}
if self._args.common_query_options is not None:
for query_option_and_value in self._args.common_query_options:
try:
query_option, value = query_option_and_value.split('=')
except ValueError:
LOG.error(
"Could not parse --common-query-options: '{common_query_options}'".format(
common_query_options=self._args.common_query_options))
exit(1)
query_option = query_option.upper()
if query_option in common_query_options:
LOG.error(
"Query option '{query_option}' already defined in --common-query-options: "
"'{common_query_options}'".format(
query_option=query_option,
common_query_options=self._args.common_query_options))
exit(1)
elif query_option in IGNORE_QUERY_OPTIONS:
LOG.warn(
"Ignoring '{query_option}' in common query options: '{opt}': "
"The stress test algorithm needs control of this option.".format(
query_option=query_option, opt=self._args.common_query_options))
else:
common_query_options[query_option] = value
LOG.debug("Common query option '{query_option}' set to '{value}'".format(
query_option=query_option, value=value))
self._common_query_options = common_query_options
return self._common_query_options
@property
def runtime_info_path(self):
runtime_info_path = self._args.runtime_info_path
if "{cm_host}" in runtime_info_path:
runtime_info_path = runtime_info_path.format(cm_host=self._args.cm_host)
return runtime_info_path
def create_and_start_daemon_thread(fn, name):
thread = Thread(target=fn, name=name)
thread.error = None
thread.daemon = True
thread.start()
return thread
def increment(counter):
with counter.get_lock():
counter.value += 1
def print_stacks(*_):
"""Print the stacks of all threads from this script to stderr."""
thread_names = dict([(t.ident, t.name) for t in threading.enumerate()])
stacks = list()
for thread_id, stack in sys._current_frames().items():
stacks.append(
"\n# Thread: %s(%d)"
% (thread_names.get(thread_id, "No name"), thread_id))
for filename, lineno, name, line in traceback.extract_stack(stack):
stacks.append('File: "%s", line %d, in %s' % (filename, lineno, name))
if line:
stacks.append(" %s" % (line.strip(), ))
print("\n".join(stacks), file=sys.stderr)
# To help debug hangs, the stacks of all threads can be printed by sending signal USR1
# to each process.
signal.signal(signal.SIGUSR1, print_stacks)
def print_crash_info_if_exists(impala, start_time):
"""If any impalads are found not running, they will assumed to have crashed and an
error message will be printed to stderr for each stopped impalad. Returns a value
that evaluates to True if any impalads are stopped.
"""
max_attempts = 5
for remaining_attempts in xrange(max_attempts - 1, -1, -1):
try:
crashed_impalads = impala.find_crashed_impalads(start_time)
break
except Timeout as e:
LOG.info(
"Timeout checking if impalads crashed: %s."
% e + (" Will retry." if remaining_attempts else ""))
else:
LOG.error(
"Aborting after %s failed attempts to check if impalads crashed", max_attempts)
raise e
for message in crashed_impalads.itervalues():
print(message, file=sys.stderr)
return crashed_impalads
class QueryReport(object):
"""Holds information about a single query run."""
def __init__(self, query):
self.query = query
self.result_hash = None
self.runtime_secs = None
self.mem_was_spilled = False
# not_enough_memory includes conditions like "Memory limit exceeded", admission
# control rejecting because not enough memory, etc.
self.not_enough_memory = False
# ac_rejected is true if the query was rejected by admission control.
# It is mutually exclusive with not_enough_memory - if the query is rejected by
# admission control because the memory limit is too low, it is counted as
# not_enough_memory.
# TODO: reconsider whether they should be mutually exclusive
self.ac_rejected = False
self.ac_timedout = False
self.other_error = None
self.timed_out = False
self.was_cancelled = False
self.profile = None
self.query_id = None
def __str__(self):
return dedent("""
<QueryReport
result_hash: %(result_hash)s
runtime_secs: %(runtime_secs)s
mem_was_spilled: %(mem_was_spilled)s
not_enough_memory: %(not_enough_memory)s
ac_rejected: %(ac_rejected)s
ac_timedout: %(ac_timedout)s
other_error: %(other_error)s
timed_out: %(timed_out)s
was_cancelled: %(was_cancelled)s
query_id: %(query_id)s
>
""".strip() % self.__dict__)
def has_query_error(self):
"""Return true if any kind of error status was returned from the query (i.e.
the query didn't run to completion, time out or get cancelled)."""
return (self.not_enough_memory or self.ac_rejected or self.ac_timedout
or self.other_error)
def write_query_profile(self, directory, prefix=None):
"""
Write out the query profile bound to this object to a given directory.
The file name is generated and will contain the query ID. Use the optional prefix
parameter to set a prefix on the filename.
Example return:
tpcds_300_decimal_parquet_q21_00000001_a38c8331_profile.txt
Parameters:
directory (str): Directory to write profile.
prefix (str): Prefix for filename.
"""
if not (self.profile and self.query_id):
return
if prefix is not None:
file_name = prefix + '_'
else:
file_name = ''
file_name += self.query.logical_query_id + '_'
file_name += self.query_id.replace(":", "_") + "_profile.txt"
profile_log_path = os.path.join(directory, file_name)
with open(profile_log_path, "w") as profile_log:
profile_log.write(self.profile)
class MemBroker(object):
"""Provides memory usage coordination for clients running in different processes.
The broker fulfills reservation requests by blocking as needed so total memory
used by clients never exceeds the total available memory (including an
'overcommitable' amount).
The lock built in to _available is also used to protect access to other members.
The state stored in this class is actually an encapsulation of part of the state
of the StressRunner class below. The state here is separated for clarity.
"""
def __init__(self, real_mem_mb, overcommitable_mem_mb):
"""'real_mem_mb' memory should be the amount of memory that each impalad is able
to use. 'overcommitable_mem_mb' is the amount of memory that will be dispensed
over the 'real' amount.
"""
self._total_mem_mb = real_mem_mb + overcommitable_mem_mb
self._available = Value("i", self._total_mem_mb)
self._max_overcommitment = overcommitable_mem_mb
# Each reservation will be assigned an id. Ids are monotonically increasing. When
# a reservation crosses the overcommitment threshold, the corresponding reservation
# id will be stored in '_last_overcommitted_reservation_id' so clients can check
# to see if memory was overcommitted since their reservation was made (this is a race
# but an incorrect result will be on the conservative side).
self._next_reservation_id = Value("L", 0)
self._last_overcommitted_reservation_id = Value("L", 0)
@property
def total_mem_mb(self):
return self._total_mem_mb
@property
def overcommitted_mem_mb(self):
return max(self._max_overcommitment - self._available.value, 0)
@property
def available_mem_mb(self):
return self._available.value
@property
def last_overcommitted_reservation_id(self):
return self._last_overcommitted_reservation_id.value
@contextmanager
def reserve_mem_mb(self, mem_mb):
"""Blocks until the requested amount of memory is available and taken for the caller.
This function should be used in a 'with' block. The taken memory will
automatically be released when the 'with' context exits. A numeric id is returned
so clients can compare against 'last_overcommitted_reservation_id' to see if
memory was overcommitted since the reservation was obtained.
with broker.reserve_mem_mb(100) as reservation_id:
# Run query using 100 MB of memory
if <query failed>:
# Immediately check broker.was_overcommitted(reservation_id) to see if
# memory was overcommitted.
"""
reservation_id = self._wait_until_reserved(mem_mb)
try:
yield reservation_id
finally:
self._release(mem_mb)
def _wait_until_reserved(self, req):
while True:
with self._available.get_lock():
if req <= self._available.value:
self._available.value -= req
LOG.debug(
"Reserved %s MB; %s MB available; %s MB overcommitted",
req, self._available.value, self.overcommitted_mem_mb)
reservation_id = self._next_reservation_id.value
increment(self._next_reservation_id)
if self.overcommitted_mem_mb > 0:
self._last_overcommitted_reservation_id.value = reservation_id
return reservation_id
sleep(0.1)
def _release(self, req):
with self._available.get_lock():
self._available.value += req
LOG.debug(
"Released %s MB; %s MB available; %s MB overcommitted",
req, self._available.value, self.overcommitted_mem_mb)
def was_overcommitted(self, reservation_id):
"""Returns True if memory was overcommitted since the given reservation was made.
For an accurate return value, this should be called just after the query ends
or while the query is still running.
"""
return reservation_id <= self._last_overcommitted_reservation_id.value
class StressRunner(object):
"""This class contains functionality related to producing/consuming queries for the
purpose of stress testing Impala.
Queries will be executed in separate processes since python threading is limited
to the use of a single CPU.
"""
# This is the point at which the work queue will block because it is full.
WORK_QUEUE_CAPACITY = 10
def __init__(self):
self.use_kerberos = False
self.common_query_options = {}
self.test_admission_control = False
self._mem_broker = None
self._verify_results = True
self._select_probability = None
# Synchronized blocking work queue for producer/consumers.
self._query_queue = Queue(self.WORK_QUEUE_CAPACITY)
# The Value class provides cross-process shared memory.
self._mem_mb_needed_for_next_query = Value("i", 0)
# This lock provides a way to stop new queries from running. This lock must be
# acquired before writing to the NUM_QUERIES_SUBMITTED metric for the query_runner,
# which is incremented before every query submission.Reading NUM_QUERIES_SUBMITTED is
# allowed without taking this lock.
self._submit_query_lock = Lock()
self.leak_check_interval_mins = None
self._next_leak_check_unix_time = Value("i", 0)
self._max_mem_mb_reported_usage = Value("i", -1) # -1 => Unknown
self._max_mem_mb_usage = Value("i", -1) # -1 => Unknown
self.cancel_probability = 0
self.spill_probability = 0
self.startup_queries_per_sec = 1.0
self.num_successive_errors_needed_to_abort = 1
self._num_successive_errors = Value("i", 0)
self.results_dir = gettempdir()
self._status_headers = [
"Done", "Active", "Executing", "Mem Lmt Ex", "AC Reject", "AC Timeout",
"Cancel", "Err", "Incorrect", "Next Qry Mem Lmt",
"Tot Qry Mem Lmt", "Tracked Mem", "RSS Mem"]
self._num_queries_to_run = None
self._query_producer_thread = None
# This lock is used to synchronize access to the '_query_runners' list and also to all
# the '_past_runners*' members.
self._query_runners_lock = Lock()
self._query_runners = []
# These are the cumulative values of all the queries that have started/finished/-
# dequeued, etc. on runners that have already died. Every time we notice that a query
# runner has died, we update these values.
self._past_runner_metrics = defaultdict(lambda: Value("i", 0))
self._query_consumer_thread = None
self._mem_polling_thread = None
def _record_runner_metrics_before_evict(self, query_runner):
""" Before removing 'query_runner' from the self._query_runners list, record its
metrics. Must only be called if 'query_runner' is to be removed from the list.
MUST hold '_query_runners_lock' before calling.
"""
for key, synchronized_val in query_runner._metrics.iteritems():
self._past_runner_metrics[key].value += synchronized_val.value
def _calc_total_runner_metrics(self):
""" Calculate the total of metrics across past and active query runners. """
totals = defaultdict(lambda: 0)
with self._query_runners_lock:
for key in self._past_runner_metrics:
totals[key] = self._past_runner_metrics[key].value
for query_runner in self._query_runners:
for key, synchronized_val in query_runner._metrics.iteritems():
totals[key] += synchronized_val.value
return totals
def _calc_total_runner_metric(self, key):
""" Calculate the total of metric 'key' across past and active query runners. """
with self._query_runners_lock:
return self._calc_total_runner_metric_no_lock(key)
def _calc_total_runner_metric_no_lock(self, key):
""" TODO: Get rid of this function after reformatting how we obtain query indices.
_query_runners_lock MUST be taken before calling this function.
"""
total = self._past_runner_metrics[key].value
for runner in self._query_runners:
total += runner._metrics[key].value
return total
def _total_num_queries_submitted(self):
return self._calc_total_runner_metric(NUM_QUERIES_SUBMITTED)
def _total_num_queries_active(self):
"""The number of queries that are currently active (i.e. submitted to a query runner
and haven't yet completed)."""
metrics = self._calc_total_runner_metrics()
num_running = metrics[NUM_QUERIES_SUBMITTED] - metrics[NUM_QUERIES_FINISHED]
assert num_running >= 0, "The number of running queries is negative"
return num_running
def _num_runners_remaining(self):
return len(self._query_runners)
def run_queries(
self, queries, impala, num_queries_to_run, mem_overcommit_pct, should_print_status,
verify_results, select_probability
):
"""Runs queries randomly chosen from 'queries' and stops after 'num_queries_to_run'
queries have completed. 'select_probability' should be float between 0 and 1, it
determines the likelihood of choosing a select query (as opposed to a DML query,
for example).
Before a query is run, a mem limit will be chosen. 'spill_probability' determines
the likelihood of choosing a mem limit that will cause spilling. To induce
spilling, a value is randomly chosen below the min memory needed to avoid spilling
but above the min memory needed with spilling. So the min/max query memory
requirements must be determined before calling this method.
If 'mem_overcommit_pct' is zero, an exception will be raised if any queries
fail for any reason other than cancellation (controlled by the 'cancel_probability'
property), since each query should have enough memory to run successfully. If
non-zero, failures due to insufficient memory will be ignored if memory was
overcommitted at any time during execution.
If a query completes without error, the result will be verified if 'verify_results'
is True. An error will be raised upon a result mismatch. 'verify_results' should be
false for the case where the expected results are not known in advance, if we are
running DML queries, for example.
"""
# TODO: The state from a previous run should be cleared out. This isn't really a
# problem now because the one caller (main()) never calls a second time.
if self.startup_queries_per_sec <= 0:
raise Exception("Startup queries per second must be positive")
if self.leak_check_interval_mins is not None and self.leak_check_interval_mins <= 0:
raise Exception("Memory leak check interval must be positive")
# If there is a crash, start looking for errors starting from this time.
self.start_time = datetime.now()
self._mem_broker = MemBroker(
impala.min_impalad_mem_mb,
int(impala.min_impalad_mem_mb * mem_overcommit_pct / 100))
self._verify_results = verify_results
self._select_probability = select_probability
# Print the status to show the state before starting.
if should_print_status:
self._print_status(print_header=True)
self._num_queries_to_run = num_queries_to_run
self._start_polling_mem_usage(impala)
self._start_producing_queries(queries)
self._start_consuming_queries(impala)
# Wait for everything to finish.
self._wait_for_test_to_finish(impala, should_print_status)
# And print the final state.
if should_print_status:
self._print_status()
self._check_for_test_failure()
self.print_duration()
def _start_producing_queries(self, queries):
def enqueue_queries():
# Generate a dict(query type -> list of queries).
queries_by_type = {}
for query in queries:
if query.query_type not in queries_by_type:
queries_by_type[query.query_type] = []
queries_by_type[query.query_type].append(query)
try:
for _ in xrange(self._num_queries_to_run):
# First randomly determine a query type, then choose a random query of that
# type.
if (
QueryType.SELECT in queries_by_type and
(len(queries_by_type.keys()) == 1 or random() < self._select_probability)
):
result = choice(queries_by_type[QueryType.SELECT])
else:
query_type = choice([
key for key in queries_by_type if key != QueryType.SELECT])
result = choice(queries_by_type[query_type])
self._query_queue.put(result)
except Exception as e:
LOG.error("Error producing queries: %s", e)
current_thread().error = e
raise e
LOG.info("Producing thread completed job. Exiting...")
self._query_producer_thread = create_and_start_daemon_thread(
enqueue_queries, "Query Producer")
def _start_consuming_queries(self, impala):
def start_additional_runners_if_needed():
try:
while self._total_num_queries_submitted() < self._num_queries_to_run:
# TODO: sleeping for the below amount leads to slower submission than the goal,
# because it does not factor in the time spent by this thread outside of the
# sleep() call.
sleep(1.0 / self.startup_queries_per_sec)
# Remember num dequeued/started are cumulative.
with self._submit_query_lock:
metrics = self._calc_total_runner_metrics()
num_dequeued = metrics[NUM_QUERIES_DEQUEUED]
num_submitted = metrics[NUM_QUERIES_SUBMITTED]
LOG.debug("Submitted {0} queries. Dequeued {1} queries".format(
num_submitted, num_dequeued))
if num_dequeued != num_submitted:
# Assume dequeued queries are stuck waiting for cluster resources so there
# is no point in starting an additional runner.
continue
num_coordinators = len(impala.impalads)
if self.max_coordinators > 0:
num_coordinators = min(num_coordinators, self.max_coordinators)
impalad = impala.impalads[len(self._query_runners) % num_coordinators]
query_runner = QueryRunner()
query_runner.impalad = impalad
query_runner.results_dir = self.results_dir
query_runner.use_kerberos = self.use_kerberos
query_runner.common_query_options = self.common_query_options
query_runner.test_admission_control = self.test_admission_control
query_runner.proc = \
Process(target=self._start_single_runner, args=(query_runner, ))
query_runner.proc.daemon = True
with self._query_runners_lock:
self._query_runners.append(query_runner)
query_runner.proc.start()
LOG.info("Consuming thread completed job. Exiting...")
except Exception as e:
LOG.error("Error consuming queries: %s", e)
current_thread().error = e
raise e
self._query_consumer_thread = create_and_start_daemon_thread(
start_additional_runners_if_needed, "Query Consumer")
def _start_polling_mem_usage(self, impala):
def poll_mem_usage():
if self.leak_check_interval_mins:
self._next_leak_check_unix_time.value = int(
time() + 60 * self.leak_check_interval_mins)
query_submission_is_locked = False
# Query submission will be unlocked after a memory report has been collected twice
# while no queries were running.
ready_to_unlock = None
try:
while self._total_num_queries_submitted() < self._num_queries_to_run:
if ready_to_unlock:
assert query_submission_is_locked, "Query submission not yet locked"
assert not self._total_num_queries_active(), "Queries are still running"
LOG.debug("Resuming query submission")
self._next_leak_check_unix_time.value = int(
time() + 60 * self.leak_check_interval_mins)
self._submit_query_lock.release()
query_submission_is_locked = False
ready_to_unlock = None
if (
not query_submission_is_locked and
self.leak_check_interval_mins and
time() > self._next_leak_check_unix_time.value
):
assert self._total_num_queries_active() <= self._num_runners_remaining(), \
"Each running query should belong to a runner"
LOG.debug("Stopping query submission")
self._submit_query_lock.acquire()
query_submission_is_locked = True
max_reported, max_actual = self._get_mem_usage_values()
if max_reported != -1 and max_actual != -1:
# Value were already retrieved but haven't been used yet. Assume newer
# values aren't wanted and check again later.
sleep(1)
continue
try:
max_reported = max(impala.find_impalad_mem_mb_reported_usage())
except Timeout:
LOG.debug("Timeout collecting reported mem usage")
max_reported = -1
try:
max_actual = max(impala.find_impalad_mem_mb_actual_usage())
except Timeout:
LOG.debug("Timeout collecting reported actual usage")
max_actual = -1
self._set_mem_usage_values(max_reported, max_actual)
if query_submission_is_locked and not self._total_num_queries_active():
if ready_to_unlock is None:
ready_to_unlock = False
else:
ready_to_unlock = True
except Exception:
LOG.debug("Error collecting impalad mem usage", exc_info=True)
if query_submission_is_locked:
LOG.debug("Resuming query submission")
self._submit_query_lock.release()
self._mem_polling_thread = create_and_start_daemon_thread(
poll_mem_usage, "Mem Usage Poller")
def _get_mem_usage_values(self, reset=False):
reported = None
actual = None
with self._max_mem_mb_reported_usage.get_lock():
with self._max_mem_mb_usage.get_lock():
reported = self._max_mem_mb_reported_usage.value
actual = self._max_mem_mb_usage.value
if reset:
self._max_mem_mb_reported_usage.value = -1
self._max_mem_mb_usage.value = -1
return reported, actual
def _set_mem_usage_values(self, reported, actual):
with self._max_mem_mb_reported_usage.get_lock():
with self._max_mem_mb_usage.get_lock():
self._max_mem_mb_reported_usage.value = reported
self._max_mem_mb_usage.value = actual
def _start_single_runner(self, query_runner):
"""Consumer function to take a query of the queue and run it. This is intended to
run in a separate process so validating the result set can use a full CPU.
"""
LOG.debug("New query runner started")
# The query runner should already be set up. We just need to connect() before using
# the runner.
query_runner.connect()
while not self._query_queue.empty():
try:
query = self._query_queue.get(True, 1)
except Empty:
continue
except EOFError:
LOG.debug("Query running aborting due to closed query queue")
break
LOG.debug("Getting query_idx")
with self._query_runners_lock:
query_idx = self._calc_total_runner_metric_no_lock(NUM_QUERIES_DEQUEUED)
increment(query_runner._metrics[NUM_QUERIES_DEQUEUED])
LOG.debug("Query_idx: {0} | PID: {1}".format(query_idx, query_runner.proc.pid))
if not query.required_mem_mb_without_spilling:
mem_limit = query.required_mem_mb_with_spilling
solo_runtime = query.solo_runtime_secs_with_spilling
elif self.spill_probability < random():
mem_limit = query.required_mem_mb_without_spilling
solo_runtime = query.solo_runtime_secs_without_spilling
else:
mem_limit = randrange(
query.required_mem_mb_with_spilling,
query.required_mem_mb_without_spilling + 1)
solo_runtime = query.solo_runtime_secs_with_spilling
LOG.debug("Waiting for other query runners to start their queries")
while query_idx > self._total_num_queries_submitted():
sleep(0.1)
self._mem_mb_needed_for_next_query.value = mem_limit
LOG.debug("Requesting memory reservation")
with self._mem_broker.reserve_mem_mb(mem_limit) as reservation_id:
LOG.debug("Received memory reservation")
with self._submit_query_lock:
increment(query_runner._metrics[NUM_QUERIES_SUBMITTED])
should_cancel = self.cancel_probability > random()
if should_cancel:
timeout = randrange(1, max(int(solo_runtime), 2))
else:
# Let the query run as long as necessary - it is nearly impossible to pick a
# good value that won't have false positives under load - see IMPALA-8222.
timeout = maxint
report = query_runner.run_query(query, mem_limit, timeout_secs=timeout,
should_cancel=should_cancel)
LOG.debug("Got execution report for query")
if report.timed_out and should_cancel:
report.was_cancelled = True
query_runner.update_from_query_report(report)
if report.other_error:
error_msg = str(report.other_error)
# There is a possible race during cancellation. If a fetch request fails (for
# example due to hitting a mem limit), just before the cancellation request, the
# server may have already unregistered the query as part of the fetch failure.
# In that case the server gives an error response saying the handle is invalid.
if "Invalid query handle" in error_msg and report.timed_out:
self._num_successive_errors.value = 0
continue
# Occasionally the network connection will fail, and depending on when the
# failure occurred during run_query(), an attempt to get the profile may be
# made which results in "Invalid session id" since the server destroyed the
# session upon disconnect.
if "Invalid session id" in error_msg:
self._num_successive_errors.value = 0
continue
# The server may fail to respond to clients if the load is high. An error
# message with "connect()...Connection timed out" comes from the impalad so
# that will not be ignored.
if (
("Connection timed out" in error_msg and "connect()" not in error_msg) or
"ECONNRESET" in error_msg or
"couldn't get a client" in error_msg or
"timeout: timed out" in error_msg
):
self._num_successive_errors.value = 0
continue
increment(self._num_successive_errors)
increment(query_runner._metrics[NUM_OTHER_ERRORS])
self._write_query_profile(report, PROFILES_DIR, prefix='error')
raise Exception("Query {query} ID {id} failed: {mesg}".format(
query=query.logical_query_id,
id=report.query_id,
mesg=error_msg))
if (
report.not_enough_memory and (self.test_admission_control or
not self._mem_broker.was_overcommitted(reservation_id))
):
increment(self._num_successive_errors)
self._write_query_profile(
report, PROFILES_DIR, prefix='unexpected_mem_exceeded')
raise Exception("Unexpected mem limit exceeded; mem was not overcommitted. "
"Query ID: {0}".format(report.query_id))
if (
not report.timed_out and not report.has_query_error() and
(self._verify_results and report.result_hash != query.result_hash)
):
increment(self._num_successive_errors)
increment(query_runner._metrics[NUM_RESULT_MISMATCHES])
self._write_query_profile(report, PROFILES_DIR, prefix='incorrect_results')
raise Exception(dedent("""\
Result hash mismatch; expected {expected}, got {actual}
Query ID: {id}
Query: {query}""".format(expected=query.result_hash,
actual=report.result_hash,
id=report.query_id,
query=query.logical_query_id)))
if report.timed_out and not should_cancel:
self._write_query_profile(report, PROFILES_DIR, prefix='timed_out')
raise Exception(
"Query {query} unexpectedly timed out. Query ID: {id}".format(
query=query.logical_query_id, id=report.query_id))
self._num_successive_errors.value = 0
LOG.debug("Query runner completed...")
def _print_status_header(self):
print(" | ".join(self._status_headers))
def _print_status(self, print_header=False):
if print_header:
self._print_status_header()
metrics = self._calc_total_runner_metrics()
reported_mem, actual_mem = self._get_mem_usage_values(reset=True)
status_format = " | ".join(["%%%ss" % len(header) for header in self._status_headers])
print(status_format % (
# Done
metrics[NUM_QUERIES_FINISHED],
# Active
metrics[NUM_QUERIES_SUBMITTED] - metrics[NUM_QUERIES_FINISHED],
# Executing
metrics[NUM_QUERIES_STARTED_RUNNING_OR_CANCELLED] -
metrics[NUM_QUERIES_FINISHED],
# Mem Lmt Ex
metrics[NUM_QUERIES_EXCEEDED_MEM_LIMIT],
# AC Rejected
metrics[NUM_QUERIES_AC_REJECTED],
# AC Timed Out
metrics[NUM_QUERIES_AC_TIMEDOUT],
# Cancel
metrics[NUM_QUERIES_CANCELLED],
# Err
metrics[NUM_OTHER_ERRORS],
# Incorrect
metrics[NUM_RESULT_MISMATCHES],
# Next Qry Mem Lmt
self._mem_mb_needed_for_next_query.value,
# Total Qry Mem Lmt
self._mem_broker.total_mem_mb - self._mem_broker.available_mem_mb,
# Tracked Mem
"" if reported_mem == -1 else reported_mem,
# RSS Mem
"" if actual_mem == -1 else actual_mem))
def _write_query_profile(self, report, subdir, prefix=None):
report.write_query_profile(
os.path.join(self.results_dir, subdir),
prefix)
def _check_successive_errors(self):
if (self._num_successive_errors.value >= self.num_successive_errors_needed_to_abort):
print(
"Aborting due to %s successive errors encountered"
% self._num_successive_errors.value, file=sys.stderr)
self.print_duration()
sys.exit(1)
def _check_for_test_failure(self):
metrics = self._calc_total_runner_metrics()
if metrics[NUM_OTHER_ERRORS] > 0 or metrics[NUM_RESULT_MISMATCHES] > 0:
LOG.error("Failing the stress test due to unexpected errors, incorrect results, or "
"timed out queries. See the report line above for details.")
self.print_duration()
sys.exit(1)
def _wait_for_test_to_finish(self, impala, should_print_status):
last_report_secs = 0
lines_printed = 1
sleep_secs = 0.1
num_runners_remaining = self._num_runners_remaining()
while (
self._query_producer_thread.is_alive() or
self._query_consumer_thread.is_alive() or
num_runners_remaining
):
if self._query_producer_thread.error or self._query_consumer_thread.error:
# This is bad enough to abort early. A failure here probably means there's a
# bug in this script. The mem poller could be checked for an error too. It is
# not critical so is ignored.
LOG.error("Aborting due to error in producer/consumer")
sys.exit(1)
do_check_for_impala_crashes = False
with self._query_runners_lock:
for idx, runner in enumerate(self._query_runners):
if runner.proc.exitcode is not None:
if runner.proc.exitcode != 0:
# Since at least one query runner process failed, make sure to check for
# crashed impalads.
do_check_for_impala_crashes = True
# TODO: Handle case for num_queries_dequeued != num_queries_submitted
num_submitted = runner._metrics[NUM_QUERIES_SUBMITTED].value
num_started_or_cancelled = \
runner._metrics[NUM_QUERIES_STARTED_RUNNING_OR_CANCELLED]
num_finished = runner._metrics[NUM_QUERIES_FINISHED].value
if num_submitted != num_finished:
# The query runner process may have crashed before updating the number
# of finished queries but after it incremented the number of queries
# submitted.
assert num_submitted - num_finished == 1
increment(runner._metrics[NUM_QUERIES_FINISHED])
if num_submitted != num_started_or_cancelled:
assert num_submitted - num_started_or_cancelled == 1
increment(runner._metrics[NUM_QUERIES_STARTED_RUNNING_OR_CANCELLED])
# Since we know that the runner crashed while trying to run a query, we
# count it as an 'other error'
increment(runner._metrics[NUM_OTHER_ERRORS])
self._check_successive_errors()
assert runner._metrics[NUM_QUERIES_SUBMITTED].value == \
runner._metrics[NUM_QUERIES_FINISHED].value, \
str([(k, v.value) for k, v in runner._metrics.iteritems()])
# Make sure to record all the metrics before removing this runner from the
# list.
print("Query runner ({0}) exited with exit code {1}".format(
runner.proc.pid, runner.proc.exitcode))
self._record_runner_metrics_before_evict(self._query_runners[idx])
# Remove the query runner from the list.
del self._query_runners[idx]
if do_check_for_impala_crashes:
# Since we know that at least one query runner failed, check if any of the Impala
# daemons themselves crashed.
LOG.info("Checking for Impala crashes")
if print_crash_info_if_exists(impala, self.start_time):