/src/fluent-bit/plugins/out_gcs/gcs_store.c

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

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2026 The Fluent Bit Authors
 *
 *  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 <fluent-bit/flb_info.h>
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_log.h>
#include <fluent-bit/flb_time.h>

#include "gcs.h"
#include "gcs_store.h"

static void normalize_stream_suffix(char *out, size_t out_size, const char *in)
{
    size_t i;
    char ch;

    if (!out || out_size == 0) {
        return;
    }

    if (!in) {
        out[0] = '\0';
        return;
    }

    for (i = 0; i < out_size - 1 && in[i] != '\0'; i++) {
        ch = in[i];
        if ((ch >= 'a' && ch <= 'z') ||
            (ch >= 'A' && ch <= 'Z') ||
            (ch >= '0' && ch <= '9') ||
            ch == '_' || ch == '-' || ch == '.') {
            out[i] = ch;
        }
        else {
            out[i] = '_';
        }
    }
    out[i] = '\0';
}

static flb_sds_t gen_store_filename(void)
{
    unsigned long hash;
    flb_sds_t hash_str;
    flb_sds_t tmp;
    struct flb_time tm;

    flb_time_get(&tm);

    hash = (unsigned long) tm.tm.tv_sec * tm.tm.tv_nsec;

    hash_str = flb_sds_create_size(64);
    if (!hash_str) {
        flb_errno();
        return NULL;
    }

    tmp = flb_sds_printf(&hash_str, "%lu", hash);
    if (!tmp) {
        flb_errno();
        flb_sds_destroy(hash_str);
        return NULL;
    }

    return tmp;
}

int gcs_store_init(struct flb_gcs *ctx)
{
    const char *instance_name;
    char stream_suffix[96];
    flb_sds_t stream_name;
    flb_sds_t tmp;

    stream_name = flb_sds_create_size(64);
    if (!stream_name) {
       flb_errno();
       return -1;
    }

    ctx->fs = flb_fstore_create(ctx->store_dir, FLB_FSTORE_FS);
    if (!ctx->fs) {
        flb_sds_destroy(stream_name);
        return -1;
    }

    instance_name = ctx->ins->alias ? ctx->ins->alias : ctx->ins->name;
    normalize_stream_suffix(stream_suffix, sizeof(stream_suffix), instance_name);

    tmp = flb_sds_printf(&stream_name, "gcs_upload_buffer_%s", stream_suffix);
    if (!tmp) {
        flb_sds_destroy(stream_name);
        flb_fstore_destroy(ctx->fs);
        ctx->fs = NULL;

        return -1;
    }
    stream_name = tmp;

    ctx->fs_stream_name = stream_name;
    ctx->fs_stream = flb_fstore_stream_create(ctx->fs, ctx->fs_stream_name);
    if (!ctx->fs_stream) {
        flb_sds_destroy(ctx->fs_stream_name);
        ctx->fs_stream_name = NULL;
        flb_fstore_destroy(ctx->fs);
        ctx->fs = NULL;

        return -1;
    }

    return 0;
}

int gcs_store_exit(struct flb_gcs *ctx)
{
    if (ctx->fs_stream_name) {
        flb_sds_destroy(ctx->fs_stream_name);
        ctx->fs_stream_name = NULL;
    }

    if (ctx->fs) {
        flb_fstore_destroy(ctx->fs);
        ctx->fs = NULL;
    }
    ctx->fs_stream = NULL;
    return 0;
}

int gcs_store_has_data(struct flb_gcs *ctx)
{
    if (!ctx || !ctx->fs_stream) {
        return FLB_FALSE;
    }
    return mk_list_size(&ctx->fs_stream->files) > 0 ? FLB_TRUE : FLB_FALSE;
}

struct gcs_file *gcs_store_file_get(struct flb_gcs *ctx, const char *tag, int tag_len)
{
    struct mk_list *head;
    struct flb_fstore_file *fsf;
    struct gcs_file *chunk;

    mk_list_foreach(head, &ctx->fs_stream->files) {
        fsf = mk_list_entry(head, struct flb_fstore_file, _head);
        if (fsf->meta_size != tag_len) {
            continue;
        }

        chunk = fsf->data;
        if (!chunk || chunk->locked == FLB_TRUE) {
            continue;
        }
        if (strncmp(fsf->meta_buf, tag, tag_len) == 0) {
            return chunk;
        }
    }
    return NULL;
}

