glusterfs

1
#!/usr/bin/python3
2

3
from __future__ import absolute_import
4
from __future__ import division
5
from __future__ import unicode_literals
6
from __future__ import print_function
7
import re
8
import sys
9
import fcntl
10
import base64
11
import threading
12
import socket
13
import os
14
import shlex
15
import argparse
16
import subprocess
17
import time
18
import SimpleXMLRPCServer
19
import xmlrpclib
20
import md5
21
import httplib
22
import uuid
23

24
DEFAULT_PORT = 9999
25
TEST_TIMEOUT_S = 15 * 60
26
CLIENT_CONNECT_TIMEOUT_S = 10
27
CLIENT_TIMEOUT_S = 60
28
PATCH_FILE_UID = str(uuid.uuid4())
29
SSH_TIMEOUT_S = 10
30
MAX_ATTEMPTS = 3
31
ADDRESS_FAMILY = 'IPv4'
32

33

34
def socket_instance(address_family):
35
    if address_family.upper() == 'ipv4'.upper():
36
        return socket.socket(socket.AF_INET, socket.SOCK_STREAM)
37
    elif address_family.upper() == 'ipv6'.upper():
38
        return socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
39
    else:
40
        Log.error("Invalid IP address family")
41
        sys.exit(1)
42

43

44
def patch_file():
45
    return "/tmp/%s-patch.tar.gz" % PATCH_FILE_UID
46

47
# ..............................................................................
48
# SimpleXMLRPCServer IPvX Wrapper
49
# ..............................................................................
50

51

52
class GeneralXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer):
53
    def __init__(self, addr):
54
        SimpleXMLRPCServer.SimpleXMLRPCServer.__init__(self, addr)
55

56
    def server_bind(self):
57
        if self.socket:
58
            self.socket.close()
59
        self.socket = socket_instance(args.address_family)
60
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
61
        SimpleXMLRPCServer.SimpleXMLRPCServer.server_bind(self)
62

63

64
class HTTPConnection(httplib.HTTPConnection):
65
    def __init__(self, host):
66
        self.host = host
67
        httplib.HTTPConnection.__init__(self, host)
68

69
    def connect(self):
70
        old_timeout = socket.getdefaulttimeout()
71
        self.sock = socket.create_connection((self.host, self.port),
72
                                             timeout=CLIENT_CONNECT_TIMEOUT_S)
73
        self.sock.settimeout(old_timeout)
74

75

76
class IPTransport(xmlrpclib.Transport):
77
    def __init__(self, *args, **kwargs):
78
        xmlrpclib.Transport.__init__(self, *args, **kwargs)
79

80
    def make_connection(self, host):
81
        return HTTPConnection(host)
82

83

84
# ..............................................................................
85
# Common
86
# ..............................................................................
87

88

89
class Timer:
90
    def __init__(self):
91
        self.start = time.time()
92

93
    def elapsed_s(self):
94
        return int(time.time() - self.start)
95

96
    def reset(self):
97
        ret = self.elapsed_s()
98
        self.start = time.time()
99
        return ret
100

101

102
def encode(buf):
103
    return base64.b16encode(buf)
104

105

106
def decode(buf):
107
    return base64.b16decode(buf)
108

109

110
def get_file_content(path):
111
    with open(path, "r") as f:
112
        return f.read()
113

114

115
def write_to_file(path, data):
116
    with open(path, "w") as f:
117
        f.write(data)
118

119

120
def failsafe(fn, args=()):
121
    try:
122
        return (True, fn(*args))
123
    except (xmlrpclib.Fault, xmlrpclib.ProtocolError, xmlrpclib.ResponseError,
124
            Exception) as err:
125
        Log.debug(str(err))
126
    return (False, None)
127

128

129
class LogLevel:
130
    DEBUG = 2
131
    ERROR = 1
132
    CLI = 0
133

134

135
class Log:
136
    LOGLEVEL = LogLevel.ERROR
137

138
    @staticmethod
139
    def _normalize(msg):
140
        return msg[:100]
141

142
    @staticmethod
143
    def debug(msg):
144
        if Log.LOGLEVEL >= LogLevel.DEBUG:
