<commit_before>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2018 Treasure Data Inc.
 *
 *  Licensed 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
 *
 *      http://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.
 */

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>

#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_parser.h>

#include "tail.h"
#include "tail_file.h"
#include "tail_config.h"
#include "tail_db.h"
#include "tail_signal.h"
#include "tail_multiline.h"
#include "tail_scan.h"

static inline void consume_bytes(char *buf, int bytes, int length)
{
    memmove(buf, buf + bytes, length - bytes);
}

static int unpack_and_pack(msgpack_packer *pck, msgpack_object *root,
                           char *key, size_t key_len,
                           char *val, size_t val_len)
{
    int i;
    int size = root->via.map.size;

    msgpack_pack_map(pck, size + 1);

    /* Append new k/v */
    msgpack_pack_str(pck, key_len);
    msgpack_pack_str_body(pck, key, key_len);
    msgpack_pack_str(pck, val_len);
    msgpack_pack_str_body(pck, val, val_len);

    for (i = 0; i < size; i++) {
        msgpack_object k = root->via.map.ptr[i].key;
        msgpack_object v = root->via.map.ptr[i].val;;

        msgpack_pack_object(pck, k);
        msgpack_pack_object(pck, v);
    }

    return 0;
}

static int append_record_to_map(char **data, size_t *data_size,
                                char *key,  size_t key_len,
                                char *val,  size_t val_len)
{
    int ret;
    msgpack_unpacked result;
    msgpack_object   root;
    msgpack_sbuffer sbuf;
    msgpack_packer  pck;
    size_t off = 0;

    msgpack_sbuffer_init(&sbuf);
    msgpack_packer_init(&pck, &sbuf, msgpack_sbuffer_write);
    msgpack_unpacked_init(&result);

    ret = msgpack_unpack_next(&result, *data, *data_size, &off);
    if (ret != MSGPACK_UNPACK_SUCCESS) {
        msgpack_unpacked_destroy(&result);
        msgpack_sbuffer_destroy(&sbuf);
        return -1;
    }

    root = result.data;
    ret = unpack_and_pack(&pck, &root,
                          key, key_len, val, val_len);
    if (ret < 0) {
        /* fail! */
        msgpack_unpacked_destroy(&result);
        msgpack_sbuffer_destroy(&sbuf);
        return -1;
    }
    else {
        /* success !*/
        flb_free(*data);
        *data      = sbuf.data;
        *data_size = sbuf.size;
    }

    msgpack_unpacked_destroy(&result);
    return 0;
}

static inline int pack_line_map(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                                struct flb_time *time, char **data,
                                size_t *data_size, struct flb_tail_file *file)
{
    int map_num = 1;

    if (file->config->path_key != NULL) {
        map_num++; /* to append path_key */
    }

    if (file->config->path_key != NULL) {
        append_record_to_map(data, data_size,
                             file->config->path_key,
                             file->config->path_key_len,
                             file->name, file->name_len);
    }

    msgpack_pack_array(mp_pck, 2);
    flb_time_append_to_msgpack(time, mp_pck, 0);
    msgpack_sbuffer_write(mp_sbuf, *data, *data_size);

    return 0;
}

int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                            struct flb_time *time, char *data, size_t data_size,
                            struct flb_tail_file *file)
{
    int map_num = 1;
    struct flb_tail_config *ctx = file->config;

    if (file->config->path_key != NULL) {
        map_num++; /* to append path_key */
    }
    msgpack_pack_array(mp_pck, 2);
    flb_time_append_to_msgpack(time, mp_pck, 0);
    msgpack_pack_map(mp_pck, map_num);

    if (file->config->path_key != NULL) {
        /* append path_key */
        msgpack_pack_str(mp_pck, file->config->path_key_len);
        msgpack_pack_str_body(mp_pck, file->config->path_key,
                              file->config->path_key_len);
        msgpack_pack_str(mp_pck, file->name_len);
        msgpack_pack_str_body(mp_pck, file->name, file->name_len);
    }

    msgpack_pack_str(mp_pck, ctx->key_len);
    msgpack_pack_str_body(mp_pck, ctx->key, ctx->key_len);
    msgpack_pack_str(mp_pck, data_size);
    msgpack_pack_str_body(mp_pck, data, data_size);

    return 0;
}