int gcs_store_buffer_put(struct flb_gcs *ctx, struct gcs_file *chunk,
                         const char *tag, int tag_len, char *data, size_t bytes)
{
    int ret;
    flb_sds_t name;
    struct flb_fstore_file *fsf;
    size_t space_remaining;

    if (ctx->store_dir_limit_size > 0 &&
        (ctx->current_buffer_size > ctx->store_dir_limit_size ||
         bytes > ctx->store_dir_limit_size - ctx->current_buffer_size)) {
        flb_plg_error(ctx->ins,
                      "Buffer is full: current_buffer_size=%zu, new_data=%zu, "
                      "store_dir_limit_size=%zu bytes",
                      ctx->current_buffer_size, bytes, ctx->store_dir_limit_size);
        return -1;
    }

    if (!chunk) {
        if (ctx->store_chunk_limit > 0 &&
            mk_list_size(&ctx->fs_stream->files) >= ctx->store_chunk_limit) {
            flb_plg_error(ctx->ins,
                          "gcs local buffer chunk limit reached: limit=%d, dropping",
                          ctx->store_chunk_limit);
            return -1;
        }

        name = gen_store_filename();
        if (!name) {
            return -1;
        }
        fsf = flb_fstore_file_create(ctx->fs, ctx->fs_stream, name, bytes);
        flb_sds_destroy(name);
        if (!fsf) {
            return -1;
        }

        ret = flb_fstore_file_meta_set(ctx->fs, fsf, (char *) tag, tag_len);
        if (ret == -1) {
            flb_fstore_file_delete(ctx->fs, fsf);
            return -1;
        }
        chunk = flb_calloc(1, sizeof(struct gcs_file));
        if (!chunk) {
            flb_fstore_file_delete(ctx->fs, fsf);
            return -1;
        }
        chunk->fsf = fsf;
        chunk->create_time = time(NULL);
        fsf->data = chunk;
    }
    else {
        fsf = chunk->fsf;
    }

    ret = flb_fstore_file_append(fsf, data, bytes);
    if (ret != 0) {
        return -1;
    }

    chunk->size += bytes;
    ctx->current_buffer_size += bytes;

    if (ctx->store_dir_limit_size > 0) {
        space_remaining = ctx->store_dir_limit_size - ctx->current_buffer_size;
        if ((space_remaining * 20) < ctx->store_dir_limit_size) {
            flb_plg_warn(ctx->ins,
                         "Buffer is almost full: current_buffer_size=%zu, "
                         "store_dir_limit_size=%zu bytes",
                         ctx->current_buffer_size, ctx->store_dir_limit_size);
        }
    }
    return 0;
}

int gcs_store_file_read(struct flb_gcs *ctx, struct gcs_file *chunk,
                        char **out_buf, size_t *out_size)
{
    return flb_fstore_file_content_copy(ctx->fs, chunk->fsf, (void **) out_buf, out_size);
}

void gcs_store_file_lock(struct gcs_file *chunk)
{
    chunk->locked = FLB_TRUE;
}

void gcs_store_file_unlock(struct gcs_file *chunk)
{
    chunk->locked = FLB_FALSE;
}

int gcs_store_file_delete(struct flb_gcs *ctx, struct gcs_file *chunk)
{
    struct flb_fstore_file *fsf;

    if (!chunk) {
        return 0;
    }

    fsf = chunk->fsf;
    if (chunk->size <= ctx->current_buffer_size) {
        ctx->current_buffer_size -= chunk->size;
    }
    else {
        ctx->current_buffer_size = 0;
    }
    flb_free(chunk);
    flb_fstore_file_delete(ctx->fs, fsf);

    return 0;
}

Coverage Report

Created: 2026-08-09 07:14

