<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 <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_error.h>
#include <fluent-bit/flb_utils.h>

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

static inline int consume_byte(int fd)
{
    int ret;
    uint64_t val;

    /* We need to consume the byte */
    ret = read(fd, &val, sizeof(val));
    if (ret <= 0) {
        flb_errno();
        return -1;
    }

    return 0;
}

/* cb_collect callback */
static int in_tail_collect_pending(struct flb_input_instance *i_ins,
                                   struct flb_config *config, void *in_context)
{
    int ret;
    int active = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_tail_config *ctx = in_context;
    struct flb_tail_file *file;
    struct stat st;

    /* Iterate promoted event files with pending bytes */
    mk_list_foreach_safe(head, tmp, &ctx->files_event) {
        file = mk_list_entry(head, struct flb_tail_file, _head);
        if (file->pending_bytes <= 0) {
            continue;
        }

        /* Gather current file size */
        ret = fstat(file->fd, &st);
        if (ret == -1) {
            flb_errno();
            flb_tail_file_remove(file);
            continue;
        }

        ret = flb_tail_file_chunk(file);
        switch (ret) {
        case FLB_TAIL_ERROR:
            /* Could not longer read the file */
            flb_tail_file_remove(file);
            break;
        case FLB_TAIL_OK:
        case FLB_TAIL_BUSY:
            /*
             * Adjust counter to verify if we need a further read(2) later.
             * For more details refer to tail_fs_inotify.c:96.
             */
            if (file->offset < st.st_size) {
                file->pending_bytes = (st.st_size - file->offset);
                active++;
            }
            else {
                file->pending_bytes = 0;
            }
            break;
        }
    }

    /* If no more active files, consume pending signal so we don't get called again. */
    if (active == 0) {
        tail_consume_pending(ctx);
    }

    return 0;
}

/* cb_collect callback */
static int in_tail_collect_static(struct flb_input_instance *i_ins,
                                  struct flb_config *config, void *in_context)
{
    int ret;
    int active = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_tail_config *ctx = in_context;
    struct flb_tail_file *file;

    /* Do a data chunk collection for each file */
    mk_list_foreach_safe(head, tmp, &ctx->files_static) {
        file = mk_list_entry(head, struct flb_tail_file, _head);
        ret = flb_tail_file_chunk(file);
        switch (ret) {
        case FLB_TAIL_ERROR:
            /* Could not longer read the file */
            flb_tail_file_remove(file);
            break;
        case FLB_TAIL_OK:
        case FLB_TAIL_BUSY:
            active++;
            continue;
        case FLB_TAIL_WAIT:
            if (file->config->exit_on_eof) {
                flb_info("[in_tail] file=%s ended, stop", file->name);
                flb_engine_shutdown(config);
                exit(0);
            }
            /* Promote file to 'events' type handler */
            flb_debug("[in_tail] file=%s promote to TAIL_EVENT", file->name);
            ret = flb_tail_file_to_event(file);<commit_msg>In tail input, on 'Exit_On_EOF' case, call flb_engine_exit()<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 <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_error.h>
#include <fluent-bit/flb_utils.h>

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

static inline int consume_byte(int fd)
{
    int ret;
    uint64_t val;

    /* We need to consume the byte */
    ret = read(fd, &val, sizeof(val));
    if (ret <= 0) {
        flb_errno();
        return -1;
    }

    return 0;
}

/* cb_collect callback */
static int in_tail_collect_pending(struct flb_input_instance *i_ins,
                                   struct flb_config *config, void *in_context)
{
    int ret;
    int active = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_tail_config *ctx = in_context;
    struct flb_tail_file *file;
    struct stat st;

    /* Iterate promoted event files with pending bytes */
    mk_list_foreach_safe(head, tmp, &ctx->files_event) {
        file = mk_list_entry(head, struct flb_tail_file, _head);
        if (file->pending_bytes <= 0) {
            continue;
        }

        /* Gather current file size */
        ret = fstat(file->fd, &st);
        if (ret == -1) {
            flb_errno();
            flb_tail_file_remove(file);
            continue;
        }

        ret = flb_tail_file_chunk(file);
        switch (ret) {
        case FLB_TAIL_ERROR:
            /* Could not longer read the file */
            flb_tail_file_remove(file);
            break;
        case FLB_TAIL_OK:
        case FLB_TAIL_BUSY:
            /*
             * Adjust counter to verify if we need a further read(2) later.
             * For more details refer to tail_fs_inotify.c:96.
             */
            if (file->offset < st.st_size) {
                file->pending_bytes = (st.st_size - file->offset);
                active++;
            }
            else {
                file->pending_bytes = 0;
            }
            break;
        }
    }

    /* If no more active files, consume pending signal so we don't get called again. */
    if (active == 0) {
        tail_consume_pending(ctx);
    }

    return 0;
}

/* cb_collect callback */
static int in_tail_collect_static(struct flb_input_instance *i_ins,
                                  struct flb_config *config, void *in_context)
{
    int ret;
    int active = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_tail_config *ctx = in_context;
    struct flb_tail_file *file;

    /* Do a data chunk collection for each file */
    mk_list_foreach_safe(head, tmp, &ctx->files_static) {
        file = mk_list_entry(head, struct flb_tail_file, _head);
        ret = flb_tail_file_chunk(file);
        switch (ret) {
        case FLB_TAIL_ERROR:
            /* Could not longer read the file */
            flb_tail_file_remove(file);
            break;
        case FLB_TAIL_OK:
        case FLB_TAIL_BUSY:
            active++;
            continue;
        case FLB_TAIL_WAIT:
            if (file->config->exit_on_eof) {
                flb_info("[in_tail] file=%s ended, stop", file->name);
                flb_engine_exit(config);
            }
            /* Promote file to 'events' type handler */
            flb_debug("[in_tail] file=%s promote to TAIL_EVENT", file->name);
            ret = flb_tail_file_to_event(file);
{'fraction_non_alphanumeric': 0.1076341127922971, 'fraction_numerical': 0.0067629527739569, 'mean_word_length': 2.139618567830155, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5487416', 'n_tokens_mistral': 2862, 'n_tokens_neox': 2782, 'n_words': 769}