145
            sys.stdout.write("<debug> %s\n" % Log._normalize(msg))
146
            sys.stdout.flush()
147

148
    @staticmethod
149
    def error(msg):
150
        sys.stderr.write("<error> %s\n" % Log._normalize(msg))
151

152
    @staticmethod
153
    def header(msg):
154
        sys.stderr.write("* %s *\n" % Log._normalize(msg))
155

156
    @staticmethod
157
    def cli(msg):
158
        sys.stderr.write("%s\n" % msg)
159

160

161
class Shell:
162
    def __init__(self, cwd=None, logpath=None):
163
        self.cwd = cwd
164
        self.shell = True
165
        self.redirect = open(os.devnull if not logpath else logpath, "wr+")
166

167
    def __del__(self):
168
        self.redirect.close()
169

170
    def cd(self, cwd):
171
        Log.debug("cd %s" % cwd)
172
        self.cwd = cwd
173

174
    def truncate(self):
175
        self.redirect.truncate(0)
176

177
    def read_logs(self):
178
        self.redirect.seek(0)
179
        return self.redirect.read()
180

181
    def check_call(self, cmd):
182
        status = self.call(cmd)
183
        if status:
184
            raise Exception("Error running command %s. status=%s"
185
                            % (cmd, status))
186

187
    def call(self, cmd):
188
        if isinstance(cmd, list):
189
            return self._calls(cmd)
190

191
        return self._call(cmd)
192

193
    def ssh(self, hostname, cmd, id_rsa=None):
194
        flags = "" if not id_rsa else "-i " + id_rsa
195
        return self.call("timeout %s ssh %s root@%s \"%s\"" %
196
                            (SSH_TIMEOUT_S, flags, hostname, cmd))
197

198
    def scp(self, hostname, src, dest, id_rsa=None):
199
        flags = "" if not id_rsa else "-i " + id_rsa
200
        return self.call("timeout %s scp %s %s root@%s:%s" %
201
                            (SSH_TIMEOUT_S, flags, src, hostname, dest))
202

203
    def output(self, cmd, cwd=None):
204
        Log.debug("%s> %s" % (cwd, cmd))
205
        return subprocess.check_output(shlex.split(cmd), cwd=self.cwd)
206

207
    def _calls(self, cmds):
208
        Log.debug("Running commands. %s" % cmds)
209
        for c in cmds:
210
            status = self.call(c)
211
            if status:
212
                Log.error("Commands failed with %s" % status)
213
                return status
214
        return 0
215

216
    def _call(self, cmd):
217
        if not self.shell:
218
            cmd = shlex.split(cmd)
219

220
        Log.debug("%s> %s" % (self.cwd, cmd))
221

222
        status = subprocess.call(cmd, cwd=self.cwd, shell=self.shell,
223
                                 stdout=self.redirect, stderr=self.redirect)
224

225
        Log.debug("return %s" % status)
226
        return status
227

228

229
# ..............................................................................
230
# Server role
231
# ..............................................................................
232

233
class TestServer:
234
    def __init__(self, port, scratchdir):
235
        self.port = port
236
        self.scratchdir = scratchdir
237
        self.shell = Shell()
238
        self.rpc = None
239
        self.pidf = None
240

241
        self.shell.check_call("mkdir -p %s" % self.scratchdir)
242
        self._process_lock()
243

244
    def __del__(self):
245
        if self.pidf:
246
            self.pidf.close()
247

248
    def init(self):
249
        Log.debug("Starting xmlrpc server on port %s" % self.port)
250
        self.rpc = GeneralXMLRPCServer(("", self.port))
251
        self.rpc.register_instance(Handlers(self.scratchdir))
252

253
    def serve(self):
254
        (status, _) = failsafe(self.rpc.serve_forever)
255
        Log.cli("== End ==")
256

257
    def _process_lock(self):
258
        pid_filename = os.path.basename(__file__).replace("/", "-")
259
        pid_filepath = "%s/%s.pid" % (self.scratchdir, pid_filename)
260
        self.pidf = open(pid_filepath, "w")
261
        try:
262
            fcntl.lockf(self.pidf, fcntl.LOCK_EX | fcntl.LOCK_NB)
263
            # We have the lock, kick anybody listening on this port
