Adds option to delete extracted tag attributes (#78)

Author: sambart19

README.md

@@ -102,6 +102,7 @@ As `fluent-plugin-datadog` is a buffered output plugin, you can set all of the b
 | **port** | Proxy port when logs are not directly forwarded to Datadog and ssl is not used | 80 |
 | **host** | Proxy endpoint when logs are not directly forwarded to Datadog | http-intake.logs.datadoghq.com |
 | **http_proxy** | HTTP proxy, only takes effect if HTTP forwarding is enabled (`use_http`). Defaults to `HTTP_PROXY`/`http_proxy` env vars. | nil |
+| **delete_extracted_tag_attributes** | When true, removes `kubernetes` and `docker` attributes from log records after extracting them as tags. Useful to avoid duplicate data in Datadog logs UI. | false |
 
 ### Docker and Kubernetes tags
 
@@ -115,6 +116,8 @@ If your logs contain any of the following attributes, it will automatically be a
 * kubernetes.pod_name
 * docker.container_id
 
+**Note:** By default, these values will appear twice in the Datadog logs UI, once as tags (e.g., `container_name:myapp`) and once as attributes (e.g., `@kubernetes.container_name`). If you prefer to avoid this duplication, set `delete_extracted_tag_attributes` to `true` in your configuration. This will remove the `kubernetes` and `docker` attributes from the log record after the tags have been extracted.
+
 If the Datadog Agent collect them automatically, FluentD requires a plugin for this. We recommend using [fluent-plugin-kubernetes_metadata_filter](https://github.com/fabric8io/fluent-plugin-kubernetes_metadata_filter) to collect Docker and Kubernetes metadata.
 
 Configuration example:

lib/fluent/plugin/out_datadog.rb

@@ -43,6 +43,7 @@ class RetryableError < StandardError;
   config_param :dd_source, :string, :default => nil
   config_param :dd_tags, :string, :default => nil
   config_param :dd_hostname, :string, :default => nil
+  config_param :delete_extracted_tag_attributes, :bool, :default => false
 
   # Connection settings
   config_param :host, :string, :default => DD_DEFAULT_HTTP_ENDPOINT
@@ -254,6 +255,12 @@ def enrich_record(tag, time, record)
         record["ddtags"] = record["ddtags"] + "," + container_tags
       end
     end
+
+    if @delete_extracted_tag_attributes
+      record.delete('kubernetes')
+      record.delete('docker')
+    end
+
     record
   end
 

test/plugin/test_out_datadog.rb

@@ -150,6 +150,83 @@ def create_valid_subject
     end
   end
 
+  sub_test_case "delete_extracted_tag_attributes" do
+    test "should not delete kubernetes/docker attributes by default" do
+      plugin = create_driver(%[
+        api_key foo
+      ]).instance
+      record = {
+        "message" => "test",
+        "kubernetes" => {
+          "container_name" => "myapp",
+          "namespace_name" => "default",
+          "pod_name" => "myapp-abc123"
+        },
+        "docker" => {
+          "container_id" => "abc123"
+        }
+      }
+      result = plugin.enrich_record(nil, 12345, record)
+      assert_not_nil result["kubernetes"]
+      assert_not_nil result["docker"]
+    end
+
+    test "should delete kubernetes/docker attributes when delete_extracted_tag_attributes is true" do
+      plugin = create_driver(%[
+        api_key foo
+        delete_extracted_tag_attributes true
+      ]).instance
+      record = {
+        "message" => "test",
+        "kubernetes" => {
+          "container_name" => "myapp",
+          "namespace_name" => "default",
+          "pod_name" => "myapp-abc123"
+        },
+        "docker" => {
+          "container_id" => "abc123"
+        }
+      }
+      result = plugin.enrich_record(nil, 12345, record)
+      assert_nil result["kubernetes"]
+      assert_nil result["docker"]
+      # Verify tags were still extracted
+      assert_true result["ddtags"].include?("container_name:myapp")
+      assert_true result["ddtags"].include?("kube_namespace:default")
+      assert_true result["ddtags"].include?("pod_name:myapp-abc123")
+      assert_true result["ddtags"].include?("container_id:abc123")
+    end
+
+    test "should handle records without kubernetes/docker attributes when delete_extracted_tag_attributes is true" do
+      plugin = create_driver(%[
+        api_key foo
+        delete_extracted_tag_attributes true
+      ]).instance
+      record = {"message" => "test"}
+      result = plugin.enrich_record(nil, 12345, record)
+      assert_nil result["kubernetes"]
+      assert_nil result["docker"]
+    end
+
+    test "should preserve other record attributes when delete_extracted_tag_attributes is true" do
+      plugin = create_driver(%[
+        api_key foo
+        delete_extracted_tag_attributes true
+      ]).instance
+      record = {
+        "message" => "test",
+        "custom_field" => "custom_value",
+        "kubernetes" => {
+          "container_name" => "myapp"
+        }
+      }
+      result = plugin.enrich_record(nil, 12345, record)
+      assert_nil result["kubernetes"]
+      assert_equal "custom_value", result["custom_field"]
+      assert_equal "test", result["message"]
+    end
+  end
+
   sub_test_case "truncation" do
     test "truncate messages of the given length" do
       plugin = create_valid_subject