Optimized Rule Deployment Logic
[holmes/engine-management.git] / engine-d / src / main / java / org / onap / holmes / engine / manager / DroolsEngine.java
index bb3795d..7472457 100644 (file)
-/**\r
- * Copyright 2017 ZTE Corporation.\r
- *\r
- * Licensed under the Apache License, Version 2.0 (the "License");\r
- * you may not use this file except in compliance with the License.\r
- * You may obtain a copy of the License at\r
- *\r
- * http://www.apache.org/licenses/LICENSE-2.0\r
- *\r
- * Unless required by applicable law or agreed to in writing, software\r
- * distributed under the License is distributed on an "AS IS" BASIS,\r
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * See the License for the specific language governing permissions and\r
- * limitations under the License.\r
- */\r
-package org.onap.holmes.engine.manager;\r
-import java.io.StringReader;\r
-import java.util.ArrayList;\r
-import java.util.HashSet;\r
-import java.util.List;\r
-import java.util.Locale;\r
-import java.util.Set;\r
-import javax.annotation.PostConstruct;\r
-import javax.inject.Inject;\r
-import lombok.extern.slf4j.Slf4j;\r
-import org.drools.compiler.kie.builder.impl.InternalKieModule;\r
-import org.jvnet.hk2.annotations.Service;\r
-\r
-import org.kie.api.KieBase;\r
-import org.kie.api.KieServices;\r
-import org.kie.api.builder.KieBuilder;\r
-import org.kie.api.builder.KieFileSystem;\r
-import org.kie.api.builder.KieRepository;\r
-import org.kie.api.builder.Message;\r
-import org.kie.api.builder.Message.Level;\r
-import org.kie.api.builder.model.KieBaseModel;\r
-import org.kie.api.builder.model.KieModuleModel;\r
-import org.kie.api.builder.model.KieSessionModel;\r
-import org.kie.api.conf.EqualityBehaviorOption;\r
-import org.kie.api.conf.EventProcessingOption;\r
-import org.kie.api.definition.KiePackage;\r
-import org.kie.api.io.KieResources;\r
-import org.kie.api.io.ResourceType;\r
-import org.kie.api.runtime.KieContainer;\r
-import org.kie.api.runtime.KieSession;\r
-import org.kie.api.runtime.conf.ClockTypeOption;\r
-import org.kie.api.runtime.rule.FactHandle;\r
-\r
-import org.onap.holmes.common.api.entity.AlarmInfo;\r
-\r
-import org.onap.holmes.common.api.stat.VesAlarm;\r
-import org.onap.holmes.common.dmaap.DmaapService;\r
-import org.onap.holmes.common.exception.AlarmInfoException;\r
-import org.onap.holmes.common.utils.DbDaoUtil;\r
-import org.onap.holmes.engine.db.AlarmInfoDao;\r
-import org.onap.holmes.engine.request.DeployRuleRequest;\r
-import org.onap.holmes.common.api.entity.CorrelationRule;\r
-import org.onap.holmes.common.exception.CorrelationException;\r
-import org.onap.holmes.common.utils.ExceptionUtil;\r
-import org.onap.holmes.engine.wrapper.RuleMgtWrapper;\r
-\r
-@Slf4j\r
-@Service\r
-public class DroolsEngine {\r
-\r
-    private static final int ENABLE = 1;\r
-    public static final String UTF_8 = "UTF-8";\r
-    public static final String K_BASE = "KBase";\r
-    private static final String RULES_FILE_NAME = "src/main/resources/rules/rule.drl";\r
-    private final Set<String> packageNames = new HashSet<String>();\r
-\r
-    @Inject\r
-    private RuleMgtWrapper ruleMgtWrapper;\r
-\r
-\r
-    private KieBase kieBase;\r
-    private KieSession kieSession;\r
-    private KieContainer kieContainer;\r
-    private KieFileSystem kfs;\r
-    private KieServices ks;\r
-    private KieBuilder kieBuilder;\r
-    private KieResources resources;\r
-    private KieRepository kieRepository;\r
-\r
-    private AlarmInfoDao alarmInfoDao;\r
-    @Inject\r
-    private DbDaoUtil daoUtil;\r
-\r
-\r
-    @PostConstruct\r
-    private void init() {\r
-        alarmInfoDao = daoUtil.getJdbiDaoByOnDemand(AlarmInfoDao.class);\r
-        try {\r
-            // start engine\r
-            start();\r
-        } catch (Exception e) {\r
-            log.error("Failed to start the service: " + e.getMessage(), e);\r
-            throw ExceptionUtil.buildExceptionResponse("Failed to start the drools engine!");\r
-        }\r
-    }\r
-\r
-    private void start() throws AlarmInfoException {\r
-        log.info("Drools Engine Initialize Beginning...");\r
-\r
-        initEngineParameter();\r
-        alarmSynchronization();\r
-//        initDeployRule();\r
-\r
-        log.info("Alarm synchronization Successfully.");\r
-    }\r
-\r
-    public void stop() {\r
-        this.kieSession.dispose();\r
-    }\r
-\r
-    public void initEngineParameter() {\r
-        this.ks = KieServices.Factory.get();\r
-        this.resources = ks.getResources();\r
-        this.kieRepository = ks.getRepository();\r
-        this.kfs = createKieFileSystemWithKProject(ks);\r
-\r
-        this.kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
-        this.kieContainer = ks.newKieContainer(kieRepository.getDefaultReleaseId());\r
-\r
-        this.kieBase = kieContainer.getKieBase();\r
-        this.kieSession = kieContainer.newKieSession();\r
-    }\r
-\r
-    private void initDeployRule() throws CorrelationException {\r
-        List<CorrelationRule> rules = ruleMgtWrapper.queryRuleByEnable(ENABLE);\r
-\r
-        if (rules.isEmpty()) {\r
-            return;\r
-        }\r
-        for (CorrelationRule rule : rules) {\r
-            if (rule.getContent() != null) {\r
-                deployRuleFromDB(rule.getContent());\r
-                DmaapService.loopControlNames.put(rule.getPackageName(), rule.getClosedControlLoopName());\r
-            }\r
-        }\r
-    }\r
-\r
-    private void deployRuleFromDB(String ruleContent) throws CorrelationException {\r
-        avoidDeployBug();\r
-        StringReader reader = new StringReader(ruleContent);\r
-        kfs.write(RULES_FILE_NAME,\r
-                this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
-        kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
-        try {\r
-            InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();\r
-            kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
-        } catch (Exception e) {\r
-            throw new CorrelationException(e.getMessage(), e);\r
-        }\r
-        kieSession.fireAllRules();\r
-    }\r
-\r
-    public synchronized String deployRule(DeployRuleRequest rule, Locale locale)\r
-        throws CorrelationException {\r
-        avoidDeployBug();\r
-        StringReader reader = new StringReader(rule.getContent());\r
-        kfs.write(RULES_FILE_NAME,\r
-                this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
-        kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
-\r
-        judgeRuleContent(locale, kieBuilder, true);\r
-\r
-        InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();;\r
-        String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
-        try {\r
-            kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
-        } catch (Exception e) {\r
-            throw new CorrelationException("Failed to deploy the rule.", e);\r
-        }\r
-        packageNames.add(packageName);\r
-        kieSession.fireAllRules();\r
-        return packageName;\r
-    }\r
-\r
-    public synchronized void undeployRule(String packageName, Locale locale)\r
-        throws CorrelationException {\r
-        KiePackage kiePackage = kieBase.getKiePackage(packageName);\r
-        if (null == kiePackage) {\r
-            throw new CorrelationException("The rule " + packageName + " does not exist!");\r
-        }\r
-        try {\r
-            kieBase.removeKiePackage(kiePackage.getName());\r
-        } catch (Exception e) {\r
-            throw new CorrelationException("Failed to delete the rule: " + packageName, e);\r
-        }\r
-        packageNames.remove(kiePackage.getName());\r
-    }\r
-\r
-    public void compileRule(String content, Locale locale)\r
-        throws CorrelationException {\r
-        StringReader reader = new StringReader(content);\r
-\r
-        kfs.write(RULES_FILE_NAME,\r
-                this.resources.newReaderResource(reader, UTF_8).setResourceType(ResourceType.DRL));\r
-\r
-        kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
-\r
-        judgeRuleContent(locale, kieBuilder, false);\r
-    }\r
-\r
-    private void judgeRuleContent(Locale locale, KieBuilder kbuilder, boolean judgePackageName)\r
-        throws CorrelationException {\r
-        if (kbuilder.getResults().hasMessages(Message.Level.ERROR)) {\r
-            String errorMsg = "There are errors in the rule: " + kbuilder.getResults()\r
-                    .getMessages(Level.ERROR).toString();\r
-            log.error(errorMsg);\r
-            throw new CorrelationException(errorMsg);\r
-        }\r
-        InternalKieModule internalKieModule = null;\r
-        try {\r
-            internalKieModule = (InternalKieModule) kbuilder.getKieModule();\r
-        } catch (Exception e) {\r
-            throw new CorrelationException("There are errors in the rule!" + e.getMessage(), e);\r
-        }\r
-        if (internalKieModule == null) {\r
-            throw new CorrelationException("There are errors in the rule!");\r
-        }\r
-        String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
-\r
-        if (queryAllPackage().contains(packageName) && judgePackageName) {\r
-            throw new CorrelationException("The rule " + packageName + " already exists in the drools engine.");\r
-        }\r
-    }\r
-\r
-    public void putRaisedIntoStream(VesAlarm alarm) {\r
-        FactHandle factHandle = this.kieSession.getFactHandle(alarm);\r
-        if (factHandle != null) {\r
-            Object obj = this.kieSession.getObject(factHandle);\r
-            if (obj != null && obj instanceof VesAlarm) {\r
-                alarm.setRootFlag(((VesAlarm) obj).getRootFlag());\r
-            }\r
-            this.kieSession.delete(factHandle);\r
-            \r
-            if (alarm.getAlarmIsCleared() == 1) {\r
-                alarmInfoDao.deleteClearedAlarm(convertVesAlarm2AlarmInfo(alarm));\r
-            }\r
-        } else {\r
-            this.kieSession.insert(alarm);\r
-        }\r
-\r
-        this.kieSession.fireAllRules();\r
-\r
-    }\r
-\r
-    public List<String> queryAllPackage() {\r
-        List<KiePackage> kiePackages = (List<KiePackage>)kieBase.getKiePackages();\r
-        List<String> list = new ArrayList<>();\r
-        for(KiePackage kiePackage : kiePackages) {\r
-            list.add(kiePackage.getName());\r
-        }\r
-        return list;\r
-    }\r
-\r
-    private KieFileSystem createKieFileSystemWithKProject(KieServices ks) {\r
-        KieModuleModel kieModuleModel = ks.newKieModuleModel();\r
-        KieBaseModel kieBaseModel = kieModuleModel.newKieBaseModel(K_BASE)\r
-                .addPackage("rules")\r
-                .setDefault(true)\r
-                .setEqualsBehavior(EqualityBehaviorOption.EQUALITY)\r
-                .setEventProcessingMode(EventProcessingOption.STREAM);\r
-        KieSessionModel kieSessionModel = kieBaseModel.newKieSessionModel("KSession")\r
-                .setDefault( true )\r
-                .setType( KieSessionModel.KieSessionType.STATEFUL )\r
-                .setClockType( ClockTypeOption.get("realtime") );\r
-        KieFileSystem kfs = ks.newKieFileSystem();\r
-        kfs.writeKModuleXML(kieModuleModel.toXML());\r
-        return kfs;\r
-    }\r
-\r
-    private void avoidDeployBug() {\r
-        String tmp = Math.random() + "";\r
-        String rule = "package justInOrderToAvoidDeployBug" + tmp.substring(2);\r
-        kfs.write(RULES_FILE_NAME, rule);\r
-        kieBuilder = ks.newKieBuilder(kfs).buildAll();\r
-        InternalKieModule internalKieModule = (InternalKieModule)kieBuilder.getKieModule();\r
-        String packageName = internalKieModule.getKnowledgePackagesForKieBase(K_BASE).iterator().next().getName();\r
-        kieRepository.addKieModule(internalKieModule);\r
-        kieContainer.updateToVersion(internalKieModule.getReleaseId());\r
-\r
-        KiePackage kiePackage = kieBase.getKiePackage(packageName);\r
-        kieBase.removeKiePackage(kiePackage.getName());\r
-    }\r
-\r
-    public void alarmSynchronization() throws AlarmInfoException {\r
-        alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> alarmInfoDao.deleteClearedAlarm(alarmInfo));\r
-        alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> putRaisedIntoStream(convertAlarmInfo2VesAlarm(alarmInfo)));\r
-    }\r
-\r
-    private VesAlarm convertAlarmInfo2VesAlarm(AlarmInfo alarmInfo) {\r
-        VesAlarm vesAlarm = new VesAlarm();\r
-        vesAlarm.setEventId(alarmInfo.getEventId());\r
-        vesAlarm.setEventName(alarmInfo.getEventName());\r
-        vesAlarm.setStartEpochMicrosec(alarmInfo.getStartEpochMicroSec());\r
-        vesAlarm.setSourceId(alarmInfo.getSourceId());\r
-        vesAlarm.setSourceName(alarmInfo.getSourceName());\r
-        vesAlarm.setRootFlag(alarmInfo.getRootFlag());\r
-        vesAlarm.setAlarmIsCleared(alarmInfo.getAlarmIsCleared());\r
-        vesAlarm.setLastEpochMicrosec(alarmInfo.getLastEpochMicroSec());\r
-        return vesAlarm;\r
-    }\r
-\r
-    private AlarmInfo convertVesAlarm2AlarmInfo(VesAlarm vesAlarm){\r
-        AlarmInfo alarmInfo = new AlarmInfo();\r
-        alarmInfo.setEventId(vesAlarm.getEventId());\r
-        alarmInfo.setEventName(vesAlarm.getEventName());\r
-        alarmInfo.setStartEpochMicroSec(vesAlarm.getStartEpochMicrosec());\r
-        alarmInfo.setLastEpochMicroSec(vesAlarm.getLastEpochMicrosec());\r
-        alarmInfo.setSourceId(vesAlarm.getSourceId());\r
-        alarmInfo.setSourceName(vesAlarm.getSourceName());\r
-        alarmInfo.setAlarmIsCleared(vesAlarm.getAlarmIsCleared());\r
-        alarmInfo.setRootFlag(vesAlarm.getRootFlag());\r
-\r
-        return alarmInfo;\r
-    }\r
-\r
-}\r
+/**
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.holmes.engine.manager;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import javax.annotation.PostConstruct;
+import javax.inject.Inject;
+
+import lombok.extern.slf4j.Slf4j;
+import org.drools.compiler.kie.builder.impl.InternalKieModule;
+import org.drools.core.util.StringUtils;
+import org.jvnet.hk2.annotations.Service;
+
+import org.kie.api.KieServices;
+import org.kie.api.builder.*;
+import org.kie.api.builder.Message.Level;
+import org.kie.api.io.Resource;
+import org.kie.api.runtime.KieContainer;
+import org.kie.api.runtime.KieSession;
+import org.kie.api.runtime.rule.FactHandle;
+
+import org.onap.holmes.common.api.entity.AlarmInfo;
+
+import org.onap.holmes.common.api.stat.VesAlarm;
+import org.onap.holmes.common.dmaap.DmaapService;
+import org.onap.holmes.common.exception.AlarmInfoException;
+import org.onap.holmes.common.utils.DbDaoUtil;
+import org.onap.holmes.engine.db.AlarmInfoDao;
+import org.onap.holmes.engine.request.DeployRuleRequest;
+import org.onap.holmes.common.api.entity.CorrelationRule;
+import org.onap.holmes.common.exception.CorrelationException;
+import org.onap.holmes.common.utils.ExceptionUtil;
+import org.onap.holmes.engine.wrapper.RuleMgtWrapper;
+
+@Slf4j
+@Service
+public class DroolsEngine {
+
+    @Inject
+    private RuleMgtWrapper ruleMgtWrapper;
+    @Inject
+    private DbDaoUtil daoUtil;
+
+    private final static int ENABLE = 1;
+    private AlarmInfoDao alarmInfoDao;
+    private final Map<String, String> deployed = new ConcurrentHashMap<>();
+    private KieServices ks = KieServices.Factory.get();
+    private ReleaseId releaseId = ks.newReleaseId("org.onap.holmes", "rules", "1.0.0-SNAPSHOT");
+    private ReleaseId compilationRelease = ks.newReleaseId("org.onap.holmes", "compilation", "1.0.0-SNAPSHOT");
+    private KieContainer container;
+    private KieSession session;
+
+    @PostConstruct
+    private void init() {
+        alarmInfoDao = daoUtil.getJdbiDaoByOnDemand(AlarmInfoDao.class);
+        try {
+            log.info("Drools engine initializing...");
+            initEngine();
+            log.info("Drools engine initialized.");
+
+            log.info("Start deploy existing rules...");
+            initRules();
+            log.info("All rules were deployed.");
+
+            log.info("Synchronizing alarms...");
+            syncAlarms();
+            log.info("Alarm synchronization succeeded.");
+        } catch (Exception e) {
+            log.error("Failed to startup the engine of Holmes: " + e.getMessage(), e);
+            throw ExceptionUtil.buildExceptionResponse("Failed to startup Drools!");
+        }
+    }
+
+    public void stop() {
+        session.dispose();
+    }
+
+    public void initEngine() {
+        KieModule km = null;
+        try {
+            String drl = "package holmes;";
+            deployed.put(getPackageName(drl), drl);
+            km = createAndDeployJar(ks, releaseId, new ArrayList<>(deployed.values()));
+        } catch (Exception e) {
+            log.error("Failed to initialize the engine service module.", e);
+        }
+        container = ks.newKieContainer(km.getReleaseId());
+        session = container.newKieSession();
+        deployed.clear();
+    }
+
+    private void initRules() throws CorrelationException {
+        List<CorrelationRule> rules = ruleMgtWrapper.queryRuleByEnable(ENABLE);
+        if (rules.isEmpty()) {
+            return;
+        }
+
+        for (CorrelationRule rule : rules) {
+            if (!StringUtils.isEmpty(rule.getContent())) {
+                deployRule(rule.getContent());
+                DmaapService.loopControlNames.put(rule.getPackageName(), rule.getClosedControlLoopName());
+            }
+        }
+
+        session.fireAllRules();
+    }
+
+    public void syncAlarms() throws AlarmInfoException {
+        alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> alarmInfoDao.deleteClearedAlarm(alarmInfo));
+        alarmInfoDao.queryAllAlarm().forEach(alarmInfo -> putRaisedIntoStream(convertAlarmInfo2VesAlarm(alarmInfo)));
+    }
+
+    public String deployRule(DeployRuleRequest rule) throws CorrelationException {
+        return deployRule(rule.getContent());
+    }
+
+    private synchronized String deployRule(String rule) throws CorrelationException {
+        final String packageName = getPackageName(rule);
+
+        if (StringUtils.isEmpty(packageName)) {
+            throw new CorrelationException("The package name can not be empty.");
+        }
+
+        if (deployed.containsKey(packageName)) {
+            throw new CorrelationException("A rule with the same package name already exists in the system.");
+        }
+
+        if (!StringUtils.isEmpty(rule)) {
+            deployed.put(packageName, rule);
+            try {
+                refreshInMemRules();
+            } catch (CorrelationException e) {
+                deployed.remove(packageName);
+                throw e;
+            }
+            session.fireAllRules();
+        }
+
+        return packageName;
+    }
+
+    public synchronized void undeployRule(String packageName) throws CorrelationException {
+
+        if (StringUtils.isEmpty(packageName)) {
+            throw new CorrelationException("The package name should not be null.");
+        }
+
+        if (!deployed.containsKey(packageName)) {
+            throw new CorrelationException("The rule " + packageName + " does not exist!");
+        }
+
+        String removed = deployed.remove(packageName);
+        try {
+            refreshInMemRules();
+        } catch (Exception e) {
+            deployed.put(packageName, removed);
+            throw new CorrelationException("Failed to delete the rule: " + packageName, e);
+        }
+    }
+
+    private void refreshInMemRules() throws CorrelationException {
+        KieModule km = createAndDeployJar(ks, releaseId, new ArrayList<>(deployed.values()));
+        container.updateToVersion(km.getReleaseId());
+    }
+
+    public void compileRule(String content)
+            throws CorrelationException {
+
+        KieFileSystem kfs = ks.newKieFileSystem().generateAndWritePomXML(compilationRelease);
+        kfs.write("src/main/resources/rules/rule.drl", content);
+        KieBuilder builder = ks.newKieBuilder(kfs).buildAll();
+        if (builder.getResults().hasMessages(Message.Level.ERROR)) {
+            String errorMsg = "There are errors in the rule: " + builder.getResults()
+                    .getMessages(Level.ERROR).toString();
+            log.info("Compilation failure: " + errorMsg);
+            throw new CorrelationException(errorMsg);
+        }
+
+        if (deployed.containsKey(getPackageName(content))) {
+            throw new CorrelationException("There's no compilation error. But a rule with the same package name already " +
+                    "exists in the engine, which may cause a deployment failure.");
+        }
+
+        ks.getRepository().removeKieModule(compilationRelease);
+    }
+
+    public void putRaisedIntoStream(VesAlarm alarm) {
+        FactHandle factHandle = this.session.getFactHandle(alarm);
+        if (factHandle != null) {
+            Object obj = this.session.getObject(factHandle);
+            if (obj != null && obj instanceof VesAlarm) {
+                alarm.setRootFlag(((VesAlarm) obj).getRootFlag());
+            }
+            this.session.delete(factHandle);
+
+            if (alarm.getAlarmIsCleared() == 1) {
+                alarmInfoDao.deleteClearedAlarm(convertVesAlarm2AlarmInfo(alarm));
+            }
+        } else {
+            this.session.insert(alarm);
+        }
+
+        this.session.fireAllRules();
+    }
+
+    public List<String> queryPackagesFromEngine() {
+        return container.getKieBase().getKiePackages().stream()
+                .filter(pkg -> pkg.getRules().size() != 0)
+                .map(pkg -> pkg.getName())
+                .collect(Collectors.toList());
+    }
+
+
+
+    private VesAlarm convertAlarmInfo2VesAlarm(AlarmInfo alarmInfo) {
+        VesAlarm vesAlarm = new VesAlarm();
+        vesAlarm.setEventId(alarmInfo.getEventId());
+        vesAlarm.setEventName(alarmInfo.getEventName());
+        vesAlarm.setStartEpochMicrosec(alarmInfo.getStartEpochMicroSec());
+        vesAlarm.setSourceId(alarmInfo.getSourceId());
+        vesAlarm.setSourceName(alarmInfo.getSourceName());
+        vesAlarm.setRootFlag(alarmInfo.getRootFlag());
+        vesAlarm.setAlarmIsCleared(alarmInfo.getAlarmIsCleared());
+        vesAlarm.setLastEpochMicrosec(alarmInfo.getLastEpochMicroSec());
+        return vesAlarm;
+    }
+
+    private AlarmInfo convertVesAlarm2AlarmInfo(VesAlarm vesAlarm) {
+        AlarmInfo alarmInfo = new AlarmInfo();
+        alarmInfo.setEventId(vesAlarm.getEventId());
+        alarmInfo.setEventName(vesAlarm.getEventName());
+        alarmInfo.setStartEpochMicroSec(vesAlarm.getStartEpochMicrosec());
+        alarmInfo.setLastEpochMicroSec(vesAlarm.getLastEpochMicrosec());
+        alarmInfo.setSourceId(vesAlarm.getSourceId());
+        alarmInfo.setSourceName(vesAlarm.getSourceName());
+        alarmInfo.setAlarmIsCleared(vesAlarm.getAlarmIsCleared());
+        alarmInfo.setRootFlag(vesAlarm.getRootFlag());
+
+        return alarmInfo;
+    }
+
+    private String getPackageName(String contents) {
+        String ret = contents.trim();
+        StringBuilder stringBuilder = new StringBuilder();
+        if (ret.startsWith("package")) {
+            ret = ret.substring(7).trim();
+            for (int i = 0; i < ret.length(); i++) {
+                char tmp = ret.charAt(i);
+                if (tmp == ';' || tmp == ' ' || tmp == '\n') {
+                    break;
+                }
+                stringBuilder.append(tmp);
+            }
+        }
+        return stringBuilder.toString();
+    }
+
+    private KieModule createAndDeployJar(KieServices ks, ReleaseId releaseId, List<String> drls) throws CorrelationException {
+        byte[] jar = createJar(ks, releaseId, drls);
+        KieModule km = deployJarIntoRepository(ks, jar);
+        return km;
+    }
+
+    private byte[] createJar(KieServices ks, ReleaseId releaseId, List<String> drls) throws CorrelationException {
+        KieFileSystem kfs = ks.newKieFileSystem().generateAndWritePomXML(releaseId);
+        int i = 0;
+        for (String drl : drls) {
+            if (!StringUtils.isEmpty(drl)) {
+                kfs.write("src/main/resources/" + getPackageName(drl) + ".drl", drl);
+            }
+        }
+        KieBuilder kb = ks.newKieBuilder(kfs).buildAll();
+        if (kb.getResults().hasMessages(Message.Level.ERROR)) {
+            StringBuilder sb = new StringBuilder();
+            for (Message msg : kb.getResults().getMessages()) {
+                sb.append(String.format("[%s]Line: %d, Col: %d\t%s\n", msg.getLevel().toString(), msg.getLine(),
+                        msg.getColumn(), msg.getText()));
+            }
+            throw new CorrelationException("Failed to compile JAR. Details: \n" + sb.toString());
+        }
+
+        InternalKieModule kieModule = (InternalKieModule) ks.getRepository()
+                .getKieModule(releaseId);
+
+        return kieModule.getBytes();
+    }
+
+    private KieModule deployJarIntoRepository(KieServices ks, byte[] jar) {
+        Resource jarRes = ks.getResources().newByteArrayResource(jar);
+        return ks.getRepository().addKieModule(jarRes);
+    }
+
+}