tests/regression/tools/live/live_test.c

   1 /*
   2  * Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License as published by as
   6  * published by the Free Software Foundation; only version 2 of the License.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _GNU_SOURCE
  19 #include <assert.h>
  20 #include <errno.h>
  21 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <unistd.h>
  25 #include <time.h>
  26 #include <sys/types.h>
  27 #include <inttypes.h>
  28 #include <stdlib.h>
  29 #include <sys/socket.h>
  30 #include <netinet/in.h>
  31 #include <netdb.h>
  32 #include <fcntl.h>
  33 #include <sys/mman.h>
  34 #include <sys/stat.h>
  35
  36 #include <tap/tap.h>
  37 #include <lttng/lttng.h>
  38
  39 #include <urcu/list.h>
  40 #include <bin/lttng-sessiond/session.h>
  41 #include <common/common.h>
  42
  43 #include <bin/lttng-relayd/lttng-viewer-abi.h>
  44 #include <common/index/ctf-index.h>
  45
  46 #define SESSION1 "test1"
  47 #define RELAYD_URL "net://localhost"
  48 #define LIVE_TIMER 2000000
  49
  50 /* Number of TAP tests in this file */
  51 #define NUM_TESTS 8
  52 #define mmap_size 524288
  53
  54 int ust_consumerd32_fd;
  55 int ust_consumerd64_fd;
  56
  57 static int control_sock;
  58 struct live_session *session;
  59
  60 static int first_packet_offset;
  61 static int first_packet_len;
  62 static int first_packet_stream_id;
  63
  64 struct viewer_stream {
  65         uint64_t id;
  66         uint64_t ctf_trace_id;
  67         void *mmap_base;
  68         int fd;
  69         int metadata_flag;
  70         int first_read;
  71         char path[PATH_MAX];
  72 };
  73
  74 struct live_session {
  75         struct viewer_stream *streams;
  76         uint64_t live_timer_interval;
  77         uint64_t stream_count;
  78 };
  79
  80 static
  81 int connect_viewer(char *hostname)
  82 {
  83         struct hostent *host;
  84         struct sockaddr_in server_addr;
  85         int ret;
  86
  87         host = gethostbyname(hostname);
  88         if (!host) {
  89                 ret = -1;
  90                 goto end;
  91         }
  92
  93         if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  94                 perror("Socket");
  95                 ret = -1;
  96                 goto end;
  97         }
  98
  99         server_addr.sin_family = AF_INET;
 100         server_addr.sin_port = htons(5344);
 101         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 102         bzero(&(server_addr.sin_zero), 8);
 103
 104         if (connect(control_sock, (struct sockaddr *) &server_addr,
 105                                 sizeof(struct sockaddr)) == -1) {
 106                 perror("Connect");
 107                 ret = -1;
 108                 goto end;
 109         }
 110
 111         server_addr.sin_family = AF_INET;
 112         server_addr.sin_port = htons(5345);
 113         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 114         bzero(&(server_addr.sin_zero), 8);
 115
 116         ret = 0;
 117
 118 end:
 119         return ret;
 120 }
 121
 122 int establish_connection(void)
 123 {
 124         struct lttng_viewer_cmd cmd;
 125         struct lttng_viewer_connect connect;
 126         int ret;
 127
 128         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 129         cmd.data_size = sizeof(connect);
 130         cmd.cmd_version = 0;
 131
 132         memset(&connect, 0, sizeof(connect));
 133         connect.major = htobe32(VERSION_MAJOR);
 134         connect.minor = htobe32(VERSION_MINOR);
 135         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 136
 137         do {
 138                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 139         } while (ret < 0 && errno == EINTR);
 140         if (ret < 0) {
 141                 fprintf(stderr, "Error sending cmd\n");
 142                 goto error;
 143         }
 144         do {
 145                 ret = send(control_sock, &connect, sizeof(connect), 0);
 146         } while (ret < 0 && errno == EINTR);
 147         if (ret < 0) {
 148                 fprintf(stderr, "Error sending version\n");
 149                 goto error;
 150         }
 151
 152         do {
 153                 ret = recv(control_sock, &connect, sizeof(connect), 0);
 154         } while (ret < 0 && errno == EINTR);
 155         if (ret < 0) {
 156                 fprintf(stderr, "Error receiving version\n");
 157                 goto error;
 158         }
 159         ret = 0;
 160
 161 error:
 162         return ret;
 163 }
 164
 165 /*
 166  * Returns the number of sessions, should be 1 during the unit test.
 167  */
 168 int list_sessions(int *session_id)
 169 {
 170         struct lttng_viewer_cmd cmd;
 171         struct lttng_viewer_list_sessions list;
 172         struct lttng_viewer_session lsession;
 173         int i, ret;
 174         int first_session = 0;
 175
 176         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 177         cmd.data_size = 0;
 178         cmd.cmd_version = 0;
 179
 180         do {
 181                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 182         } while (ret < 0 && errno == EINTR);
 183         if (ret < 0) {
 184                 fprintf(stderr, "Error sending cmd\n");
 185                 goto error;
 186         }
 187
 188         do {
 189                 ret = recv(control_sock, &list, sizeof(list), 0);
 190         } while (ret < 0 && errno == EINTR);
 191         if (ret < 0) {
 192                 fprintf(stderr, "Error receiving session list\n");
 193                 goto error;
 194         }
 195
 196         for (i = 0; i < be32toh(list.sessions_count); i++) {
 197                 do {
 198                         ret = recv(control_sock, &lsession, sizeof(lsession), 0);
 199                 } while (ret < 0 && errno == EINTR);
 200                 if (ret < 0) {
 201                         fprintf(stderr, "Error receiving session\n");
 202                         goto error;
 203                 }
 204                 if (lsession.streams > 0 && first_session <= 0) {
 205                         first_session = be64toh(lsession.id);
 206                         *session_id = first_session;
 207                 }
 208         }
 209
 210         ret = be32toh(list.sessions_count);
 211
 212 error:
 213         return ret;
 214 }
 215
 216 int create_viewer_session()
 217 {
 218         struct lttng_viewer_cmd cmd;
 219         struct lttng_viewer_create_session_response resp;
 220         int ret;
 221         ssize_t ret_len;
 222
 223         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 224         cmd.data_size = 0;
 225         cmd.cmd_version = 0;
 226
 227         do {
 228                 ret_len = send(control_sock, &cmd, sizeof(cmd), 0);
 229         } while (ret_len < 0 && errno == EINTR);
 230         if (ret_len < 0) {
 231                 fprintf(stderr, "[error] Error sending cmd\n");
 232                 ret = ret_len;
 233                 goto error;
 234         }
 235         assert(ret_len == sizeof(cmd));
 236
 237         do {
 238                 ret_len = recv(control_sock, &resp, sizeof(resp), 0);
 239         } while (ret_len < 0 && errno == EINTR);
 240         if (ret_len < 0) {
 241                 fprintf(stderr, "[error] Error receiving create session reply\n");
 242                 ret = ret_len;
 243                 goto error;
 244         }
 245         assert(ret_len == sizeof(resp));
 246
 247         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
 248                 fprintf(stderr, "[error] Error creating viewer session\n");
 249                 ret = -1;
 250                 goto error;
 251         }
 252         ret = 0;
 253
 254 error:
 255         return ret;
 256 }
 257
 258 int attach_session(int id)
 259 {
 260         struct lttng_viewer_cmd cmd;
 261         struct lttng_viewer_attach_session_request rq;
 262         struct lttng_viewer_attach_session_response rp;
 263         struct lttng_viewer_stream stream;
 264         int ret, i;
 265
 266         session = zmalloc(sizeof(struct live_session));
 267         if (!session) {
 268                 ret = -1;
 269                 goto error;
 270         }
 271
 272         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 273         cmd.data_size = sizeof(rq);
 274         cmd.cmd_version = 0;
 275
 276         memset(&rq, 0, sizeof(rq));
 277         rq.session_id = htobe64(id);
 278         rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 279
 280         do {
 281                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 282         } while (ret < 0 && errno == EINTR);
 283         if (ret < 0) {
 284                 fprintf(stderr, "Error sending cmd\n");
 285                 goto error;
 286         }
 287         do {
 288                 ret = send(control_sock, &rq, sizeof(rq), 0);
 289         } while (ret < 0 && errno == EINTR);
 290         if (ret < 0) {
 291                 fprintf(stderr, "Error sending attach request\n");
 292                 goto error;
 293         }
 294
 295         do {
 296                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 297         } while (ret < 0 && errno == EINTR);
 298         if (ret < 0) {
 299                 fprintf(stderr, "Error receiving attach response\n");
 300                 goto error;
 301         }
 302         if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
 303                 ret = -1;
 304                 goto end;
 305         }
 306
 307         session->stream_count = be32toh(rp.streams_count);
 308         session->streams = zmalloc(session->stream_count *
 309                         sizeof(struct viewer_stream));
 310         if (!session->streams) {
 311                 ret = -1;
 312                 goto error;
 313         }
 314
 315         for (i = 0; i < be32toh(rp.streams_count); i++) {
 316                 do {
 317                         ret = recv(control_sock, &stream, sizeof(stream), 0);
 318                 } while (ret < 0 && errno == EINTR);
 319                 if (ret < 0) {
 320                         fprintf(stderr, "Error receiving stream\n");
 321                         goto error;
 322                 }
 323                 session->streams[i].id = be64toh(stream.id);
 324
 325                 session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
 326                 session->streams[i].first_read = 1;
 327                 session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ | PROT_WRITE,
 328                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 329                 if (session->streams[i].mmap_base == MAP_FAILED) {
 330                         fprintf(stderr, "mmap error\n");
 331                         ret = -1;
 332                         goto error;
 333                 }
 334
 335                 if (be32toh(stream.metadata_flag)) {
 336                         session->streams[i].metadata_flag = 1;
 337                 }
 338         }
 339         ret = session->stream_count;
 340
 341 end:
 342 error:
 343         return ret;
 344 }
 345
 346 int get_metadata(void)
 347 {
 348         struct lttng_viewer_cmd cmd;
 349         struct lttng_viewer_get_metadata rq;
 350         struct lttng_viewer_metadata_packet rp;
 351         int ret;
 352         uint64_t i;
 353         char *data = NULL;
 354         uint64_t len = 0;
 355         int metadata_stream_id = -1;
 356
 357         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 358         cmd.data_size = sizeof(rq);
 359         cmd.cmd_version = 0;
 360
 361         for (i = 0; i < session->stream_count; i++) {
 362                 if (session->streams[i].metadata_flag) {
 363                         metadata_stream_id = i;
 364                         break;
 365                 }
 366         }
 367
 368         if (metadata_stream_id < 0) {
 369                 fprintf(stderr, "No metadata stream found\n");
 370                 ret = -1;
 371                 goto error;
 372         }
 373
 374         rq.stream_id = htobe64(session->streams[metadata_stream_id].id);
 375
 376         do {
 377                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 378         } while (ret < 0 && errno == EINTR);
 379         if (ret < 0) {
 380                 fprintf(stderr, "Error sending cmd\n");
 381                 goto error;
 382         }
 383         do {
 384                 ret = send(control_sock, &rq, sizeof(rq), 0);
 385         } while (ret < 0 && errno == EINTR);
 386         if (ret < 0) {
 387                 fprintf(stderr, "Error sending get_metadata request\n");
 388                 goto error;
 389         }
 390         do {
 391                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 392         } while (ret < 0 && errno == EINTR);
 393         if (ret < 0) {
 394                 fprintf(stderr, "Error receiving metadata response\n");
 395                 goto error;
 396         }
 397         switch (be32toh(rp.status)) {
 398                 case LTTNG_VIEWER_METADATA_OK:
 399                         break;
 400                 case LTTNG_VIEWER_NO_NEW_METADATA:
 401                         fprintf(stderr, "NO NEW\n");
 402                         ret = -1;
 403                         goto end;
 404                 case LTTNG_VIEWER_METADATA_ERR:
 405                         fprintf(stderr, "ERR\n");
 406                         ret = -1;
 407                         goto end;
 408                 default:
 409                         fprintf(stderr, "UNKNOWN\n");
 410                         ret = -1;
 411                         goto end;
 412         }
 413
 414         len = be64toh(rp.len);
 415         if (len <= 0) {
 416                 goto end;
 417         }
 418
 419         data = zmalloc(len);
 420         if (!data) {
 421                 perror("relay data zmalloc");
 422                 goto error;
 423         }
 424         do {
 425                 ret = recv(control_sock, data, len, MSG_WAITALL);
 426         } while (ret < 0 && errno == EINTR);
 427         if (ret < 0) {
 428                 fprintf(stderr, "Error receiving trace packet\n");
 429                 free(data);
 430                 goto error;
 431         }
 432         free(data);
 433
 434         ret = (int) len;
 435 end:
 436 error:
 437         return ret;
 438 }
 439
 440 int get_next_index(void)
 441 {
 442         struct lttng_viewer_cmd cmd;
 443         struct lttng_viewer_get_next_index rq;
 444         struct lttng_viewer_index rp;
 445         int ret;
 446         int id;
 447
 448         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
 449         cmd.data_size = sizeof(rq);
 450         cmd.cmd_version = 0;
 451
 452         for (id = 0; id < session->stream_count; id++) {
 453                 if (session->streams[id].metadata_flag) {
 454                         continue;
 455                 }
 456                 rq.stream_id = htobe64(session->streams[id].id);
 457
 458 retry:
 459                 do {
 460                         ret = send(control_sock, &cmd, sizeof(cmd), 0);
 461                 } while (ret < 0 && errno == EINTR);
 462                 if (ret < 0) {
 463                         fprintf(stderr, "Error sending cmd\n");
 464                         goto error;
 465                 }
 466                 do {
 467                         ret = send(control_sock, &rq, sizeof(rq), 0);
 468                 } while (ret < 0 && errno == EINTR);
 469                 if (ret < 0) {
 470                         fprintf(stderr, "Error sending get_next_index request\n");
 471                         goto error;
 472                 }
 473                 do {
 474                         ret = recv(control_sock, &rp, sizeof(rp), 0);
 475                 } while (ret < 0 && errno == EINTR);
 476                 if (ret < 0) {
 477                         fprintf(stderr, "Error receiving index response\n");
 478                         goto error;
 479                 }
 480
 481                 rp.flags = be32toh(rp.flags);
 482
 483                 switch (be32toh(rp.status)) {
 484                         case LTTNG_VIEWER_INDEX_INACTIVE:
 485                                 fprintf(stderr, "(INACTIVE)\n");
 486                                 break;
 487                         case LTTNG_VIEWER_INDEX_OK:
 488                                 break;
 489                         case LTTNG_VIEWER_INDEX_RETRY:
 490                                 sleep(1);
 491                                 goto retry;
 492                         case LTTNG_VIEWER_INDEX_HUP:
 493                                 fprintf(stderr, "(HUP)\n");
 494                                 session->streams[id].id = -1ULL;
 495                                 session->streams[id].fd = -1;
 496                                 break;
 497                         case LTTNG_VIEWER_INDEX_ERR:
 498                                 fprintf(stderr, "(ERR)\n");
 499                                 ret = -1;
 500                                 goto error;
 501                         default:
 502                                 fprintf(stderr, "SHOULD NOT HAPPEN\n");
 503                                 ret = -1;
 504                                 goto error;
 505                 }
 506                 if (!first_packet_stream_id) {
 507                         first_packet_offset = be64toh(rp.offset);
 508                         first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
 509                         first_packet_stream_id = id;
 510                 }
 511         }
 512         ret = 0;
 513
 514 error:
 515         return ret;
 516 }
 517
 518 static
 519 int get_data_packet(int id, uint64_t offset,
 520                 uint64_t len)
 521 {
 522         struct lttng_viewer_cmd cmd;
 523         struct lttng_viewer_get_packet rq;
 524         struct lttng_viewer_trace_packet rp;
 525         int ret;
 526
 527         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 528         cmd.data_size = sizeof(rq);
 529         cmd.cmd_version = 0;
 530
 531         rq.stream_id = htobe64(session->streams[id].id);
 532         /* Already in big endian. */
 533         rq.offset = offset;
 534         rq.len = htobe32(len);
 535
 536         do {
 537                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 538         } while (ret < 0 && errno == EINTR);
 539         if (ret < 0) {
 540                 fprintf(stderr, "Error sending cmd\n");
 541                 goto error;
 542         }
 543         do {
 544                 ret = send(control_sock, &rq, sizeof(rq), 0);
 545         } while (ret < 0 && errno == EINTR);
 546         if (ret < 0) {
 547                 fprintf(stderr, "Error sending get_data_packet request\n");
 548                 goto error;
 549         }
 550         do {
 551                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 552         } while (ret < 0 && errno == EINTR);
 553         if (ret < 0) {
 554                 fprintf(stderr, "Error receiving data response\n");
 555                 goto error;
 556         }
 557         rp.flags = be32toh(rp.flags);
 558
 559         switch (be32toh(rp.status)) {
 560         case LTTNG_VIEWER_GET_PACKET_OK:
 561                 break;
 562         case LTTNG_VIEWER_GET_PACKET_RETRY:
 563                 fprintf(stderr, "RETRY\n");
 564                 ret = -1;
 565                 goto end;
 566         case LTTNG_VIEWER_GET_PACKET_ERR:
 567                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 568                         fprintf(stderr, "NEW_METADATA\n");
 569                         ret = 0;
 570                         goto end;
 571                 }
 572                 fprintf(stderr, "ERR\n");
 573                 ret = -1;
 574                 goto end;
 575         default:
 576                 fprintf(stderr, "UNKNOWN\n");
 577                 ret = -1;
 578                 goto end;
 579         }
 580
 581         len = be32toh(rp.len);
 582         if (len <= 0) {
 583                 goto end;
 584         }
 585
 586         if (len > mmap_size) {
 587                 fprintf(stderr, "mmap_size not big enough\n");
 588                 ret = -1;
 589                 goto error;
 590         }
 591
 592         do {
 593                 ret = recv(control_sock, session->streams[id].mmap_base, len, MSG_WAITALL);
 594         } while (ret < 0 && errno == EINTR);
 595         if (ret < 0) {
 596                 fprintf(stderr, "Error receiving trace packet\n");
 597                 goto error;
 598         }
 599         ret = len;
 600
 601 end:
 602 error:
 603         return ret;
 604 }
 605
 606 int main(int argc, char **argv)
 607 {
 608         int ret;
 609         int session_id;
 610
 611         plan_tests(NUM_TESTS);
 612
 613         diag("Live unit tests");
 614
 615         ret = connect_viewer("localhost");
 616         ok(ret == 0, "Connect viewer to relayd");
 617
 618         ret = establish_connection();
 619         ok(ret == 0, "Established connection and version check with %d.%d",
 620                         VERSION_MAJOR, VERSION_MINOR);
 621
 622         ret = list_sessions(&session_id);
 623         ok(ret > 0, "List sessions : %d session(s)", ret);
 624
 625         ret = create_viewer_session();
 626         ok(ret == 0, "Create viewer session");
 627
 628         ret = attach_session(session_id);
 629         ok(ret > 0, "Attach to session, %d streams received", ret);
 630
 631         ret = get_metadata();
 632         ok(ret > 0, "Get metadata, received %d bytes", ret);
 633
 634         ret = get_next_index();
 635         ok(ret == 0, "Get one index per stream");
 636
 637         ret = get_data_packet(first_packet_stream_id, first_packet_offset,
 638                         first_packet_len);
 639         ok(ret == first_packet_len,
 640                         "Get one data packet for stream %d, offset %d, len %d",
 641                         first_packet_stream_id, first_packet_offset,
 642                         first_packet_len);
 643
 644         return exit_status();
 645 }