engine-d/src/main/java/org/onap/holmes/engine/manager/DroolsEngine.java

   1 /**\r
   2  * Copyright 2017 ZTE Corporation.\r
   3  *\r
   4  * Licensed under the Apache License, Version 2.0 (the "License");\r
   5  * you may not use this file except in compliance with the License.\r
   6  * You may obtain a copy of the License at\r
   7  *\r
   8  * http://www.apache.org/licenses/LICENSE-2.0\r
   9  *\r
  10  * Unless required by applicable law or agreed to in writing, software\r
  11  * distributed under the License is distributed on an "AS IS" BASIS,\r
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  13  * See the License for the specific language governing permissions and\r
  14  * limitations under the License.\r
  15  */\r
  16 package org.onap.holmes.engine.manager;\r
  17 import java.io.StringReader;\r
  18 import java.util.ArrayList;\r
  19 import java.util.HashSet;\r
  20 import java.util.List;\r
  21 import java.util.Locale;\r
  22 import java.util.Set;\r
  23 import javax.annotation.PostConstruct;\r
  24 import javax.inject.Inject;\r
  25 import lombok.extern.slf4j.Slf4j;\r
  26 import org.drools.compiler.kie.builder.impl.InternalKieModule;\r
  27 import org.jvnet.hk2.annotations.Service;\r
  28 \r
  29 import org.kie.api.KieBase;\r
  30 import org.kie.api.KieServices;\r
  31 import org.kie.api.builder.KieBuilder;\r
  32 import org.kie.api.builder.KieFileSystem;\r
  33 import org.kie.api.builder.KieRepository;\r
  34 import org.kie.api.builder.Message;\r
  35 import org.kie.api.builder.Message.Level;\r
  36 import org.kie.api.builder.model.KieBaseModel;\r
  37 import org.kie.api.builder.model.KieModuleModel;\r
  38 import org.kie.api.builder.model.KieSessionModel;\r
  39 import org.kie.api.conf.EqualityBehaviorOption;\r
  40 import org.kie.api.conf.EventProcessingOption;\r
  41 import org.kie.api.definition.KiePackage;\r
  42 import org.kie.api.io.KieResources;\r
  43 import org.kie.api.io.ResourceType;\r
  44 import org.kie.api.runtime.KieContainer;\r
  45 import org.kie.api.runtime.KieSession;\r
  46 import org.kie.api.runtime.conf.ClockTypeOption;\r
  47 import org.kie.api.runtime.rule.FactHandle;\r
  48 \r
  49 import org.onap.holmes.common.api.entity.AlarmInfo;\r
  50 \r
  51 import org.onap.holmes.common.api.stat.VesAlarm;\r
  52 import org.onap.holmes.common.dmaap.DmaapService;\r
  53 import org.onap.holmes.common.exception.AlarmInfoException;\r
  54 import org.onap.holmes.common.utils.DbDaoUtil;\r
  55 import org.onap.holmes.engine.db.AlarmInfoDao;\r
  56 import org.onap.holmes.engine.request.DeployRuleRequest;\r
  57 import org.onap.holmes.common.api.entity.CorrelationRule;\r
  58 import org.onap.holmes.common.exception.CorrelationException;\r
  59 import org.onap.holmes.common.utils.ExceptionUtil;\r
  60 import org.onap.holmes.engine.wrapper.RuleMgtWrapper;\r
  61 \r
  62 @Slf4j\r
  63 @Service\r
  64 public class DroolsEngine {\r
  65 \r
  66     private static final int ENABLE = 1;\r
  67     public static final String UTF_8 = "UTF-8";\r
  68     public static final String K_BASE = "KBase";\r
  69     private static final String RULES_FILE_NAME = "src/main/resources/rules/rule.drl";\r
  70     private final Set<String> packageNames = new HashSet<String>();\r
  71 \r
  72     @Inject\r
  73     private RuleMgtWrapper ruleMgtWrapper;\r
  74 \r
  75 \r
  76     private KieBase kieBase;\r
  77     private KieSession kieSession;\r
  78     private KieContainer kieContainer;\r
  79     private KieFileSystem kfs;\r
  80     private KieServices ks;\r
  81     private KieBuilder kieBuilder;\r
  82     private KieResources resources;\r
  83     private KieRepository kieRepository;\r
  84 \r
  85     private AlarmInfoDao alarmInfoDao;\r
  86     @Inject\r
  87     private DbDaoUtil daoUtil;\r
  88 \r
  89 \r
  90     @PostConstruct\r
  91     private void init() {\r
  92         alarmInfoDao = daoUtil.getJdbiDaoByOnDemand(AlarmInfoDao.class);\r
  93         try {\r
  94             // start engine\r
  95             start();\r
  96         } catch (Exception e) {\r
  97             log.error("Failed to start the service: " + e.getMessage(), e);\r
  98             throw ExceptionUtil.buildExceptionResponse("Failed to start the drools engine!");\r
  99         }\r
 100     }\r
 101 \r
 102     private void start() throws AlarmInfoException {\r
 103         log.info("Drools Engine Initialize Beginning...");\r
 104 \r
 105         initEngineParameter();\r
 106         alarmSynchronization();\r
 107 //        initDeployRule();\r
 108 \r
 109         log.info("Alarm synchronization Successfully.");\r
 110     }\r
 111 \r
 112     public void stop() {\r
 113         this.kieSession.dispose();\r
 114     }\r
 115 \r
 116     public void initEngineParameter() {\r
 117         this.ks = KieServices.Factory.get();\r
 118         this.resources = ks.getResources();\r
 119         this.kieRepository = ks.getRepository();\r
 120         this.kfs = createKieFileSystemWithKProject(ks);\r
 121 \r
 122         this.kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
 123         this.kieContainer = ks.newKieContainer(kieRepository.getDefaultReleaseId());\r
 124 \r
 125         this.kieBase = kieContainer.getKieBase();\r
 126         this.kieSession = kieContainer.newKieSession();\r
 127     }\r
 128 \r
 129     private void initDeployRule() throws CorrelationException {\r
 130         List<CorrelationRule> rules = ruleMgtWrapper.queryRuleByEnable(ENABLE);\r
 131 \r
 132         if (rules.isEmpty()) {\r
 133             return;\r
 134         }\r
 135         for (CorrelationRule rule : rules) {\r
 136             if (rule.getContent() != null) {\r
 137                 deployRuleFromDB(rule.getContent());\r
 138                 DmaapService.loopControlNames.put(rule.getPackageName(), rule.getClosedControlLoopName());\r
 139             }\r
 140         }\r
 141     }\r
 142 \r
 143     private void deployRuleFromDB(String ruleContent) throws CorrelationException {\r
 144         avoidDeployBug();\r
 145         StringReader reader = new StringReader(ruleContent);\r
 146         kfs.write(RULES_FILE_NAME,\r
 147                 this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
 148         kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
 149         try {\r
 150             InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();\r
 151             kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
 152         } catch (Exception e) {\r
 153             throw new CorrelationException(e.getMessage(), e);\r
 154         }\r
 155         kieSession.fireAllRules();\r
 156     }\r
 157 \r
 158     public synchronized String deployRule(DeployRuleRequest rule, Locale locale)\r
 159         throws CorrelationException {\r
 160         avoidDeployBug();\r
 161         StringReader reader = new StringReader(rule.getContent());\r
 162         kfs.write(RULES_FILE_NAME,\r
 163                 this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
 164         kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
 165 \r
 166         judgeRuleContent(locale, kieBuilder, true);\r
 167 \r
 168         InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();;\r
 169         String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
 170         try {\r
 171             kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
 172         } catch (Exception e) {\r
 173             throw new CorrelationException("Failed to deploy the rule.", e);\r
 174         }\r
 175         packageNames.add(packageName);\r
 176         kieSession.fireAllRules();\r
 177         return packageName;\r
 178     }\r
 179 \r
 180     public synchronized void undeployRule(String packageName, Locale locale)\r
 181         throws CorrelationException {\r
 182         KiePackage kiePackage = kieBase.getKiePackage(packageName);\r
 183         if (null == kiePackage) {\r
 184             throw new CorrelationException("The rule " + packageName + " does not exist!");\r
 185         }\r
 186         try {\r
 187             kieBase.removeKiePackage(kiePackage.getName());\r
 188         } catch (Exception e) {\r
 189             throw new CorrelationException("Failed to delete the rule: " + packageName, e);\r
 190         }\r
 191         packageNames.remove(kiePackage.getName());\r
 192     }\r
 193 \r
 194     public void compileRule(String content, Locale locale)\r
 195         throws CorrelationException {\r
 196         StringReader reader = new StringReader(content);\r
 197 \r
 198         kfs.write(RULES_FILE_NAME,\r
 199                 this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
 200 \r
 201         kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
 202 \r
 203         judgeRuleContent(locale, kieBuilder, false);\r
 204     }\r
 205 \r
 206     private void judgeRuleContent(Locale locale, KieBuilder kbuilder, boolean judgePackageName)\r
 207         throws CorrelationException {\r
 208         if (kbuilder.getResults().hasMessages(Message.Level.ERROR)) {\r
 209             String errorMsg = "There are errors in the rule: " + kbuilder.getResults()\r
 210                     .getMessages(Level.ERROR).toString();\r
 211             log.error(errorMsg);\r
 212             throw new CorrelationException(errorMsg);\r
 213         }\r
 214         InternalKieModule internalKieModule = null;\r
 215         try {\r
 216             internalKieModule = (InternalKieModule) kbuilder.getKieModule();\r
 217         } catch (Exception e) {\r
 218             throw new CorrelationException("There are errors in the rule!" + e.getMessage(), e);\r
 219         }\r
 220         if (internalKieModule == null) {\r
 221             throw new CorrelationException("There are errors in the rule!");\r
 222         }\r
 223         String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
 224 \r
 225         if (queryAllPackage().contains(packageName) && judgePackageName) {\r
 226             throw new CorrelationException("The rule " + packageName + " already exists in the drools engine.");\r
 227         }\r
 228     }\r
 229 \r
 230     public void putRaisedIntoStream(VesAlarm alarm) {\r
 231         FactHandle factHandle = this.kieSession.getFactHandle(alarm);\r
 232         if (factHandle != null) {\r
 233             Object obj = this.kieSession.getObject(factHandle);\r
 234             if (obj != null && obj instanceof VesAlarm) {\r
 235                 alarm.setRootFlag(((VesAlarm) obj).getRootFlag());\r
 236             }\r
 237             this.kieSession.delete(factHandle);\r
 238             \r
 239             if (alarm.getAlarmIsCleared() == 1) {\r
 240                 alarmInfoDao.deleteClearedAlarm(convertVesAlarm2AlarmInfo(alarm));\r
 241             }\r
 242         } else {\r
 243             this.kieSession.insert(alarm);\r
 244         }\r
 245 \r
 246         this.kieSession.fireAllRules();\r
 247 \r
 248     }\r
 249 \r
 250     public List<String> queryAllPackage() {\r
 251         List<KiePackage> kiePackages = (List<KiePackage>)kieBase.getKiePackages();\r
 252         List<String> list = new ArrayList<>();\r
 253         for(KiePackage kiePackage : kiePackages) {\r
 254             list.add(kiePackage.getName());\r
 255         }\r
 256         return list;\r
 257     }\r
 258 \r
 259     private KieFileSystem createKieFileSystemWithKProject(KieServices ks) {\r
 260         KieModuleModel kieModuleModel = ks.newKieModuleModel();\r
 261         KieBaseModel kieBaseModel = kieModuleModel.newKieBaseModel(K_BASE)\r
 262                 .addPackage("rules")\r
 263                 .setDefault(true)\r
 264                 .setEqualsBehavior(EqualityBehaviorOption.EQUALITY)\r
 265                 .setEventProcessingMode(EventProcessingOption.STREAM);\r
 266         KieSessionModel kieSessionModel = kieBaseModel.newKieSessionModel("KSession")\r
 267                 .setDefault( true )\r
 268                 .setType( KieSessionModel.KieSessionType.STATEFUL )\r
 269                 .setClockType( ClockTypeOption.get("realtime") );\r
 270         KieFileSystem kfs = ks.newKieFileSystem();\r
 271         kfs.writeKModuleXML(kieModuleModel.toXML());\r
 272         return kfs;\r
 273     }\r
 274 \r
 275     private void avoidDeployBug() {\r
 276         String tmp = Math.random() + "";\r
 277         String rule = "package justInOrderToAvoidDeployBug" + tmp.substring(2);\r
 278         kfs.write(RULES_FILE_NAME, rule);\r
 279         kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
 280         InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();\r
 281         String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
 282         kieRepository.addKieModule(internalKieModule);\r
 283         kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
 284 \r
 285         KiePackage kiePackage = kieBase.getKiePackage(packageName);\r
 286         kieBase.removeKiePackage(kiePackage.getName());\r
 287     }\r
 288 \r
 289     public void alarmSynchronization() throws AlarmInfoException {\r
 290         alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> alarmInfoDao.deleteClearedAlarm(alarmInfo));\r
 291         alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> putRaisedIntoStream(convertAlarmInfo2VesAlarm(alarmInfo)));\r
 292     }\r
 293 \r
 294     private VesAlarm convertAlarmInfo2VesAlarm(AlarmInfo alarmInfo) {\r
 295         VesAlarm vesAlarm = new VesAlarm();\r
 296         vesAlarm.setEventId(alarmInfo.getEventId());\r
 297         vesAlarm.setEventName(alarmInfo.getEventName());\r
 298         vesAlarm.setStartEpochMicrosec(alarmInfo.getStartEpochMicroSec());\r
 299         vesAlarm.setSourceId(alarmInfo.getSourceId());\r
 300         vesAlarm.setSourceName(alarmInfo.getSourceName());\r
 301         vesAlarm.setRootFlag(alarmInfo.getRootFlag());\r
 302         vesAlarm.setAlarmIsCleared(alarmInfo.getAlarmIsCleared());\r
 303         vesAlarm.setLastEpochMicrosec(alarmInfo.getLastEpochMicroSec());\r
 304         return vesAlarm;\r
 305     }\r
 306 \r
 307     private AlarmInfo convertVesAlarm2AlarmInfo(VesAlarm vesAlarm){\r
 308         AlarmInfo alarmInfo = new AlarmInfo();\r
 309         alarmInfo.setEventId(vesAlarm.getEventId());\r
 310         alarmInfo.setEventName(vesAlarm.getEventName());\r
 311         alarmInfo.setStartEpochMicroSec(vesAlarm.getStartEpochMicrosec());\r
 312         alarmInfo.setLastEpochMicroSec(vesAlarm.getLastEpochMicrosec());\r
 313         alarmInfo.setSourceId(vesAlarm.getSourceId());\r
 314         alarmInfo.setSourceName(vesAlarm.getSourceName());\r
 315         alarmInfo.setAlarmIsCleared(vesAlarm.getAlarmIsCleared());\r
 316         alarmInfo.setRootFlag(vesAlarm.getRootFlag());\r
 317 \r
 318         return alarmInfo;\r
 319     }\r
 320 \r
 321 }\r