Skip to content

Fix Windows container inventory cache refresh races #1787

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
suyadav1 wants to merge 2 commits into
base: ci_prod
Choose a base branch
from
+354 −93
Draft

Fix Windows container inventory cache refresh races #1787

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
197958a
fix: preserve Windows inventory cache during refresh
suyadav1 Sep 23, 2026
032352a
fix: decouple pod inventory startup from Windows discovery
suyadav1 Sep 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 53 additions & 93 deletions source/plugins/ruby/in_kube_podinventory.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ def initialize

@watchWinNodesThread = nil
@windowsNodeNameListCache = []
@podCacheRefreshRequested = false
@windowsContainerRecordsCacheSizeBytes = 0

@kubeservicesTag = "oneagent.containerInsights.KUBE_SERVICES_BLOB"
Expand Down Expand Up @@ -742,25 +743,22 @@ def getServiceNameFromLabels(namespace, labels, serviceRecords)

def watch_pods
$log.info("in_kube_podinventory::watch_pods:Start @ #{Time.now.utc.iso8601}")
return if @finished

podsResourceVersion = nil
# invoke getWindowsNodes to handle scenario where windowsNodeNameCache not populated yet on containerstart
winNodes = KubernetesApiClient.getWindowsNodesArray()
if winNodes.length > 0
@windowsNodeNameCacheMutex.synchronize {
@windowsNodeNameListCache = winNodes.dup
}
end
loop do
begin
@windowsNodeNameCacheMutex.synchronize {
if @podCacheRefreshRequested
podsResourceVersion = nil
@podCacheRefreshRequested = false
end
}
if podsResourceVersion.nil?
# clear cache before filling the cache with list
@podCacheMutex.synchronize {
@podItemsCache.clear()
}
currentWindowsNodeNameList = []
@windowsNodeNameCacheMutex.synchronize {
currentWindowsNodeNameList = @windowsNodeNameListCache.dup
}
continuationToken = nil
resourceUri = "pods?limit=#{@PODS_CHUNK_SIZE}"
$log.info("in_kube_podinventory::watch_pods:Getting pods from Kube API: #{resourceUri} @ #{Time.now.utc.iso8601}")
Expand All @@ -777,13 +775,9 @@ def watch_pods
key = item["metadata"]["uid"]
if !key.nil? && !key.empty?
nodeName = (!item["spec"].nil? && !item["spec"]["nodeName"].nil?) ? item["spec"]["nodeName"] : ""
isWindowsPodItem = false
if !nodeName.empty? &&
!currentWindowsNodeNameList.nil? &&
!currentWindowsNodeNameList.empty? &&
currentWindowsNodeNameList.include?(nodeName)
isWindowsPodItem = true
end
isWindowsPodItem = @windowsNodeNameCacheMutex.synchronize {
@windowsNodeNameListCache.include?(nodeName)
}
podItem = KubernetesApiClient.getOptimizedItem("pods", item, isWindowsPodItem)
if !podItem.nil? && !podItem.empty?
@podCacheMutex.synchronize {
Expand Down Expand Up @@ -816,13 +810,9 @@ def watch_pods
key = item["metadata"]["uid"]
if !key.nil? && !key.empty?
nodeName = (!item["spec"].nil? && !item["spec"]["nodeName"].nil?) ? item["spec"]["nodeName"] : ""
isWindowsPodItem = false
if !nodeName.empty? &&
!currentWindowsNodeNameList.nil? &&
!currentWindowsNodeNameList.empty? &&
currentWindowsNodeNameList.include?(nodeName)
isWindowsPodItem = true
end
isWindowsPodItem = @windowsNodeNameCacheMutex.synchronize {
@windowsNodeNameListCache.include?(nodeName)
}
podItem = KubernetesApiClient.getOptimizedItem("pods", item, isWindowsPodItem)
if !podItem.nil? && !podItem.empty?
@podCacheMutex.synchronize {
Expand All @@ -843,6 +833,8 @@ def watch_pods
end
end
end
next if @windowsNodeNameCacheMutex.synchronize { @podCacheRefreshRequested }

if podsResourceVersion.nil? || podsResourceVersion.empty? || podsResourceVersion == "0"
# https://github.com/kubernetes/kubernetes/issues/74022
$log.warn("in_kube_podinventory::watch_pods:received podsResourceVersion either nil or empty or 0 @ #{Time.now.utc.iso8601}")
Expand All @@ -856,6 +848,8 @@ def watch_pods
$log.warn("in_kube_podinventory::watch_pods:watch API returned nil watcher for watch connection with resource version: #{podsResourceVersion} @ #{Time.now.utc.iso8601}")
else
watcher.each do |notice|
break if @windowsNodeNameCacheMutex.synchronize { @podCacheRefreshRequested }

case notice["type"]
when "ADDED", "MODIFIED", "DELETED", "BOOKMARK"
item = notice["object"]
Expand All @@ -874,18 +868,10 @@ def watch_pods
if ((notice["type"] == "ADDED") || (notice["type"] == "MODIFIED"))
key = item["metadata"]["uid"]
if !key.nil? && !key.empty?
currentWindowsNodeNameList = []
@windowsNodeNameCacheMutex.synchronize {
currentWindowsNodeNameList = @windowsNodeNameListCache.dup
}
isWindowsPodItem = false
nodeName = (!item["spec"].nil? && !item["spec"]["nodeName"].nil?) ? item["spec"]["nodeName"] : ""
if !nodeName.empty? &&
!currentWindowsNodeNameList.nil? &&
!currentWindowsNodeNameList.empty? &&
currentWindowsNodeNameList.include?(nodeName)
isWindowsPodItem = true
end
isWindowsPodItem = @windowsNodeNameCacheMutex.synchronize {
@windowsNodeNameListCache.include?(nodeName)
}
podItem = KubernetesApiClient.getOptimizedItem("pods", item, isWindowsPodItem)
if !podItem.nil? && !podItem.empty?
@podCacheMutex.synchronize {
Expand Down Expand Up @@ -1070,67 +1056,39 @@ def watch_windows_nodes
loop do
begin
if nodesResourceVersion.nil?
@windowsNodeNameCacheMutex.synchronize {
@windowsNodeNameListCache.clear()
}
windowsNodeNameList = []
listResourceVersion = nil
continuationToken = nil
resourceUri = KubernetesApiClient.getNodesResourceUri("nodes?labelSelector=kubernetes.io%2Fos%3Dwindows&limit=#{@NODES_CHUNK_SIZE}")
$log.info("in_kube_podinventory::watch_windows_nodes:Getting windows nodes from Kube API: #{resourceUri} @ #{Time.now.utc.iso8601}")
continuationToken, nodeInventory, responseCode = KubernetesApiClient.getResourcesAndContinuationTokenV2(resourceUri)
if responseCode.nil? || responseCode != "200"
$log.info("in_kube_podinventory::watch_windows_nodes:Getting windows nodes from Kube API: #{resourceUri} failed with statuscode: #{responseCode} @ #{Time.now.utc.iso8601}")
else
$log.info("in_kube_podinventory::watch_windows_nodes:Done getting windows nodes from Kube API @ #{Time.now.utc.iso8601}")
if (!nodeInventory.nil? && !nodeInventory.empty?)
nodesResourceVersion = nodeInventory["metadata"]["resourceVersion"]
if (nodeInventory.key?("items") && !nodeInventory["items"].nil? && !nodeInventory["items"].empty?)
$log.info("in_kube_podinventory::watch_windows_nodes: number of windows node items :#{nodeInventory["items"].length} from Kube API @ #{Time.now.utc.iso8601}")
nodeInventory["items"].each do |item|
key = item["metadata"]["name"]
if !key.nil? && !key.empty?
@windowsNodeNameCacheMutex.synchronize {
if !@windowsNodeNameListCache.include?(key)
@windowsNodeNameListCache.push(key)
end
}
else
$log.warn "in_kube_podinventory::watch_windows_nodes:Received node name either nil or empty @ #{Time.now.utc.iso8601}"
end
end
end
else
$log.warn "in_kube_podinventory::watch_windows_nodes:Received empty nodeInventory @ #{Time.now.utc.iso8601}"
begin
pageResourceUri = resourceUri
if !continuationToken.nil? && !continuationToken.empty?
pageResourceUri += "&continue=#{continuationToken}"
end
while (!continuationToken.nil? && !continuationToken.empty?)
continuationToken, nodeInventory, responseCode = KubernetesApiClient.getResourcesAndContinuationTokenV2(resourceUri + "&continue=#{continuationToken}")
if responseCode.nil? || responseCode != "200"
$log.info("in_kube_podinventory::watch_windows_nodes:Getting windows nodes from Kube API: #{resourceUri}&continue=#{continuationToken} failed with statuscode: #{responseCode} @ #{Time.now.utc.iso8601}")
nodesResourceVersion = nil
break # break, if any of the pagination call failed so that full cache can be rebuild with LIST again
else
if (!nodeInventory.nil? && !nodeInventory.empty?)
nodesResourceVersion = nodeInventory["metadata"]["resourceVersion"]
if (nodeInventory.key?("items") && !nodeInventory["items"].nil? && !nodeInventory["items"].empty?)
$log.info("in_kube_podinventory::watch_windows_nodes : number of windows node items :#{nodeInventory["items"].length} from Kube API @ #{Time.now.utc.iso8601}")
nodeInventory["items"].each do |item|
key = item["metadata"]["name"]
if !key.nil? && !key.empty?
@windowsNodeNameCacheMutex.synchronize {
if !@windowsNodeNameListCache.include?(key)
@windowsNodeNameListCache.push(key)
end
}
else
$log.warn "in_kube_podinventory::watch_windows_nodes:Received node name either nil or empty @ #{Time.now.utc.iso8601}"
end
end
end
else
$log.warn "in_kube_podinventory::watch_windows_nodes:Received empty nodeInventory @ #{Time.now.utc.iso8601}"
end
end
$log.info("in_kube_podinventory::watch_windows_nodes:Getting windows nodes from Kube API: #{pageResourceUri} @ #{Time.now.utc.iso8601}")
continuationToken, nodeInventory, responseCode = KubernetesApiClient.getResourcesAndContinuationTokenV2(pageResourceUri)
unless responseCode == "200" && nodeInventory.is_a?(Hash) &&
nodeInventory["metadata"].is_a?(Hash) && nodeInventory["items"].is_a?(Array)
raise "Invalid windows node LIST response for #{pageResourceUri}, statuscode: #{responseCode}"
end
end
pageResourceVersion = nodeInventory["metadata"]["resourceVersion"]
unless pageResourceVersion.is_a?(String) && !pageResourceVersion.empty? && pageResourceVersion != "0" &&
(listResourceVersion.nil? || listResourceVersion == pageResourceVersion)
raise "Invalid or inconsistent windows node LIST resourceVersion for #{pageResourceUri}: #{pageResourceVersion}"
end
listResourceVersion = pageResourceVersion
nodeInventory["items"].each do |item|
nodeName = item["metadata"]["name"]
raise "Invalid windows node name in LIST response for #{pageResourceUri}" unless nodeName.is_a?(String) && !nodeName.empty?
windowsNodeNameList.push(nodeName)
end
end while !continuationToken.nil? && !continuationToken.empty?
windowsNodeNameList.uniq!
@windowsNodeNameCacheMutex.synchronize {
@podCacheRefreshRequested = true unless (windowsNodeNameList - @windowsNodeNameListCache).empty?
@windowsNodeNameListCache = windowsNodeNameList
}
nodesResourceVersion = listResourceVersion
end
if nodesResourceVersion.nil? || nodesResourceVersion.empty? || nodesResourceVersion == "0"
# https://github.com/kubernetes/kubernetes/issues/74022
Expand Down Expand Up @@ -1165,6 +1123,7 @@ def watch_windows_nodes
@windowsNodeNameCacheMutex.synchronize {
if !@windowsNodeNameListCache.include?(key)
@windowsNodeNameListCache.push(key)
@podCacheRefreshRequested = true
end
}
elsif notice["type"] == "DELETED"
Expand Down Expand Up @@ -1196,6 +1155,7 @@ def watch_windows_nodes
rescue => errorStr
$log.warn("in_kube_podinventory::watch_windows_nodes:failed with an error: #{errorStr} @ #{Time.now.utc.iso8601}")
nodesResourceVersion = nil
sleep(30)
end
end
$log.info("in_kube_podinventory::watch_windows_nodes:End @ #{Time.now.utc.iso8601}")
Expand Down
Loading
Loading