extract common code and check node-type offering if present during upgrade and scale

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java

diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
index e352a69..6220b3a 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
@@ -1418,14 +1418,10 @@
                     throw new InvalidParameterValueException("Failed to find service offering ID: " + serviceOfferingId);
                 }
                 checkServiceOfferingForNodesScale(serviceOffering, kubernetesCluster, clusterVersion);
-                Long nodeTypeOfferingId = getExistingServiceOfferingIdForNodeType(key, kubernetesCluster);
-                if (nodeTypeOfferingId == null) {
-                    if (getNodeCountForType(KubernetesClusterNodeType.valueOf(key), kubernetesCluster) <=0) {
-                        continue;
-                    }
-                    nodeTypeOfferingId = kubernetesCluster.getServiceOfferingId();
+                final ServiceOffering existingServiceOffering = getEffectiveServiceOfferingForNodeType(key, kubernetesCluster);
+                if (existingServiceOffering == null) {
+                    continue;
                 }
-                final ServiceOffering existingServiceOffering = serviceOfferingDao.findById(nodeTypeOfferingId);
                 if (KubernetesCluster.State.Running.equals(kubernetesCluster.getState()) && (serviceOffering.getRamSize() < existingServiceOffering.getRamSize() ||
                         serviceOffering.getCpu() * serviceOffering.getSpeed() < existingServiceOffering.getCpu() * existingServiceOffering.getSpeed())) {
                     logAndThrow(Level.WARN, String.format("Kubernetes cluster cannot be scaled down for service offering. Service offering : %s offers lesser resources as compared to service offering : %s of Kubernetes cluster : %s",
@@ -1446,6 +1442,33 @@
         return kubernetesCluster.getServiceOfferingId();
     }
 
+    // Resolves the offering backing a node type, falling back to the legacy cluster-wide offering; null if the cluster has no nodes of this type.
+    protected ServiceOffering getEffectiveServiceOfferingForNodeType(String key, KubernetesClusterVO kubernetesCluster) {
+        if (getNodeCountForType(KubernetesClusterNodeType.valueOf(key), kubernetesCluster) <= 0) {
+            return null;
+        }
+        Long offeringId = getExistingServiceOfferingIdForNodeType(key, kubernetesCluster);
+        if (offeringId == null) {
+            offeringId = kubernetesCluster.getServiceOfferingId();
+        }
+        if (offeringId == null) {
+            return null;
+        }
+        return serviceOfferingDao.findByIdIncludingRemoved(offeringId);
+    }
+
+    // Effective offerings for every node type actually present on the cluster.
+    protected List<ServiceOffering> getEffectiveServiceOfferingsForCluster(KubernetesClusterVO kubernetesCluster) {
+        List<ServiceOffering> offerings = new ArrayList<>();
+        for (String key : CLUSTER_NODES_TYPES_LIST) {
+            ServiceOffering offering = getEffectiveServiceOfferingForNodeType(key, kubernetesCluster);
+            if (offering != null) {
+                offerings.add(offering);
+            }
+        }
+        return offerings;
+    }
+
     protected void checkServiceOfferingForNodesScale(ServiceOffering serviceOffering, KubernetesClusterVO kubernetesCluster, KubernetesSupportedVersion clusterVersion) {
         if (serviceOffering.isDynamic()) {
             throw new InvalidParameterValueException(String.format("Custom service offerings are not supported for Kubernetes clusters. Kubernetes cluster : %s, service offering : %s", kubernetesCluster.getName(), serviceOffering.getName()));
@@ -1510,17 +1533,19 @@
             throw new InvalidParameterValueException(String.format("Invalid Kubernetes version associated with cluster : %s",
                     kubernetesCluster.getName()));
         }
-        final ServiceOffering serviceOffering = serviceOfferingDao.findByIdIncludingRemoved(kubernetesCluster.getServiceOfferingId());
-        if (serviceOffering == null) {
+        final List<ServiceOffering> effectiveServiceOfferings = getEffectiveServiceOfferingsForCluster(kubernetesCluster);
+        if (CollectionUtils.isEmpty(effectiveServiceOfferings)) {
             throw new CloudRuntimeException(String.format("Invalid service offering associated with Kubernetes cluster : %s", kubernetesCluster.getName()));
         }
-        if (serviceOffering.getCpu() < upgradeVersion.getMinimumCpu()) {
-            throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s cannot be upgraded with Kubernetes version : %s which needs minimum %d vCPUs while associated service offering : %s offers only %d vCPUs",
-                    kubernetesCluster.getName(), upgradeVersion.getName(), upgradeVersion.getMinimumCpu(), serviceOffering.getName(), serviceOffering.getCpu()));
-        }
-        if (serviceOffering.getRamSize() < upgradeVersion.getMinimumRamSize()) {
-            throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s cannot be upgraded with Kubernetes version : %s which needs minimum %d MB RAM while associated service offering : %s offers only %d MB RAM",
-                    kubernetesCluster.getName(), upgradeVersion.getName(), upgradeVersion.getMinimumRamSize(), serviceOffering.getName(), serviceOffering.getRamSize()));
+        for (ServiceOffering serviceOffering : effectiveServiceOfferings) {
+            if (serviceOffering.getCpu() < upgradeVersion.getMinimumCpu()) {
+                throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s cannot be upgraded with Kubernetes version : %s which needs minimum %d vCPUs while associated service offering : %s offers only %d vCPUs",
+                        kubernetesCluster.getName(), upgradeVersion.getName(), upgradeVersion.getMinimumCpu(), serviceOffering.getName(), serviceOffering.getCpu()));
+            }
+            if (serviceOffering.getRamSize() < upgradeVersion.getMinimumRamSize()) {
+                throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s cannot be upgraded with Kubernetes version : %s which needs minimum %d MB RAM while associated service offering : %s offers only %d MB RAM",
+                        kubernetesCluster.getName(), upgradeVersion.getName(), upgradeVersion.getMinimumRamSize(), serviceOffering.getName(), serviceOffering.getRamSize()));
+            }
         }
         // Check upgradeVersion is either patch upgrade or immediate minor upgrade
         try {