text: /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
  
  #define _GNU_SOURCE /* for accept4 */
  #include <fluent-bit.h>
  #include <fluent-bit/flb_info.h>
  #include "flb_tests_runtime.h"
  
  #include <sys/types.h>
  #include <sys/stat.h>
  #include <unistd.h>
  
  struct kube_test {
      flb_ctx_t *flb;
  };
  
  struct kube_test_result {
      char *target;
      char *suffix;
      int   type;
      int   nmatched;
  };
  
  /* Test target mode */
  #define KUBE_TAIL     0
  #define KUBE_SYSTEMD  1
  
  #ifdef FLB_HAVE_SYSTEMD
  int flb_test_systemd_send(void);
  char kube_test_id[64];
  #endif
  
  /* Constants */
  #define KUBE_IP          "127.0.0.1"
  #define KUBE_PORT        "8002"
  #define KUBE_URL         "http://" KUBE_IP ":" KUBE_PORT
  #define DPATH            FLB_TESTS_DATA_PATH "/data/kubernetes/"
  #define STD_PARSER       "../conf/parsers.conf"
  
  /*
   * Data files
   * ==========
   */
  #define T_APACHE_LOGS           DPATH "apache-logs_default"
  #define T_APACHE_LOGS_ANN       DPATH "apache-logs-annotated_default"
  #define T_APACHE_LOGS_ANN_INV   DPATH "apache-logs-annotated-invalid"
  #define T_APACHE_LOGS_ANN_MERGE DPATH "apache-logs-annotated-merge"
  #define T_APACHE_LOGS_ANN_EXCL  DPATH "apache-logs-annotated-exclude"
  #define T_JSON_LOGS             DPATH "json-logs_default"
  #define T_JSON_LOGS_NO_KEEP     DPATH "json-logs-no-keep"
  #define T_JSON_LOGS_INV         DPATH "json-logs-invalid"
  #define T_JSON_LOGS_INV_NO_KEEP DPATH "json-logs-invalid"
  #define T_SYSTEMD_SIMPLE        DPATH "kairosdb-914055854-b63vq"
  
  #define T_MULTI_INIT            DPATH "session-db-fdd649d68-cq5sp_socks_istio-init-"
  #define T_MULTI_PROXY           DPATH "session-db-fdd649d68-cq5sp_socks_istio-proxy-"
  #define T_MULTI_REDIS           DPATH "session-db-fdd649d68-cq5sp_socks_istio-session-db-"
  
  static int file_to_buf(char *path, char **out_buf, size_t *out_size)
  {
      int ret;
      long bytes;
      char *buf;
      FILE *fp;
      struct stat st;
  
      ret = stat(path, &st);
      if (ret == -1) {
          return -1;
      }
  
      fp = fopen(path, "r");
      if (!fp) {
          return -1;
      }
  
      buf = flb_malloc(st.st_size);
      if (!buf) {
          flb_errno();
          fclose(fp);
          return -1;
      }
  
      bytes = fread(buf, st.st_size, 1, fp);
      if (bytes != 1) {
          flb_errno();
          flb_free(buf);
          fclose(fp);
          return -1;
      }
  
      fclose(fp);
      *out_buf = buf;
      *out_size = st.st_size;
  
      return 0;
  }
  
  /* Given a target, lookup the .out file and return it content in a new buffer */
  static char *get_out_file_content(char *target, char *suffix)
  {
      int ret;
      char file[PATH_MAX];
      char *p;
      char *out_buf;
      size_t out_size;
  
      snprintf(file, sizeof(file) - 1, "%s%s.out", target,suffix);
  
      ret = file_to_buf(file, &out_buf, &out_size);
      if (ret != 0) {
          flb_error("no output file found '%s'", file);
          exit(EXIT_FAILURE);
      }
  
      /* Sanitize content, get rid of ending \n */
      p = out_buf + (out_size - 1);
      while (*p == '\n' || *p == '\r') p--;
      *++p = '\0';
  
      return out_buf;
  }
  
  static int cb_check_result(void *record, size_t size, void *data)
  {
      struct kube_test_result *result;
      char *out;
      char *check;
      char streamfilter[64];
  
      result = (struct kube_test_result *) data;
      out = get_out_file_content(result->target, result->suffix);
      if (!out) {
          exit(EXIT_FAILURE);
      }
      if (result->type == KUBE_SYSTEMD) {
          char *skip_record, *skip_out;
          /* Skip the other records since some are created by systemd,
             only check the kubernetes annotations
           */
          skip_out = strstr(out, "\"kubernetes\":");
          skip_record = strstr(record, "\"kubernetes\":");
          if (skip_out && skip_record) {
              TEST_CHECK(strcmp(skip_record, skip_out) == 0);
              result->nmatched++;
          }
      } else {
          sprintf(streamfilter, "\"stream\":\"%s\"", result->suffix);
          if (!result->suffix ||
              !*result->suffix ||
              strstr(record, streamfilter)) {
  
              /*
               * Our validation is: check that the content of out file is found
               * in the output record.
               */
              check = strstr(record, out);
              if (!check) {
                  fprintf(stderr, "Validator mismatch::\n"
                                  "Target: <<%s>>, Suffix: <<%s>\n"
                                  "Filtered record: <<%s>>\n"
                                  "Expected record: <<%s>>\n",
                                  result->target, result->suffix,
                                  (char *)record, out);
              }
              TEST_CHECK(check != NULL);
              result->nmatched++;
          }
      }
      if (size > 0) {
          flb_free(record);
      }
      flb_free(out);
      return 0;
  }
  
  static void kube_test_destroy(struct kube_test *ctx)
  {
      flb_stop(ctx->flb);
      flb_destroy(ctx->flb);
      flb_free(ctx);
  }
  
  static void kube_test_create(char *target, int type, char *suffix, char *parserconf, int nExpected, ...)
  {
      int i;
      int ret;
      int in_ffd;
      int filter_ffd;
      int out_ffd;
      char *key;
      char *value;
      char path[PATH_MAX];
      char tag_prefix[PATH_MAX];
      va_list va;
      struct kube_test *ctx;
      struct flb_lib_out_cb cb_data;
      struct kube_test_result result = {0};
  
      result.nmatched = 0;
      result.target = target;
      result.suffix = suffix;
      result.type = type;
  
      /* Compose path pattern based on target */
      snprintf(path, sizeof(path) - 1, "%s*.log", target);
  
  
      /* Kubernetes tag prefix */
      if (type == KUBE_TAIL) {
          ret = snprintf(tag_prefix, sizeof(tag_prefix) - 1,
                         "kube%s.data.kubernetes.", FLB_TESTS_DATA_PATH);
          for (i = 0; i < ret; i++) {
              if (tag_prefix[i] == '/') {
                  tag_prefix[i] = '.';
              }
          }
      }
  
      ctx = flb_malloc(sizeof(struct kube_test));
      if (!ctx) {
          flb_errno();
          TEST_CHECK(ctx != NULL);
          exit(EXIT_FAILURE);
      }
  
      ctx->flb = flb_create();
      flb_service_set(ctx->flb,
                      "Flush", "1",
                      "Grace", "1",
                      "Parsers_File", parserconf,
                      NULL);
  
      if (type == KUBE_TAIL) {
          in_ffd = flb_input(ctx->flb, "tail", NULL);
          ret = flb_input_set(ctx->flb, in_ffd,
                              "Tag", "kube.*",
                              "Path", path,
                              "Parser", "docker",
                              "Decode_Field", "json log",
                              NULL);
          TEST_CHECK(ret == 0);
      }
  #ifdef FLB_HAVE_SYSTEMD
      else if (type == KUBE_SYSTEMD) {
          sprintf(kube_test_id, "KUBE_TEST=%u%lu", getpid(), random());
          in_ffd = flb_input(ctx->flb, "systemd", NULL);
          ret = flb_input_set(ctx->flb, in_ffd,
                              "Tag", "kube.*",
                              "Systemd_Filter", kube_test_id,
                              NULL);
          TEST_CHECK(ret == 0);
      }
  #endif
  
      filter_ffd = flb_filter(ctx->flb, "kubernetes", NULL);
      ret = flb_filter_set(ctx->flb, filter_ffd,
                           "Match", "kube.*",
                           "Kube_URL", KUBE_URL,
                           "k8s-logging.parser", "On",
                           "k8s-logging.exclude", "On",
                           "Kube_Meta_Preload_Cache_Dir", "../tests/runtime/data/kubernetes",
                           NULL);
  
      /* Iterate number of arguments for filter_kubernetes additional options */
      va_start(va, nExpected);
      while ((key = va_arg(va, char *))) {
          value = va_arg(va, char *);
          if (!value) {
              /* Wrong parameter */
              break;
          }
          flb_filter_set(ctx->flb, filter_ffd, key, value, NULL);
      }
      va_end(va);
  
      if (type == KUBE_TAIL) {
          ret = flb_filter_set(ctx->flb, filter_ffd,
                               "Regex_Parser", "filter-kube-test",
                               "Kube_Tag_Prefix", tag_prefix,
                               NULL);
      }
  #ifdef FLB_HAVE_SYSTEMD
      else if (type == KUBE_SYSTEMD) {
          flb_filter_set(ctx->flb, filter_ffd,
                         "Use_Journal", "On",
                         NULL);
      }
  #endif
  
      /* Prepare output callback context*/
      cb_data.cb = cb_check_result;
      cb_data.data = &result;
  
      /* Output */
      out_ffd = flb_output(ctx->flb, "lib", (void *) &cb_data);
      TEST_CHECK(out_ffd >= 0);
      flb_output_set(ctx->flb, out_ffd,
                     "Match", "kube.*",
                     "format", "json",
                     NULL);
  
  #ifdef FLB_HAVE_SYSTEMD
      /*
       *  If the source of data is Systemd, just let the output lib plugin
       * to process one record only, otherwise when the test case stop after
       * the first callback it destroy the contexts, but out_lib still have
       * pending data to flush. This option solves the problem.
       */
      if (type == KUBE_SYSTEMD) {
          flb_output_set(ctx->flb, out_ffd,
                         "Max_Records", "1",
                         NULL);
      }
  #endif
  
      /* Start the engine */
      ret = flb_start(ctx->flb);
      TEST_CHECK(ret == 0);
      if (ret == -1) {
          exit(EXIT_FAILURE);
      }
  #ifdef FLB_HAVE_SYSTEMD
      if (type == KUBE_SYSTEMD) {
          TEST_CHECK_(flb_test_systemd_send() >= 0,
                      "Error sending sample message to journal");
      }
  #endif
  
      /* Poll for up to 2 seconds or until we got a match */
      for (ret = 0; ret < 2000 && result.nmatched == 0; ret++) {
          usleep(1000);
      }
      TEST_CHECK(result.nmatched == nExpected);
  
      kube_test_destroy(ctx);
  }
  
  void flb_test_apache_logs()
  {
      kube_test_create(T_APACHE_LOGS, KUBE_TAIL, "", STD_PARSER, 1, NULL);
  }
  
  void flb_test_apache_logs_merge()
  {
      kube_test_create(T_APACHE_LOGS, KUBE_TAIL, "", STD_PARSER,
                       1,
                       "Merge_Log", "On",
                       "Merge_Log_Key", "merge",
                       NULL);
  }
  
  void flb_test_apache_logs_annotated()
  {
      kube_test_create(T_APACHE_LOGS_ANN, KUBE_TAIL, "", STD_PARSER,
                       1,
                       "Merge_Log", "On",
                       NULL);
  }
  
  void flb_test_apache_logs_annotated_invalid()
  {
      kube_test_create(T_APACHE_LOGS_ANN_INV, KUBE_TAIL, "", STD_PARSER, 1, NULL);
  }
  
  void flb_test_apache_logs_annotated_exclude()
  {
      kube_test_create(T_APACHE_LOGS_ANN_EXCL, KUBE_TAIL, "", STD_PARSER, 0, NULL);
  }
  
  void flb_test_apache_logs_annotated_merge()
  {
      kube_test_create(T_APACHE_LOGS_ANN_MERGE, KUBE_TAIL, "", STD_PARSER, 1,
                       "Merge_Log", "On",
                       "Merge_Log_Key", "merge", NULL);
  }
  
  void flb_test_json_logs()
  {
      kube_test_create(T_JSON_LOGS, KUBE_TAIL, "", STD_PARSER, 1,
                       "Merge_Log", "On",
                       NULL);
  }
  
  void flb_test_json_logs_no_keep()
  {
      kube_test_create(T_JSON_LOGS_NO_KEEP, KUBE_TAIL, "", STD_PARSER, 1,
                       "Merge_Log", "On",
                       "Keep_Log", "Off",
                       NULL);
  }
  
  void flb_test_json_logs_invalid()
  {
      kube_test_create(T_JSON_LOGS_INV, KUBE_TAIL, "", STD_PARSER, 1, NULL);
  }
  
  void flb_test_json_logs_invalid_no_keep()
  {
      kube_test_create(T_JSON_LOGS_INV_NO_KEEP, KUBE_TAIL, "", STD_PARSER, 1,
                       "Merge_Log", "On",
                       "Keep_Log", "Off",
                       NULL);
  }
  
  #ifdef FLB_HAVE_SYSTEMD
  #define CONTAINER_NAME "CONTAINER_NAME=k8s_kairosdb_kairosdb-914055854-b63vq_default_d6c53deb-05a4-11e8-a8c4-080027435fb7_23"
  #include <systemd/sd-journal.h>
  
  int flb_test_systemd_send()
  {
      return sd_journal_send(
              "@timestamp=2018-02-23T08:58:45.0Z",
              "PRIORITY=6",
              CONTAINER_NAME,
              "CONTAINER_TAG=",
              "CONTAINER_ID=56e257661383",
              "CONTAINER_ID_FULL=56e257661383836fac4cd90a23ee8a7a02ee1538c8f35657d1a90f3de1065a22",
              "MESSAGE=08:58:45.839 [qtp151442075-47] DEBUG [HttpParser.java:281] - filled 157/157",
              kube_test_id,
              NULL);
  }
  
  void flb_test_systemd_logs()
  {
      struct stat statb;
      /* We want to avoid possibly getting a log from a previous run,
         so create a unique-id for each 'send' */
      sprintf(kube_test_id, "KUBE_TEST=%u%lu", getpid(), random());
  
      if (stat("/run/systemd/journal/socket", &statb) == 0 &&
          statb.st_mode & S_IFSOCK) {
  
          int r;
          sd_journal *journal;
          r = sd_journal_open(&journal, 0);
          if (r < 0) {
              flb_error("Skip test: journal error: ", strerror(-r));
              return;
          }
  
          r = sd_journal_get_fd(journal);
          if (r < 0) {
              flb_error("Skip test: journal fd error: ", strerror(-r));
              sd_journal_close(journal);
              return;
          }
          sd_journal_add_match(journal, CONTAINER_NAME, 0);
          sd_journal_seek_tail(journal);
  
          /*
           * Send test message to Journal. If this fails (e.g. journal is
           * not running then skip the test.
           */
          if (flb_test_systemd_send() < 0) {
  
              flb_error("Skip test: journal send error: ", strerror(-r));
              sd_journal_close(journal);
              return;
          }
  
          r = sd_journal_previous(journal);
          if (r < 0) {
              flb_error("Skip test: journal previous error: ", strerror(-r));
              sd_journal_close(journal);
              return;
          }
  
          r = sd_journal_next(journal);
          if (r < 0) {
              flb_error("Skip test: journal next error: ", strerror(-r));
              sd_journal_close(journal);
              return;
          }
  
          r = sd_journal_wait(journal, 2000);
          if (r < 0) {
              flb_error("Skip test: journal wait error: ", strerror(-r));
              sd_journal_close(journal);
              return;
          }
          sd_journal_close(journal);
  
          kube_test_create(T_SYSTEMD_SIMPLE, KUBE_SYSTEMD, "", STD_PARSER, 1,
                           "Merge_Log", "On",
                           NULL);
      }
  }
  #endif
  
  void flb_test_multi_logs(char *log, char *suffix)
  {
      flb_info("\n");
      flb_info("Multi test: log <%s>", log);
      kube_test_create(log, KUBE_TAIL, suffix,
                       "../tests/runtime/data/kubernetes/multi-parsers.conf", 1,
                       "Merge_Log", "On", NULL);
  }
  
  void flb_test_multi_init_stdout() { flb_test_multi_logs(T_MULTI_INIT, "stdout"); }
  void flb_test_multi_init_stderr() { flb_test_multi_logs(T_MULTI_INIT, "stderr"); }
  void flb_test_multi_proxy() { flb_test_multi_logs(T_MULTI_PROXY, ""); }
  void flb_test_multi_redis() { flb_test_multi_logs(T_MULTI_REDIS, ""); }
  
  TEST_LIST = {
      {"kube_apache_logs", flb_test_apache_logs},
      {"kube_apache_logs_merge", flb_test_apache_logs_merge},
      {"kube_apache_logs_annotated", flb_test_apache_logs_annotated},
      {"kube_apache_logs_annotated_invalid", flb_test_apache_logs_annotated_invalid},
      {"kube_apache_logs_annotated_exclude", flb_test_apache_logs_annotated_exclude},
      {"kube_apache_logs_annotated_merge_log", flb_test_apache_logs_annotated_merge},
      {"kube_json_logs", flb_test_json_logs},
      {"kube_json_logs_no_keep", flb_test_json_logs_no_keep},
      {"kube_json_logs_invalid", flb_test_json_logs_invalid},
      {"kube_json_logs_invalid_no_keep", flb_test_json_logs_invalid_no_keep},
  #ifdef FLB_HAVE_SYSTEMD
      {"kube_systemd_logs", flb_test_systemd_logs},
  #endif
      {"kube_multi_init_stdout", flb_test_multi_init_stdout},
      {"kube_multi_init_stderr", flb_test_multi_init_stderr},
      {"kube_multi_proxy", flb_test_multi_proxy},
      {"kube_multi_redis", flb_test_multi_redis},
      {NULL, NULL}
  };
  