plugins/in_tail/tail_file.c
/* Append custom keys and report the number of records processed */
Append custom keys and report the number of records processed
Append custom keys and report the number of records processed
static int record_append_custom_keys(struct flb_tail_file *file,
                                     size_t processed_bytes,
                                     char *in_data, size_t in_size,
                                     char **out_data, size_t *out_size)
{
    int i;
    int ok = MSGPACK_UNPACK_SUCCESS;
    int len;
    int records = 0;
    size_t off = 0;
    size_t total;
    msgpack_unpacked result;
    msgpack_object time;
    msgpack_object map;
    msgpack_object k;
    msgpack_object v;
    msgpack_sbuffer mp_sbuf;
    msgpack_packer mp_pck;
    struct flb_mp_map_header mh;
    msgpack_sbuffer_init(&mp_sbuf);
    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
    msgpack_unpacked_init(&result);
    while ((msgpack_unpack_next(&result, in_data, in_size, &off) == ok)) {
        time = result.data.via.array.ptr[0];
        map = result.data.via.array.ptr[1];
        msgpack_pack_array(&mp_pck, 2);
        msgpack_pack_object(&mp_pck, time);
        flb_mp_map_header_init(&mh, &mp_pck);
        for (i = 0; i < map.via.map.size; i++) {
            k = map.via.map.ptr[i].key;
            v = map.via.map.ptr[i].val;
            flb_mp_map_header_append(&mh);
            msgpack_pack_object(&mp_pck, k);
            msgpack_pack_object(&mp_pck, v);
        }
        if (file->config->path_key) {
            len = flb_sds_len(file->config->path_key);
            flb_mp_map_header_append(&mh);
            msgpack_pack_str(&mp_pck, len);
            msgpack_pack_str_body(&mp_pck, file->config->path_key, len);
            msgpack_pack_str(&mp_pck, file->name_len);
            msgpack_pack_str_body(&mp_pck, file->name, file->name_len);
        }
        if (file->config->offset_key) {
            len = flb_sds_len(file->config->offset_key);
            flb_mp_map_header_append(&mh);
            msgpack_pack_str(&mp_pck, len);
            msgpack_pack_str_body(&mp_pck, file->config->offset_key, len);
            total = file->offset + processed_bytes;
            msgpack_pack_uint64(&mp_pck, total);
        }
        flb_mp_map_header_end(&mh);
        records++;
    }
    *out_data = mp_sbuf.data;
    *out_size = mp_sbuf.size;
    return records;
}
static int record_append_custom_keys ( struct flb_tail_file * file , size_t processed_bytes , char * in_data , size_t in_size , char * * out_data , size_t * out_size ) { int i ; int ok = MSGPACK_UNPACK_SUCCESS ; int len ; int records = 0 ; size_t off = 0 ; size_t total ; msgpack_unpacked result ; msgpack_object time ; msgpack_object map ; msgpack_object k ; msgpack_object v ; msgpack_sbuffer mp_sbuf ; msgpack_packer mp_pck ; struct flb_mp_map_header mh ; msgpack_sbuffer_init ( & mp_sbuf ) ; msgpack_packer_init ( & mp_pck , & mp_sbuf , msgpack_sbuffer_write ) ; msgpack_unpacked_init ( & result ) ; while ( ( msgpack_unpack_next ( & result , in_data , in_size , & off ) == ok ) ) { time = result . data . via . array . ptr [ 0 ] ; map = result . data . via . array . ptr [ 1 ] ; msgpack_pack_array ( & mp_pck , 2 ) ; msgpack_pack_object ( & mp_pck , time ) ; flb_mp_map_header_init ( & mh , & mp_pck ) ; for ( i = 0 ; i < map . via . map . size ; i ++ ) { k = map . via . map . ptr [ i ] . key ; v = map . via . map . ptr [ i ] . val ; flb_mp_map_header_append ( & mh ) ; msgpack_pack_object ( & mp_pck , k ) ; msgpack_pack_object ( & mp_pck , v ) ; } if ( file -> config -> path_key ) { len = flb_sds_len ( file -> config -> path_key ) ; flb_mp_map_header_append ( & mh ) ; msgpack_pack_str ( & mp_pck , len ) ; msgpack_pack_str_body ( & mp_pck , file -> config -> path_key , len ) ; msgpack_pack_str ( & mp_pck , file -> name_len ) ; msgpack_pack_str_body ( & mp_pck , file -> name , file -> name_len ) ; } if ( file -> config -> offset_key ) { len = flb_sds_len ( file -> config -> offset_key ) ; flb_mp_map_header_append ( & mh ) ; msgpack_pack_str ( & mp_pck , len ) ; msgpack_pack_str_body ( & mp_pck , file -> config -> offset_key , len ) ; total = file -> offset + processed_bytes ; msgpack_pack_uint64 ( & mp_pck , total ) ; } flb_mp_map_header_end ( & mh ) ; records ++ ; } * out_data = mp_sbuf . data ; * out_size = mp_sbuf . size ; return records ; }
Append custom keys and report the number of records processed
Append custom keys and report the number of records processed
/* init new buffers */ /* Some extra content will be added... */ /* pack map */ /* append previous map keys */ /* path_key */ /* key */ /* val */ /* offset_key */ /* key */ /* val */ /* finalize map */ /* counter */