264
            self.shell.call("kill $(lsof -t -i:%s)" % self.port)
265
        except IOError:
266
            Log.error("Another process instance is running")
267
            sys.exit(0)
268

269
#
270
# Server Handler
271
#
272

273

274
handler_lock = threading.Lock()
275
handler_serving_since = Timer()
276

277

278
def synchronized(func):
279
    def decorator(*args, **kws):
280
        handler_lock.acquire()
281
        h = args[0]
282
        try:
283
            h.shell.truncate()
284
            ret = func(*args, **kws)
285
            return ret
286
        except Exception() as err:
287
            Log.error(str(err))
288
            Log.error(decode(h._log_content()))
289
            raise
290
        finally:
291
            handler_lock.release()
292
            handler_serving_since.reset()
293

294
    return decorator
295

296

297
class Handlers:
298
    def __init__(self, scratchdir):
299
        self.client_id = None
300
        self.scratchdir = scratchdir
301
        self.gluster_root = "%s/glusterfs" % self.scratchdir
302
        self.shell = Shell(logpath="%s/test-handlers.log" % self.scratchdir)
303

304
    def hello(self, id):
305
        if not handler_lock.acquire(False):
306
            return False
307
        try:
308
            return self._hello_locked(id)
309
        finally:
310
            handler_lock.release()
311

312
    def _hello_locked(self, id):
313
        if handler_serving_since.elapsed_s() > CLIENT_TIMEOUT_S:
314
            Log.debug("Disconnected client %s" % self.client_id)
315
            self.client_id = None
316

317
        if not self.client_id:
318
            self.client_id = id
319
            handler_serving_since.reset()
320
            return True
321

322
        return (id == self.client_id)
323

324
    @synchronized
325
    def ping(self, id=None):
326
        if id:
327
            return id == self.client_id
328
        return True
329

330
    @synchronized
331
    def bye(self, id):
332
        assert id == self.client_id
333
        self.client_id = None
334
        handler_serving_since.reset()
335
        return True
336

337
    @synchronized
338
    def cleanup(self, id):
339
        assert id == self.client_id
340
        self.shell.cd(self.gluster_root)
341
        self.shell.check_call("PATH=.:$PATH; sudo ./clean_gfs_devserver.sh")
342
        return True
343

344
    @synchronized
345
    def copy(self, id, name, content):
346
        with open("%s/%s" % (self.scratchdir, name), "w+") as f:
347
            f.write(decode(content))
348
        return True
349

350
    @synchronized
351
    def copygzip(self, id, content):
352
        assert id == self.client_id
353
        gzipfile = "%s/tmp.tar.gz" % self.scratchdir
354
        tarfile = "%s/tmp.tar" % self.scratchdir
355
        self.shell.check_call("rm -f %s" % gzipfile)
356
        self.shell.check_call("rm -f %s" % tarfile)
357
        write_to_file(gzipfile, decode(content))
358

359
        self.shell.cd(self.scratchdir)
360
        self.shell.check_call("rm -r -f %s" % self.gluster_root)
361
        self.shell.check_call("mkdir -p %s" % self.gluster_root)
362

363
        self.shell.cd(self.gluster_root)
364
        cmds = [
365
            "gunzip -f -q %s" % gzipfile,
366
            "tar -xvf %s" % tarfile
367
        ]
368
        return self.shell.call(cmds) == 0
369

370
    @synchronized
371
    def build(self, id, asan=False):
372
        assert id == self.client_id
373
        self.shell.cd(self.gluster_root)
374
        self.shell.call("make clean")
375
        env = "ASAN_ENABLED=1" if asan else ""
376
        return self.shell.call(
377
		"%s ./extras/distributed-testing/distributed-test-build.sh" % env) == 0
378

379
    @synchronized
380
    def install(self, id):
381
        assert id == self.client_id
382
        self.shell.cd(self.gluster_root)
383
        return self.shell.call("make install") == 0
384

385
    @synchronized
386
    def prove(self, id, test, timeout, valgrind="no", asan_noleaks=True):
387
        assert id == self.client_id
388
        self.shell.cd(self.gluster_root)
389
        env = "DEBUG=1 "
