diff --git a/README.md b/README.md
index c97f8eab6..bcffbf8a5 100644
--- a/README.md
+++ b/README.md
@@ -11,6 +11,25 @@ additional questions or comments.
 
 Note : The agent version(s) below has dates (ciprod<mmddyyyy>), which indicate the agent build dates (not release dates)
 
+### 03/02/2020 -
+##### Version microsoft/oms:ciprod03022020 Version mcr.microsoft.com/azuremonitor/containerinsights/ciprod:ciprod03022020
+##### Code change log
+- Collection of GPU metrics as InsightsMetrics
+- Enable config map settings to enable collection of 'Normal' kube events
+- Fix kubehealth exceptions to handle empty/nil kube api responses
+- Get resource limits for health and MDM from kubelet instead of kube api
+- Bug fix for windows node image collection where image name contains multiple slashes
+- Exclude ARO master node for data collection
+- Telemetry for kube events flushed
+- Changes to support msi for mdm if service principal doesnt exist
+- Changes for AKS telemetry to ping ods endpoint first and then network check
+- KubeEvents bug fix for KubeEvent type
+
+##### Customer Impact
+- Providing capability for customers to collect 'Normal' kube events using config map
+- Metrics for GPU are collected and ingested to customers workspace if they have GPU enabled nodes 
+- Bug fix for windows container image collection allows customers to get the right data in the ContainerInventory table for windows       containers.
+
 ### 01/07/2020 -
 ##### Version microsoft/oms:ciprod01072020 Version mcr.microsoft.com/azuremonitor/containerinsights/ciprod:ciprod01072020
 ##### Code change log
diff --git a/installer/conf/azm-containers-parser.conf b/installer/conf/azm-containers-parser.conf
new file mode 100644
index 000000000..c0b146143
--- /dev/null
+++ b/installer/conf/azm-containers-parser.conf
@@ -0,0 +1,19 @@
+[PARSER]
+    Name         docker
+    Format       json
+    Time_Key     time
+    Time_Format  %Y-%m-%dT%H:%M:%S.%L
+    Time_Keep    On
+    # Command      |  Decoder | Field | Optional Action
+    # =============|==================|=================
+    Decode_Field_As   escaped    log
+
+[PARSER]
+    # http://rubular.com/r/tjUt3Awgg4
+    Name cri
+    Format regex
+    Regex ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<log>.*)$
+    Time_Key    time
+    Time_Format %Y-%m-%dT%H:%M:%S.%L
+    Time_Keep    On   
+    
\ No newline at end of file
diff --git a/installer/conf/td-agent-bit-rs.conf b/installer/conf/td-agent-bit-rs.conf
index 7839b0eee..55ab6e81a 100644
--- a/installer/conf/td-agent-bit-rs.conf
+++ b/installer/conf/td-agent-bit-rs.conf
@@ -1,7 +1,7 @@
 [SERVICE]
     Flush         30
     Log_Level     info
-    Parsers_File  /etc/td-agent-bit/parsers.conf
+    Parsers_File  /etc/opt/microsoft/docker-cimprov/azm-containers-parser.conf
     Log_File      /var/opt/microsoft/docker-cimprov/log/fluent-bit.log
 
 [INPUT]
diff --git a/installer/conf/td-agent-bit.conf b/installer/conf/td-agent-bit.conf
index 6a1bf3e3e..5007c18db 100644
--- a/installer/conf/td-agent-bit.conf
+++ b/installer/conf/td-agent-bit.conf
@@ -2,7 +2,7 @@
     #Default service flush interval is 15 seconds
     ${SERVICE_FLUSH_INTERVAL}
     Log_Level     info
-    Parsers_File  /etc/td-agent-bit/parsers.conf
+    Parsers_File  /etc/opt/microsoft/docker-cimprov/azm-containers-parser.conf
     Log_File      /var/opt/microsoft/docker-cimprov/log/fluent-bit.log
 
 [INPUT]
diff --git a/installer/conf/telegraf-rs.conf b/installer/conf/telegraf-rs.conf
index 3450ab88f..64b43d0a9 100644
--- a/installer/conf/telegraf-rs.conf
+++ b/installer/conf/telegraf-rs.conf
@@ -56,7 +56,7 @@
 
   ## Default flushing interval for all outputs. You shouldn't set this below
   ## interval. Maximum flush_interval will be flush_interval + flush_jitter
-  flush_interval = "60s"
+  flush_interval = "15s"
   ## Jitter the flush interval by a random amount. This is primarily to avoid
   ## large write spikes for users running a large number of telegraf instances.
   ## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
diff --git a/installer/conf/telegraf.conf b/installer/conf/telegraf.conf
index f9dc3fb6a..b5c4cb64c 100644
--- a/installer/conf/telegraf.conf
+++ b/installer/conf/telegraf.conf
@@ -56,7 +56,7 @@
 
   ## Default flushing interval for all outputs. You shouldn't set this below
   ## interval. Maximum flush_interval will be flush_interval + flush_jitter
-  flush_interval = "60s"
+  flush_interval = "15s"
   ## Jitter the flush interval by a random amount. This is primarily to avoid
   ## large write spikes for users running a large number of telegraf instances.
   ## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
@@ -386,7 +386,7 @@
 #  report_active = true
 #  fieldpass = ["usage_active","cluster","node","host","device"]
 #  taginclude = ["cluster","cpu","node"]
-  
+
 
 
 # Read metrics about disk usage by mount point
@@ -395,7 +395,7 @@
   ## By default stats will be gathered for all mount points.
   ## Set mount_points will restrict the stats to only the specified mount points.
   # mount_points = ["/"]
-  
+
   ## Ignore mount points by filesystem type.
   ignore_fs = ["tmpfs", "devtmpfs", "devfs", "overlay", "aufs", "squashfs"]
   fieldpass = ["free", "used", "used_percent"]
@@ -532,7 +532,8 @@
   name_prefix="container.azm.ms/"
   ## An array of urls to scrape metrics from.
   urls = ["$CADVISOR_METRICS_URL"]
-  fieldpass = ["kubelet_docker_operations", "kubelet_docker_operations_errors"]
+  ## Include "$KUBELET_RUNTIME_OPERATIONS_TOTAL_METRIC", "$KUBELET_RUNTIME_OPERATIONS_ERRORS_TOTAL_METRIC" when we add for support for 1.18
+  fieldpass = ["$KUBELET_RUNTIME_OPERATIONS_METRIC", "$KUBELET_RUNTIME_OPERATIONS_ERRORS_METRIC"]
 
   metric_version = 2
   url_tag = "scrapeUrl"
@@ -578,7 +579,7 @@
   urls = $AZMON_DS_PROM_URLS
 
   fieldpass = $AZMON_DS_PROM_FIELDPASS
-  
+
   fielddrop = $AZMON_DS_PROM_FIELDDROP
 
   metric_version = 2
diff --git a/installer/datafiles/base_container.data b/installer/datafiles/base_container.data
index e011dddf9..a7fe8dc69 100644
--- a/installer/datafiles/base_container.data
+++ b/installer/datafiles/base_container.data
@@ -45,6 +45,8 @@ MAINTAINER:              'Microsoft Corporation'
 /opt/microsoft/omsagent/plugin/DockerApiClient.rb;												source/code/plugin/DockerApiClient.rb;				644; root; root
 /opt/microsoft/omsagent/plugin/DockerApiRestHelper.rb;											source/code/plugin/DockerApiRestHelper.rb;			644; root; root
 /opt/microsoft/omsagent/plugin/in_containerinventory.rb;										source/code/plugin/in_containerinventory.rb;		644; root; root
+/opt/microsoft/omsagent/plugin/kubernetes_container_inventory.rb;							source/code/plugin/kubernetes_container_inventory.rb;		644; root; root
+
 
 /opt/microsoft/omsagent/plugin/out_mdm.rb;                                                  source/code/plugin/out_mdm.rb; 644; root; root
 /opt/microsoft/omsagent/plugin/filter_cadvisor2mdm.rb;                                      source/code/plugin/filter_cadvisor2mdm.rb; 644; root; root
@@ -106,6 +108,7 @@ MAINTAINER:              'Microsoft Corporation'
 /opt/td-agent-bit/bin/out_oms.so;                                intermediate/${{BUILD_CONFIGURATION}}/out_oms.so;  	755; root; root
 /etc/opt/microsoft/docker-cimprov/td-agent-bit.conf;			    installer/conf/td-agent-bit.conf;                    644; root; root
 /etc/opt/microsoft/docker-cimprov/td-agent-bit-rs.conf;			    installer/conf/td-agent-bit-rs.conf;                    644; root; root
+/etc/opt/microsoft/docker-cimprov/azm-containers-parser.conf;	    installer/conf/azm-containers-parser.conf;                    644; root; root
 /etc/opt/microsoft/docker-cimprov/out_oms.conf;			            installer/conf/out_oms.conf;                    644; root; root
 /etc/opt/microsoft/docker-cimprov/telegraf.conf;			        installer/conf/telegraf.conf;                    644; root; root
 /etc/opt/microsoft/docker-cimprov/telegraf-rs.conf;			        installer/conf/telegraf-rs.conf;                    644; root; root
