aws: protect credential providers with pthread_mutex

Signed-off-by: Wesley Pettit <[email protected]>

Author: PettitWesley

include/fluent-bit/flb_aws_credentials.h

@@ -118,12 +118,13 @@ struct flb_aws_provider_vtable {
  */
 struct flb_aws_provider {
     /*
-     * Fluent Bit is single-threaded but asynchonous. Co-routines are paused
-     * and resumed during blocking IO calls.
-     *
+     * Fluent Bit now has multi-threads/workers, need to a mutex to protect cred provider.
      * When a refresh is needed, only one co-routine should refresh.
+     * When one thread refreshes, the cached creds are freed and reset, there could be a double
+     * free without a lock.
+     * We use trylock to prevent deadlock.
      */
-    int locked;
+    pthread_mutex_t lock;
 
     struct flb_aws_provider_vtable *provider_vtable;
 

src/aws/flb_aws_credentials.c

@@ -537,6 +537,8 @@ static struct flb_aws_provider *standard_chain_create(struct flb_config
         return NULL;
     }
 
+    pthread_mutex_init(&provider->lock, NULL);
+
     implementation = flb_calloc(1, sizeof(struct flb_aws_provider_chain));
 
     if (!implementation) {
@@ -768,6 +770,8 @@ void flb_aws_provider_destroy(struct flb_aws_provider *provider)
             provider->provider_vtable->destroy(provider);
         }
 
+        pthread_mutex_destroy(&provider->lock);
+
         /* free managed dependencies */
         if (provider->base_aws_provider) {
             flb_aws_provider_destroy(provider->base_aws_provider);
@@ -834,27 +838,25 @@ time_t flb_aws_cred_expiration(const char *timestamp)
 }
 
 /*
- * Fluent Bit is single-threaded but asynchonous. Only one co-routine will
- * be running at a time, and they only pause/resume for IO.
- *
- * Thus, while synchronization is needed (to prevent multiple co-routines
- * from duplicating effort and performing the same work), it can be obtained
- * using a simple integer flag on the provider.
+ * Fluent Bit is now multi-threaded and asynchonous with coros. 
+ * The trylock prevents deadlock, and protects the provider
+ * when a cred refresh happens. The refresh frees and 
+ * sets the shared cred cache, a double free could occur
+ * if two threads do it at the same exact time.
  */
 
 /* Like a traditional try lock- it does not block if the lock is not obtained */
 int try_lock_provider(struct flb_aws_provider *provider)
 {
-    if (provider->locked == FLB_TRUE) {
+    int ret = 0;
+    ret = pthread_mutex_trylock(&provider->lock);
+    if (ret != 0) {
         return FLB_FALSE;
     }
-    provider->locked = FLB_TRUE;
     return FLB_TRUE;
 }
 
 void unlock_provider(struct flb_aws_provider *provider)
 {
-    if (provider->locked == FLB_TRUE) {
-        provider->locked = FLB_FALSE;
-    }
+    pthread_mutex_unlock(&provider->lock);
 }

src/aws/flb_aws_credentials_ec2.c

@@ -229,6 +229,8 @@ struct flb_aws_provider *flb_ec2_provider_create(struct flb_config *config,
         return NULL;
     }
 
+    pthread_mutex_init(&provider->lock, NULL);
+
     implementation = flb_calloc(1, sizeof(struct flb_aws_provider_ec2));
 
     if (!implementation) {

src/aws/flb_aws_credentials_http.c

@@ -250,6 +250,8 @@ struct flb_aws_provider *flb_http_provider_create(struct flb_config *config,
         return NULL;
     }
 
+    pthread_mutex_init(&provider->lock, NULL);
+
     implementation = flb_calloc(1, sizeof(struct flb_aws_provider_http));
 
     if (!implementation) {

src/aws/flb_aws_credentials_profile.c

@@ -103,10 +103,16 @@ struct flb_aws_credentials *get_credentials_fn_profile(struct flb_aws_provider
         time(NULL) >= implementation->next_refresh)) {
         AWS_CREDS_DEBUG("Retrieving credentials for AWS Profile %s",
                         implementation->profile);
-        ret = refresh_credentials(implementation, FLB_FALSE);
-        if (ret < 0) {
-            AWS_CREDS_ERROR("Failed to retrieve credentials for AWS Profile %s",
-                            implementation->profile);
+        if (try_lock_provider(provider) == FLB_TRUE) {
+            ret = refresh_credentials(implementation, FLB_FALSE);
+            unlock_provider(provider);
+            if (ret < 0) {
+                AWS_CREDS_ERROR("Failed to retrieve credentials for AWS Profile %s",
+                                implementation->profile);
+                return NULL;
+            }
+        } else {
+            AWS_CREDS_WARN("Another thread is refreshing credentials, will retry");
             return NULL;
         }
     }
@@ -152,15 +158,27 @@ struct flb_aws_credentials *get_credentials_fn_profile(struct flb_aws_provider
 int refresh_fn_profile(struct flb_aws_provider *provider)
 {
     struct flb_aws_provider_profile *implementation = provider->implementation;
+    int ret = -1;
     AWS_CREDS_DEBUG("Refresh called on the profile provider");
-    return refresh_credentials(implementation, FLB_FALSE);
+    if (try_lock_provider(provider) == FLB_TRUE) {
+        ret = refresh_credentials(implementation, FLB_FALSE);
+        unlock_provider(provider);
+        return ret;
+    }
+    return ret;
 }
 
 int init_fn_profile(struct flb_aws_provider *provider)
 {
     struct flb_aws_provider_profile *implementation = provider->implementation;
+    int ret = -1;
     AWS_CREDS_DEBUG("Init called on the profile provider");
-    return refresh_credentials(implementation, FLB_TRUE);
+    if (try_lock_provider(provider) == FLB_TRUE) {
+        ret = refresh_credentials(implementation, FLB_TRUE);
+        unlock_provider(provider);
+        return ret;
+    }
+    return ret;
 }
 
 /*
@@ -234,6 +252,8 @@ struct flb_aws_provider *flb_profile_provider_create(char* profile)
         goto error;
     }
 
+    pthread_mutex_init(&provider->lock, NULL);
+
     implementation = flb_calloc(1,
                                 sizeof(
                                 struct flb_aws_provider_profile));

src/aws/flb_aws_credentials_sts.c

@@ -308,6 +308,8 @@ struct flb_aws_provider *flb_sts_provider_create(struct flb_config *config,
         return NULL;
     }
 
+    pthread_mutex_init(&provider->lock, NULL);
+
     implementation = flb_calloc(1, sizeof(struct flb_aws_provider_sts));
     if (!implementation) {
         goto error;
@@ -578,7 +580,9 @@ struct flb_aws_provider *flb_eks_provider_create(struct flb_config *config,
         flb_errno();
         return NULL;
     }
-
+    
+    pthread_mutex_init(&provider->lock, NULL);
+    
     implementation = flb_calloc(1, sizeof(struct flb_aws_provider_eks));
 
     if (!implementation) {