Implement HTTPS pod to service association endpoint calls with mtls (f…

…luent#16)

.gitignore

@@ -4,8 +4,11 @@
 *~
 _book/
 lib/jemalloc
+cmake-build-debug/
 tests/internal/flb_tests_internal.h
 tests/runtime/flb_tests_runtime.h
+tests/internal/cmake-build-debug/
+tests/runtime/cmake-build-debug/
 build/*
 include/fluent-bit/flb_info.h
 include/fluent-bit/flb_plugins.h

plugins/filter_kubernetes/kube_conf.c

@@ -235,6 +235,14 @@ void flb_kube_conf_destroy(struct flb_kube *ctx)
         flb_upstream_destroy(ctx->upstream);
     }
 
+    if(ctx->pod_association_tls) {
+        flb_tls_destroy(ctx->pod_association_tls);
+    }
+
+    if (ctx->pod_association_upstream) {
+        flb_upstream_destroy(ctx->pod_association_upstream);
+    }
+
 #ifdef FLB_HAVE_TLS
     if (ctx->tls) {
         flb_tls_destroy(ctx->tls);

plugins/filter_kubernetes/kube_conf.h

@@ -192,6 +192,15 @@ struct flb_kube {
     int pod_service_map_ttl;
     int pod_service_map_refresh_interval;
     flb_sds_t pod_service_preload_cache_dir;
+    struct flb_upstream *pod_association_upstream;
+
+    //Agent TLS certs
+    struct flb_tls *pod_association_tls;
+    char *pod_association_host_server_ca_file;
+    char *pod_association_host_client_cert_file;
+    char *pod_association_host_client_key_file;
+    int pod_association_host_tls_debug;
+    int pod_association_host_tls_verify;
 
     struct flb_tls *tls;
 

plugins/filter_kubernetes/kube_meta.c

@@ -1580,11 +1580,38 @@ static int wait_for_dns(struct flb_kube *ctx)
     return -1;
 }
 
+int flb_kube_pod_association_init(struct flb_kube *ctx, struct flb_config *config) {
+    ctx->pod_association_tls = flb_tls_create(ctx->pod_association_host_tls_verify,
+                                              ctx->pod_association_host_tls_debug,
+                                              NULL, NULL,
+                                              ctx->pod_association_host_server_ca_file,
+                                              ctx->pod_association_host_client_cert_file, ctx->pod_association_host_client_key_file, NULL);
+    if (!ctx->pod_association_tls) {
+        flb_plg_error(ctx->ins, "[kube_meta] could not create TLS config for pod association host");
+        return -1;
+    }
+    ctx->pod_association_upstream = flb_upstream_create(config,
+                                                        ctx->pod_association_host,
+                                                        ctx->pod_association_port,
+                                                        FLB_IO_TLS, ctx->pod_association_tls);
+    if (!ctx->pod_association_upstream) {
+        flb_plg_error(ctx->ins, "kube network init create pod association upstream failed");
+        flb_tls_destroy(ctx->pod_association_tls);
+        ctx->pod_association_tls = NULL;
+        return -1;
+    }
+    flb_upstream_thread_safe(ctx->pod_association_upstream);
+    mk_list_init(&ctx->pod_association_upstream->_head);
+    return 0;
+}
+
 static int flb_kube_network_init(struct flb_kube *ctx, struct flb_config *config)
 {
     int io_type = FLB_IO_TCP;
 
     ctx->upstream = NULL;
+    ctx->pod_association_upstream = NULL;
+    ctx->pod_association_tls = NULL;
 
     if (ctx->api_https == FLB_TRUE) {
         if (!ctx->tls_ca_path && !ctx->tls_ca_file) {
@@ -1618,6 +1645,11 @@ static int flb_kube_network_init(struct flb_kube *ctx, struct flb_config *config
     /* Remove async flag from upstream */
     ctx->upstream->flags &= ~(FLB_IO_ASYNC);
 
+    /* Continue the filter kubernetes plugin functionality if the pod_association fails */
+    if(ctx->use_pod_association) {
+        flb_kube_pod_association_init(ctx, config);
+    }
+
     return 0;
 }
 

plugins/filter_kubernetes/kube_meta.h

@@ -69,5 +69,6 @@ int flb_kube_meta_get(struct flb_kube *ctx,
                       struct flb_kube_meta *meta,
                       struct flb_kube_props *props);
 int flb_kube_meta_release(struct flb_kube_meta *meta);
+int flb_kube_pod_association_init(struct flb_kube *ctx, struct flb_config *config);
 
 #endif

plugins/filter_kubernetes/kubernetes.c