static int process_content(struct flb_tail_file *file, off_t *bytes)
{
    int len;
    int lines = 0;
    int ret;
    off_t processed_bytes = 0;
    char *data;
    char *p;
    void *out_buf;
    size_t out_size;
    time_t now = time(NULL);
    struct flb_time out_time = {};
    msgpack_sbuffer mp_sbuf;
    msgpack_packer mp_pck;
    msgpack_sbuffer *out_sbuf;
    msgpack_packer *out_pck;
    struct flb_tail_config *ctx = file->config;

    /* Create a temporal msgpack buffer */
    msgpack_sbuffer_init(&mp_sbuf);
    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
    out_sbuf = &mp_sbuf;
    out_pck  = &mp_pck;

    /* Parse the data content */
    data = file->buf_data;
    while ((p = strchr(data, '\n'))) {
        len = (p - data);

        if (file->skip_next == FLB_TRUE) {
            data += len + 1;
            processed_bytes += len;
            file->skip_next = FLB_FALSE;
            continue;
        }

        /* Empty line (just \n) */<commit_msg>in_tail: include NULL bytes and don't stop processing (#427)<commit_after>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2018 Treasure Data Inc.
 *
 *  Licensed 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
 *
 *      http://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.
 */

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>

#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_parser.h>

#include "tail.h"
#include "tail_file.h"
#include "tail_config.h"
#include "tail_db.h"
#include "tail_signal.h"
#include "tail_multiline.h"
#include "tail_scan.h"

static inline void consume_bytes(char *buf, int bytes, int length)
{
    memmove(buf, buf + bytes, length - bytes);
}

static int unpack_and_pack(msgpack_packer *pck, msgpack_object *root,
                           char *key, size_t key_len,
                           char *val, size_t val_len)
{
    int i;
    int size = root->via.map.size;

    msgpack_pack_map(pck, size + 1);

    /* Append new k/v */
    msgpack_pack_str(pck, key_len);
    msgpack_pack_str_body(pck, key, key_len);
    msgpack_pack_str(pck, val_len);
    msgpack_pack_str_body(pck, val, val_len);

    for (i = 0; i < size; i++) {
        msgpack_object k = root->via.map.ptr[i].key;
        msgpack_object v = root->via.map.ptr[i].val;;

        msgpack_pack_object(pck, k);
        msgpack_pack_object(pck, v);
    }

    return 0;
}

static int append_record_to_map(char **data, size_t *data_size,
                                char *key,  size_t key_len,
                                char *val,  size_t val_len)
{
    int ret;
    msgpack_unpacked result;
    msgpack_object   root;
    msgpack_sbuffer sbuf;
    msgpack_packer  pck;
    size_t off = 0;

    msgpack_sbuffer_init(&sbuf);
    msgpack_packer_init(&pck, &sbuf, msgpack_sbuffer_write);
    msgpack_unpacked_init(&result);

    ret = msgpack_unpack_next(&result, *data, *data_size, &off);
    if (ret != MSGPACK_UNPACK_SUCCESS) {
        msgpack_unpacked_destroy(&result);
        msgpack_sbuffer_destroy(&sbuf);
        return -1;
    }

    root = result.data;
    ret = unpack_and_pack(&pck, &root,
                          key, key_len, val, val_len);
    if (ret < 0) {
        /* fail! */
        msgpack_unpacked_destroy(&result);
        msgpack_sbuffer_destroy(&sbuf);
        return -1;
    }
    else {
        /* success !*/
        flb_free(*data);
        *data      = sbuf.data;
        *data_size = sbuf.size;
    }

    msgpack_unpacked_destroy(&result);
    return 0;
}

static inline int pack_line_map(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                                struct flb_time *time, char **data,
                                size_t *data_size, struct flb_tail_file *file)
{
    int map_num = 1;

    if (file->config->path_key != NULL) {
        map_num++; /* to append path_key */
    }

    if (file->config->path_key != NULL) {
        append_record_to_map(data, data_size,
                             file->config->path_key,
                             file->config->path_key_len,
                             file->name, file->name_len);
    }

    msgpack_pack_array(mp_pck, 2);
    flb_time_append_to_msgpack(time, mp_pck, 0);
    msgpack_sbuffer_write(mp_sbuf, *data, *data_size);

    return 0;
}

int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                            struct flb_time *time, char *data, size_t data_size,
                            struct flb_tail_file *file)
{
    int map_num = 1;
    struct flb_tail_config *ctx = file->config;

    if (file->config->path_key != NULL) {
        map_num++; /* to append path_key */
    }
    msgpack_pack_array(mp_pck, 2);
    flb_time_append_to_msgpack(time, mp_pck, 0);
    msgpack_pack_map(mp_pck, map_num);

    if (file->config->path_key != NULL) {
        /* append path_key */
        msgpack_pack_str(mp_pck, file->config->path_key_len);
        msgpack_pack_str_body(mp_pck, file->config->path_key,
                              file->config->path_key_len);
        msgpack_pack_str(mp_pck, file->name_len);
        msgpack_pack_str_body(mp_pck, file->name, file->name_len);
    }

    msgpack_pack_str(mp_pck, ctx->key_len);
    msgpack_pack_str_body(mp_pck, ctx->key, ctx->key_len);
    msgpack_pack_str(mp_pck, data_size);
    msgpack_pack_str_body(mp_pck, data, data_size);

    return 0;
}

static int process_content(struct flb_tail_file *file, off_t *bytes)
{
    int len;
    int lines = 0;
    int ret;
    off_t processed_bytes = 0;
    char *data;
    char *end;
    char *p;
    void *out_buf;
    size_t out_size;
    time_t now = time(NULL);
    struct flb_time out_time = {};
    msgpack_sbuffer mp_sbuf;
    msgpack_packer mp_pck;
    msgpack_sbuffer *out_sbuf;
    msgpack_packer *out_pck;
    struct flb_tail_config *ctx = file->config;

    /* Create a temporal msgpack buffer */
    msgpack_sbuffer_init(&mp_sbuf);
    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
    out_sbuf = &mp_sbuf;
    out_pck  = &mp_pck;

    /* Parse the data content */
    data = file->buf_data;
    end = data + file->buf_len;
    while ((p = memchr(data, '\n', end - data))) {
        len = (p - data);

        if (file->skip_next == FLB_TRUE) {
            data += len + 1;
            processed_bytes += len;
            file->skip_next = FLB_FALSE;
            continue;
        }

        /* Empty line (just \n) */
{'fraction_non_alphanumeric': 0.11677852348993288, 'fraction_numerical': 0.005788590604026846, 'mean_word_length': 2.4503617945007234, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 88, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6059366', 'n_tokens_mistral': 4320, 'n_tokens_neox': 4204, 'n_words': 927}