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 7
  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(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(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(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 attach_session(int id)
 217 {
 218         struct lttng_viewer_cmd cmd;
 219         struct lttng_viewer_attach_session_request rq;
 220         struct lttng_viewer_attach_session_response rp;
 221         struct lttng_viewer_stream stream;
 222         int ret, i;
 223
 224         session = zmalloc(sizeof(struct live_session));
 225         if (!session) {
 226                 ret = -1;
 227                 goto error;
 228         }
 229
 230         cmd.cmd = htobe32(VIEWER_ATTACH_SESSION);
 231         cmd.data_size = sizeof(rq);
 232         cmd.cmd_version = 0;
 233
 234         memset(&rq, 0, sizeof(rq));
 235         rq.session_id = htobe64(id);
 236         rq.seek = htobe32(VIEWER_SEEK_BEGINNING);
 237
 238         do {
 239                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 240         } while (ret < 0 && errno == EINTR);
 241         if (ret < 0) {
 242                 fprintf(stderr, "Error sending cmd\n");
 243                 goto error;
 244         }
 245         do {
 246                 ret = send(control_sock, &rq, sizeof(rq), 0);
 247         } while (ret < 0 && errno == EINTR);
 248         if (ret < 0) {
 249                 fprintf(stderr, "Error sending attach request\n");
 250                 goto error;
 251         }
 252
 253         do {
 254                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 255         } while (ret < 0 && errno == EINTR);
 256         if (ret < 0) {
 257                 fprintf(stderr, "Error receiving attach response\n");
 258                 goto error;
 259         }
 260         if (be32toh(rp.status) != VIEWER_ATTACH_OK) {
 261                 ret = -1;
 262                 goto end;
 263         }
 264
 265         session->stream_count = be32toh(rp.streams_count);
 266         session->streams = zmalloc(session->stream_count *
 267                         sizeof(struct viewer_stream));
 268         if (!session->streams) {
 269                 ret = -1;
 270                 goto error;
 271         }
 272
 273         for (i = 0; i < be32toh(rp.streams_count); i++) {
 274                 do {
 275                         ret = recv(control_sock, &stream, sizeof(stream), 0);
 276                 } while (ret < 0 && errno == EINTR);
 277                 if (ret < 0) {
 278                         fprintf(stderr, "Error receiving stream\n");
 279                         goto error;
 280                 }
 281                 session->streams[i].id = be64toh(stream.id);
 282
 283                 session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
 284                 session->streams[i].first_read = 1;
 285                 session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ | PROT_WRITE,
 286                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 287                 if (session->streams[i].mmap_base == MAP_FAILED) {
 288                         fprintf(stderr, "mmap error\n");
 289                         ret = -1;
 290                         goto error;
 291                 }
 292
 293                 if (be32toh(stream.metadata_flag)) {
 294                         session->streams[i].metadata_flag = 1;
 295                 }
 296         }
 297         ret = session->stream_count;
 298
 299 end:
 300 error:
 301         return ret;
 302 }
 303
 304 int get_metadata(void)
 305 {
 306         struct lttng_viewer_cmd cmd;
 307         struct lttng_viewer_get_metadata rq;
 308         struct lttng_viewer_metadata_packet rp;
 309         int ret;
 310         uint64_t i;
 311         char *data = NULL;
 312         uint64_t len = 0;
 313         int metadata_stream_id = -1;
 314
 315         cmd.cmd = htobe32(VIEWER_GET_METADATA);
 316         cmd.data_size = sizeof(rq);
 317         cmd.cmd_version = 0;
 318
 319         for (i = 0; i < session->stream_count; i++) {
 320                 if (session->streams[i].metadata_flag) {
 321                         metadata_stream_id = i;
 322                         break;
 323                 }
 324         }
 325
 326         if (metadata_stream_id < 0) {
 327                 fprintf(stderr, "No metadata stream found\n");
 328                 ret = -1;
 329                 goto error;
 330         }
 331
 332         rq.stream_id = htobe64(session->streams[metadata_stream_id].id);
 333
 334         do {
 335                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 336         } while (ret < 0 && errno == EINTR);
 337         if (ret < 0) {
 338                 fprintf(stderr, "Error sending cmd\n");
 339                 goto error;
 340         }
 341         do {
 342                 ret = send(control_sock, &rq, sizeof(rq), 0);
 343         } while (ret < 0 && errno == EINTR);
 344         if (ret < 0) {
 345                 fprintf(stderr, "Error sending get_metadata request\n");
 346                 goto error;
 347         }
 348         do {
 349                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 350         } while (ret < 0 && errno == EINTR);
 351         if (ret < 0) {
 352                 fprintf(stderr, "Error receiving metadata response\n");
 353                 goto error;
 354         }
 355         switch (be32toh(rp.status)) {
 356                 case VIEWER_METADATA_OK:
 357                         break;
 358                 case VIEWER_NO_NEW_METADATA:
 359                         fprintf(stderr, "NO NEW\n");
 360                         ret = -1;
 361                         goto end;
 362                 case VIEWER_METADATA_ERR:
 363                         fprintf(stderr, "ERR\n");
 364                         ret = -1;
 365                         goto end;
 366                 default:
 367                         fprintf(stderr, "UNKNOWN\n");
 368                         ret = -1;
 369                         goto end;
 370         }
 371
 372         len = be64toh(rp.len);
 373         if (len <= 0) {
 374                 goto end;
 375         }
 376
 377         data = zmalloc(len);
 378         if (!data) {
 379                 perror("relay data zmalloc");
 380                 goto error;
 381         }
 382         do {
 383                 ret = recv(control_sock, data, len, MSG_WAITALL);
 384         } while (ret < 0 && errno == EINTR);
 385         if (ret < 0) {
 386                 fprintf(stderr, "Error receiving trace packet\n");
 387                 free(data);
 388                 goto error;
 389         }
 390         free(data);
 391
 392         ret = (int) len;
 393 end:
 394 error:
 395         return ret;
 396 }
 397
 398 int get_next_index(void)
 399 {
 400         struct lttng_viewer_cmd cmd;
 401         struct lttng_viewer_get_next_index rq;
 402         struct lttng_viewer_index rp;
 403         int ret;
 404         int id;
 405
 406         cmd.cmd = htobe32(VIEWER_GET_NEXT_INDEX);
 407         cmd.data_size = sizeof(rq);
 408         cmd.cmd_version = 0;
 409
 410         for (id = 0; id < session->stream_count; id++) {
 411                 if (session->streams[id].metadata_flag) {
 412                         continue;
 413                 }
 414                 rq.stream_id = htobe64(session->streams[id].id);
 415
 416 retry:
 417                 do {
 418                         ret = send(control_sock, &cmd, sizeof(cmd), 0);
 419                 } while (ret < 0 && errno == EINTR);
 420                 if (ret < 0) {
 421                         fprintf(stderr, "Error sending cmd\n");
 422                         goto error;
 423                 }
 424                 do {
 425                         ret = send(control_sock, &rq, sizeof(rq), 0);
 426                 } while (ret < 0 && errno == EINTR);
 427                 if (ret < 0) {
 428                         fprintf(stderr, "Error sending get_next_index request\n");
 429                         goto error;
 430                 }
 431                 do {
 432                         ret = recv(control_sock, &rp, sizeof(rp), 0);
 433                 } while (ret < 0 && errno == EINTR);
 434                 if (ret < 0) {
 435                         fprintf(stderr, "Error receiving index response\n");
 436                         goto error;
 437                 }
 438
 439                 rp.flags = be32toh(rp.flags);
 440
 441                 switch (be32toh(rp.status)) {
 442                         case VIEWER_INDEX_INACTIVE:
 443                                 fprintf(stderr, "(INACTIVE)\n");
 444                                 break;
 445                         case VIEWER_INDEX_OK:
 446                                 break;
 447                         case VIEWER_INDEX_RETRY:
 448                                 sleep(1);
 449                                 goto retry;
 450                         case VIEWER_INDEX_HUP:
 451                                 fprintf(stderr, "(HUP)\n");
 452                                 session->streams[id].id = -1ULL;
 453                                 session->streams[id].fd = -1;
 454                                 break;
 455                         case VIEWER_INDEX_ERR:
 456                                 fprintf(stderr, "(ERR)\n");
 457                                 ret = -1;
 458                                 goto error;
 459                         default:
 460                                 fprintf(stderr, "SHOULD NOT HAPPEN\n");
 461                                 ret = -1;
 462                                 goto error;
 463                 }
 464                 if (!first_packet_stream_id) {
 465                         first_packet_offset = be64toh(rp.offset);
 466                         first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
 467                         first_packet_stream_id = id;
 468                 }
 469         }
 470         ret = 0;
 471
 472 error:
 473         return ret;
 474 }
 475
 476 static
 477 int get_data_packet(int id, uint64_t offset,
 478                 uint64_t len)
 479 {
 480         struct lttng_viewer_cmd cmd;
 481         struct lttng_viewer_get_packet rq;
 482         struct lttng_viewer_trace_packet rp;
 483         int ret;
 484
 485         cmd.cmd = htobe32(VIEWER_GET_PACKET);
 486         cmd.data_size = sizeof(rq);
 487         cmd.cmd_version = 0;
 488
 489         rq.stream_id = htobe64(session->streams[id].id);
 490         /* Already in big endian. */
 491         rq.offset = offset;
 492         rq.len = htobe32(len);
 493
 494         do {
 495                 ret = send(control_sock, &cmd, sizeof(cmd), 0);
 496         } while (ret < 0 && errno == EINTR);
 497         if (ret < 0) {
 498                 fprintf(stderr, "Error sending cmd\n");
 499                 goto error;
 500         }
 501         do {
 502                 ret = send(control_sock, &rq, sizeof(rq), 0);
 503         } while (ret < 0 && errno == EINTR);
 504         if (ret < 0) {
 505                 fprintf(stderr, "Error sending get_data_packet request\n");
 506                 goto error;
 507         }
 508         do {
 509                 ret = recv(control_sock, &rp, sizeof(rp), 0);
 510         } while (ret < 0 && errno == EINTR);
 511         if (ret < 0) {
 512                 fprintf(stderr, "Error receiving data response\n");
 513                 goto error;
 514         }
 515         rp.flags = be32toh(rp.flags);
 516
 517         switch (be32toh(rp.status)) {
 518         case VIEWER_GET_PACKET_OK:
 519                 break;
 520         case VIEWER_GET_PACKET_RETRY:
 521                 fprintf(stderr, "RETRY\n");
 522                 ret = -1;
 523                 goto end;
 524         case VIEWER_GET_PACKET_ERR:
 525                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 526                         fprintf(stderr, "NEW_METADATA\n");
 527                         ret = 0;
 528                         goto end;
 529                 }
 530                 fprintf(stderr, "ERR\n");
 531                 ret = -1;
 532                 goto end;
 533         default:
 534                 fprintf(stderr, "UNKNOWN\n");
 535                 ret = -1;
 536                 goto end;
 537         }
 538
 539         len = be32toh(rp.len);
 540         if (len <= 0) {
 541                 goto end;
 542         }
 543
 544         if (len > mmap_size) {
 545                 fprintf(stderr, "mmap_size not big enough\n");
 546                 ret = -1;
 547                 goto error;
 548         }
 549
 550         do {
 551                 ret = recv(control_sock, session->streams[id].mmap_base, len, MSG_WAITALL);
 552         } while (ret < 0 && errno == EINTR);
 553         if (ret < 0) {
 554                 fprintf(stderr, "Error receiving trace packet\n");
 555                 goto error;
 556         }
 557         ret = len;
 558
 559 end:
 560 error:
 561         return ret;
 562 }
 563
 564 int main(int argc, char **argv)
 565 {
 566         int ret;
 567         int session_id;
 568
 569         plan_tests(NUM_TESTS);
 570
 571         diag("Live unit tests");
 572
 573         ret = connect_viewer("localhost");
 574         ok(ret == 0, "Connect viewer to relayd");
 575
 576         ret = establish_connection();
 577         ok(ret == 0, "Established connection and version check with %d.%d",
 578                         VERSION_MAJOR, VERSION_MINOR);
 579
 580         ret = list_sessions(&session_id);
 581         ok(ret > 0, "List sessions : %d session(s)", ret);
 582
 583         ret = attach_session(session_id);
 584         ok(ret > 0, "Attach to session, %d streams received", ret);
 585
 586         ret = get_metadata();
 587         ok(ret > 0, "Get metadata, received %d bytes", ret);
 588
 589         ret = get_next_index();
 590         ok(ret == 0, "Get one index per stream");
 591
 592         ret = get_data_packet(first_packet_stream_id, first_packet_offset,
 593                         first_packet_len);
 594         ok(ret == first_packet_len,
 595                         "Get one data packet for stream %d, offset %d, len %d",
 596                         first_packet_stream_id, first_packet_offset,
 597                         first_packet_len);
 598
 599         return exit_status();
 600 }