diff --git a/generator/test_case_generator.go b/generator/test_case_generator.go
index 925878746..8e2a67f9f 100644
--- a/generator/test_case_generator.go
+++ b/generator/test_case_generator.go
@@ -231,6 +231,10 @@ var testTypeToTestConfig = map[string][]testConfig{
 			testDir: "./test/awsneuron", terraformDir: "terraform/eks/daemon/awsneuron",
 			targets: map[string]map[string]struct{}{"arc": {"amd64": {}}},
 		},
+		{
+			testDir: "./test/entity", terraformDir: "terraform/eks/daemon/entity",
+			targets: map[string]map[string]struct{}{"arc": {"amd64": {}}},
+		},
 	},
 	"eks_deployment": {
 		{testDir: "./test/metric_value_benchmark"},
diff --git a/terraform/eks/daemon/entity/main.tf b/terraform/eks/daemon/entity/main.tf
new file mode 100644
index 000000000..b8d80bfd4
--- /dev/null
+++ b/terraform/eks/daemon/entity/main.tf
@@ -0,0 +1,252 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+module "common" {
+  source = "../../../common"
+}
+
+module "basic_components" {
+  source = "../../../basic_components"
+
+  region = var.region
+}
+
+data "aws_eks_cluster_auth" "this" {
+  name = aws_eks_cluster.this.name
+}
+
+resource "aws_eks_cluster" "this" {
+  name     = "cwagent-eks-integ-${module.common.testing_id}"
+  role_arn = module.basic_components.role_arn
+  version  = var.k8s_version
+  vpc_config {
+    subnet_ids         = module.basic_components.public_subnet_ids
+    security_group_ids = [module.basic_components.security_group]
+  }
+}
+
+# EKS Node Groups
+resource "aws_eks_node_group" "this" {
+  cluster_name    = aws_eks_cluster.this.name
+  node_group_name = "cwagent-eks-integ-node-${module.common.testing_id}"
+  node_role_arn   = aws_iam_role.node_role.arn
+  subnet_ids      = module.basic_components.public_subnet_ids
+
+  scaling_config {
+    desired_size = 1
+    max_size     = 1
+    min_size     = 1
+  }
+
+  ami_type       = var.ami_type
+  capacity_type  = "ON_DEMAND"
+  disk_size      = 20
+  instance_types = [var.instance_type]
+
+  depends_on = [
+    aws_iam_role_policy_attachment.node_AmazonEC2ContainerRegistryReadOnly,
+    aws_iam_role_policy_attachment.node_AmazonEKS_CNI_Policy,
+    aws_iam_role_policy_attachment.node_AmazonEKSWorkerNodePolicy,
+    aws_iam_role_policy_attachment.node_CloudWatchAgentServerPolicy,
+    aws_iam_role_policy_attachment.node_AWSXRayDaemonWriteAccess
+  ]
+}
+
+# EKS Node IAM Role
+resource "aws_iam_role" "node_role" {
+  name = "cwagent-eks-Worker-Role-${module.common.testing_id}"
+
+  assume_role_policy = <<POLICY
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "ec2.amazonaws.com"
+      },
+      "Action": "sts:AssumeRole"
+    }
+  ]
+}
+POLICY
+}
+
+resource "aws_iam_role_policy_attachment" "node_AmazonEKSWorkerNodePolicy" {
+  policy_arn = "arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy"
+  role       = aws_iam_role.node_role.name
+}
+
+resource "aws_iam_role_policy_attachment" "node_AmazonEKS_CNI_Policy" {
+  policy_arn = "arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy"
+  role       = aws_iam_role.node_role.name
+}
+
+resource "aws_iam_role_policy_attachment" "node_AmazonEC2ContainerRegistryReadOnly" {
+  policy_arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly"
+  role       = aws_iam_role.node_role.name
+}
+
+resource "aws_iam_role_policy_attachment" "node_CloudWatchAgentServerPolicy" {
+  policy_arn = "arn:aws:iam::aws:policy/CloudWatchAgentServerPolicy"
+  role       = aws_iam_role.node_role.name
+}
+resource "aws_iam_role_policy_attachment" "node_AWSXRayDaemonWriteAccess" {
+  policy_arn = "arn:aws:iam::aws:policy/AWSXRayDaemonWriteAccess"
+  role       = aws_iam_role.node_role.name
+}
+
+# TODO: these security groups be created once and then reused
+# EKS Cluster Security Group
+resource "aws_security_group" "eks_cluster_sg" {
+  name        = "cwagent-eks-cluster-sg-${module.common.testing_id}"
+  description = "Cluster communication with worker nodes"
+  vpc_id      = module.basic_components.vpc_id
+}
+
+resource "aws_security_group_rule" "cluster_inbound" {
+  description              = "Allow worker nodes to communicate with the cluster API Server"
+  from_port                = 443
+  protocol                 = "tcp"
+  security_group_id        = aws_security_group.eks_cluster_sg.id
+  source_security_group_id = aws_security_group.eks_nodes_sg.id
+  to_port                  = 443
+  type                     = "ingress"
+}
+
+resource "aws_security_group_rule" "cluster_outbound" {
+  description              = "Allow cluster API Server to communicate with the worker nodes"
+  from_port                = 1024
+  protocol                 = "tcp"
+  security_group_id        = aws_security_group.eks_cluster_sg.id
+  source_security_group_id = aws_security_group.eks_nodes_sg.id
+  to_port                  = 65535
+  type                     = "egress"
+}
+
+
+# EKS Node Security Group
+resource "aws_security_group" "eks_nodes_sg" {
+  name        = "cwagent-eks-node-sg-${module.common.testing_id}"
+  description = "Security group for all nodes in the cluster"
+  vpc_id      = module.basic_components.vpc_id
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+}
+
+resource "aws_security_group_rule" "nodes_internal" {
+  description              = "Allow nodes to communicate with each other"
+  from_port                = 0
+  protocol                 = "-1"
+  security_group_id        = aws_security_group.eks_nodes_sg.id
+  source_security_group_id = aws_security_group.eks_nodes_sg.id
+  to_port                  = 65535
+  type                     = "ingress"
+}
+
+resource "aws_security_group_rule" "nodes_cluster_inbound" {
+  description              = "Allow worker Kubelets and pods to receive communication from the cluster control plane"
+  from_port                = 1025
+  protocol                 = "tcp"
+  security_group_id        = aws_security_group.eks_nodes_sg.id
+  source_security_group_id = aws_security_group.eks_cluster_sg.id
+  to_port                  = 65535
+  type                     = "ingress"
+}
+
+resource "null_resource" "clone_helm_chart" {
+  provisioner "local-exec" {
+    command = <<-EOT
+      if [ ! -d "./helm-charts" ]; then
+        git clone -b ${var.helm_chart_branch} https://github.com/aws-observability/helm-charts.git ./helm-charts
+      fi
+    EOT
+  }
+}
+
+resource "helm_release" "aws_observability" {
+  name             = "amazon-cloudwatch-observability"
+  chart            = "./helm-charts/charts/amazon-cloudwatch-observability"
+  namespace        = "amazon-cloudwatch"
+  create_namespace = true
+
+  set {
+    name  = "clusterName"
+    value = aws_eks_cluster.this.name
+  }
+
+  set {
+    name  = "region"
+    value = "us-west-2"
+  }
+  depends_on = [
+    aws_eks_cluster.this,
+    aws_eks_node_group.this,
+  null_resource.clone_helm_chart]
+}
+
+resource "kubernetes_pod" "log_generator" {
+  depends_on = [aws_eks_node_group.this]
+  metadata {
+    name      = "log-generator"
+    namespace = "default"
+  }
+
+  spec {
+    container {
+      name  = "log-generator"
+      image = "busybox"
+
+      # Run shell script that generate a log line every second
+      command = ["/bin/sh", "-c"]
+      args    = ["while true; do echo \"Log entry at $(date)\"; sleep 1; done"]
+    }
+    restart_policy = "Always"
+  }
+}
+
+
+# Get the single instance ID of the node in the node group
+data "aws_instances" "eks_node" {
+  depends_on = [
+    aws_eks_node_group.this
+  ]
+  filter {
+    name   = "tag:eks:nodegroup-name"
+    values = [aws_eks_node_group.this.node_group_name]
+  }
+}
+
+# Retrieve details of the single instance to get private DNS
+data "aws_instance" "eks_node_detail" {
+  depends_on = [
+    data.aws_instances.eks_node
+  ]
+  instance_id = data.aws_instances.eks_node.ids[0]
+}
+
+resource "null_resource" "validator" {
+  depends_on = [
+    aws_eks_node_group.this,
+    helm_release.aws_observability,
+    kubernetes_pod.log_generator
+  ]
+
+  triggers = {
+    always_run = timestamp()
+  }
+
+  provisioner "local-exec" {
+    command = <<-EOT
+      echo "Validating EKS logs for entity fields"
+      cd ../../../..
+      go test ${var.test_dir} -timeout 1h -eksClusterName=${aws_eks_cluster.this.name} -computeType=EKS -v -eksDeploymentStrategy=DAEMON -instanceId=${data.aws_instance.eks_node_detail.instance_id}
+    EOT
+  }
+}
+
diff --git a/terraform/eks/daemon/entity/providers.tf b/terraform/eks/daemon/entity/providers.tf
new file mode 100644
index 000000000..42ec47cab
--- /dev/null
+++ b/terraform/eks/daemon/entity/providers.tf
@@ -0,0 +1,29 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+provider "aws" {
+  region = var.region
+}
+
+provider "kubernetes" {
+  exec {
+    api_version = "client.authentication.k8s.io/v1beta1"
+    command     = "aws"
+    args        = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name]
+  }
+  host                   = aws_eks_cluster.this.endpoint
+  cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data)
+  token                  = data.aws_eks_cluster_auth.this.token
+}
+
+provider "helm" {
+  kubernetes {
+    host                   = aws_eks_cluster.this.endpoint
+    cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data)
+    exec {
+      api_version = "client.authentication.k8s.io/v1beta1"
+      args        = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name]
+      command     = "aws"
+    }
+  }
+}
\ No newline at end of file
diff --git a/terraform/eks/daemon/entity/variables.tf b/terraform/eks/daemon/entity/variables.tf
new file mode 100644
index 000000000..3063f072e
--- /dev/null
+++ b/terraform/eks/daemon/entity/variables.tf
@@ -0,0 +1,42 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+variable "region" {
+  type    = string
+  default = "us-west-2"
+}
+
+variable "test_dir" {
+  type    = string
+  default = "./test/entity"
+}
+
+variable "k8s_version" {
+  type    = string
+  default = "1.30"
+}
+
+variable "ami_type" {
+  type    = string
+  default = "AL2_x86_64"
+}
+
+variable "instance_type" {
+  type    = string
+  default = "t3a.medium"
+}
+
+variable "helm_chart_branch" {
+  type    = string
+  default = "main"
+}
+
+variable "cwagent_image_repo" {
+  type    = string
+  default = "public.ecr.aws/cloudwatch-agent/cloudwatch-agent"
+}
+
+variable "cwagent_image_tag" {
+  type    = string
+  default = "latest"
+}
\ No newline at end of file
diff --git a/test/entity/entity_test.go b/test/entity/entity_test.go
new file mode 100644
index 000000000..7751336a2
--- /dev/null
+++ b/test/entity/entity_test.go
@@ -0,0 +1,222 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package entity
+
+import (
+	"fmt"
+	"log"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/aws/amazon-cloudwatch-agent-test/environment"
+	"github.com/aws/amazon-cloudwatch-agent-test/util/awsservice"
+)
+
+const (
+	sleepForFlush = 180 * time.Second
+
+	entityType        = "@entity.KeyAttributes.Type"
+	entityName        = "@entity.KeyAttributes.Name"
+	entityEnvironment = "@entity.KeyAttributes.Environment"
+
+	entityPlatform          = "@entity.Attributes.PlatformType"
+	entityInstanceId        = "@entity.Attributes.EC2.InstanceId"
+	entityEKSCluster        = "@entity.Attributes.EKS.Cluster"
+	entityK8sNode           = "@entity.Attributes.K8s.Node"
+	entityK8sNamespace      = "@entity.Attributes.K8s.Namespace"
+	entityK8sWorkload       = "@entity.Attributes.K8s.Workload"
+	entityServiceNameSource = "@entity.Attributes.AWS.ServiceNameSource"
+)
+
+type EntityValidator struct {
+	requiredFields  map[string]bool
+	expectedEntity  expectedEntity
+	platformType    string
+	fieldValidators map[string]func(fieldValue string) bool
+}
+
+// NewEntityValidator initializes the validator based on the platform type.
+func NewEntityValidator(platformType string, expected expectedEntity) *EntityValidator {
+	ev := &EntityValidator{
+		expectedEntity:  expected,
+		platformType:    platformType,
+		requiredFields:  make(map[string]bool),
+		fieldValidators: make(map[string]func(fieldValue string) bool),
+	}
+
+	// Define platform-specific required fields and validators
+	if platformType == "EC2" {
+		ev.requiredFields = map[string]bool{
+			entityType:        false,
+			entityName:        false,
+			entityEnvironment: false,
+			entityPlatform:    false,
+			entityInstanceId:  false,
+		}
+		ev.fieldValidators = map[string]func(fieldValue string) bool{
+			entityType:        func(v string) bool { return v == ev.expectedEntity.entityType },
+			entityName:        func(v string) bool { return v == ev.expectedEntity.name },
+			entityEnvironment: func(v string) bool { return v == ev.expectedEntity.environment },
+			entityPlatform:    func(v string) bool { return v == ev.expectedEntity.platformType },
+			entityInstanceId:  func(v string) bool { return v == ev.expectedEntity.instanceId },
+		}
+	} else if platformType == "EKS" {
+		ev.requiredFields = map[string]bool{
+			entityType:              false,
+			entityName:              false,
+			entityEnvironment:       false,
+			entityPlatform:          false,
+			entityEKSCluster:        false,
+			entityK8sNode:           false,
+			entityK8sNamespace:      false,
+			entityK8sWorkload:       false,
+			entityServiceNameSource: false,
+		}
+		ev.fieldValidators = map[string]func(fieldValue string) bool{
+			entityType:              func(v string) bool { return v == ev.expectedEntity.entityType },
+			entityName:              func(v string) bool { return v == ev.expectedEntity.name },
+			entityEnvironment:       func(v string) bool { return v == ev.expectedEntity.environment },
+			entityPlatform:          func(v string) bool { return v == ev.expectedEntity.platformType },
+			entityEKSCluster:        func(v string) bool { return v == ev.expectedEntity.eksCluster },
+			entityK8sNode:           func(v string) bool { return v == ev.expectedEntity.k8sNode },
+			entityK8sNamespace:      func(v string) bool { return v == ev.expectedEntity.k8sNamespace },
+			entityK8sWorkload:       func(v string) bool { return v == ev.expectedEntity.k8sWorkload },
+			entityServiceNameSource: func(v string) bool { return v == ev.expectedEntity.serviceNameSource },
+		}
+	}
+	return ev
+}
+
+// ValidateField checks if a field is expected and matches the expected value.
+func (ev *EntityValidator) ValidateField(field, value string, t *testing.T) {
+	if validator, ok := ev.fieldValidators[field]; ok {
+		ev.requiredFields[field] = true
+		assert.True(t, validator(value), "Validation failed for field %s", field)
+	}
+}
+
+// AllFieldsPresent ensures all required fields are found.
+func (ev *EntityValidator) AllFieldsPresent() bool {
+	for _, present := range ev.requiredFields {
+		if !present {
+			return false
+		}
+	}
+	return true
+}
+
+type expectedEntity struct {
+	entityType        string
+	name              string
+	environment       string
+	platformType      string
+	k8sWorkload       string
+	k8sNode           string
+	k8sNamespace      string
+	eksCluster        string
+	serviceNameSource string
+	instanceId        string
+}
+
+func init() {
+	environment.RegisterEnvironmentMetaDataFlags()
+}
+
+// TestPutLogEventEntityEKS checks if entity is emitted correctly in EKS
+// through FluentBit
+func TestPutLogEventEntityEKS(t *testing.T) {
+	var instancePrivateDNS *string
+
+	env := environment.GetEnvironmentMetaData()
+	assert.NotEmpty(t, env.InstanceId)
+	assert.NotEmpty(t, env.EKSClusterName)
+	assert.NotEmpty(t, env.ComputeType)
+	if env.InstanceId != "" {
+		var err error
+		instancePrivateDNS, err = awsservice.GetInstancePrivateIpDns(env.InstanceId)
+		assert.NoError(t, err)
+		assert.NotEmpty(t, instancePrivateDNS)
+	}
+
+	// ensure that there is enough time from the "start" time and the first log line,
+	// so we don't miss it in the StartQuery call
+	log.Printf("Sleeping for %f seconds to ensure log group is ready for query", sleepForFlush.Seconds())
+	time.Sleep(sleepForFlush)
+	end := time.Now()
+
+	testCases := map[string]struct {
+		agentConfigPath string
+		podName         string
+		useEC2Tag       bool
+		expectedEntity  expectedEntity
+	}{
+		"Entity/K8sWorkloadServiceNameSource": {
+			agentConfigPath: filepath.Join("resources", "compass_default_log.json"),
+			podName:         "log-generator",
+			expectedEntity: expectedEntity{
+				entityType:        "Service",
+				name:              "log-generator",
+				environment:       "eks:" + env.EKSClusterName + "/" + "default",
+				platformType:      "AWS::EKS",
+				k8sWorkload:       "log-generator",
+				k8sNode:           *instancePrivateDNS,
+				k8sNamespace:      "default",
+				eksCluster:        env.EKSClusterName,
+				instanceId:        env.InstanceId,
+				serviceNameSource: "K8sWorkload",
+			},
+		},
+	}
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			var podApplicationLogStream string
+
+			appLogGroup := fmt.Sprintf("/aws/containerinsights/%s/%s", env.EKSClusterName, "application")
+			logStreamNames := awsservice.GetLogStreamNames(appLogGroup)
+			assert.NotZero(t, len(logStreamNames))
+			for _, streamName := range logStreamNames {
+				if strings.Contains(streamName, testCase.podName) {
+					podApplicationLogStream = streamName
+					log.Printf("Found log stream %s that matches pattern %s", streamName, testCase.podName)
+				}
+			}
+			assert.NotEmpty(t, podApplicationLogStream)
+			// check CWL to ensure we got the expected entities in the log group
+			queryString := fmt.Sprintf("fields @message, @entity.KeyAttributes.Type, @entity.KeyAttributes.Name, @entity.KeyAttributes.Environment, @entity.Attributes.PlatformType, @entity.Attributes.EKS.Cluster, @entity.Attributes.K8s.Node, @entity.Attributes.K8s.Namespace, @entity.Attributes.K8s.Workload, @entity.Attributes.AWS.ServiceNameSource, @entity.Attributes.EC2.InstanceId | filter @logStream == \"%s\"", podApplicationLogStream)
+			ValidateLogEntity(t, appLogGroup, podApplicationLogStream, &end, queryString, testCase.expectedEntity, string(env.ComputeType))
+		})
+	}
+}
+
+// ValidateLogEntity performs the entity validation for PutLogEvents.
+func ValidateLogEntity(t *testing.T, logGroup, logStream string, end *time.Time, queryString string, expectedEntity expectedEntity, entityPlatformType string) {
+	log.Printf("Checking log group/stream: %s/%s", logGroup, logStream)
+	if !awsservice.IsLogGroupExists(logGroup) {
+		t.Fatalf("application log group used for entity validation doesn't exist: %s", logGroup)
+	}
+
+	begin := end.Add(-sleepForFlush * 2)
+	log.Printf("Start time is %s and end time is %s", begin.String(), end.String())
+
+	result, err := awsservice.GetLogQueryResults(logGroup, begin.Unix(), end.Unix(), queryString)
+	assert.NoError(t, err)
+	if !assert.NotZero(t, len(result)) {
+		return
+	}
+
+	validator := NewEntityValidator(entityPlatformType, expectedEntity)
+	for _, field := range result[0] {
+		fieldName := aws.ToString(field.Field)
+		fieldValue := aws.ToString(field.Value)
+		validator.ValidateField(fieldName, fieldValue, t)
+		fmt.Printf("%s: %s\n", fieldName, fieldValue)
+	}
+
+	assert.True(t, validator.AllFieldsPresent(), "Not all required fields were found")
+}
diff --git a/util/awsservice/cloudwatchlogs.go b/util/awsservice/cloudwatchlogs.go
index c801b7880..132907922 100644
--- a/util/awsservice/cloudwatchlogs.go
+++ b/util/awsservice/cloudwatchlogs.go
@@ -197,6 +197,47 @@ func GetLogQueryStats(logGroupName string, startTime, endTime int64, queryString
 	}
 }
 
