X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=rulemgt%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fholmes%2Frulemgt%2FRuleAllocator.java;h=79ffa31ae5ffd80e99bef9d2dd021c4da2667ef3;hb=a796bda2f8e2db093ce9f528f6e5960e4aa1b22c;hp=2dc05ee1a9c7b041d6503fa6a24fa9799f59f190;hpb=348ce6e112876f552a939e58d74376704537344e;p=holmes%2Frule-management.git

diff --git a/rulemgt/src/main/java/org/onap/holmes/rulemgt/RuleAllocator.java b/rulemgt/src/main/java/org/onap/holmes/rulemgt/RuleAllocator.java
index 2dc05ee..79ffa31 100644
--- a/rulemgt/src/main/java/org/onap/holmes/rulemgt/RuleAllocator.java
+++ b/rulemgt/src/main/java/org/onap/holmes/rulemgt/RuleAllocator.java
@@ -1,5 +1,5 @@
 /**
- * Copyright 2017-2020 ZTE Corporation.
+ * Copyright 2017-2021 ZTE Corporation.
  * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,44 +17,45 @@
 package org.onap.holmes.rulemgt;
 
 import lombok.extern.slf4j.Slf4j;
-import org.jvnet.hk2.annotations.Service;
 import org.onap.holmes.common.api.entity.CorrelationRule;
 import org.onap.holmes.common.exception.CorrelationException;
-import org.onap.holmes.common.utils.DbDaoUtil;
 import org.onap.holmes.rulemgt.bolt.enginebolt.EngineWrapper;
-import org.onap.holmes.rulemgt.db.CorrelationRuleDao;
+import org.onap.holmes.rulemgt.db.CorrelationRuleService;
 import org.onap.holmes.rulemgt.tools.EngineTools;
 import org.onap.holmes.rulemgt.wrapper.RuleMgtWrapper;
 import org.onap.holmes.rulemgt.wrapper.RuleQueryWrapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
-import javax.inject.Inject;
 import java.util.*;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 @Slf4j
-@Service
+@Component
 public class RuleAllocator {
     private static final Logger LOGGER = LoggerFactory.getLogger(RuleAllocator.class);
 
     public final static int ENABLE = 1;
+    public final static int RETRY_TIMES = 5;
+    public final static long RETRY_INTERVAL_SEC = 15;
     private RuleMgtWrapper ruleMgtWrapper;
     private RuleQueryWrapper ruleQueryWrapper;
     private EngineWrapper engineWrapper;
     private EngineTools engineTools;
-    private CorrelationRuleDao correlationRuleDao;
+    private CorrelationRuleService correlationRuleService;
 
-    @Inject
+    @Autowired
     public RuleAllocator(RuleMgtWrapper ruleMgtWrapper, RuleQueryWrapper ruleQueryWrapper,
-                         EngineWrapper engineWrapper, EngineTools engineTools, DbDaoUtil daoUtil) {
+                         EngineWrapper engineWrapper, EngineTools engineTools, CorrelationRuleService correlationRuleService) {
         this.ruleMgtWrapper = ruleMgtWrapper;
         this.ruleQueryWrapper = ruleQueryWrapper;
         this.engineWrapper = engineWrapper;
         this.engineTools = engineTools;
-        correlationRuleDao = daoUtil.getJdbiDaoByOnDemand(CorrelationRuleDao.class);
+        this.correlationRuleService = correlationRuleService;
     }
 
     @PostConstruct
@@ -87,15 +88,16 @@ public class RuleAllocator {
             return;
         }
 
-        if (legacyEngineInstances.size() < numOfEngines) {
-            //extend
+        if (legacyEngineInstances.size() < numOfEngines) { // extend
             List<CorrelationRule> rules2Allocate = calculateRule(legacyEngineInstances, numOfEngines);
             List<CorrelationRule> rules2Delete = copyList(rules2Allocate);
             List<String> newInstanceIds = sortOutNewEngineInstances(engines, legacyEngineInstances);
             distributeRules(newInstanceIds, rules2Allocate);
             cleanUpRulesFromEngines(rules2Delete, legacyEngineInstances);
-        } else {
-            //destroy
+        } else { // destroy
+            // If new engine instances share the same IP addresses with the old ones, the
+            // rule management module will simply leave the them to cope with the legacy rules.
+            // Here, it only takes care of the rules that need to be moved from one IP address to another.
             List<String> destroyed = getDestroyedEngines(engines, legacyEngineInstances);
             distributeRules(getRemainingEngines(engines, destroyed), getRules(destroyed));
         }
@@ -194,7 +196,7 @@ public class RuleAllocator {
 
     // Sorted by the number of rules each engine contains, in a descending order.
     private List<String> sortIpByRuleNumDesc(List<String> ips) {
-        List<CorrelationRule> rules = null;
+        List<CorrelationRule> rules;
         Map<String, Integer> ruleNumOfEngines = new HashMap();
 
         try {
@@ -219,12 +221,27 @@ public class RuleAllocator {
     }
 
     private void allocateRule(CorrelationRule rule, String ip) throws CorrelationException {
-        try {
-            ruleMgtWrapper.deployRule2Engine(rule, ip);
-            correlationRuleDao.updateRule(rule);
-        } catch (CorrelationException e) {
-            throw new CorrelationException(String.format("Failed to allocate rule <%s> to <%s>",
-                    rule.getName(), ip), e);
+        // Retry for a couple of times in case of deployment failure
+        // due to unfinished initialization procedures of engine instances.
+        for (int i = 0; i <= RETRY_TIMES; ++i) {
+            try {
+                ruleMgtWrapper.deployRule2Engine(rule, ip);
+                correlationRuleService.updateRule(rule);
+                // If the codes reach here, it means everything's okay. There's no need to run the loop more.
+                break;
+            } catch (CorrelationException e) {
+                LOGGER.warn(String.format("Failed to allocate rule <%s> to <%s>. Retry: %d.",
+                        rule.getName(), ip, i), e);
+                if (i == RETRY_TIMES) {
+                    throw new CorrelationException(String.format("Failed to allocate rule <%s> to <%s>",
+                            rule.getName(), ip), e);
+                }
+                try {
+                    SECONDS.sleep(RETRY_INTERVAL_SEC * (i + 1));
+                } catch (InterruptedException interruptedException) {
+                    LOGGER.info(interruptedException.getMessage(), interruptedException);
+                }
+            }
         }
     }