Skip to content

Commit 1365417

Browse files
cosmo0920edsiper
cosmo0920
authored and
edsiper
committed
out_chronicle: Add testing interface for formatter
Signed-off-by: Hiroshi Hatake <[email protected]>
1 parent c54abf0 commit 1365417

File tree

2 files changed

+125
-68
lines changed

2 files changed

+125
-68
lines changed

‎plugins/out_chronicle/chronicle.c

Lines changed: 84 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -373,7 +373,7 @@ static int check_chronicle_log_type(struct flb_chronicle *ctx, struct flb_config
373373
{
374374
int ret;
375375
size_t b_sent;
376-
flb_sds_t token;
376+
flb_sds_t token = NULL;
377377
struct flb_connection *u_conn;
378378
struct flb_http_client *c;
379379

@@ -493,21 +493,23 @@ static int cb_chronicle_init(struct flb_output_instance *ins,
493493
}
494494
flb_output_upstream_set(ctx->u, ins);
495495

496-
/* Get or renew the OAuth2 token */
497-
token = get_google_token(ctx);
496+
if (ins->test_mode == FLB_FALSE) {
497+
/* Get or renew the OAuth2 token */
498+
token = get_google_token(ctx);
498499

499-
if (!token) {
500-
flb_plg_warn(ctx->ins, "token retrieval failed");
501-
}
502-
else {
503-
flb_sds_destroy(token);
504-
}
500+
if (!token) {
501+
flb_plg_warn(ctx->ins, "token retrieval failed");
502+
}
503+
else {
504+
flb_sds_destroy(token);
505+
}
505506

506-
ret = check_chronicle_log_type(ctx, config);
507-
if (ret != 0) {
508-
flb_plg_error(ctx->ins, "Validate log_type failed. '%s' is not supported. ret = %d",
509-
ctx->log_type, ret);
510-
return -1;
507+
ret = check_chronicle_log_type(ctx, config);
508+
if (ret != 0) {
509+
flb_plg_error(ctx->ins, "Validate log_type failed. '%s' is not supported. ret = %d",
510+
ctx->log_type, ret);
511+
return -1;
512+
}
511513
}
512514

513515
return 0;
@@ -607,6 +609,8 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, uint64_t by
607609
if (log_key_found == FLB_FALSE) {
608610
flb_plg_error(ctx->ins, "Could not find log_key '%s' in record",
609611
ctx->log_key);
612+
flb_free(val_buf);
613+
return NULL;
610614
}
611615

612616
/* If nothing was read, destroy buffer */
@@ -684,6 +688,7 @@ static int chronicle_format(const void *data, size_t bytes,
684688
msgpack_packer mp_pck;
685689
flb_sds_t log_text = NULL;
686690
int log_text_size;
691+
char *json_str;
687692

688693
array_size = count_mp_with_threshold(last_offset, threshold, log_decoder, ctx);
689694

@@ -764,11 +769,29 @@ static int chronicle_format(const void *data, size_t bytes,
764769
msgpack_pack_str_body(&mp_pck, "log_text", 8);
765770
if (ctx->log_key != NULL) {
766771
log_text = flb_pack_msgpack_extract_log_key(ctx, bytes, log_event);
772+
if (log_text == NULL) {
773+
flb_plg_error(ctx->ins, "log_key extraction failed, skipping record");
774+
msgpack_sbuffer_destroy(&mp_sbuf);
775+
return -1;
776+
}
767777
log_text_size = flb_sds_len(log_text);
768778
}
769779
else {
770-
log_text = flb_msgpack_to_json_str(alloc_size, log_event.body);
771-
log_text_size = strlen(log_text);
780+
json_str = flb_msgpack_to_json_str(alloc_size, log_event.body);
781+
if (json_str == NULL) {
782+
flb_plg_error(ctx->ins, "Could not convert record to json string");
783+
msgpack_sbuffer_destroy(&mp_sbuf);
784+
return -1;
785+
}
786+
log_text = flb_sds_create(json_str);
787+
flb_free(json_str); /* free the original buffer */
788+
789+
if (log_text == NULL) {
790+
flb_plg_error(ctx->ins, "Could not create sds string for log");
791+
msgpack_sbuffer_destroy(&mp_sbuf);
792+
return -1;
793+
}
794+
log_text_size = flb_sds_len(log_text);
772795
}
773796

774797
if (log_text == NULL) {
@@ -778,12 +801,7 @@ static int chronicle_format(const void *data, size_t bytes,
778801
msgpack_pack_str(&mp_pck, log_text_size);
779802
msgpack_pack_str_body(&mp_pck, log_text, log_text_size);
780803

781-
if (ctx->log_key != NULL) {
782-
flb_sds_destroy(log_text);
783-
}
784-
else {
785-
flb_free(log_text);
786-
}
804+
flb_sds_destroy(log_text);
787805
/* timestamp */
788806
msgpack_pack_str(&mp_pck, 10);
789807
msgpack_pack_str_body(&mp_pck, "ts_rfc3339", 10);
@@ -825,6 +843,36 @@ static int chronicle_format(const void *data, size_t bytes,
825843
return 0;
826844
}
827845

846+
static int cb_chronicle_format_test(struct flb_config *config,
847+
struct flb_input_instance *ins,
848+
void *plugin_context,
849+
void *flush_ctx,
850+
int event_type,
851+
const char *tag, int tag_len,
852+
const void *data, size_t bytes,
853+
void **out_data, size_t *out_size)
854+
{
855+
struct flb_chronicle *ctx = plugin_context;
856+
struct flb_log_event_decoder log_decoder;
857+
int ret;
858+
size_t out_offset;
859+
860+
ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes);
861+
if (ret != FLB_EVENT_DECODER_SUCCESS) {
862+
flb_plg_error(ctx->ins, "log event decoder init error");
863+
return -1;
864+
}
865+
866+
ret = chronicle_format(data, bytes, tag, tag_len,
867+
(char **)out_data, out_size,
868+
0, bytes, &out_offset,
869+
&log_decoder, ctx);
870+
871+
flb_log_event_decoder_destroy(&log_decoder);
872+
return ret;
873+
}
874+
875+
828876
static void cb_chronicle_flush(struct flb_event_chunk *event_chunk,
829877
struct flb_output_flush *out_flush,
830878
struct flb_input_instance *i_ins,
@@ -859,13 +907,15 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk,
859907
FLB_OUTPUT_RETURN(FLB_RETRY);
860908
}
861909

862-
/* Get or renew Token */
863-
token = get_google_token(ctx);
910+
if (ctx->ins->test_mode == FLB_FALSE) {
911+
/* Get or renew Token */
912+
token = get_google_token(ctx);
864913

865-
if (!token) {
866-
flb_plg_error(ctx->ins, "cannot retrieve oauth2 token");
867-
flb_upstream_conn_release(u_conn);
868-
FLB_OUTPUT_RETURN(FLB_RETRY);
914+
if (!token) {
915+
flb_plg_error(ctx->ins, "cannot retrieve oauth2 token");
916+
flb_upstream_conn_release(u_conn);
917+
FLB_OUTPUT_RETURN(FLB_RETRY);
918+
}
869919
}
870920

871921
flb_plg_trace(ctx->ins, "msgpack payload size is %zu", event_chunk->size);
@@ -906,7 +956,8 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk,
906956
event_chunk->tag, flb_sds_len(event_chunk->tag),
907957
&payload_buf, &payload_size,
908958
offset, threshold, &out_offset,
909-
&log_decoder, ctx);
959+
&log_decoder,
960+
ctx);
910961
if (ret != 0) {
911962
flb_upstream_conn_release(u_conn);
912963
flb_sds_destroy(token);
@@ -1019,6 +1070,7 @@ static int cb_chronicle_exit(void *data, struct flb_config *config)
10191070
if (ctx->u) {
10201071
flb_upstream_destroy(ctx->u);
10211072
}
1073+
pthread_mutex_destroy(&ctx->token_mutex);
10221074

10231075
flb_chronicle_conf_destroy(ctx);
10241076
return 0;
@@ -1078,6 +1130,9 @@ struct flb_output_plugin out_chronicle_plugin = {
10781130
.cb_flush = cb_chronicle_flush,
10791131
.cb_exit = cb_chronicle_exit,
10801132
.config_map = config_map,
1133+
1134+
/* Test*/
1135+
.test_formatter.callback = cb_chronicle_format_test,
10811136
/* Plugin flags */
10821137
.flags = FLB_OUTPUT_NET | FLB_IO_TLS,
10831138
};

‎plugins/out_chronicle/chronicle_conf.c

Lines changed: 41 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -211,55 +211,57 @@ struct flb_chronicle *flb_chronicle_conf_create(struct flb_output_instance *ins,
211211
}
212212
}
213213

214-
if (ctx->credentials_file) {
215-
ret = flb_chronicle_read_credentials_file(ctx,
216-
ctx->credentials_file,
217-
ctx->oauth_credentials);
218-
if (ret != 0) {
219-
flb_chronicle_conf_destroy(ctx);
220-
return NULL;
221-
}
222-
}
223-
else if (!ctx->credentials_file) {
224-
/*
225-
* If no credentials file has been defined, do manual lookup of the
226-
* client email and the private key.
227-
*/
228-
229-
/* Service Account Email */
230-
tmp = flb_output_get_property("service_account_email", ins);
231-
if (tmp) {
232-
creds->client_email = flb_sds_create(tmp);
214+
if (ins->test_mode == FLB_FALSE) {
215+
if (ctx->credentials_file) {
216+
ret = flb_chronicle_read_credentials_file(ctx,
217+
ctx->credentials_file,
218+
ctx->oauth_credentials);
219+
if (ret != 0) {
220+
flb_chronicle_conf_destroy(ctx);
221+
return NULL;
222+
}
233223
}
234-
else {
235-
tmp = getenv("SERVICE_ACCOUNT_EMAIL");
224+
else if (!ctx->credentials_file) {
225+
/*
226+
* If no credentials file has been defined, do manual lookup of the
227+
* client email and the private key.
228+
*/
229+
230+
/* Service Account Email */
231+
tmp = flb_output_get_property("service_account_email", ins);
236232
if (tmp) {
237233
creds->client_email = flb_sds_create(tmp);
238234
}
239-
}
235+
else {
236+
tmp = getenv("SERVICE_ACCOUNT_EMAIL");
237+
if (tmp) {
238+
creds->client_email = flb_sds_create(tmp);
239+
}
240+
}
240241

241-
/* Service Account Secret */
242-
tmp = flb_output_get_property("service_account_secret", ins);
243-
if (tmp) {
244-
creds->private_key = flb_sds_create(tmp);
245-
}
246-
else {
247-
tmp = getenv("SERVICE_ACCOUNT_SECRET");
242+
/* Service Account Secret */
243+
tmp = flb_output_get_property("service_account_secret", ins);
248244
if (tmp) {
249245
creds->private_key = flb_sds_create(tmp);
250246
}
251-
}
247+
else {
248+
tmp = getenv("SERVICE_ACCOUNT_SECRET");
249+
if (tmp) {
250+
creds->private_key = flb_sds_create(tmp);
251+
}
252+
}
252253

253-
if (!creds->client_email) {
254-
flb_plg_error(ctx->ins, "service_account_email/client_email is not defined");
255-
flb_chronicle_conf_destroy(ctx);
256-
return NULL;
257-
}
254+
if (!creds->client_email) {
255+
flb_plg_error(ctx->ins, "service_account_email/client_email is not defined");
256+
flb_chronicle_conf_destroy(ctx);
257+
return NULL;
258+
}
258259

259-
if (!creds->private_key) {
260-
flb_plg_error(ctx->ins, "service_account_secret/private_key is not defined");
261-
flb_chronicle_conf_destroy(ctx);
262-
return NULL;
260+
if (!creds->private_key) {
261+
flb_plg_error(ctx->ins, "service_account_secret/private_key is not defined");
262+
flb_chronicle_conf_destroy(ctx);
263+
return NULL;
264+
}
263265
}
264266
}
265267

0 commit comments

Comments
 (0)