diff --git a/source/code/go/src/plugins/glide.lock b/source/code/go/src/plugins/glide.lock
index fc147fe74..b322898e5 100644
--- a/source/code/go/src/plugins/glide.lock
+++ b/source/code/go/src/plugins/glide.lock
@@ -1,6 +1,8 @@
-hash: a6a873d09ed9c3d890a70122e61efba992ead9850fe48f6fcb020d86800d4ade
-updated: 2018-10-10T13:37:51.9703908-07:00
+hash: b00ffe7239c0f8d9d87e0afe8512c75ff9952c41cd2b3b79dc30fe6d3f0e3ec8
+updated: 2020-03-17T15:46:37.8805517-07:00
 imports:
+- name: code.cloudfoundry.org/clock
+  version: 86534f4ca3a5cbac1685968132dfaefeb8e8cad2
 - name: github.com/fluent/fluent-bit-go
   version: c4a158a6e3a793166c6ecfa2d5c80d71eada8959
   subpackages:
@@ -26,6 +28,8 @@ imports:
   version: 7d79101e329e5a3adf994758c578dab82b90c017
 - name: github.com/google/gofuzz
   version: 44d81051d367757e1c7c6a5a86423ece9afcf63c
+- name: github.com/google/uuid
+  version: 0cd6bf5da1e1c83f8b45653022c74f71af0538a4
 - name: github.com/googleapis/gnostic
   version: 0c5108395e2debce0d731cf0287ddf7242066aba
   subpackages:
@@ -42,12 +46,15 @@ imports:
   version: d2df5d440eda5372f24fcac03839a64d6cb5f7e5
   subpackages:
   - appinsights
+  - appinsights/contracts
 - name: github.com/modern-go/concurrent
   version: bacd9c7ef1dd9b15be4a9909b8ac7a4e313eec94
 - name: github.com/modern-go/reflect2
   version: 05fbef0ca5da472bbf96c9322b84a53edc03c9fd
 - name: github.com/peterbourgon/diskv
   version: 5f041e8faa004a95c88a202771f4cc3e991971e6
+- name: github.com/satori/go.uuid
+  version: b2ce2384e17bbe0c6d34077efa39dbab3e09123b
 - name: github.com/ugorji/go
   version: 00b869d2f4a5e27445c2d916fa106fc72c106d4c
   subpackages:
diff --git a/source/code/go/src/plugins/glide.yaml b/source/code/go/src/plugins/glide.yaml
index b2829391b..9d0f9d2ac 100644
--- a/source/code/go/src/plugins/glide.yaml
+++ b/source/code/go/src/plugins/glide.yaml
@@ -3,6 +3,8 @@ import:
 - package: github.com/fluent/fluent-bit-go
   subpackages:
   - output
+- package: github.com/google/uuid
+  version: ^1.1.0
 - package: gopkg.in/natefinch/lumberjack.v2
   version: ^2.1.0
 - package: k8s.io/apimachinery
diff --git a/source/code/go/src/plugins/oms.go b/source/code/go/src/plugins/oms.go
index 8dfaf0e7e..3dfe539e2 100644
--- a/source/code/go/src/plugins/oms.go
+++ b/source/code/go/src/plugins/oms.go
@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/fluent/fluent-bit-go/output"
+	"github.com/google/uuid"
 
 	lumberjack "gopkg.in/natefinch/lumberjack.v2"
 
@@ -37,6 +38,9 @@ const ResourceIdEnv = "AKS_RESOURCE_ID"
 //env variable which has ResourceName for NON-AKS
 const ResourceNameEnv = "ACS_RESOURCE_NAME"
 
+//env variable which has container run time name
+const ContainerRuntimeEnv = "CONTAINER_RUNTIME"
+
 // Origin prefix for telegraf Metrics (used as prefix for origin field & prefix for azure monitor specific tags and also for custom-metrics telemetry )
 const TelegrafMetricOriginPrefix = "container.azm.ms"
 