390
        if valgrind == "memcheck" or valgrind == "yes":
391
            cmd = "valgrind"
392
            cmd += " --tool=memcheck --leak-check=full --track-origins=yes"
393
            cmd += " --show-leak-kinds=all -v prove -v"
394
        elif valgrind == "drd":
395
            cmd = "valgrind"
396
            cmd += " --tool=drd -v prove -v"
397
        elif asan_noleaks:
398
            cmd = "prove -v"
399
            env += "ASAN_OPTIONS=detect_leaks=0 "
400
        else:
401
            cmd = "prove -v"
402

403
        status = self.shell.call(
404
		"%s timeout %s %s %s" % (env, timeout, cmd, test))
405

406
        if status != 0:
407
            return (False, self._log_content())
408
        return (True, "")
409

410
    def _log_content(self):
411
        return encode(self.shell.read_logs())
412

413
# ..............................................................................
414
# Cli role
415
# ..............................................................................
416

417

418
class RPCConnection((threading.Thread)):
419
    def __init__(self, host, port, path, cb):
420
        threading.Thread.__init__(self)
421
        self.host = host
422
        self.port = port
423
        self.path = path
424
        self.shell = Shell()
425
        self.cb = cb
426
        self.stop = False
427
        self.proxy = None
428
        self.logid = "%s:%s" % (self.host, self.port)
429

430
    def connect(self):
431
        (status, ret) = failsafe(self._connect)
432
        return (status and ret)
433

434
    def _connect(self):
435
        url = "http://%s:%s" % (self.host, self.port)
436
        self.proxy = xmlrpclib.ServerProxy(url, transport=IPTransport())
437
        return self.proxy.hello(self.cb.id)
438

439
    def disconnect(self):
440
        self.stop = True
441

442
    def ping(self):
443
        return self.proxy.ping()
444

445
    def init(self):
446
        return self._copy() and self._compile_and_install()
447

448
    def run(self):
449
        (status, ret) = failsafe(self.init)
450
        if not status:
451
            self.cb.note_lost_connection(self)
452
            return
453
        elif not ret:
454
            self.cb.note_setup_failed(self)
455
            return
456

457
        while not self.stop:
458
            (status, ret) = failsafe(self._run)
459
            if not status or not ret:
460
                self.cb.note_lost_connection(self)
461
                break
462
            time.sleep(0)
463

464
        failsafe(self.proxy.bye, (self.cb.id,))
465
        Log.debug("%s connection thread stopped" % self.host)
466

467
    def _run(self):
468
        test = self.cb.next_test()
469
        (status, _) = failsafe(self._execute_next, (test,))
470
        if not status:
471
            self.cb.note_retry(test)
472
            return False
473
        return True
474

475
    def _execute_next(self, test):
476
        if not test:
477
            time.sleep(1)
478
            return
479

480
        (status, error) = self.proxy.prove(self.cb.id, test,
481
                                           self.cb.test_timeout,
482
                                           self.cb.valgrind,
483
                                           self.cb.asan_noleaks)
484
        if status:
485
            self.cb.note_done(test)
486
        else:
487
            self.cb.note_error(test, error)
488

489
    def _compile_and_install(self):
490
        Log.debug("<%s> Build " % self.logid)
491
        asan = self.cb.asan or self.cb.asan_noleaks
492
        return (self.proxy.build(self.cb.id, asan) and
493
                self.proxy.install(self.cb.id))
494

495
    def _copy(self):
496
        return self._copy_gzip()
497

498
    def _copy_gzip(self):
499
        Log.cli("<%s> copying and compiling %s to remote" %
500
                 (self.logid, self.path))
501
        data = encode(get_file_content(patch_file()))
502
        Log.debug("GZIP size = %s B" % len(data))
503
        return self.proxy.copygzip(self.cb.id, data)
504

505

506
class RPCConnectionPool:
507
    def __init__(self, gluster_path, hosts, n, id_rsa):
508
        self.gluster_path = gluster_path
509
        self.hosts = hosts
510
        self.conns = []
511
        self.faulty = []
512
        self.n = int(len(hosts) / 2) + 1 if not n else n
513
        self.id_rsa = id_rsa
514
        self.stop = False
