diff --git a/enricher/ecs/ecs.go b/enricher/ecs/ecs.go
new file mode 100644
index 0000000..d6b2b62
--- /dev/null
+++ b/enricher/ecs/ecs.go
@@ -0,0 +1,102 @@
+package ecs
+
+import (
+	"os"
+	"regexp"
+	"strconv"
+	"time"
+
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher"
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher/mappings"
+	"github.com/sirupsen/logrus"
+)
+
+type Enricher struct {
+	canvaAWSAccount string
+	canvaAppName    string
+	logGroup        string
+	ecsTaskFamily   string
+	ecsTaskRevision int
+}
+
+var _ enricher.IEnricher = (*Enricher)(nil)
+
+func NewEnricher() *Enricher {
+	ecsTaskDefinition := os.Getenv("ECS_TASK_DEFINITION")
+	re := regexp.MustCompile(`^(?P<ecs_task_family>[^ ]*):(?P<ecs_task_revision>[\d]+)$`)
+	ecsTaskDefinitionParts := re.FindStringSubmatch(ecsTaskDefinition)
+	var (
+		ecsTaskFamily   string
+		ecsTaskRevision int
+	)
+	ecsTaskFamilyIndex := re.SubexpIndex("ecs_task_family")
+	ecsTaskRevisionIndex := re.SubexpIndex("ecs_task_revision")
+
+	if len(ecsTaskDefinitionParts) >= ecsTaskFamilyIndex {
+		ecsTaskFamily = ecsTaskDefinitionParts[ecsTaskFamilyIndex]
+	}
+	if len(ecsTaskDefinitionParts) >= ecsTaskRevisionIndex {
+		var err error
+		ecsTaskRevision, err = strconv.Atoi(ecsTaskDefinitionParts[re.SubexpIndex("ecs_task_revision")])
+		if err != nil {
+			logrus.Warnf("[kinesis] ecs_task_revision not found for ECS_TASK_DEFINITION=%s", ecsTaskDefinition)
+		}
+	}
+
+	return &Enricher{
+		canvaAWSAccount: os.Getenv("CANVA_AWS_ACCOUNT"),
+		canvaAppName:    os.Getenv("CANVA_APP_NAME"),
+		logGroup:        os.Getenv("LOG_GROUP"),
+		ecsTaskFamily:   ecsTaskFamily,
+		ecsTaskRevision: ecsTaskRevision,
+	}
+}
+
+// EnrichRecord modifies existing record.
+func (enr *Enricher) EnrichRecord(r map[interface{}]interface{}, t time.Time) map[interface{}]interface{} {
+	resource := map[interface{}]interface{}{
+		mappings.RESOURCE_CLOUD_ACCOUNT_ID: enr.canvaAWSAccount,
+		"service.name":                     enr.canvaAppName,
+		"cloud.platform":                   "aws_ecs",
+		"aws.ecs.launchtype":               "EC2",
+		"aws.ecs.task.family":              enr.ecsTaskFamily,
+		"aws.ecs.task.revision":            enr.ecsTaskRevision,
+		"aws.log.group.names":              enr.logGroup,
+	}
+	body := make(map[interface{}]interface{})
+
+	var (
+		ok        bool
+		strVal    string
+		timestamp interface{}
+	)
+	for k, v := range r {
+		strVal, ok = k.(string)
+		if ok {
+			switch strVal {
+			case "ecs_task_definition":
+				// Skip
+			case "timestamp":
+				timestamp = v
+			case "ec2_instance_id":
+				resource["host.id"] = v
+			case "ecs_cluster":
+				resource["aws.ecs.cluster.name"] = v
+			case "ecs_task_arn":
+				resource["aws.ecs.task.arn"] = v
+			case "container_id":
+				resource["container.id"] = v
+			case "container_name":
+				resource["container.name"] = v
+			default:
+				body[k] = v
+			}
+		}
+	}
+	return map[interface{}]interface{}{
+		"resource":          resource,
+		"body":              body,
+		"timestamp":         timestamp,
+		"observedTimestamp": t.UnixMilli(),
+	}
+}
diff --git a/enricher/ecs/ecs_test.go b/enricher/ecs/ecs_test.go
new file mode 100644
index 0000000..f16a342
--- /dev/null
+++ b/enricher/ecs/ecs_test.go
@@ -0,0 +1,80 @@
+package ecs
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher"
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher/mappings"
+)
+
+func TestEnrichRecords(t *testing.T) {
+	type args struct {
+		r map[interface{}]interface{}
+		t time.Time
+	}
+	tests := []struct {
+		name string
+		enr  enricher.IEnricher
+		args args
+		want map[interface{}]interface{}
+	}{
+		{
+			name: "enrich",
+			enr: &Enricher{
+				canvaAWSAccount: "canva_aws_account_val",
+				canvaAppName:    "canva_app_name_val",
+				logGroup:        "log_group_val",
+				ecsTaskFamily:   "ecs_task_family_val",
+				ecsTaskRevision: 10001,
+			},
+			args: args{
+				map[interface{}]interface{}{
+					"ec2_instance_id":     "ec2_instance_id_val",
+					"ecs_cluster":         "ecs_cluster_val",
+					"ecs_task_arn":        "ecs_task_arn_val",
+					"container_id":        "container_id_val",
+					"container_name":      "container_name_val",
+					"other_key_1":         "other_value_1",
+					"other_key_2":         "other_value_2",
+					"other_key_3":         "other_value_3",
+					"timestamp":           "1234567890",
+					"ecs_task_definition": "ecs_task_definition_val",
+				},
+				time.Date(2009, time.November, 10, 23, 7, 5, 432000000, time.UTC),
+			},
+			want: map[interface{}]interface{}{
+				"resource": map[interface{}]interface{}{
+					mappings.RESOURCE_CLOUD_ACCOUNT_ID: "canva_aws_account_val",
+					"service.name":                     "canva_app_name_val",
+					"cloud.platform":                   "aws_ecs",
+					"aws.ecs.launchtype":               "EC2",
+					"aws.ecs.task.family":              "ecs_task_family_val",
+					"aws.ecs.task.revision":            10001,
+					"aws.log.group.names":              "log_group_val",
+					"host.id":                          "ec2_instance_id_val",
+					"aws.ecs.cluster.name":             "ecs_cluster_val",
+					"aws.ecs.task.arn":                 "ecs_task_arn_val",
+					"container.id":                     "container_id_val",
+					"container.name":                   "container_name_val",
+				},
+				"body": map[interface{}]interface{}{
+					"other_key_1": "other_value_1",
+					"other_key_2": "other_value_2",
+					"other_key_3": "other_value_3",
+				},
+				"timestamp":         "1234567890",
+				"observedTimestamp": int64(1257894425432),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.enr.EnrichRecord(tt.args.r, tt.args.t)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("enricher.enrichRecord() = %+v, want %+v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/enricher/eks/eks.go b/enricher/eks/eks.go
new file mode 100644
index 0000000..fad8160
--- /dev/null
+++ b/enricher/eks/eks.go
@@ -0,0 +1,37 @@
+package eks
+
+import (
+	"time"
+
+	"github.com/caarlos0/env/v7"
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher"
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher/mappings"
+)
+
+type Enricher struct {
+	// AWS Account ID
+	AccountId string `env:"CANVA_AWS_ACCOUNT,required"`
+	// Canva Account Group Function
+	CanvaAccountFunction string `env:"CANVA_ACCOUNT_FUNCTION,required"`
+}
+
+func NewEnricher() (*Enricher, error) {
+	enricher := Enricher{}
+	if err := env.Parse(&enricher); err != nil {
+		return nil, err
+	}
+
+	return &enricher, nil
+}
+
+var _ enricher.IEnricher = (*Enricher)(nil)
+
+func (e Enricher) EnrichRecord(r map[interface{}]interface{}, _ time.Time) map[interface{}]interface{} {
+	// add resource attributes
+	r["resource"] = map[interface{}]interface{}{
+		mappings.RESOURCE_CLOUD_ACCOUNT_ID: e.AccountId,
+		mappings.RESOURCE_ACCOUNT_GROUP:    e.CanvaAccountFunction,
+	}
+
+	return r
+}
diff --git a/enricher/eks/eks_test.go b/enricher/eks/eks_test.go
new file mode 100644
index 0000000..1bfc6f0
--- /dev/null
+++ b/enricher/eks/eks_test.go
@@ -0,0 +1,119 @@
+package eks
+
+import (
+	"testing"
+	"time"
+
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher/mappings"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidNewEnricher(t *testing.T) {
+	var cases = []struct {
+		Name     string
+		Env      map[string]string
+		Expected *Enricher
+	}{
+		{
+			Name: "Gets AccountId",
+			Env: map[string]string{
+				mappings.ENV_ACCOUNT_ID:    "1234567890",
+				mappings.ENV_ACCOUNT_GROUP: DummyAccountGroup,
+			},
+			Expected: &Enricher{
+				AccountId:            "1234567890",
+				CanvaAccountFunction: DummyAccountGroup,
+			},
+		},
+		{
+			Name: "Gets Account Group",
+			Env: map[string]string{
+				mappings.ENV_ACCOUNT_ID:    DummyAccountId,
+				mappings.ENV_ACCOUNT_GROUP: "PII",
+			},
+			Expected: &Enricher{
+				AccountId:            DummyAccountId,
+				CanvaAccountFunction: "PII",
+			},
+		},
+	}
+
+	for _, v := range cases {
+		t.Run(v.Name, func(tt *testing.T) {
+			for k, v := range v.Env {
+				tt.Setenv(k, v)
+			}
+			actual, err := NewEnricher()
+
+			assert.NoError(tt, err)
+
+			assert.Equal(tt, v.Expected, actual)
+
+			tt.Cleanup(func() {})
+		})
+	}
+}
+
+func TestInvalidNewEnricher(t *testing.T) {
+	enricher, err := NewEnricher()
+
+	assert.Nil(t, enricher)
+	assert.Error(t, err)
+}
+
+func TestEnrichRecordsWithAccountId(t *testing.T) {
+	var cases = []struct {
+		Name     string
+		Enricher Enricher
+		Input    map[interface{}]interface{}
+		Expected map[interface{}]interface{}
+	}{
+		{
+			Name: "Adds Account Id",
+			Enricher: Enricher{
+				AccountId:            "1234567",
+				CanvaAccountFunction: DummyAccountGroup,
+			},
+			Input: map[interface{}]interface{}{
+				"log": "hello world",
+			},
+			Expected: map[interface{}]interface{}{
+				"log": "hello world",
+				"resource": map[interface{}]interface{}{
+					mappings.RESOURCE_CLOUD_ACCOUNT_ID: "1234567",
+					mappings.RESOURCE_ACCOUNT_GROUP:    DummyAccountGroup,
+				},
+			},
+		},
+		{
+			Name: "Adds Account Group",
+			Enricher: Enricher{
+				AccountId:            DummyAccountId,
+				CanvaAccountFunction: "PII",
+			},
+			Input: map[interface{}]interface{}{
+				"log": "hello world",
+			},
+			Expected: map[interface{}]interface{}{
+				"log": "hello world",
+				"resource": map[interface{}]interface{}{
+					mappings.RESOURCE_CLOUD_ACCOUNT_ID: DummyAccountId,
+					mappings.RESOURCE_ACCOUNT_GROUP:    "PII",
+				},
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.Name, func(tt *testing.T) {
+			actual := c.Enricher.EnrichRecord(c.Input, DummyTime)
+			assert.Equal(tt, c.Expected, actual)
+		})
+	}
+}
+
+var (
+	DummyTime         = time.Now()
+	DummyAccountGroup = "general"
+	DummyAccountId    = "Account_Id"
+)
diff --git a/enricher/enricher.go b/enricher/enricher.go
index f7ffec1..4ad089f 100644
--- a/enricher/enricher.go
+++ b/enricher/enricher.go
@@ -1,117 +1,30 @@
 package enricher
 
-import (
-	"os"
-	"regexp"
-	"strconv"
-	"time"
+import "time"
 
-	"github.com/sirupsen/logrus"
-)
-
-// EnrichRecord modifies existing record.
-func EnrichRecord(r map[interface{}]interface{}, t time.Time) map[interface{}]interface{} {
-	return defaultEnricher.enrichRecord(r, t)
-}
-
-type enricher struct {
-	enable          bool
-	canvaAWSAccount string
-	canvaAppName    string
-	logGroup        string
-	ecsTaskFamily   string
-	ecsTaskRevision int
+type IEnricher interface {
+	EnrichRecord(r map[interface{}]interface{}, t time.Time) map[interface{}]interface{}
 }
 
-var defaultEnricher *enricher
+type Enricher struct {
+	Enable bool
 
-// Init will initialise defaultEnricher instance.
-func Init(enable bool) {
-	if !enable {
-		defaultEnricher = new(enricher)
-		return
-	}
-
-	ecsTaskDefinition := os.Getenv("ECS_TASK_DEFINITION")
-	re := regexp.MustCompile(`^(?P<ecs_task_family>[^ ]*):(?P<ecs_task_revision>[\d]+)$`)
-	ecsTaskDefinitionParts := re.FindStringSubmatch(ecsTaskDefinition)
-	var (
-		ecsTaskFamily   string
-		ecsTaskRevision int
-	)
-	ecsTaskFamilyIndex := re.SubexpIndex("ecs_task_family")
-	ecsTaskRevisionIndex := re.SubexpIndex("ecs_task_revision")
+	enricher IEnricher
+}
 
-	if len(ecsTaskDefinitionParts) >= ecsTaskFamilyIndex {
-		ecsTaskFamily = ecsTaskDefinitionParts[ecsTaskFamilyIndex]
-	}
-	if len(ecsTaskDefinitionParts) >= ecsTaskRevisionIndex {
-		var err error
-		ecsTaskRevision, err = strconv.Atoi(ecsTaskDefinitionParts[re.SubexpIndex("ecs_task_revision")])
-		if err != nil {
-			logrus.Warnf("[kinesis] ecs_task_revision not found for ECS_TASK_DEFINITION=%s", ecsTaskDefinition)
-		}
-	}
+var _ IEnricher = (*Enricher)(nil)
 
-	defaultEnricher = &enricher{
-		enable:          true,
-		canvaAWSAccount: os.Getenv("CANVA_AWS_ACCOUNT"),
-		canvaAppName:    os.Getenv("CANVA_APP_NAME"),
-		logGroup:        os.Getenv("LOG_GROUP"),
-		ecsTaskFamily:   ecsTaskFamily,
-		ecsTaskRevision: ecsTaskRevision,
+func NewEnricher(enable bool, enricher IEnricher) *Enricher {
+	return &Enricher{
+		Enable:   enable,
+		enricher: enricher,
 	}
 }
 
-// enrichRecord modifies existing record.
-func (enr *enricher) enrichRecord(r map[interface{}]interface{}, t time.Time) map[interface{}]interface{} {
-	if !enr.enable {
+func (e *Enricher) EnrichRecord(r map[interface{}]interface{}, t time.Time) map[interface{}]interface{} {
+	if !e.Enable {
 		return r
 	}
 
-	resource := map[interface{}]interface{}{
-		"cloud.account.id":      enr.canvaAWSAccount,
-		"service.name":          enr.canvaAppName,
-		"cloud.platform":        "aws_ecs",
-		"aws.ecs.launchtype":    "EC2",
-		"aws.ecs.task.family":   enr.ecsTaskFamily,
-		"aws.ecs.task.revision": enr.ecsTaskRevision,
-		"aws.log.group.names":   enr.logGroup,
-	}
-	body := make(map[interface{}]interface{})
-
-	var (
-		ok        bool
-		strVal    string
-		timestamp interface{}
-	)
-	for k, v := range r {
-		strVal, ok = k.(string)
-		if ok {
-			switch strVal {
-			case "ecs_task_definition":
-				// Skip
-			case "timestamp":
-				timestamp = v
-			case "ec2_instance_id":
-				resource["host.id"] = v
-			case "ecs_cluster":
-				resource["aws.ecs.cluster.name"] = v
-			case "ecs_task_arn":
-				resource["aws.ecs.task.arn"] = v
-			case "container_id":
-				resource["container.id"] = v
-			case "container_name":
-				resource["container.name"] = v
-			default:
-				body[k] = v
-			}
-		}
-	}
-	return map[interface{}]interface{}{
-		"resource":          resource,
-		"body":              body,
-		"timestamp":         timestamp,
-		"observedTimestamp": t.UnixMilli(),
-	}
+	return e.enricher.EnrichRecord(r, t)
 }
diff --git a/enricher/enricher_test.go b/enricher/enricher_test.go
index 026be01..7bdd7d0 100644
--- a/enricher/enricher_test.go
+++ b/enricher/enricher_test.go
@@ -1,111 +1,62 @@
-package enricher
+package enricher_test
 
 import (
-	"reflect"
 	"testing"
 	"time"
+
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher"
+	"github.com/stretchr/testify/assert"
 )
 
-func Test_enricher_enrichRecord(t *testing.T) {
-	type args struct {
-		r map[interface{}]interface{}
-		t time.Time
-	}
-	tests := []struct {
-		name string
-		enr  *enricher
-		args args
-		want map[interface{}]interface{}
+func TestEnrichRecord(t *testing.T) {
+	var cases = []struct {
+		Name     string
+		Enabled  bool
+		Enricher enricher.IEnricher
+		Input    map[interface{}]interface{}
+		Expected map[interface{}]interface{}
 	}{
 		{
-			name: "enable",
-			enr: &enricher{
-				enable:          true,
-				canvaAWSAccount: "canva_aws_account_val",
-				canvaAppName:    "canva_app_name_val",
-				logGroup:        "log_group_val",
-				ecsTaskFamily:   "ecs_task_family_val",
-				ecsTaskRevision: 10001,
+			Name:     "Disabled",
+			Enabled:  false,
+			Enricher: &DummyEnricher{},
+			Input: map[interface{}]interface{}{
+				"message": "hello world",
 			},
-			args: args{
-				map[interface{}]interface{}{
-					"ec2_instance_id":     "ec2_instance_id_val",
-					"ecs_cluster":         "ecs_cluster_val",
-					"ecs_task_arn":        "ecs_task_arn_val",
-					"container_id":        "container_id_val",
-					"container_name":      "container_name_val",
-					"other_key_1":         "other_value_1",
-					"other_key_2":         "other_value_2",
-					"other_key_3":         "other_value_3",
-					"timestamp":           "1234567890",
-					"ecs_task_definition": "ecs_task_definition_val",
-				},
-				time.Date(2009, time.November, 10, 23, 7, 5, 432000000, time.UTC),
-			},
-			want: map[interface{}]interface{}{
-				"resource": map[interface{}]interface{}{
-					"cloud.account.id":      "canva_aws_account_val",
-					"service.name":          "canva_app_name_val",
-					"cloud.platform":        "aws_ecs",
-					"aws.ecs.launchtype":    "EC2",
-					"aws.ecs.task.family":   "ecs_task_family_val",
-					"aws.ecs.task.revision": 10001,
-					"aws.log.group.names":   "log_group_val",
-					"host.id":               "ec2_instance_id_val",
-					"aws.ecs.cluster.name":  "ecs_cluster_val",
-					"aws.ecs.task.arn":      "ecs_task_arn_val",
-					"container.id":          "container_id_val",
-					"container.name":        "container_name_val",
-				},
-				"body": map[interface{}]interface{}{
-					"other_key_1": "other_value_1",
-					"other_key_2": "other_value_2",
-					"other_key_3": "other_value_3",
-				},
-				"timestamp":         "1234567890",
-				"observedTimestamp": int64(1257894425432),
+			Expected: map[interface{}]interface{}{
+				"message": "hello world",
 			},
 		},
 		{
-			name: "disable",
-			enr: &enricher{
-				enable: false,
-			},
-			args: args{
-				map[interface{}]interface{}{
-					"ec2_instance_id":     "ec2_instance_id_val",
-					"ecs_cluster":         "ecs_cluster_val",
-					"ecs_task_arn":        "ecs_task_arn_val",
-					"container_id":        "container_id_val",
-					"container_name":      "container_name_val",
-					"other_key_1":         "other_value_1",
-					"other_key_2":         "other_value_2",
-					"other_key_3":         "other_value_3",
-					"timestamp":           "1234567890",
-					"ecs_task_definition": "ecs_task_definition_val",
-				},
-				time.Date(2009, time.November, 10, 23, 7, 5, 432000000, time.UTC),
-			},
-			want: map[interface{}]interface{}{
-				"ec2_instance_id":     "ec2_instance_id_val",
-				"ecs_cluster":         "ecs_cluster_val",
-				"ecs_task_arn":        "ecs_task_arn_val",
-				"container_id":        "container_id_val",
-				"container_name":      "container_name_val",
-				"other_key_1":         "other_value_1",
-				"other_key_2":         "other_value_2",
-				"other_key_3":         "other_value_3",
-				"timestamp":           "1234567890",
-				"ecs_task_definition": "ecs_task_definition_val",
+			Name:     "Enabled",
+			Enabled:  true,
+			Enricher: &DummyEnricher{},
+			Input: map[interface{}]interface{}{
+				"message": "hello world",
 			},
+			Expected: DummyRecord,
 		},
 	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := tt.enr.enrichRecord(tt.args.r, tt.args.t)
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("enricher.enrichRecord() = %+v, want %+v", got, tt.want)
-			}
+
+	for _, v := range cases {
+		t.Run(v.Name, func(t *testing.T) {
+			en := enricher.NewEnricher(v.Enabled, v.Enricher)
+
+			actual := en.EnrichRecord(v.Input, DummyTime)
+
+			assert.Equal(t, v.Expected, actual)
 		})
 	}
 }
+
+type DummyEnricher struct{}
+
+func (d DummyEnricher) EnrichRecord(_ map[interface{}]interface{}, _ time.Time) map[interface{}]interface{} {
+	return DummyRecord
+}
+
+var DummyRecord = map[interface{}]interface{}{
+	"message": "I am enriched",
+}
+
+var DummyTime = time.Now()
diff --git a/enricher/mappings/mappings.go b/enricher/mappings/mappings.go
new file mode 100644
index 0000000..43b29c5
--- /dev/null
+++ b/enricher/mappings/mappings.go
@@ -0,0 +1,11 @@
+package mappings
+
+const (
+	RESOURCE_CLOUD_ACCOUNT_ID = "cloud.account.id"
+	RESOURCE_ACCOUNT_GROUP    = "canva.account.function"
+)
+
+const (
+	ENV_ACCOUNT_ID    = "CANVA_AWS_ACCOUNT"
+	ENV_ACCOUNT_GROUP = "CANVA_ACCOUNT_FUNCTION"
+)
diff --git a/fluent-bit-kinesis.go b/fluent-bit-kinesis.go
index 8160920..8c39854 100644
--- a/fluent-bit-kinesis.go
+++ b/fluent-bit-kinesis.go
@@ -15,12 +15,15 @@ package main
 
 import (
 	"C"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 	"unsafe"
 
+	"github.com/canva/amazon-kinesis-streams-for-fluent-bit/enricher/eks"
+
 	"github.com/aws/amazon-kinesis-firehose-for-fluent-bit/plugins"
 	kinesisAPI "github.com/aws/aws-sdk-go/service/kinesis"
 	"github.com/fluent/fluent-bit-go/output"
@@ -40,6 +43,7 @@ const (
 
 var (
 	pluginInstances []*kinesis.OutputPlugin
+	enr             *enricher.Enricher
 )
 
 func addPluginInstance(ctx unsafe.Pointer) error {
@@ -107,6 +111,9 @@ func newKinesisOutput(ctx unsafe.Pointer, pluginID int) (*kinesis.OutputPlugin,
 	enrichRecords := output.FLBPluginConfigKey(ctx, "enrich_records")
 	logrus.Infof("[kinesis %d] plugin parameter enrich_records = %q", pluginID, enrichRecords)
 
+	enrichEKSRecords := output.FLBPluginConfigKey(ctx, "enrich_eks_records")
+	logrus.Infof("[kinesis %d] plugin parameter enrich_eks_records = %q", pluginID, enrichEKSRecords)
+
 	if stream == "" || region == "" {
 		return nil, fmt.Errorf("[kinesis %d] stream and region are required configuration parameters", pluginID)
 	}
@@ -216,12 +223,32 @@ func newKinesisOutput(ctx unsafe.Pointer, pluginID int) (*kinesis.OutputPlugin,
 		}
 	}
 
-	enricherEnable := false
+	var e enricher.IEnricher
+
+	var enricherEnable bool
+	// ECS Enricher
 	if strings.ToLower(enrichRecords) == "true" {
 		enricherEnable = true
 	}
 
-	enricher.Init(enricherEnable)
+	var enricherEksEnable bool
+	// EKS Enricher
+	if strings.ToLower(enrichEKSRecords) == "true" {
+		enricherEksEnable = true
+		e, err = eks.NewEnricher()
+
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if enricherEnable && enricherEksEnable {
+		enr = nil
+		return nil, errors.New("both enrichers cannot be enabled at the same time")
+	}
+
+	enr = enricher.NewEnricher(enricherEnable || enricherEksEnable, e)
+
 	compress.Init(&compress.Config{
 		Format: aggregationCompressionFormat,
 		Level:  aggregationCompressionLevelInt,
@@ -311,7 +338,7 @@ func unpackRecords(kinesisOutput *kinesis.OutputPlugin, data unsafe.Pointer, len
 			timestamp = time.Now()
 		}
 
-		record = enricher.EnrichRecord(record, timestamp)
+		record = enr.EnrichRecord(record, timestamp)
 
 		retCode := kinesisOutput.AddRecord(&records, record, &timestamp)
 		if retCode != output.FLB_OK {
diff --git a/go.mod b/go.mod
index 5b5e0d3..4727418 100644
--- a/go.mod
+++ b/go.mod
@@ -5,6 +5,7 @@ go 1.19
 require (
 	github.com/aws/amazon-kinesis-firehose-for-fluent-bit v1.7.0
 	github.com/aws/aws-sdk-go v1.44.91
+	github.com/caarlos0/env/v7 v7.0.0
 	github.com/fluent/fluent-bit-go v0.0.0-20220311094233-780004bf5562
 	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.2
diff --git a/go.sum b/go.sum
index b8b0b62..130aa36 100644
--- a/go.sum
+++ b/go.sum
@@ -2,6 +2,8 @@ github.com/aws/amazon-kinesis-firehose-for-fluent-bit v1.7.0 h1:EZDlDm9kmBWr8PyT
 github.com/aws/amazon-kinesis-firehose-for-fluent-bit v1.7.0/go.mod h1:260TjZZVmdgzRc/csalI+v+lN7x3qmnmvvl4eAZxfOU=
 github.com/aws/aws-sdk-go v1.44.91 h1:SRWmuX7PTyhBdLuvSfM7KWrWISJsrRsUPcFDSFduRxY=
 github.com/aws/aws-sdk-go v1.44.91/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
+github.com/caarlos0/env/v7 v7.0.0 h1:cyczlTd/zREwSr9ch/mwaDl7Hse7kJuUY8hvHfXu5WI=
+github.com/caarlos0/env/v7 v7.0.0/go.mod h1:LPPWniDUq4JaO6Q41vtlyikhMknqymCLBw0eX4dcH1E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=