diff --git a/plugins/out_chronicle/chronicle.c b/plugins/out_chronicle/chronicle.c index 10afd972d6f..d674315597d 100644 --- a/plugins/out_chronicle/chronicle.c +++ b/plugins/out_chronicle/chronicle.c @@ -193,10 +193,15 @@ static int chronicle_get_oauth2_token(struct flb_chronicle *ctx) time_t issued; time_t expires; char payload[1024]; + const char *scope = FLB_CHRONICLE_SCOPE; /* Clear any previous oauth2 payload content */ flb_oauth2_payload_clear(ctx->o); + if (ctx->api == FLB_CHRONICLE_API_CHRONICLE) { + scope = FLB_CHRONICLE_API_SCOPE; + } + /* JWT encode for oauth2 */ issued = time(NULL); expires = issued + FLB_CHRONICLE_TOKEN_REFRESH; @@ -204,7 +209,7 @@ static int chronicle_get_oauth2_token(struct flb_chronicle *ctx) snprintf(payload, sizeof(payload) - 1, "{\"iss\": \"%s\", \"scope\": \"%s\", " "\"aud\": \"%s\", \"exp\": %lu, \"iat\": %lu}", - ctx->oauth_credentials->client_email, FLB_CHRONICLE_SCOPE, + ctx->oauth_credentials->client_email, scope, FLB_CHRONICLE_AUTH_URL, expires, issued); @@ -504,11 +509,17 @@ static int cb_chronicle_init(struct flb_output_instance *ins, flb_sds_destroy(token); } - ret = check_chronicle_log_type(ctx, config); - if (ret != 0) { - flb_plg_error(ctx->ins, "Validate log_type failed. '%s' is not supported. ret = %d", - ctx->log_type, ret); - return -1; + /* + * The supported log types can only be validated with the legacy API. + * The Chronicle API rejects unknown log types on logs:import. + */ + if (ctx->api == FLB_CHRONICLE_API_LEGACY) { + ret = check_chronicle_log_type(ctx, config); + if (ret != 0) { + flb_plg_error(ctx->ins, "Validate log_type failed. '%s' is not supported. ret = %d", + ctx->log_type, ret); + return -1; + } } } @@ -906,6 +917,324 @@ static int chronicle_metadata_resolve(struct flb_chronicle *ctx, return 0; } +/* Format a timestamp as an RFC 3339 UTC string with nanosecond precision */ +static int chronicle_format_rfc3339(struct flb_time *tms, char *buf, size_t size) +{ + int len; + size_t s; + struct tm tm; + + if (gmtime_r(&tms->tm.tv_sec, &tm) == NULL) { + return -1; + } + + s = strftime(buf, size, FLB_STD_TIME_FMT, &tm); + if (s == 0) { + return -1; + } + + len = snprintf(buf + s, size - s, ".%09" PRIu64 "Z", (uint64_t) tms->tm.tv_nsec); + if (len < 0 || len >= size - s) { + return -1; + } + + return s + len; +} + +/* + * Pack root map (unstructured log) for the legacy Ingestion API: + * see: https://cloud.google.com/chronicle/docs/reference/ingestion-api#request_body_2 + * { + * "customer_id": "c8c65bfa-5f2c-42d4-9189-64bb7b939f2c", + * "log_type": "BIND_DNS", + * "entries": [ + * { + * "log_text": "26-Feb-2019 13:35:02.187 client 10.120.20.32#4238: query: altostrat.com IN A + (203.0.113.102)", + * "ts_epoch_microseconds": 1551188102187000 + * }, + * { + * "log_text": "26-Feb-2019 13:37:04.523 client 10.50.100.33#1116: query: examplepetstore.com IN A + (203.0.113.102)", + * "ts_rfc3339": "2019-26-02T13:37:04.523-08:00" + * }, + * { + * "log_text": "26-Feb-2019 13:39:01.115 client 10.1.2.3#3333: query: www.example.com IN A + (203.0.113.102)" + * }, + * ] + * } + */ +static int chronicle_pack_unstructured_logs(struct flb_chronicle *ctx, + msgpack_packer *mp_pck, + struct cfl_list *entry_list, + flb_sds_t namespace, + struct mk_list *labels, + int label_count) +{ + int len; + int map_size = 3; + char time_formatted[255]; + struct cfl_list *head; + struct mk_list *label_head; + struct chronicle_entry *entry; + struct chronicle_resolved_label *label; + + if (namespace) { + map_size++; + } + + if (label_count > 0) { + map_size++; + } + + msgpack_pack_map(mp_pck, map_size); + + msgpack_pack_str(mp_pck, 11); + msgpack_pack_str_body(mp_pck, "customer_id", 11); + + msgpack_pack_str(mp_pck, strlen(ctx->customer_id)); + msgpack_pack_str_body(mp_pck, ctx->customer_id, strlen(ctx->customer_id)); + + msgpack_pack_str(mp_pck, 8); + msgpack_pack_str_body(mp_pck, "log_type", 8); + + msgpack_pack_str(mp_pck, strlen(ctx->log_type)); + msgpack_pack_str_body(mp_pck, ctx->log_type, strlen(ctx->log_type)); + + if (namespace) { + msgpack_pack_str(mp_pck, 9); + msgpack_pack_str_body(mp_pck, "namespace", 9); + + msgpack_pack_str(mp_pck, flb_sds_len(namespace)); + msgpack_pack_str_body(mp_pck, namespace, flb_sds_len(namespace)); + } + + if (label_count > 0) { + msgpack_pack_str(mp_pck, 6); + msgpack_pack_str_body(mp_pck, "labels", 6); + + msgpack_pack_array(mp_pck, label_count); + + mk_list_foreach(label_head, labels) { + label = mk_list_entry(label_head, struct chronicle_resolved_label, _head); + + msgpack_pack_map(mp_pck, 2); + + msgpack_pack_str(mp_pck, 3); + msgpack_pack_str_body(mp_pck, "key", 3); + msgpack_pack_str(mp_pck, flb_sds_len(label->key)); + msgpack_pack_str_body(mp_pck, label->key, flb_sds_len(label->key)); + + msgpack_pack_str(mp_pck, 5); + msgpack_pack_str_body(mp_pck, "value", 5); + msgpack_pack_str(mp_pck, flb_sds_len(label->value)); + msgpack_pack_str_body(mp_pck, label->value, flb_sds_len(label->value)); + } + } + + msgpack_pack_str(mp_pck, 7); + msgpack_pack_str_body(mp_pck, "entries", 7); + + /* Append entries */ + msgpack_pack_array(mp_pck, cfl_list_size(entry_list)); + + /* Iterate the list of valid entries and pack them */ + cfl_list_foreach(head, entry_list) { + entry = cfl_list_entry(head, struct chronicle_entry, _head); + /* + * Pack entries + * + * { + * "log_text": {...}, + * "ts_rfc3339": "..." + * } + * + */ + msgpack_pack_map(mp_pck, 2); + + /* log_text */ + msgpack_pack_str(mp_pck, 8); + msgpack_pack_str_body(mp_pck, "log_text", 8); + msgpack_pack_str(mp_pck, entry->log_text_size); + msgpack_pack_str_body(mp_pck, entry->log_text, entry->log_text_size); + + /* timestamp */ + len = chronicle_format_rfc3339(&entry->timestamp, time_formatted, + sizeof(time_formatted)); + if (len < 0) { + flb_plg_error(ctx->ins, "could not format the record timestamp"); + return -1; + } + + msgpack_pack_str(mp_pck, 10); + msgpack_pack_str_body(mp_pck, "ts_rfc3339", 10); + + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, time_formatted, len); + } + + return 0; +} + +/* + * Pack root map for the Chronicle API logs:import method: + * see: https://cloud.google.com/chronicle/docs/reference/rest/v1/projects.locations.instances.logTypes.logs/import + * { + * "inlineSource": { + * "logs": [ + * { + * "data": "", + * "logEntryTime": "2019-02-26T13:35:02.187000000Z", + * "collectionTime": "2019-02-26T13:35:03.000000000Z", + * "environmentNamespace": "namespace", + * "labels": { + * "key": {"value": "value"} + * } + * } + * ] + * } + * } + */ +static int chronicle_pack_import_logs(struct flb_chronicle *ctx, + msgpack_packer *mp_pck, + struct cfl_list *entry_list, + flb_sds_t namespace, + struct mk_list *labels, + int label_count) +{ + int ret; + int len; + int map_size; + size_t olen; + size_t data_size; + char *data; + char entry_time[64]; + char collection_time[64]; + uint64_t entry_ns; + struct flb_time now; + struct flb_time collected; + struct cfl_list *head; + struct mk_list *label_head; + struct chronicle_entry *entry; + struct chronicle_resolved_label *label; + + flb_time_get(&now); + + msgpack_pack_map(mp_pck, 1); + msgpack_pack_str(mp_pck, 12); + msgpack_pack_str_body(mp_pck, "inlineSource", 12); + + msgpack_pack_map(mp_pck, 1); + msgpack_pack_str(mp_pck, 4); + msgpack_pack_str_body(mp_pck, "logs", 4); + + msgpack_pack_array(mp_pck, cfl_list_size(entry_list)); + + cfl_list_foreach(head, entry_list) { + entry = cfl_list_entry(head, struct chronicle_entry, _head); + + map_size = 3; + if (namespace) { + map_size++; + } + if (label_count > 0) { + map_size++; + } + + msgpack_pack_map(mp_pck, map_size); + + /* data: the raw log line must be base64 encoded */ + data_size = ((entry->log_text_size + 2) / 3) * 4 + 1; + data = flb_malloc(data_size); + if (!data) { + flb_errno(); + return -1; + } + + ret = flb_base64_encode((unsigned char *) data, data_size, &olen, + (unsigned char *) entry->log_text, + entry->log_text_size); + if (ret != 0) { + flb_plg_error(ctx->ins, "could not base64 encode the log entry"); + flb_free(data); + return -1; + } + + msgpack_pack_str(mp_pck, 4); + msgpack_pack_str_body(mp_pck, "data", 4); + msgpack_pack_str(mp_pck, olen); + msgpack_pack_str_body(mp_pck, data, olen); + flb_free(data); + + /* logEntryTime */ + len = chronicle_format_rfc3339(&entry->timestamp, entry_time, + sizeof(entry_time)); + if (len < 0) { + flb_plg_error(ctx->ins, "could not format the record timestamp"); + return -1; + } + + msgpack_pack_str(mp_pck, 12); + msgpack_pack_str_body(mp_pck, "logEntryTime", 12); + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, entry_time, len); + + /* + * collectionTime: it must be later than the log entry time, so records + * that are not in the past are collected one millisecond after them. + */ + collected = now; + entry_ns = flb_time_to_nanosec(&entry->timestamp); + if (entry_ns >= flb_time_to_nanosec(&now)) { + ret = flb_time_from_uint64(&collected, + entry_ns + FLB_CHRONICLE_API_COLLECTION_DELAY_NS); + if (ret != 0) { + flb_plg_error(ctx->ins, "could not compute the collection timestamp"); + return -1; + } + } + + len = chronicle_format_rfc3339(&collected, collection_time, + sizeof(collection_time)); + if (len < 0) { + flb_plg_error(ctx->ins, "could not format the collection timestamp"); + return -1; + } + + msgpack_pack_str(mp_pck, 14); + msgpack_pack_str_body(mp_pck, "collectionTime", 14); + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, collection_time, len); + + if (namespace) { + msgpack_pack_str(mp_pck, 20); + msgpack_pack_str_body(mp_pck, "environmentNamespace", 20); + msgpack_pack_str(mp_pck, flb_sds_len(namespace)); + msgpack_pack_str_body(mp_pck, namespace, flb_sds_len(namespace)); + } + + if (label_count > 0) { + msgpack_pack_str(mp_pck, 6); + msgpack_pack_str_body(mp_pck, "labels", 6); + + msgpack_pack_map(mp_pck, label_count); + + mk_list_foreach(label_head, labels) { + label = mk_list_entry(label_head, struct chronicle_resolved_label, _head); + + msgpack_pack_str(mp_pck, flb_sds_len(label->key)); + msgpack_pack_str_body(mp_pck, label->key, flb_sds_len(label->key)); + + msgpack_pack_map(mp_pck, 1); + msgpack_pack_str(mp_pck, 5); + msgpack_pack_str_body(mp_pck, "value", 5); + msgpack_pack_str(mp_pck, flb_sds_len(label->value)); + msgpack_pack_str_body(mp_pck, label->value, flb_sds_len(label->value)); + } + } + } + + return 0; +} + static int chronicle_format(const void *data, size_t bytes, const char *tag, size_t tag_len, char **out_data, size_t *out_size, @@ -915,16 +1244,11 @@ static int chronicle_format(const void *data, size_t bytes, struct flb_chronicle *ctx, struct flb_config *config) { - int len; int ret; int array_size = 0; size_t off = 0; size_t last_off = last_offset; size_t alloc_size = 0; - size_t s; - char time_formatted[255]; - /* Parameters for Timestamp */ - struct tm tm; flb_sds_t out_buf; struct flb_log_event log_event; msgpack_sbuffer mp_sbuf; @@ -934,16 +1258,12 @@ static int chronicle_format(const void *data, size_t bytes, char *json_str; struct cfl_list entry_list; struct chronicle_entry *entry; - struct cfl_list *head; struct mk_list resolved_labels; struct mk_list record_labels; - struct mk_list *label_head; - struct chronicle_resolved_label *label; flb_sds_t namespace = NULL; flb_sds_t record_namespace = NULL; int label_count = 0; int record_label_count = 0; - int map_size = 3; int metadata_resolved = FLB_FALSE; size_t record_start; @@ -1095,130 +1415,30 @@ static int chronicle_format(const void *data, size_t bytes, msgpack_sbuffer_init(&mp_sbuf); msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); - /* - * Pack root map (unstructured log): - * see: https://cloud.google.com/chronicle/docs/reference/ingestion-api#request_body_2 - * { - * "customer_id": "c8c65bfa-5f2c-42d4-9189-64bb7b939f2c", - * "log_type": "BIND_DNS", - * "entries": [ - * { - * "log_text": "26-Feb-2019 13:35:02.187 client 10.120.20.32#4238: query: altostrat.com IN A + (203.0.113.102)", - * "ts_epoch_microseconds": 1551188102187000 - * }, - * { - * "log_text": "26-Feb-2019 13:37:04.523 client 10.50.100.33#1116: query: examplepetstore.com IN A + (203.0.113.102)", - * "ts_rfc3339": "2019-26-02T13:37:04.523-08:00" - * }, - * { - * "log_text": "26-Feb-2019 13:39:01.115 client 10.1.2.3#3333: query: www.example.com IN A + (203.0.113.102)" - * }, - * ] - * } - */ - if (namespace) { - map_size++; - } - - if (label_count > 0) { - map_size++; - } - - msgpack_pack_map(&mp_pck, map_size); - - msgpack_pack_str(&mp_pck, 11); - msgpack_pack_str_body(&mp_pck, "customer_id", 11); - - msgpack_pack_str(&mp_pck, strlen(ctx->customer_id)); - msgpack_pack_str_body(&mp_pck, ctx->customer_id, strlen(ctx->customer_id)); - - msgpack_pack_str(&mp_pck, 8); - msgpack_pack_str_body(&mp_pck, "log_type", 8); - - msgpack_pack_str(&mp_pck, strlen(ctx->log_type)); - msgpack_pack_str_body(&mp_pck, ctx->log_type, strlen(ctx->log_type)); - - if (namespace) { - msgpack_pack_str(&mp_pck, 9); - msgpack_pack_str_body(&mp_pck, "namespace", 9); - - msgpack_pack_str(&mp_pck, flb_sds_len(namespace)); - msgpack_pack_str_body(&mp_pck, namespace, flb_sds_len(namespace)); + if (ctx->api == FLB_CHRONICLE_API_CHRONICLE) { + ret = chronicle_pack_import_logs(ctx, &mp_pck, &entry_list, + namespace, &resolved_labels, + label_count); } - - if (label_count > 0) { - msgpack_pack_str(&mp_pck, 6); - msgpack_pack_str_body(&mp_pck, "labels", 6); - - msgpack_pack_array(&mp_pck, label_count); - - mk_list_foreach(label_head, &resolved_labels) { - label = mk_list_entry(label_head, struct chronicle_resolved_label, _head); - - msgpack_pack_map(&mp_pck, 2); - - msgpack_pack_str(&mp_pck, 3); - msgpack_pack_str_body(&mp_pck, "key", 3); - msgpack_pack_str(&mp_pck, flb_sds_len(label->key)); - msgpack_pack_str_body(&mp_pck, label->key, flb_sds_len(label->key)); - - msgpack_pack_str(&mp_pck, 5); - msgpack_pack_str_body(&mp_pck, "value", 5); - msgpack_pack_str(&mp_pck, flb_sds_len(label->value)); - msgpack_pack_str_body(&mp_pck, label->value, flb_sds_len(label->value)); - } + else { + ret = chronicle_pack_unstructured_logs(ctx, &mp_pck, &entry_list, + namespace, &resolved_labels, + label_count); } - msgpack_pack_str(&mp_pck, 7); - msgpack_pack_str_body(&mp_pck, "entries", 7); - - /* Append entries */ - msgpack_pack_array(&mp_pck, cfl_list_size(&entry_list)); - - /* Iterate the list of valid entries and pack them */ - cfl_list_foreach(head, &entry_list) { - entry = cfl_list_entry(head, struct chronicle_entry, _head); - /* - * Pack entries - * - * { - * "log_text": {...}, - * "ts_rfc3339": "..." - * } - * - */ - msgpack_pack_map(&mp_pck, 2); - - /* log_text */ - msgpack_pack_str(&mp_pck, 8); - msgpack_pack_str_body(&mp_pck, "log_text", 8); - msgpack_pack_str(&mp_pck, entry->log_text_size); - msgpack_pack_str_body(&mp_pck, entry->log_text, entry->log_text_size); - - /* timestamp */ - msgpack_pack_str(&mp_pck, 10); - msgpack_pack_str_body(&mp_pck, "ts_rfc3339", 10); - - gmtime_r(&entry->timestamp.tm.tv_sec, &tm); - s = strftime(time_formatted, sizeof(time_formatted) - 1, - FLB_STD_TIME_FMT, &tm); - len = snprintf(time_formatted + s, sizeof(time_formatted) - 1 - s, - ".%03" PRIu64 "Z", - (uint64_t) entry->timestamp.tm.tv_nsec); - s += len; - - msgpack_pack_str(&mp_pck, s); - msgpack_pack_str_body(&mp_pck, time_formatted, s); + chronicle_resolved_labels_destroy(&resolved_labels); + flb_sds_destroy(namespace); + chronicle_entries_destroy(&entry_list); + if (ret != 0) { + msgpack_sbuffer_destroy(&mp_sbuf); + return -1; } /* Convert from msgpack to JSON */ out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size, config->json_escape_unicode); msgpack_sbuffer_destroy(&mp_sbuf); - chronicle_resolved_labels_destroy(&resolved_labels); - flb_sds_destroy(namespace); - chronicle_entries_destroy(&entry_list); if (!out_buf) { flb_plg_error(ctx->ins, "error formatting JSON payload"); @@ -1295,6 +1515,44 @@ static int cb_chronicle_format_test(struct flb_config *config, } +/* + * Map the HTTP status of an ingestion request to a flush result. With the + * Chronicle API, client errors such as an unknown log type or missing + * permissions fail again on every retry, so they are not retried, except for + * authentication, timeout and rate limit errors. + */ +static int chronicle_response_result(struct flb_chronicle *ctx, int status) +{ + if (ctx->api == FLB_CHRONICLE_API_LEGACY) { + if (status == 200) { + return FLB_OK; + } + return FLB_RETRY; + } + + if (status >= 200 && status < 300) { + return FLB_OK; + } + + if (status >= 400 && status < 500 && + status != 401 && status != 408 && status != 429) { + return FLB_ERROR; + } + + return FLB_RETRY; +} + +static int cb_chronicle_response_test(struct flb_config *config, + void *plugin_context, + int status, + const void *data, size_t bytes, + void **out_data, size_t *out_size) +{ + struct flb_chronicle *ctx = plugin_context; + + return chronicle_response_result(ctx, status); +} + static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, struct flb_output_flush *out_flush, struct flb_input_instance *i_ins, @@ -1304,7 +1562,8 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, (void) i_ins; (void) config; int ret; - int ret_code = FLB_RETRY; + int result; + int ret_code = FLB_OK; size_t b_sent; flb_sds_t token; flb_sds_t payload_buf; @@ -1313,13 +1572,13 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, struct flb_connection *u_conn; struct flb_http_client *c; struct flb_log_event_decoder log_decoder; - size_t threshold = 0.8 * 1024 * 1024; + size_t threshold = 0.8 * ctx->max_payload_size; size_t offset = 0; size_t out_offset = 0; int need_loop = FLB_TRUE; const int retry_limit = 8; int retries = 0; - const size_t one_mebibyte = 1024 * 1024; + const size_t max_payload_size = ctx->max_payload_size; flb_plg_trace(ctx->ins, "flushing bytes %zu", event_chunk->size); @@ -1370,7 +1629,7 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, * For 8th attempt, it won't happen. Just give up for * formating though. :) */ - threshold = (retry_limit - retries)/10.0 * one_mebibyte; + threshold = (retry_limit - retries)/10.0 * max_payload_size; } /* Reformat msgpack to chronicle JSON payload */ @@ -1391,7 +1650,7 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, flb_plg_debug(ctx->ins, "the last offset of msgpack decoder is %zu", out_offset); - if (payload_size >= one_mebibyte) { + if (payload_size >= max_payload_size) { retries++; if (retries >= retry_limit) { flb_plg_error(ctx->ins, "Retry limit is exeeced for chronicle_format"); @@ -1406,7 +1665,7 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, flb_plg_debug(ctx->ins, "HTTP request body is exeeded to %zd bytes. actual: %zu. left attempt(s): %d", - one_mebibyte, payload_size, retry_limit - retries); + max_payload_size, payload_size, retry_limit - retries); flb_sds_destroy(payload_buf); goto retry; @@ -1441,20 +1700,21 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, /* validate response */ if (ret != 0) { flb_plg_warn(ctx->ins, "http_do=%i", ret); - ret_code = FLB_RETRY; + result = FLB_RETRY; } else { /* The request was issued successfully, validate the 'error' field */ flb_plg_debug(ctx->ins, "HTTP Status=%i", c->resp.status); - if (c->resp.status == 200) { - ret_code = FLB_OK; - } - else { + result = chronicle_response_result(ctx, c->resp.status); + if (result != FLB_OK) { if (c->resp.payload && c->resp.payload_size > 0) { /* we got an error */ flb_plg_warn(ctx->ins, "response\n%s", c->resp.payload); } - ret_code = FLB_RETRY; + } + if (result == FLB_ERROR) { + flb_plg_error(ctx->ins, "request rejected with HTTP status %i, " + "the records will not be retried", c->resp.status); } } @@ -1466,6 +1726,21 @@ static void cb_chronicle_flush(struct flb_event_chunk *event_chunk, flb_sds_destroy(payload_buf); flb_http_client_destroy(c); + /* + * A chunk can be sent in several requests, so a later success must + * not hide a failed request. A retry sends the whole chunk again, + * so stop at the first request to retry. A rejected request is not + * retried, so keep sending the rest of the chunk and report the + * error once done. + */ + if (result == FLB_RETRY) { + ret_code = FLB_RETRY; + break; + } + if (result == FLB_ERROR) { + ret_code = FLB_ERROR; + } + /* The next loop uses the returned offset */ offset = out_offset; } @@ -1534,7 +1809,15 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_STR, "region", (char *)NULL, 0, FLB_TRUE, offsetof(struct flb_chronicle, region), - "Set the region" + "Set the region. With the legacy API: US, EU, UK or ASIA. With the " + "Chronicle API: the location of the Google SecOps instance, e.g: us, " + "europe or europe-west2 (default: us)" + }, + { + FLB_CONFIG_MAP_STR, "api", "legacy", + 0, FLB_TRUE, offsetof(struct flb_chronicle, api_str), + "Set the ingestion API: 'legacy' for the deprecated Ingestion API or " + "'chronicle' for the Chronicle API (logs:import)" }, { FLB_CONFIG_MAP_STR, "log_key", NULL, @@ -1570,6 +1853,7 @@ struct flb_output_plugin out_chronicle_plugin = { /* Test*/ .test_formatter.callback = cb_chronicle_format_test, + .test_response.callback = cb_chronicle_response_test, /* Plugin flags */ .flags = FLB_OUTPUT_NET | FLB_IO_TLS, }; diff --git a/plugins/out_chronicle/chronicle.h b/plugins/out_chronicle/chronicle.h index 7e6bcbc346f..ece0764e77e 100644 --- a/plugins/out_chronicle/chronicle.h +++ b/plugins/out_chronicle/chronicle.h @@ -45,6 +45,26 @@ #define FLB_CHRONICLE_URL_BASE_UK "https://europe-west2-malachiteingestion-pa.googleapis.com" #define FLB_CHRONICLE_URL_BASE_ASIA "https://asia-southeast1-malachiteingestion-pa.googleapis.com" +/* Maximum size of a request body for the legacy Ingestion API */ +#define FLB_CHRONICLE_MAX_PAYLOAD_SIZE (1024 * 1024) + +/* Ingestion APIs */ +#define FLB_CHRONICLE_API_LEGACY 0 +#define FLB_CHRONICLE_API_CHRONICLE 1 + +/* Chronicle API (logs:import) */ +#define FLB_CHRONICLE_API_SCOPE "https://www.googleapis.com/auth/cloud-platform" +#define FLB_CHRONICLE_API_DEFAULT_LOCATION "us" +#define FLB_CHRONICLE_API_URL_BASE "https://%s-chronicle.googleapis.com" +#define FLB_CHRONICLE_API_IMPORT_ENDPOINT \ + "/v1/projects/%s/locations/%s/instances/%s/logTypes/%s/logs:import" + +/* Maximum size of a request body for the Chronicle API (4 MB) */ +#define FLB_CHRONICLE_API_MAX_PAYLOAD_SIZE (4 * 1000 * 1000) + +/* Delay added to the collection time of records that are not in the past */ +#define FLB_CHRONICLE_API_COLLECTION_DELAY_NS 1000000 + struct flb_chronicle_oauth_credentials { /* parsed credentials file */ flb_sds_t type; @@ -75,10 +95,16 @@ struct flb_chronicle { flb_sds_t customer_id; flb_sds_t log_type; + /* ingestion API: legacy Ingestion API or Chronicle API */ + flb_sds_t api_str; + int api; + size_t max_payload_size; + flb_sds_t uri; flb_sds_t health_uri; flb_sds_t endpoint; flb_sds_t region; + flb_sds_t location; flb_sds_t log_key; flb_sds_t namespace; flb_sds_t namespace_key; diff --git a/plugins/out_chronicle/chronicle_conf.c b/plugins/out_chronicle/chronicle_conf.c index b0c3f222a58..a849daeed18 100644 --- a/plugins/out_chronicle/chronicle_conf.c +++ b/plugins/out_chronicle/chronicle_conf.c @@ -25,6 +25,7 @@ #include #include +#include #include #include @@ -108,6 +109,23 @@ static int chronicle_label_add(struct flb_chronicle *ctx, return 0; } +static int chronicle_label_exists(struct flb_chronicle *ctx, flb_sds_t key) +{ + struct mk_list *head; + struct flb_chronicle_label *label; + + mk_list_foreach(head, &ctx->labels) { + label = mk_list_entry(head, struct flb_chronicle_label, _head); + + if (flb_sds_len(label->key) == flb_sds_len(key) && + strncmp(label->key, key, flb_sds_len(key)) == 0) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + static int chronicle_configure_labels(struct flb_chronicle *ctx) { int ret; @@ -128,6 +146,14 @@ static int chronicle_configure_labels(struct flb_chronicle *ctx) val = mk_list_entry_last(mv->val.list, struct flb_slist_entry, _head); + /* The Chronicle API sends labels as a map keyed by label name */ + if (ctx->api == FLB_CHRONICLE_API_CHRONICLE && + chronicle_label_exists(ctx, key->str) == FLB_TRUE) { + flb_plg_error(ctx->ins, "duplicate label '%s' is not supported " + "by the Chronicle API", key->str); + return -1; + } + ret = chronicle_label_add(ctx, key->str, flb_sds_len(key->str), val->str, flb_sds_len(val->str)); @@ -270,6 +296,100 @@ static int flb_chronicle_read_credentials_file(struct flb_chronicle *ctx, return 0; } +/* Check that a value can be used as-is as a segment of the request path */ +static int chronicle_api_check_path_segment(struct flb_chronicle *ctx, + const char *name, flb_sds_t value) +{ + size_t i; + char c; + + if (flb_sds_len(value) == 0) { + flb_plg_error(ctx->ins, "property '%s' cannot be empty", name); + return -1; + } + + for (i = 0; i < flb_sds_len(value); i++) { + c = value[i]; + if (!isalnum((unsigned char) c) && + c != '-' && c != '.' && c != '_' && c != '~' && c != ':') { + flb_plg_error(ctx->ins, "property '%s' has an invalid character: '%s'", + name, value); + return -1; + } + } + + return 0; +} + +/* Compose the base URI and the logs:import endpoint for the Chronicle API */ +static int chronicle_api_configure(struct flb_chronicle *ctx) +{ + size_t i; + char c; + flb_sds_t tmp; + + /* The region is the location of the Google SecOps instance, e.g: 'us' */ + if (ctx->region == NULL) { + ctx->location = flb_sds_create(FLB_CHRONICLE_API_DEFAULT_LOCATION); + } + else { + ctx->location = flb_sds_create(ctx->region); + } + if (!ctx->location) { + flb_errno(); + return -1; + } + + /* The location is also part of the hostname */ + if (flb_sds_len(ctx->location) == 0) { + flb_plg_error(ctx->ins, "property 'region' cannot be empty"); + return -1; + } + + for (i = 0; i < flb_sds_len(ctx->location); i++) { + c = tolower((unsigned char) ctx->location[i]); + if (!isalnum((unsigned char) c) && c != '-') { + flb_plg_error(ctx->ins, "invalid region '%s' for the Chronicle API", + ctx->region); + return -1; + } + ctx->location[i] = c; + } + + if (chronicle_api_check_path_segment(ctx, "project_id", ctx->project_id) != 0 || + chronicle_api_check_path_segment(ctx, "customer_id", ctx->customer_id) != 0 || + chronicle_api_check_path_segment(ctx, "log_type", ctx->log_type) != 0) { + return -1; + } + + ctx->uri = flb_sds_create_size(64); + if (!ctx->uri) { + flb_errno(); + return -1; + } + + tmp = flb_sds_printf(&ctx->uri, FLB_CHRONICLE_API_URL_BASE, ctx->location); + if (!tmp) { + return -1; + } + + ctx->endpoint = flb_sds_create_size(256); + if (!ctx->endpoint) { + flb_errno(); + return -1; + } + + tmp = flb_sds_printf(&ctx->endpoint, FLB_CHRONICLE_API_IMPORT_ENDPOINT, + ctx->project_id, ctx->location, + ctx->customer_id, ctx->log_type); + if (!tmp) { + return -1; + } + + ctx->max_payload_size = FLB_CHRONICLE_API_MAX_PAYLOAD_SIZE; + + return 0; +} struct flb_chronicle *flb_chronicle_conf_create(struct flb_output_instance *ins, struct flb_config *config) @@ -296,6 +416,20 @@ struct flb_chronicle *flb_chronicle_conf_create(struct flb_output_instance *ins, return NULL; } + /* config: 'api' */ + if (strcasecmp(ctx->api_str, "legacy") == 0) { + ctx->api = FLB_CHRONICLE_API_LEGACY; + } + else if (strcasecmp(ctx->api_str, "chronicle") == 0) { + ctx->api = FLB_CHRONICLE_API_CHRONICLE; + } + else { + flb_plg_error(ins, "invalid api '%s', expected 'legacy' or 'chronicle'", + ctx->api_str); + flb_free(ctx); + return NULL; + } + /* Lookup credentials file */ creds = flb_calloc(1, sizeof(struct flb_chronicle_oauth_credentials)); if (!creds) { @@ -449,6 +583,22 @@ struct flb_chronicle *flb_chronicle_conf_create(struct flb_output_instance *ins, } } + if (ctx->api == FLB_CHRONICLE_API_CHRONICLE) { + ret = chronicle_api_configure(ctx); + if (ret != 0) { + flb_chronicle_conf_destroy(ctx); + return NULL; + } + + flb_plg_info(ctx->ins, "api='chronicle' project='%s' customer_id='%s' " + "location='%s'", + ctx->project_id, ctx->customer_id, ctx->location); + + return ctx; + } + + ctx->max_payload_size = FLB_CHRONICLE_MAX_PAYLOAD_SIZE; + /* Create the target endpoint URI */ ctx->endpoint = flb_sds_create_size(sizeof(FLB_CHRONICLE_UNSTRUCTURED_ENDPOINT)); if (!ctx->endpoint) { @@ -536,6 +686,7 @@ int flb_chronicle_conf_destroy(struct flb_chronicle *ctx) flb_sds_destroy(ctx->endpoint); flb_sds_destroy(ctx->uri); + flb_sds_destroy(ctx->location); if (ctx->o) { flb_oauth2_destroy(ctx->o); diff --git a/tests/runtime/out_chronicle.c b/tests/runtime/out_chronicle.c index ddfd54412a0..8da4dc3e037 100644 --- a/tests/runtime/out_chronicle.c +++ b/tests/runtime/out_chronicle.c @@ -119,6 +119,29 @@ static void cb_check_format_no_log_key(void *ctx, int ffd, flb_sds_destroy(res_data); } +static void cb_check_format_subsecond_timestamp(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + char *out_json = res_data; + char *p; + + if (res_ret != 0 || out_json == NULL) { + set_callback_error("formatter returned an error or no output"); + flb_sds_destroy(res_data); + return; + } + + /* The fraction of 1700000000.0625 must keep its leading zero */ + p = strstr(out_json, "\"ts_rfc3339\":\"2023-11-14T22:13:20.062500000Z\""); + if (p == NULL) { + set_callback_error("expected sub-second timestamp was not found"); + } + + increment_output_invoked(); + flb_sds_destroy(res_data); +} + static void cb_check_format_with_log_key(void *ctx, int ffd, int res_ret, void *res_data, size_t res_size, void *data) @@ -366,6 +389,148 @@ static void cb_check_format_split_on_metadata_change(void *ctx, int ffd, flb_sds_destroy(res_data); } +static void cb_check_format_chronicle_api(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + char *out_json = res_data; + char *p; + + if (res_ret != 0 || out_json == NULL) { + set_callback_error("formatter returned an error or no output"); + flb_sds_destroy(res_data); + return; + } + + p = strstr(out_json, "{\"inlineSource\":{\"logs\":[{"); + if (p != out_json) { + set_callback_error("expected inlineSource.logs root was not found"); + } + + /* base64 of {"message":"hello world"} */ + p = strstr(out_json, "\"data\":\"eyJtZXNzYWdlIjoiaGVsbG8gd29ybGQifQ==\""); + if (p == NULL) { + set_callback_error("expected base64 encoded data was not found"); + } + + p = strstr(out_json, "\"logEntryTime\":\"2023-11-14T22:13:20.062500000Z\""); + if (p == NULL) { + set_callback_error("expected logEntryTime was not found"); + } + + p = strstr(out_json, "\"collectionTime\":\""); + if (p == NULL) { + set_callback_error("expected collectionTime was not found"); + } + + p = strstr(out_json, "customer_id"); + if (p != NULL) { + set_callback_error("unexpected legacy customer_id was found"); + } + + p = strstr(out_json, "log_text"); + if (p != NULL) { + set_callback_error("unexpected legacy log_text was found"); + } + + p = strstr(out_json, "environmentNamespace"); + if (p != NULL) { + set_callback_error("unexpected environmentNamespace was found"); + } + + p = strstr(out_json, "labels"); + if (p != NULL) { + set_callback_error("unexpected labels were found"); + } + + increment_output_invoked(); + flb_sds_destroy(res_data); +} + +static void cb_check_format_chronicle_api_log_key(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + char *out_json = res_data; + char *p; + + if (res_ret != 0 || out_json == NULL) { + set_callback_error("formatter returned an error or no output"); + flb_sds_destroy(res_data); + return; + } + + /* base64 of: This is the target message. */ + p = strstr(out_json, "\"data\":\"VGhpcyBpcyB0aGUgdGFyZ2V0IG1lc3NhZ2Uu\""); + if (p == NULL) { + set_callback_error("expected base64 encoded log_key value was not found"); + } + + increment_output_invoked(); + flb_sds_destroy(res_data); +} + +static void cb_check_format_chronicle_api_namespace_and_labels(void *ctx, int ffd, + int res_ret, + void *res_data, + size_t res_size, + void *data) +{ + char *out_json = res_data; + char *p; + + if (res_ret != 0 || out_json == NULL) { + set_callback_error("formatter returned an error or no output"); + flb_sds_destroy(res_data); + return; + } + + p = strstr(out_json, "\"environmentNamespace\":\"tenant-a\""); + if (p == NULL) { + set_callback_error("expected environmentNamespace was not found"); + } + + p = strstr(out_json, + "\"labels\":{\"env\":{\"value\":\"production\"}," + "\"cluster_name\":{\"value\":\"blue\"}}"); + if (p == NULL) { + set_callback_error("expected labels map was not found"); + } + + increment_output_invoked(); + flb_sds_destroy(res_data); +} + +static void cb_check_format_chronicle_api_future_timestamp(void *ctx, int ffd, + int res_ret, + void *res_data, + size_t res_size, + void *data) +{ + char *out_json = res_data; + char *p; + + if (res_ret != 0 || out_json == NULL) { + set_callback_error("formatter returned an error or no output"); + flb_sds_destroy(res_data); + return; + } + + p = strstr(out_json, "\"logEntryTime\":\"2100-01-01T00:00:00.999755859Z\""); + if (p == NULL) { + set_callback_error("expected logEntryTime was not found"); + } + + /* The collection time must be later than the log entry time */ + p = strstr(out_json, "\"collectionTime\":\"2100-01-01T00:00:01.000755859Z\""); + if (p == NULL) { + set_callback_error("collectionTime is not later than logEntryTime"); + } + + increment_output_invoked(); + flb_sds_destroy(res_data); +} + void test_format_no_log_key() { flb_ctx_t *ctx; @@ -399,6 +564,39 @@ void test_format_no_log_key() stop_and_check(ctx, 1); } +void test_format_subsecond_timestamp() +{ + flb_ctx_t *ctx; + int in_ffd, out_ffd; + char *record = "[1700000000.0625, {\"message\": \"hello world\"}]"; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + NULL); + + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_subsecond_timestamp, NULL, NULL); + + flb_start(ctx); + clear_output_invoked(); + + flb_lib_push(ctx, in_ffd, record, strlen(record)); + + sleep(1); + + stop_and_check(ctx, 1); +} + void test_format_with_log_key_found() { flb_ctx_t *ctx; @@ -728,9 +926,354 @@ void test_format_split_on_metadata_change() stop_and_check(ctx, 2); } +void test_format_chronicle_api() +{ + flb_ctx_t *ctx; + int in_ffd, out_ffd; + char *record = "[1700000000.0625, {\"message\": \"hello world\"}]"; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "api", "chronicle", + "region", "EUROPE-WEST2", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + NULL); + + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_chronicle_api, NULL, NULL); + + flb_start(ctx); + clear_output_invoked(); + + flb_lib_push(ctx, in_ffd, record, strlen(record)); + + sleep(1); + + stop_and_check(ctx, 1); +} + +void test_format_chronicle_api_log_key() +{ + flb_ctx_t *ctx; + int in_ffd, out_ffd; + char record[1024]; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "api", "chronicle", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + "log_key", "message", + NULL); + + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_chronicle_api_log_key, NULL, NULL); + + flb_start(ctx); + clear_output_invoked(); + + snprintf(record, sizeof(record) - 1, + "[%ld, {\"other_key\": \"some value\", \"message\": \"This is the target message.\"}]", + (long) time(NULL)); + flb_lib_push(ctx, in_ffd, record, strlen(record)); + + sleep(1); + + stop_and_check(ctx, 1); +} + +void test_format_chronicle_api_namespace_and_labels() +{ + flb_ctx_t *ctx; + int in_ffd, out_ffd; + char record[1024]; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "api", "chronicle", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + "namespace_key", "$tenant_namespace", + "namespace", "fallback-namespace", + "label", "env production", + "label", "cluster_name $cluster['name']", + NULL); + + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_chronicle_api_namespace_and_labels, + NULL, NULL); + + flb_start(ctx); + clear_output_invoked(); + + snprintf(record, sizeof(record) - 1, + "[%ld, {\"message\": \"hello world\", \"tenant_namespace\": \"tenant-a\", " + "\"cluster\": {\"name\": \"blue\"}}]", + (long) time(NULL)); + flb_lib_push(ctx, in_ffd, record, strlen(record)); + + sleep(1); + + stop_and_check(ctx, 1); +} + +void test_format_chronicle_api_future_timestamp() +{ + flb_ctx_t *ctx; + int in_ffd, out_ffd; + /* A future record whose collection time crosses a second boundary */ + char *record = "[4102444800.999755859375, {\"message\": \"hello world\"}]"; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "api", "chronicle", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + NULL); + + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_chronicle_api_future_timestamp, NULL, NULL); + + flb_start(ctx); + clear_output_invoked(); + + flb_lib_push(ctx, in_ffd, record, strlen(record)); + + sleep(1); + + stop_and_check(ctx, 1); +} + +static flb_ctx_t *create_chronicle_api_ctx(int *out_ffd) +{ + flb_ctx_t *ctx; + int in_ffd; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + *out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, *out_ffd, + "match", "test", + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + NULL); + + return ctx; +} + +static void check_start_error(flb_ctx_t *ctx, int out_ffd, const char *reason) +{ + int ret; + + /* The test formatter enables the test mode, so credentials are skipped */ + flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_format_chronicle_api, NULL, NULL); + + ret = flb_start(ctx); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("%s", reason); + flb_stop(ctx); + } + + flb_destroy(ctx); +} + +void test_chronicle_api_invalid_api() +{ + flb_ctx_t *ctx; + int out_ffd; + + ctx = create_chronicle_api_ctx(&out_ffd); + flb_output_set(ctx, out_ffd, + "api", "backstory", + "log_type", "TEST_LOG", + NULL); + + check_start_error(ctx, out_ffd, "an unknown api must be rejected"); +} + +void test_chronicle_api_invalid_region() +{ + flb_ctx_t *ctx; + int out_ffd; + + ctx = create_chronicle_api_ctx(&out_ffd); + flb_output_set(ctx, out_ffd, + "api", "chronicle", + "region", "us.example.com/", + "log_type", "TEST_LOG", + NULL); + + check_start_error(ctx, out_ffd, "a region that is not a hostname label must be rejected"); +} + +void test_chronicle_api_invalid_log_type() +{ + flb_ctx_t *ctx; + int out_ffd; + + ctx = create_chronicle_api_ctx(&out_ffd); + flb_output_set(ctx, out_ffd, + "api", "chronicle", + "log_type", "TEST_LOG/../../other", + NULL); + + check_start_error(ctx, out_ffd, "a log_type that is not a path segment must be rejected"); +} + +struct response_case { + int status; + int result; +}; + +static int last_response_result = -1; + +static void cb_store_response_result(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + pthread_mutex_lock(&result_mutex); + last_response_result = res_ret; + pthread_mutex_unlock(&result_mutex); +} + +static void check_response_results(const char *api, + struct response_case *cases, int count) +{ + flb_ctx_t *ctx; + int i; + int ret; + int in_ffd, out_ffd; + int result; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "0.2", "grace", "1", "log_level", "error", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "chronicle", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "api", api, + "customer_id", "test-customer", + "project_id", "TESTING_FORMAT", + "log_type", "TEST_LOG", + NULL); + + ret = flb_output_set_http_test(ctx, out_ffd, "response", + cb_store_response_result, NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + for (i = 0; i < count; i++) { + ret = flb_lib_response(ctx, out_ffd, cases[i].status, "{}", 2); + TEST_CHECK(ret == 0); + + pthread_mutex_lock(&result_mutex); + result = last_response_result; + pthread_mutex_unlock(&result_mutex); + + if (!TEST_CHECK(result == cases[i].result)) { + TEST_MSG("api=%s status=%d: got %d, expected %d", + api, cases[i].status, result, cases[i].result); + } + } + + flb_stop(ctx); + flb_destroy(ctx); +} + +void test_chronicle_api_response_status() +{ + /* Client errors are not retried, except for auth, timeout and quota */ + struct response_case cases[] = { + { 200, FLB_OK }, + { 400, FLB_ERROR }, + { 401, FLB_RETRY }, + { 403, FLB_ERROR }, + { 404, FLB_ERROR }, + { 408, FLB_RETRY }, + { 429, FLB_RETRY }, + { 500, FLB_RETRY }, + { 503, FLB_RETRY }, + }; + + check_response_results("chronicle", cases, sizeof(cases) / sizeof(cases[0])); +} + +void test_legacy_api_response_status() +{ + /* The legacy API keeps retrying every failed request */ + struct response_case cases[] = { + { 200, FLB_OK }, + { 400, FLB_RETRY }, + { 404, FLB_RETRY }, + { 503, FLB_RETRY }, + }; + + check_response_results("legacy", cases, sizeof(cases) / sizeof(cases[0])); +} + +void test_chronicle_api_duplicate_label() +{ + flb_ctx_t *ctx; + int out_ffd; + + ctx = create_chronicle_api_ctx(&out_ffd); + flb_output_set(ctx, out_ffd, + "api", "chronicle", + "log_type", "TEST_LOG", + "label", "env production", + "label", "env staging", + NULL); + + check_start_error(ctx, out_ffd, "duplicate label keys must be rejected"); +} + TEST_LIST = { { "format_no_log_key", test_format_no_log_key }, + { "format_subsecond_timestamp", test_format_subsecond_timestamp }, { "format_with_log_key_found", test_format_with_log_key_found }, { "format_with_log_key_not_found", test_format_with_log_key_not_found }, { "format_with_log_key_conversion_error", test_format_with_log_key_conversion_error }, @@ -740,5 +1283,17 @@ TEST_LIST = { { "format_namespace_fallback_and_missing_label", test_format_namespace_fallback_and_missing_label }, { "format_split_on_metadata_change", test_format_split_on_metadata_change }, + { "format_chronicle_api", test_format_chronicle_api }, + { "format_chronicle_api_log_key", test_format_chronicle_api_log_key }, + { "format_chronicle_api_namespace_and_labels", + test_format_chronicle_api_namespace_and_labels }, + { "format_chronicle_api_future_timestamp", + test_format_chronicle_api_future_timestamp }, + { "chronicle_api_invalid_api", test_chronicle_api_invalid_api }, + { "chronicle_api_invalid_region", test_chronicle_api_invalid_region }, + { "chronicle_api_invalid_log_type", test_chronicle_api_invalid_log_type }, + { "chronicle_api_duplicate_label", test_chronicle_api_duplicate_label }, + { "chronicle_api_response_status", test_chronicle_api_response_status }, + { "legacy_api_response_status", test_legacy_api_response_status }, { NULL, NULL } };