515
        self.scanner = threading.Thread(target=self._scan_hosts_loop)
516
        self.kicker = threading.Thread(target=self._kick_hosts_loop)
517
        self.shell = Shell()
518
        self.since_start = Timer()
519

520
        self.shell.check_call("rm -f %s" % patch_file())
521
        self.shell.check_call("tar -zcvf %s ." % patch_file())
522
        self.id = md5.new(get_file_content(patch_file())).hexdigest()
523
        Log.cli("client UID %s" % self.id)
524
        Log.cli("patch UID %s" % PATCH_FILE_UID)
525

526
    def __del__(self):
527
        self.shell.check_call("rm -f %s" % patch_file())
528

529
    def pool_status(self):
530
        elapsed_m = int(self.since_start.elapsed_s() / 60)
531
        return "%s/%s connected, %smin elapsed" % (len(self.conns), self.n,
532
                                                   elapsed_m)
533

534
    def connect(self):
535
        Log.debug("Starting scanner")
536
        self.scanner.start()
537
        self.kicker.start()
538

539
    def disconnect(self):
540
        self.stop = True
541
        for conn in self.conns:
542
            conn.disconnect()
543

544
    def note_lost_connection(self, conn):
545
        Log.cli("lost connection to %s" % conn.host)
546
        self.conns.remove(conn)
547
        self.hosts.append((conn.host, conn.port))
548

549
    def note_setup_failed(self, conn):
550
        Log.error("Setup failed on %s:%s" % (conn.host, conn.port))
551
        self.conns.remove(conn)
552
        self.faulty.append((conn.host, conn.port))
553

554
    def _scan_hosts_loop(self):
555
        Log.debug("Scanner thread started")
556
        while not self.stop:
557
            failsafe(self._scan_hosts)
558
            time.sleep(5)
559

560
    def _scan_hosts(self):
561
        if len(self.hosts) == 0 and len(self.conns) == 0:
562
            Log.error("no more hosts available to loadbalance")
563
            sys.exit(1)
564

565
        for (host, port) in self.hosts:
566
            if (len(self.conns) >= self.n) or self.stop:
567
                break
568
            self._scan_host(host, port)
569

570
    def _scan_host(self, host, port):
571
        Log.debug("scanning %s:%s" % (host, port))
572
        c = RPCConnection(host, port, self.gluster_path, self)
573
        (status, result) = failsafe(c.connect)
574
        if status and result:
575
            self.hosts.remove((host, port))
576
            Log.debug("Connected to %s:%s" % (host, port))
577
            self.conns.append(c)
578
            c.start()
579
            Log.debug("%s / %s connected" % (len(self.conns), self.n))
580
        else:
581
            Log.debug("Failed to connect to %s:%s" % (host, port))
582

583
    def _kick_hosts_loop(self):
584
        Log.debug("Kick thread started")
585
        while not self.stop:
586
            time.sleep(10)
587
            failsafe(self._kick_hosts)
588

589
        Log.debug("Kick thread stopped")
590

591
    def _is_pingable(self, host, port):
592
        c = RPCConnection(host, port, self.gluster_path, self)
593
        failsafe(c.connect)
594
        (status, result) = failsafe(c.ping)
595
        return status and result
596

597
    def _kick_hosts(self):
598
        # Do not kick hosts if we have the optimal number of connections
599
        if (len(self.conns) >= self.n) or self.stop:
600
            Log.debug("Skip kicking hosts")
601
            return
602

603
        # Check and if dead kick all hosts
604
        for (host, port) in self.hosts:
605
            if self.stop:
606
                Log.debug("Break kicking hosts")
607
                break
608

609
            if self._is_pingable(host, port):
610
                Log.debug("Host=%s is alive. Won't kick" % host)
611
                continue
612

613
            Log.debug("Kicking %s" % host)
614
            mypath = sys.argv[0]
615
            myname = os.path.basename(mypath)
616
            destpath = "/tmp/%s" % myname
617
            sh = Shell()
618
            sh.scp(host, mypath, destpath, self.id_rsa)
619
            sh.ssh(host, "nohup %s --server &>> %s.log &" %
620
                         (destpath, destpath), self.id_rsa)
