[out_oracle_log_analytics] refactor OCI_logan plugin code for clarity and consistency

Signed-off-by: rghouzra <[email protected]>

Author: rghouzra

plugins/out_oracle_log_analytics/oci_logan.c

@@ -197,11 +197,8 @@ static int build_headers(struct flb_http_client *c, struct flb_oci_logan *ctx,
     flb_sds_t encoded_uri = NULL;
     flb_sds_t signature = NULL;
     flb_sds_t auth_header_str = NULL;
-
     flb_sds_t tmp_ref = NULL;
-
     size_t tmp_len = 0;
-
     unsigned char sha256_buf[32] = { 0 };
 
     tmp_sds = flb_sds_create_size(512);
@@ -576,6 +573,7 @@ static int cb_oci_logan_init(struct flb_output_instance *ins,
                              struct flb_config *config, void *data)
 {
     struct flb_oci_logan *ctx;
+
     ctx = flb_oci_logan_conf_create(ins, config);
     if (!ctx) {
         flb_plg_error(ins, "cannot initialize plugin");
@@ -663,29 +661,33 @@ static flb_sds_t compose_uri(struct flb_oci_logan *ctx,
     return full_uri;
 }
 
-// Checks if the current security token needs refresh(expires within 5min)
+/* Checks if the current security token needs refresh(expires within 5min) */
 static int token_needs_refresh(struct flb_oci_logan *ctx)
 {
-    time_t now = time(NULL);
+    time_t now;
 
+    now = time(NULL);
     return (ctx->security_token.expires_at - now) < 300;
 }
 
-// Refreshes the security token if it's close to expiration
+/* Refreshes the security token if it's close to expiration */
 static int refresh_security_token_if_needed(struct flb_oci_logan *ctx)
 {
+    flb_sds_t json_payload;
+    flb_sds_t response;
+
     if (!token_needs_refresh(ctx)) {
         return 0;
     }
 
-    flb_sds_t json_payload = create_federation_payload(ctx);
+    json_payload = create_federation_payload(ctx);
     if (!json_payload) {
         flb_plg_error(ctx->ins,
                       "failed to create federation payload for token refresh");
         return -1;
     }
 
-    flb_sds_t response = sign_and_send_federation_request(ctx, json_payload);
+    response = sign_and_send_federation_request(ctx, json_payload);
     flb_sds_destroy(json_payload);
 
     if (!response) {
@@ -722,9 +724,10 @@ static flb_sds_t sign_oci_request_with_security_token_for_logging(struct
     unsigned char *b64_out = NULL;
     size_t sig_len = 0;
     BIO *bio = NULL;
+    size_t b64_size, olen;
     EVP_PKEY *pkey = NULL;
     EVP_MD_CTX *md_ctx = NULL;
-
+    const char *headers_list;
 
     string_to_sign = flb_sds_create_size(2048);
     if (!string_to_sign) {
@@ -784,8 +787,8 @@ static flb_sds_t sign_oci_request_with_security_token_for_logging(struct
         goto cleanup;
     }
 
-    size_t b64_size = ((sig_len + 2) / 3) * 4 + 1;
-    size_t olen = 0;
+    b64_size = ((sig_len + 2) / 3) * 4 + 1;
+    olen = 0;
     b64_out = flb_malloc(b64_size);
     if (!b64_out) {
         goto cleanup;
@@ -801,7 +804,7 @@ static flb_sds_t sign_oci_request_with_security_token_for_logging(struct
         goto cleanup;
     }
 
-    const char *headers_list = payload && payload_size > 0 ?
+    headers_list = payload && payload_size > 0 ?
         "date (request-target) host x-content-sha256 content-type content-length"
         : "date (request-target) host x-content-sha256 content-length";
 
@@ -878,7 +881,8 @@ struct flb_http_client *create_oci_signed_request_for_logging(struct
     char date_buf[128];
     time_t now;
     struct tm *tm_info;
-
+    size_t date_len;
+    const char *user_agent;
 
     if (refresh_security_token_if_needed(ctx) < 0) {
         flb_plg_error(ctx->ins, "refresh_security_token_if_needed failed ");
@@ -897,7 +901,7 @@ struct flb_http_client *create_oci_signed_request_for_logging(struct
         return NULL;
     }
 
-    size_t date_len =
+    date_len =
         strftime(date_buf, sizeof(date_buf) - 1, "%a, %d %b %Y %H:%M:%S GMT",
                  tm_info);
     if (date_len == 0) {
@@ -941,7 +945,7 @@ struct flb_http_client *create_oci_signed_request_for_logging(struct
 
     flb_http_add_header(client, "opc-retry-token", 15, content_sha256,
                         strlen(content_sha256));
-    const char *user_agent = "fluent-bit-oci-plugin/1.0";
+    user_agent = "fluent-bit-oci-plugin/1.0";
     flb_http_add_header(client, "User-Agent", 10, user_agent,
                         strlen(user_agent));
 
@@ -981,19 +985,22 @@ static void dump_payload_to_file(struct flb_oci_logan *ctx, flb_sds_t payload,
 {
     char hash_in_hex[66];
     char filename[1024];
+    char *content_sha256;
+    int i;
+    size_t payload_size;
+    FILE *fp;
 
     if (!ctx->payload_files_location) {
         flb_plg_error(ctx->ins,
                       "directory path for dumping should be specified");
         return;
     }
-    size_t payload_size = flb_sds_len(payload);
-    char *content_sha256 =
-        calculate_content_sha256_b64(payload, payload_size);
+    payload_size = flb_sds_len(payload);
+    content_sha256 = calculate_content_sha256_b64(payload, payload_size);
     if (!content_sha256) {
         return;
     }
-    int i;
+
     for (i = 0; i < SHA256_DIGEST_LENGTH; i++) {
         sprintf(hash_in_hex + (i * 2), "%02x", content_sha256[i]);
     }
@@ -1007,7 +1014,7 @@ static void dump_payload_to_file(struct flb_oci_logan *ctx, flb_sds_t payload,
         return;
     }
 
-    FILE *fp = fopen(filename, "w");
+    fp = fopen(filename, "w");
     if (!fp) {
         flb_plg_error(ctx->ins, "cant open file -> %s", filename);
         flb_free(content_sha256);
@@ -1031,6 +1038,7 @@ static int flush_to_endpoint(struct flb_oci_logan *ctx,
     flb_sds_t full_uri;
     struct flb_http_client *c = NULL;
     struct flb_connection *u_conn;
+    bool should_retry;
 
     if (!payload) {
         return FLB_ERROR;
@@ -1053,7 +1061,7 @@ static int flush_to_endpoint(struct flb_oci_logan *ctx,
         flb_sds_destroy(full_uri);
         return FLB_ERROR;
     }
-    bool should_retry = 0;
+    should_retry = 0;
     if (strcmp(ctx->auth_type, "instance_principal") == 0) {
         c = create_oci_signed_request_for_logging(ctx, u_conn,
                                                   full_uri,
@@ -1346,6 +1354,8 @@ static void pack_oci_fields(msgpack_packer *packer, struct flb_oci_logan *ctx,
 static int check_metadata_dot_notation(msgpack_object key,
                                        char **metadata_key)
 {
+    int meta_key_start, meta_key_len;
+
     if (key.type != MSGPACK_OBJECT_STR) {
         return FLB_FALSE;
     }
@@ -1363,8 +1373,8 @@ static int check_metadata_dot_notation(msgpack_object key,
     if (key.via.str.ptr[FLB_OCI_METADATA_KEY_SIZE] != '.') {
         return FLB_FALSE;
     }
-    int meta_key_start = FLB_OCI_METADATA_KEY_SIZE + 1;
-    int meta_key_len = key.via.str.size - meta_key_start;
+    meta_key_start = FLB_OCI_METADATA_KEY_SIZE + 1;
+    meta_key_len = key.via.str.size - meta_key_start;
     if (meta_key_len <= 0) {
         return FLB_FALSE;
     }
@@ -1406,11 +1416,16 @@ static int get_and_pack_oci_fields_from_record(msgpack_packer *packer,
     const char *log_path_value = NULL;
     int log_path_len = 0;
 
-    //handling metadata
+    char tz_str[256];
+    int tz_len;
     struct mk_list dot_metadata_list;
     struct metadata_obj *dot_meta_obj;
     char *dot_meta_key;
     int has_dot_metadata = 0;
+    int total_meta_fields;
+    struct mk_list *head;
+    struct mk_list *tmp;
+    int j;
 
     mk_list_init(&dot_metadata_list);
 
@@ -1620,8 +1635,7 @@ static int get_and_pack_oci_fields_from_record(msgpack_packer *packer,
     }
 
     if (log_timezone) {
-        char tz_str[256];
-        int tz_len = log_timezone->via.str.size;
+        tz_len = log_timezone->via.str.size;
         if (tz_len >= sizeof(tz_str)) {
             tz_len = sizeof(tz_str) - 1;
         }
@@ -1649,7 +1663,7 @@ static int get_and_pack_oci_fields_from_record(msgpack_packer *packer,
         msgpack_pack_str_body(packer, FLB_OCI_LOG_METADATA,
                               FLB_OCI_LOG_METADATA_SIZE);
 
-        int total_meta_fields = 0;
+        total_meta_fields = 0;
         if (metadata) {
             total_meta_fields += metadata->via.map.size;
         }
@@ -1660,15 +1674,13 @@ static int get_and_pack_oci_fields_from_record(msgpack_packer *packer,
         msgpack_pack_map(packer, total_meta_fields);
 
         if (metadata) {
-            int j;
             for (j = 0; j < metadata->via.map.size; j++) {
                 msgpack_pack_object(packer, metadata->via.map.ptr[j].key);
                 msgpack_pack_object(packer, metadata->via.map.ptr[j].val);
             }
         }
 
         if (has_dot_metadata) {
-            struct mk_list *head;
             mk_list_foreach(head, &dot_metadata_list) {
                 dot_meta_obj =
                     mk_list_entry(head, struct metadata_obj, _head);
@@ -1698,8 +1710,6 @@ static int get_and_pack_oci_fields_from_record(msgpack_packer *packer,
         }
     }
     if (has_dot_metadata) {
-        struct mk_list *tmp;
-        struct mk_list *head;
         mk_list_foreach_safe(head, tmp, &dot_metadata_list) {
             dot_meta_obj = mk_list_entry(head, struct metadata_obj, _head);
             if (dot_meta_obj->key)
@@ -1729,6 +1739,9 @@ static int send_batch_with_count(struct flb_oci_logan *ctx,
     msgpack_object map;
     char *tag;
     int tag_len;
+    int skip;
+    int packed_records;
+    int map_size;
 
     tag = event_chunk->tag;
     tag_len = ((event_chunk->tag) ? flb_sds_len(event_chunk->tag) : 0);
@@ -1753,7 +1766,6 @@ static int send_batch_with_count(struct flb_oci_logan *ctx,
         log_set_id = ctx->oci_la_log_set_id;
     }
     else {
-        int skip;
         for (skip = 0; skip < start_record; skip++) {
             if (flb_log_event_decoder_next(&log_decoder, &log_event) !=
                 FLB_EVENT_DECODER_SUCCESS) {
@@ -1803,14 +1815,14 @@ static int send_batch_with_count(struct flb_oci_logan *ctx,
         current_record++;
     }
 
-    int packed_records = 0;
+    packed_records = 0;
     while (packed_records < record_count &&
            flb_log_event_decoder_next(&log_decoder,
                                       &log_event) ==
            FLB_EVENT_DECODER_SUCCESS) {
 
         map = *log_event.body;
-        int map_size = map.via.map.size;
+        map_size = map.via.map.size;
 
         /* lookupfor message or log field */
         msg = -1;
@@ -1851,9 +1863,8 @@ static int send_batch_with_count(struct flb_oci_logan *ctx,
 
     flb_log_event_decoder_destroy(&log_decoder);
 
-    out_buf =
-        flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
-                                    config->json_escape_unicode);
+    out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
+                                          config->json_escape_unicode);
     flb_plg_debug(ctx->ins, "out->buf [%s]", out_buf);
     msgpack_sbuffer_destroy(&mp_sbuf);
 
@@ -1887,6 +1898,9 @@ static int total_flush(struct flb_event_chunk *event_chunk,
     size_t estimated_total_size = 0;
     flb_sds_t log_group_id = NULL;
     flb_sds_t log_set_id = NULL;
+    int batches_needed, records_per_batch, start_record, remaining_records,
+        batch_size;
+    msgpack_object map;
 
     /* intialize decoder for  counting record and estimate size */
     ret = flb_log_event_decoder_init(&log_decoder, (char *) event_chunk->data,
@@ -1899,7 +1913,7 @@ static int total_flush(struct flb_event_chunk *event_chunk,
 
     while (flb_log_event_decoder_next(&log_decoder, &log_event) ==
            FLB_EVENT_DECODER_SUCCESS) {
-        msgpack_object map = *log_event.body;
+        map = *log_event.body;
 
         for (i = 0; i < map.via.map.size; i++) {
             if (map.via.map.ptr[i].val.type == MSGPACK_OBJECT_STR) {
@@ -1926,12 +1940,13 @@ static int total_flush(struct flb_event_chunk *event_chunk,
     }
 
     /* calculate batching parameters */
-    int batches_needed = (estimated_total_size + MAX_PAYLOAD_SIZE_BYTES - 1) / MAX_PAYLOAD_SIZE_BYTES;
+    batches_needed = (estimated_total_size + MAX_PAYLOAD_SIZE_BYTES -
+                      1) / MAX_PAYLOAD_SIZE_BYTES;
     if (batches_needed < 1) {
         batches_needed = 1;
     }
-    
-    int records_per_batch = (total_records + batches_needed - 1) / batches_needed;
+
+    records_per_batch = (total_records + batches_needed - 1) / batches_needed;
     if (records_per_batch < 1) {
         records_per_batch = 1;
     }
@@ -1941,13 +1956,13 @@ static int total_flush(struct flb_event_chunk *event_chunk,
                  batches_needed, records_per_batch);
 
     /* send data in batches */
-    int start_record = 0;
-    int remaining_records = total_records;
+    start_record = 0;
+    remaining_records = total_records;
 
     while (remaining_records > 0) {
-        int batch_size =
-            (remaining_records >
-             records_per_batch) ? records_per_batch : remaining_records;
+        batch_size = (remaining_records >
+                      records_per_batch) ? records_per_batch :
+            remaining_records;
 
         ret =
             send_batch_with_count(ctx, event_chunk, start_record, batch_size,