@@ -191,7 +191,6 @@ static int fetch_pod_service_map(struct flb_kube *ctx, char *api_server_url) {
     struct flb_http_client *c;
     size_t b_sent;
     struct flb_upstream_conn *u_conn;
-    struct flb_upstream  *u;
     char *buffer = {0};
 
     flb_plg_debug(ctx->ins, "fetch pod to service map");
@@ -203,15 +202,25 @@ static int fetch_pod_service_map(struct flb_kube *ctx, char *api_server_url) {
     }
     else {
         /* Get upstream context and connection */
-        u = flb_upstream_create(ctx->config,
-                                ctx->pod_association_host,
-                                ctx->pod_association_port,
-                                FLB_IO_TCP, NULL);
-        u_conn = flb_upstream_conn_get(u);
+        /* if block handles the TLS certificates update, as the Fluent-bit connection gets net timeout error, it destroys the upstream
+         * On the next call to fetch_pod_service_map, it creates a new pod association upstream with latest TLS certs */
+        if (!ctx->pod_association_upstream) {
+            flb_plg_debug(ctx->ins, "[kubernetes] upstream object for pod association is NULL. Making a new one now");
+            ret = flb_kube_pod_association_init(ctx,ctx->config);
+            if( ret == -1) {
+                return -1;
+            }
+        }
+
+        u_conn = flb_upstream_conn_get(ctx->pod_association_upstream);
         if (!u_conn) {
-            flb_plg_error(ctx->ins, "no upstream connections available to %s:%i",
-                          u->tcp_host, u->tcp_port);
-            return FLB_RETRY;
+            flb_plg_error(ctx->ins, "[kubernetes] no upstream connections available to %s:%i",
+                          ctx->pod_association_upstream->tcp_host, ctx->pod_association_upstream->tcp_port);
+            flb_upstream_destroy(ctx->pod_association_upstream);
+            flb_tls_destroy(ctx->pod_association_tls);
+            ctx->pod_association_upstream = NULL;
+            ctx->pod_association_tls = NULL;
+            return -1;
         }
 
         /* Create HTTP client */
@@ -221,7 +230,12 @@ static int fetch_pod_service_map(struct flb_kube *ctx, char *api_server_url) {
                             ctx->pod_association_port, NULL, 0);
 
         if (!c) {
-            flb_error("[kube_meta] could not create HTTP client");
+            flb_error("[kubernetes] could not create HTTP client");
+            flb_upstream_conn_release(u_conn);
+            flb_upstream_destroy(ctx->pod_association_upstream);
+            flb_tls_destroy(ctx->pod_association_tls);
+            ctx->pod_association_upstream = NULL;
+            ctx->pod_association_tls = NULL;
             return -1;
         }
 
@@ -1159,7 +1173,7 @@ static struct flb_config_map config_map[] = {
     * Will only check when "use_pod_association" config is set to true
     */
     {
-       FLB_CONFIG_MAP_STR, "pod_association_host", "127.0.0.1",
+       FLB_CONFIG_MAP_STR, "pod_association_host", "cloudwatch-agent.amazon-cloudwatch",
        0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host),
        "host to connect with when performing pod to service name association"
     },
@@ -1202,7 +1216,32 @@ static struct flb_config_map config_map[] = {
         0, FLB_TRUE, offsetof(struct flb_kube, pod_service_preload_cache_dir),
         "set directory with pod to service map files"
     },
-
+    {
+    FLB_CONFIG_MAP_STR, "pod_association_host_server_ca_file", "/etc/amazon-cloudwatch-observability-agent-server-cert/tls-ca.crt",
+    0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host_server_ca_file),
+    "TLS CA certificate path for communication with agent server"
+    },
+    {
+    FLB_CONFIG_MAP_STR, "pod_association_host_client_cert_file", "/etc/amazon-cloudwatch-observability-agent-client-cert/client.crt",
+    0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host_client_cert_file),
+    "Client Certificate path for enabling mTLS on calls to agent server"
+    },
+    {
+    FLB_CONFIG_MAP_STR, "pod_association_host_client_key_file", "/etc/amazon-cloudwatch-observability-agent-client-cert/client.key",
+    0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host_client_key_file),
+    "Client Certificate Key path for enabling mTLS on calls to agent server"
+    },
+    {
+    FLB_CONFIG_MAP_INT, "pod_association_host_tls_debug", "0",
+    0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host_tls_debug),
+    "set TLS debug level: 0 (no debug), 1 (error), "
+    "2 (state change), 3 (info) and 4 (verbose)"
+    },
+    {
+    FLB_CONFIG_MAP_BOOL, "pod_association_host_tls_verify", "true",
+    0, FLB_TRUE, offsetof(struct flb_kube, pod_association_host_tls_verify),
+    "enable or disable verification of TLS peer certificate"
+   },
     /* EOF */
     {0}
 };

tests/runtime/filter_kubernetes.c

@@ -354,6 +354,25 @@ static void kube_test(const char *target, int type, const char *suffix, int nExp
         clear_file(path);
     }
 
+    //Testing the default values setup
+    struct mk_list *head;
+    struct flb_filter_instance *f_ins;
+    mk_list_foreach(head, &ctx.flb->config->filters) {
+        f_ins = mk_list_entry(head, struct flb_filter_instance, _head);
+        if (strstr(f_ins->p->name, "kubernetes")) {
+            TEST_CHECK(strcmp(f_ins->p->config_map[39].name, "pod_association_host_server_ca_file") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[39].def_value, "/etc/amazon-cloudwatch-observability-agent-server-cert/tls-ca.crt") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[40].name, "pod_association_host_client_cert_file") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[40].def_value, "/etc/amazon-cloudwatch-observability-agent-client-cert/client.crt") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[41].name, "pod_association_host_client_key_file") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[41].def_value, "/etc/amazon-cloudwatch-observability-agent-client-cert/client.key") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[42].name, "pod_association_host_tls_debug") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[42].def_value, "0") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[43].name, "pod_association_host_tls_verify") == 0);
+            TEST_CHECK(strcmp(f_ins->p->config_map[43].def_value, "true") == 0);
+        }
+    }
+
     /* Start the engine */
     ret = flb_start(ctx.flb);
     TEST_CHECK_(ret == 0, "starting engine");
@@ -951,6 +970,9 @@ static void flb_test_annotations_exclude_multiple_4_container_4_stderr()
         "use_kubelet", "true", \
         "kubelet_port", "8002", \
         "Pod_Service_Preload_Cache_Dir", DPATH "/servicemap", \
+        "pod_association_host_server_ca_file", "/tst/ca.crt", \
+        "pod_association_host_client_cert_file", "/tst/client.crt", \
+        "pod_association_host_client_key_file", "/tst/client.key", \
         NULL); \
 
 static void kube_options_use_pod_association_enabled()