From 0c0457654dfe1ba05e0de63c55d0f974011d0efe Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Thu, 3 Dec 2020 14:03:17 -0500
Subject: [PATCH 01/14] Get boost::asio to build with openssl

---
 CMakeLists.txt                                | 15 +++
 plugins/out_stackdriver/CMakeLists.txt        |  6 +-
 plugins/out_stackdriver/stackdriver.c         |  5 +
 plugins/out_stackdriver/stackdriver_flush.cpp | 94 +++++++++++++++++++
 plugins/out_stackdriver/stackdriver_flush.h   | 10 ++
 5 files changed, 129 insertions(+), 1 deletion(-)
 create mode 100644 plugins/out_stackdriver/stackdriver_flush.cpp
 create mode 100644 plugins/out_stackdriver/stackdriver_flush.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 77c501d3556..86db74799c9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -9,6 +9,21 @@ set(FLB_VERSION_STR "${FLB_VERSION_MAJOR}.${FLB_VERSION_MINOR}.${FLB_VERSION_PAT
 
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 
+# Find openssl
+find_package(OpenSSL REQUIRED)
+include_directories(${OpenSSL_INCLUDE_DIRS})
+
+find_library(SSL_LIB libssl.a REQUIRED)
+find_library(CRYPTO_LIB libcrypto.a REQUIRED)
+
+# Find boost
+find_package(Boost REQUIRED system)
+set(Boost_USE_STATIC_LIBS    ON)
+set(Boost_USE_STATIC_RUNTIME    ON)
+include_directories(${Boost_INCLUDE_DIRS})
+
+
+
 # Define macro to identify Windows system (without Cygwin)
 if(CMAKE_SYSTEM_NAME MATCHES "Windows")
   set(FLB_SYSTEM_WINDOWS On)
diff --git a/plugins/out_stackdriver/CMakeLists.txt b/plugins/out_stackdriver/CMakeLists.txt
index 9c80c7a2d9e..dedec23d0a7 100644
--- a/plugins/out_stackdriver/CMakeLists.txt
+++ b/plugins/out_stackdriver/CMakeLists.txt
@@ -7,6 +7,10 @@ set(src
   stackdriver_http_request.c
   stackdriver_timestamp.c
   stackdriver_helper.c
+  stackdriver_flush.cpp
   )
 
-FLB_PLUGIN(out_stackdriver "${src}" "")
+
+FLB_PLUGIN(out_stackdriver "${src}" "${Boost_LIBRARIES}")
+target_link_libraries(flb-plugin-out_stackdriver OpenSSL::Crypto)
+target_link_libraries(flb-plugin-out_stackdriver OpenSSL::SSL)
diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
index e2db6f3deb8..50a68672f64 100644
--- a/plugins/out_stackdriver/stackdriver.c
+++ b/plugins/out_stackdriver/stackdriver.c
@@ -36,6 +36,8 @@
 #include "stackdriver_http_request.h"
 #include "stackdriver_timestamp.h"
 #include "stackdriver_helper.h"
+#include "stackdriver_flush.h"
+
 #include <mbedtls/base64.h>
 #include <mbedtls/sha256.h>
 
@@ -855,6 +857,9 @@ static int cb_stackdriver_init(struct flb_output_instance *ins,
         io_flags |= FLB_IO_IPV6;
     }
 
+    int res = flush_data(2);
+    printf("C++ returned: %d\n", res);
+
     /* Create Upstream context for Stackdriver Logging (no oauth2 service) */
     ctx->u = flb_upstream_create_url(config, FLB_STD_WRITE_URL,
                                      io_flags, &ins->tls);
diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
new file mode 100644
index 00000000000..c43ec974e5b
--- /dev/null
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -0,0 +1,94 @@
+#define BOOST_NETWORK_ENABLE_HTTPS
+#include "stackdriver_flush.h"
+
+#include <string>
+#include <iostream>
+#include <boost/asio.hpp>
+#include <openssl/ssl.h>
+#include <boost/asio/ssl.hpp>
+
+
+using boost::asio::ip::tcp;
+
+extern "C" int flush_data(int potato){
+    try {
+
+    boost::asio::io_service io_service;
+    // Get a list of endpoints corresponding to the server name.
+    tcp::resolver resolver(io_service);
+    tcp::resolver::query query("www.google.com", "https");
+    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
+
+    // Try each endpoint until we successfully establish a connection.
+    boost::asio::ssl::context ssl_ctx(boost::asio::ssl::context::method::sslv23_client);
+
+    boost::asio::ssl::stream<tcp::socket> socket(io_service, ssl_ctx);
+    boost::asio::connect(socket.lowest_layer(), endpoint_iterator);
+    socket.handshake(boost::asio::ssl::stream_base::handshake_type::client);
+
+    // Form the request. We specify the "Connection: close" header so that the
+    // server will close the socket after transmitting the response. This will
+    // allow us to treat all data up until the EOF as the content.
+    boost::asio::streambuf request;
+    std::ostream request_stream(&request);
+    request_stream << "GET " << "/" << " HTTP/1.0\r\n";
+    request_stream << "Host: " << "fluent-bit" << "\r\n";
+    request_stream << "Accept: */*\r\n";
+    request_stream << "Connection: close\r\n\r\n";
+
+    // Send the request.
+    boost::asio::write(socket, request);
+
+    // Read the response status line. The response streambuf will automatically
+    // grow to accommodate the entire line. The growth may be limited by passing
+    // a maximum size to the streambuf constructor.
+    boost::asio::streambuf response;
+    boost::asio::read_until(socket, response, "\r\n");
+
+    // Check that response is OK.
+    std::istream response_stream(&response);
+    std::string http_version;
+    response_stream >> http_version;
+    unsigned int status_code;
+    response_stream >> status_code;
+    std::string status_message;
+    std::getline(response_stream, status_message);
+    if (!response_stream || http_version.substr(0, 5) != "HTTP/")
+    {
+      std::cout << "Invalid response\n";
+      return 1;
+    }
+    if (status_code != 200)
+    {
+      std::cout << "Response returned with status code " << status_code << "\n";
+      return 1;
+    }
+
+    // Read the response headers, which are terminated by a blank line.
+    boost::asio::read_until(socket, response, "\r\n\r\n");
+
+    // Process the response headers.
+    std::string header;
+    while (std::getline(response_stream, header) && header != "\r")
+      std::cout << header << "\n";
+    std::cout << "\n";
+
+    // Write whatever content we already have to output.
+    if (response.size() > 0)
+      std::cout << &response;
+
+    // Read until EOF, writing data to output as we go.
+    boost::system::error_code error;
+    while (boost::asio::read(socket, response,
+          boost::asio::transfer_at_least(1), error))
+      std::cout << &response;
+    if (error != boost::asio::error::eof)
+      throw boost::system::system_error(error);
+  }
+  catch (std::exception& e)
+  {
+    std::cout << "Exception: " << e.what() << "\n";
+  }
+
+  return potato + 2;
+}
diff --git a/plugins/out_stackdriver/stackdriver_flush.h b/plugins/out_stackdriver/stackdriver_flush.h
new file mode 100644
index 00000000000..1a937e3c76f
--- /dev/null
+++ b/plugins/out_stackdriver/stackdriver_flush.h
@@ -0,0 +1,10 @@
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int flush_data(int potato);
+
+
+#ifdef __cplusplus
+}
+#endif