@@ -93,7 +97,9 @@ var (
 	//KubeMonAgentEvents skip first flush
 	skipKubeMonEventsFlush bool
 	// enrich container logs (when true this will add the fields - timeofcommand, containername & containerimage)
-	enrichContainerLogs bool
+	enrichContainerLogs bool		
+	// container runtime engine configured on the kubelet
+	containerRuntime string
 )
 
 var (
@@ -133,6 +139,12 @@ var (
 	Log = FLBLogger.Printf
 )
 
+var (
+	dockerCimprovVersion = "9.0.0.0"
+	agentName = "ContainerAgent"
+	userAgent = ""
+)
+
 // DataItem represents the object corresponding to the json that is sent by fluentbit tail plugin
 type DataItem struct {
 	LogEntry              string `json:"LogEntry"`
@@ -513,6 +525,9 @@ func flushKubeMonAgentEventRecords() {
 				} else {
 					req, _ := http.NewRequest("POST", OMSEndpoint, bytes.NewBuffer(marshalled))
 					req.Header.Set("Content-Type", "application/json")
+					req.Header.Set("User-Agent", userAgent )
+					reqId := uuid.New().String()
+					req.Header.Set("X-Request-ID", reqId)
 					//expensive to do string len for every request, so use a flag
 					if ResourceCentric == true {
 						req.Header.Set("x-ms-AzureResourceId", ResourceID)
@@ -527,7 +542,7 @@ func flushKubeMonAgentEventRecords() {
 						Log("Failed to flush %d records after %s", len(laKubeMonAgentEventsRecords), elapsed)
 					} else if resp == nil || resp.StatusCode != 200 {
 						if resp != nil {
-							Log("Status %s Status Code %d", resp.Status, resp.StatusCode)
+							Log(" RequestId %s Status %s Status Code %d", reqId, resp.Status, resp.StatusCode)
 						}
 						Log("Failed to flush %d records after %s", len(laKubeMonAgentEventsRecords), elapsed)
 					} else {
@@ -656,6 +671,9 @@ func PostTelegrafMetricsToLA(telegrafRecords []map[interface{}]interface{}) int
 
 	//set headers
 	req.Header.Set("x-ms-date", time.Now().Format(time.RFC3339))
+	req.Header.Set("User-Agent", userAgent )
+	reqId := uuid.New().String()
+	req.Header.Set("X-Request-ID", reqId)
 
 	//expensive to do string len for every request, so use a flag
 	if ResourceCentric == true {
@@ -669,31 +687,34 @@ func PostTelegrafMetricsToLA(telegrafRecords []map[interface{}]interface{}) int
 	if err != nil {
 		message := fmt.Sprintf("PostTelegrafMetricsToLA::Error:(retriable) when sending %v metrics. duration:%v err:%q \n", len(laMetrics), elapsed, err.Error())
 		Log(message)
-		UpdateNumTelegrafMetricsSentTelemetry(0, 1)
+		UpdateNumTelegrafMetricsSentTelemetry(0, 1, 0)
 		return output.FLB_RETRY
 	}
 
 	if resp == nil || resp.StatusCode != 200 {
 		if resp != nil {
-			Log("PostTelegrafMetricsToLA::Error:(retriable) Response Status %v Status Code %v", resp.Status, resp.StatusCode)
+			Log("PostTelegrafMetricsToLA::Error:(retriable) RequestID %s Response Status %v Status Code %v", reqId, resp.Status, resp.StatusCode)
+		}
+		if resp != nil && resp.StatusCode == 429 {
+			UpdateNumTelegrafMetricsSentTelemetry(0, 1, 1)
 		}
-		UpdateNumTelegrafMetricsSentTelemetry(0, 1)
 		return output.FLB_RETRY
 	}
 
 	defer resp.Body.Close()
 
 	numMetrics := len(laMetrics)
-	UpdateNumTelegrafMetricsSentTelemetry(numMetrics, 0)
+	UpdateNumTelegrafMetricsSentTelemetry(numMetrics, 0, 0)
 	Log("PostTelegrafMetricsToLA::Info:Successfully flushed %v records in %v", numMetrics, elapsed)
 
 	return output.FLB_OK
 }
 
-func UpdateNumTelegrafMetricsSentTelemetry(numMetricsSent int, numSendErrors int) {
+func UpdateNumTelegrafMetricsSentTelemetry(numMetricsSent int, numSendErrors int, numSend429Errors int) {
 	ContainerLogTelemetryMutex.Lock()
 	TelegrafMetricsSentCount += float64(numMetricsSent)
 	TelegrafMetricsSendErrorCount += float64(numSendErrors)
+	TelegrafMetricsSend429ErrorCount += float64(numSend429Errors)
 	ContainerLogTelemetryMutex.Unlock()
 }
 
@@ -734,9 +755,11 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 
 		stringMap := make(map[string]string)
 
-		stringMap["LogEntry"] = ToString(record["log"])
+		logEntry := ToString(record["log"])
+		logEntryTimeStamp := ToString(record["time"])			
+		stringMap["LogEntry"] = logEntry
 		stringMap["LogEntrySource"] = logEntrySource
-		stringMap["LogEntryTimeStamp"] = ToString(record["time"])
+		stringMap["LogEntryTimeStamp"] = logEntryTimeStamp
 		stringMap["SourceSystem"] = "Containers"
 		stringMap["Id"] = containerID
 
@@ -781,7 +804,7 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 		if dataItem.LogEntryTimeStamp != "" {
 			loggedTime, e := time.Parse(time.RFC3339, dataItem.LogEntryTimeStamp)
 			if e != nil {
-				message := fmt.Sprintf("Error while converting LogEntryTimeStamp for telemetry purposes: %s", e.Error())
+				message := fmt.Sprintf("containerId: %s Error while converting LogEntryTimeStamp for telemetry purposes: %s", dataItem.ID, e.Error())
 				Log(message)
 				SendException(message)
 			} else {
@@ -810,6 +833,9 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 
 		req, _ := http.NewRequest("POST", OMSEndpoint, bytes.NewBuffer(marshalled))
 		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("User-Agent", userAgent )
+		reqId := uuid.New().String()
+		req.Header.Set("X-Request-ID", reqId)
 		//expensive to do string len for every request, so use a flag
 		if ResourceCentric == true {
 			req.Header.Set("x-ms-AzureResourceId", ResourceID)
@@ -830,7 +856,7 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 
 		if resp == nil || resp.StatusCode != 200 {
 			if resp != nil {
-				Log("Status %s Status Code %d", resp.Status, resp.StatusCode)
+				Log("RequestId %s Status %s Status Code %d", reqId, resp.Status, resp.StatusCode)
 			}
 			return output.FLB_RETRY
 		}
@@ -958,6 +984,21 @@ func InitializePlugin(pluginConfPath string, agentVersion string) {
 		Log("ResourceID=%s", ResourceID)
 		Log("ResourceName=%s", ResourceName)
 	}
+	
+	// log runtime info for debug purpose
+	containerRuntime = os.Getenv(ContainerRuntimeEnv)		
+	Log("Container Runtime engine %s", containerRuntime)	
+	
+
+	//set useragent to be used by ingestion 
+	docker_cimprov_version := strings.TrimSpace(os.Getenv("DOCKER_CIMPROV_VERSION"))
+	if len(docker_cimprov_version) > 0 {
+		dockerCimprovVersion = docker_cimprov_version
+	}
+
+	userAgent = fmt.Sprintf("%s/%s", agentName, dockerCimprovVersion)
+
+	Log("Usage-Agent = %s \n", userAgent)
 
 	// Initialize image,name map refresh ticker
 	containerInventoryRefreshInterval, err := strconv.Atoi(pluginConfig["container_inventory_refresh_interval"])
diff --git a/source/code/go/src/plugins/telemetry.go b/source/code/go/src/plugins/telemetry.go
index d5675187f..b31bac1c7 100644
--- a/source/code/go/src/plugins/telemetry.go
+++ b/source/code/go/src/plugins/telemetry.go
@@ -34,6 +34,8 @@ var (
 	TelegrafMetricsSentCount float64
 	//Tracks the number of send errors between telemetry ticker periods (uses ContainerLogTelemetryTicker)
 	TelegrafMetricsSendErrorCount float64
+    //Tracks the number of 429 (throttle) errors between telemetry ticker periods (uses ContainerLogTelemetryTicker)
+	TelegrafMetricsSend429ErrorCount float64
 )
 
 const (
@@ -49,6 +51,7 @@ const (
 	metricNameAgentLogProcessingMaxLatencyMs          = "ContainerLogsAgentSideLatencyMs"
 	metricNameNumberofTelegrafMetricsSentSuccessfully = "TelegrafMetricsSentCount"
 	metricNameNumberofSendErrorsTelegrafMetrics       = "TelegrafMetricsSendErrorCount"
+	metricNameNumberofSend429ErrorsTelegrafMetrics    = "TelegrafMetricsSend429ErrorCount"
 
 	defaultTelemetryPushIntervalSeconds = 300
 
@@ -78,8 +81,10 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 		logSizeRate := FlushedRecordsSize / float64(elapsed/time.Second)
 		telegrafMetricsSentCount := TelegrafMetricsSentCount
 		telegrafMetricsSendErrorCount := TelegrafMetricsSendErrorCount
+		telegrafMetricsSend429ErrorCount := TelegrafMetricsSend429ErrorCount
 		TelegrafMetricsSentCount = 0.0
 		TelegrafMetricsSendErrorCount = 0.0
+		TelegrafMetricsSend429ErrorCount = 0.0
 		FlushedRecordsCount = 0.0
 		FlushedRecordsSize = 0.0
 		FlushedRecordsTimeTaken = 0.0
@@ -103,7 +108,12 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 			TelemetryClient.Track(logLatencyMetric)
 		}
 		TelemetryClient.Track(appinsights.NewMetricTelemetry(metricNameNumberofTelegrafMetricsSentSuccessfully, telegrafMetricsSentCount))
-		TelemetryClient.Track(appinsights.NewMetricTelemetry(metricNameNumberofSendErrorsTelegrafMetrics, telegrafMetricsSendErrorCount))
+		if telegrafMetricsSendErrorCount > 0.0 {
+			TelemetryClient.Track(appinsights.NewMetricTelemetry(metricNameNumberofSendErrorsTelegrafMetrics, telegrafMetricsSendErrorCount))
+		}
+		if telegrafMetricsSend429ErrorCount > 0.0 {
+			TelemetryClient.Track(appinsights.NewMetricTelemetry(metricNameNumberofSend429ErrorsTelegrafMetrics, telegrafMetricsSend429ErrorCount))
+		}
 		start = time.Now()
 	}
 }
diff --git a/source/code/plugin/ApplicationInsightsUtility.rb b/source/code/plugin/ApplicationInsightsUtility.rb
index f7bd806a0..27f048520 100644
--- a/source/code/plugin/ApplicationInsightsUtility.rb
+++ b/source/code/plugin/ApplicationInsightsUtility.rb
@@ -6,7 +6,7 @@ class ApplicationInsightsUtility
   require_relative "omslog"
   require_relative "DockerApiClient"
   require_relative "oms_common"
-  require 'yajl/json_gem'
+  require "yajl/json_gem"
   require "base64"
 
   @@HeartBeat = "HeartBeatEvent"
@@ -20,6 +20,7 @@ class ApplicationInsightsUtility
   @@EnvApplicationInsightsKey = "APPLICATIONINSIGHTS_AUTH"
   @@EnvApplicationInsightsEndpoint = "APPLICATIONINSIGHTS_ENDPOINT"
   @@EnvControllerType = "CONTROLLER_TYPE"
+  @@EnvContainerRuntime = "CONTAINER_RUNTIME"
 
   @@CustomProperties = {}
   @@Tc = nil
@@ -73,7 +74,7 @@ def initializeUtility()
           @@Tc = ApplicationInsights::TelemetryClient.new
         elsif !encodedAppInsightsKey.nil?
           decodedAppInsightsKey = Base64.decode64(encodedAppInsightsKey)
-          
+
           #override ai endpoint if its available otherwise use default.
           if appInsightsEndpoint && !appInsightsEndpoint.nil? && !appInsightsEndpoint.empty?
             $log.info("AppInsightsUtility: Telemetry client uses overrided endpoint url : #{appInsightsEndpoint}")
@@ -110,11 +111,17 @@ def initializeUtility()
       end
     end
 
-    def getDockerInfo()
-      dockerInfo = DockerApiClient.dockerInfo
-      if (!dockerInfo.nil? && !dockerInfo.empty?)
-        @@CustomProperties["DockerVersion"] = dockerInfo["Version"]
-        #@@CustomProperties["DockerApiVersion"] = dockerInfo["ApiVersion"]
+    def getContainerRuntimeInfo()
+      containerRuntime = ENV[@@EnvContainerRuntime]
+      if !containerRuntime.nil? && !containerRuntime.empty?
+        # DockerVersion field holds either containerRuntime for non-docker or Dockerversion if its docker
+        @@CustomProperties["DockerVersion"] = containerRuntime
+        if containerRuntime.casecmp("docker") == 0
+          dockerInfo = DockerApiClient.dockerInfo
+          if (!dockerInfo.nil? && !dockerInfo.empty?)
+            @@CustomProperties["DockerVersion"] = dockerInfo["Version"]          
+          end
+        end
       end
     end
 
@@ -169,7 +176,7 @@ def sendExceptionTelemetry(errorStr, properties = nil)
         if @@CustomProperties.empty? || @@CustomProperties.nil?
           initializeUtility()
         elsif @@CustomProperties["DockerVersion"].nil?
-          getDockerInfo()
+          getContainerRuntimeInfo()
         end
         telemetryProps = {}
         # add common dimensions
@@ -193,7 +200,7 @@ def sendTelemetry(pluginName, properties)
         if @@CustomProperties.empty? || @@CustomProperties.nil?
           initializeUtility()
         elsif @@CustomProperties["DockerVersion"].nil?
-          getDockerInfo()
+          getContainerRuntimeInfo()
         end
         @@CustomProperties["Computer"] = properties["Computer"]
         sendHeartBeatEvent(pluginName)
@@ -213,7 +220,7 @@ def sendMetricTelemetry(metricName, metricValue, properties)
         if @@CustomProperties.empty? || @@CustomProperties.nil?
           initializeUtility()
         elsif @@CustomProperties["DockerVersion"].nil?
-          getDockerInfo()
+          getContainerRuntimeInfo()
         end
         telemetryProps = {}
         # add common dimensions
diff --git a/source/code/plugin/CAdvisorMetricsAPIClient.rb b/source/code/plugin/CAdvisorMetricsAPIClient.rb
index 53139ea4e..bdd7750f5 100644
--- a/source/code/plugin/CAdvisorMetricsAPIClient.rb
+++ b/source/code/plugin/CAdvisorMetricsAPIClient.rb
@@ -73,6 +73,11 @@ def getNodeCapacityFromCAdvisor(winNode: nil)
       return getResponse(winNode, relativeUri)
     end
 
+    def getPodsFromCAdvisor(winNode: nil)
+      relativeUri = "/pods"
+      return getResponse(winNode, relativeUri)
+    end
+
     def getBaseCAdvisorUri(winNode)
         cAdvisorSecurePort = isCAdvisorOnSecurePort()
 
diff --git a/source/code/plugin/in_containerinventory.rb b/source/code/plugin/in_containerinventory.rb
index d107047b4..c1126aa4e 100644
--- a/source/code/plugin/in_containerinventory.rb
+++ b/source/code/plugin/in_containerinventory.rb
@@ -5,20 +5,17 @@ module Fluent
   class Container_Inventory_Input < Input
     Plugin.register_input("containerinventory", self)
 
-    @@PluginName = "ContainerInventory"
-    @@RunningState = "Running"
-    @@FailedState = "Failed"
-    @@StoppedState = "Stopped"
-    @@PausedState = "Paused"
+    @@PluginName = "ContainerInventory"   
 
     def initialize
       super
-      require 'yajl/json_gem'
-      require "time"
-      require_relative "DockerApiClient"
+      require "yajl/json_gem"
+      require "time"      
       require_relative "ContainerInventoryState"
       require_relative "ApplicationInsightsUtility"
       require_relative "omslog"
+      require_relative "CAdvisorMetricsAPIClient"
+      require_relative "kubernetes_container_inventory"      
     end
 
     config_param :run_interval, :time, :default => 60
@@ -46,213 +43,78 @@ def shutdown
         }
         @thread.join
       end
-    end
-
-    def obtainContainerConfig(instance, container, clusterCollectEnvironmentVar)
-      begin
-        configValue = container["Config"]
-        if !configValue.nil?
-          instance["ContainerHostname"] = configValue["Hostname"]
-          # Check to see if the environment variable collection is disabled at the cluster level - This disables env variable collection for all containers.
-          if !clusterCollectEnvironmentVar.nil? && !clusterCollectEnvironmentVar.empty? && clusterCollectEnvironmentVar.casecmp("false") == 0
-            instance["EnvironmentVar"] = ["AZMON_CLUSTER_COLLECT_ENV_VAR=FALSE"]
-          else
-            envValue = configValue["Env"]
-            envValueString = (envValue.nil?) ? "" : envValue.to_s
-            # Skip environment variable processing if it contains the flag AZMON_COLLECT_ENV=FALSE
-            # Check to see if the environment variable collection is disabled for this container.
-            if /AZMON_COLLECT_ENV=FALSE/i.match(envValueString)
-              envValueString = ["AZMON_COLLECT_ENV=FALSE"]
-              $log.warn("Environment Variable collection for container: #{container["Id"]} skipped because AZMON_COLLECT_ENV is set to false")
-            end
-            # Restricting the ENV string value to 200kb since the size of this string can go very high
-            if envValueString.length > 200000
-              envValueStringTruncated = envValueString.slice(0..200000)
-              lastIndex = envValueStringTruncated.rindex("\", ")
-              if !lastIndex.nil?
-                envValueStringTruncated = envValueStringTruncated.slice(0..lastIndex) + "]"
-              end
-              instance["EnvironmentVar"] = envValueStringTruncated
-            else
-              instance["EnvironmentVar"] = envValueString
-            end
-          end
-
-          cmdValue = configValue["Cmd"]
-          cmdValueString = (cmdValue.nil?) ? "" : cmdValue.to_s
-          instance["Command"] = cmdValueString
-
-          instance["ComposeGroup"] = ""
-          labelsValue = configValue["Labels"]
-          if !labelsValue.nil? && !labelsValue.empty?
-            instance["ComposeGroup"] = labelsValue["com.docker.compose.project"]
-          end
-        else
-          $log.warn("Attempt in ObtainContainerConfig to get container: #{container["Id"]} config information returned null")
-        end
-      rescue => errorStr
-        $log.warn("Exception in obtainContainerConfig: #{errorStr}")
-      end
-    end
-
-    def obtainContainerState(instance, container)
-      begin
-        stateValue = container["State"]
-        if !stateValue.nil?
-          exitCodeValue = stateValue["ExitCode"]
-          # Exit codes less than 0 are not supported by the engine
-          if exitCodeValue < 0
-            exitCodeValue = 128
-            $log.info("obtainContainerState::Container: #{container["Id"]} returned negative exit code")
-          end
-          instance["ExitCode"] = exitCodeValue
-          if exitCodeValue > 0
-            instance["State"] = @@FailedState
-          else
-            # Set the Container status : Running/Paused/Stopped
-            runningValue = stateValue["Running"]
-            if runningValue
-              pausedValue = stateValue["Paused"]
-              # Checking for paused within running is true state because docker returns true for both Running and Paused fields when the container is paused
-              if pausedValue
-                instance["State"] = @@PausedState
-              else
-                instance["State"] = @@RunningState
-              end
-            else
-              instance["State"] = @@StoppedState
-            end
-          end
-          instance["StartedTime"] = stateValue["StartedAt"]
-          instance["FinishedTime"] = stateValue["FinishedAt"]
-        else
-          $log.info("Attempt in ObtainContainerState to get container: #{container["Id"]} state information returned null")
-        end
-      rescue => errorStr
-        $log.warn("Exception in obtainContainerState: #{errorStr}")
-      end
-    end
-
-    def obtainContainerHostConfig(instance, container)
-      begin
-        hostConfig = container["HostConfig"]
-        if !hostConfig.nil?
-          links = hostConfig["Links"]
-          instance["Links"] = ""
-          if !links.nil?
-            linksString = links.to_s
-            instance["Links"] = (linksString == "null") ? "" : linksString
-          end
-          portBindings = hostConfig["PortBindings"]
-          instance["Ports"] = ""
-          if !portBindings.nil?
-            portBindingsString = portBindings.to_s
-            instance["Ports"] = (portBindingsString == "null") ? "" : portBindingsString
-          end
-        else
-          $log.info("Attempt in ObtainContainerHostConfig to get container: #{container["Id"]} host config information returned null")
-        end
-      rescue => errorStr
-        $log.warn("Exception in obtainContainerHostConfig: #{errorStr}")
-      end
-    end
-
-    def inspectContainer(id, nameMap, clusterCollectEnvironmentVar)
-      containerInstance = {}
-      begin
-        container = DockerApiClient.dockerInspectContainer(id)
-        if !container.nil? && !container.empty?
-          containerInstance["InstanceID"] = container["Id"]
-          containerInstance["CreatedTime"] = container["Created"]
-          containerName = container["Name"]
-          if !containerName.nil? && !containerName.empty?
-            # Remove the leading / from the name if it exists (this is an API issue)
-            containerInstance["ElementName"] = (containerName[0] == "/") ? containerName[1..-1] : containerName
-          end
-          imageValue = container["Image"]
-          if !imageValue.nil? && !imageValue.empty?
-            repoImageTagArray = nameMap[imageValue]
-            if nameMap.has_key? imageValue
-              containerInstance["Repository"] = repoImageTagArray[0]
-              containerInstance["Image"] = repoImageTagArray[1]
-              containerInstance["ImageTag"] = repoImageTagArray[2]
-              # Setting the image id to the id in the remote repository
-              containerInstance["ImageId"] = repoImageTagArray[3]
-            end
-          end
-          obtainContainerConfig(containerInstance, container, clusterCollectEnvironmentVar)
-          obtainContainerState(containerInstance, container)
-          obtainContainerHostConfig(containerInstance, container)
-        end
-      rescue => errorStr
-        $log.warn("Exception in inspectContainer: #{errorStr} for container: #{id}")
-      end
-      return containerInstance
-    end
-
+    end   
+  
     def enumerate
       currentTime = Time.now
       emitTime = currentTime.to_f
       batchTime = currentTime.utc.iso8601
       containerInventory = Array.new
+      eventStream = MultiEventStream.new
+      hostName = ""
       $log.info("in_container_inventory::enumerate : Begin processing @ #{Time.now.utc.iso8601}")
-      hostname = DockerApiClient.getDockerHostName
       begin
-        containerIds = DockerApiClient.listContainers
-        if !containerIds.nil? && !containerIds.empty?
-          eventStream = MultiEventStream.new
-          nameMap = DockerApiClient.getImageIdMap
-          clusterCollectEnvironmentVar = ENV["AZMON_CLUSTER_COLLECT_ENV_VAR"]
-          if !clusterCollectEnvironmentVar.nil? && !clusterCollectEnvironmentVar.empty? && clusterCollectEnvironmentVar.casecmp("false") == 0
-            $log.warn("Environment Variable collection disabled for cluster")
-          end
-          containerIds.each do |containerId|
-            inspectedContainer = {}
-            inspectedContainer = inspectContainer(containerId, nameMap, clusterCollectEnvironmentVar)
-            inspectedContainer["Computer"] = hostname
-            inspectedContainer["CollectionTime"] = batchTime #This is the time that is mapped to become TimeGenerated
-            containerInventory.push inspectedContainer
-            ContainerInventoryState.writeContainerState(inspectedContainer)
-          end
-          # Update the state for deleted containers
-          deletedContainers = ContainerInventoryState.getDeletedContainers(containerIds)
-          if !deletedContainers.nil? && !deletedContainers.empty?
-            deletedContainers.each do |deletedContainer|
-              container = ContainerInventoryState.readContainerState(deletedContainer)
-              if !container.nil?
-                container.each { |k, v| container[k] = v }
-                container["State"] = "Deleted"
-                containerInventory.push container
+        containerRuntimeEnv = ENV["CONTAINER_RUNTIME"]
+        $log.info("in_container_inventory::enumerate : container runtime : #{containerRuntimeEnv}")
+        clusterCollectEnvironmentVar = ENV["AZMON_CLUSTER_COLLECT_ENV_VAR"]
+        $log.info("in_container_inventory::enumerate : using cadvisor apis")                    
+        containerIds = Array.new
+        response = CAdvisorMetricsAPIClient.getPodsFromCAdvisor(winNode: nil)
+        if !response.nil? && !response.body.nil?
+            podList = JSON.parse(response.body)
+            if !podList.nil? && !podList.empty? && podList.key?("items") && !podList["items"].nil? && !podList["items"].empty?
+              podList["items"].each do |item|
+                containerInventoryRecords = KubernetesContainerInventory.getContainerInventoryRecords(item, batchTime, clusterCollectEnvironmentVar)
+                containerInventoryRecords.each do |containerRecord|
+                  ContainerInventoryState.writeContainerState(containerRecord)
+                  if hostName.empty? && !containerRecord["Computer"].empty?
+                    hostName = containerRecord["Computer"]
+                  end
+                  containerIds.push containerRecord["InstanceID"]
+                  containerInventory.push containerRecord
+                end           
               end
-            end
-          end
-
-          containerInventory.each do |record|
-            wrapper = {
-              "DataType" => "CONTAINER_INVENTORY_BLOB",
-              "IPName" => "ContainerInsights",
-              "DataItems" => [record.each { |k, v| record[k] = v }],
-            }
-            eventStream.add(emitTime, wrapper) if wrapper
-          end
-          router.emit_stream(@tag, eventStream) if eventStream
-          @@istestvar = ENV["ISTEST"]
-          if (!@@istestvar.nil? && !@@istestvar.empty? && @@istestvar.casecmp("true") == 0 && eventStream.count > 0)
-            $log.info("containerInventoryEmitStreamSuccess @ #{Time.now.utc.iso8601}")
-          end
-          $log.info("in_container_inventory::enumerate : Processing complete - emitted stream @ #{Time.now.utc.iso8601}")
+            end  
+        end                          
+        # Update the state for deleted containers
+        deletedContainers = ContainerInventoryState.getDeletedContainers(containerIds)
+        if !deletedContainers.nil? && !deletedContainers.empty?
+          deletedContainers.each do |deletedContainer|
+            container = ContainerInventoryState.readContainerState(deletedContainer)
+            if !container.nil?
+              container.each { |k, v| container[k] = v }
+              container["State"] = "Deleted"   
+              KubernetesContainerInventory.deleteCGroupCacheEntryForDeletedContainer(container["InstanceID"])
+              containerInventory.push container
+             end
+           end
+        end        
+        containerInventory.each do |record|
+          wrapper = {
+            "DataType" => "CONTAINER_INVENTORY_BLOB",
+            "IPName" => "ContainerInsights",
+            "DataItems" => [record.each { |k, v| record[k] = v }],
+          }
+          eventStream.add(emitTime, wrapper) if wrapper
+        end
+        router.emit_stream(@tag, eventStream) if eventStream
+        @@istestvar = ENV["ISTEST"]
+        if (!@@istestvar.nil? && !@@istestvar.empty? && @@istestvar.casecmp("true") == 0 && eventStream.count > 0)
+          $log.info("containerInventoryEmitStreamSuccess @ #{Time.now.utc.iso8601}")
         end
+        $log.info("in_container_inventory::enumerate : Processing complete - emitted stream @ #{Time.now.utc.iso8601}")
         timeDifference = (DateTime.now.to_time.to_i - @@telemetryTimeTracker).abs
         timeDifferenceInMinutes = timeDifference / 60
         if (timeDifferenceInMinutes >= 5)
           @@telemetryTimeTracker = DateTime.now.to_time.to_i
           telemetryProperties = {}
-          telemetryProperties["Computer"] = hostname
+          telemetryProperties["Computer"] = hostName
           telemetryProperties["ContainerCount"] = containerInventory.length
           ApplicationInsightsUtility.sendTelemetry(@@PluginName, telemetryProperties)
         end
       rescue => errorStr
         $log.warn("Exception in enumerate container inventory: #{errorStr}")
+        $log.debug_backtrace(errorStr.backtrace)
         ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
       end
     end
@@ -281,6 +143,7 @@ def run_periodic
             $log.info("in_container_inventory::run_periodic.enumerate.end @ #{Time.now.utc.iso8601}")
           rescue => errorStr
             $log.warn "in_container_inventory::run_periodic: Failed in enumerate container inventory: #{errorStr}"
+            $log.debug_backtrace(errorStr.backtrace)
           end
         end
         @mutex.lock
diff --git a/source/code/plugin/in_kube_events.rb b/source/code/plugin/in_kube_events.rb
index bb0ab6f05..6f59a3fc1 100644
--- a/source/code/plugin/in_kube_events.rb
+++ b/source/code/plugin/in_kube_events.rb
@@ -103,8 +103,6 @@ def enumerate
 
         # Flush AppInsights telemetry once all the processing is done, only if the number of events flushed is greater than 0
         if (@eventsCount > 0)
-          telemetryProperties = {}
-          telemetryProperties["CollectAllKubeEvents"] = @collectAllKubeEvents
           ApplicationInsightsUtility.sendMetricTelemetry("EventCount", @eventsCount, {})
         end
       rescue => errorStr
@@ -141,7 +139,7 @@ def parse_and_emit_records(events, eventQueryState, newEventQueryState, batchTim
           record["Name"] = items["involvedObject"]["name"]
           record["Reason"] = items["reason"]
           record["Message"] = items["message"]
-          record["Type"] = items["type"]
+          record["KubeEventType"] = items["type"]
           record["TimeGenerated"] = items["metadata"]["creationTimestamp"]
           record["SourceComponent"] = items["source"]["component"]
           record["FirstSeen"] = items["firstTimestamp"]
diff --git a/source/code/plugin/in_kube_nodes.rb b/source/code/plugin/in_kube_nodes.rb
index 4242a8dba..fe28efad6 100644
--- a/source/code/plugin/in_kube_nodes.rb
+++ b/source/code/plugin/in_kube_nodes.rb
@@ -7,6 +7,7 @@ class Kube_nodeInventory_Input < Input
 
     @@ContainerNodeInventoryTag = "oms.containerinsights.ContainerNodeInventory"
     @@MDMKubeNodeInventoryTag = "mdm.kubenodeinventory"
+    @@configMapMountPath = "/etc/config/settings/log-data-collection-settings"
     @@promConfigMountPath = "/etc/config/settings/prometheus-data-collection-settings"
     @@AzStackCloudFileName = "/etc/kubernetes/host/azurestackcloud.json"
     @@kubeperfTag = "oms.api.KubePerf"
@@ -18,6 +19,7 @@ class Kube_nodeInventory_Input < Input
     @@rsPromUrlCount = ENV["TELEMETRY_RS_PROM_URLS_LENGTH"]
     @@rsPromMonitorPods = ENV["TELEMETRY_RS_PROM_MONITOR_PODS"]
     @@rsPromMonitorPodsNamespaceLength = ENV["TELEMETRY_RS_PROM_MONITOR_PODS_NS_LENGTH"]
+    @@collectAllKubeEvents = ENV["AZMON_CLUSTER_COLLECT_ALL_KUBE_EVENTS"]
 
     def initialize
       super
@@ -163,9 +165,13 @@ def parse_and_emit_records(nodeInventory, batchTime = Time.utc.iso8601)
           record["KubeletVersion"] = nodeInfo["kubeletVersion"]
           record["KubeProxyVersion"] = nodeInfo["kubeProxyVersion"]
           containerNodeInventoryRecord["OperatingSystem"] = nodeInfo["osImage"]
-          dockerVersion = nodeInfo["containerRuntimeVersion"]
-          dockerVersion.slice! "docker://"
-          containerNodeInventoryRecord["DockerVersion"] = dockerVersion
+          containerRuntimeVersion = nodeInfo["containerRuntimeVersion"]
+          if containerRuntimeVersion.downcase.start_with?("docker://")
+            containerNodeInventoryRecord["DockerVersion"] = containerRuntimeVersion.split("//")[1]
+          else
+            # using containerRuntimeVersion as DockerVersion as is for non docker runtimes
+            containerNodeInventoryRecord["DockerVersion"] = containerRuntimeVersion
+          end
           # ContainerNodeInventory data for docker version and operating system.
           containerNodeInventoryWrapper = {
             "DataType" => "CONTAINER_NODE_INVENTORY_BLOB",
@@ -188,7 +194,12 @@ def parse_and_emit_records(nodeInventory, batchTime = Time.utc.iso8601)
             properties["Computer"] = record["Computer"]
             properties["KubeletVersion"] = record["KubeletVersion"]
             properties["OperatingSystem"] = nodeInfo["operatingSystem"]
-            properties["DockerVersion"] = dockerVersion
+            # DockerVersion field holds docker version if runtime is docker/moby else <runtime>://<version>
+            if containerRuntimeVersion.downcase.start_with?("docker://")
+              properties["DockerVersion"] = containerRuntimeVersion.split("//")[1]
+            else
+              properties["DockerVersion"] = containerRuntimeVersion
+            end
             properties["KubernetesProviderID"] = record["KubernetesProviderID"]
             properties["KernelVersion"] = nodeInfo["kernelVersion"]
             properties["OSImage"] = nodeInfo["osImage"]
@@ -210,6 +221,11 @@ def parse_and_emit_records(nodeInventory, batchTime = Time.utc.iso8601)
               ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
             end
 
+            # Telemetry for data collection config for replicaset
+            if (File.file?(@@configMapMountPath))
+              properties["collectAllKubeEvents"] = @@collectAllKubeEvents
+            end
+
             #telemetry about prometheus metric collections settings for replicaset
             if (File.file?(@@promConfigMountPath))
               properties["rsPromInt"] = @@rsPromInterval
@@ -230,7 +246,7 @@ def parse_and_emit_records(nodeInventory, batchTime = Time.utc.iso8601)
         if telemetrySent == true
           @@nodeTelemetryTimeTracker = DateTime.now.to_time.to_i
         end
-        
+
         if (!@@istestvar.nil? && !@@istestvar.empty? && @@istestvar.casecmp("true") == 0 && eventStream.count > 0)
           $log.info("kubeNodeInventoryEmitStreamSuccess @ #{Time.now.utc.iso8601}")
         end
@@ -281,7 +297,7 @@ def parse_and_emit_records(nodeInventory, batchTime = Time.utc.iso8601)
             $log.warn "Failed when processing GPU metrics in_kube_nodes : #{errorStr}"
             $log.debug_backtrace(errorStr.backtrace)
             ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
-          end 
+          end
           #end GPU InsightsMetrics items
         rescue => errorStr
           $log.warn "Failed in enumerate for KubePerf from in_kube_nodes : #{errorStr}"
diff --git a/source/code/plugin/in_kube_podinventory.rb b/source/code/plugin/in_kube_podinventory.rb
index 29438d076..e4ba6f15e 100644
--- a/source/code/plugin/in_kube_podinventory.rb
+++ b/source/code/plugin/in_kube_podinventory.rb
@@ -2,7 +2,7 @@
 # frozen_string_literal: true
 
 module Fluent
-  require_relative "podinventory_to_mdm"
+  require_relative "podinventory_to_mdm"      
 
   class Kube_PodInventory_Input < Input
     Plugin.register_input("kubepodinventory", self)
@@ -19,7 +19,8 @@ def initialize
       require "yajl"
       require "set"
       require "time"
-
+      
+      require_relative "kubernetes_container_inventory"
       require_relative "KubernetesApiClient"
       require_relative "ApplicationInsightsUtility"
       require_relative "oms_common"
@@ -137,126 +138,7 @@ def enumerate(podList = nil)
         $log.debug_backtrace(errorStr.backtrace)
         ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
       end
-    end
-
-    def populateWindowsContainerInventoryRecord(container, record, containerEnvVariableHash, batchTime)
-      begin
-        containerInventoryRecord = {}
-        containerName = container["name"]
-        containerInventoryRecord["InstanceID"] = record["ContainerID"]
-        containerInventoryRecord["CollectionTime"] = batchTime #This is the time that is mapped to become TimeGenerated
-        containerInventoryRecord["Computer"] = record["Computer"]
-        containerInventoryRecord["ContainerHostname"] = record["Computer"]
-        containerInventoryRecord["ElementName"] = containerName
-
-        # Find delimiters in the string of format repository/image:imagetag
-        imageValue = container["image"]
-        if !imageValue.empty?
-          slashLocation = imageValue.index("/")
-          colonLocation = imageValue.index(":")
-          if !colonLocation.nil?
-            if slashLocation.nil?
-              # image:imagetag
-              containerInventoryRecord["Image"] = imageValue[0..(colonLocation - 1)]
-            else
-              # repository/image:imagetag
-              containerInventoryRecord["Repository"] = imageValue[0..(slashLocation - 1)]
-              containerInventoryRecord["Image"] = imageValue[(slashLocation + 1)..(colonLocation - 1)]
-            end
-            containerInventoryRecord["ImageTag"] = imageValue[(colonLocation + 1)..-1]
-          end
-        end
-
-        imageIdInfo = container["imageID"]
-        imageIdSplitInfo = imageIdInfo.split("@")
-        if !imageIdSplitInfo.nil?
-          containerInventoryRecord["ImageId"] = imageIdSplitInfo[1]
-        end
-        # Get container state
-        containerStatus = container["state"]
-        if containerStatus.keys[0] == "running"
-          containerInventoryRecord["State"] = "Running"
-          containerInventoryRecord["StartedTime"] = container["state"]["running"]["startedAt"]
-        elsif containerStatus.keys[0] == "terminated"
-          containerExitCode = container["state"]["terminated"]["exitCode"]
-          containerStartTime = container["state"]["terminated"]["startedAt"]
-          containerFinishTime = container["state"]["terminated"]["finishedAt"]
-          if containerExitCode < 0
-            # Exit codes less than 0 are not supported by the engine
-            containerExitCode = 128
-          end
-          if containerExitCode > 0
-            containerInventoryRecord["State"] = "Failed"
-          else
-            containerInventoryRecord["State"] = "Stopped"
-          end
-          containerInventoryRecord["ExitCode"] = containerExitCode
-          containerInventoryRecord["StartedTime"] = containerStartTime
-          containerInventoryRecord["FinishedTime"] = containerFinishTime
-        elsif containerStatus.keys[0] == "waiting"
-          containerInventoryRecord["State"] = "Waiting"
-        end
-        if !containerEnvVariableHash.nil? && !containerEnvVariableHash.empty?
-          containerInventoryRecord["EnvironmentVar"] = containerEnvVariableHash[containerName]
-        end
-        return containerInventoryRecord
-      rescue => errorStr
-        $log.warn "Failed in populateWindowsContainerInventoryRecord: #{errorStr}"
-        $log.debug_backtrace(errorStr.backtrace)
-        ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
-      end
-    end
-
-    def getContainerEnvironmentVariables(pod, clusterCollectEnvironmentVar)
-      begin
-        podSpec = pod["spec"]
-        containerEnvHash = {}
-        podContainersEnv = []
-        if !podSpec["containers"].nil? && !podSpec["containers"].empty?
-          podContainersEnv = podContainersEnv + podSpec["containers"]
-        end
-        # Adding init containers to the record list as well.
-        if !podSpec["initContainers"].nil? && !podSpec["initContainers"].empty?
-          podContainersEnv = podContainersEnv + podSpec["initContainers"]
-        end
-        if !podContainersEnv.nil? && !podContainersEnv.empty?
-          podContainersEnv.each do |container|
-            if !clusterCollectEnvironmentVar.nil? && !clusterCollectEnvironmentVar.empty? && clusterCollectEnvironmentVar.casecmp("false") == 0
-              containerEnvHash[container["name"]] = ["AZMON_CLUSTER_COLLECT_ENV_VAR=FALSE"]
-            else
-              envVarsArray = []
-              containerEnvArray = container["env"]
-              # Parsing the environment variable array of hashes to a string value
-              # since that is format being sent by container inventory workflow in daemonset
-              # Keeping it in the same format because the workflow expects it in this format
-              # and the UX expects an array of string for environment variables
-              if !containerEnvArray.nil? && !containerEnvArray.empty?
-                containerEnvArray.each do |envVarHash|
-                  envName = envVarHash["name"]
-                  envValue = envVarHash["value"]
-                  if !envName.nil? && !envValue.nil?
-                    envArrayElement = envName + "=" + envValue
-                    envVarsArray.push(envArrayElement)
-                  end
-                end
-              end
-              # Skip environment variable processing if it contains the flag AZMON_COLLECT_ENV=FALSE
-              envValueString = envVarsArray.to_s
-              if /AZMON_COLLECT_ENV=FALSE/i.match(envValueString)
-                envValueString = ["AZMON_COLLECT_ENV=FALSE"]
-                $log.warn("Environment Variable collection for container: #{container["name"]} skipped because AZMON_COLLECT_ENV is set to false")
-              end
-              containerEnvHash[container["name"]] = envValueString
-            end
-          end
-        end
-        return containerEnvHash
-      rescue => errorStr
-        $log.warn "Failed in getContainerEnvironmentVariables: #{errorStr}"
-        $log.debug_backtrace(errorStr.backtrace)
-        ApplicationInsightsUtility.sendExceptionTelemetry(errorStr)
-      end
-    end
+    end    
 
     def parse_and_emit_records(podInventory, serviceList, continuationToken, batchTime = Time.utc.iso8601)
       currentTime = Time.now
@@ -269,8 +151,7 @@ def parse_and_emit_records(podInventory, serviceList, continuationToken, batchTi
         # Getting windows nodes from kubeapi
         winNodes = KubernetesApiClient.getWindowsNodesArray
 
-        podInventory["items"].each do |items| #podInventory block start
-          sendWindowsContainerInventoryRecord = false
+        podInventory["items"].each do |items| #podInventory block start          
           containerInventoryRecords = []
           records = []
           record = {}
@@ -343,11 +224,11 @@ def parse_and_emit_records(podInventory, serviceList, continuationToken, batchTi
           if winNodes.length > 0
             if (!record["Computer"].empty? && (winNodes.include? record["Computer"]))
               clusterCollectEnvironmentVar = ENV["AZMON_CLUSTER_COLLECT_ENV_VAR"]
-              if !clusterCollectEnvironmentVar.nil? && !clusterCollectEnvironmentVar.empty? && clusterCollectEnvironmentVar.casecmp("false") == 0
-                $log.warn("WindowsContainerInventory: Environment Variable collection disabled for cluster")
-              end
-              sendWindowsContainerInventoryRecord = true
-              containerEnvVariableHash = getContainerEnvironmentVariables(items, clusterCollectEnvironmentVar)
+              #Generate ContainerInventory records for windows nodes so that we can get image and image tag in property panel
+              containerInventoryRecordsInPodItem = KubernetesContainerInventory.getContainerInventoryRecords(items, batchTime, clusterCollectEnvironmentVar, true)  
+              containerInventoryRecordsInPodItem.each do |containerRecord|
+                containerInventoryRecords.push(containerRecord)          
+              end              
             end
           end
 
@@ -459,13 +340,7 @@ def parse_and_emit_records(podInventory, serviceList, continuationToken, batchTi
               end
 
               podRestartCount += containerRestartCount
-              records.push(record.dup)
-
-              #Generate ContainerInventory records for windows nodes so that we can get image and image tag in property panel
-              if sendWindowsContainerInventoryRecord == true
-                containerInventoryRecord = populateWindowsContainerInventoryRecord(container, record, containerEnvVariableHash, batchTime)
-                containerInventoryRecords.push(containerInventoryRecord)
-              end
+              records.push(record.dup)            
             end
           else # for unscheduled pods there are no status.containerStatuses, in this case we still want the pod
             records.push(record)
diff --git a/source/code/plugin/kubelet_utils.rb b/source/code/plugin/kubelet_utils.rb
index 6d97e30a9..fecb339b9 100644
--- a/source/code/plugin/kubelet_utils.rb
+++ b/source/code/plugin/kubelet_utils.rb
@@ -7,7 +7,7 @@
 class KubeletUtils
     class << self
         def get_node_capacity
-            
+
             cpu_capacity = 1.0
             memory_capacity = 1.0
 
diff --git a/source/code/plugin/kubernetes_container_inventory.rb b/source/code/plugin/kubernetes_container_inventory.rb
new file mode 100644
index 000000000..207514c1e
--- /dev/null
+++ b/source/code/plugin/kubernetes_container_inventory.rb
@@ -0,0 +1,321 @@
+#!/usr/local/bin/ruby
+# frozen_string_literal: true
+
+class KubernetesContainerInventory
+  require "yajl/json_gem"
+  require "time"
+  require_relative "omslog"
+  require_relative "ApplicationInsightsUtility"
+  
+  # cache the container and cgroup parent process
+  @@containerCGroupCache = Hash.new
+
+  def initialize
+  end
+
+  class << self
+    def getContainerInventoryRecords(podItem, batchTime, clusterCollectEnvironmentVar, isWindows = false)
+      containerInventoryRecords = Array.new
+      begin
+        containersInfoMap = getContainersInfoMap(podItem, isWindows)
+        podContainersStatuses = []
+        if !podItem["status"]["containerStatuses"].nil? && !podItem["status"]["containerStatuses"].empty?
+          podContainersStatuses = podItem["status"]["containerStatuses"]
+        end
+        if !podItem["status"]["initContainerStatuses"].nil? && !podItem["status"]["initContainerStatuses"].empty?
+          podContainersStatuses = podContainersStatuses + podItem["status"]["initContainerStatuses"]
+        end
+        containerInventoryRecord = {}
+        if !podContainersStatuses.empty?
+          podContainersStatuses.each do |containerStatus|
+            containerInventoryRecord["CollectionTime"] = batchTime #This is the time that is mapped to become TimeGenerated
+            containerName = containerStatus["name"]
+            # containeId format is <containerRuntime>://<containerId>
+            containerRuntime = ""
+            containerId = ""
+            if !containerStatus["containerID"].nil?
+              containerRuntime = containerStatus["containerID"].split(":")[0]
+              containerId = containerStatus["containerID"].split("//")[1]
+              containerInventoryRecord["InstanceID"] = containerId
+            else
+              # for containers that have image issues (like invalid image/tag etc..) this will be empty. do not make it all 0
+              containerInventoryRecord["InstanceID"] = containerId
+            end
+            # imagedId is of the format - repo@sha256:imageid
+            imageIdValue = containerStatus["imageID"]
+            if !imageIdValue.nil? && !imageIdValue.empty?
+              atLocation = imageIdValue.index("@")
+              if !atLocation.nil?
+                containerInventoryRecord["ImageId"] = imageIdValue[(atLocation + 1)..-1]
+              end
+            end
+            # image is of the format - repository/image:imagetag
+            imageValue = containerStatus["image"]
+            if !imageValue.nil? && !imageValue.empty?
+              # Find delimiters in the string of format repository/image:imagetag
+              slashLocation = imageValue.index("/")
+              colonLocation = imageValue.index(":")
+              if !colonLocation.nil?
+                if slashLocation.nil?
+                  # image:imagetag
+                  containerInventoryRecord["Image"] = imageValue[0..(colonLocation - 1)]
+                else
+                  # repository/image:imagetag
+                  containerInventoryRecord["Repository"] = imageValue[0..(slashLocation - 1)]
+                  containerInventoryRecord["Image"] = imageValue[(slashLocation + 1)..(colonLocation - 1)]
+                end
+                containerInventoryRecord["ImageTag"] = imageValue[(colonLocation + 1)..-1]
+              end
+            elsif !imageIdValue.nil? && !imageIdValue.empty?
+              # Getting repo information from imageIdValue when no tag in ImageId
+              if !atLocation.nil?
+                containerInventoryRecord["Repository"] = imageIdValue[0..(atLocation - 1)]
+              end
+            end
+            containerInventoryRecord["ExitCode"] = 0
+            isContainerTerminated = false
+            if !containerStatus["state"].nil? && !containerStatus["state"].empty?
+              containerState = containerStatus["state"]
+              if containerState.key?("running")
+                containerInventoryRecord["State"] = "Running"
+                containerInventoryRecord["StartedTime"] = containerState["running"]["startedAt"]
+              elsif containerState.key?("terminated")
+                containerInventoryRecord["State"] = "Terminated"
+                containerInventoryRecord["StartedTime"] = containerState["terminated"]["startedAt"]
+                containerInventoryRecord["FinishedTime"] = containerState["terminated"]["finishedAt"]
+                containerInventoryRecord["ExitCode"] = containerState["terminated"]["exitCode"]
+                isContainerTerminated = true
+              elsif containerState.key?("waiting")
+                containerInventoryRecord["State"] = "Waiting"
+              end
+            end
+            containerInfoMap = containersInfoMap[containerName]
+            containerInventoryRecord["ElementName"] = containerInfoMap["ElementName"]
+            containerInventoryRecord["Computer"] = containerInfoMap["Computer"]
+            containerInventoryRecord["ContainerHostname"] = containerInfoMap["ContainerHostname"]
+            containerInventoryRecord["CreatedTime"] = containerInfoMap["CreatedTime"]
+            containerInventoryRecord["EnvironmentVar"] = containerInfoMap["EnvironmentVar"]
+            containerInventoryRecord["Ports"] = containerInfoMap["Ports"]
+            containerInventoryRecord["Command"] = containerInfoMap["Command"]
+            if !clusterCollectEnvironmentVar.nil? && !clusterCollectEnvironmentVar.empty? && clusterCollectEnvironmentVar.casecmp("false") == 0
+              containerInventoryRecord["EnvironmentVar"] = ["AZMON_CLUSTER_COLLECT_ENV_VAR=FALSE"]
+            elsif isWindows
+              containerInventoryRecord["EnvironmentVar"] = containerInfoMap["EnvironmentVar"]
+            else
+              if containerId.nil? || containerId.empty? || isContainerTerminated || containerRuntime.nil? || containerRuntime.empty?
+                containerInventoryRecord["EnvironmentVar"] = ""
+              else
+                if containerRuntime.casecmp("cri-o") == 0
+                  # crio containers have conmon as parent process and we only to need get container main process envvars
+                  containerInventoryRecord["EnvironmentVar"] = obtainContainerEnvironmentVars("crio-#{containerId}")
+                else
+                  containerInventoryRecord["EnvironmentVar"] = obtainContainerEnvironmentVars(containerId)
+                end
+              end
+            end
+            containerInventoryRecords.push containerInventoryRecord
+          end
+        end
+      rescue => error
+        $log.warn("KubernetesContainerInventory::getContainerInventoryRecords : Get Container Inventory Records failed: #{error}")
+        $log.debug_backtrace(error.backtrace)
+        ApplicationInsightsUtility.sendExceptionTelemetry(error)
+      end
+      return containerInventoryRecords
+    end
+
+    def getContainersInfoMap(podItem, isWindows)
+      containersInfoMap = {}
+      begin
+        nodeName = (!podItem["spec"]["nodeName"].nil?) ? podItem["spec"]["nodeName"] : ""
+        createdTime = podItem["metadata"]["creationTimestamp"]
+        if !podItem.nil? && !podItem.empty? && podItem.key?("spec") && !podItem["spec"].nil? && !podItem["spec"].empty?
+          podContainers = []
+          if !podItem["spec"]["containers"].nil? && !podItem["spec"]["containers"].empty?
+            podContainers = podItem["spec"]["containers"]
+          end
+          if !podItem["spec"]["initContainers"].nil? && !podItem["spec"]["initContainers"].empty?
+            podContainers = podContainers + podItem["spec"]["initContainers"]
+          end
+          if !podContainers.empty?
+            podContainers.each do |container|
+              containerInfoMap = {}
+              containerName = container["name"]
+              containerInfoMap["ElementName"] = containerName
+              containerInfoMap["Computer"] = nodeName
+              containerInfoMap["ContainerHostname"] = nodeName
+              containerInfoMap["CreatedTime"] = createdTime
+              portsValue = container["ports"]
+              portsValueString = (portsValue.nil?) ? "" : portsValue.to_s
+              containerInfoMap["Ports"] = portsValueString
+              cmdValue = container["command"]
+              cmdValueString = (cmdValue.nil?) ? "" : cmdValue.to_s
+              containerInfoMap["Command"] = cmdValueString
+              containerInfoMap["EnvironmentVar"] = ""
+              if isWindows
+                containerInfoMap["EnvironmentVar"] = obtainWindowsContainerEnvironmentVars(podItem, container)
+              end
+              containersInfoMap[containerName] = containerInfoMap
+            end
+          end
+        end
+      rescue => error
+        $log.warn("KubernetesContainerInventory::getContainersInfoMap : Get Container Info Maps failed: #{error}")
+        $log.debug_backtrace(error.backtrace)
+        ApplicationInsightsUtility.sendExceptionTelemetry(error)
+      end
+      return containersInfoMap
+    end
+
+    def obtainContainerEnvironmentVars(containerId)
+      $log.info("KubernetesContainerInventory::obtainContainerEnvironmentVars @ #{Time.now.utc.iso8601}")
+      envValueString = ""
+      begin
+        unless @@containerCGroupCache.has_key?(containerId)
+          $log.info("KubernetesContainerInventory::obtainContainerEnvironmentVars fetching cGroup parent pid @ #{Time.now.utc.iso8601} for containerId: #{containerId}")
+          Dir["/hostfs/proc/*/cgroup"].each do |filename|
+            if File.file?(filename) && File.foreach(filename).grep(/#{containerId}/).any?
+              # file full path is /hostfs/proc/<cGroupPid>/cgroup            
+              cGroupPid = filename.split("/")[3]
+              if @@containerCGroupCache.has_key?(containerId)
+                tempCGroupPid = @@containerCGroupCache[containerId]
+                if tempCGroupPid > cGroupPid
+                  @@containerCGroupCache[containerId] = cGroupPid
+                end
+              else
+                @@containerCGroupCache[containerId] = cGroupPid
+              end
+            end
+          end
+        end
+        cGroupPid = @@containerCGroupCache[containerId]
+        if !cGroupPid.nil? && !cGroupPid.empty?
+          environFilePath = "/hostfs/proc/#{cGroupPid}/environ"
+          $log.info("KubernetesContainerInventory::obtainContainerEnvironmentVars cGroupPid: #{cGroupPid} environFilePath: #{environFilePath} for containerId: #{containerId}")
+          if File.exist?(environFilePath)
+            # Skip environment variable processing if it contains the flag AZMON_COLLECT_ENV=FALSE
+            # Check to see if the environment variable collection is disabled for this container.
+            if File.foreach(environFilePath).grep(/AZMON_COLLECT_ENV=FALSE/i).any?
+              envValueString = ["AZMON_COLLECT_ENV=FALSE"]
+              $log.warn("Environment Variable collection for container: #{containerId} skipped because AZMON_COLLECT_ENV is set to false")
+            else
+              # Restricting the ENV string value to 200kb since the size of this string can go very high             
+              envVars = File.read(environFilePath, 200000)             
+              if !envVars.nil? && !envVars.empty?
+                envVars = envVars.split("\0")
+                envValueString = envVars.to_s
+                envValueStringLength = envValueString.length
+                $log.info("KubernetesContainerInventory::environment vars filename @ #{environFilePath} envVars size @ #{envValueStringLength}")
+                if envValueStringLength >= 200000
+                  lastIndex = envValueString.rindex("\", ")
+                  if !lastIndex.nil?
+                    envValueStringTruncated = envValueString.slice(0..lastIndex) + "]"
+                    envValueString = envValueStringTruncated
+                  end
+                end
+              end
+            end
+          end
+        else
+          $log.warn("KubernetesContainerInventory::obtainContainerEnvironmentVars: cGroupPid is NIL or empty for containerId: #{containerId}")  
+        end
+      rescue => error
+        $log.warn("KubernetesContainerInventory::obtainContainerEnvironmentVars: obtain Container Environment vars failed: #{error} for containerId: #{containerId}")
+        $log.debug_backtrace(error.backtrace)
+        ApplicationInsightsUtility.sendExceptionTelemetry(error)
+      end
+      return envValueString
+    end
+
+    def obtainWindowsContainerEnvironmentVars(pod, container)
+      envValueString = ""
+      begin
+        envVars = []
+        envVarsJSON = container["env"]
+        if !pod.nil? && !pod.empty? && !envVarsJSON.nil? && !envVarsJSON.empty?
+          envVarsJSON.each do |envVar|
+            key = envVar["name"]
+            value = ""
+            if !envVar["value"].nil?
+              value = envVar["value"]
+            elsif !envVar["valueFrom"].nil?
+              valueFrom = envVar["valueFrom"]
+              # https://kubernetes.io/docs/tasks/inject-data-application/environment-variable-expose-pod-information/#use-pod-fields-as-values-for-environment-variables
+              if valueFrom.key?("fieldRef") && !valueFrom["fieldRef"]["fieldPath"].nil? && !valueFrom["fieldRef"]["fieldPath"].empty?
+                fieldPath = valueFrom["fieldRef"]["fieldPath"]
+                fields = fieldPath.split(".")
+                if fields.length() == 2
+                  if !fields[1].nil? && !fields[1].empty? && fields[1].end_with?("]")
+                    indexFields = fields[1].split("[")
+                    hashMapValue = pod[fields[0]][indexFields[0]]
+                    if !hashMapValue.nil? && !hashMapValue.empty?
+                      subField = indexFields[1].chomp("]").delete("\\'")
+                      value = hashMapValue[subField]
+                    end
+                  else
+                    value = pod[fields[0]][fields[1]]
+                  end
+                end
+                # https://kubernetes.io/docs/tasks/inject-data-application/environment-variable-expose-pod-information/#use-container-fields-as-values-for-environment-variables
+              elsif valueFrom.key?("resourceFieldRef") && !valueFrom["resourceFieldRef"]["resource"].nil? && !valueFrom["resourceFieldRef"]["resource"].empty?
+                resource = valueFrom["resourceFieldRef"]["resource"]
+                resourceFields = resource.split(".")
+                containerResources = container["resources"]
+                if !containerResources.nil? && !containerResources.empty? && resourceFields.length() == 2
+                  value = containerResources[resourceFields[0]][resourceFields[1]]
+                end
+                # https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables
+              elsif valueFrom.key?("secretKeyRef")
+                secretName = valueFrom["secretKeyRef"]["name"]
+                secretKey = valueFrom["secretKeyRef"]["key"]
+                # This is still secret not the plaintext. Flatten value so that CI Ux can show that
+                if !secretName.nil? && !secretName.empty? && !secretKey.nil? && !secretKey.empty?
+                  value = "secretKeyRef_#{secretName}_#{secretKey}"
+                end
+              else
+                value = envVar["valueFrom"].to_s
+              end
+            end
+            envVars.push("#{key}=#{value}")
+          end
+          envValueString = envVars.to_s
+          containerName = container["name"]
+          # Skip environment variable processing if it contains the flag AZMON_COLLECT_ENV=FALSE
+          # Check to see if the environment variable collection is disabled for this container.
+          if /AZMON_COLLECT_ENV=FALSE/i.match(envValueString)
+            envValueString = ["AZMON_COLLECT_ENV=FALSE"]
+            $log.warn("Environment Variable collection for container: #{containerName} skipped because AZMON_COLLECT_ENV is set to false")
+          else
+            # Restricting the ENV string value to 200kb since the size of this string can go very high
+            if envValueString.length > 200000
+              envValueStringTruncated = envValueString.slice(0..200000)
+              lastIndex = envValueStringTruncated.rindex("\", ")
+              if !lastIndex.nil?
+                envValueString = envValueStringTruncated.slice(0..lastIndex) + "]"
+              else
+                envValueString = envValueStringTruncated
+              end
+            end
+          end
+        end
+      rescue => error
+        $log.warn("KubernetesContainerInventory::obtainWindowsContainerEnvironmentVars: parsing of EnvVars failed: #{error}")
+        $log.debug_backtrace(error.backtrace)
+        ApplicationInsightsUtility.sendExceptionTelemetry(error)
+      end
+      return envValueString
+    end
+
+    def deleteCGroupCacheEntryForDeletedContainer(containerId)
+      begin
+        if !containerId.nil? && !containerId.empty? && !@@containerCGroupCache.nil? && @@containerCGroupCache.length > 0 && @@containerCGroupCache.key?(containerId)
+          @@containerCGroupCache.delete(containerId)
+        end      
+      rescue => error
+        $log.warn("KubernetesContainerInventory::deleteCGroupCacheEntryForDeletedContainer: deleting of cache entry failed: #{error}")
+        $log.debug_backtrace(error.backtrace)
+        ApplicationInsightsUtility.sendExceptionTelemetry(error)
+      end 
+    end
+  end
+end
diff --git a/source/code/plugin/out_mdm.rb b/source/code/plugin/out_mdm.rb
index 243251bca..135da982e 100644
--- a/source/code/plugin/out_mdm.rb
+++ b/source/code/plugin/out_mdm.rb
@@ -204,6 +204,7 @@ def write(chunk)
           count = post_body.size
           while count > 0
             current_batch = post_body.first(@@record_batch_size)
+            post_body = post_body.drop(current_batch.size)
             count -= current_batch.size
             send_to_mdm current_batch
           end