+// GetLogQueryResults for the log group between start/end (in epoch seconds) for the
+// query string.
+func GetLogQueryResults(logGroupName string, startTime, endTime int64, queryString string) ([][]types.ResultField, error) {
+	output, err := CwlClient.StartQuery(ctx, &cloudwatchlogs.StartQueryInput{
+		LogGroupName: aws.String(logGroupName),
+		StartTime:    aws.Int64(startTime),
+		EndTime:      aws.Int64(endTime),
+		QueryString:  aws.String(queryString),
+	})
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to start query for log group (%s): %w", logGroupName, err)
+	}
+
+	// Sleep a fixed amount of time after making the query to give it time to
+	// process the request.
+	time.Sleep(retryInterval)
+
+	var attempts int
+	for {
+		results, err := CwlClient.GetQueryResults(ctx, &cloudwatchlogs.GetQueryResultsInput{
+			QueryId: output.QueryId,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to get query results for log group (%s): %w", logGroupName, err)
+		}
+		switch results.Status {
+		case types.QueryStatusScheduled, types.QueryStatusRunning, types.QueryStatusUnknown:
+			if attempts >= StandardRetries {
+				return nil, fmt.Errorf("attempted get query results after %s without success. final status: %v", time.Duration(attempts)*retryInterval, results.Status)
+			}
+			attempts++
+			time.Sleep(retryInterval)
+		case types.QueryStatusComplete:
+			return results.Results, nil
+		default:
+			return nil, fmt.Errorf("unexpected query status: %v", results.Status)
+		}
+	}
+}
+
 func GetLogStreams(logGroupName string) []types.LogStream {
 	for i := 0; i < logStreamRetry; i++ {
 		describeLogStreamsOutput, err := CwlClient.DescribeLogStreams(ctx, &cloudwatchlogs.DescribeLogStreamsInput{