621

622
    def join(self):
623
        self.scanner.join()
624
        self.kicker.join()
625
        for c in self.conns:
626
            c.join()
627

628

629
# ..............................................................................
630
# test role
631
# ..............................................................................
632

633
class TestRunner(RPCConnectionPool):
634
    def __init__(self, gluster_path, hosts, n, tests, flaky_tests, valgrind,
635
                 asan, asan_noleaks, id_rsa, test_timeout):
636
        RPCConnectionPool.__init__(self, gluster_path, self._parse_hosts(hosts),
637
                                   n, id_rsa)
638
        self.flaky_tests = flaky_tests.split(" ")
639
        self.pending = []
640
        self.done = []
641
        self.error = []
642
        self.retry = {}
643
        self.error_logs = []
644
        self.stats_timer = Timer()
645
        self.valgrind = valgrind
646
        self.asan = asan
647
        self.asan_noleaks = asan_noleaks
648
        self.test_timeout = test_timeout
649

650
        self.tests = self._get_tests(tests)
651

652
        Log.debug("tests: %s" % self.tests)
653

654
    def _get_tests(self, tests):
655
        if not tests or tests == "all":
656
            return self._not_flaky(self._all())
657
        elif tests == "flaky":
658
            return self.flaky_tests
659
        else:
660
            return self._not_flaky(tests.strip().split(" "))
661

662
    def run(self):
663
        self.connect()
664
        self.join()
665
        return len(self.error)
666

667
    def _pretty_print(self, data):
668
        if isinstance(data, list):
669
            str = ""
670
            for i in data:
671
                str = "%s %s" % (str, i)
672
            return str
673
        return "%s" % data
674

675
    def print_result(self):
676
        Log.cli("== RESULTS ==")
677
        Log.cli("SUCCESS  : %s" % len(self.done))
678
        Log.cli("ERRORS   : %s" % len(self.error))
679
        Log.cli("== ERRORS ==")
680
        Log.cli(self._pretty_print(self.error))
681
        Log.cli("== LOGS ==")
682
        Log.cli(self._pretty_print(self.error_logs))
683
        Log.cli("== END ==")
684

685
    def next_test(self):
686
        if len(self.tests):
687
            test = self.tests.pop()
688
            self.pending.append(test)
689
            return test
690

691
        if not len(self.pending):
692
            self.disconnect()
693

694
        return None
695

696
    def _pct_completed(self):
697
        total = len(self.tests) + len(self.pending) + len(self.done)
698
        total += len(self.error)
699
        completed = len(self.done) + len(self.error)
700
        return 0 if not total else int(completed / total * 100)
701

702
    def note_done(self, test):
703
        Log.cli("%s PASS (%s%% done) (%s)" % (test, self._pct_completed(),
704
                                              self.pool_status()))
705
        self.pending.remove(test)
706
        self.done.append(test)
707
        if test in self.retry:
708
            del self.retry[test]
709

710
    def note_error(self, test, errstr):
711
        Log.cli("%s FAIL" % test)
712
        self.pending.remove(test)
713
        if test not in self.retry:
714
            self.retry[test] = 1
715

716
        if errstr:
717
            path = "%s/%s-%s.log" % ("/tmp", test.replace("/", "-"),
718
                                     self.retry[test])
719
            failsafe(write_to_file, (path, decode(errstr),))
720
            self.error_logs.append(path)
721

722
        if self.retry[test] < MAX_ATTEMPTS:
723
            self.retry[test] += 1
724
            Log.debug("retry test %s attempt %s" % (test, self.retry[test]))
725
            self.tests.append(test)
726
        else:
727
            Log.debug("giveup attempt test %s" % test)
728
            del self.retry[test]
729
            self.error.append(test)
730

731
    def note_retry(self, test):
732
        Log.cli("retry %s on another host" % test)
733
        self.pending.remove(test)
734
        self.tests.append(test)
735

736
    #
737
    # test classifications
738
    #
739
    def _all(self):
740
        return self._list_tests(["tests"], recursive=True)
741

742
    def _not_flaky(self, tests):
743
        for t in self.flaky_tests:
744
            if t in tests:
745
                tests.remove(t)
746
        return tests