Line	Count	Source
1		/* -- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -- */
2
3		/* Fluent Bit
4		* ==========
5		* Copyright (C) 2015-2026 The Fluent Bit Authors
6		*
7		* Licensed under the Apache License, Version 2.0 (the "License");
8		* you may not use this file except in compliance with the License.
9		* You may obtain a copy of the License at
10		*
11		* http://www.apache.org/licenses/LICENSE-2.0
12		*
13		* Unless required by applicable law or agreed to in writing, software
14		* distributed under the License is distributed on an "AS IS" BASIS,
15		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16		* See the License for the specific language governing permissions and
17		* limitations under the License.
18		*/
19
20		#include <fluent-bit/flb_info.h>
21		#include <fluent-bit/flb_utils.h>
22		#include <fluent-bit/flb_log.h>
23		#include <fluent-bit/flb_time.h>
24
25		#include "gcs.h"
26		#include "gcs_store.h"
27
28		static void normalize_stream_suffix(char out, size_t out_size, const char in)
29	0	{
30	0	size_t i;
31	0	char ch;
32
33	0	if (!out \|\| out_size == 0) {
34	0	return;
35	0	}
36
37	0	if (!in) {
38	0	out[0] = '\0';
39	0	return;
40	0	}
41
42	0	for (i = 0; i < out_size - 1 && in[i] != '\0'; i++) {
43	0	ch = in[i];
44	0	if ((ch >= 'a' && ch <= 'z') \|\|
45	0	(ch >= 'A' && ch <= 'Z') \|\|
46	0	(ch >= '0' && ch <= '9') \|\|
47	0	ch == '_' \|\| ch == '-' \|\| ch == '.') {
48	0	out[i] = ch;
49	0	}
50	0	else {
51	0	out[i] = '_';
52	0	}
53	0	}
54	0	out[i] = '\0';
55	0	}
56
57		static flb_sds_t gen_store_filename(void)
58	0	{
59	0	unsigned long hash;
60	0	flb_sds_t hash_str;
61	0	flb_sds_t tmp;
62	0	struct flb_time tm;
63
64	0	flb_time_get(&tm);
65
66	0	hash = (unsigned long) tm.tm.tv_sec * tm.tm.tv_nsec;
67
68	0	hash_str = flb_sds_create_size(64);
69	0	if (!hash_str) {
70	0	flb_errno();
71	0	return NULL;
72	0	}
73
74	0	tmp = flb_sds_printf(&hash_str, "%lu", hash);
75	0	if (!tmp) {
76	0	flb_errno();
77	0	flb_sds_destroy(hash_str);
78	0	return NULL;
79	0	}
80
81	0	return tmp;
82	0	}
83
84		int gcs_store_init(struct flb_gcs *ctx)
85	0	{
86	0	const char *instance_name;
87	0	char stream_suffix[96];
88	0	flb_sds_t stream_name;
89	0	flb_sds_t tmp;
90
91	0	stream_name = flb_sds_create_size(64);
92	0	if (!stream_name) {
93	0	flb_errno();
94	0	return -1;
95	0	}
96
97	0	ctx->fs = flb_fstore_create(ctx->store_dir, FLB_FSTORE_FS);
98	0	if (!ctx->fs) {
99	0	flb_sds_destroy(stream_name);
100	0	return -1;
101	0	}
102
103	0	instance_name = ctx->ins->alias ? ctx->ins->alias : ctx->ins->name;
104	0	normalize_stream_suffix(stream_suffix, sizeof(stream_suffix), instance_name);
105
106	0	tmp = flb_sds_printf(&stream_name, "gcs_upload_buffer_%s", stream_suffix);
107	0	if (!tmp) {
108	0	flb_sds_destroy(stream_name);
109	0	flb_fstore_destroy(ctx->fs);
110	0	ctx->fs = NULL;
111
112	0	return -1;
113	0	}
114	0	stream_name = tmp;
115
116	0	ctx->fs_stream_name = stream_name;
117	0	ctx->fs_stream = flb_fstore_stream_create(ctx->fs, ctx->fs_stream_name);
118	0	if (!ctx->fs_stream) {
119	0	flb_sds_destroy(ctx->fs_stream_name);
120	0	ctx->fs_stream_name = NULL;
121	0	flb_fstore_destroy(ctx->fs);
122	0	ctx->fs = NULL;
123
124	0	return -1;
125	0	}
126
127	0	return 0;
128	0	}
129
130		int gcs_store_exit(struct flb_gcs *ctx)
131	0	{
132	0	if (ctx->fs_stream_name) {
133	0	flb_sds_destroy(ctx->fs_stream_name);
134	0	ctx->fs_stream_name = NULL;
135	0	}
136
137	0	if (ctx->fs) {
138	0	flb_fstore_destroy(ctx->fs);
139	0	ctx->fs = NULL;
140	0	}
141	0	ctx->fs_stream = NULL;
142	0	return 0;
143	0	}
144
145		int gcs_store_has_data(struct flb_gcs *ctx)
146	0	{
147	0	if (!ctx \|\| !ctx->fs_stream) {
148	0	return FLB_FALSE;
149	0	}
150	0	return mk_list_size(&ctx->fs_stream->files) > 0 ? FLB_TRUE : FLB_FALSE;
151	0	}
152
153		struct gcs_file gcs_store_file_get(struct flb_gcs ctx, const char *tag, int tag_len)
154	0	{
155	0	struct mk_list *head;
156	0	struct flb_fstore_file *fsf;
157	0	struct gcs_file *chunk;
158
159	0	mk_list_foreach(head, &ctx->fs_stream->files) {
160	0	fsf = mk_list_entry(head, struct flb_fstore_file, _head);
161	0	if (fsf->meta_size != tag_len) {
162	0	continue;
163	0	}
164
165	0	chunk = fsf->data;
166	0	if (!chunk \|\| chunk->locked == FLB_TRUE) {
167	0	continue;
168	0	}
169	0	if (strncmp(fsf->meta_buf, tag, tag_len) == 0) {
170	0	return chunk;
171	0	}
172	0	}
173	0	return NULL;
174	0	}
175
176		int gcs_store_buffer_put(struct flb_gcs ctx, struct gcs_file chunk,
177		const char tag, int tag_len, char data, size_t bytes)
178	0	{
179	0	int ret;
180	0	flb_sds_t name;
181	0	struct flb_fstore_file *fsf;
182	0	size_t space_remaining;
183
184	0	if (ctx->store_dir_limit_size > 0 &&
185	0	(ctx->current_buffer_size > ctx->store_dir_limit_size \|\|
186	0	bytes > ctx->store_dir_limit_size - ctx->current_buffer_size)) {
187	0	flb_plg_error(ctx->ins,
188	0	"Buffer is full: current_buffer_size=%zu, new_data=%zu, "
189	0	"store_dir_limit_size=%zu bytes",
190	0	ctx->current_buffer_size, bytes, ctx->store_dir_limit_size);
191	0	return -1;
192	0	}
193
194	0	if (!chunk) {
195	0	if (ctx->store_chunk_limit > 0 &&
196	0	mk_list_size(&ctx->fs_stream->files) >= ctx->store_chunk_limit) {
197	0	flb_plg_error(ctx->ins,
198	0	"gcs local buffer chunk limit reached: limit=%d, dropping",
199	0	ctx->store_chunk_limit);
200	0	return -1;
201	0	}
202
203	0	name = gen_store_filename();
204	0	if (!name) {
205	0	return -1;
206	0	}
207	0	fsf = flb_fstore_file_create(ctx->fs, ctx->fs_stream, name, bytes);
208	0	flb_sds_destroy(name);
209	0	if (!fsf) {
210	0	return -1;
211	0	}
212
213	0	ret = flb_fstore_file_meta_set(ctx->fs, fsf, (char *) tag, tag_len);
214	0	if (ret == -1) {
215	0	flb_fstore_file_delete(ctx->fs, fsf);
216	0	return -1;
217	0	}
218	0	chunk = flb_calloc(1, sizeof(struct gcs_file));
219	0	if (!chunk) {
220	0	flb_fstore_file_delete(ctx->fs, fsf);
221	0	return -1;
222	0	}
223	0	chunk->fsf = fsf;
224	0	chunk->create_time = time(NULL);
225	0	fsf->data = chunk;
226	0	}
227	0	else {
228	0	fsf = chunk->fsf;
229	0	}
230
231	0	ret = flb_fstore_file_append(fsf, data, bytes);
232	0	if (ret != 0) {
233	0	return -1;
234	0	}
235
236	0	chunk->size += bytes;
237	0	ctx->current_buffer_size += bytes;
238
239	0	if (ctx->store_dir_limit_size > 0) {
240	0	space_remaining = ctx->store_dir_limit_size - ctx->current_buffer_size;
241	0	if ((space_remaining * 20) < ctx->store_dir_limit_size) {
242	0	flb_plg_warn(ctx->ins,
243	0	"Buffer is almost full: current_buffer_size=%zu, "
244	0	"store_dir_limit_size=%zu bytes",
245	0	ctx->current_buffer_size, ctx->store_dir_limit_size);
246	0	}
247	0	}
248	0	return 0;
249	0	}
250
251		int gcs_store_file_read(struct flb_gcs ctx, struct gcs_file chunk,
252		char *out_buf, size_t out_size)
253	0	{
254	0	return flb_fstore_file_content_copy(ctx->fs, chunk->fsf, (void **) out_buf, out_size);
255	0	}
256
257		void gcs_store_file_lock(struct gcs_file *chunk)
258	0	{
259	0	chunk->locked = FLB_TRUE;
260	0	}
261
262		void gcs_store_file_unlock(struct gcs_file *chunk)
263	0	{
264	0	chunk->locked = FLB_FALSE;
265	0	}
266
267		int gcs_store_file_delete(struct flb_gcs ctx, struct gcs_file chunk)
268	0	{
269	0	struct flb_fstore_file *fsf;
270
271	0	if (!chunk) {
272	0	return 0;
273	0	}
274
275	0	fsf = chunk->fsf;
276	0	if (chunk->size <= ctx->current_buffer_size) {
277	0	ctx->current_buffer_size -= chunk->size;
278	0	}
279	0	else {
280	0	ctx->current_buffer_size = 0;
281	0	}
282	0	flb_free(chunk);
283	0	flb_fstore_file_delete(ctx->fs, fsf);
284
285	0	return 0;
286	0	}