From 1d757468455354ca99a530d780b1b29c7b1b8734 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Thu, 3 Dec 2020 18:13:49 -0500
Subject: [PATCH 02/14] Fix http protocol issue

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index c43ec974e5b..7809d4e88a3 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -16,7 +16,9 @@ extern "C" int flush_data(int potato){
     boost::asio::io_service io_service;
     // Get a list of endpoints corresponding to the server name.
     tcp::resolver resolver(io_service);
-    tcp::resolver::query query("www.google.com", "https");
+    std::string website = "pantheon.corp.google.com";
+    std::string resource = "/";
+    tcp::resolver::query query(website, "https");
     tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
 
     // Try each endpoint until we successfully establish a connection.
@@ -31,8 +33,8 @@ extern "C" int flush_data(int potato){
     // allow us to treat all data up until the EOF as the content.
     boost::asio::streambuf request;
     std::ostream request_stream(&request);
-    request_stream << "GET " << "/" << " HTTP/1.0\r\n";
-    request_stream << "Host: " << "fluent-bit" << "\r\n";
+    request_stream << "GET "<< resource <<" HTTP/1.0\r\n";
+    request_stream << "Host: " << website << "\r\n";
     request_stream << "Accept: */*\r\n";
     request_stream << "Connection: close\r\n\r\n";
 

From 7cadaec591332c85a95db3a02f14badb5a7c4560 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Thu, 3 Dec 2020 18:14:06 -0500
Subject: [PATCH 03/14] Add locking to metrics code

This might be accessed by multiple threads during FLB_OUTPUT_RETURN
---
 src/flb_metrics.c | 20 +++++++++++++++++++-
 1 file changed, 19 insertions(+), 1 deletion(-)

diff --git a/src/flb_metrics.c b/src/flb_metrics.c
index 15d53daa428..fa2bd495a1d 100644
--- a/src/flb_metrics.c
+++ b/src/flb_metrics.c
@@ -29,6 +29,11 @@
 #include <fluent-bit/flb_metrics.h>
 #include <msgpack.h>
 
+#include <pthread.h>
+
+
+static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+
 static int id_exists(int id, struct flb_metrics *metrics)
 {
     struct mk_list *head;
@@ -116,6 +121,7 @@ int flb_metrics_add(int id, const char *title, struct flb_metrics *metrics)
 {
     int ret;
     struct flb_metric *m;
+    pthread_mutex_lock(&mutex);
 
     /* Create context */
     m = flb_malloc(sizeof(struct flb_metric));
@@ -153,12 +159,15 @@ int flb_metrics_add(int id, const char *title, struct flb_metrics *metrics)
     m->id = id;
     metrics->count++;
 
+    pthread_mutex_unlock(&mutex);
+
     return id;
 }
 
 int flb_metrics_sum(int id, size_t val, struct flb_metrics *metrics)
 {
     struct flb_metric *m;
+    pthread_mutex_lock(&mutex);
 
     m = flb_metrics_get_id(id, metrics);
     if (!m) {
@@ -166,6 +175,7 @@ int flb_metrics_sum(int id, size_t val, struct flb_metrics *metrics)
     }
 
     m->val += val;
+    pthread_mutex_unlock(&mutex);
     return 0;
 }
 
@@ -175,6 +185,7 @@ int flb_metrics_destroy(struct flb_metrics *metrics)
     struct mk_list *tmp;
     struct mk_list *head;
     struct flb_metric *m;
+    pthread_mutex_lock(&mutex);
 
     mk_list_foreach_safe(head, tmp, &metrics->list) {
         m = mk_list_entry(head, struct flb_metric, _head);
@@ -184,6 +195,7 @@ int flb_metrics_destroy(struct flb_metrics *metrics)
     }
 
     flb_free(metrics);
+    pthread_mutex_unlock(&mutex);
     return count;
 }
 
@@ -191,6 +203,7 @@ int flb_metrics_print(struct flb_metrics *metrics)
 {
     struct mk_list *head;
     struct flb_metric *m;
+    pthread_mutex_lock(&mutex);
 
     printf("[metric dump] title => '%s'", metrics->title);
 
@@ -199,7 +212,8 @@ int flb_metrics_print(struct flb_metrics *metrics)
         printf(", '%s' => %lu", m->title, m->val);
     }
     printf("\n");
-
+    pthread_mutex_unlock(&mutex);
+    
     return 0;
 }
 
@@ -212,6 +226,8 @@ int flb_metrics_dump_values(char **out_buf, size_t *out_size,
     msgpack_sbuffer mp_sbuf;
     msgpack_packer mp_pck;
 
+    pthread_mutex_lock(&mutex);
+
     /* Prepare new outgoing buffer */
     msgpack_sbuffer_init(&mp_sbuf);
     msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
@@ -228,5 +244,7 @@ int flb_metrics_dump_values(char **out_buf, size_t *out_size,
     *out_buf  = mp_sbuf.data;
     *out_size = mp_sbuf.size;
 
+    pthread_mutex_unlock(&mutex);
+
     return 0;
 }

From 84a9ca9baa0315caa02b739b4f534b6e714f23f5 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Fri, 4 Dec 2020 20:01:26 -0500
Subject: [PATCH 04/14] Prototype for multithreading in cpp

---
 plugins/out_stackdriver/stackdriver.c         | 10 +++--
 plugins/out_stackdriver/stackdriver_flush.cpp | 42 +++++++++++++++++--
 plugins/out_stackdriver/stackdriver_flush.h   | 24 ++++++++++-
 3 files changed, 68 insertions(+), 8 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
index 50a68672f64..2b2ee006bbf 100644
--- a/plugins/out_stackdriver/stackdriver.c
+++ b/plugins/out_stackdriver/stackdriver.c
@@ -857,9 +857,12 @@ static int cb_stackdriver_init(struct flb_output_instance *ins,
         io_flags |= FLB_IO_IPV6;
     }
 
-    int res = flush_data(2);
-    printf("C++ returned: %d\n", res);
-
+    printf("Before!\n");
+    StackdriverFlushContext* fctx = stackdriver_cpp_init(2);
+    stackdriver_cpp_flush(fctx);
+    printf("After!\n");
+    return 0;
+        
     /* Create Upstream context for Stackdriver Logging (no oauth2 service) */
     ctx->u = flb_upstream_create_url(config, FLB_STD_WRITE_URL,
                                      io_flags, &ins->tls);
@@ -1891,6 +1894,7 @@ static void cb_stackdriver_flush(const void *data, size_t bytes,
     struct flb_stackdriver *ctx = out_context;
     struct flb_upstream_conn *u_conn;
     struct flb_http_client *c;
+    FLB_OUTPUT_RETURN(FLB_RETRY);
 
     /* Get upstream connection */
     u_conn = flb_upstream_conn_get(ctx->u);
diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 7809d4e88a3..483c03e3f8a 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -3,14 +3,48 @@
 
 #include <string>
 #include <iostream>
-#include <boost/asio.hpp>
-#include <openssl/ssl.h>
+#include <thread>
+#include <vector>
+
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/connect.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/strand.hpp>
 #include <boost/asio/ssl.hpp>
 
 
 using boost::asio::ip::tcp;
 
-extern "C" int flush_data(int potato){
+
+extern "C" StackdriverFlushContext* stackdriver_cpp_init(int num_threads) {
+
+  StackdriverFlushContext* ctx =  new StackdriverFlushContext();
+  ctx->workers.reserve(num_threads);
+  for(int i = 0; i < num_threads; ++i)
+  {
+    ctx->workers.emplace_back(
+        [ctx] {
+          // Stops the asio event loop from running out of work
+          boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
+              work_guard = boost::asio::make_work_guard(ctx->ioc);
+
+          ctx->ioc.run();
+        });
+  }
+
+  return ctx;
+}
+
+extern "C" int stackdriver_cpp_flush(StackdriverFlushContext* ctx) {
+  boost::asio::post(ctx->ioc, [](){
+      std::cout<<"I am in a post call!!\n\n\n"<<std::endl;
+    });
+  return 0;
+
+}
+
+extern "C" int flush_data(){
     try {
 
     boost::asio::io_service io_service;
@@ -92,5 +126,5 @@ extern "C" int flush_data(int potato){
     std::cout << "Exception: " << e.what() << "\n";
   }
 
-  return potato + 2;
+  return 123;
 }
diff --git a/plugins/out_stackdriver/stackdriver_flush.h b/plugins/out_stackdriver/stackdriver_flush.h
index 1a937e3c76f..5284d01e9ca 100644
--- a/plugins/out_stackdriver/stackdriver_flush.h
+++ b/plugins/out_stackdriver/stackdriver_flush.h
@@ -1,10 +1,32 @@
+#ifndef STACKDRIVER_CPP_FLUSH_H
+#define STACKDRIVER_CPP_FLUSH_H
+
+#ifdef __cplusplus
+#include <boost/asio.hpp>
+#include <thread>
+#include <vector>
+
+class StackdriverFlushContext {
+public:
+  boost::asio::io_context ioc;
+  std::vector<std::thread> workers;
+};
+#else
+typedef struct StackdriverFlushContext StackdriverFlushContext;
+#endif
+
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-int flush_data(int potato);
+StackdriverFlushContext* stackdriver_cpp_init(int num_threads);
+int stackdriver_cpp_flush(StackdriverFlushContext* ctx);
+int stackdriver_cpp_stop(StackdriverFlushContext* ctx);
 
 
 #ifdef __cplusplus
 }
 #endif
+
+#endif /*STACKDRIVER_CPP_FLUSH_H*/

From 152f9c86e94d77da8df09ec82953367511644480 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 16:30:17 -0500
Subject: [PATCH 05/14] Not finished prototype check in

---
 plugins/out_stackdriver/stackdriver.c         | 39 +++----------
 plugins/out_stackdriver/stackdriver.h         | 10 ++++
 plugins/out_stackdriver/stackdriver_flush.cpp | 57 +++++++++++++++++--
 plugins/out_stackdriver/stackdriver_flush.h   | 14 ++---
 4 files changed, 77 insertions(+), 43 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
index 2b2ee006bbf..f76bd3cea8f 100644
--- a/plugins/out_stackdriver/stackdriver.c
+++ b/plugins/out_stackdriver/stackdriver.c
@@ -265,7 +265,7 @@ static int get_oauth2_token(struct flb_stackdriver *ctx)
     return 0;
 }
 
-static char *get_google_token(struct flb_stackdriver *ctx)
+char *get_google_token(struct flb_stackdriver *ctx)
 {
     int ret = 0;
 
@@ -859,7 +859,6 @@ static int cb_stackdriver_init(struct flb_output_instance *ins,
 
     printf("Before!\n");
     StackdriverFlushContext* fctx = stackdriver_cpp_init(2);
-    stackdriver_cpp_flush(fctx);
     printf("After!\n");
     return 0;
         
@@ -1271,13 +1270,10 @@ static int pack_json_payload(int insert_id_extracted,
         return ret;
 }
 
-static int stackdriver_format(struct flb_config *config,
-                              struct flb_input_instance *ins,
-                              void *plugin_context,
-                              void *flush_ctx,
+int stackdriver_format(struct flb_stackdriver *ctx,
                               const char *tag, int tag_len,
-                              const void *data, size_t bytes,
-                              void **out_data, size_t *out_size)
+                              const char *data, size_t bytes,
+                              flb_sds_t* out_data, size_t *out_size)
 {
     int len;
     int ret;
@@ -1297,7 +1293,6 @@ static int stackdriver_format(struct flb_config *config,
     msgpack_sbuffer mp_sbuf;
     msgpack_packer mp_pck;
     flb_sds_t out_buf;
-    struct flb_stackdriver *ctx = plugin_context;
 
     /* Parameters for severity */
     int severity_extracted = FLB_FALSE;
@@ -1889,32 +1884,14 @@ static void cb_stackdriver_flush(const void *data, size_t bytes,
     char *token;
     flb_sds_t payload_buf;
     size_t payload_size;
-    void *out_buf;
-    size_t out_size;
     struct flb_stackdriver *ctx = out_context;
     struct flb_upstream_conn *u_conn;
     struct flb_http_client *c;
-    FLB_OUTPUT_RETURN(FLB_RETRY);
 
-    /* Get upstream connection */
-    u_conn = flb_upstream_conn_get(ctx->u);
-    if (!u_conn) {
-        FLB_OUTPUT_RETURN(FLB_RETRY);
-    }
-
-    /* Reformat msgpack to stackdriver JSON payload */
-    ret = stackdriver_format(config, i_ins,
-                             ctx, NULL,
-                             tag, tag_len,
-                             data, bytes,
-                             &out_buf, &out_size);
-    if (ret != 0) {
-        flb_upstream_conn_release(u_conn);
-        FLB_OUTPUT_RETURN(FLB_RETRY);
-    }
+    struct flb_thread *cur_thread = (struct flb_thread *) pthread_getspecific(flb_thread_key);
+    stackdriver_cpp_flush(ctx, cur_thread, data, bytes, tag, tag_len);
+    FLB_OUTPUT_RETURN(FLB_RETRY);
 
-    payload_buf = (flb_sds_t) out_buf;
-    payload_size = out_size;
 
     /* Get or renew Token */
     token = get_google_token(ctx);
@@ -1994,7 +1971,7 @@ struct flb_output_plugin out_stackdriver_plugin = {
     .cb_exit      = cb_stackdriver_exit,
 
     /* Test */
-    .test_formatter.callback = stackdriver_format,
+    //.test_formatter.callback = stackdriver_format,
 
     /* Plugin flags */
     .flags          = FLB_OUTPUT_NET | FLB_IO_TLS,
diff --git a/plugins/out_stackdriver/stackdriver.h b/plugins/out_stackdriver/stackdriver.h
index 32778117e01..bd90d34e172 100644
--- a/plugins/out_stackdriver/stackdriver.h
+++ b/plugins/out_stackdriver/stackdriver.h
@@ -129,6 +129,9 @@ struct flb_stackdriver {
 
     /* Fluent Bit context */
     struct flb_config *config;
+
+    /* C++ flush context context */
+    struct StackdriverFlushContext* flush_ctx;
 };
 
 typedef enum {
@@ -154,4 +157,11 @@ typedef enum {
     INSERTID_NOT_PRESENT = 2
 } insert_id_status;
 
+char *get_google_token(struct flb_stackdriver *ctx);
+
+int stackdriver_format(struct flb_stackdriver *ctx,
+                              const char *tag, int tag_len,
+                              const char *data, size_t bytes,
+                              flb_sds_t* out_data, size_t *out_size);
+
 #endif
diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 483c03e3f8a..40e36444eff 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -1,10 +1,14 @@
 #define BOOST_NETWORK_ENABLE_HTTPS
+extern "C" {
+#include "stackdriver.h"
+}
 #include "stackdriver_flush.h"
 
 #include <string>
 #include <iostream>
 #include <thread>
 #include <vector>
+#include <mutex>
 
 #include <boost/asio/bind_executor.hpp>
 #include <boost/asio/connect.hpp>
@@ -12,16 +16,30 @@
 #include <boost/asio/post.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/ssl.hpp>
+#include <boost/bind.hpp>
+
+#include <fluent-bit/flb_output.h>
+#include <fluent-bit/flb_output_plugin.h>
+#include <fluent-bit/flb_thread.h>
 
 
 using boost::asio::ip::tcp;
 
+extern "C" char *get_google_token(struct flb_stackdriver *ctx);
+
+
+extern "C" int stackdriver_format(struct flb_stackdriver *ctx,
+                                  const char *tag, int tag_len,
+                                  const char *data, size_t bytes,
+                                  flb_sds_t* out_data, size_t *out_size);
+
+
 
 extern "C" StackdriverFlushContext* stackdriver_cpp_init(int num_threads) {
 
   StackdriverFlushContext* ctx =  new StackdriverFlushContext();
   ctx->workers.reserve(num_threads);
-  for(int i = 0; i < num_threads; ++i)
+  for(int i = 0; i < num_threads; i++)
   {
     ctx->workers.emplace_back(
         [ctx] {
@@ -36,10 +54,39 @@ extern "C" StackdriverFlushContext* stackdriver_cpp_init(int num_threads) {
   return ctx;
 }
 
-extern "C" int stackdriver_cpp_flush(StackdriverFlushContext* ctx) {
-  boost::asio::post(ctx->ioc, [](){
-      std::cout<<"I am in a post call!!\n\n\n"<<std::endl;
-    });
+void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len){
+  StackdriverFlushContext* ctx = plg_ctx->flush_ctx;
+  std::cout<<"I am in a post call!!\n\n\n"<<std::endl;
+
+  /* Get the authorization token */
+  std::unique_lock<std::mutex> lock(ctx->mutex);
+  char* token = get_google_token(plg_ctx);
+  if (!token) {
+    // flb_plg_.. ids are macros, not funcitons.
+    flb_plg_error(plg_ctx->ins, "cannot retrieve oauth2 token");
+    flb_output_return(FLB_RETRY, calling_thread);
+    return;
+  }
+
+  flb_sds_t payload_buf = NULL;
+  size_t payload_size = 0;
+  /* Reformat msgpack to stackdriver JSON payload */
+  /*
+  int ret = stackdriver_format(plg_ctx, tag, tag_len,
+                              data, data_len,
+                              &payload_buf, &payload_size);
+  if (ret != 0) {
+    flb_plg_error(plg_ctx->ins, "cannot format payload JSON");
+    flb_output_return(FLB_RETRY, calling_thread);
+    return;
+    } */
+
+}
+
+
+extern "C" int stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len) {
+  StackdriverFlushContext* ctx = plg_ctx->flush_ctx;
+  boost::asio::post(ctx->ioc, boost::bind(cpp_internal_flush, plg_ctx, calling_thread, data, data_len, tag, tag_len));
   return 0;
 
 }
diff --git a/plugins/out_stackdriver/stackdriver_flush.h b/plugins/out_stackdriver/stackdriver_flush.h
index 5284d01e9ca..efc979088b3 100644
--- a/plugins/out_stackdriver/stackdriver_flush.h
+++ b/plugins/out_stackdriver/stackdriver_flush.h
@@ -5,24 +5,24 @@
 #include <boost/asio.hpp>
 #include <thread>
 #include <vector>
+#include <mutex>
 
 class StackdriverFlushContext {
 public:
   boost::asio::io_context ioc;
   std::vector<std::thread> workers;
+  std::mutex mutex;
+
 };
+extern "C" {
 #else
 typedef struct StackdriverFlushContext StackdriverFlushContext;
 #endif
 
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
+#include "stackdriver.h"
 StackdriverFlushContext* stackdriver_cpp_init(int num_threads);
-int stackdriver_cpp_flush(StackdriverFlushContext* ctx);
-int stackdriver_cpp_stop(StackdriverFlushContext* ctx);
+int stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len);
+int stackdriver_cpp_stop(struct flb_stackdriver * plg_ctx, StackdriverFlushContext* flush_ctx);
 
 
 #ifdef __cplusplus

From edb2d8cff9e98f2863e462e895bcfe04b28bcc7b Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 18:45:59 -0500
Subject: [PATCH 06/14] Hopefully runnable prototype

---
 plugins/out_stackdriver/stackdriver.c         |  15 +-
 plugins/out_stackdriver/stackdriver.h         |   7 +-
 plugins/out_stackdriver/stackdriver_conf.c    |   9 +-
 plugins/out_stackdriver/stackdriver_flush.cpp | 149 +++++++-----------
 plugins/out_stackdriver/stackdriver_flush.h   |   4 +-
 5 files changed, 67 insertions(+), 117 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
index f76bd3cea8f..5380deb4f28 100644
--- a/plugins/out_stackdriver/stackdriver.c
+++ b/plugins/out_stackdriver/stackdriver.c
@@ -857,20 +857,9 @@ static int cb_stackdriver_init(struct flb_output_instance *ins,
         io_flags |= FLB_IO_IPV6;
     }
 
-    printf("Before!\n");
-    StackdriverFlushContext* fctx = stackdriver_cpp_init(2);
-    printf("After!\n");
-    return 0;
-        
-    /* Create Upstream context for Stackdriver Logging (no oauth2 service) */
-    ctx->u = flb_upstream_create_url(config, FLB_STD_WRITE_URL,
-                                     io_flags, &ins->tls);
+    ctx->flush_ctx = stackdriver_cpp_init(2);
     ctx->metadata_u = flb_upstream_create_url(config, "http://metadata.google.internal",
-                                     FLB_IO_TCP, NULL);
-    if (!ctx->u) {
-        flb_plg_error(ctx->ins, "upstream creation failed");
-        return -1;
-    }
+                                              FLB_IO_TCP, NULL);
     if (!ctx->metadata_u) {
         flb_plg_error(ctx->ins, "metadata upstream creation failed");
         return -1;
diff --git a/plugins/out_stackdriver/stackdriver.h b/plugins/out_stackdriver/stackdriver.h
index bd90d34e172..8665e9b86f9 100644
--- a/plugins/out_stackdriver/stackdriver.h
+++ b/plugins/out_stackdriver/stackdriver.h
@@ -37,8 +37,8 @@
 
 /* Stackdriver Logging 'write' end-point */
 #define FLB_STD_WRITE_URI "/v2/entries:write"
-#define FLB_STD_WRITE_URL \
-    "https://logging.googleapis.com" FLB_STD_WRITE_URI
+#define FLB_STD_WRITE_DOMAIN "logging.googleapis.com"
+#define FLB_STD_WRITE_URL ("https://" FLB_STD_WRITE_DOMAIN FLB_STD_WRITE_URI)
 
 /* Timestamp format */
 #define FLB_STD_TIME_FMT  "%Y-%m-%dT%H:%M:%S"
@@ -118,9 +118,6 @@ struct flb_stackdriver {
     /* oauth2 context */
     struct flb_oauth2 *o;
 
-    /* upstream context for stackdriver write end-point */
-    struct flb_upstream *u;
-
     /* upstream context for metadata end-point */
     struct flb_upstream *metadata_u;
 
diff --git a/plugins/out_stackdriver/stackdriver_conf.c b/plugins/out_stackdriver/stackdriver_conf.c
index 57c9262b668..dc587cf9e44 100644
--- a/plugins/out_stackdriver/stackdriver_conf.c
+++ b/plugins/out_stackdriver/stackdriver_conf.c
@@ -31,6 +31,7 @@
 
 #include "stackdriver.h"
 #include "stackdriver_conf.h"
+#include "stackdriver_flush.h"
 
 static inline int key_cmp(const char *str, int len, const char *cmp) {
 
@@ -351,6 +352,10 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx)
         return -1;
     }
 
+    if (ctx->flush_ctx) {
+        stackdriver_cpp_destroy(ctx);
+    }
+
     if (ctx->k8s_resource_type){
         flb_sds_destroy(ctx->namespace_name);
         flb_sds_destroy(ctx->pod_name);
@@ -390,9 +395,7 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx)
         flb_upstream_destroy(ctx->metadata_u);
     }
 
-    if (ctx->u) {
-        flb_upstream_destroy(ctx->u);
-    }
+
     flb_free(ctx);
 
     return 0;
diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 40e36444eff..4337340799e 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -1,6 +1,9 @@
 #define BOOST_NETWORK_ENABLE_HTTPS
 extern "C" {
 #include "stackdriver.h"
+#include <fluent-bit/flb_output.h>
+#include <fluent-bit/flb_output_plugin.h>
+#include <fluent-bit/flb_thread.h>
 }
 #include "stackdriver_flush.h"
 
@@ -16,13 +19,16 @@ extern "C" {
 #include <boost/asio/post.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/ssl.hpp>
+
 #include <boost/bind.hpp>
 
-#include <fluent-bit/flb_output.h>
-#include <fluent-bit/flb_output_plugin.h>
-#include <fluent-bit/flb_thread.h>
+#include <boost/beast/core.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/beast/version.hpp>
 
 
+namespace beast = boost::beast;
+namespace http = beast::http;
 using boost::asio::ip::tcp;
 
 extern "C" char *get_google_token(struct flb_stackdriver *ctx);
@@ -60,118 +66,73 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
 
   /* Get the authorization token */
   std::unique_lock<std::mutex> lock(ctx->mutex);
-  char* token = get_google_token(plg_ctx);
-  if (!token) {
+  char* c_token = get_google_token(plg_ctx);
+  if (!c_token) {
     // flb_plg_.. ids are macros, not funcitons.
     flb_plg_error(plg_ctx->ins, "cannot retrieve oauth2 token");
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
+  std::string token = c_token;
+  lock.release();
 
   flb_sds_t payload_buf = NULL;
   size_t payload_size = 0;
   /* Reformat msgpack to stackdriver JSON payload */
-  /*
   int ret = stackdriver_format(plg_ctx, tag, tag_len,
-                              data, data_len,
-                              &payload_buf, &payload_size);
+                               data, data_len,
+                               &payload_buf, &payload_size);
   if (ret != 0) {
     flb_plg_error(plg_ctx->ins, "cannot format payload JSON");
     flb_output_return(FLB_RETRY, calling_thread);
     return;
-    } */
+  }
+
+  try {
+  // look up endpoint
+  tcp::resolver resolver(ctx->ioc);
+  tcp::resolver::query query(FLB_STD_WRITE_DOMAIN, "https");
+  tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
+
+  // handshake
+  boost::asio::ssl::context ssl_ctx(boost::asio::ssl::context::method::sslv23_client);
+  boost::asio::ssl::stream<tcp::socket> stream(ctx->ioc, ssl_ctx);
+  boost::asio::connect(stream.lowest_layer(), endpoint_iterator);
+  stream.handshake(boost::asio::ssl::stream_base::handshake_type::client);
+
+  // HTTP request
+  http::request<http::string_body> req{http::verb::post, FLB_STD_WRITE_URI, 11};
+  req.set(http::field::host, FLB_STD_WRITE_DOMAIN);
+  req.set(http::field::user_agent, "Fluent-Bit");
+  req.set(http::field::content_type, "application/json");
+  req.set(http::field::authorization, token);
+  req.set(http::field::content_length, payload_size);
+  req.set(http::field::body, payload_buf);
+  http::write(stream, req);
+
+  // Receive the HTTP response
+  beast::flat_buffer buffer;
+  http::response<http::dynamic_body> res;
+  http::read(stream, buffer, res);
+
+  // Write the message to standard out
+  std::cout << res << std::endl;
+  } catch (std::exception& e) {
+    flb_plg_error(plg_ctx->ins, "https request failed: ", e.what());
+    flb_output_return(FLB_RETRY, calling_thread);
+  }
+  flb_sds_destroy(payload_buf);
 
 }
 
 
-extern "C" int stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len) {
+extern "C" void stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len) {
   StackdriverFlushContext* ctx = plg_ctx->flush_ctx;
   boost::asio::post(ctx->ioc, boost::bind(cpp_internal_flush, plg_ctx, calling_thread, data, data_len, tag, tag_len));
-  return 0;
-
+  flb_thread_yield(calling_thread, FLB_FALSE);
 }
 
-extern "C" int flush_data(){
-    try {
-
-    boost::asio::io_service io_service;
-    // Get a list of endpoints corresponding to the server name.
-    tcp::resolver resolver(io_service);
-    std::string website = "pantheon.corp.google.com";
-    std::string resource = "/";
-    tcp::resolver::query query(website, "https");
-    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
-
-    // Try each endpoint until we successfully establish a connection.
-    boost::asio::ssl::context ssl_ctx(boost::asio::ssl::context::method::sslv23_client);
-
-    boost::asio::ssl::stream<tcp::socket> socket(io_service, ssl_ctx);
-    boost::asio::connect(socket.lowest_layer(), endpoint_iterator);
-    socket.handshake(boost::asio::ssl::stream_base::handshake_type::client);
-
-    // Form the request. We specify the "Connection: close" header so that the
-    // server will close the socket after transmitting the response. This will
-    // allow us to treat all data up until the EOF as the content.
-    boost::asio::streambuf request;
-    std::ostream request_stream(&request);
-    request_stream << "GET "<< resource <<" HTTP/1.0\r\n";
-    request_stream << "Host: " << website << "\r\n";
-    request_stream << "Accept: */*\r\n";
-    request_stream << "Connection: close\r\n\r\n";
-
-    // Send the request.
-    boost::asio::write(socket, request);
-
-    // Read the response status line. The response streambuf will automatically
-    // grow to accommodate the entire line. The growth may be limited by passing
-    // a maximum size to the streambuf constructor.
-    boost::asio::streambuf response;
-    boost::asio::read_until(socket, response, "\r\n");
-
-    // Check that response is OK.
-    std::istream response_stream(&response);
-    std::string http_version;
-    response_stream >> http_version;
-    unsigned int status_code;
-    response_stream >> status_code;
-    std::string status_message;
-    std::getline(response_stream, status_message);
-    if (!response_stream || http_version.substr(0, 5) != "HTTP/")
-    {
-      std::cout << "Invalid response\n";
-      return 1;
-    }
-    if (status_code != 200)
-    {
-      std::cout << "Response returned with status code " << status_code << "\n";
-      return 1;
-    }
-
-    // Read the response headers, which are terminated by a blank line.
-    boost::asio::read_until(socket, response, "\r\n\r\n");
-
-    // Process the response headers.
-    std::string header;
-    while (std::getline(response_stream, header) && header != "\r")
-      std::cout << header << "\n";
-    std::cout << "\n";
-
-    // Write whatever content we already have to output.
-    if (response.size() > 0)
-      std::cout << &response;
-
-    // Read until EOF, writing data to output as we go.
-    boost::system::error_code error;
-    while (boost::asio::read(socket, response,
-          boost::asio::transfer_at_least(1), error))
-      std::cout << &response;
-    if (error != boost::asio::error::eof)
-      throw boost::system::system_error(error);
-  }
-  catch (std::exception& e)
-  {
-    std::cout << "Exception: " << e.what() << "\n";
-  }
+extern "C" void stackdriver_cpp_destroy(struct flb_stackdriver * plg_ctx) {
+  // Just leak for now
 
-  return 123;
 }
diff --git a/plugins/out_stackdriver/stackdriver_flush.h b/plugins/out_stackdriver/stackdriver_flush.h
index efc979088b3..a3f05f4f3d2 100644
--- a/plugins/out_stackdriver/stackdriver_flush.h
+++ b/plugins/out_stackdriver/stackdriver_flush.h
@@ -21,8 +21,8 @@ typedef struct StackdriverFlushContext StackdriverFlushContext;
 
 #include "stackdriver.h"
 StackdriverFlushContext* stackdriver_cpp_init(int num_threads);
-int stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len);
-int stackdriver_cpp_stop(struct flb_stackdriver * plg_ctx, StackdriverFlushContext* flush_ctx);
+void stackdriver_cpp_flush(struct flb_stackdriver * plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len);
+void stackdriver_cpp_destroy(struct flb_stackdriver * plg_ctx);
 
 
 #ifdef __cplusplus

From 683d1e884cd1f35b815790dc1c6038b5b344dd0e Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 23:25:22 -0500
Subject: [PATCH 07/14] Multithreaded sync'd prototype

---
 CMakeLists.txt                                |  1 +
 plugins/out_stackdriver/stackdriver_flush.cpp | 72 ++++++++++---------
 2 files changed, 39 insertions(+), 34 deletions(-)

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 86db74799c9..1cb5e06e5ba 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -8,6 +8,7 @@ set(FLB_VERSION_PATCH  0)
 set(FLB_VERSION_STR "${FLB_VERSION_MAJOR}.${FLB_VERSION_MINOR}.${FLB_VERSION_PATCH}")
 
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)
+set (CMAKE_CXX_STANDARD 11)
 
 # Find openssl
 find_package(OpenSSL REQUIRED)
diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 4337340799e..f55296b5eb2 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -1,3 +1,4 @@
+#define BOOST_ASIO_ENABLE_HANDLER_TRACKING
 #define BOOST_NETWORK_ENABLE_HTTPS
 extern "C" {
 #include "stackdriver.h"
@@ -13,7 +14,6 @@ extern "C" {
 #include <vector>
 #include <mutex>
 
-#include <boost/asio/bind_executor.hpp>
 #include <boost/asio/connect.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/post.hpp>
@@ -25,6 +25,8 @@ extern "C" {
 #include <boost/beast/core.hpp>
 #include <boost/beast/http.hpp>
 #include <boost/beast/version.hpp>
+#include <boost/lexical_cast.hpp>
+
 
 
 namespace beast = boost::beast;
@@ -62,7 +64,7 @@ extern "C" StackdriverFlushContext* stackdriver_cpp_init(int num_threads) {
 
 void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len){
   StackdriverFlushContext* ctx = plg_ctx->flush_ctx;
-  std::cout<<"I am in a post call!!\n\n\n"<<std::endl;
+  std::cout<<"Callback start\n";
 
   /* Get the authorization token */
   std::unique_lock<std::mutex> lock(ctx->mutex);
@@ -74,54 +76,56 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     return;
   }
   std::string token = c_token;
-  lock.release();
+  lock.unlock();
 
-  flb_sds_t payload_buf = NULL;
+  flb_sds_t c_payload_buf = NULL;
   size_t payload_size = 0;
   /* Reformat msgpack to stackdriver JSON payload */
   int ret = stackdriver_format(plg_ctx, tag, tag_len,
                                data, data_len,
-                               &payload_buf, &payload_size);
+                               &c_payload_buf, &payload_size);
   if (ret != 0) {
     flb_plg_error(plg_ctx->ins, "cannot format payload JSON");
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
+  std::string payload(c_payload_buf, payload_size);
+  flb_sds_destroy(c_payload_buf);
 
   try {
-  // look up endpoint
-  tcp::resolver resolver(ctx->ioc);
-  tcp::resolver::query query(FLB_STD_WRITE_DOMAIN, "https");
-  tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
-
-  // handshake
-  boost::asio::ssl::context ssl_ctx(boost::asio::ssl::context::method::sslv23_client);
-  boost::asio::ssl::stream<tcp::socket> stream(ctx->ioc, ssl_ctx);
-  boost::asio::connect(stream.lowest_layer(), endpoint_iterator);
-  stream.handshake(boost::asio::ssl::stream_base::handshake_type::client);
-
-  // HTTP request
-  http::request<http::string_body> req{http::verb::post, FLB_STD_WRITE_URI, 11};
-  req.set(http::field::host, FLB_STD_WRITE_DOMAIN);
-  req.set(http::field::user_agent, "Fluent-Bit");
-  req.set(http::field::content_type, "application/json");
-  req.set(http::field::authorization, token);
-  req.set(http::field::content_length, payload_size);
-  req.set(http::field::body, payload_buf);
-  http::write(stream, req);
-
-  // Receive the HTTP response
-  beast::flat_buffer buffer;
-  http::response<http::dynamic_body> res;
-  http::read(stream, buffer, res);
-
-  // Write the message to standard out
-  std::cout << res << std::endl;
+    // look up endpoint
+    tcp::resolver resolver(ctx->ioc);
+    tcp::resolver::query query(FLB_STD_WRITE_DOMAIN, "https");
+    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
+
+    // handshake
+    boost::asio::ssl::context ssl_ctx(boost::asio::ssl::context::method::sslv23_client);
+    boost::asio::ssl::stream<tcp::socket> stream(ctx->ioc, ssl_ctx);
+    boost::asio::connect(stream.lowest_layer(), endpoint_iterator);
+    stream.handshake(boost::asio::ssl::stream_base::handshake_type::client);
+
+    // HTTP request
+    http::request<http::string_body> req{http::verb::post, FLB_STD_WRITE_URI, 11};
+    req.set(http::field::host, FLB_STD_WRITE_DOMAIN);
+    req.set(http::field::user_agent, "Fluent-Bit");
+    req.set(http::field::content_type, "application/json");
+    req.set(http::field::authorization, std::string("Bearer ") + token);
+    req.set(http::field::content_length, boost::lexical_cast<std::string>(payload_size));
+    req.body() = payload;
+    req.prepare_payload();
+
+    http::write(stream, req);
+
+
+    // Receive the HTTP response
+    beast::flat_buffer buffer;
+    http::response<http::dynamic_body> res;
+    http::read(stream, buffer, res);
+
   } catch (std::exception& e) {
     flb_plg_error(plg_ctx->ins, "https request failed: ", e.what());
     flb_output_return(FLB_RETRY, calling_thread);
   }
-  flb_sds_destroy(payload_buf);
 
 }
 

From 9c83c4336aeab68a85b51a9a4012f805a66341bc Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 23:27:23 -0500
Subject: [PATCH 08/14] Remove debugging output

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index f55296b5eb2..550d3a62d97 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -1,4 +1,3 @@
-#define BOOST_ASIO_ENABLE_HANDLER_TRACKING
 #define BOOST_NETWORK_ENABLE_HTTPS
 extern "C" {
 #include "stackdriver.h"

From 37cdc14292855db6740f87b9b9cec1ec07f25998 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 23:29:09 -0500
Subject: [PATCH 09/14] Remove debug print

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 550d3a62d97..657b0bf78c8 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -63,7 +63,6 @@ extern "C" StackdriverFlushContext* stackdriver_cpp_init(int num_threads) {
 
 void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* calling_thread, const char* data, size_t data_len, const char* tag, int tag_len){
   StackdriverFlushContext* ctx = plg_ctx->flush_ctx;
-  std::cout<<"Callback start\n";
 
   /* Get the authorization token */
   std::unique_lock<std::mutex> lock(ctx->mutex);

From 7695dda066a5c68048cae4ff286735b3a31a9373 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Mon, 7 Dec 2020 23:56:18 -0500
Subject: [PATCH 10/14] Try to parse return codes

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 657b0bf78c8..87fb1d24890 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -23,6 +23,7 @@ extern "C" {
 
 #include <boost/beast/core.hpp>
 #include <boost/beast/http.hpp>
+#include <boost/beast/http/status.hpp>
 #include <boost/beast/version.hpp>
 #include <boost/lexical_cast.hpp>
 
@@ -117,13 +118,27 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
 
     // Receive the HTTP response
     beast::flat_buffer buffer;
-    http::response<http::dynamic_body> res;
-    http::read(stream, buffer, res);
+    http::response<http::dynamic_body> resp;
+    http::read(stream, buffer, resp);
+    http::status_class status_class = http::to_status_class(resp.result());
+    if (status_class == http::status_class::successful){
+      flb_output_return(FLB_OK, calling_thread);
+      return;
+    } else if (status_class == http::status_class::server_error){
+      flb_output_return(FLB_RETRY, calling_thread);
+      return;
+    } else {
+      flb_output_return(FLB_ERROR, calling_thread);
+      return;
+    }
 
   } catch (std::exception& e) {
     flb_plg_error(plg_ctx->ins, "https request failed: ", e.what());
     flb_output_return(FLB_RETRY, calling_thread);
+    return;
   }
+  flb_plg_error(plg_ctx->ins, "This line shouldn't be executed");
+  flb_output_return(FLB_ERROR, calling_thread);
 
 }
 

From b79cb3c1a7a57990cb30543f0b81c9e0564982ae Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Tue, 8 Dec 2020 16:36:34 -0500
Subject: [PATCH 11/14] Hopefully small perf improvement

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 87fb1d24890..aa5a47e76ed 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -89,7 +89,6 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     return;
   }
   std::string payload(c_payload_buf, payload_size);
-  flb_sds_destroy(c_payload_buf);
 
   try {
     // look up endpoint
@@ -110,8 +109,7 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     req.set(http::field::content_type, "application/json");
     req.set(http::field::authorization, std::string("Bearer ") + token);
     req.set(http::field::content_length, boost::lexical_cast<std::string>(payload_size));
-    req.body() = payload;
-    req.prepare_payload();
+    req.body() = payload_buf;
 
     http::write(stream, req);
 
@@ -122,21 +120,26 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     http::read(stream, buffer, resp);
     http::status_class status_class = http::to_status_class(resp.result());
     if (status_class == http::status_class::successful){
+      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_OK, calling_thread);
       return;
     } else if (status_class == http::status_class::server_error){
+      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_RETRY, calling_thread);
       return;
     } else {
+      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_ERROR, calling_thread);
       return;
     }
 
   } catch (std::exception& e) {
+    flb_sds_destroy(c_payload_buf);
     flb_plg_error(plg_ctx->ins, "https request failed: ", e.what());
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
+  flb_sds_destroy(c_payload_buf);
   flb_plg_error(plg_ctx->ins, "This line shouldn't be executed");
   flb_output_return(FLB_ERROR, calling_thread);
 

From ca30d0ae26773fd1257769987218ee7b2182a9d0 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Tue, 8 Dec 2020 16:41:28 -0500
Subject: [PATCH 12/14] Fix typo

---
 plugins/out_stackdriver/stackdriver_flush.cpp | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index aa5a47e76ed..128968e9bb4 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -88,7 +88,6 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
-  std::string payload(c_payload_buf, payload_size);
 
   try {
     // look up endpoint
@@ -109,7 +108,7 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     req.set(http::field::content_type, "application/json");
     req.set(http::field::authorization, std::string("Bearer ") + token);
     req.set(http::field::content_length, boost::lexical_cast<std::string>(payload_size));
-    req.body() = payload_buf;
+    req.body() = c_payload_buf;
 
     http::write(stream, req);
 

From 5bd73543a646a200a18a06e311ce2ab139233380 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Tue, 8 Dec 2020 18:12:20 -0500
Subject: [PATCH 13/14] Revert previous 2 commits

This reverts commit b79cb3c1a7a57990cb30543f0b81c9e0564982ae.
---
 plugins/out_stackdriver/stackdriver_flush.cpp | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/plugins/out_stackdriver/stackdriver_flush.cpp b/plugins/out_stackdriver/stackdriver_flush.cpp
index 128968e9bb4..87fb1d24890 100644
--- a/plugins/out_stackdriver/stackdriver_flush.cpp
+++ b/plugins/out_stackdriver/stackdriver_flush.cpp
@@ -88,6 +88,8 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
+  std::string payload(c_payload_buf, payload_size);
+  flb_sds_destroy(c_payload_buf);
 
   try {
     // look up endpoint
@@ -108,7 +110,8 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     req.set(http::field::content_type, "application/json");
     req.set(http::field::authorization, std::string("Bearer ") + token);
     req.set(http::field::content_length, boost::lexical_cast<std::string>(payload_size));
-    req.body() = c_payload_buf;
+    req.body() = payload;
+    req.prepare_payload();
 
     http::write(stream, req);
 
@@ -119,26 +122,21 @@ void cpp_internal_flush(struct flb_stackdriver* plg_ctx, struct flb_thread* call
     http::read(stream, buffer, resp);
     http::status_class status_class = http::to_status_class(resp.result());
     if (status_class == http::status_class::successful){
-      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_OK, calling_thread);
       return;
     } else if (status_class == http::status_class::server_error){
-      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_RETRY, calling_thread);
       return;
     } else {
-      flb_sds_destroy(c_payload_buf);
       flb_output_return(FLB_ERROR, calling_thread);
       return;
     }
 
   } catch (std::exception& e) {
-    flb_sds_destroy(c_payload_buf);
     flb_plg_error(plg_ctx->ins, "https request failed: ", e.what());
     flb_output_return(FLB_RETRY, calling_thread);
     return;
   }
-  flb_sds_destroy(c_payload_buf);
   flb_plg_error(plg_ctx->ins, "This line shouldn't be executed");
   flb_output_return(FLB_ERROR, calling_thread);
 

From e9461a4fae462e136cd988bb0c36bc68465bad69 Mon Sep 17 00:00:00 2001
From: Henrique S Matulis <69014250+hsmatulisgoogle@users.noreply.github.com>
Date: Tue, 8 Dec 2020 21:18:05 -0500
Subject: [PATCH 14/14] Add some debug code

---
 plugins/out_stackdriver/stackdriver.c | 2 +-
 src/flb_pack.c                        | 3 +++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/plugins/out_stackdriver/stackdriver.c b/plugins/out_stackdriver/stackdriver.c
index 5380deb4f28..e8a359dd709 100644
--- a/plugins/out_stackdriver/stackdriver.c
+++ b/plugins/out_stackdriver/stackdriver.c
@@ -857,7 +857,7 @@ static int cb_stackdriver_init(struct flb_output_instance *ins,
         io_flags |= FLB_IO_IPV6;
     }
 
-    ctx->flush_ctx = stackdriver_cpp_init(2);
+    ctx->flush_ctx = stackdriver_cpp_init(1);
     ctx->metadata_u = flb_upstream_create_url(config, "http://metadata.google.internal",
                                               FLB_IO_TCP, NULL);
     if (!ctx->metadata_u) {
diff --git a/src/flb_pack.c b/src/flb_pack.c
index 81c3b44f0a4..db2ba8db06a 100644
--- a/src/flb_pack.c
+++ b/src/flb_pack.c
@@ -682,9 +682,11 @@ flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size)
     }
 
     root = &result.data;
+    int increase_count = 0;
     while (1) {
         ret = flb_msgpack_to_json(out_buf, out_size, root);
         if (ret <= 0) {
+            increase_count++;
             tmp_buf = flb_sds_increase(out_buf, 256);
             if (tmp_buf) {
                 out_buf = tmp_buf;
@@ -701,6 +703,7 @@ flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size)
             break;
         }
     }
+    printf("increase_count: %d\tin_size: %lu\tout_size: %d\n", increase_count, in_size, ret);
 
     msgpack_unpacked_destroy(&result);
     flb_sds_len_set(out_buf, ret);