747

748
    def _list_tests(self, prefixes, recursive=False, ignore_ifnotexist=False):
749
        tests = []
750
        for prefix in prefixes:
751
            real_path = "%s/%s" % (self.gluster_path, prefix)
752
            if not os.path.exists(real_path) and ignore_ifnotexist:
753
                continue
754
            for f in os.listdir(real_path):
755
                if os.path.isdir(real_path + "/" + f):
756
                    if recursive:
757
                        tests += self._list_tests([prefix + "/" + f], recursive)
758
                else:
759
                    if re.match(r".*\.t$", f):
760
                        tests += [prefix + "/" + f]
761
        return tests
762

763
    def _parse_hosts(self, hosts):
764
        ret = []
765
        for h in args.hosts.split(" "):
766
            ret.append((h, DEFAULT_PORT))
767
        Log.debug(ret)
768
        return ret
769

770
# ..............................................................................
771
# Roles entry point
772
# ..............................................................................
773

774

775
def run_as_server(args):
776
    if not args.server_path:
777
        Log.error("please provide server path")
778
        return 1
779

780
    server = TestServer(args.port, args.server_path)
781
    server.init()
782
    server.serve()
783
    return 0
784

785

786
def run_as_tester(args):
787
    Log.header("GLUSTER TEST CLI")
788

789
    Log.debug("args = %s" % args)
790

791
    tests = TestRunner(args.gluster_path, args.hosts, args.n, args.tests,
792
                       args.flaky_tests, valgrind=args.valgrind,
793
                       asan=args.asan, asan_noleaks=args.asan_noleaks,
794
                       id_rsa=args.id_rsa, test_timeout=args.test_timeout)
795
    result = tests.run()
796
    tests.print_result()
797
    return result
798

799
# ..............................................................................
800
# main
801
# ..............................................................................
802

803

804
def main(args):
805
    if args.v:
806
        Log.LOGLEVEL = LogLevel.DEBUG
807

808
    if args.server and args.tester:
809
        Log.error("Invalid arguments. More than one role specified")
810
        sys.exit(1)
811

812
    if args.server:
813
        sys.exit(run_as_server(args))
814
    elif args.tester:
815
        sys.exit(run_as_tester(args))
816
    else:
817
        Log.error("please specify a mode for CI")
818
        parser.print_help()
819
        sys.exit(1)
820

821

822
parser = argparse.ArgumentParser(description="Gluster CI")
823

824
# server role
825
parser.add_argument("--server", help="start server", action="store_true")
826
parser.add_argument("--server_path", help="server scratch space",
827
                    default="/tmp/gluster-test")
828
parser.add_argument("--host", help="server address to listen", default="")
829
parser.add_argument("--port", help="server port to listen",
830
                    type=int, default=DEFAULT_PORT)
831
# test role
832
parser.add_argument("--tester", help="start tester", action="store_true")
833
parser.add_argument("--valgrind[=memcheck,drd]",
834
                    help="run tests with valgrind tool 'memcheck' or 'drd'",
835
                    default="no")
836
parser.add_argument("--asan", help="test with asan enabled",
837
                    action="store_true")
838
parser.add_argument("--asan-noleaks", help="test with asan but no mem leaks",
839
                    action="store_true")
840
parser.add_argument("--tests", help="all/flaky/list of tests", default=None)
841
parser.add_argument("--flaky_tests", help="list of flaky tests", default=None)
842
parser.add_argument("--n", help="max number of machines to use", type=int,
843
                    default=0)
844
parser.add_argument("--hosts", help="list of worker machines")
845
parser.add_argument("--gluster_path", help="gluster path to test",
846
                    default=os.getcwd())
847
parser.add_argument("--id-rsa", help="private key to use for ssh",
848
                    default=None)
849
parser.add_argument("--test-timeout",
850
                    help="test timeout in sec (default 15min)",
851
                    default=TEST_TIMEOUT_S)
852
# general
853
parser.add_argument("-v", help="verbose", action="store_true")
854
parser.add_argument("--address_family", help="IPv6 or IPv4 to use",
855
                    default=ADDRESS_FAMILY)
856

857
args = parser.parse_args()
858

859
main(args)
860